From e7e89ba7f94d38d098c766ca3f24e63ac123bcdc Mon Sep 17 00:00:00 2001 From: jakeross Date: Mon, 17 Aug 2026 09:33:22 -0700 Subject: [PATCH] chore(auth): harden Authentik authorization Six related fixes to the authorization layer, plus two defects surfaced while adding coverage for it. Fail fast on a misconfigured bypass. assert_auth_configuration() runs at app creation and refuses to boot when AUTHENTIK_DISABLE_AUTHENTICATION=1 outside MODE=development. The previous guard was per-request and keyed on MODE == "production", so a deploy with MODE unset served every endpoint anonymously and logged nothing. Collapses the two disagreeing reads of that variable (an import-time snapshot governing JWKS, a per-request read governing enforcement) into authentication_disabled(), and makes settings.mode a fresh env read so it no longer depends on which module called load_dotenv() first. Enforce the role hierarchy in code. authenticated() gains any_of=, and the tiers in core/dependencies.py list every group that satisfies them, so Admin satisfies an editor- or viewer-gated route. Previously the documented Admin > Editor > Viewer hierarchy existed nowhere in code and held only as long as whoever provisioned Authentik granted all three tiers; an Admin without the Viewer group was denied on every read. Families stay orthogonal -- general Admin confers nothing in AMP or Lexicon. Rename @public_route to @in_public_schema. It only ever controlled anonymous OpenAPI visibility, but read like an authorization decorator: two /thing routes carried it alongside a viewer_dependency. Removed from those, added to the routes that genuinely have no dependency (/ngwmn/*, polled by the federal NGWMN harvester, and /disclaimer, advertised as terms_of_service by both pygeoapi mounts). Verify the iss claim against AUTHENTIK_URL, accepting both trailing-slash spellings. Drop the dead scope= parameter, which was unused and would have substring-matched had it been used, since the OIDC scope claim is a space-delimited string rather than a list. Share one _decode() helper between the Depends path and the internal-OGC ASGI middleware, so a request decodes its token once instead of twice. Give the JWKS cache a TTL (AUTHENTIK_JWKS_TTL_SECONDS, default 3600) and force one refresh on an unrecognized kid. It was an unbounded lru_cache, so an Authentik key rotation 401'd every request until the process was redeployed. Insufficient groups now returns 403 rather than 401. This matters to OcotilloUI, whose axios-auth-refresh interceptor fires on 401 only: the old status sent an under-permissioned user through a token refresh and retry, and on to a forced logout. Two defects the new tests caught: POST /feedback was fully unauthenticated. It declared `_user=viewer_dependency` -- the dependency alias as a default value rather than a type annotation -- so FastAPI treated _user as a query parameter and never ran the dependency. Anyone could post arbitrary content that the server relays into Jira and Slack under its own credentials. public_openapi() could not see routes added via include_router. Those live inside opaque _IncludedRouter branches rather than being flattened into app.routes, so matching schema paths against app.routes by .path found only endpoints declared directly on the app, and the public schema contained /health alone. It now walks iter_route_contexts(), the same helper get_openapi() uses. tests/test_authorization.py covers all of it: an inventory test over every route's dependency tree against an explicit allowlist of intentionally anonymous routes (authorization is opt-in per endpoint, so nothing else notices an omission), agreement between that set and the anonymous OpenAPI schema, the group-membership logic, the startup guard, and JWKS caching. Co-Authored-By: Claude Opus 5 --- .env.example | 11 ++ CLAUDE.md | 23 +++ api/disclaimer.py | 2 + api/feedback.py | 2 +- api/ngwmn.py | 9 + api/thing.py | 3 - core/app.py | 55 ++++-- core/dependencies.py | 29 ++- core/factory.py | 9 + core/internal_ogc_auth.py | 14 +- core/permissions.py | 265 +++++++++++++++++-------- core/settings.py | 14 +- tests/test_authorization.py | 376 ++++++++++++++++++++++++++++++++++++ 13 files changed, 687 insertions(+), 125 deletions(-) create mode 100644 tests/test_authorization.py diff --git a/.env.example b/.env.example index 2e2bd4556..5fa1ad8ba 100644 --- a/.env.example +++ b/.env.example @@ -80,17 +80,28 @@ MODE=development # ENABLE_PG_CRON=0 # disable authentication (for development only) +# +# Honored ONLY when MODE=development. With any other MODE (including unset or +# "staging"), the app refuses to start -- core.permissions.assert_auth_configuration() +# raises AuthConfigurationError rather than serving every endpoint anonymously. AUTHENTIK_DISABLE_AUTHENTICATION=1 # erase and rebuild the database for step tests REBUILD_DB=1 # authentik +# AUTHENTIK_URL is both the JWKS base and the expected `iss` claim; trailing +# slash optional, both spellings are accepted. AUTHENTIK_URL= AUTHENTIK_CLIENT_ID= AUTHENTIK_AUTHORIZE_URL= AUTHENTIK_TOKEN_URL= +# How long a fetched JWKS document is trusted, in seconds (default 3600). +# An unrecognized `kid` forces one immediate refresh regardless, so this only +# bounds how long a revoked key stays usable. +# AUTHENTIK_JWKS_TTL_SECONDS=3600 + # feedback endpoint (POST /feedback) — bug reports and feature requests JIRA_BASE_URL=https://nmbgmr.atlassian.net diff --git a/CLAUDE.md b/CLAUDE.md index 77cb84105..88802a2a0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -155,8 +155,31 @@ The system uses **Authentik** for OAuth2 authentication with role-based access c - **Editor**: Can modify existing records (includes Viewer permissions) - **Admin**: Can create new records (includes Editor + Viewer permissions) +The hierarchy is enforced in code, via `authenticated(any_of=[...])` group lists — +`Admin` satisfies an editor- or viewer-gated route without needing all three +Authentik groups granted. + **AMP-Specific Roles**: `AMPAdmin`, `AMPEditor`, `AMPViewer` for legacy AMPAPI integration +**Role families are orthogonal**: general `Admin` confers nothing in the AMP or +Lexicon families. Only tiers *within* a family nest. + +**Authorization is opt-in per endpoint** — a `user: _dependency` parameter +in the signature, not a router-level `dependencies=[...]`. Omitting it produces a +fully public endpoint with no error. `tests/test_authorization.py` holds the +allowlist of intentionally anonymous routes and fails on anything else. Note the +annotation must be a *type annotation* (`user: viewer_dependency`), never a +default value (`user=viewer_dependency`) — the latter silently disables the +dependency and FastAPI treats it as a query parameter. + +**Development bypass**: `AUTHENTIK_DISABLE_AUTHENTICATION=1` is honored only when +`MODE=development`. Any other `MODE` (including unset) makes +`assert_auth_configuration()` abort startup. + +**`@in_public_schema`** (`core/app.py`) controls anonymous OpenAPI visibility +only — it grants no access and removes no dependency. Apply it only to routes +that genuinely have none. + ### Database Configuration The application supports two database modes (configured via `DB_DRIVER` in `.env`): diff --git a/api/disclaimer.py b/api/disclaimer.py index 2e94981d1..b0ec7e836 100644 --- a/api/disclaimer.py +++ b/api/disclaimer.py @@ -31,6 +31,7 @@ from fastapi import APIRouter, Query, Request from fastapi.responses import HTMLResponse, JSONResponse +from core.app import in_public_schema from core.disclaimer import ( DISCLAIMER_CONTACT_EMAIL, DISCLAIMER_PARAGRAPHS, @@ -84,6 +85,7 @@ def _render_html() -> str: ) +@in_public_schema @router.get( "/disclaimer", response_class=HTMLResponse, diff --git a/api/feedback.py b/api/feedback.py index 68f632b2f..ce3d3473a 100644 --- a/api/feedback.py +++ b/api/feedback.py @@ -225,7 +225,7 @@ def _build_slack_payload(payload: FeedbackCreate, jira_key: str, jira_url: str) @router.post("", response_model=FeedbackResponse) async def create_feedback( payload: FeedbackCreate, - _user=viewer_dependency, + _user: viewer_dependency, ): jira_base = os.environ["JIRA_BASE_URL"] jira_email = os.environ["JIRA_EMAIL"] diff --git a/api/ngwmn.py b/api/ngwmn.py index 7fc2e1d51..c954fe788 100644 --- a/api/ngwmn.py +++ b/api/ngwmn.py @@ -16,6 +16,7 @@ from fastapi import APIRouter from starlette.responses import Response +from core.app import in_public_schema from core.dependencies import session_dependency from services.ngwmn_helper import ( make_waterlevels_response, @@ -25,7 +26,13 @@ router = APIRouter(prefix="/ngwmn", tags=["NGWMN"]) +# These three routes are intentionally anonymous: the federal NGWMN harvester +# polls them without credentials. @in_public_schema documents that (and lists +# them in /openapi.json) so tests/test_authorization.py can tell an intentional +# public route from an endpoint that simply forgot its `user:` dependency. + +@in_public_schema @router.get( "/waterlevels/{pointid}", summary="Get waterlevels for a given pointid in the NGWMN format", @@ -35,6 +42,7 @@ def read_ngwmn_waterlevels(pointid: str, db: session_dependency): return Response(content=data, media_type="application/xml") +@in_public_schema @router.get( "/wellconstruction/{pointid}", summary="Get wellconstruction for a given pointid in the NGWMN format", @@ -44,6 +52,7 @@ def read_ngwmn_wellconstruction(pointid: str, db: session_dependency): return Response(content=data, media_type="application/xml") +@in_public_schema @router.get( "/lithology/{pointid}", summary="Get lithology for a given pointid in the NGWMN format", diff --git a/api/thing.py b/api/thing.py index baeed59e7..b1176071d 100644 --- a/api/thing.py +++ b/api/thing.py @@ -27,7 +27,6 @@ ) from api.pagination import CustomPage -from core.app import public_route from core.dependencies import ( session_dependency, admin_dependency, @@ -349,7 +348,6 @@ def get_thing_id_links( return paginate(query=sql, conn=session) -@public_route @router.get("/id-link/{link_id}", summary="Get thing links by link ID") def get_thing_id_links( user: viewer_dependency, @@ -362,7 +360,6 @@ def get_thing_id_links( return simple_get_by_id(session, ThingIdLink, link_id) -@public_route @router.get("", summary="Get all things", status_code=HTTP_200_OK) def get_things( user: viewer_dependency, diff --git a/core/app.py b/core/app.py index d14ccecf3..dd392f08b 100644 --- a/core/app.py +++ b/core/app.py @@ -27,6 +27,7 @@ get_swagger_ui_oauth2_redirect_html, ) from fastapi.openapi.utils import get_openapi +from fastapi.routing import iter_route_contexts from sqlalchemy import text from sqlalchemy.orm import Session @@ -123,27 +124,32 @@ def public_openapi(): routes=app.routes, ) - # Keep only operations where the endpoint function is marked public. + # Collect the operations whose endpoint carries @in_public_schema. + # + # This walks iter_route_contexts() rather than app.routes. Routes added + # via app.include_router() are not flattened into app.routes -- they + # live inside opaque _IncludedRouter branches -- so the previous + # `next(r for r in app.routes if r.path == path)` lookup matched + # nothing but the few endpoints declared directly on `app`, and + # silently dropped every decorated router route from the public schema. + # iter_route_contexts() is the same helper get_openapi() itself walks, + # so prefixes resolve identically to the paths in `schema`. + public_operations = set() + for route_context in iter_route_contexts(app.routes): + if not getattr(route_context.endpoint, "_in_public_schema", False): + continue + route_path = route_context.path_format or route_context.path + for route_method in route_context.methods or (): + public_operations.add((route_path, route_method.lower())) + new_paths = {} for path, path_item in schema["paths"].items(): new_methods = {} for method, operation in path_item.items(): - route = next( - ( - r - for r in app.routes - if getattr(r, "path", None) == path - and method.upper() in getattr(r, "methods", set()) - ), - None, - ) - if not route: + if (path, method.lower()) not in public_operations: continue - - endpoint = getattr(route, "endpoint", None) - if getattr(endpoint, "_is_public", False): - operation["security"] = [] - new_methods[method] = operation + operation["security"] = [] + new_methods[method] = operation if new_methods: new_paths[path] = new_methods @@ -224,7 +230,7 @@ async def warmup(): return {"status": "ok"} @app.get("/health", tags=["meta"]) - @public_route + @in_public_schema def health(response: Response, session: Session = Depends(get_db_session)): # Ping the database so a 200 actually proves PostGIS is reachable, not # just that the process is up. Uptime monitors / status pages assert on @@ -248,9 +254,18 @@ def health(response: Response, session: Session = Depends(get_db_session)): return app -def public_route(func): - """Mark a route as public for OpenAPI filtering.""" - setattr(func, "_is_public", True) +def in_public_schema(func): + """Advertise a route in the anonymous OpenAPI schema (/openapi.json). + + Schema visibility only -- this grants no access and removes no dependency. + It was previously named `public_route`, which read like an authorization + decorator; two `/thing` endpoints carried it *and* a `viewer_dependency`, + so the public schema advertised operations that 401 for anonymous callers. + + Apply it only to routes that genuinely have no auth dependency. + tests/test_authorization.py asserts the two sets match exactly. + """ + setattr(func, "_in_public_schema", True) return func diff --git a/core/dependencies.py b/core/dependencies.py index 6372804a9..09e7c3f79 100644 --- a/core/dependencies.py +++ b/core/dependencies.py @@ -34,26 +34,35 @@ Admin, can do everything Editor and Viewer can do + create new objects +That hierarchy is enforced here, by `any_of=` group lists rather than by +Authentik group membership overlap: an Admin-only account satisfies an +editor- or viewer-gated route because "Admin" appears in those lists. Before +this was explicit, `authenticated(permissions=["Viewer"])` required the +literal Viewer group, so the hierarchy held only as long as whoever +provisioned the Authentik groups granted all three tiers to every admin. + +The three families below are deliberately orthogonal -- general `Admin` does +not confer `AMPAdmin` or `LexiconAdmin`. Only tiers *within* a family nest. """ # General Purpose Authentication/Permissions ----------------------------------- -admin_function = authenticated(permissions=["Admin"]) -editor_function = authenticated(permissions=["Editor"]) -viewer_function = authenticated(permissions=["Viewer"]) +admin_function = authenticated(any_of=["Admin"]) +editor_function = authenticated(any_of=["Admin", "Editor"]) +viewer_function = authenticated(any_of=["Admin", "Editor", "Viewer"]) # AMP-Specific Authentication/Permissions -------------------------------------- -amp_admin_function = authenticated(permissions=["AMPAdmin"]) -amp_editor_function = authenticated(permissions=["AMPEditor"]) -amp_viewer_function = authenticated(permissions=["AMPViewer"]) +amp_admin_function = authenticated(any_of=["AMPAdmin"]) +amp_editor_function = authenticated(any_of=["AMPAdmin", "AMPEditor"]) +amp_viewer_function = authenticated(any_of=["AMPAdmin", "AMPEditor", "AMPViewer"]) # Lexicon-Specific Authentication/Permissions ---------------------------------- -lexicon_admin_function = authenticated(permissions=["LexiconAdmin"]) -lexicon_editor_function = authenticated(permissions=["LexiconEditor"]) +lexicon_admin_function = authenticated(any_of=["LexiconAdmin"]) +lexicon_editor_function = authenticated(any_of=["LexiconAdmin", "LexiconEditor"]) # OGC-Internal Authentication/Permissions -------------------------------------- @@ -63,7 +72,9 @@ # Testing-Specific Authentication/Permissions ---------------------------------- -no_permission_function = authenticated(permissions=["NoPermission"]) +# A group nobody is ever granted, so this dependency always 403s. Used to +# assert that group enforcement is actually wired up. +no_permission_function = authenticated(any_of=["NoPermission"]) # Permissions Dependencies ----------------------------------------------------- diff --git a/core/factory.py b/core/factory.py index f85b6f4cf..79a347e73 100644 --- a/core/factory.py +++ b/core/factory.py @@ -40,6 +40,15 @@ def initialize_runtime() -> None: def create_api_app(): initialize_runtime() + + # After initialize_runtime()'s load_dotenv(), so MODE and + # AUTHENTIK_DISABLE_AUTHENTICATION are both resolved. Raises + # AuthConfigurationError -- boot fails loudly rather than serving every + # endpoint anonymously. + from core.permissions import assert_auth_configuration + + assert_auth_configuration() + app = create_base_app() register_api_routes(app) from core.pygeoapi import mount_pygeoapi, mount_pygeoapi_internal diff --git a/core/internal_ogc_auth.py b/core/internal_ogc_auth.py index 8edd767cf..85f1ee3dd 100644 --- a/core/internal_ogc_auth.py +++ b/core/internal_ogc_auth.py @@ -29,7 +29,6 @@ """ import json -import os from starlette.types import ASGIApp, Receive, Scope, Send @@ -77,18 +76,15 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: await self.app(scope, receive, send) return - if int(os.environ.get("AUTHENTIK_DISABLE_AUTHENTICATION", 0)): - if settings.mode == "production": + if permissions.authentication_disabled(): + if settings.mode != permissions.BYPASS_ALLOWED_MODE: # HTTPException(424) (what core.permissions.authenticated() # raises for this same misconfiguration) means nothing from # raw ASGI code -- send the response directly so a - # misconfigured production box degrades to "internal mount - # always 424s" rather than crashing the worker. + # misconfigured box degrades to "internal mount always 424s" + # rather than crashing the worker. await _send_json( - send, - 424, - "Authentication is disabled in production mode. Set " - "AUTHENTIK_DISABLE_AUTHENTICATION=0 to enable authentication.", + send, 424, permissions.bypass_misconfiguration_detail() ) return await self.app(scope, receive, send) diff --git a/core/permissions.py b/core/permissions.py index fec27d37f..ad3406f41 100644 --- a/core/permissions.py +++ b/core/permissions.py @@ -14,13 +14,13 @@ # limitations under the License. # =============================================================================== import os -from functools import lru_cache -from typing import Optional, List, Union, cast, Callable +import threading +import time +from typing import Optional, List, Sequence, Tuple, Union, cast, Callable import httpx from fastapi import Depends, HTTPException, status from fastapi.security import HTTPAuthorizationCredentials, OAuth2AuthorizationCodeBearer -from fastapi.security import OAuth2PasswordBearer from jose import jwt from jose.exceptions import JWTError from jwt.algorithms import RSAAlgorithm @@ -29,33 +29,133 @@ from core.settings import settings -AUTHENTIK_ISSUER = os.environ.get("AUTHENTIK_URL") ALGORITHMS = ["RS256"] -jwks = {} -auth_disabled = int(os.environ.get("AUTHENTIK_DISABLE_AUTHENTICATION", 0)) -if AUTHENTIK_ISSUER and not auth_disabled: - JWKS_URL = f"{AUTHENTIK_ISSUER}jwks/" +# The only MODE in which AUTHENTIK_DISABLE_AUTHENTICATION=1 is honored. +# assert_auth_configuration() refuses to boot anywhere else, so a box with an +# unset or mislabeled MODE can never come up with authentication switched off. +BYPASS_ALLOWED_MODE = "development" -@lru_cache(maxsize=1) -def get_jwks(): - if not AUTHENTIK_ISSUER or auth_disabled: +# How long a fetched JWKS document is trusted. Authentik rotates its signing +# keys; the cache used to be an unbounded lru_cache, so a rotation 401'd every +# request until the process was redeployed. get_public_key() also forces one +# refresh on an unrecognized `kid`, which covers rotations inside the window. +JWKS_TTL_SECONDS = int(os.environ.get("AUTHENTIK_JWKS_TTL_SECONDS", "3600")) + + +def _issuer() -> str: + """Authentik issuer URL, read lazily so it survives late load_dotenv().""" + return (os.environ.get("AUTHENTIK_URL") or "").strip() + + +def _accepted_issuers() -> Tuple[str, ...]: + """Issuer values accepted for the `iss` claim. + + Authentik's issuer is the provider URL, which operators configure with or + without a trailing slash depending on where they copied it from. Accept + both spellings rather than making token validation depend on that. + """ + issuer = _issuer() + if not issuer: + return () + return (issuer.rstrip("/"), issuer.rstrip("/") + "/") + + +def authentication_disabled() -> bool: + """Whether the development authentication bypass is switched on. + + Read fresh from the environment on every call. This used to be an + import-time snapshot (`auth_disabled`) that the per-request check in + authenticated() did not share: flipping the variable after import left + JWKS fetching disabled while token verification stayed live, so every + request failed with "Invalid signing key" instead of either enforcing or + bypassing cleanly. + """ + raw = (os.environ.get("AUTHENTIK_DISABLE_AUTHENTICATION") or "0").strip() + try: + return bool(int(raw)) + except ValueError: + return raw.lower() in {"true", "yes", "on"} + + +class AuthConfigurationError(RuntimeError): + """Raised at startup when the auth bypass is enabled outside development.""" + + +def bypass_misconfiguration_detail() -> str: + return ( + "AUTHENTIK_DISABLE_AUTHENTICATION is enabled but MODE is " + f"{settings.mode or ''!r}. The bypass is only permitted when " + f"MODE={BYPASS_ALLOWED_MODE!r}. Set " + "AUTHENTIK_DISABLE_AUTHENTICATION=0, or set MODE=development." + ) + + +def assert_auth_configuration() -> None: + """Fail fast when the app is configured to serve traffic without auth. + + Called from core.factory.create_api_app() after load_dotenv(). The old + guard was per-request and keyed on `settings.mode == "production"`, so a + deploy with MODE unset served every endpoint anonymously and logged + nothing. Two independent variables had to be right; now a wrong one stops + the process at boot. + """ + if authentication_disabled() and settings.mode != BYPASS_ALLOWED_MODE: + raise AuthConfigurationError(bypass_misconfiguration_detail()) + + +_jwks_lock = threading.Lock() +_jwks_cache: dict = {"payload": None, "fetched_at": 0.0} + + +def reset_jwks_cache() -> None: + """Drop the cached JWKS. Test hook and manual-invalidation escape hatch.""" + with _jwks_lock: + _jwks_cache["payload"] = None + _jwks_cache["fetched_at"] = 0.0 + + +def get_jwks(force_refresh: bool = False) -> dict: + if not _issuer() or authentication_disabled(): return {} - resp = httpx.get(JWKS_URL, timeout=10.0) + if not force_refresh: + with _jwks_lock: + cached = _jwks_cache["payload"] + age = time.monotonic() - _jwks_cache["fetched_at"] + if cached is not None and age < JWKS_TTL_SECONDS: + return cached + + resp = httpx.get(f"{_issuer().rstrip('/')}/jwks/", timeout=10.0) resp.raise_for_status() - return resp.json() + payload = resp.json() + with _jwks_lock: + _jwks_cache["payload"] = payload + _jwks_cache["fetched_at"] = time.monotonic() + return payload -def get_public_key(token): - unverified_header = jwt.get_unverified_header(token) - for key in get_jwks().get("keys", []): - if key["kid"] == unverified_header["kid"]: - return RSAAlgorithm.from_jwk(key) - raise HTTPException(status_code=401, detail="Invalid signing key") +def _find_signing_key(jwks: dict, kid: Optional[str]) -> Optional[dict]: + if not kid: + return None + for key in jwks.get("keys", []): + if key.get("kid") == kid: + return key + return None -oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") + +def get_public_key(token): + kid = jwt.get_unverified_header(token).get("kid") + + key = _find_signing_key(get_jwks(), kid) + if key is None: + # Unknown kid: Authentik may have rotated inside the TTL window. + # Refetch once before rejecting an otherwise valid token. + key = _find_signing_key(get_jwks(force_refresh=True), kid) + if key is None: + raise HTTPException(status_code=401, detail="Invalid signing key") + return RSAAlgorithm.from_jwk(key) TokenType = Union[str, HTTPAuthorizationCredentials] @@ -69,84 +169,93 @@ def get_public_key(token): ) +def authorize_groups( + payload: dict, + require_all: Optional[Sequence[str]] = None, + require_any: Optional[Sequence[str]] = None, +) -> bool: + """Check a decoded token's `groups` claim against a group requirement. + + `require_all` demands every listed group; `require_any` demands at least + one. Role tiers in core/dependencies.py use `require_any` so that an Admin + satisfies an Editor- or Viewer-gated route. The old check was all-of only, + which meant the documented Admin > Editor > Viewer hierarchy existed + nowhere in code -- it worked solely because operators happened to grant + overlapping groups in Authentik, and an Admin without the Viewer group got + a 403 on every read. + """ + groups = payload.get("groups") or [] + if require_all and not all(group in groups for group in require_all): + return False + if require_any and not any(group in groups for group in require_any): + return False + return True + + def authenticated( optional: bool = False, - scope: Optional[List[str]] = None, permissions: Optional[List[str]] = None, + any_of: Optional[List[str]] = None, ): + """Build a FastAPI dependency enforcing a bearer token and group membership. - def _authenicated( + `permissions` requires every listed Authentik group, `any_of` requires at + least one. Returns the decoded token payload on success so endpoints can + read claims; returns True when the development bypass is active. + """ + + def _authenticated( request: Request, response: Response, token: TokenType = Depends(cast(Callable, scheme)), ): - # def _authenicated(request: Request, response: Response): - # def _authenicated(): - """ - A placeholder for the authentication logic. - This function should check if the user is authenticated and has the required permissions. - If `optional` is True, it should allow unauthenticated access. - """ - - if int(os.environ.get("AUTHENTIK_DISABLE_AUTHENTICATION", 0)): - if settings.mode == "production": + if authentication_disabled(): + # assert_auth_configuration() already rejected this combination at + # startup; this is the belt for a variable flipped at runtime. + if settings.mode != BYPASS_ALLOWED_MODE: raise HTTPException( status_code=status.HTTP_424_FAILED_DEPENDENCY, - detail="Authentication is disabled in production mode. Set AUTHENTIK_DISABLE_AUTHENTICATION=0 to enable authentication.", + detail=bypass_misconfiguration_detail(), ) return True - if optional and not token: - return True - - # Here you would typically check the token against your authentication system - # and verify the user's permissions. - - if not token or not verify_token(token, scope, permissions): + if not token: + if optional: + return True raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized" ) - # this is a placeholder for the actual authentication logic - return _get_token_payload(token) if token else None + # Decoded once and reused. The previous flow decoded the JWT twice per + # request: verify_token() decoded to read groups, then the caller + # decoded again to build the return value. + payload = _get_token_payload(token) - return _authenicated + if not authorize_groups(payload, require_all=permissions, require_any=any_of): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, detail="Forbidden" + ) + return payload -def verify_token( - token: TokenType, scope: Optional[List[str]], permissions: Optional[List[str]] -) -> bool: - """ - Placeholder function to verify the token. - This should contain the logic to check if the token is valid and has the required permissions. - """ - # Implement your token verification logic here + return _authenticated - payload = _get_token_payload(token) - # Optionally check scopes and permissions in payload - if scope: - if not all(s in payload.get("scope", []) for s in scope): - return False - if permissions: - if not all(p in payload.get("groups", []) for p in permissions): - return False - return True +def _decode(token: str) -> dict: + """Verify signature, audience, and issuer, returning the claims.""" + return jwt.decode( + token, + get_public_key(token), + algorithms=ALGORITHMS, + audience=os.environ.get("AUTHENTIK_CLIENT_ID"), # Authentik application + issuer=_accepted_issuers() or None, + ) -def _get_token_payload(token: str = Depends(oauth2_scheme)): +def _get_token_payload(token: str) -> dict: try: - public_key = get_public_key(token) - payload = jwt.decode( - token, - public_key, - algorithms=ALGORITHMS, - audience=os.environ.get( - "AUTHENTIK_CLIENT_ID" - ), # Must match Authentik application - ) - return payload - except JWTError as e: + return _decode(token) + except JWTError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", @@ -170,17 +279,11 @@ class TokenInvalid(Exception): def decode_token_payload(token: str) -> dict: - """Same JWT verification as _get_token_payload (get_public_key/JWKS/ - jwt.decode), but raises TokenInvalid instead of HTTPException(401). + """Same JWT verification as _get_token_payload (shared _decode helper), + but raises TokenInvalid instead of HTTPException(401). """ try: - public_key = get_public_key(token) - return jwt.decode( - token, - public_key, - algorithms=ALGORITHMS, - audience=os.environ.get("AUTHENTIK_CLIENT_ID"), - ) + return _decode(token) except (JWTError, HTTPException) as e: raise TokenInvalid(str(e)) from e diff --git a/core/settings.py b/core/settings.py index 95ea93b68..c29c5719c 100644 --- a/core/settings.py +++ b/core/settings.py @@ -30,8 +30,18 @@ def _resolve_version() -> str: class Settings: version = _resolve_version() - def __init__(self): - self.mode = os.getenv("MODE", "") # Default mode + @property + def mode(self) -> str: + """Deployment mode, read fresh from the environment on every access. + + This used to be snapshotted in __init__. Settings() is instantiated + while core.app is imported, which happens before core.factory calls + load_dotenv() -- so whether MODE was visible depended on which module + happened to call load_dotenv() first. Reading it lazily makes the + value independent of import order, which matters because + core.permissions gates the authentication bypass on it. + """ + return os.getenv("MODE", "") def get_enum(self, name: str): if name == "MODE": diff --git a/tests/test_authorization.py b/tests/test_authorization.py new file mode 100644 index 000000000..97608ae9c --- /dev/null +++ b/tests/test_authorization.py @@ -0,0 +1,376 @@ +# =============================================================================== +# Copyright 2026 ross +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# =============================================================================== +"""Authorization wiring tests. + +CI runs with AUTHENTIK_DISABLE_AUTHENTICATION=1, so no test here can exercise +a real Authentik token. These tests cover the parts that are verifiable +without one: + +* which routes have no authentication dependency at all (an inventory test -- + authorization is opt-in per endpoint, so a forgotten `user:` parameter + silently publishes an endpoint and nothing else would catch it), +* that the anonymous OpenAPI schema advertises exactly those routes, +* the pure group-membership logic behind the role tiers, +* the startup guard on the development bypass. +""" + +import pytest +from fastapi.routing import iter_route_contexts + +from core import dependencies, permissions +from tests import client + +# Routes that are allowed to have no authentication dependency. Adding an entry +# here is a deliberate decision to publish an endpoint anonymously -- it is not +# a formality to satisfy the test. +# +# /health, /_ah/warmup uptime monitors and App Engine warmup +# /docs-auth, /openapi-auth Swagger UI and schema, no data +# /disclaimer advertised as terms_of_service by both pygeoapi +# mounts, so OGC clients fetch it uncredentialed +# /ngwmn/* polled by the federal NGWMN harvester +# +# Not listed, because they never reach this scan: +# /openapi.json, /docs, /redoc are bare Starlette routes with no dependant. +# /ogcapi is a Mount -- anonymous by design; /ogcapi-internal is gated by +# core.internal_ogc_auth.InternalOGCAuthMiddleware, outside Depends(). +EXPECTED_ANONYMOUS_ROUTES = { + ("GET", "/health"), + ("GET", "/_ah/warmup"), + ("GET", "/docs-auth"), + ("GET", "/docs-auth/oauth2-redirect"), + ("GET", "/openapi-auth.json"), + ("GET", "/disclaimer"), + ("GET", "/ngwmn/waterlevels/{pointid}"), + ("GET", "/ngwmn/wellconstruction/{pointid}"), + ("GET", "/ngwmn/lithology/{pointid}"), +} + +# Every dependency callable built by core.permissions.authenticated(). +AUTH_DEPENDENCY_CALLABLES = frozenset( + { + dependencies.admin_function, + dependencies.editor_function, + dependencies.viewer_function, + dependencies.amp_admin_function, + dependencies.amp_editor_function, + dependencies.amp_viewer_function, + dependencies.lexicon_admin_function, + dependencies.lexicon_editor_function, + dependencies.no_permission_function, + } +) + + +def _has_auth_dependency(dependant) -> bool: + """Walk a route's dependency tree looking for an auth dependency.""" + if dependant.call in AUTH_DEPENDENCY_CALLABLES: + return True + return any(_has_auth_dependency(sub) for sub in dependant.dependencies) + + +def _anonymous_routes() -> set: + """Every (method, path) with no authentication dependency. + + Walks iter_route_contexts() rather than app.routes: routes registered via + include_router() are not flattened into app.routes in this FastAPI version, + so a plain scan would see only the endpoints declared directly on `app` and + this test would pass while reporting on ~5 of ~130 routes. + """ + found = set() + for route_context in iter_route_contexts(client.app.routes): + dependant = getattr(route_context, "dependant", None) + if dependant is None: + # Not an APIRoute (Mounts, bare Starlette routes such as /docs). + continue + if _has_auth_dependency(dependant): + continue + path = route_context.path_format or route_context.path + for method in route_context.methods or (): + if method in ("HEAD", "OPTIONS"): + continue + found.add((method, path)) + return found + + +def test_no_unintended_anonymous_routes(): + """Every route without an auth dependency is one we chose to publish. + + Authorization is declared per endpoint as a `user: _dependency` + parameter, not at the router level, so omitting it produces a fully public + endpoint with no error anywhere. This test is the only thing that notices. + """ + unexpected = _anonymous_routes() - EXPECTED_ANONYMOUS_ROUTES + assert not unexpected, ( + "These routes have no authentication dependency. Add a `user: " + "_dependency` parameter, or add them to " + f"EXPECTED_ANONYMOUS_ROUTES if they are meant to be public: " + f"{sorted(unexpected)}" + ) + + +def test_expected_anonymous_routes_still_exist(): + """Keeps EXPECTED_ANONYMOUS_ROUTES from rotting into a stale allowlist.""" + stale = EXPECTED_ANONYMOUS_ROUTES - _anonymous_routes() + assert not stale, ( + "EXPECTED_ANONYMOUS_ROUTES lists routes that no longer exist or now " + f"require authentication -- remove them: {sorted(stale)}" + ) + + +def test_public_schema_advertises_only_anonymous_routes(): + """@in_public_schema must not advertise an authenticated operation. + + Two /thing routes used to carry the decorator (then named @public_route) + alongside a viewer_dependency, so the anonymous schema described endpoints + that 401 for anonymous callers. + """ + schema = client.get("/openapi.json").json() + advertised = { + (method.upper(), path) + for path, item in schema["paths"].items() + for method in item + } + assert advertised <= EXPECTED_ANONYMOUS_ROUTES, ( + "The anonymous OpenAPI schema advertises routes that require " + "authentication. Remove @in_public_schema from them: " + f"{sorted(advertised - EXPECTED_ANONYMOUS_ROUTES)}" + ) + + +# Group membership logic ------------------------------------------------------- + + +@pytest.mark.parametrize( + "groups, expected", + [ + (["Admin"], True), + (["Editor"], True), + (["Viewer"], True), + (["AMPAdmin"], False), + ([], False), + ], +) +def test_admin_satisfies_viewer_tier(groups, expected): + """Admin > Editor > Viewer is enforced in code, not by Authentik overlap.""" + assert ( + permissions.authorize_groups( + {"groups": groups}, require_any=["Admin", "Editor", "Viewer"] + ) + is expected + ) + + +@pytest.mark.parametrize( + "groups, expected", + [ + (["Admin"], True), + (["Editor"], False), + (["Viewer"], False), + ], +) +def test_admin_tier_does_not_accept_lower_roles(groups, expected): + assert ( + permissions.authorize_groups({"groups": groups}, require_any=["Admin"]) + is expected + ) + + +def test_role_families_stay_orthogonal(): + """General Admin confers nothing in the AMP or Lexicon families.""" + payload = {"groups": ["Admin"]} + assert not permissions.authorize_groups( + payload, require_any=["AMPAdmin", "AMPEditor", "AMPViewer"] + ) + assert not permissions.authorize_groups( + payload, require_any=["LexiconAdmin", "LexiconEditor"] + ) + + +def test_require_all_demands_every_group(): + assert permissions.authorize_groups( + {"groups": ["Admin", "AMPAdmin"]}, require_all=["Admin", "AMPAdmin"] + ) + assert not permissions.authorize_groups( + {"groups": ["Admin"]}, require_all=["Admin", "AMPAdmin"] + ) + + +def test_missing_groups_claim_denies(): + """A token with no `groups` claim must not satisfy a role requirement.""" + assert not permissions.authorize_groups({}, require_any=["Viewer"]) + assert not permissions.authorize_groups({"groups": None}, require_any=["Viewer"]) + + +# Bypass configuration guard --------------------------------------------------- + + +@pytest.fixture +def auth_env(monkeypatch): + """Set MODE and AUTHENTIK_DISABLE_AUTHENTICATION for one test.""" + + def _set(mode, disabled): + monkeypatch.setenv("MODE", mode) + monkeypatch.setenv("AUTHENTIK_DISABLE_AUTHENTICATION", disabled) + + return _set + + +@pytest.mark.parametrize("mode", ["production", "staging", "", "Development"]) +def test_bypass_outside_development_refuses_to_boot(auth_env, mode): + """The bypass is honored in development only. + + The old guard only rejected MODE=="production", so a deploy with MODE + unset served every endpoint anonymously and said nothing about it. + """ + auth_env(mode, "1") + with pytest.raises(permissions.AuthConfigurationError) as exc: + permissions.assert_auth_configuration() + assert "AUTHENTIK_DISABLE_AUTHENTICATION" in str(exc.value) + + +def test_bypass_allowed_in_development(auth_env): + auth_env("development", "1") + permissions.assert_auth_configuration() + + +@pytest.mark.parametrize("mode", ["production", "staging", "", "development"]) +def test_any_mode_boots_with_auth_enabled(auth_env, mode): + auth_env(mode, "0") + permissions.assert_auth_configuration() + + +@pytest.mark.parametrize( + "raw, expected", + [ + ("1", True), + ("0", False), + ("", False), + ("true", True), + ("TRUE", True), + ("on", True), + ("no", False), + ("garbage", False), + ], +) +def test_authentication_disabled_parsing(monkeypatch, raw, expected): + """A non-numeric value must not crash the guard into a bypass.""" + monkeypatch.setenv("AUTHENTIK_DISABLE_AUTHENTICATION", raw) + assert permissions.authentication_disabled() is expected + + +def test_authentication_disabled_defaults_to_enforcing(monkeypatch): + monkeypatch.delenv("AUTHENTIK_DISABLE_AUTHENTICATION", raising=False) + assert permissions.authentication_disabled() is False + + +def test_mode_is_read_fresh_from_environment(monkeypatch): + """settings.mode used to be an import-time snapshot, so whether MODE was + visible depended on which module called load_dotenv() first.""" + from core.settings import settings + + monkeypatch.setenv("MODE", "sentinel-mode") + assert settings.mode == "sentinel-mode" + + +# JWKS caching ----------------------------------------------------------------- + + +def test_jwks_cache_expires(monkeypatch): + """A TTL'd cache, so an Authentik key rotation does not require a redeploy. + + The cache was an unbounded lru_cache: once a rotation invalidated the + cached keys every request 401'd with "Invalid signing key" until the + process restarted. + """ + permissions.reset_jwks_cache() + monkeypatch.setenv("AUTHENTIK_URL", "https://authentik.example/application/o/x/") + monkeypatch.setenv("AUTHENTIK_DISABLE_AUTHENTICATION", "0") + + fetches = [] + + class _Resp: + def raise_for_status(self): + pass + + def json(self): + return {"keys": [{"kid": f"k{len(fetches)}"}]} + + def _fake_get(url, **kwargs): + fetches.append(url) + return _Resp() + + monkeypatch.setattr(permissions.httpx, "get", _fake_get) + + clock = {"now": 1000.0} + monkeypatch.setattr(permissions.time, "monotonic", lambda: clock["now"]) + + permissions.get_jwks() + permissions.get_jwks() + assert len(fetches) == 1, "within the TTL the cached document is reused" + assert fetches[0] == "https://authentik.example/application/o/x/jwks/" + + clock["now"] += permissions.JWKS_TTL_SECONDS + 1 + permissions.get_jwks() + assert len(fetches) == 2, "past the TTL the document is refetched" + + permissions.reset_jwks_cache() + + +def test_jwks_not_fetched_when_bypass_active(monkeypatch): + permissions.reset_jwks_cache() + monkeypatch.setenv("AUTHENTIK_URL", "https://authentik.example/application/o/x/") + monkeypatch.setenv("AUTHENTIK_DISABLE_AUTHENTICATION", "1") + + def _explode(*args, **kwargs): + raise AssertionError("JWKS must not be fetched while auth is bypassed") + + monkeypatch.setattr(permissions.httpx, "get", _explode) + assert permissions.get_jwks() == {} + + +def test_accepted_issuers_covers_both_slash_spellings(monkeypatch): + """The `iss` claim is now verified; operators configure AUTHENTIK_URL with + and without a trailing slash, so both spellings must be accepted.""" + monkeypatch.setenv("AUTHENTIK_URL", "https://authentik.example/application/o/x/") + assert set(permissions._accepted_issuers()) == { + "https://authentik.example/application/o/x", + "https://authentik.example/application/o/x/", + } + + monkeypatch.setenv("AUTHENTIK_URL", "https://authentik.example/application/o/x") + assert set(permissions._accepted_issuers()) == { + "https://authentik.example/application/o/x", + "https://authentik.example/application/o/x/", + } + + +def test_accepted_issuers_empty_when_unconfigured(monkeypatch): + """Empty tuple, which _decode() passes to jose as issuer=None.""" + monkeypatch.delenv("AUTHENTIK_URL", raising=False) + assert permissions._accepted_issuers() == () + + +def test_dead_scope_parameter_is_gone(): + """`scope=` was never used and was wrong if it had been: the OIDC scope + claim is a space-delimited string, so `s in payload["scope"]` was substring + matching.""" + import inspect + + assert "scope" not in inspect.signature(permissions.authenticated).parameters + + +# ============= EOF =============================================