Merge remote-tracking branch 'origin/main' into claude/security-code-review-C1C5t

Author: JSv4

CHANGELOG.md

@@ -7,13 +7,18 @@
 """
 
 import logging
+from typing import TYPE_CHECKING, Any, Optional
 
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.contrib.auth.backends import ModelBackend
+from django.http import HttpRequest
+
+if TYPE_CHECKING:
+    from opencontractserver.users.models import User
 
 logger = logging.getLogger(__name__)
-User = get_user_model()
+UserModel = get_user_model()
 
 
 class Auth0AdminBackend(ModelBackend):
@@ -27,32 +32,47 @@ class Auth0AdminBackend(ModelBackend):
     the frontend Auth0 login flow, then back to admin.
     """
 
-    def authenticate(self, request, auth0_user_id=None, **kwargs):
+    def authenticate(
+        self,
+        request: Optional[HttpRequest],
+        username: Optional[str] = None,
+        password: Optional[str] = None,
+        **kwargs: Any,
+    ) -> Optional["User"]:
         """
         Authenticate a user by their Auth0 user ID (sub claim).
 
         This is called after the user has authenticated via Auth0 on the
-        frontend and the token has been validated.
+        frontend and the token has been validated. ``username`` and
+        ``password`` are declared to match :class:`ModelBackend`'s
+        signature (Django calls every backend with those kwargs) but are
+        ignored here — credential-based login falls through to the
+        default backend. The Auth0-specific ``auth0_user_id`` is passed
+        via ``**kwargs``.
 
         Args:
             request: The HTTP request object.
-            auth0_user_id: The Auth0 user ID (sub claim from JWT).
-            **kwargs: Additional keyword arguments (ignored).
+            username: Accepted for :class:`ModelBackend` compatibility; ignored.
+            password: Accepted for :class:`ModelBackend` compatibility; ignored.
+            **kwargs: Additional credentials. The ``auth0_user_id`` key
+                (Auth0 JWT ``sub`` claim) drives the lookup; other keys
+                are ignored.
 
         Returns:
-            User: The authenticated user if valid and is_staff, None otherwise.
+            The authenticated user if valid and ``is_staff``, otherwise ``None``.
         """
         if not getattr(settings, "USE_AUTH0", False):
             return None
 
+        auth0_user_id: Optional[str] = kwargs.get("auth0_user_id")
         if not auth0_user_id:
             return None
 
         try:
-            user = User.objects.get(username=auth0_user_id)
+            user = UserModel.objects.get(username=auth0_user_id)
             if user.is_active and user.is_staff:
                 logger.info(
-                    "Auth0 admin authentication successful for user ID %s", user.id
+                    "Auth0 admin authentication successful for user ID %s", user.pk
                 )
                 return user
             else:
@@ -63,13 +83,13 @@ def authenticate(self, request, auth0_user_id=None, **kwargs):
                     user.is_staff,
                 )
                 return None
-        except User.DoesNotExist:
+        except UserModel.DoesNotExist:
             logger.warning("Auth0 admin auth failed: user %s not found", auth0_user_id)
             return None
 
-    def get_user(self, user_id):
+    def get_user(self, user_id: int) -> Optional["User"]:
         """Retrieve user by primary key."""
         try:
-            return User.objects.get(pk=user_id)
-        except User.DoesNotExist:
+            return UserModel.objects.get(pk=user_id)
+        except UserModel.DoesNotExist:
             return None

config/admin_auth/backends.py

@@ -7,12 +7,22 @@
 """
 
 import logging
+from typing import TYPE_CHECKING, Optional
 from urllib.parse import urlencode
 
 from django.conf import settings
 from django.contrib import admin, messages
 from django.contrib.auth import authenticate, login, logout
-from django.http import HttpResponse, HttpResponseNotAllowed, JsonResponse
+from django.http import (
+    HttpRequest,
+    HttpResponse,
+    HttpResponseBase,
+    HttpResponseNotAllowed,
+    JsonResponse,
+)
+
+if TYPE_CHECKING:
+    from opencontractserver.users.models import User
 from django.shortcuts import redirect, render
 from django.urls import reverse
 from django.utils.decorators import method_decorator
@@ -35,17 +45,17 @@
 ADMIN_LOGIN_PAGE_RATE = RateLimits.ADMIN_LOGIN_PAGE
 
 
-def _get_login_url():
+def _get_login_url() -> str:
     """Get the admin login URL using reverse() for proper URL resolution."""
     return reverse("admin_auth0_login")
 
 
-def _get_admin_index_url():
+def _get_admin_index_url() -> str:
     """Get the admin index URL using reverse() for proper URL resolution."""
     return reverse("admin:index")
 
 
-def _get_next_url_from_request(request):
+def _get_next_url_from_request(request: HttpRequest) -> Optional[str]:
     """
     Extract the 'next' parameter from request, checking POST first then GET.
 
