Merge pull request #1370 from Open-Source-Legal/claude/resolve-issue-1333-22Y0E

Add type annotations to auth, users, and notifications packages

Author: JSv4

CHANGELOG.md

@@ -38,6 +38,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - `.github/workflows/backend.yml`: `Run mypy` step added to the `linter` job; the preceding `pip install -r requirements/local.txt` step satisfies the django-stubs plugin.
   - `docs/typing/README.md`: how to run locally / via pre-commit / in the test container, plus the per-file graduation workflow.
   - `docs/typing/mypy_baseline.txt`: frozen error list (sorted for stable diffs) so follow-up issues can measure progress.
+- **Mypy: typed the auth, users, and notifications packages** (Issue #1333, follow-up to #1331): Brought `config/admin_auth`, `config/graphql_api_token_auth`, `opencontractserver/users`, and `opencontractserver/notifications` to full return-annotation coverage and graduated them out of the mypy baseline. These packages sit on hot paths for authentication, token verification, session handling, and notification dispatch; typing them documents invariants that used to live only in `CLAUDE.md`.
+  - Signal handlers (`opencontractserver/users/signals.py`, `opencontractserver/notifications/signals.py`) now declare `sender: type[Model]`, `instance: Model`, `created: bool`, `**kwargs: Any`, and `-> None` so drive-by edits can't silently change the signal contract.
+  - `opencontractserver/notifications/__init__.py` gained a module-level docstring calling out that the app deliberately diverges from `AnnotatePermissionsForReadMixin` — a single `recipient` FK is the authoritative visibility gate. The design note in `CLAUDE.md` is referenced so changes to one file flag the other.
+  - `mypy.ini`: dropped `[mypy-config.admin_auth.*]`, `[mypy-config.graphql_api_token_auth.backends]`, and `[mypy-opencontractserver.users.models]` `ignore_errors` sections. Added a narrow `disable_error_code = django-manager-missing` on `opencontractserver.users.models` only, documented inline: `django-tree-queries`' `as_manager(with_tree_fields=True)` pattern used by `Corpus`, `CorpusFolder`, and `DocumentPath` creates manager classes at runtime that the `mypy_django_plugin` can't introspect, so reverse `_set` accessors on `User` can't resolve — a known plugin limitation, not a code error. Graduating those three models with explicit manager typing will let us delete the disable.
+  - `docs/typing/mypy_baseline.txt`: pruned all 20 entries for the graduated modules.
+  - Bug fixes uncovered by typing:
+    - `config/graphql_api_token_auth/backends.py`: `ApiKeyBackend.authenticate_header` referenced `self.keyword` but the attribute was never declared. DRF calls this on 401 responses to build the `WWW-Authenticate` header, so an unauthenticated request would have hit `AttributeError` at response time. Added `keyword: str = "Token"` as a class attribute.
+    - `opencontractserver/notifications/signals.py`: `create_moderation_notification` dereferenced `action.moderator.username` unconditionally, but the `ModerationAction.moderator` FK permits `NULL`. A moderation action created without a moderator would crash the signal handler (and therefore the originating save). Added an explicit early return with a debug log when `action.moderator is None`.
 
 ### Changed
 

config/admin_auth/backends.py

@@ -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/views.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/graphql_api_token_auth/backends.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