"""Every admin route is guarded, on all three axes (task 8.1, add-admin-access-control D11). The point of the split architecture is that authorization lives in the API. Three separate things have to be true of every admin endpoint, and each has its own failure mode: 1. **Capability** — a `require(...)` dependency. Without it the route is reachable by anyone holding a session, whatever their role. 2. **Scope** — the records it touches are narrowed to the caller's own events. Without it a Host reads another customer's guest list, and nothing about the response looks wrong. 3. **Confinement** — it is refused while the caller holds a temporary password. Without it, an admin-issued password that travelled over WhatsApp is a full session. The list of routes is **derived, not written down**. A hand-maintained list of 42 endpoints goes stale the first time somebody adds one in a hurry, and the failure mode of the one that was missed is a silent cross-customer read. So a new admin route with no entry in `SCOPE_COVERAGE` fails this file until someone states how it is scoped. """ from functools import lru_cache from typing import Any import pytest from app.main import app from app.routers import ( admin_cards, admin_data, admin_events, admin_guests, admin_messaging, admin_stats, admin_users, admin_wedding, ) from app.services.policy import Action ADMIN_ROUTERS = ( admin_stats, admin_guests, admin_data, admin_messaging, admin_events, admin_cards, admin_users, admin_wedding, ) #: How each admin route is narrowed. The value is documentation that fails when it stops #: being true of the route list, not an assertion about the query itself — that is what #: `test_scoping.py` exercises against a real database. #: #: * `event` — resolves an event, guest, invitation or design and applies the caller's scope #: * `list` — a query whose WHERE carries a scope fragment #: * `roster` — Super Admin only by capability; scope has no meaning (it *assigns* scope) #: * `system` — a wedding-wide setting, Super Admin only via `scope.require_system_wide` #: * `self` — acts only on the caller's own account SCOPE_COVERAGE: dict[str, str] = { "/admin/stats": "list", "/admin/stats/trend": "list", "/admin/events": "list", "/admin/events/{event_id}": "event", "/admin/events/{event_id}/delete-impact": "event", "/admin/events/{event_id}/broadcast-date-change": "event", "/admin/events/{event_id}/transfer": "roster", "/admin/events/{event_id}/guests": "event", "/admin/events/{event_id}/guest-ids": "event", "/admin/events/{event_id}/invitations/compose": "event", "/admin/events/{event_id}/invitations/send-batch": "event", "/admin/events/{event_id}/guests/import": "event", "/admin/events/{event_id}/guests/import/preview": "event", "/admin/events/{event_id}/card-designs": "event", "/admin/guests": "list", "/admin/guests/{guest_id}": "event", "/admin/invitations/{invitation_id}/rsvp": "event", "/admin/invitations/{invitation_id}/message": "event", "/admin/invitations/{invitation_id}/link-copied": "event", "/admin/invitations/{invitation_id}/send": "event", "/admin/send-batches/{batch_id}": "list", "/admin/card-designs/{design_id}": "event", "/admin/card-designs/{design_id}/publish": "event", "/admin/card-designs/{design_id}/unpublish": "event", "/admin/export": "list", "/admin/qr/{event_slug}": "event", "/admin/qr/guest/{invitation_id}": "event", "/admin/send/preview": "list", "/admin/send": "list", "/admin/messages": "list", "/admin/messages/retry": "list", "/admin/reminders/preview": "list", "/admin/reminders/{schedule_id}": "event", "/admin/users": "roster", "/admin/users/{admin_id}": "roster", "/admin/users/{admin_id}/activate": "roster", "/admin/users/{admin_id}/reject": "roster", "/admin/users/{admin_id}/withdraw": "roster", "/admin/users/{admin_id}/temporary-password": "roster", "/admin/wedding": "system", "/admin/wedding/invitation-messages": "system", } @lru_cache def _openapi() -> dict[str, Any]: return app.openapi() def _admin_operations() -> list[tuple[str, str, dict[str, Any]]]: found = [] for path, entry in _openapi()["paths"].items(): if not path.startswith("/api/admin"): continue for method, operation in entry.items(): if method == "parameters": continue found.append((method.upper(), path, operation)) return found def _admin_routes() -> list[tuple[str, Any]]: """(path, endpoint) for every declared admin route, across every admin router.""" routes = [] for module in ADMIN_ROUTERS: for route in module.router.routes: endpoint = getattr(route, "endpoint", None) path = getattr(route, "path", "") if endpoint is not None and path.startswith("/admin"): routes.append((path, endpoint)) return routes def _guards_of(endpoint: Any) -> list[Any]: """The `require(...)` closures a route declares through its parameter defaults.""" defaults = getattr(endpoint, "__defaults__", None) or () return [ d for d in defaults if getattr(getattr(d, "dependency", None), "__name__", "") == "_dependency" ] def test_admin_surface_exists() -> None: """A guard test that silently matches nothing would pass forever.""" assert len(_admin_operations()) >= 8 def test_every_admin_route_has_a_capability_dependency() -> None: """Inspect the actual dependency tree rather than the schema — this is the real check.""" missing = [] for path, endpoint in _admin_routes(): if _guards_of(endpoint): continue # A router-level dependency counts too: `admin_users` guards its whole router with # `require(Action.MANAGE_ADMINS)` rather than repeating it on nine endpoints. if any( getattr(getattr(d, "dependency", None), "__name__", "") == "_dependency" for module in ADMIN_ROUTERS for d in module.router.dependencies if any(p == path for p, _ in _routes_of(module)) ): continue missing.append(f"{path} ({endpoint.__name__})") assert not missing, f"admin routes without a capability guard: {missing}" def _routes_of(module: Any) -> list[tuple[str, Any]]: return [ (getattr(r, "path", ""), r) for r in module.router.routes if getattr(r, "endpoint", None) is not None ] def test_every_admin_route_declares_how_it_is_scoped() -> None: """The load-bearing one. A new admin endpoint with no entry above fails here. That is the whole point: the cost of forgetting to scope one is another customer's guest list, and the cost of this test failing is thirty seconds spent adding a line. """ declared = set(SCOPE_COVERAGE) actual = {path for path, _ in _admin_routes()} undeclared = sorted(actual - declared) assert not undeclared, ( "These admin routes do not say how they are scoped. Add each to SCOPE_COVERAGE " f"once you have scoped it in the handler: {undeclared}" ) stale = sorted(declared - actual) assert not stale, f"SCOPE_COVERAGE names routes that no longer exist: {stale}" def test_scoped_routes_reference_the_scope_service() -> None: """Every route declared `event` or `list` must actually mention the scope service. A cheap structural check on the handler's source, not a proof — the behavioural proof is `test_scoping.py`. What this catches is the specific regression of somebody adding a route to `SCOPE_COVERAGE` to make the test above pass without scoping the handler. """ import inspect unscoped = [] for path, endpoint in _admin_routes(): if SCOPE_COVERAGE.get(path) not in {"event", "list"}: continue source = inspect.getsource(endpoint) module_source = inspect.getsource(inspect.getmodule(endpoint)) # Either the handler applies scope itself, or it delegates to a helper in its module # that does (`_load_guest`, `_load_event_for_send`, `_filtered_guests`, ...). if "scope." in source or ("scope." in module_source and "admin" in source): continue unscoped.append(f"{path} ({endpoint.__name__})") assert not unscoped, f"declared scoped but never consults the scope service: {unscoped}" @pytest.mark.parametrize( "path", [p for p, kind in SCOPE_COVERAGE.items() if kind not in {"self"}], ) def test_admin_routes_are_confined_by_a_temporary_password(path: str) -> None: """No admin surface is reachable while the caller must change their password (D14). `require(...)` depends on `get_unconfined_admin`, so every capability-guarded route inherits the confinement. This asserts the chain rather than trusting it, because the obvious "fix" for an awkward test is to depend on `get_current_admin` directly — which would silently opt that route out. """ from app.services.auth import get_unconfined_admin endpoints = [e for p, e in _admin_routes() if p == path] assert endpoints, f"{path} vanished from the router" for endpoint in endpoints: guards = _guards_of(endpoint) if not guards: # Router-level guard (the roster). Its dependency chain is asserted once below. continue for guard in guards: sub = guard.dependency.__wrapped__ if hasattr(guard.dependency, "__wrapped__") else None del sub # the closure is inspected through its cell contents instead # `require` closes over `get_unconfined_admin` as its own dependency's default. defaults = guards[0].dependency.__defaults__ or () assert any(getattr(d, "dependency", None) is get_unconfined_admin for d in defaults), ( f"{path} is capability-guarded but not confined by the temporary-password gate" ) def test_the_roster_router_is_super_admin_only() -> None: """The roster assigns scope, so reaching it is reaching everything.""" guards = [ d for d in admin_users.router.dependencies if getattr(getattr(d, "dependency", None), "__name__", "") == "_dependency" ] assert guards, "the admin roster router has no capability guard" def test_export_requires_the_export_capability_specifically() -> None: """CSV export is PII leaving the system; it must not ride on a generic view grant.""" export_route = next( r for r in admin_data.router.routes if getattr(r, "path", "") == "/admin/export" ) assert _guards_of(export_route.endpoint), "export route has no capability guard" def test_actions_enum_covers_the_prd_matrix() -> None: """Every capability the PRD names has an Action, so none is enforced ad hoc.""" required = { "view_dashboard", "add_edit_guests", "delete_guests", "send_messages", "export_csv", "edit_content", "configure_gateways", "manage_admins", "view_audit_log", } assert required <= {str(a) for a in Action}