From a30b8527890d37c076ae980495f7c5a1ca0ef59e Mon Sep 17 00:00:00 2001 From: dev_agent <17+dev_agent@code.cubecraftcreations.com> Date: Wed, 23 Sep 2026 00:59:39 -0400 Subject: [PATCH] Admin auth: nestquest-admins group gate and service-token rejection (#157) --- api/auth.py | 100 ++++++++++++++++++++++++--- api/routes_admin.py | 34 ++++----- tests/test_api_admin_auth.py | 129 ++++++++++++++++++++++++++++++++++- 3 files changed, 237 insertions(+), 26 deletions(-) diff --git a/api/auth.py b/api/auth.py index 8111ebb..c4b0bc1 100644 --- a/api/auth.py +++ b/api/auth.py @@ -35,16 +35,20 @@ the real cause is visible in the server log instead of disguised as a bad credential; a short backoff window after a failed fetch fails fast (re-raising the recorded fault) rather than fetching once per request. -This module validates the TOKEN only — it is authentication, not -authorization. Membership in the ``nestquest-admins`` group is the -NEXT task, which extends :func:`require_admin_jwt` (the verified -claims are already its return value, so no rewrite is needed). +Beyond authentication, :func:`require_admin` is the ONE dependency +every admin route uses: it REJECTS the panel service token outright — +that credential authenticates the panel plane only, so treating it as +a JWT candidate would blur two credential types into one check — and +then requires the verified JWT's ``groups`` claim to name the +``nestquest-admins`` group (authorization, 403, distinct from the +uniform authentication 401 so the two failure classes stay apart). """ from __future__ import annotations import asyncio import binascii import json +import secrets import time from collections.abc import Awaitable, Callable from typing import Annotated @@ -70,6 +74,11 @@ _REJECTION_DETAIL = "Missing or invalid admin credentials" #: Challenge header returned with a 401, per the Bearer auth scheme. _WWW_AUTHENTICATE = {"WWW-Authenticate": "Bearer"} +#: The authorization group a verified admin JWT must name in its +#: ``groups`` claim. A module constant (the value Authentik's +#: nestquest-admins group maps to) so it is clearly named in one place. +ADMIN_GROUP = "nestquest-admins" + #: Seconds a fetched JWKS document stays trusted before the next #: fetch. Five minutes bounds how long a rotated-away key keeps #: verifying while staying far above a per-request fetch. @@ -266,6 +275,19 @@ class JwksCache: return document +#: 403 detail for the panel service token on an admin route. Deliberately +#: non-leaking: it names the credential TYPE, not its value or where it came +#: from. 403 (not 401) because this is a REFUSED credential type, not an +#: absent/invalid one — challenging re-authentication would suggest the same +#: credential could work with a retry, which it never can here. +_PANEL_TOKEN_REJECTION_DETAIL = "This credential type is not accepted for admin access" + +#: 403 detail for a valid JWT that is not a member of the admin group. +#: Distinct from the uniform 401 on purpose: the token AUTHENTICATED, but +#: the caller is not AUTHORIZED, so the response may say so. +_FORBIDDEN_DETAIL = "Admin access requires membership in the nestquest-admins group" + + def _reject() -> HTTPException: """Build the single 401 the dependency raises for every failure.""" return HTTPException( @@ -391,9 +413,9 @@ async def require_admin_jwt( :func:`verify_admin_token`, returning the verified claims for the route to use (the probe route surfaces the token's ``sub``). - The nestquest-admins GROUP check is the NEXT task: it extends this - dependency (e.g. by inspecting the returned claims) so every admin - route inherits it — the 401 contract stays exactly this uniform. + The nestquest-admins GROUP check lives in :func:`require_admin`, + which wraps this dependency; ``require_admin_jwt`` remains the plain + JWT-validation contract — the 401 path stays exactly this uniform. """ if authorization is None: raise _reject() @@ -403,4 +425,66 @@ async def require_admin_jwt( return await verify_admin_token(request, credential.strip()) -__all__ = ["JwksCache", "JwksFetcher", "require_admin_jwt", "verify_admin_token"] +async def require_admin( + request: Request, + authorization: Annotated[str | None, Header()] = None, +) -> dict[str, object]: + """The ONE admin-plane dependency: service-token refusal + group gate. + + Order matters. FIRST the Bearer credential is compared against the + configured panel service token (``app.state.config.panel_token``): + a match is refused OUTRIGHT with 403 — BEFORE any JWT parsing — so + the panel credential is never even treated as a JWT candidate, let + alone a non-admin JWT. (403 rather than 401 because this is a + refused credential TYPE, not an invalid one: challenging + re-authentication would imply retrying could ever succeed; the + detail names the type only, never the value.) A config with an + EMPTY panel token never matches anything, so it cannot lock every + Bearer request out. + + Otherwise the credential goes through the existing JWT path + (:func:`require_admin_jwt` → :func:`verify_admin_token`): a missing + or malformed header and any invalid JWT still yield the SAME + uniform 401 as before. A VALID JWT is then checked for + authorization: its ``groups`` claim must be a list containing + :data:`ADMIN_GROUP` — a missing, empty, or non-list ``groups`` or + one without the group raises 403 (distinct from the 401 on + purpose: the token authenticated, the caller did not authorize). + + Returns the verified claims for the route. + """ + if authorization is not None: + scheme, _, credential = authorization.partition(" ") + if scheme.lower() == _BEARER_SCHEME and credential.strip(): + config = getattr(request.app.state, "config", None) + if ( + config is not None + and config.panel_token + # compare_digest on bytes (as require_panel_token does) + # so the refusal does not short-circuit on the first + # mismatching byte — a constant-time comparison keeps a + # timing side channel on the credential off the table. + and secrets.compare_digest( + credential.strip().encode("utf-8"), + config.panel_token.encode("utf-8"), + ) + ): + raise HTTPException( + status_code=403, + detail=_PANEL_TOKEN_REJECTION_DETAIL, + ) + claims = await require_admin_jwt(request, authorization) + groups = claims.get("groups") + if not isinstance(groups, list) or ADMIN_GROUP not in groups: + raise HTTPException(status_code=403, detail=_FORBIDDEN_DETAIL) + return claims + + +__all__ = [ + "ADMIN_GROUP", + "JwksCache", + "JwksFetcher", + "require_admin", + "require_admin_jwt", + "verify_admin_token", +] diff --git a/api/routes_admin.py b/api/routes_admin.py index 7a2e2a2..b7164c7 100644 --- a/api/routes_admin.py +++ b/api/routes_admin.py @@ -2,17 +2,17 @@ The admin plane (``/api/v1/admin/*``, D-012) serves the admin PWA. Every route on :data:`router` requires a valid Authentik OIDC JWT -through the ONE reusable dependency :func:`api.auth.require_admin_jwt` -(declared once at the router level, so each later admin route added to -this router inherits the check). +whose ``groups`` claim names ``nestquest-admins``, through the ONE +reusable dependency :func:`api.auth.require_admin` (declared once at +the router level, so each later admin route added to this router +inherits the check). The panel service token is refused outright on +this plane (403), never treated as a JWT. This task is the AUTH probe only: the ping route is a thin adapter that performs NO business logic and NO auth of its own — it reuses the router's already-verified claims (FastAPI caches a dependency's result per request, so the check runs exactly once) and returns the status -plus the token's subject. Authorization — membership in the -``nestquest-admins`` group — is the NEXT task, layered onto -:func:`api.auth.require_admin_jwt` so every admin route inherits it. +plus the token's subject. """ from __future__ import annotations @@ -21,13 +21,15 @@ from typing import Annotated from fastapi import APIRouter, Depends from pydantic import BaseModel -from api.auth import require_admin_jwt +from api.auth import require_admin -#: All admin-plane routes share this router; the JWT check is attached -#: HERE so every admin route (current and later) requires it. +#: All admin-plane routes share this router; the ONE admin dependency +#: (service-token refusal + JWT verification + nestquest-admins group +#: gate) is attached HERE so every admin route (current and later) +#: requires it. router = APIRouter( prefix="/api/v1/admin", - dependencies=[Depends(require_admin_jwt)], + dependencies=[Depends(require_admin)], ) @@ -42,14 +44,14 @@ class AdminPingResponse(BaseModel): @router.get("/ping", summary="Admin plane probe") async def admin_ping( - claims: Annotated[dict[str, object], Depends(require_admin_jwt)], + claims: Annotated[dict[str, object], Depends(require_admin)], ) -> AdminPingResponse: - """Return 200 for a verified admin JWT (the plane's smoke probe). + """Return 200 for a verified JWT naming the nestquest-admins group. - Requires a valid Authentik JWT (checked by the router's shared - :func:`~api.auth.require_admin_jwt` dependency — the ONE token - check; this handler performs NO auth of its own). ``claims`` is - the SAME verified-claims dict that dependency produced (FastAPI's + Requires a valid Authentik JWT with the admin group (checked by the + router's shared :func:`~api.auth.require_admin` dependency — the + ONE check; this handler performs NO auth of its own). ``claims`` + is the SAME verified-claims dict that dependency produced (FastAPI's per-request dependency cache), so the token is verified exactly once per request. """ diff --git a/tests/test_api_admin_auth.py b/tests/test_api_admin_auth.py index ff161f5..e1a19ad 100644 --- a/tests/test_api_admin_auth.py +++ b/tests/test_api_admin_auth.py @@ -49,7 +49,7 @@ import pytest from cryptography.hazmat.primitives.asymmetric import rsa from api.app import create_app -from api.auth import JwksCache +from api.auth import ADMIN_GROUP, JwksCache from api.config import ApiConfig, ConfigError #: The configured OIDC settings the test apps run with (values a real @@ -67,6 +67,16 @@ SUBJECT = "admin-user-1" #: The uniform 401 detail every rejection must carry. REJECTION_DETAIL = "Missing or invalid admin credentials" +#: The 403 detail for the panel service token on an admin route. +PANEL_TOKEN_REJECTION_DETAIL = ( + "This credential type is not accepted for admin access" +) + +#: The 403 detail for a valid JWT without the admin group. +FORBIDDEN_DETAIL = ( + "Admin access requires membership in the nestquest-admins group" +) + def _generate_key() -> rsa.RSAPrivateKey: """Generate one local RSA key pair (no network, no fixtures).""" @@ -163,10 +173,20 @@ def _make_token( sub: str = SUBJECT, expires_in: int | None = 300, nbf_in: int | None = -1, + groups: object = (ADMIN_GROUP,), ) -> str: - """Mint one RS256 JWT the way Authentik would (its claims, our key).""" + """Mint one RS256 JWT the way Authentik would (its claims, our key). + + ``groups`` defaults to the admin group (what a real admin user's + token carries); tests pass a different value — or ``None`` for NO + groups claim at all — to prove the group gate rejects it. + """ now = int(time.time()) claims: dict[str, object] = {"iss": iss, "aud": aud, "sub": sub, "iat": now} + if groups is not None: + claims["groups"] = ( + list(groups) if isinstance(groups, (list, tuple)) else groups + ) if nbf_in is not None: claims["nbf"] = now + nbf_in if expires_in is not None: @@ -789,3 +809,108 @@ def test_from_env_requires_oidc_setting(env_var: str) -> None: env[env_var] = " " with pytest.raises(ConfigError): ApiConfig.from_env(env) + + +# --- the nestquest-admins group gate (task 0f2afb29) ------------------------- + + +async def test_admin_jwt_with_group_returns_200( + admin_client: httpx.AsyncClient, +) -> None: + """A valid JWT whose groups claim names nestquest-admins returns 200.""" + response = await admin_client.get( + "/api/v1/admin/ping", + headers={"Authorization": f"Bearer {_make_token()}"}, + ) + assert response.status_code == 200 + assert response.json() == {"status": "ok", "sub": SUBJECT} + + +async def test_non_admin_jwt_is_forbidden( + admin_client: httpx.AsyncClient, +) -> None: + """A valid JWT WITHOUT nestquest-admins in groups returns 403.""" + token = _make_token(groups=("some-other-group",)) + response = await admin_client.get( + "/api/v1/admin/ping", + headers={"Authorization": f"Bearer {token}"}, + ) + assert response.status_code == 403 + assert response.json()["detail"] == FORBIDDEN_DETAIL + + +async def test_empty_groups_is_forbidden( + admin_client: httpx.AsyncClient, +) -> None: + """A valid JWT with an EMPTY groups list returns 403.""" + token = _make_token(groups=()) + response = await admin_client.get( + "/api/v1/admin/ping", + headers={"Authorization": f"Bearer {token}"}, + ) + assert response.status_code == 403 + + +@pytest.mark.parametrize("groups", ["nestquest-admins", None]) +async def test_malformed_groups_is_forbidden( + admin_client: httpx.AsyncClient, groups: object +) -> None: + """A valid JWT with a non-list (or missing) groups claim returns 403. + + Even a plain STRING naming the group is not enough — the claim must + be a list, so a token crafted with a scalar ``groups`` cannot pass. + """ + token = _make_token(groups=groups) + response = await admin_client.get( + "/api/v1/admin/ping", + headers={"Authorization": f"Bearer {token}"}, + ) + assert response.status_code == 403 + assert response.json()["detail"] == FORBIDDEN_DETAIL + + +async def test_panel_service_token_is_refused_outright( + temp_db_path: str, +) -> None: + """The panel service token on an admin route is 403, never a JWT. + + 403 (not the uniform 401) proves the credential was refused as a + TYPE, not run through JWT verification — and the JWKS fetcher is + never called, proving it was not treated as a token to verify. + """ + runner = _AdminRunner(temp_db_path, _jwks_for(ADMIN_KEY, KID)) + async with runner as client: + response = await client.get( + "/api/v1/admin/ping", + headers={"Authorization": "Bearer admin-test-panel-token"}, + ) + assert response.status_code == 403 + assert response.json()["detail"] == PANEL_TOKEN_REJECTION_DETAIL + assert runner.fetch.calls == 0 + + +async def test_admin_router_is_wired_through_require_admin( + temp_db_path: str, +) -> None: + """The admin router's dependency IS the one require_admin dependency. + + The router-level dependency list carries + :func:`api.auth.require_admin`, so every admin route (current and + later) inherits the service-token refusal and the group gate. + """ + from api.auth import require_admin + from api.routes_admin import router as admin_router + + dependencies = [d.dependency for d in admin_router.dependencies] + assert require_admin in dependencies + + # Behavioral proof: the probe returns 200 for an admin JWT through + # the same dependency. + runner = _AdminRunner(temp_db_path, _jwks_for(ADMIN_KEY, KID)) + async with runner as client: + response = await client.get( + "/api/v1/admin/ping", + headers={"Authorization": f"Bearer {_make_token()}"}, + ) + assert response.status_code == 200 +