@@ -60,7 +70,11 @@ def _get_next_url_from_request(request):
     return request.POST.get("next") or request.GET.get("next")
 
 
-def _get_safe_redirect_url(request, url=None, default=None):
+def _get_safe_redirect_url(
+    request: HttpRequest,
+    url: Optional[str] = None,
+    default: Optional[str] = None,
+) -> str:
     """
     Validate and return a safe redirect URL.
 
@@ -103,7 +117,7 @@ def _get_safe_redirect_url(request, url=None, default=None):
     return default
 
 
-def _get_safe_logout_return_url(request):
+def _get_safe_logout_return_url(request: HttpRequest) -> str:
     """
     Get a safe return URL for Auth0 logout.
 
@@ -163,7 +177,7 @@ class Auth0AdminLoginView(View):
 
     @method_decorator(csrf_protect)
     @method_decorator(view_ratelimit(rate=ADMIN_LOGIN_PAGE_RATE, block=False))
-    def get(self, request):
+    def get(self, request: HttpRequest) -> HttpResponseBase:
         """Display the appropriate login form."""
         if getattr(request, "limited", False):
             logger.warning("Rate limit exceeded for admin login page GET")
@@ -203,7 +217,7 @@ def get(self, request):
 
     @method_decorator(csrf_protect)
     @method_decorator(view_ratelimit(rate=ADMIN_LOGIN_RATE, block=False))
-    def post(self, request):
+    def post(self, request: HttpRequest) -> HttpResponseBase:
         """Handle token-based login via POST or password authentication."""
         if getattr(request, "limited", False):
             logger.warning("Rate limit exceeded for admin login POST")
@@ -236,7 +250,7 @@ def post(self, request):
         return redirect(_get_login_url())
 
     @staticmethod
-    def _get_csp_nonce(request):
+    def _get_csp_nonce(request: HttpRequest) -> str:
         """Return the CSP nonce from the request, warning if it is missing.
 
         An empty nonce would produce an invalid CSP directive that silently
@@ -252,12 +266,14 @@ def _get_csp_nonce(request):
             return ""
         return nonce
 
-    def _authenticate_with_token(self, request, token):
+    def _authenticate_with_token(
+        self, request: HttpRequest, token: str
+    ) -> HttpResponseBase:
         """Authenticate user with Auth0 JWT token."""
         from config.jwt_utils import get_user_from_jwt_token
 
         try:
-            user = get_user_from_jwt_token(token)
+            user: Optional["User"] = get_user_from_jwt_token(token)
 
             if user and user.is_active:
                 # Sync admin claims from token (only during admin login, not API requests)
@@ -282,12 +298,12 @@ def _authenticate_with_token(self, request, token):
                 )
                 # Validate redirect URL to prevent open redirect attacks
                 next_url = _get_safe_redirect_url(request)
-                logger.info("Admin login successful for user ID %s", user.id)
+                logger.info("Admin login successful for user ID %s", user.pk)
                 return redirect(next_url)
             else:
                 logger.warning(
                     "User ID %s denied admin access",
-                    user.id if user else "unknown",
+                    user.pk if user else "unknown",
                 )
                 messages.error(
                     request, "You do not have permission to access the admin."
@@ -299,7 +315,7 @@ def _authenticate_with_token(self, request, token):
             messages.error(request, "Authentication failed. Please try again.")
             return redirect(_get_login_url())
 
-    def _sync_admin_claims(self, user, token):
+    def _sync_admin_claims(self, user: "User", token: str) -> bool:
         """
         Sync admin claims from Auth0 token to user model.
 
@@ -324,7 +340,7 @@ def _sync_admin_claims(self, user, token):
             return sync_admin_claims_from_payload(user, payload)
         except Exception as e:
             # Log but don't fail authentication - claim sync is secondary
-            logger.warning("Failed to sync admin claims for user ID %s: %s", user.id, e)
+            logger.warning("Failed to sync admin claims for user ID %s: %s", user.pk, e)
             return False
 
 
@@ -337,7 +353,7 @@ class Auth0AdminLogoutView(View):
     """
 
     @method_decorator(csrf_protect)
-    def post(self, request):
+    def post(self, request: HttpRequest) -> HttpResponseBase:
         """Log out the user and redirect appropriately."""
         logout(request)
 
@@ -355,7 +371,7 @@ def post(self, request):
 
         return redirect(_get_admin_index_url())
 
-    def get(self, request):
+    def get(self, request: HttpRequest) -> HttpResponse:
         """
         Reject GET requests for logout.
 

config/admin_auth/views.py

