Add authenticate_oidc_access_token() for auth with out-of-band access token #590

Author: soxofaan

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Add experimental `openeo.testing.results` subpackage with reusable test utilities for comparing batch job results with reference data
 - `MultiBackendJobManager`: add initial support for storing job metadata in Parquet file (instead of CSV) ([#571](https://github.com/Open-EO/openeo-python-client/issues/571))
+- Add `Connection.authenticate_oidc_access_token()` to set up authorization headers with an access token that is obtained "out-of-band" ([#598](https://github.com/Open-EO/openeo-python-client/issues/598))
 
 ### Changed
 

openeo/rest/connection.py

@@ -382,13 +382,19 @@ def authenticate_basic(self, username: Optional[str] = None, password: Optional[
         self.auth = BasicBearerAuth(access_token=resp["access_token"])
         return self
 
-    def _get_oidc_provider(self, provider_id: Union[str, None] = None) -> Tuple[str, OidcProviderInfo]:
+    def _get_oidc_provider(
+        self, provider_id: Union[str, None] = None, parse_info: bool = True
+    ) -> Tuple[str, Union[OidcProviderInfo, None]]:
         """
-        Get OpenID Connect discovery URL for given provider_id
+        Get provider id and info, based on context.
+        If provider_id is given, verify it against backend's list of providers.
+        If not given, find a suitable provider based on env vars, config or backend's default.
 
         :param provider_id: id of OIDC provider as specified by backend (/credentials/oidc).
             Can be None if there is just one provider.
-        :return: updated provider_id and provider info object
+        :param parse_info: whether to parse the provider info into an :py:class:`OidcProviderInfo` object
+            (which involves a ".well-known/openid-configuration" request)
+        :return: resolved/verified provider_id and provider info object (unless ``parse_info`` is False)
         """
         oidc_info = self.get("/credentials/oidc", expected_status=200).json()
         providers = OrderedDict((p["id"], p) for p in oidc_info["providers"])
@@ -434,8 +440,10 @@ def _get_oidc_provider(self, provider_id: Union[str, None] = None) -> Tuple[str,
                 _log.info(
                     f"No OIDC provider given. Using first provider {provider_id!r} as advertised by backend."
                 )
-        provider = OidcProviderInfo.from_dict(provider)
-        return provider_id, provider
+
+        provider_info = OidcProviderInfo.from_dict(provider) if parse_info else None
+
+        return provider_id, provider_info
 
     def _get_oidc_provider_and_client_info(
         self,
@@ -766,6 +774,26 @@ def authenticate_oidc(
         print("Authenticated using device code flow.")
         return con
 
+    def authenticate_oidc_access_token(self, access_token: str, provider_id: Optional[str] = None) -> None:
+        """
+        Set up authorization headers directly with an OIDC access token.
+
+        :py:class:`Connection` provides multiple methods to handle various OIDC authentication flows end-to-end.
+        If you already obtained a valid OIDC access token in another "out-of-band" way, you can use this method to
+        set up the authorization headers appropriately.
+
+        :param access_token: OIDC access token
+        :param provider_id: id of the OIDC provider as listed by the openEO backend (``/credentials/oidc``).
+            If not specified, the first (default) OIDC provider will be used.
+        :param skip_verification: Skip clients-side verification of the provider_id
+            against the backend's list of providers to avoid and related OIDC configuration
+
+        .. versionadded:: 0.31.0
+        """
+        provider_id, _ = self._get_oidc_provider(provider_id=provider_id, parse_info=False)
+        self.auth = OidcBearerAuth(provider_id=provider_id, access_token=access_token)
+        self._oidc_auth_renewer = None
+
     def request(
         self,
         method: str,

tests/rest/test_connection.py

@@ -2349,6 +2349,36 @@ def test_authenticate_oidc_auto_renew_expired_access_token_initial_client_creden
     assert "Failed to obtain new access token (grant 'client_credentials')" in caplog.text
 
 
+class TestAuthenticateOidcAccessToken:
+    @pytest.fixture(autouse=True)
+    def _setup(self, requests_mock):
+        requests_mock.get(API_URL, json=build_capabilities())
+        requests_mock.get(
+            API_URL + "credentials/oidc",
+            json={"providers": [{"id": "oi", "issuer": "https://oidc.test", "title": "example", "scopes": ["openid"]}]},
+        )
+
+    def test_authenticate_oidc_access_token_default_provider(self):
+        connection = Connection(API_URL)
+        connection.authenticate_oidc_access_token(access_token="Th3Tok3n!@#")
+        assert isinstance(connection.auth, BearerAuth)
+        assert connection.auth.bearer == "oidc/oi/Th3Tok3n!@#"
+
+    def test_authenticate_oidc_access_token_with_provider(self):
+        connection = Connection(API_URL)
+        connection.authenticate_oidc_access_token(access_token="Th3Tok3n!@#", provider_id="oi")
+        assert isinstance(connection.auth, BearerAuth)
+        assert connection.auth.bearer == "oidc/oi/Th3Tok3n!@#"
+
+    def test_authenticate_oidc_access_token_wrong_provider(self):
+        connection = Connection(API_URL)
+        with pytest.raises(
+            OpenEoClientException,
+            match=re.escape("Requested OIDC provider 'nope' not available. Should be one of ['oi']."),
+        ):
+            connection.authenticate_oidc_access_token(access_token="Th3Tok3n!@#", provider_id="nope")
+
+
 def test_load_collection_arguments_100(requests_mock):
     requests_mock.get(API_URL, json={"api_version": "1.0.0"})
     conn = Connection(API_URL)