Admin routes: require_admin gates nestquest-admins group, refuses panel token

WHY: the admin plane was authenticated (any valid JWT passed) but not
authorized, and the panel service token would have been run through
JWT verification like any other credential. require_admin is now the
ONE dependency every admin route uses: it refuses the panel service
token outright with 403 BEFORE any JWT parsing (a refused credential
type, not an invalid one — a 401 challenge would imply a retry could
ever succeed), then verifies the JWT via the existing require_admin_jwt
path (uniform 401 unchanged) and requires the groups claim to name
nestquest-admins (403 for a missing/empty/non-list claim or one
without the group).

api/auth.py

@@ -35,10 +35,13 @@ 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
 
@@ -70,6 +73,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 +274,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 +412,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 +424,59 @@ 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
+                and credential.strip() == config.panel_token
+            ):
+                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",
+]

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.
     """

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
+