Add separate methods for token_info to avoid breaking middleware

Author: Deepthi-Chand

api/utils/keycloak_utils.py

@@ -102,9 +102,15 @@ def validate_token(self, token: str) -> Dict[str, Any]:
             logger.error(f"Error validating token: {e}")
             return {}
 
-    def get_user_roles(self, token_info: dict) -> list[str]:
+    def get_user_roles_from_token_info(self, token_info: dict) -> list[str]:
         """
-        Extract roles from a Keycloak token.
+        Extract roles from token introspection data.
+
+        Args:
+            token_info: Token introspection response
+
+        Returns:
+            List of role names
         """
         roles: list[str] = []
 
@@ -122,19 +128,85 @@ def get_user_roles(self, token_info: dict) -> list[str]:
 
         return roles
 
-    def get_user_organizations(self, token_info: dict) -> List[Dict[str, Any]]:
+    def get_user_organizations_from_token_info(self, token_info: dict) -> List[Dict[str, Any]]:
         """
-        Get the organizations a user belongs to from their token info.
+        Get organizations from token introspection data.
+
+        Args:
+            token_info: Token introspection response
+
+        Returns:
+            List of organization information
+        """
+        try:
+            # Get organization info from resource_access or attributes
+            resource_access = token_info.get("resource_access", {})
+            client_roles = resource_access.get(self.client_id, {}).get("roles", [])
+
+            # Extract organization info from roles
+            organizations = []
+            for role in client_roles:
+                if role.startswith("org_"):
+                    parts = role.split("_")
+                    if len(parts) >= 3:
+                        org_id = parts[1]
+                        role_name = parts[2]
+                        organizations.append({"organization_id": org_id, "role": role_name})
+
+            return organizations
+        except Exception as e:
+            logger.error(f"Error getting user organizations: {e}")
+            return []
+
+    def get_user_roles(self, token: str) -> list[str]:
+        """
+        Extract roles from a Keycloak token.
+
+        Args:
+            token: The user's token
+
+        Returns:
+            List of role names
+        """
+        try:
+            # Decode the token to get user info
+            token_info = self.keycloak_openid.decode_token(token)
+
+            roles: list[str] = []
+
+            # Extract realm roles
+            realm_access = token_info.get("realm_access", {})
+            if realm_access and "roles" in realm_access:
+                roles.extend(realm_access["roles"])  # type: ignore[no-any-return]
+
+            # Extract client roles
+            resource_access = token_info.get("resource_access", {})
+            client_id = settings.KEYCLOAK_CLIENT_ID
+            if resource_access and client_id in resource_access:
+                client_roles = resource_access[client_id].get("roles", [])
+                roles.extend(client_roles)
+
+            return roles
+        except Exception as e:
+            logger.error(f"Error getting user roles: {e}")
+            return []
+
+    def get_user_organizations(self, token: str) -> List[Dict[str, Any]]:
+        """
+        Get the organizations a user belongs to from their token.
         This assumes that organization information is stored in the token
         as client roles or in user attributes.
 
         Args:
-            token_info: The decoded token information
+            token: The user's token
 
         Returns:
             List of organization information
         """
         try:
+            # Decode the token to get user info
+            token_info = self.keycloak_openid.decode_token(token)
+
             # Get organization info from resource_access or attributes
             # This implementation depends on how organizations are represented in Keycloak
             # This is a simplified example - adjust based on your Keycloak configuration

api/views/auth.py

@@ -35,9 +35,9 @@ def post(self, request: Request) -> Response:
         # Get token introspection data for roles and organizations
         token_info = keycloak_manager.keycloak_openid.introspect(keycloak_token)
 
-        # Get user roles and organizations from the token
-        roles = keycloak_manager.get_user_roles(token_info)
-        organizations = keycloak_manager.get_user_organizations(token_info)
+        # Get user roles and organizations from the token introspection data
+        roles = keycloak_manager.get_user_roles_from_token_info(token_info)
+        organizations = keycloak_manager.get_user_organizations_from_token_info(token_info)
 
         # Sync the user information with our database
         user = keycloak_manager.sync_user_from_keycloak(user_info, roles, organizations)

authorization/middleware_utils.py

@@ -50,9 +50,7 @@ def get_user_from_keycloak_token(request: HttpRequest) -> User:
                 # Use the raw header value, but check if it might be a raw token
                 # (no 'Bearer ' prefix but still a valid JWT format)
                 token = auth_header
-                logger.debug(
-                    f"Using raw Authorization header as token, length: {len(token)}"
-                )
+                logger.debug(f"Using raw Authorization header as token, length: {len(token)}")
 
         # If no token found, return anonymous user
         if not token:
@@ -91,14 +89,10 @@ def get_user_from_keycloak_token(request: HttpRequest) -> User:
             return cast(User, AnonymousUser())
 
         # Log the user info for debugging
-        logger.debug(
-            f"User info from token: {user_info.keys() if user_info else 'None'}"
-        )
+        logger.debug(f"User info from token: {user_info.keys() if user_info else 'None'}")
         logger.debug(f"User sub: {user_info.get('sub', 'None')}")
         logger.debug(f"User email: {user_info.get('email', 'None')}")
-        logger.debug(
-            f"User preferred_username: {user_info.get('preferred_username', 'None')}"
-        )
+        logger.debug(f"User preferred_username: {user_info.get('preferred_username', 'None')}")
 
         # Get user roles and organizations from the token
         roles = keycloak_manager.get_user_roles(token)
@@ -113,9 +107,7 @@ def get_user_from_keycloak_token(request: HttpRequest) -> User:
             logger.warning("User synchronization failed, returning anonymous user")
             return cast(User, AnonymousUser())
 
-        logger.debug(
-            f"Successfully authenticated user: {user.username} (ID: {user.id})"
-        )
+        logger.debug(f"Successfully authenticated user: {user.username} (ID: {user.id})")
 
         # Return the authenticated user
         logger.debug(f"Returning authenticated user: {user.username}")