Verify Cloudflare Access assertions at the origin

backend/core/users.py

@@ -21,6 +21,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
+from core.access_identity import AccessIdentity
 from core.env_parsing import parse_comma_list
 
 # ── Config ────────────────────────────────────────────────────────────────────
@@ -65,6 +66,19 @@ def _derive_auth_flags(env: Mapping[str, str]) -> tuple[bool, bool]:
 
 AUTH_ENABLED, AUTH_BYPASS = _derive_auth_flags(os.environ)
 
+#: The header Cloudflare sets on every request its Access applications admit.
+ACCESS_ASSERTION_HEADER = "Cf-Access-Jwt-Assertion"
+
+#: One instance, so the JWKS cache and its lock are shared across requests.
+#: The audience differs per environment and comes from the compose overlays; the
+#: team domain is the same everywhere and comes from the base compose file.
+#: Either unset means unconfigured, and an unconfigured verifier is never
+#: consulted rather than refusing assertions nobody sent.
+access_identity = AccessIdentity(
+    team_domain=os.getenv("CF_ACCESS_TEAM_DOMAIN", ""),
+    audience=os.getenv("CF_ACCESS_AUD", ""),
+)
+
 _DATA_DIR = Path(__file__).resolve().parent.parent / "data"
 _DATA_DIR.mkdir(parents=True, exist_ok=True)
 # RETINA_DB_PATH exists so tests and one-off migrations can point at a scratch
@@ -312,8 +326,42 @@ async def _read_user_from_request(request: Request) -> User | None:
     return user
 
 
+def _access_user_dict(email: str) -> dict:
+    """A user dict for a verified Access identity, with no database row.
+
+    Membership of the Access group is what grants the console, so anyone whose
+    assertion verifies for this environment's audience is an administrator; a
+    second list in AUTH_ADMIN_EMAILS would only be one more thing to drift.
+
+    The id is derived from the email rather than allocated, so the same person
+    is the same id across requests and restarts and the destructive endpoints
+    stay attributable in /api/admin/events.
+    """
+    return {
+        "id": str(uuid.uuid5(uuid.NAMESPACE_URL, f"mailto:{email}")),
+        "email": email,
+        "name": email.split("@")[0],
+        "avatar": "",
+        "provider": "cloudflare-access",
+        "role": "admin",
+        "is_superuser": True,
+        "created_at": 0,
+    }
+
+
+async def _access_user_from_request(request: Request) -> dict | None:
+    """The verified Access identity for this request, or None."""
+    if not access_identity.is_configured():
+        return None
+    email = await access_identity.identity(request.headers.get(ACCESS_ASSERTION_HEADER))
+    return _access_user_dict(email) if email else None
+
+
 async def get_current_user(request: Request) -> dict:
     """Return user dict or raise 401. Returns anonymous admin where AUTH_BYPASS is opted into."""
+    access = await _access_user_from_request(request)
+    if access is not None:
+        return access
     if AUTH_BYPASS:
         return dict(ANONYMOUS_USER)
     user = await _read_user_from_request(request)
@@ -324,6 +372,11 @@ async def get_current_user(request: Request) -> dict:
 
 async def require_admin(request: Request) -> dict:
     """Like get_current_user but also enforces superuser/admin role."""
+    # Ahead of the bypass: where both are available the real person is the better
+    # answer, since "Admin (no auth)" is not an attribution.
+    access = await _access_user_from_request(request)
+    if access is not None:
+        return access
     if AUTH_BYPASS:
         return dict(ANONYMOUS_USER)
     user = await _read_user_from_request(request)

backend/requirements.txt

@@ -9,6 +9,10 @@ pyyaml==6.0.3
 cryptography==46.0.6
 boto3==1.42.89
 fastapi-users[sqlalchemy]==13.0.0
+# Direct since core/access_identity.py imports it to verify Cloudflare Access
+# assertions; it had arrived only as a transitive of fastapi-users, which pins
+# this exact version, so it is not a free choice. [crypto] for RS256.
+pyjwt[crypto]==2.8.0
 aiosqlite==0.20.0
 alembic==1.13.2
 pyarrow==16.1.0

backend/routes/auth.py

@@ -27,7 +27,6 @@
 )
 from core.users import (
     ANONYMOUS_USER,
-    AUTH_BYPASS,
     JWT_LIFETIME_SECONDS,
     JWT_SECRET,
     get_current_user,
@@ -231,10 +230,13 @@ async def callback_github(request: Request, code: str = "", state: str = ""):
 
 @router.get("/me")
 async def me(request: Request):
-    if AUTH_BYPASS:
-        return {**ANONYMOUS_USER, "auth_enabled": False}
     user_dict = await get_current_user(request)
-    return {**user_dict, "auth_enabled": True}
+    # Delegated rather than short-circuiting on AUTH_BYPASS, so this agrees with
+    # what require_admin decided: a verified Access assertion outranks the
+    # bypass, and answering "Admin (no auth)" while the admin routes attribute a
+    # real person would make the console wrong about its own session.
+    anonymous = user_dict["id"] == ANONYMOUS_USER["id"]
+    return {**user_dict, "auth_enabled": not anonymous}
 
 
 @router.post("/logout")