Issue #225: Add support for OIDC device auth grant without PKCE nor c…

…lient secret

CHANGELOG.md

@@ -10,6 +10,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - Add command line tool `openeo-auth token-clear` to remove OIDC refresh token cache
+- Add support for OIDC device authorization grant without PKCE nor client secret,
+  ([#225](https://github.com/Open-EO/openeo-python-client/issues/225), [openeo-api#410](https://github.com/Open-EO/openeo-api/issues/410))
 
 ### Changed
 

openeo/rest/auth/oidc.py

@@ -222,11 +222,18 @@ class DefaultOidcClientGrant(enum.Enum):
     Enum with possible values for "grant_types" field of default OIDC clients provided by backend.
     """
     IMPLICIT = "implicit"
+    AUTH_CODE = "authorization_code"
     AUTH_CODE_PKCE = "authorization_code+pkce"
+    DEVICE_CODE = "urn:ietf:params:oauth:grant-type:device_code"
     DEVICE_CODE_PKCE = "urn:ietf:params:oauth:grant-type:device_code+pkce"
     REFRESH_TOKEN = "refresh_token"
 
 
+# Type hint for function that checks if given list of OIDC grant types (DefaultOidcClientGrant enum values)
+# fulfills a criterion.
+GrantsChecker = Union[List[DefaultOidcClientGrant], Callable[[List[DefaultOidcClientGrant]], bool]]
+
+
 class OidcProviderInfo:
     """OpenID Connect Provider information, as provided by an openEO back-end (endpoint `/credentials/oidc`)"""
 
@@ -274,12 +281,28 @@ def get_scopes_string(self, request_refresh_token: bool = False):
         log.debug("Using scopes: {s}".format(s=scopes))
         return " ".join(sorted(scopes))
 
-    def get_default_client_id(self, grant_types: List[DefaultOidcClientGrant]) -> Union[str, None]:
-        """Get first default client supporting the given grant types"""
+    def get_default_client_id(self, grant_check: GrantsChecker) -> Union[str, None]:
+        """
+        Get first default client that supports (as stated by provider's `grant_types`)
+        the desired grant types (as implemented by `grant_check`)
+        """
+        if isinstance(grant_check, list):
+            # Simple `grant_check` mode: just provide list of grants that all must be supported.
+            desired_grants = grant_check
+            grant_check = lambda grants: all(g in grants for g in desired_grants)
+
+        def normalize_grants(grants: List[str]):
+            for grant in grants:
+                try:
+                    yield DefaultOidcClientGrant(grant)
+                except ValueError:
+                    log.warning(f"Invalid OIDC grant type {grant!r}.")
+
         for client in self.default_clients or []:
             client_id = client.get("id")
             supported_grants = client.get("grant_types")
-            if client_id and supported_grants and all(g.value in supported_grants for g in grant_types):
+            supported_grants = list(normalize_grants(supported_grants))
+            if client_id and supported_grants and grant_check(supported_grants):
                 return client_id
 
 
@@ -296,6 +319,13 @@ def __init__(self, client_id: str, provider: OidcProviderInfo, client_secret: st
 
     # TODO: load from config file
 
+    def guess_device_flow_pkce_support(self):
+        """Best effort guess if PKCE should be used for device auth grant"""
+        # Check if this client is also defined as default client with device_code+pkce
+        default_clients = [c for c in self.provider.default_clients or [] if c["id"] == self.client_id]
+        grant_types = set(g for c in default_clients for g in c.get("grant_types", []))
+        return any("device_code+pkce" in g for g in grant_types)
+
 
 class OidcAuthenticator:
     """
@@ -614,13 +644,10 @@ def __init__(
         # Allow to specify/override device code URL for cases when it is not available in OIDC discovery doc.
         self._device_code_url = device_code_url or self._provider_config.get("device_authorization_endpoint")
         if not self._device_code_url:
-            raise OidcException("No support for device code flow")
+            raise OidcException("No support for device authorization grant")
         self._max_poll_time = max_poll_time
         if use_pkce is None:
-            # TODO: better auto-detection if PKCE should/can be used, e.g.:
-            #       does OIDC provider supports device flow + PKCE? Get this from `OidcProviderInfo`?
-            #       (also see https://github.com/Open-EO/openeo-api/pull/366)
-            use_pkce = client_info.client_secret is None
+            use_pkce = client_info.client_secret is None and client_info.guess_device_flow_pkce_support()
         self._pkce = PkceCode() if use_pkce else None
 
     def _get_verification_info(self, request_refresh_token: bool = False) -> VerificationInfo:

openeo/rest/connection.py

@@ -26,7 +26,7 @@
 from openeo.rest.auth.config import RefreshTokenStore, AuthConfig
 from openeo.rest.auth.oidc import OidcClientCredentialsAuthenticator, OidcAuthCodePkceAuthenticator, \
     OidcClientInfo, OidcAuthenticator, OidcRefreshTokenAuthenticator, OidcResourceOwnerPasswordAuthenticator, \
-    OidcDeviceAuthenticator, OidcProviderInfo, OidcException, DefaultOidcClientGrant
+    OidcDeviceAuthenticator, OidcProviderInfo, OidcException, DefaultOidcClientGrant, GrantsChecker
 from openeo.rest.datacube import DataCube
 from openeo.rest.imagecollectionclient import ImageCollectionClient
 from openeo.rest.job import RESTJob
@@ -338,7 +338,7 @@ def _get_oidc_provider(self, provider_id: Union[str, None] = None) -> Tuple[str,
     def _get_oidc_provider_and_client_info(
             self, provider_id: str,
             client_id: Union[str, None], client_secret: Union[str, None],
-            default_client_grant_types: Union[None, List[DefaultOidcClientGrant]] = None
+            default_client_grant_check: Union[None, GrantsChecker] = None
     ) -> Tuple[str, OidcClientInfo]:
         """
         Resolve provider_id and client info (as given or from config)
@@ -357,10 +357,10 @@ def _get_oidc_provider_and_client_info(
             )
             if client_id:
                 _log.info("Using client_id {c!r} from config (provider {p!r})".format(c=client_id, p=provider_id))
-        if client_id is None and default_client_grant_types:
+        if client_id is None and default_client_grant_check:
             # Try "default_client" from backend's provider info.
             _log.debug("No client_id given: checking default client in backend's provider info")
-            client_id = provider.get_default_client_id(grant_types=default_client_grant_types)
+            client_id = provider.get_default_client_id(grant_check=default_client_grant_check)
             if client_id:
                 _log.info("Using default client_id {c!r} from OIDC provider {p!r} info.".format(
                     c=client_id, p=provider_id
@@ -414,7 +414,7 @@ def authenticate_oidc_authorization_code(
         """
         provider_id, client_info = self._get_oidc_provider_and_client_info(
             provider_id=provider_id, client_id=client_id, client_secret=client_secret,
-            default_client_grant_types=[DefaultOidcClientGrant.AUTH_CODE_PKCE],
+            default_client_grant_check=[DefaultOidcClientGrant.AUTH_CODE_PKCE],
         )
         authenticator = OidcAuthCodePkceAuthenticator(
             client_info=client_info,
@@ -466,7 +466,7 @@ def authenticate_oidc_refresh_token(
         """
         provider_id, client_info = self._get_oidc_provider_and_client_info(
             provider_id=provider_id, client_id=client_id, client_secret=client_secret,
-            default_client_grant_types=[DefaultOidcClientGrant.REFRESH_TOKEN],
+            default_client_grant_check=[DefaultOidcClientGrant.REFRESH_TOKEN],
         )
 
         if refresh_token is None:
@@ -495,9 +495,10 @@ def authenticate_oidc_device(
 
         .. versionchanged:: 0.5.1 Add :py:obj:`use_pkce` argument
         """
+        _g = DefaultOidcClientGrant  # alias for compactness
         provider_id, client_info = self._get_oidc_provider_and_client_info(
             provider_id=provider_id, client_id=client_id, client_secret=client_secret,
-            default_client_grant_types=[DefaultOidcClientGrant.DEVICE_CODE_PKCE],
+            default_client_grant_check=(lambda grants: _g.DEVICE_CODE in grants or _g.DEVICE_CODE_PKCE in grants),
         )
         authenticator = OidcDeviceAuthenticator(client_info=client_info, use_pkce=use_pkce, **kwargs)
         return self._authenticate_oidc(authenticator, provider_id=provider_id, store_refresh_token=store_refresh_token)
@@ -506,16 +507,20 @@ def authenticate_oidc(
             self,
             provider_id: str = None,
             client_id: Union[str, None] = None, client_secret: Union[str, None] = None,
-            store_refresh_token: bool = True
+            store_refresh_token: bool = True,
+            use_pkce: Union[bool, None] = None,
     ):
         """
         Do OpenID Connect authentication, first trying refresh tokens and falling back on device code flow.
 
         .. versionadded:: 0.6.0
         """
+        _g = DefaultOidcClientGrant  # alias for compactness
         provider_id, client_info = self._get_oidc_provider_and_client_info(
             provider_id=provider_id, client_id=client_id, client_secret=client_secret,
-            default_client_grant_types=[DefaultOidcClientGrant.DEVICE_CODE_PKCE, DefaultOidcClientGrant.REFRESH_TOKEN]
+            default_client_grant_check=lambda grants: (
+                    _g.REFRESH_TOKEN in grants and (_g.DEVICE_CODE in grants or _g.DEVICE_CODE_PKCE in grants)
+            )
         )
 
         # Try refresh token first.
@@ -539,7 +544,7 @@ def authenticate_oidc(
         # Fall back on device code flow
         # TODO: make it possible to do other fallback flows too?
         _log.info("Trying device code flow.")
-        authenticator = OidcDeviceAuthenticator(client_info=client_info)
+        authenticator = OidcDeviceAuthenticator(client_info=client_info, use_pkce=use_pkce)
         con = self._authenticate_oidc(authenticator, provider_id=provider_id, store_refresh_token=store_refresh_token)
         print("Authenticated using device code flow.")
         return con