Issue #530 Initial support for authenticate_oidc() without refresh_…

…token support
Open-EO · Jan 19, 2024 · 2e3499f · 2e3499f
1 parent 7ce9292
commit 2e3499f
Show file tree

Hide file tree

Showing 3 changed files with 64 additions and 5 deletions.
diff --git a/openeo/rest/auth/testing.py b/openeo/rest/auth/testing.py
@@ -110,7 +110,7 @@ def token_callback(
         grant_type = params["grant_type"]
         self.grant_request_history.append({"grant_type": grant_type})
         if self.expected_grant_type:
-            assert grant_type == self.expected_grant_type
+            assert grant_type == self.expected_grant_type, f"{grant_type} != {self.expected_grant_type}"
         callback = {
             "authorization_code": self.token_callback_authorization_code,
             "client_credentials": self.token_callback_client_credentials,

diff --git a/openeo/rest/connection.py b/openeo/rest/connection.py
@@ -705,11 +705,19 @@ def authenticate_oidc(
             raise ValueError(f"Unhandled auth method {auth_method}")
 
         _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_check=lambda grants: (
-                    _g.REFRESH_TOKEN in grants and (_g.DEVICE_CODE in grants or _g.DEVICE_CODE_PKCE in grants)
+        # TODO: need for dedicated `use_refresh_token` option instead of `store_refresh_token` here,
+        #       for better distinction of intention?
+        if store_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)
             )
+        else:
+            default_client_grant_check = lambda grants: (_g.DEVICE_CODE in grants or _g.DEVICE_CODE_PKCE in grants)
+        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_check=default_client_grant_check,
         )
 
         # Try refresh token first.

diff --git a/tests/rest/test_connection.py b/tests/rest/test_connection.py
@@ -2350,6 +2350,57 @@ 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
 
 
+def test_authenticate_oidc_default_client_handling_and_refresh_token_support(
+    requests_mock, refresh_token_store, oidc_device_code_flow_checker
+):
+    """
+    https://github.com/Open-EO/openeo-python-client/issues/530
+    """
+    requests_mock.get(API_URL, json={"api_version": "1.0.0"})
+    issuer = "https://oidc.test"
+    requests_mock.get(
+        API_URL + "credentials/oidc",
+        json={
+            "providers": [
+                {
+                    "id": "oi",
+                    "issuer": issuer,
+                    "title": "example",
+                    "scopes": ["openid"],
+                    "default_clients": [
+                        # TODO: parameterize with/without PKCE
+                        {"id": "client_123", "grant_types": ["urn:ietf:params:oauth:grant-type:device_code+pkce"]}
+                    ],
+                }
+            ]
+        },
+    )
+    oidc_mock = OidcMock(
+        requests_mock=requests_mock,
+        expected_client_id="client_123",
+        expected_grant_type="urn:ietf:params:oauth:grant-type:device_code",
+        oidc_issuer=issuer,
+        expected_fields={
+            "scope": "openid",
+            "code_verifier": True,
+            "code_challenge": True,
+        },
+    )
+
+    # With all this set up, kick off the openid connect flow
+    conn = Connection(API_URL, refresh_token_store=refresh_token_store)
+    assert isinstance(conn.auth, NullAuth)
+    oidc_mock.state["device_code_callback_timeline"] = ["great success"]
+    with oidc_device_code_flow_checker():
+        # TODO: parameterize store_refresh_token
+        conn.authenticate_oidc(store_refresh_token=False)
+    assert isinstance(conn.auth, BearerAuth)
+    assert conn.auth.bearer == "oidc/oi/" + oidc_mock.state["access_token"]
+    assert [r["grant_type"] for r in oidc_mock.grant_request_history] == [
+        "urn:ietf:params:oauth:grant-type:device_code"
+    ]
+
+
 def test_load_collection_arguments_100(requests_mock):
     requests_mock.get(API_URL, json={"api_version": "1.0.0"})
     conn = Connection(API_URL)