refactor(api): extracted common code in utils.py

Author: klajdicaushi

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "cardo-python-utils"
-version = "0.5.dev10"
+version = "0.5.dev11"
 description = "Python library enhanced with a wide range of functions for different scenarios."
 readme = "README.rst"
 requires-python = ">=3.8"

python_utils/django/keycloak/api/drf.py

@@ -1,31 +1,19 @@
-import jwt
-
 from django.conf import settings
-from django.contrib.auth import get_user_model
-from jwt import PyJWKClient
 from jwt.exceptions import InvalidTokenError
 
 from rest_framework import authentication
 from rest_framework.exceptions import AuthenticationFailed
 from rest_framework.permissions import BasePermission
 
-
-jwks_client = PyJWKClient(getattr(settings, "JWKS_URL", ""))
+from .utils import create_or_update_user, decode_jwt
 
 
 class AuthenticationBackend(authentication.TokenAuthentication):
     keyword = "Bearer"
 
     def authenticate_credentials(self, token: str):
-        signing_key = jwks_client.get_signing_key_from_jwt(token)
-
         try:
-            payload = jwt.decode(
-                token,
-                signing_key.key,
-                algorithms=["RS256"],
-                audience=getattr(settings, "JWT_AUDIENCE", None),
-            )
+            payload = decode_jwt(token)
         except InvalidTokenError as e:
             raise AuthenticationFailed(f"Invalid token: {str(e)}") from e
 
@@ -36,43 +24,9 @@ def authenticate_credentials(self, token: str):
                 "Invalid token: preferred_username not present."
             ) from e
 
-        user = self._get_user(username, payload)
+        user = create_or_update_user(username, payload)
         return user, payload
 
-    def _get_user(self, username: str, payload: dict):
-        """
-        Get or create a user based on the JWT payload.
-        If the user exists, update their details.
-        """
-        user_model = get_user_model()
-        user_data = {
-            "first_name": payload.get("given_name") or "",
-            "last_name": payload.get("family_name") or "",
-            "email": payload.get("email") or "",
-            "is_staff": payload.get("is_staff", False),
-        }
-        if hasattr(user_model, "is_demo"):
-            user_data["is_demo"] = payload.get("is_demo", False)
-
-        user = user_model.objects.filter(username=username).first()
-        if user:
-            update_needed = False
-
-            for field, value in user_data.items():
-                if getattr(user, field) != value:
-                    setattr(user, field, value)
-                    update_needed = True
-
-            if update_needed:
-                user.save(update_fields=list(user_data.keys()))
-
-            return user
-        else:
-            return user_model.objects.create(
-                username=username,
-                **user_data,
-            )
-
 
 class HasScope(BasePermission):
     """

python_utils/django/keycloak/api/ninja.py

@@ -1,27 +1,18 @@
 from typing import Literal
 
-from jwt import PyJWKClient, decode as jwt_decode
 from jwt.exceptions import InvalidTokenError
 
 from django.conf import settings
-from django.contrib.auth import get_user_model
 from ninja.security import HttpBearer
 from ninja.errors import AuthenticationError, HttpError
 
-jwks_client = PyJWKClient(getattr(settings, "JWKS_URL", ""))
+from .utils import create_or_update_user, decode_jwt
 
 
 class AuthBearer(HttpBearer):
     def authenticate(self, request, token):
-        signing_key = jwks_client.get_signing_key_from_jwt(token)
-
         try:
-            payload = jwt_decode(
-                token,
-                signing_key.key,
-                algorithms=["RS256"],
-                audience=getattr(settings, "JWT_AUDIENCE", None),
-            )
+            payload = decode_jwt(token)
         except InvalidTokenError as e:
             raise AuthenticationError(f"Invalid token: {str(e)}") from e
 
@@ -32,46 +23,12 @@ def authenticate(self, request, token):
                 "Invalid token: preferred_username not present."
             ) from e
 
-        user = self._get_user(username, payload)
+        user = create_or_update_user(username, payload)
 
         self._verify_scopes(request, payload)
 
         return user
 
-    def _get_user(self, username: str, payload: dict):
-        """
-        Get or create a user based on the JWT payload.
-        If the user exists, update their details.
-        """
-        user_model = get_user_model()
-        user_data = {
-            "first_name": payload.get("given_name") or "",
-            "last_name": payload.get("family_name") or "",
-            "email": payload.get("email") or "",
-            "is_staff": payload.get("is_staff", False),
-        }
-        if hasattr(user_model, "is_demo"):
-            user_data["is_demo"] = payload.get("is_demo", False)
-
-        user = user_model.objects.filter(username=username).first()
-        if user:
-            update_needed = False
-
-            for field, value in user_data.items():
-                if getattr(user, field) != value:
-                    setattr(user, field, value)
-                    update_needed = True
-
-            if update_needed:
-                user.save(update_fields=list(user_data.keys()))
-
-            return user
-        else:
-            return user_model.objects.create(
-                username=username,
-                **user_data,
-            )
-
     def _verify_scopes(self, request, token_payload):
         allowed_scopes = self._get_view_allowed_scopes(request)
 

python_utils/django/keycloak/api/utils.py

@@ -0,0 +1,77 @@
+from typing import TypedDict
+
+from django.conf import settings
+from django.contrib.auth import get_user_model
+from jwt import decode, PyJWKClient
+
+jwks_client = PyJWKClient(getattr(settings, "JWKS_URL", ""))
+
+
+class TokenPayload(TypedDict, total=False):
+    exp: int
+    iat: int
+    jti: str
+    iss: str
+    aud: str | list[str]
+    typ: str
+    azp: str
+    sid: str
+    scope: str
+    preferred_username: str
+    given_name: str
+    family_name: str
+    email: str
+    is_staff: bool
+    is_demo: bool
+    groups: list[str]  # Full path of the user group, e.g. "/group1/subgroup1"
+
+
+def decode_jwt(token: str) -> TokenPayload:
+    """
+    Decode a JWT token using the public certificate of the Auth Server.
+
+    Raises:
+        jwt.exceptions.InvalidTokenError: If the token is invalid or cannot be decoded.
+    """
+    signing_key = jwks_client.get_signing_key_from_jwt(token)
+
+    return decode(
+        token,
+        signing_key.key,
+        algorithms=["RS256"],
+        audience=getattr(settings, "JWT_AUDIENCE", None),
+    )
+
+
+def create_or_update_user(username: str, payload: TokenPayload):
+    """
+    Create or update a user based on the JWT payload.
+    """
+    user_model = get_user_model()
+    user_data = {
+        "first_name": payload.get("given_name") or "",
+        "last_name": payload.get("family_name") or "",
+        "email": payload.get("email") or "",
+        "is_staff": payload.get("is_staff", False),
+    }
+    if hasattr(user_model, "is_demo"):
+        user_data["is_demo"] = payload.get("is_demo", False)
+
+    user = user_model.objects.filter(username=username).first()
+    if user:
+        update_needed = False
+
+        for field, value in user_data.items():
+            if getattr(user, field) != value:
+                setattr(user, field, value)
+                update_needed = True
+
+        if update_needed:
+            user.save(update_fields=list(user_data.keys()))
+
+        return user
+    else:
+        return user_model.objects.create(
+            username=username,
+            **user_data,
+        )