chore(auth): harden Authentik authorization

.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

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: <role>_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`):

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,

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

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

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,

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
 
 

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

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

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)