feat: add ENFORCE_OPERATIONAL_ROUTE_AUTH rollout toggle (PER-15243)

horizon/authentication.py

@@ -1,8 +1,11 @@
 import hmac
+import threading
+import time
 from typing import Annotated
 
-from fastapi import Depends, HTTPException, status
+from fastapi import Depends, HTTPException, Request, status
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from loguru import logger
 
 from horizon.config import MOCK_API_KEY, sidecar_config
 from horizon.startup.api_keys import get_env_api_key
@@ -65,6 +68,84 @@ def enforce_pdp_token(credentials: PdpCredentials = None):
         raise HTTPException(status.HTTP_401_UNAUTHORIZED, detail="Invalid PDP token")
 
 
+# Rate-limit for the warn-and-allow path of enforce_pdp_token_operational. While
+# ENFORCE_OPERATIONAL_ROUTE_AUTH is off, every unauthenticated request to a governed route would
+# otherwise emit one WARNING - which on the high-QPS /kong decision endpoint floods the synchronous
+# stdout log sink (and adds a blocking write to the request path). Instead we coalesce to one line
+# per route per interval that reports how many would-be rejections were allowed; that count is also
+# the rollout signal - watch it fall to zero, then flip the flag on. The gate runs in a threadpool
+# (it is a sync dependency), so the shared counters are guarded by a plain threading.Lock.
+_OPERATIONAL_WARN_INTERVAL_SECONDS = 60.0
+_operational_warn_lock = threading.Lock()
+# path -> (would-be rejections accumulated since the last emitted line, monotonic time of that emit)
+_operational_warn_state: dict[str, tuple[int, float]] = {}
+
+
+def reset_operational_warn_throttle() -> None:
+    """Clear the throttle state. For tests, whose asserted warnings must not be suppressed by a prior test."""
+    with _operational_warn_lock:
+        _operational_warn_state.clear()
+
+
+def _operational_warn_should_emit(path: str) -> tuple[bool, int]:
+    """Decide whether to emit the warn-and-allow line for ``path`` now, coalescing bursts.
+
+    Returns ``(emit_now, count_since_last_emit)``. Every call counts as one would-be rejection; a
+    line is emitted on the first hit for a path and then at most once per
+    ``_OPERATIONAL_WARN_INTERVAL_SECONDS``, carrying the number of rejections coalesced into it.
+    """
+    now = time.monotonic()
+    with _operational_warn_lock:
+        state = _operational_warn_state.get(path)
+        if state is None:
+            _operational_warn_state[path] = (0, now)
+            return True, 1
+        pending, last_emit = state
+        pending += 1
+        if now - last_emit >= _OPERATIONAL_WARN_INTERVAL_SECONDS:
+            _operational_warn_state[path] = (0, now)
+            return True, pending
+        _operational_warn_state[path] = (pending, last_emit)
+        return False, pending
+
+
+def enforce_pdp_token_operational(request: Request, credentials: PdpCredentials = None):
+    """PDP-token gate for the operational routes hardened by PER-15244/PER-15245, with a rollout default.
+
+    Governed by ``ENFORCE_OPERATIONAL_ROUTE_AUTH``. When it is true this is exactly ``enforce_pdp_token``.
+    When it is false - the default, for a safe fleet rollout - a request that would be rejected is allowed
+    through but logged, so callers that don't yet send the PDP token keep working while the logs surface
+    them before enforcement is switched on.
+
+    The flag is read per-request (never captured at import) so a cloud control-plane override takes
+    effect and tests can toggle it. Kept as a distinct, named module-level function because the
+    fail-closed route audit recognises auth gates by callable name - a bare ``enforce_pdp_token`` here
+    could not carry the conditional behaviour, and an inline lambda would be invisible to the audit.
+    """
+    # Reuse enforce_pdp_token's exact reject logic in one place. When the flag is on, a rejection is
+    # honoured (re-raised). When it is off - the rollout default - the rejection is downgraded to
+    # warn-and-allow so no caller breaks; every would-be rejection is still counted, and surfaced in a
+    # per-route coalesced warning (see _operational_warn_should_emit) rather than one line per request.
+    try:
+        enforce_pdp_token(credentials)
+    except HTTPException as exc:
+        if sidecar_config.ENFORCE_OPERATIONAL_ROUTE_AUTH:
+            raise
+        emit, coalesced = _operational_warn_should_emit(request.url.path)
+        if emit:
+            logger.warning(
+                "ENFORCE_OPERATIONAL_ROUTE_AUTH is off: allowed {count} request(s) without a valid PDP token "
+                "(missing or invalid) to {method} {path} in the last ~{interval}s that would otherwise be "
+                "rejected (e.g. {detail}). Set ENFORCE_OPERATIONAL_ROUTE_AUTH=true to enforce the PDP token "
+                "on this route.",
+                count=coalesced,
+                method=request.method,
+                path=request.url.path,
+                interval=int(_OPERATIONAL_WARN_INTERVAL_SECONDS),
+                detail=exc.detail,
+            )
+
+
 def enforce_pdp_control_key(credentials: PdpCredentials = None):
     if sidecar_config.CONTAINER_CONTROL_KEY == MOCK_API_KEY:
         raise HTTPException(

horizon/config.py

@@ -262,6 +262,18 @@ def parse_plugins(value: Any) -> dict[str, dict[str, int | bool | str]]:
     # enables debug ouptut for the Kong integration endpoint
     KONG_INTEGRATION_DEBUG = confi.bool("KONG_INTEGRATION_DEBUG", False)
 
+    ENFORCE_OPERATIONAL_ROUTE_AUTH = confi.bool(
+        "ENFORCE_OPERATIONAL_ROUTE_AUTH",
+        False,
+        description="When true, enforce the PDP token on the operational routes hardened by the auth-hardening "
+        "rollout: the update-trigger routes (/policy-updater/trigger, /data-updater/trigger, /update_policy, "
+        "/update_policy_data) and the /kong decision endpoint. Defaults to FALSE for a safe fleet rollout: those "
+        "routes accept unauthenticated requests and only LOG the ones that would be rejected, so callers that do not "
+        "yet send the token keep working while you watch the logs. Flip to true (per-fleet via the cloud control "
+        "plane, or PDP_ENFORCE_OPERATIONAL_ROUTE_AUTH) once every caller sends the token. Every other PDP route is "
+        "always enforced regardless of this flag.",
+    )
+
     LOCAL_FACTS_WAIT_TIMEOUT = confi.float(
         "LOCAL_FACTS_WAIT_TIMEOUT",
         10,

horizon/enforcer/api.py

@@ -277,6 +277,12 @@ async def health():
 def init_enforcer_api_router(policy_store: BasePolicyStoreClient = None):  # noqa: C901
     policy_store = policy_store or DEFAULT_POLICY_STORE_GETTER()
     router = APIRouter()
+    # /kong lives on its own router so it can be mounted under the ENFORCE_OPERATIONAL_ROUTE_AUTH-aware
+    # gate (enforce_pdp_token_operational) while every other enforcer route keeps the plain
+    # enforce_pdp_token gate. Same closure as the rest of the enforcer routes: /kong's handler closes
+    # over kong_routes_table and the nested _is_allowed helper, so a router split here avoids a much
+    # larger refactor. Mounted separately in PermitPDP._configure_api_routes.
+    kong_router = APIRouter()
     if sidecar_config.KONG_INTEGRATION:
         with Path(KONG_ROUTES_TABLE_FILE).open() as f:
             kong_routes_table_raw = json.load(f)
@@ -543,7 +549,7 @@ async def is_allowed_nginx(
             )
             return {"allow": False, "result": False}
 
-    @router.post(
+    @kong_router.post(
         "/kong",
         response_model=KongAuthorizationResult,
         status_code=status.HTTP_200_OK,
@@ -620,7 +626,7 @@ async def is_allowed_kong(request: Request, query: KongAuthorizationQuery):
             )
             return {"allow": False, "result": False}
 
-    return router
+    return router, kong_router
 
 
 def _extract_regex_attributes(pattern: str, url: str) -> dict:

horizon/pdp.py

@@ -26,7 +26,7 @@
 from opal_common.logging_utils.formatter import Formatter
 from scalar_fastapi import get_scalar_api_reference
 
-from horizon.authentication import enforce_pdp_token
+from horizon.authentication import enforce_pdp_token, enforce_pdp_token_operational
 from horizon.config import MOCK_API_KEY, sidecar_config
 from horizon.connectivity.api import init_connectivity_router
 from horizon.enforcer.api import init_enforcer_api_router, init_enforcer_health_router, stats_manager
@@ -108,28 +108,32 @@ def apply_config(overrides_dict: dict, config_object: Confi):
 
 
 def _gate_opal_trigger_routes(app: FastAPI) -> None:
-    """Inject ``Depends(enforce_pdp_token)`` into the OPAL-mounted trigger routes.
+    """Inject ``Depends(enforce_pdp_token_operational)`` into the OPAL-mounted trigger routes.
 
     OpalClient mounts ``POST /policy-updater/trigger`` and ``POST /data-updater/trigger``
     on the app before ``PermitPDP`` gains control (opal_client.client._configure_api_routes),
     so the include_router-level dependencies used for every PDP-owned router cannot reach
-    them. We inject the standard PDP-token dependency into the already-mounted route objects
-    instead, mirroring what FastAPI itself does at ``APIRoute.__init__`` (fastapi/routing.py):
-    insert a parameterless sub-dependant at the head of ``route.dependant.dependencies``.
+    them. We inject the PDP-token dependency into the already-mounted route objects instead,
+    mirroring what FastAPI itself does at ``APIRoute.__init__`` (fastapi/routing.py): insert a
+    parameterless sub-dependant at the head of ``route.dependant.dependencies``.
+
+    The gate is the operational wrapper, so ``ENFORCE_OPERATIONAL_ROUTE_AUTH`` governs these routes
+    like the sibling operational routes: unauthenticated-but-logged by default, enforced when set.
 
     The Dependant is mutated IN PLACE - the route's request handler closes over that exact
     object, so the check is enforced on every request; do NOT reassign ``route.dependant``.
-    ``enforce_pdp_token`` only reads a header, so the route's body field needs no rebuild.
+    ``enforce_pdp_token_operational`` only reads the header and request, so the route's body field
+    needs no rebuild.
 
     Fails loud if a target route is missing (e.g. an OPAL upgrade renamed it): a silently
-    skipped injection would leave an update-trigger endpoint unauthenticated.
+    skipped injection would leave an update-trigger endpoint ungated even when enforcement is on.
     """
     gated: set[str] = set()
     for route in app.routes:
         if isinstance(route, APIRoute) and route.path in OPAL_TRIGGER_ROUTE_PATHS:
             route.dependant.dependencies.insert(
                 0,
-                get_parameterless_sub_dependant(depends=Depends(enforce_pdp_token), path=route.path_format),
+                get_parameterless_sub_dependant(depends=Depends(enforce_pdp_token_operational), path=route.path_format),
             )
             gated.add(route.path)
 
@@ -143,6 +147,23 @@ def _gate_opal_trigger_routes(app: FastAPI) -> None:
         raise SystemExit(GUNICORN_EXIT_APP)
 
 
+def _warn_if_operational_route_auth_disabled() -> None:
+    """Warn when ``ENFORCE_OPERATIONAL_ROUTE_AUTH`` is off and the operational routes accept unauthenticated calls.
+
+    This is the safe-rollout default (see SidecarConfig.ENFORCE_OPERATIONAL_ROUTE_AUTH): the token is not
+    enforced on the update-trigger routes or /kong, only logged. Surfaced at startup as well as per-request
+    so a fleet still in the permissive rollout phase stays visible; flip the flag on once every caller sends
+    the token.
+    """
+    if not sidecar_config.ENFORCE_OPERATIONAL_ROUTE_AUTH:
+        logger.warning(
+            "ENFORCE_OPERATIONAL_ROUTE_AUTH is OFF: the update-trigger routes (/policy-updater/trigger, "
+            "/data-updater/trigger, /update_policy, /update_policy_data) and /kong accept requests WITHOUT "
+            "a valid PDP token (missing or invalid) - would-be rejections are only logged. This is the safe "
+            "fleet-rollout default; set ENFORCE_OPERATIONAL_ROUTE_AUTH=true to enforce the PDP token on these routes."
+        )
+
+
 def _warn_if_opal_verifier_disabled(opal_client: OpalClient) -> None:
     """Warn loudly when the OPAL-authenticated routes are effectively open.
 
@@ -447,7 +468,7 @@ def _configure_api_routes(self, app: FastAPI):
         app.on_event("shutdown")(stats_manager.stop_tasks)
 
         enforcer_health_router = init_enforcer_health_router()
-        enforcer_router = init_enforcer_api_router(policy_store=self._opal.policy_store)
+        enforcer_router, kong_router = init_enforcer_api_router(policy_store=self._opal.policy_store)
         local_router = init_local_cache_api_router(policy_store=self._opal.policy_store)
         # Init system router
         system_router = init_system_api_router()
@@ -460,6 +481,14 @@ def _configure_api_routes(self, app: FastAPI):
             tags=["Authorization API"],
             dependencies=[Depends(enforce_pdp_token)],
         )
+        # /kong is gated with the operational wrapper so ENFORCE_OPERATIONAL_ROUTE_AUTH governs it
+        # during a fleet rollout (Kong's OPA plugin may not yet forward the PDP token) without
+        # touching the rest of the enforcer router.
+        app.include_router(
+            kong_router,
+            tags=["Authorization API"],
+            dependencies=[Depends(enforce_pdp_token_operational)],
+        )
 
         app.include_router(
             local_router,
@@ -498,11 +527,13 @@ def _configure_api_routes(self, app: FastAPI):
             )
 
         # TODO: remove this when clients update sdk version (legacy routes)
+        # Gated with the operational wrapper: these are aliases of the OPAL trigger routes, so they
+        # follow the same ENFORCE_OPERATIONAL_ROUTE_AUTH rollout default.
         @app.post(
             "/update_policy",
             status_code=status.HTTP_200_OK,
             include_in_schema=False,
-            dependencies=[Depends(enforce_pdp_token)],
+            dependencies=[Depends(enforce_pdp_token_operational)],
         )
         async def legacy_trigger_policy_update():
             logger.info("triggered policy update from api (legacy route)")
@@ -515,7 +546,7 @@ async def legacy_trigger_policy_update():
             "/update_policy_data",
             status_code=status.HTTP_200_OK,
             include_in_schema=False,
-            dependencies=[Depends(enforce_pdp_token)],
+            dependencies=[Depends(enforce_pdp_token_operational)],
         )
         async def legacy_trigger_data_update():
             logger.info("triggered policy data update from api (legacy route)")
@@ -534,6 +565,9 @@ async def legacy_trigger_data_update():
         # High-signal warning if the OPAL-authenticated routes are left open by a disabled
         # verifier (must never happen in a managed PDP).
         _warn_if_opal_verifier_disabled(self._opal)
+        # High-signal warning while ENFORCE_OPERATIONAL_ROUTE_AUTH is off (the rollout default) and
+        # the update-trigger and /kong routes accept unauthenticated requests.
+        _warn_if_operational_route_auth_disabled()
 
     @property
     def app(self):

horizon/tests/conftest.py

@@ -22,8 +22,23 @@
 from unittest.mock import Mock
 
 import aioresponses.core as _aioresponses_core
+import horizon.authentication as _authentication
+import pytest
 from aiohttp.client_reqrep import ClientResponse as _ClientResponse
 
+
+@pytest.fixture(autouse=True)
+def _reset_operational_warn_throttle():
+    """Give each test a clean warn-and-allow throttle window.
+
+    enforce_pdp_token_operational coalesces its "unauthenticated but allowed" warning to one line per
+    route per interval, and that state is process-global. Without this reset, a test that asserts the
+    warning could be silently suppressed by an earlier test that already logged for the same path.
+    """
+    _authentication.reset_operational_warn_throttle()
+    yield
+
+
 if "stream_writer" in inspect.signature(_ClientResponse.__init__).parameters:
 
     class _StreamWriterCompatClientResponse(_ClientResponse):