Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 94 additions & 1 deletion horizon/tests/test_route_auth_audit.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,10 @@
concern (see ``_warn_if_opal_verifier_disabled`` in horizon/pdp.py).
"""

from typing import Annotated

import pytest
from fastapi import Depends, FastAPI
from fastapi import APIRouter, Depends, FastAPI
from fastapi.dependencies.utils import get_flat_dependant
from fastapi.routing import APIRoute
from horizon.authentication import PUBLIC_ROUTE_PATHS, enforce_pdp_token
Expand Down Expand Up @@ -134,3 +136,94 @@ def test_audit_flags_a_mounted_subapp():
app = FastAPI()
app.mount("/sub", FastAPI())
assert any("/sub" in row for row in _find_unprotected_routes(app))


# PUBLIC_ROUTE_PATHS entries the audit-built app legitimately does not mount. ``/scalar`` is
# registered in ``PermitPDP.__init__`` *after* ``_configure_api_routes`` (see the comment
# above PUBLIC_ROUTE_PATHS in horizon/authentication.py), so ``MockPermitPDP`` - which stops
# at ``_configure_api_routes`` - never sees it, yet it is a real public route in production.
#
# TRADE-OFF (accepted): an entry listed here is exempted from the dead-entry check below, so
# if its real route is ever deleted while the PUBLIC_ROUTE_PATHS entry stays, that now-dead
# entry will NOT be flagged - and a later ungated route re-claiming the path would read as
# public. Keep this set as small as possible. The exemption-free alternative is to make
# MockPermitPDP mirror the post-_configure_api_routes registration so the path is really
# mounted; we keep the exemption for now to stay test-only and not reshape production wiring.
ALLOWLIST_ENTRIES_NOT_IN_AUDIT_APP: frozenset[str] = frozenset({"/scalar"})


def test_allowlist_has_no_dead_entries():
"""Every PUBLIC_ROUTE_PATHS entry must match a mounted route.

A dead allowlist entry is a pre-authorised hole: it exempts a path from the audit today
and silently waves through whatever route later claims that path. Matching by set
membership over ``getattr(route, "path", ...)`` covers the framework Swagger/OpenAPI
routes (plain starlette ``Route``s, not ``APIRoute``s) and paths shared across methods.
"""
mounted = {getattr(route, "path", None) for route in _sidecar._app.routes}
dead = PUBLIC_ROUTE_PATHS - mounted - ALLOWLIST_ENTRIES_NOT_IN_AUDIT_APP
assert dead == set(), (
f"PUBLIC_ROUTE_PATHS entries that match no route on the audit-built app: {sorted(dead)}. "
"Either the entry is dead (remove it, or fix the path if a route was renamed), or it is a "
"real route registered in PermitPDP.__init__ after _configure_api_routes (like /scalar) "
"that MockPermitPDP never reaches - in which case add it to ALLOWLIST_ENTRIES_NOT_IN_AUDIT_APP."
)
# Keep the exemption honest: if ``/scalar`` ever becomes visible to the audit-built app,
# it no longer needs the special case and must be dropped from the set above.
assert ALLOWLIST_ENTRIES_NOT_IN_AUDIT_APP & mounted == set(), (
"An ALLOWLIST_ENTRIES_NOT_IN_AUDIT_APP entry is now mounted on the audit app; drop "
"it from that set so it is covered by the dead-entry check like every other path."
)


def test_router_level_dependencies_surface_in_flat_dependant():
"""Empirical FastAPI contract the whole audit rests on (proven on 0.125.0; pin is loose).

The audit detects gates by walking ``get_flat_dependant(route.dependant)``. That only
works if a dependency attached at ``include_router(dependencies=[...])`` propagates into
each child route's dependant, and if nested sub-dependencies (e.g. OPAL's
``require_listener_token`` wrapping the authenticator) are flattened. If a future FastAPI
bump changes that, router-gated routes would read as unprotected and every real run would
fail for the wrong reason - so pin the assumption here, where the failure is legible.
"""

def fake_gate(): # router-level gate
pass

def inner_gate(): # reachable ONLY as wrapper's sub-dependency - see the assertion below
pass

# Annotated-Depends (the repo's own convention, see horizon/authentication.py) keeps the
# sub-dependency in the annotation rather than the argument default - idiomatic FastAPI and
# B008-clean. inner_gate is nested one level under wrapper and attached nowhere else, so the
# nested-gated assertion genuinely exercises get_flat_dependant's recursion instead of
# passing on a directly-attached copy.
def wrapper(_: Annotated[None, Depends(inner_gate)] = None):
pass

router = APIRouter()

@router.get("/router-gated")
async def _router_gated():
return {}

@router.get("/nested-gated", dependencies=[Depends(wrapper)])
async def _nested_gated():
return {}

app = FastAPI()
app.include_router(router, dependencies=[Depends(fake_gate)])
by_path = {route.path: route for route in app.routes if isinstance(route, APIRoute)}

assert "fake_gate" in _route_auth_gates(by_path["/router-gated"]), (
"include_router(dependencies=...) no longer surfaces in get_flat_dependant - the "
"route audit's gate detection is broken for router-level gates; review it before "
"trusting a green run on this FastAPI version."
)
# inner_gate reaches this route ONLY through wrapper (wrapper is the route-level dep;
# fake_gate is router-level). If get_flat_dependant stops recursing into sub-dependencies,
# inner_gate drops out and this fails - the exact regression the assertion exists to pin.
assert {"wrapper", "inner_gate"} <= _route_auth_gates(by_path["/nested-gated"]), (
"nested Depends() is no longer flattened by get_flat_dependant - closure-wrapped "
"gates (e.g. OPAL's require_listener_token) would go undetected by the audit."
)
Loading