feat(plugins): control how credentials are post using TokenAuth

eodag/config.py

@@ -548,6 +548,10 @@ class MetadataPreMapping(TypedDict, total=False):
     #: Credentials json structure if they should be sent as POST data
     req_data: dict[str, Any]
     #: :class:`~eodag.plugins.authentication.token.TokenAuth`
+    #: Whether to send credentials as POST data. If ``True``, always sent; if ``False``, never sent;
+    #: if not set, sent only when credentials are not embedded in ``auth_uri``
+    post_credentials: bool
+    #: :class:`~eodag.plugins.authentication.token.TokenAuth`
     #: URL used to fetch the access token with a refresh token
     refresh_uri: str
     #: :class:`~eodag.plugins.authentication.token.TokenAuth`

eodag/plugins/authentication/token.py

@@ -86,6 +86,9 @@ class TokenAuth(Authentication):
           returned in case of an authentication error
         * :attr:`~eodag.config.PluginConfig.req_data` (``dict[str, Any]``): if the credentials
           should be sent as data in the post request, the json structure can be given in this parameter
+        * :attr:`~eodag.config.PluginConfig.post_credentials` (``bool``): if ``True``, credentials are always
+          sent as POST data; if ``False``, they are never sent; if not set, credentials are sent only when they are not
+          already embedded in :attr:`~eodag.config.PluginConfig.auth_uri`
         * :attr:`~eodag.config.PluginConfig.retry_total` (``int``): :class:`urllib3.util.Retry` ``total`` parameter,
           total number of retries to allow; default: ``3``
         * :attr:`~eodag.config.PluginConfig.retry_backoff_factor` (``int``): :class:`urllib3.util.Retry`
@@ -265,8 +268,17 @@ def set_request_data(call_refresh: bool) -> None:
             if method != "POST":
                 return
 
-            # append req_data to credentials if specified in config
-            data = dict(getattr(self.config, "req_data", {}), **self.config.credentials)
+            # send req_data if specified in config
+            data = getattr(self.config, "req_data", {})
+            # append credendials if needed
+            creds_in_auth_uri = all(
+                x in self.config.auth_uri for x in self.config.credentials.values()
+            )
+            post_credentials = getattr(self.config, "post_credentials", None)
+            if post_credentials is True or (
+                post_credentials is None and not creds_in_auth_uri
+            ):
+                data |= self.config.credentials
 
             # when refreshing the token, we pass only the client_id/secret if present,
             # not other parameters (username/password, scope, ...)

tests/units/test_auth_plugins.py

@@ -168,6 +168,22 @@ def setUpClass(cls):
                         "auth_error_code": 401,
                     },
                 },
+                "provider_text_token_post_credentials_true": {
+                    "products": {"foo_product": {}},
+                    "auth": {
+                        "type": "TokenAuth",
+                        "auth_uri": "http://foo.bar?username={username}",
+                        "post_credentials": True,
+                    },
+                },
+                "provider_text_token_post_credentials_false": {
+                    "products": {"foo_product": {}},
+                    "auth": {
+                        "type": "TokenAuth",
+                        "auth_uri": "http://foo.bar",
+                        "post_credentials": False,
+                    },
+                },
             }
         )
 
@@ -559,6 +575,67 @@ def test_plugins_auth_tokenauth_get_method_auth_tuple_incomplete(
         ):
             auth_plugin.authenticate()
 
+    @mock.patch(
+        "eodag.plugins.authentication.token.requests.Session.request", autospec=True
+    )
+    def test_plugins_auth_tokenauth_post_credentials_true(self, mock_requests_post):
+        """TokenAuth.authenticate must post credentials when post_credentials is True even if creds are in URI"""
+        auth_plugin = self.get_auth_plugin("provider_text_token_post_credentials_true")
+
+        auth_plugin.config.credentials = {"username": "bar", "baz": "qux"}
+
+        # mock token post request response
+        mock_requests_post.return_value = mock.Mock()
+        mock_requests_post.return_value.text = "this_is_test_token"
+
+        auth = auth_plugin.authenticate()
+        self.assertTrue(isinstance(auth, AuthBase))
+
+        # credentials must be in data even though they are embedded in auth_uri
+        args, kwargs = mock_requests_post.call_args
+        self.assertDictEqual(kwargs["data"], {"username": "bar", "baz": "qux"})
+
+    @mock.patch(
+        "eodag.plugins.authentication.token.requests.Session.request", autospec=True
+    )
+    def test_plugins_auth_tokenauth_post_credentials_false(self, mock_requests_post):
+        """TokenAuth.authenticate must not post credentials when post_credentials is False"""
+        auth_plugin = self.get_auth_plugin("provider_text_token_post_credentials_false")
+
+        auth_plugin.config.credentials = {"foo": "bar", "baz": "qux"}
+
+        # mock token post request response
+        mock_requests_post.return_value = mock.Mock()
+        mock_requests_post.return_value.text = "this_is_test_token"
+
+        auth = auth_plugin.authenticate()
+        self.assertTrue(isinstance(auth, AuthBase))
+
+        # credentials must NOT be in data even though they are not in auth_uri
+        args, kwargs = mock_requests_post.call_args
+        self.assertDictEqual(kwargs["data"], {})
+
+    @mock.patch(
+        "eodag.plugins.authentication.token.requests.Session.request", autospec=True
+    )
+    def test_plugins_auth_tokenauth_post_credentials_default(self, mock_requests_post):
+        """TokenAuth.authenticate must not post credentials by default when they are already in auth_uri"""
+        auth_plugin = self.get_auth_plugin("provider_text_token_format_url")
+
+        # use a single credential whose value will appear in the formatted auth_uri
+        auth_plugin.config.credentials = {"username": "bar"}
+
+        # mock token post request response
+        mock_requests_post.return_value = mock.Mock()
+        mock_requests_post.return_value.text = "this_is_test_token"
+
+        auth = auth_plugin.authenticate()
+        self.assertTrue(isinstance(auth, AuthBase))
+
+        # credentials must NOT be in data because all values are in auth_uri
+        args, kwargs = mock_requests_post.call_args
+        self.assertDictEqual(kwargs["data"], {})
+
     def test_plugins_auth_tokenauth_request_error(self):
         """TokenAuth.authenticate must raise an AuthenticationError if a request error occurs"""
         auth_plugin = self.get_auth_plugin("provider_text_token_auth_error_code")