@@ -1,7 +1,20 @@
+"""
+API-token authentication backend.
+
+Works with any token model that exposes ``key`` (str) and ``user``
+(AbstractBaseUser) — not limited to DRF's built-in
+:class:`rest_framework.authtoken.models.Token`. The prefix expected in
+the ``Authorization`` header is read from ``settings.API_TOKEN_PREFIX``
+so the deployment can brand it (``Api-Token``, ``Bearer``, etc.).
+"""
+
 import logging
+from typing import Any, Optional
 
 from django.conf import settings
 from django.contrib.auth import get_user_model
+from django.contrib.auth.models import AbstractBaseUser
+from django.http import HttpRequest
 from django.utils.translation import gettext_lazy as _
 from rest_framework import exceptions
 
@@ -12,10 +25,23 @@
 
 
 class ApiKeyBackend:
+    """
+    Authenticate GraphQL/DRF requests against an API-token model.
 
-    model = None
+    A custom token model may be supplied by setting
+    :attr:`ApiKeyBackend.model`. It must expose:
+
+    * ``key``  — the opaque string identifying the token
+    * ``user`` — the :class:`AbstractBaseUser` the token belongs to
+    """
 
-    def get_model(self):
+    model: Optional[type[Any]] = None
+    # Used by DRF to populate the ``WWW-Authenticate`` response header on
+    # 401 responses (see :meth:`authenticate_header`). Declared as a
+    # class attribute so mypy knows the attribute always exists.
+    keyword: str = "Token"
+
+    def get_model(self) -> type[Any]:
         logger.debug("Getting token model")
         if self.model is not None:
             return self.model
@@ -25,13 +51,9 @@ def get_model(self):
         logger.debug("Using default Token model")
         return Token
 
-    """
-    A custom token model may be used, but must have the following properties.
-    * key -- The string identifying the token
-    * user -- The user to which the token belongs
-    """
-
-    def authenticate(self, request=None, **kwargs):
+    def authenticate(
+        self, request: Optional[HttpRequest] = None, **kwargs: Any
+    ) -> Optional[AbstractBaseUser]:
         logger.debug("Starting authentication process")
 
         if request is None:
@@ -66,7 +88,7 @@ def authenticate(self, request=None, **kwargs):
 
         return self.authenticate_credentials(token)
 
-    def authenticate_credentials(self, key):
+    def authenticate_credentials(self, key: str) -> AbstractBaseUser:
         logger.debug("Authenticating credentials")
         model = self.get_model()
         try:
@@ -85,6 +107,6 @@ def authenticate_credentials(self, key):
         logger.debug(f"Successfully authenticated user: {token.user.username}")
         return token.user
 
-    def authenticate_header(self, request):
+    def authenticate_header(self, request: HttpRequest) -> str:
         logger.debug("Getting authentication header")
         return self.keyword

config/graphql_api_token_auth/backends.py

@@ -1,4 +1,9 @@
+"""GraphQL middleware that authenticates requests via API-token headers."""
+
+from typing import Any, Callable
+
 from django.contrib.auth import authenticate, get_user_model
+from django.http import HttpRequest
 
 from config.graphql_api_token_auth.utils import (
     get_http_authorization,
@@ -8,15 +13,16 @@
 User = get_user_model()
 
 
-def _context_has_user(request):
+def _context_has_user(request: HttpRequest) -> bool:
     return hasattr(request, "user") and request.user.is_authenticated
 
 
-def _authenticate(request):
+def _authenticate(request: HttpRequest) -> bool:
     """
-    Check if we should attempt API token authentication.
-    Returns False if the request has a Bearer token (which should be handled by JWT middleware)
-    or if the request is already authenticated.
+    Return True if we should attempt API-token authentication.
+
+    Returns False if the request carries a Bearer token (handled by JWT
+    middleware) or is already authenticated via another backend.
     """
     auth = request.META.get("HTTP_AUTHORIZATION", "")
     if auth.startswith("Bearer "):
@@ -27,17 +33,25 @@ def _authenticate(request):
 
 
 class ApiKeyTokenMiddleware:
-    def __init__(self):
-        self.cached_allow_any = set()
+    """Graphene middleware that authenticates via an API-token header."""
+
+    def __init__(self) -> None:
+        self.cached_allow_any: set[str] = set()
 
-    def authenticate_context(self, info, **kwargs):
+    def authenticate_context(self, info: Any, **kwargs: Any) -> bool:
         root_path = info.path[0]
 
         if root_path not in self.cached_allow_any:
             return True
         return False
 
-    def resolve(self, next, root, info, **kwargs):
+    def resolve(
+        self,
+        next: Callable[..., Any],
+        root: Any,
+        info: Any,
+        **kwargs: Any,
+    ) -> Any:
 
         # Check to see if user already on context