Add origin of token to BearerAuth class

Author: m-mohr

openeo/rest/auth/auth.py

@@ -29,8 +29,9 @@ class BearerAuth(OpenEoApiAuthBase):
     https://open-eo.github.io/openeo-api/apireference/#section/Authentication/Bearer
     """
 
-    def __init__(self, bearer: str):
+    def __init__(self, bearer: str, origin: str):
         self.bearer = bearer
+        self.origin = origin
 
     def __call__(self, req: Request) -> Request:
         # Add bearer authorization header.
@@ -42,12 +43,11 @@ class BasicBearerAuth(BearerAuth):
     """Bearer token for Basic Auth (openEO API 1.0.0 style)"""
 
     def __init__(self, access_token: str):
-        super().__init__(bearer="basic//{t}".format(t=access_token))
+        super().__init__(bearer="basic//{t}".format(t=access_token), origin="basic")
 
 
 class OidcBearerAuth(BearerAuth):
     """Bearer token for OIDC Auth (openEO API 1.0.0 style)"""
 
     def __init__(self, provider_id: str, access_token: str):
-        super().__init__(bearer="oidc/{p}/{t}".format(p=provider_id, t=access_token))
-
+        super().__init__(bearer="oidc/{p}/{t}".format(p=provider_id, t=access_token), origin="oidc")

openeo/rest/auth/testing.py

@@ -281,7 +281,7 @@ def validate_access_token(self, access_token: str):
         raise LookupError("Invalid access token")
 
     def invalidate_access_token(self):
-        self.state["access_token"] = "***invalidated***"
+        self.state["access_token"] = None
 
     def get_request_history(
         self, url: Optional[str] = None, method: Optional[str] = None

openeo/rest/connection.py

@@ -283,7 +283,7 @@ def authenticate_basic(self, username: Optional[str] = None, password: Optional[
 
         # Switch to bearer based authentication in further requests.
         if jwt_conformance:
-            self.auth = BearerAuth(bearer=resp["access_token"])
+            self.auth = BearerAuth(bearer=resp["access_token"], origin="basic")
         else:
             self.auth = BasicBearerAuth(access_token=resp["access_token"])
         return self
@@ -426,7 +426,7 @@ def _authenticate_oidc(
         # check for JWT bearer token conformance
         jwt_conformance = self.capabilities().has_conformance(CONFORMANCE_JWT_BEARER)
         if jwt_conformance:
-            self.auth = BearerAuth(bearer=token)
+            self.auth = BearerAuth(bearer=token, origin="oidc")
         else:
             self.auth = OidcBearerAuth(provider_id=provider_id, access_token=token)
         self._oidc_auth_renewer = oidc_auth_renewer
@@ -738,7 +738,7 @@ def authenticate_bearer_token(self, bearer_token: str) -> Connection:
 
         .. versionadded:: 0.38.0
         """
-        self.auth = BearerAuth(bearer=bearer_token)
+        self.auth = BearerAuth(bearer=bearer_token, origin="unknown")
         self._oidc_auth_renewer = None
         return self
 
@@ -748,7 +748,7 @@ def try_access_token_refresh(self, *, reason: Optional[str] = None) -> bool:
         Returns whether a new access token was obtained.
         """
         reason = f" Reason: {reason}" if reason else ""
-        if isinstance(self.auth, OidcBearerAuth) and self._oidc_auth_renewer:
+        if isinstance(self.auth, BearerAuth) and self.auth.origin == "oidc" and self._oidc_auth_renewer:
             try:
                 self._authenticate_oidc(
                     authenticator=self._oidc_auth_renewer,

tests/extra/artifacts/_s3sts/test_s3sts.py

@@ -93,7 +93,7 @@ def conn_with_s3sts_capabilities(
     requests_mock.get(API_URL, json={"api_version": "1.0.0", **extra_api_capabilities})
     requests_mock.get(f"{API_URL}me", json={})
     conn = Connection(API_URL)
-    conn.auth = BearerAuth("oidc/fake/token")
+    conn.auth = BearerAuth("oidc/fake/token", origin="oidc")
     yield conn
 
 

tests/rest/test_connection.py

@@ -276,7 +276,7 @@ def debug(request: requests.Request, context):
 
     requests_mock.get(url, text=debug)
 
-    con = RestApiConnection(api_root, auth=BearerAuth(secret))
+    con = RestApiConnection(api_root, auth=BearerAuth(secret, origin="unknown"))
     res = con.get(url)
     assert "hello world" in res.text
     assert "User-Agent': 'openeo-python-client/" in res.text
@@ -2379,6 +2379,7 @@ def test_authenticate_eoidc_auto_renew_expired_access_token_initial_refresh_toke
         "_used_oidc_provider": "oi",
         "_used_access_token": access_token2,
     }
+    assert False