CDD 3174 - Downloads v2 endpoint

cms/auth_content/constants.py

@@ -3,7 +3,6 @@
     get_all_theme_names_and_ids,
 )
 
-WILDCARD_ID_VALUE = "-1"
 PERMISSION_SET_FIELDS = [
     {
         "field_name": "theme",

cms/auth_content/models/permission_sets.py

@@ -6,7 +6,7 @@
 from wagtail.admin.panels import FieldPanel, mark_safe
 
 from cms.auth_content.auth_utils import _create_form_field
-from cms.auth_content.constants import PERMISSION_SET_FIELDS, WILDCARD_ID_VALUE
+from cms.auth_content.constants import PERMISSION_SET_FIELDS
 from cms.dynamic_content import help_texts
 from cms.metrics_interface.field_choices_callables import (
     get_all_geography_names_and_codes,
@@ -16,6 +16,7 @@
     get_all_theme_names_and_ids,
     get_all_topic_names_and_ids,
 )
+from common.auth.permissions import WILDCARD_ID_VALUE
 
 
 class PermissionSetForm(WagtailAdminPageForm):

cms/dashboard/viewsets.py

@@ -1,4 +1,3 @@
-import logging
 from itertools import chain
 
 from django.urls import path
@@ -10,43 +9,14 @@
 
 from caching.private_api.decorators import cache_response
 from cms.auth_content.auth_utils import is_auth_enabled
-from cms.auth_content.constants import WILDCARD_ID_VALUE
 from cms.dashboard.serializers import CMSDraftPagesSerializer, ListablePageSerializer
 from cms.metrics_documentation.models.child import MetricsDocumentationChildEntry
 from cms.topic.models import TopicPage
+from common.auth.permissions import check_page_permissions
 
-logger = logging.getLogger(__name__)
 AUTH_ENABLED = is_auth_enabled()
 
 
-def check_permissions(user_permissions, theme_id, sub_theme_id, topic_id) -> bool:
-    if not isinstance(user_permissions, list):
-        return False
-
-    for permission in user_permissions:
-        permission_theme_id = permission.get("theme", {}).get("id")
-        permission_sub_theme_id = permission.get("sub_theme", {}).get("id")
-        permission_topic_id = permission.get("topic", {}).get("id")
-
-        if permission_theme_id == WILDCARD_ID_VALUE:
-            return True
-
-        if (
-            permission_theme_id == theme_id
-            and permission_sub_theme_id == WILDCARD_ID_VALUE
-        ):
-            return True
-
-        if (
-            permission_theme_id == theme_id
-            and permission_sub_theme_id == sub_theme_id
-            and (permission_topic_id in {WILDCARD_ID_VALUE, topic_id})
-        ):
-            return True
-
-    return False
-
-
 @extend_schema(tags=["cms"])
 class CMSPagesAPIViewSet(PagesAPIViewSet):
     # This is the /pages (or proxy/pages env dependent endpoint)
@@ -110,19 +80,12 @@ def get_queryset(self):
                 filtered_queryset = is_public_pages | pages_without_is_public
 
             else:
-                logger.info(
-                    "User %s has total permission sets: %s",
-                    req.user.username,
-                    req.user.permission_sets["summary"]["total_permission_sets"],
-                )
                 has_global_access = req.user.permission_sets["summary"][
                     "has_global_access"
                 ]
 
                 if has_global_access:
-                    logger.info("User %s has global access", req.user.username)
                     filtered_queryset = queryset
-
                 else:
                     user_permissions = req.user.permission_sets["permission_sets"]
                     pages_to_check = chain(
@@ -139,11 +102,11 @@ def get_queryset(self):
                         page_id
                         for page_id, page in pages_to_check
                         if page.is_public
-                        or check_permissions(
-                            user_permissions,
-                            page.theme,
-                            page.sub_theme,
-                            page.topic,
+                        or check_page_permissions(
+                            permission_sets=user_permissions,
+                            theme_id=page.theme,
+                            sub_theme_id=page.sub_theme,
+                            topic_id=page.topic,
                         )
                     ]
 

common/auth/cognito_jwt/backend.py

@@ -8,6 +8,8 @@
 from rest_framework import HTTP_HEADER_ENCODING, exceptions
 from rest_framework.authentication import BaseAuthentication
 
+from common.auth.logging import log_user_permissions
+
 from .validator import TokenError, TokenValidator
 
 logger = logging.getLogger(__name__)
@@ -50,16 +52,22 @@ def authenticate(self, request):
             raise exceptions.AuthenticationFailed from None
 
         custom_user_manager = self.get_custom_user_manager()
+
         if custom_user_manager:
             user = custom_user_manager.get_or_create_for_cognito(jwt_payload)
         else:
             user_model = self.get_user_model()
             user = user_model.objects.get_or_create_for_cognito(jwt_payload)
+
         if not user:
             logger.debug(
                 "Unable to create user from JWT, defaulting to unauthenticated"
             )
             return None
+
+        # Log user permissions for every JWT API call
+        log_user_permissions(user)
+
         return (user, jwt_token)
 
     @staticmethod

common/auth/logging.py

@@ -0,0 +1,43 @@
+"""Utilities for logging authentication and permission information across the API."""
+
+import logging
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+def log_user_permissions(user: Any) -> None:
+    """Log permission information for an authenticated user.
+
+    This function logs the permission set summary and global access status.
+    It expects ``user.permission_sets`` to be a dict with the shape produced
+    by ``CognitoManager.get_or_create_for_cognito``:
+
+    .. code-block:: python
+
+        {
+            "permission_sets": [...],
+            "summary": {"total_permission_sets": 2, "has_global_access": False},
+        }
+
+    Args:
+        user: The authenticated user object that has a ``permission_sets`` dict.
+    """
+
+    if not hasattr(user, "username"):
+        return
+    if not hasattr(user, "permission_sets"):
+        return
+
+    username = user.username
+    permission_sets = user.permission_sets
+
+    if not isinstance(permission_sets, dict):
+        return
+
+    log_msg = f'User {username} has total permission sets {permission_sets["summary"]["total_permission_sets"]}'
+
+    if permission_sets["summary"]["has_global_access"]:
+        log_msg += " and global access"
+
+    logger.info(log_msg)