"""Cross-cutting middleware: MFA enforcement + session idle-timeout. Both middlewares only act on authenticated requests. They are mounted after ``django.contrib.auth.middleware.AuthenticationMiddleware`` and before ``apps.core.audit.middleware.AuditMiddleware`` so the audit log records the final disposition of the request (e.g. a 302 to the MFA enrolment page). """ from __future__ import annotations import time from django.conf import settings from django.contrib import messages from django.contrib.auth import logout from django.http import HttpResponse from django.shortcuts import redirect from django.urls import resolve from apps.core.authentication.mfa_policy import has_verified_mfa, mfa_required_for # Paths the user must keep reaching even when MFA is mandated but not yet # enrolled. Otherwise a privileged user without MFA could never log out or # enrol. Keep this list short — anything else loops back to the enrolment URL. _MFA_BYPASS_URL_NAMES: frozenset[str] = frozenset( { "accounts:login", "accounts:logout", "accounts:mfa_enroll", "accounts:mfa_verify", "accounts:mfa_manage", } ) _MFA_BYPASS_PATH_PREFIXES: tuple[str, ...] = ( "/static/", "/uploads/", "/media/", "/auth/mfa/", "/auth/login", "/auth/logout", ) # Path prefixes whose traffic should NOT count as "user activity" — otherwise # the SSE keepalive (added in Phase 7-core) would refresh the session timer # forever and idle-timeout would never fire. _IDLE_SKIP_PATH_PREFIXES: tuple[str, ...] = ( "/static/", "/uploads/", "/media/", "/notifications/badge/", "/notifications/stream/", ) _IDLE_KEY = "_last_activity" class MFAEnforcementMiddleware: """Redirect privileged users without MFA to the enrolment page. The user is technically logged in (so ``@login_required`` views resolve), but every non-bypassed URL bounces them to ``/auth/mfa/enroll/`` until they have a verified ``MFADevice``. This catches the edge case where a role is promoted to privileged level via Django admin or a management command while the user's session is already active — they pick up the mandate on their next request. """ def __init__(self, get_response): self.get_response = get_response def __call__(self, request): if not getattr(request, "user", None) or not request.user.is_authenticated: return self.get_response(request) if not mfa_required_for(request.user): return self.get_response(request) if has_verified_mfa(request.user): return self.get_response(request) # User is privileged + un-enrolled. Allow the small bypass list. path = request.path or "" if any(path.startswith(p) for p in _MFA_BYPASS_PATH_PREFIXES): return self.get_response(request) try: match = resolve(path) url_name = f"{match.namespace}:{match.url_name}" if match.namespace else (match.url_name or "") except Exception: # noqa: BLE001 — non-resolvable paths fall through to the redirect url_name = "" if url_name in _MFA_BYPASS_URL_NAMES: return self.get_response(request) # Force them to enrol. if not request.session.get("_mfa_mandate_notified"): messages.warning( request, "Your role requires multi-factor authentication. " "Please register an authenticator app to continue.", ) request.session["_mfa_mandate_notified"] = True return redirect("accounts:mfa_enroll") class SessionIdleTimeoutMiddleware: """Log users out after ``SESSION_IDLE_TIMEOUT_SECONDS`` of inactivity. Only requests to "real" URLs refresh the timer — static assets, the SSE notification stream, and the badge polling endpoint are excluded so a backgrounded tab does not hold the session open indefinitely. """ def __init__(self, get_response): self.get_response = get_response def __call__(self, request): if not getattr(request, "user", None) or not request.user.is_authenticated: return self.get_response(request) path = request.path or "" is_activity = not any(path.startswith(p) for p in _IDLE_SKIP_PATH_PREFIXES) timeout = int(getattr(settings, "SESSION_IDLE_TIMEOUT_SECONDS", 3600) or 3600) now = int(time.time()) last = request.session.get(_IDLE_KEY) if last is None: if is_activity: request.session[_IDLE_KEY] = now return self.get_response(request) if now - int(last) >= timeout: logout(request) messages.warning(request, "You were signed out after a period of inactivity.") login_url = "/auth/login/" try: from django.urls import reverse login_url = reverse("accounts:login") except Exception: # noqa: BLE001 pass target = f"{login_url}?reason=idle" # HTMX-driven requests must not receive a 302 — HTMX would follow it # and swap the login page HTML into the original target element, # producing a "login overlaid on the current page" effect. Sits # outside HtmxAuthRedirectMiddleware in the chain, so detect HTMX # via the HX-Request header directly. if request.headers.get("HX-Request") == "true": response = HttpResponse(status=200) response["HX-Redirect"] = target return response return redirect(target) if is_activity: request.session[_IDLE_KEY] = now return self.get_response(request) class HtmxAuthRedirectMiddleware: """Convert auth-bounce 302s on HTMX requests into HX-Redirect responses. When a session expires and an HTMX-driven request (HTMX-boosted link, hx-get, hx-post, ...) hits a login-required view, Django returns a 302 to ``LOGIN_URL``. HTMX follows the redirect and swaps the rendered login page HTML into the request's target element, producing the "login form overlaid on the current page" symptom. This middleware detects that case and replaces the 302 with a 200 carrying ``HX-Redirect: ``, which HTMX honours by doing a full-page browser navigation. Sits AFTER HtmxMiddleware so ``request.htmx`` is populated, and BEFORE AuditMiddleware so the rewritten response is what gets audited. """ LOGIN_REDIRECT_PREFIXES: tuple[str, ...] = ( "/accounts/login", "/auth/login", ) def __init__(self, get_response): self.get_response = get_response def __call__(self, request): response = self.get_response(request) if not getattr(request, "htmx", False): return response if response.status_code not in (301, 302, 303, 307, 308): return response location = response.get("Location", "") if not any(location.startswith(p) for p in self.LOGIN_REDIRECT_PREFIXES): return response replacement = HttpResponse(status=200) replacement["HX-Redirect"] = location return replacement