"""Request middleware — resolves tenant, builds context, enforces auth gates.""" from __future__ import annotations from collections.abc import Callable from typing import TYPE_CHECKING from django.http import HttpResponse, JsonResponse from simorgh.apps.iam.engine import collect_permissions_for_roles from simorgh.apps.memberships.services import active_memberships_for, assignable_node_ids from simorgh.apps.tenants.resolver import resolve_tenant from simorgh.core.context import ( RequestContext, reset_request_context, set_request_context, ) if TYPE_CHECKING: from django.http import HttpRequest # Paths where the TenantResolverMiddleware skips resolution entirely # (user is not yet authenticated; a session doesn't exist yet). _TENANT_SKIP = ( "/api/v1/auth/login", "/api/v1/auth/logout", "/api/v1/auth/refresh", "/api/v1/auth/csrf", "/api/v1/auth/otp/request", "/api/v1/auth/otp/verify", "/api/v1/localization/countries", "/api/v1/localization/languages", ) # Paths that bypass the PermissionGate entirely (no auth/tenant checks). # Auth endpoints that need tenant resolution (me, tenants, switch-tenant) # are NOT here — the resolver runs for them so _session_context is accurate, # but the gate must not block them when tenant is still null. _GATE_SKIP = _TENANT_SKIP + ( "/api/v1/auth/me", "/api/v1/auth/tenants", "/api/v1/auth/switch-tenant", # Public share-link consumption: no auth or tenant needed "/api/v1/dms/share/", ) def _skip_resolver(path: str) -> bool: return ( path.startswith("/admin/") or path.startswith("/api/health") or path.startswith("/api/schema") or path.startswith("/api/docs") or any(path.startswith(p) for p in _TENANT_SKIP) or path.startswith("/static/") or path.startswith("/media/") ) def _exempt(path: str) -> bool: """True when the PermissionGate should pass the request through unscrutinised.""" return ( path.startswith("/admin/") or path.startswith("/api/health") or path.startswith("/api/schema") or path.startswith("/api/docs") or any(path.startswith(p) for p in _GATE_SKIP) or path.startswith("/static/") or path.startswith("/media/") ) def _resolve_workspace(request: HttpRequest, tenant) -> object | None: """Read the ``X-Workspace`` header and resolve to a Workspace, or None. Quietly returns None when the header is missing, the tenant is unknown, or the slug doesn't match — workspace context is always optional (background jobs and tenant-level APIs may run without one). """ if tenant is None: return None slug = request.headers.get("X-Workspace") or request.META.get("HTTP_X_WORKSPACE") if not slug: return None from simorgh.apps.workspaces.models import Workspace return Workspace.objects.filter(tenant=tenant, slug=slug, is_active=True).first() def _resolve_org_node(request: HttpRequest, tenant, user) -> object | None: """Resolve the active OrganizationNode for this request. Priority: 1. ``X-Org-Node`` header (explicit per-request override). 2. ``UserOrgContextPreference`` (persisted user preference). 3. None (caller should use tenant root or org_node_ids). """ if tenant is None or user is None or not user.is_authenticated: return None # Explicit header override node_id = request.headers.get("X-Org-Node") or request.META.get("HTTP_X_ORG_NODE") if node_id: from simorgh.apps.organizations.models import OrganizationNode try: return OrganizationNode.objects.get(pk=int(node_id), tenant=tenant, is_active=True) except (ValueError, OrganizationNode.DoesNotExist): pass # Fall back to persisted preference from simorgh.apps.workspaces.models import UserOrgContextPreference try: pref = UserOrgContextPreference.objects.get(user=user, tenant=tenant) if pref.active_organization_node is not None: return pref.active_organization_node except UserOrgContextPreference.DoesNotExist: pass return None class TenantResolverMiddleware: """Attach the resolved `Tenant` (or None) to `request.tenant`.""" def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None: self.get_response = get_response def __call__(self, request: HttpRequest) -> HttpResponse: request.tenant = None if _skip_resolver(request.path) else resolve_tenant(request) return self.get_response(request) class RequestContextMiddleware: """Build the immutable `RequestContext` and bind it to a contextvar.""" def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None: self.get_response = get_response def __call__(self, request: HttpRequest) -> HttpResponse: tenant = getattr(request, "tenant", None) user = getattr(request, "user", None) memberships: tuple = () permissions: frozenset[str] = frozenset() org_ids: frozenset[int] = frozenset() if user is not None and user.is_authenticated and tenant is not None: memberships_qs = active_memberships_for(user.pk, tenant.pk) memberships = tuple(memberships_qs) roles = [m.role for m in memberships] permissions = frozenset(collect_permissions_for_roles(roles)) org_ids = frozenset(assignable_node_ids(memberships)) ctx = RequestContext( actor=user if user and user.is_authenticated else None, tenant=tenant, memberships=memberships, permissions=permissions, org_node_ids=org_ids, workspace=_resolve_workspace(request, tenant), active_org_node=_resolve_org_node(request, tenant, user), ) token = set_request_context(ctx) try: request.simorgh_context = ctx # Reset per-request IAM cache so cross-request leaks are impossible. from simorgh.apps.iam import cache as iam_cache iam_cache.clear() return self.get_response(request) finally: reset_request_context(token) class PermissionGateMiddleware: """Reject unauthenticated/unscoped API calls early. Health/schema/docs are exempt. Anything under `/api/` that is **not** one of the exempt paths must carry an authenticated user; tenant resolution failure surfaces as 400 so the client knows to send the `X-Tenant` header or fix its subdomain. """ def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None: self.get_response = get_response def __call__(self, request: HttpRequest) -> HttpResponse: path = request.path if not path.startswith("/api/") or _exempt(path): return self.get_response(request) user = getattr(request, "user", None) if user is None or not user.is_authenticated: return JsonResponse( {"error": {"code": "not_authenticated", "message": "Authentication required."}}, status=401, ) if getattr(request, "tenant", None) is None and not user.is_superuser: return JsonResponse( { "error": { "code": "tenant_unresolved", "message": "Tenant could not be resolved.", } }, status=400, ) return self.get_response(request)