Introduce AnonymousToken

Author: gmelikov

gcl_iam/drivers.py

@@ -142,9 +142,6 @@ def get_algorithm(
 
 
 class AnonDriver(AbstractAuthDriver):
-    def __init__(self):
-        self.algorithm_keys: tp.Dict[str, AlgorithmKeys] = {}
-
     def get_introspection_info(self, token_info, otp_code=None):
         return {
             "user_info": {
@@ -164,28 +161,7 @@ def get_algorithm(
         self,
         token_info: tokens.UnverifiedToken,
     ) -> tp.Optional[algorithms.AbstractAlgorithm]:
-        # For anonymous users, we don't need algorithm validation
-        if token_info is None:
-            return None
-
-        audience = token_info.audience_name
-        if audience not in self.algorithm_keys:
-            raise KeyError(f"Unknown audience: {audience}")
-        keys = self.algorithm_keys[audience]
-
-        if isinstance(keys, HS256AlgorithmKeys):
-            return algorithms.HS256(
-                key=keys.key,
-                previous_key=keys.previous_key,
-            )
-
-        if isinstance(keys, RS256AlgorithmKeys):
-            return algorithms.RS256VerifyOnly(
-                public_key=keys.public_key,
-                previous_public_key=keys.previous_public_key,
-            )
-
-        raise TypeError(f"Unexpected algorithm keys type: {type(keys)!r}")
+        raise NotImplementedError("AnonDriver does not support token validation.")
 
 
 class HttpDriver(AbstractAuthDriver):

gcl_iam/engines.py

@@ -45,6 +45,10 @@ def last_name(self):
     def email(self):
         return self._info["email"]
 
+    @property
+    def type(self):
+        return self._info["type"]
+
 
 class IntrospectionInfo:
     def __init__(self, info):
@@ -75,9 +79,9 @@ def __init__(self, auth_token, algorithm, driver, enforcer=None, otp_code=None):
 
         # Handle anonymous users (no auth token)
         if auth_token == "" and algorithm is None:
-            self._token_info = None
+            self._token_info = tokens.AnonymousToken()
             self._introspection_info = self._driver.get_introspection_info(
-                token_info=None,
+                token_info=self._token_info,
                 otp_code=otp_code,
             )
         else:
@@ -101,10 +105,7 @@ def __init__(self, auth_token, algorithm, driver, enforcer=None, otp_code=None):
             self._introspection_info["permissions"]
         )
 
-        if self._token_info:
-            self._introspection_info["otp_enabled"] = self._token_info.otp_enabled
-        else:
-            self._introspection_info["otp_enabled"] = False
+        self._introspection_info["otp_enabled"] = self._token_info.otp_enabled
 
     @property
     def token_info(self):

gcl_iam/middlewares.py

@@ -68,6 +68,7 @@ def __init__(
         )
         self._iam_engine_driver = iam_engine_driver
         self._skip_auth_endpoints = skip_auth_endpoints or []
+        self._anon_driver = drivers.AnonDriver()
 
     def _construct_context(self, req):
         return self._context_class(req=req, **self._context_kwargs)
@@ -81,7 +82,7 @@ def _should_skip_auth(self, req):
     def _get_auth_token(self, req):
         header_value = req.headers.get("Authorization", "")
         if header_value.lower().startswith("bearer "):
-            return header_value.split(" ")[1]
+            return header_value.split(" ", 1)[1]
         return None
 
     def _get_otp_code(self, req):
@@ -101,12 +102,10 @@ def _get_response(self, ctx, req):
                 auth_token = self._get_auth_token(req)
                 if auth_token is None:
                     # Create IamEngine with anonymous user data using AnonDriver
-
-                    anon_driver = drivers.AnonDriver()
                     iam_context = engines.IamEngine(
                         auth_token="",
                         algorithm=None,
-                        driver=anon_driver,
+                        driver=self._anon_driver,
                         otp_code=None,
                     )
                 else:

gcl_iam/tokens.py

@@ -15,6 +15,7 @@
 #    under the License.
 
 import datetime
+import time
 import typing as tp
 import uuid
 
@@ -174,3 +175,57 @@ def __init__(
             ignore_expiration=ignore_expiration,
             verify=verify,
         )
+
+
+class AnonymousToken(BaseToken):
+    """Token for anonymous users with mock JWT data"""
+
+    def __init__(self):
+        # Mock JWT token info for anonymous user
+        token_info = {
+            "exp": int(time.time()) + 360000,  # Far future expiration
+            "iat": int(time.time()),
+            "auth_time": int(time.time()),
+            "jti": "11111111-1111-1111-1111-111111111111",  # Fixed UUID for anon
+            "iss": "genesis-core-iam",
+            "aud": "anon",
+            "sub": "11111111-1111-1111-1111-111111111111",
+            "typ": "anon",
+        }
+
+        super().__init__(
+            token="",  # No actual token for anonymous users
+            token_info=token_info,
+            audience="anon",
+        )
+
+    @property
+    def expiration_datetime(self):
+        exp = self._token_info["exp"]
+        return datetime.datetime.fromtimestamp(exp, tz=datetime.timezone.utc)
+
+    @property
+    def created_at(self):
+        iat = self._token_info["iat"]
+        return datetime.datetime.fromtimestamp(iat, tz=datetime.timezone.utc)
+
+    @property
+    def issuer_url(self):
+        return self._token_info["iss"]
+
+    @property
+    def user_uuid(self):
+        return uuid.UUID(self._token_info["sub"])
+
+    @property
+    def autenticated_at(self):
+        auth_time = self._token_info["auth_time"]
+        return datetime.datetime.fromtimestamp(auth_time, tz=datetime.timezone.utc)
+
+    @property
+    def token_type(self):
+        return self._token_info["typ"]
+
+    @property
+    def otp_enabled(self):
+        return False