fix: adjust internal backend auth and add cache token

retail/internal/authenticators.py

@@ -1,8 +1,8 @@
 from mozilla_django_oidc.contrib.drf import OIDCAuthentication
 
-from retail.internal.backends import InternalOIDCAuthenticationBackend
+from retail.internal.backends import WeniOIDCAuthenticationBackend
 
 
 class InternalOIDCAuthentication(OIDCAuthentication):
     def __init__(self, backend=None):
-        super().__init__(backend or InternalOIDCAuthenticationBackend())
+        super().__init__(backend or WeniOIDCAuthenticationBackend())

retail/internal/backends.py

@@ -1,39 +1,69 @@
+import json
+
 from mozilla_django_oidc.auth import OIDCAuthenticationBackend
 from django.contrib.auth.models import Permission
 from django.contrib.contenttypes.models import ContentType
 from django.contrib.auth import get_user_model
+from django_redis import get_redis_connection
 
+from django.conf import settings
 
 User = get_user_model()
 
 
-class InternalOIDCAuthenticationBackend(OIDCAuthenticationBackend):
+class WeniOIDCAuthenticationBackend(OIDCAuthenticationBackend):  # pragma: no cover
+    cache_token = settings.OIDC_CACHE_TOKEN
+    cache_ttl = settings.OIDC_CACHE_TTL
+
+    def get_userinfo(self, access_token, *args):
+        if not self.cache_token:
+            return super().get_userinfo(access_token, *args)
+
+        redis_connection = get_redis_connection()
+        userinfo = redis_connection.get(access_token)
+
+        if userinfo is not None:
+            return json.loads(userinfo)
+
+        userinfo = super().get_userinfo(access_token, *args)
+        redis_connection.set(access_token, json.dumps(userinfo), self.cache_ttl)
+
+        return userinfo
+
     def check_module_permission(self, claims, user) -> None:
         if claims.get("can_communicate_internally", False):
-            content_type = ContentType.objects.get_for_model(self.UserModel)
-            permission, _ = Permission.objects.get_or_create(
+            content_type = ContentType.objects.get_for_model(User)
+            permission, created = Permission.objects.get_or_create(
                 codename="can_communicate_internally",
                 name="can communicate internally",
                 content_type=content_type,
             )
-            if not user.has_perm("auth.can_communicate_internally"):
+            if not user.has_perm("authentication.can_communicate_internally"):
                 user.user_permissions.add(permission)
 
-    def get_username(self, claims):
-        username = claims.get("email")
-        if username:
-            return username
-        return super().get_username(claims=claims)
+    def filter_users_by_claims(self, claims):
+        """Return all users matching the specified email."""
+        email = claims.get("email")
+        if not email:
+            return self.UserModel.objects.none()
+        return self.UserModel.objects.filter(email__iexact=email).exclude(
+            first_name="", last_name=""
+        )
 
     def create_user(self, claims):
         email = claims.get("email")
-        username = self.get_username(claims)
-
-        user, _ = self.UserModel.objects.get_or_create(email=email, username=username)
 
+        user, _ = self.UserModel.objects.get_or_create(email=email)
         user.first_name = claims.get("given_name", "")
         user.last_name = claims.get("family_name", "")
+        user.save()
+
+        self.check_module_permission(claims, user)
+
+        return user
 
+    def update_user(self, user, claims):
+        user.email = claims.get("email", "")
         user.save()
 
         self.check_module_permission(claims, user)

retail/internal/permissions.py

@@ -1,8 +1,16 @@
 from rest_framework import permissions
 
 
-class CanCommunicateInternally(permissions.BasePermission):
+class CanCommunicateInternally(permissions.IsAuthenticated):
     def has_permission(self, request, view):
-        return request.user.user_permissions.filter(
+        # Check if the user is authenticated
+        user = request.user
+        if not user.is_authenticated:
+            return False
+
+        # Check if the user has 'can_communicate_internally' permission
+        has_internal_permission = user.user_permissions.filter(
             codename="can_communicate_internally"
         ).exists()
+
+        return has_internal_permission

retail/internal/views.py

@@ -8,7 +8,6 @@
 
 class InternalGenericViewSet(GenericViewSet):
     authentication_classes = [InternalOIDCAuthentication]
-    # permission_classes = [IsAuthenticated, CanCommunicateInternally]
     permission_classes = [IsAuthenticated]
     renderer_classes = [JSONRenderer]
     throttle_classes = []

retail/settings.py

@@ -184,11 +184,19 @@
     OIDC_OP_USER_ENDPOINT = env.str("OIDC_OP_USER_ENDPOINT")
     OIDC_OP_JWKS_ENDPOINT = env.str("OIDC_OP_JWKS_ENDPOINT")
     OIDC_RP_SIGN_ALGO = env.str("OIDC_RP_SIGN_ALGO", default="RS256")
-    OIDC_DRF_AUTH_BACKEND = "retail.internal.backends.InternalOIDCAuthenticationBackend"
+    OIDC_DRF_AUTH_BACKEND = "retail.internal.backends.WeniOIDCAuthenticationBackend"
     OIDC_RP_SCOPES = env.str("OIDC_RP_SCOPES", default="openid email")
 
 INTEGRATIONS_REST_ENDPOINT = env.str("INTEGRATIONS_REST_ENDPOINT")
 
 FLOWS_REST_ENDPOINT = env.str("FLOWS_REST_ENDPOINT")
 
 EMAILS_CAN_TESTING = env.str("EMAILS_CAN_TESTING", "").split(",")
+
+
+OIDC_CACHE_TOKEN = env.bool(
+    "OIDC_CACHE_TOKEN", default=False
+)  # Enable/disable user token caching (default: False).
+OIDC_CACHE_TTL = env.int(
+    "OIDC_CACHE_TTL", default=600
+)  # Time-to-live for cached user tokens (default: 600 seconds).