diff --git a/docs/decisions/issue-1093-control-plane-offload-preflight.md b/docs/decisions/issue-1093-control-plane-offload-preflight.md new file mode 100644 index 00000000..0bf1119f --- /dev/null +++ b/docs/decisions/issue-1093-control-plane-offload-preflight.md @@ -0,0 +1,67 @@ +# Issue 1093 Control-Plane Offload Preflight + +Date: 2026-08-11 + +Issue: #1093. Requirement: API-404. + +## Decision + +The HTTP adapter remains asynchronous at the network boundary, but every +synchronous control-plane, backend, and durable-store call runs in AnyIO's +bounded worker pool. An application-scoped async lock admits one target-mutating +call at a time. Mutations waiting for that lock do not occupy worker threads, +so status, snapshot, authentication, and audit requests can still make +progress while a backend call is slow. + +This includes audit persistence on middleware rejection paths. A malformed or +oversized request is rejected first, while its audit write is awaited through a +dedicated one-worker AnyIO capacity limiter that never borrows from the default +pool used by authentication, reads, and ordinary control-plane work. The +application retains at most +`ControlPlaneSecurityConfig.max_pending_rejection_audits` such writes (default +8, including the active write). Further rejection audits are dropped with a +bounded warning rather than retaining request tasks or starving authenticated +traffic. Audit persistence is best effort at this pre-routing boundary: an +unavailable or saturated audit store must not turn a deterministic `400` or +`413` into an exception or permit the rejected request to reach a route. + +The mutation queue is bounded by +`ControlPlaneSecurityConfig.max_pending_mutations`, including the active call. +The default is 32. Admission beyond that bound returns `503` with a +`Retry-After` header instead of retaining an unbounded number of request tasks. +The value must be positive and should be sized with the deployment's upstream +connection, timeout, and retry limits. + +One `FastAPI` application represents one runtime target and one event loop. +Per-application serialization prevents concurrent HTTP mutations from racing +the target's snapshot or invoking a non-thread-safe backend concurrently. This +does not create a distributed queue or make direct concurrent library calls +safe across processes. + +Worker cancellation is deliberately non-abandoning under Starlette/AnyIO's +`run_in_threadpool` contract: cancellation of the awaiting request does not +kill a Python thread midway through a backend mutation. Idempotency and the +durable operation record remain the recovery surface. Backends must still +implement their own bounded I/O timeouts; offload is not a substitute for an +operation deadline or process isolation. + +## Verification + +Acceptance requires: + +1. a blocked provisioning submission while an authenticated snapshot read + completes within a short independent deadline; +2. one active mutation per target; +3. stable `503` overload behavior at the configured pending bound; +4. an oversized-request audit that blocks in SQLite while an independent event + loop task continues within a short deadline; +5. saturation beyond the rejection-audit pending bound while a real + authenticated route retains a default-pool worker and completes; +6. stable `400`/`413` rejection when audit persistence raises or is dropped; +7. unchanged HTTP auth, idempotency, error, participant, and workflow suites; +8. lint and repository policy; and +9. full verification before release. + +Issue #1092 separately owns local transactional storage. A future multi-host +service must use a durable broker/worker design with explicit leases and +recovery rather than treating this in-process lock as distributed scheduling. diff --git a/docs/explain/sdl/runtime-architecture.md b/docs/explain/sdl/runtime-architecture.md index 050a5ed7..eeb44909 100644 --- a/docs/explain/sdl/runtime-architecture.md +++ b/docs/explain/sdl/runtime-architecture.md @@ -484,6 +484,29 @@ header identity must pass an explicit `ControlPlaneSecurityConfig`, set `trust_proxy_identity_headers=True`, and only trust those headers behind an authenticated proxy that strips caller-supplied identity headers. +Bearer and verified-proxy authentication require the same exact target binding. +An identity with no target, and an explicitly supplied bearer that is unknown, +revoked, or scoped to another target, is rejected; a rejected bearer never +falls back to proxy headers. Request admission is also bounded before FastAPI +parses or dispatches a body: a public ASGI wrapper checks one digits-only +`Content-Length`, then consumes at most the configured byte limit and replays +one coalesced body message. This boundary does not rely on framework-private +request caches and returns a stable `413` before a route can run. + +The HTTP adapter offloads synchronous backend and store calls to AnyIO's +bounded worker pool. An application-scoped async lock serializes target +mutations without occupying a worker while requests wait; independent reads +remain responsive while a backend is slow. The pending mutation count is +bounded by `ControlPlaneSecurityConfig.max_pending_mutations` and overload +returns `503` plus `Retry-After`. This in-process execution boundary neither +replaces backend I/O timeouts nor claims durable or distributed job queuing. +Request-size rejection also offloads audit persistence; an audit-store failure +cannot admit an invalid body or replace the stable `400`/`413` response. These +pre-routing audits use a separate one-worker limiter and a bounded pending +count, so an unauthenticated rejection flood cannot consume the default AnyIO +workers required by authenticated reads; excess audit records are dropped with +an operational warning. + ## Current Scope The current runtime scope includes: diff --git a/docs/requirements/API-404/requirement.md b/docs/requirements/API-404/requirement.md index d8b74f5f..038affab 100644 --- a/docs/requirements/API-404/requirement.md +++ b/docs/requirements/API-404/requirement.md @@ -22,7 +22,16 @@ Requirement inventory phase. Status audit deferred until the full canonical grap ## Traceability - IMPLEMENTS → GITHUB_ISSUE `8` (API-404: Secure, Durable, And Idempotent Control-Plane Semantics) +- IMPLEMENTS → GITHUB_ISSUE `1090` (Fail-closed bearer-token authentication and target binding) +- IMPLEMENTS → GITHUB_ISSUE `1091` (Bounded pre-routing HTTP request admission) +- DOCUMENTS → GITHUB_ISSUE `1093` (In-process HTTP offload and rejection-audit slice) - IMPLEMENTS → SPEC `contracts/schemas/control-plane/operation-receipt-v1.json` (Operation receipt JSON Schema — submission acknowledgment contract) - IMPLEMENTS → SPEC `contracts/schemas/control-plane/operation-status-v1.json` (Operation status JSON Schema — durable operation state contract) +- IMPLEMENTS → CODE_FILE `implementations/python/packages/raes_runtime/control_plane_api/_auth.py` (Fail-closed bearer and verified-proxy authentication) +- IMPLEMENTS → CODE_FILE `implementations/python/packages/raes_runtime/control_plane_api/_offload.py` (Bounded worker offload and mutation admission) +- IMPLEMENTS → CODE_FILE `implementations/python/packages/raes_runtime/control_plane_api_guards.py` (Bounded fail-closed request-size admission and rejection-audit offload) +- IMPLEMENTS → CODE_FILE `implementations/python/packages/raes_runtime/control_plane_security.py` (HTTP admission and pending-work limits) +- DOCUMENTS → DOCUMENTATION `docs/decisions/issue-1093-control-plane-offload-preflight.md` (In-process ASGI offload and rejection semantics) - TESTS → TEST `implementations/python/tests/test_runtime_control_plane.py` (Core control-plane unit tests) - TESTS → TEST `implementations/python/tests/test_runtime_control_plane_api.py` (HTTP/JSON control-plane API tests — auth, idempotency, durability, audit) +- TESTS → TEST `implementations/python/tests/test_issue_1093_request_rejection_offload.py` (Non-blocking, saturation-bounded, fail-closed request rejection audit tests) diff --git a/implementations/python/packages/raes_runtime/control_plane_api/__init__.py b/implementations/python/packages/raes_runtime/control_plane_api/__init__.py index 52e9d3ca..ad564ac2 100644 --- a/implementations/python/packages/raes_runtime/control_plane_api/__init__.py +++ b/implementations/python/packages/raes_runtime/control_plane_api/__init__.py @@ -29,6 +29,7 @@ from ..control_plane_api_participant_retrieval import register_participant_retrieval_routes from ..control_plane_security import ControlPlaneSecurityConfig from ._auth import _ControlPlaneApiAuth +from ._offload import _ControlPlaneCallExecutor from ._operation_routes import _install_request_guards, _register_operation_routes from ._participant_routes import ( _register_participant_control_routes, @@ -69,6 +70,9 @@ def create_control_plane_app( description="Reference HTTP/JSON adapter over the repo-owned runtime control plane.", ) app.state.control_plane_api_auth = _ControlPlaneApiAuth(control_plane, security) + app.state.control_plane_call_executor = _ControlPlaneCallExecutor( + max_pending_mutations=security.max_pending_mutations + ) _install_request_guards(app, control_plane, security) _register_operation_routes(app, control_plane) _register_workflow_routes(app, control_plane) diff --git a/implementations/python/packages/raes_runtime/control_plane_api/_auth.py b/implementations/python/packages/raes_runtime/control_plane_api/_auth.py index e848ef0c..17a50dd8 100644 --- a/implementations/python/packages/raes_runtime/control_plane_api/_auth.py +++ b/implementations/python/packages/raes_runtime/control_plane_api/_auth.py @@ -2,6 +2,7 @@ from __future__ import annotations +import hmac from typing import Annotated from fastapi import Depends, HTTPException, Request @@ -54,9 +55,14 @@ def _authenticate_request(self, request: Request) -> ControlPlaneIdentity: authorization = request.headers.get("authorization", "") if authorization.lower().startswith("bearer "): token = authorization.split(" ", 1)[1].strip() - identity = self._security.bearer_tokens.get(token) - if identity is not None: - return identity + identity = self._resolve_bearer_identity(token) + # A presented-but-unresolvable token is a hard failure: falling through + # to the proxy-header path would let a revoked or bogus token keep + # working whenever header identities are trusted, and would leave the + # rejected credential out of the audit log entirely. + if identity is None: + raise HTTPException(status_code=401, detail="invalid bearer token") + return self._require_target_binding(identity) if not self._security.trust_proxy_identity_headers: raise HTTPException(status_code=401, detail="trusted proxy identity headers are not enabled") identity_name = request.headers.get(self._security.identity_header, "") @@ -66,7 +72,25 @@ def _authenticate_request(self, request: Request) -> ControlPlaneIdentity: identity = self._security.trusted_identities.get(identity_name) if identity is None: raise HTTPException(status_code=401, detail="unknown client identity") - if identity.target_name and identity.target_name != self._control_plane.target_name: + return self._require_target_binding(identity) + + def _resolve_bearer_identity(self, token: str) -> ControlPlaneIdentity | None: + """Resolve a bearer token without leaking which token matched via timing.""" + + # Compare encoded bytes: ``compare_digest`` rejects non-ASCII ``str``, so a + # non-ASCII token would raise instead of being reported as unauthorized. + # The loop never breaks early, so timing does not reveal which token matched. + presented = token.encode("utf-8") + matched: ControlPlaneIdentity | None = None + for candidate, identity in self._security.bearer_tokens.items(): + if hmac.compare_digest(candidate.encode("utf-8"), presented): + matched = identity + return matched + + def _require_target_binding(self, identity: ControlPlaneIdentity) -> ControlPlaneIdentity: + """Require an identity scoped exactly to this control-plane target.""" + + if identity.target_name != self._control_plane.target_name: raise HTTPException(status_code=403, detail="identity is not authorized for this target") return identity diff --git a/implementations/python/packages/raes_runtime/control_plane_api/_offload.py b/implementations/python/packages/raes_runtime/control_plane_api/_offload.py new file mode 100644 index 00000000..76415cd1 --- /dev/null +++ b/implementations/python/packages/raes_runtime/control_plane_api/_offload.py @@ -0,0 +1,69 @@ +"""Bounded offload for synchronous control-plane work.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Callable +from typing import ParamSpec, TypeVar + +from fastapi import HTTPException, Request +from starlette.concurrency import run_in_threadpool + +_P = ParamSpec("_P") +_T = TypeVar("_T") + + +class _ControlPlaneCallExecutor: + """Keep blocking calls off the event loop and serialize target mutation. + + FastAPI/AnyIO owns the bounded worker pool. The async lock admits only one + target-mutating call at a time without consuming worker threads while other + mutations wait. Read and audit calls may still use separate workers, so a + slow backend does not prevent status or authentication requests. + """ + + def __init__(self, *, max_pending_mutations: int) -> None: + if max_pending_mutations <= 0: + raise ValueError("max_pending_mutations must be positive") + self._mutation_lock = asyncio.Lock() + self._max_pending_mutations = max_pending_mutations + self._pending_mutations = 0 + + @staticmethod + async def run( + call: Callable[_P, _T], + /, + *args: _P.args, + **kwargs: _P.kwargs, + ) -> _T: + return await run_in_threadpool(call, *args, **kwargs) + + async def mutate( + self, + call: Callable[_P, _T], + /, + *args: _P.args, + **kwargs: _P.kwargs, + ) -> _T: + if self._pending_mutations >= self._max_pending_mutations: + raise HTTPException( + status_code=503, + detail="control-plane mutation queue is full", + headers={"Retry-After": "1"}, + ) + self._pending_mutations += 1 + try: + async with self._mutation_lock: + return await self.run(call, *args, **kwargs) + finally: + self._pending_mutations -= 1 + + +def _control_plane_calls(request: Request) -> _ControlPlaneCallExecutor: + executor = getattr(request.app.state, "control_plane_call_executor", None) + if not isinstance(executor, _ControlPlaneCallExecutor): + raise RuntimeError("control-plane call executor is not configured") + return executor + + +__all__ = ("_ControlPlaneCallExecutor", "_control_plane_calls") diff --git a/implementations/python/packages/raes_runtime/control_plane_api/_operation_routes.py b/implementations/python/packages/raes_runtime/control_plane_api/_operation_routes.py index adcd7c33..0b4e10c4 100644 --- a/implementations/python/packages/raes_runtime/control_plane_api/_operation_routes.py +++ b/implementations/python/packages/raes_runtime/control_plane_api/_operation_routes.py @@ -2,9 +2,7 @@ from __future__ import annotations -from collections.abc import Awaitable, Callable - -from fastapi import FastAPI, HTTPException, Request, Response +from fastapi import FastAPI, HTTPException, Request from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse from raes_contracts.contracts import ( @@ -17,7 +15,7 @@ ) from ..control_plane import RuntimeControlPlane -from ..control_plane_api_guards import request_size_guard_response +from ..control_plane_api_guards import RequestSizeLimitMiddleware from ..control_plane_api_models import ( _evaluation_plan, _operation_status_model, @@ -28,6 +26,7 @@ ) from ..control_plane_security import ControlPlaneSecurityConfig from ._auth import _MutatingIdentity, _ReadIdentity +from ._offload import _control_plane_calls from ._responses import _CONFLICT_RESPONSES, _NOT_FOUND_RESPONSES, _receipt_response @@ -36,23 +35,17 @@ def _install_request_guards( control_plane: RuntimeControlPlane, security: ControlPlaneSecurityConfig, ) -> None: - @app.middleware("http") - async def _limit_request_size( - request: Request, - call_next: Callable[[Request], Awaitable[Response]], - ) -> Response: - guard_response = await request_size_guard_response( - control_plane, - request, - max_request_bytes=security.max_request_bytes, - ) - if guard_response is not None: - return guard_response - return await call_next(request) + app.add_middleware( + RequestSizeLimitMiddleware, + control_plane=control_plane, + max_request_bytes=security.max_request_bytes, + max_pending_rejection_audits=security.max_pending_rejection_audits, + ) @app.exception_handler(Exception) async def _redacted_errors(request: Request, exc: Exception) -> JSONResponse: - control_plane.record_audit( + await _control_plane_calls(request).run( + control_plane.record_audit, action=request.method, identity="anonymous", allowed=False, @@ -64,7 +57,8 @@ async def _redacted_errors(request: Request, exc: Exception) -> JSONResponse: @app.exception_handler(RequestValidationError) async def _redacted_request_validation_errors(request: Request, exc: RequestValidationError) -> JSONResponse: del exc - control_plane.record_audit( + await _control_plane_calls(request).run( + control_plane.record_audit, action=request.method, identity="anonymous", allowed=False, @@ -93,8 +87,14 @@ async def submit_provisioning( identity: _MutatingIdentity, ) -> OperationReceiptModel: submitted_plan = _provisioning_plan(plan) - if submitted_plan.operations and not control_plane.is_planner_authorized_provisioning_plan(submitted_plan): - control_plane.record_audit( + calls = _control_plane_calls(request) + planner_authorized = await calls.run( + control_plane.is_planner_authorized_provisioning_plan, + submitted_plan, + ) + if submitted_plan.operations and not planner_authorized: + await calls.run( + control_plane.record_audit, action="submit_provisioning", identity=identity.identity, allowed=False, @@ -103,7 +103,8 @@ async def submit_provisioning( ) raise HTTPException(status_code=403, detail="provisioning plan is not planner-authorized") try: - receipt = control_plane.submit_provisioning( + receipt = await calls.mutate( + control_plane.submit_provisioning, submitted_plan, idempotency_key=request.headers.get("idempotency-key", ""), request_fingerprint=_request_fingerprint( @@ -113,7 +114,8 @@ async def submit_provisioning( ) except ValueError as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc - control_plane.record_audit( + await calls.run( + control_plane.record_audit, action="submit_provisioning", identity=identity.identity, allowed=True, @@ -128,8 +130,10 @@ async def submit_orchestration( plan: OrchestrationPlanModel, identity: _MutatingIdentity, ) -> OperationReceiptModel: + calls = _control_plane_calls(request) try: - receipt = control_plane.submit_orchestration( + receipt = await calls.mutate( + control_plane.submit_orchestration, _orchestration_plan(plan), idempotency_key=request.headers.get("idempotency-key", ""), request_fingerprint=_request_fingerprint( @@ -139,7 +143,8 @@ async def submit_orchestration( ) except ValueError as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc - control_plane.record_audit( + await calls.run( + control_plane.record_audit, action="submit_orchestration", identity=identity.identity, allowed=True, @@ -154,8 +159,10 @@ async def submit_evaluation( plan: EvaluationPlanModel, identity: _MutatingIdentity, ) -> OperationReceiptModel: + calls = _control_plane_calls(request) try: - receipt = control_plane.submit_evaluation( + receipt = await calls.mutate( + control_plane.submit_evaluation, _evaluation_plan(plan), idempotency_key=request.headers.get("idempotency-key", ""), request_fingerprint=_request_fingerprint( @@ -165,7 +172,8 @@ async def submit_evaluation( ) except ValueError as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc - control_plane.record_audit( + await calls.run( + control_plane.record_audit, action="submit_evaluation", identity=identity.identity, allowed=True, @@ -185,10 +193,12 @@ async def get_operation( request: Request, identity: _ReadIdentity, ) -> OperationStatusModel: - status = control_plane.get_operation(operation_id) + calls = _control_plane_calls(request) + status = await calls.run(control_plane.get_operation, operation_id) if status is None: raise HTTPException(status_code=404, detail=f"Unknown operation: {operation_id}") - control_plane.record_audit( + await calls.run( + control_plane.record_audit, action="get_operation", identity=identity.identity, allowed=True, @@ -202,23 +212,27 @@ async def get_snapshot( request: Request, identity: _ReadIdentity, ) -> RuntimeSnapshotEnvelopeModel: - control_plane.record_audit( + calls = _control_plane_calls(request) + await calls.run( + control_plane.record_audit, action="get_snapshot", identity=identity.identity, allowed=True, target=str(request.url.path), ) - return _snapshot_model(control_plane.get_snapshot()) + return await calls.run(lambda: _snapshot_model(control_plane.get_snapshot())) @app.get("/apparatus/operational-summary") async def get_operational_apparatus_summary( request: Request, identity: _ReadIdentity, ) -> dict[str, object]: - control_plane.record_audit( + calls = _control_plane_calls(request) + await calls.run( + control_plane.record_audit, action="get_operational_apparatus_summary", identity=identity.identity, allowed=True, target=str(request.url.path), ) - return control_plane.operational_apparatus_summary() + return await calls.run(control_plane.operational_apparatus_summary) diff --git a/implementations/python/packages/raes_runtime/control_plane_api/_participant_routes.py b/implementations/python/packages/raes_runtime/control_plane_api/_participant_routes.py index e5292990..96f62f3d 100644 --- a/implementations/python/packages/raes_runtime/control_plane_api/_participant_routes.py +++ b/implementations/python/packages/raes_runtime/control_plane_api/_participant_routes.py @@ -21,6 +21,7 @@ ) from ..participant_control_intents import ParticipantControlIntent from ._auth import _MutatingIdentity, _ReadIdentity +from ._offload import _control_plane_calls from ._responses import ( _BAD_REQUEST_CONFLICT_RESPONSES, _CONFLICT_RESPONSES, @@ -43,6 +44,7 @@ async def control_participant_execution( identity: _MutatingIdentity, body: _ParticipantExecutionControlBody, ) -> OperationReceiptModel: + calls = _control_plane_calls(request) try: control_request = ParticipantExecutionControlRequestModel( execution_scope_ref=execution_scope_ref, @@ -50,7 +52,8 @@ async def control_participant_execution( expected_generation=body.expected_generation, timeout_seconds=body.timeout_seconds, ) - receipt = control_plane.control_participant_execution( + receipt = await calls.mutate( + control_plane.control_participant_execution, control_request, idempotency_key=request.headers.get("idempotency-key", ""), request_fingerprint=_request_fingerprint( @@ -60,7 +63,8 @@ async def control_participant_execution( ) except ValueError as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc - control_plane.record_audit( + await calls.run( + control_plane.record_audit, action=f"participant_execution_{body.action}", identity=identity.identity, allowed=True, @@ -78,11 +82,13 @@ async def get_participant_execution_state( request: Request, identity: _ReadIdentity, ) -> ParticipantExecutionServiceStateModel: + calls = _control_plane_calls(request) try: - state = control_plane.participant_execution_state(execution_scope_ref) + state = await calls.run(control_plane.participant_execution_state, execution_scope_ref) except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc - control_plane.record_audit( + await calls.run( + control_plane.record_audit, action="get_participant_execution_state", identity=identity.identity, allowed=True, @@ -113,15 +119,18 @@ async def record_participant_control( body: ParticipantControlIntent, identity: _MutatingIdentity, ) -> OperationReceiptModel: + calls = _control_plane_calls(request) try: - receipt = control_plane.record_participant_control( + receipt = await calls.mutate( + control_plane.record_participant_control, participant_address, body, identity=identity, idempotency_key=request.headers.get("idempotency-key", ""), ) except PermissionError as exc: - control_plane.record_audit( + await calls.run( + control_plane.record_audit, action="record_participant_control", identity=identity.identity, allowed=False, @@ -152,8 +161,10 @@ async def initialize_participant_episode( body: _ParticipantInitializeBody | None = None, ) -> OperationReceiptModel: payload = body or _ParticipantInitializeBody() + calls = _control_plane_calls(request) try: - receipt = control_plane.initialize_participant_episode( + receipt = await calls.mutate( + control_plane.initialize_participant_episode, participant_address, episode_id=payload.episode_id, idempotency_key=request.headers.get("idempotency-key", ""), @@ -164,7 +175,8 @@ async def initialize_participant_episode( ) except ValueError as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc - control_plane.record_audit( + await calls.run( + control_plane.record_audit, action="initialize_participant_episode", identity=identity.identity, allowed=True, @@ -184,8 +196,10 @@ async def reset_participant_episode( body: _ParticipantResetBody | None = None, ) -> OperationReceiptModel: payload = body or _ParticipantResetBody() + calls = _control_plane_calls(request) try: - receipt = control_plane.reset_participant_episode( + receipt = await calls.mutate( + control_plane.reset_participant_episode, participant_address, episode_id=payload.episode_id, reason=payload.reason, @@ -197,7 +211,8 @@ async def reset_participant_episode( ) except ValueError as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc - control_plane.record_audit( + await calls.run( + control_plane.record_audit, action="reset_participant_episode", identity=identity.identity, allowed=True, @@ -222,8 +237,10 @@ async def restart_participant_episode( body: _ParticipantRestartBody | None = None, ) -> OperationReceiptModel: payload = body or _ParticipantRestartBody() + calls = _control_plane_calls(request) try: - receipt = control_plane.restart_participant_episode( + receipt = await calls.mutate( + control_plane.restart_participant_episode, participant_address, episode_id=payload.episode_id, reason=payload.reason, @@ -235,7 +252,8 @@ async def restart_participant_episode( ) except ValueError as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc - control_plane.record_audit( + await calls.run( + control_plane.record_audit, action="restart_participant_episode", identity=identity.identity, allowed=True, @@ -255,12 +273,14 @@ async def terminate_participant_episode( body: _ParticipantTerminateBody | None = None, ) -> OperationReceiptModel: payload = body or _ParticipantTerminateBody() + calls = _control_plane_calls(request) try: terminal_reason = ParticipantEpisodeTerminalReason(payload.terminal_reason) except ValueError as exc: raise HTTPException(status_code=400, detail=f"invalid terminal_reason: {exc}") from exc try: - receipt = control_plane.terminate_participant_episode( + receipt = await calls.mutate( + control_plane.terminate_participant_episode, participant_address, terminal_reason=terminal_reason, detail=payload.detail, @@ -272,7 +292,8 @@ async def terminate_participant_episode( ) except ValueError as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc - control_plane.record_audit( + await calls.run( + control_plane.record_audit, action="terminate_participant_episode", identity=identity.identity, allowed=True, diff --git a/implementations/python/packages/raes_runtime/control_plane_api/_workflow_routes.py b/implementations/python/packages/raes_runtime/control_plane_api/_workflow_routes.py index c4b9ac4b..39929415 100644 --- a/implementations/python/packages/raes_runtime/control_plane_api/_workflow_routes.py +++ b/implementations/python/packages/raes_runtime/control_plane_api/_workflow_routes.py @@ -8,6 +8,7 @@ from ..control_plane import RuntimeControlPlane from ..control_plane_api_models import _request_fingerprint from ._auth import _MutatingIdentity +from ._offload import _control_plane_calls from ._responses import _CONFLICT_RESPONSES, _receipt_response @@ -23,8 +24,10 @@ async def cancel_workflow( cancellation: WorkflowCancellationRequestModel | None = None, ) -> OperationReceiptModel: payload = cancellation or WorkflowCancellationRequestModel() + calls = _control_plane_calls(request) try: - receipt = control_plane.cancel_workflow( + receipt = await calls.mutate( + control_plane.cancel_workflow, workflow_address, run_id=payload.run_id, reason=payload.reason, @@ -36,7 +39,8 @@ async def cancel_workflow( ) except ValueError as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc - control_plane.record_audit( + await calls.run( + control_plane.record_audit, action="cancel_workflow", identity=identity.identity, allowed=True, @@ -50,8 +54,10 @@ async def reconcile_timeouts( request: Request, identity: _MutatingIdentity, ) -> OperationReceiptModel: + calls = _control_plane_calls(request) try: - receipt = control_plane.reconcile_workflow_timeouts( + receipt = await calls.mutate( + control_plane.reconcile_workflow_timeouts, idempotency_key=request.headers.get("idempotency-key", ""), request_fingerprint=_request_fingerprint( request, @@ -60,7 +66,8 @@ async def reconcile_timeouts( ) except ValueError as exc: raise HTTPException(status_code=409, detail=str(exc)) from exc - control_plane.record_audit( + await calls.run( + control_plane.record_audit, action="reconcile_workflow_timeouts", identity=identity.identity, allowed=True, diff --git a/implementations/python/packages/raes_runtime/control_plane_api_guards.py b/implementations/python/packages/raes_runtime/control_plane_api_guards.py index ced8ebf3..c1a14022 100644 --- a/implementations/python/packages/raes_runtime/control_plane_api_guards.py +++ b/implementations/python/packages/raes_runtime/control_plane_api_guards.py @@ -1,91 +1,180 @@ -"""HTTP request guards for the runtime control-plane API.""" +"""Bounded ASGI request admission for the runtime control-plane API.""" from __future__ import annotations -from fastapi import Request, Response +import logging +from collections.abc import Sequence +from functools import partial + +from anyio import CapacityLimiter, to_thread from fastapi.responses import JSONResponse +from starlette.types import ASGIApp, Message, Receive, Scope, Send from .control_plane import RuntimeControlPlane _REQUEST_TOO_LARGE_DETAIL = "request too large" _INVALID_CONTENT_LENGTH_DETAIL = "invalid content-length" +_LOGGER = logging.getLogger(__name__) + + +class RejectionAuditExecutor: + """Bound rejected-request audits away from AnyIO's default worker limiter.""" + def __init__(self, control_plane: RuntimeControlPlane, *, max_pending: int) -> None: + if max_pending <= 0: + raise ValueError("max_pending rejection audits must be positive") + self._control_plane = control_plane + self._max_pending = max_pending + self._pending = 0 + self._limiter = CapacityLimiter(1) -async def request_size_guard_response( - control_plane: RuntimeControlPlane, - request: Request, - *, - max_request_bytes: int, -) -> Response | None: - guard_response = _content_length_guard_response( - control_plane, - request, - max_request_bytes=max_request_bytes, - ) - if guard_response is not None: - return guard_response - return await _body_size_guard_response( - control_plane, - request, - max_request_bytes=max_request_bytes, - ) - - -def _content_length_guard_response( - control_plane: RuntimeControlPlane, - request: Request, - *, - max_request_bytes: int, -) -> JSONResponse | None: - response: JSONResponse | None = None - content_length = request.headers.get("content-length") - if content_length is not None: + async def record( + self, + *, + action: str, + identity: str, + allowed: bool, + target: str, + reason: str, + ) -> bool: + if self._pending >= self._max_pending: + _LOGGER.warning( + "control-plane rejection audit queue is full; dropping audit action=%s target=%s reason=%s", + action, + target, + reason, + ) + return False + self._pending += 1 try: - content_length_value = int(content_length) - except ValueError: - response = _invalid_content_length_response(control_plane, request) + call = partial( + self._control_plane.record_audit, + action=action, + identity=identity, + allowed=allowed, + target=target, + reason=reason, + ) + await to_thread.run_sync(call, limiter=self._limiter) + finally: + self._pending -= 1 + return True + + +class RequestSizeLimitMiddleware: + """Reject oversized HTTP bodies before FastAPI parses or dispatches them. + + The middleware buffers at most ``max_request_bytes`` bytes, then replays + accepted ASGI messages to the application. A chunk that crosses the limit + is rejected before it is copied into the buffer, keeping middleware-owned + allocation bounded without relying on Starlette's private ``Request._body`` + cache. + """ + + def __init__( + self, + app: ASGIApp, + *, + control_plane: RuntimeControlPlane, + max_request_bytes: int, + max_pending_rejection_audits: int = 8, + ) -> None: + if max_request_bytes <= 0: + raise ValueError("max_request_bytes must be positive") + self._app = app + self._rejection_audits = RejectionAuditExecutor( + control_plane, + max_pending=max_pending_rejection_audits, + ) + self._max_request_bytes = max_request_bytes + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] != "http": + await self._app(scope, receive, send) else: - if content_length_value > max_request_bytes: - response = _request_too_large_response(control_plane, request) - return response - - -async def _body_size_guard_response( - control_plane: RuntimeControlPlane, - request: Request, - *, - max_request_bytes: int, -) -> JSONResponse | None: - body = await request.body() - if len(body) > max_request_bytes: - return _request_too_large_response(control_plane, request) - request.state.raw_body = body - return None - - -def _request_too_large_response( - control_plane: RuntimeControlPlane, - request: Request, -) -> JSONResponse: - control_plane.record_audit( - action=request.method, - identity="anonymous", - allowed=False, - target=str(request.url.path), - reason=_REQUEST_TOO_LARGE_DETAIL, - ) - return JSONResponse(status_code=413, content={"detail": _REQUEST_TOO_LARGE_DETAIL}) - - -def _invalid_content_length_response( - control_plane: RuntimeControlPlane, - request: Request, -) -> JSONResponse: - control_plane.record_audit( - action=request.method, - identity="anonymous", - allowed=False, - target=str(request.url.path), - reason=_INVALID_CONTENT_LENGTH_DETAIL, - ) - return JSONResponse(status_code=400, content={"detail": _INVALID_CONTENT_LENGTH_DETAIL}) + await self._handle_http(scope, receive, send) + + async def _handle_http(self, scope: Scope, receive: Receive, send: Send) -> None: + try: + content_length = _declared_content_length(scope.get("headers", ())) + except ValueError: + await self._reject(scope, receive, send, status_code=400, detail=_INVALID_CONTENT_LENGTH_DETAIL) + return + if content_length is not None and content_length > self._max_request_bytes: + await self._reject(scope, receive, send, status_code=413, detail=_REQUEST_TOO_LARGE_DETAIL) + return + + body = bytearray() + disconnected = False + while True: + message = await receive() + if message["type"] == "http.disconnect": + disconnected = True + break + if message["type"] != "http.request": + continue + chunk = message.get("body", b"") + if len(chunk) > self._max_request_bytes - len(body): + await self._reject(scope, receive, send, status_code=413, detail=_REQUEST_TOO_LARGE_DETAIL) + return + body.extend(chunk) + if not message.get("more_body", False): + break + + raw_body = bytes(body) + scope.setdefault("state", {})["raw_body"] = raw_body + replay_message: Message | None = ( + {"type": "http.disconnect"} + if disconnected + else {"type": "http.request", "body": raw_body, "more_body": False} + ) + + async def replay_receive() -> Message: + nonlocal replay_message + if replay_message is not None: + message = replay_message + replay_message = None + return message + return await receive() + + await self._app(scope, replay_receive, send) + + async def _reject( + self, + scope: Scope, + receive: Receive, + send: Send, + *, + status_code: int, + detail: str, + ) -> None: + try: + await self._rejection_audits.record( + action=scope.get("method", ""), + identity="anonymous", + allowed=False, + target=scope.get("path", ""), + reason=detail, + ) + except Exception: + # Admission already failed closed. An unavailable audit store must + # neither dispatch the body nor replace the stable rejection. + _LOGGER.exception("control-plane rejection audit persistence failed") + response = JSONResponse(status_code=status_code, content={"detail": detail}) + await response(scope, receive, send) + + +def _declared_content_length(headers: Sequence[tuple[bytes, bytes]]) -> int | None: + values = [value for name, value in headers if name.lower() == b"content-length"] + if not values: + return None + if len(values) != 1: + raise ValueError("content-length must appear at most once") + raw_value = values[0] + if not raw_value.isdigit(): + raise ValueError("content-length must be a non-negative integer") + try: + value = int(raw_value) + except ValueError as exc: + raise ValueError("content-length must be a non-negative integer") from exc + return value diff --git a/implementations/python/packages/raes_runtime/control_plane_api_participant_retrieval.py b/implementations/python/packages/raes_runtime/control_plane_api_participant_retrieval.py index f0322485..701a34c1 100644 --- a/implementations/python/packages/raes_runtime/control_plane_api_participant_retrieval.py +++ b/implementations/python/packages/raes_runtime/control_plane_api_participant_retrieval.py @@ -13,6 +13,7 @@ ) from .control_plane import RuntimeControlPlane +from .control_plane_api._offload import _control_plane_calls from .control_plane_security import ControlPlaneIdentity, ParticipantAudienceSubjectBinding _NOT_FOUND_RESPONSES = {404: {"description": "Not found"}} @@ -45,17 +46,20 @@ async def get_participant_status_view( identity: _ReadIdentity, ) -> ParticipantStatusViewModel: audience_binding = _require_governed_audience_candidate(control_plane, identity, participant_address) - view = _governed_view( + calls = _control_plane_calls(request) + view = await calls.mutate( + _governed_view, lambda: control_plane.get_participant_status_view( participant_address, identity=identity, audience_binding=audience_binding, idempotency_key=request.headers.get("idempotency-key", ""), - ) + ), ) if view is None: raise HTTPException(status_code=404, detail=f"Unknown participant: {participant_address}") - control_plane.record_audit( + await calls.run( + control_plane.record_audit, action="get_participant_status_view", identity=identity.identity, allowed=True, @@ -74,21 +78,24 @@ async def get_participant_history_view( identity: _ReadIdentity, ) -> ParticipantHistoryViewModel: audience_binding = _require_governed_audience_candidate(control_plane, identity, participant_address) - view = _governed_view( + calls = _control_plane_calls(request) + view = await calls.mutate( + _governed_view, lambda: control_plane.get_participant_history_view( participant_address, episode_id, identity=identity, audience_binding=audience_binding, idempotency_key=request.headers.get("idempotency-key", ""), - ) + ), ) if view is None: raise HTTPException( status_code=404, detail=f"Unknown participant episode: {participant_address}/{episode_id}", ) - control_plane.record_audit( + await calls.run( + control_plane.record_audit, action="get_participant_history_view", identity=identity.identity, allowed=True, @@ -110,7 +117,9 @@ async def get_participant_context_view( payload_ref: str | None = None, ) -> ParticipantContextViewModel: audience_binding = _require_governed_audience_candidate(control_plane, identity, participant_address) - view = _governed_view( + calls = _control_plane_calls(request) + view = await calls.mutate( + _governed_view, lambda: control_plane.get_participant_context_view( participant_address, view_ref=view_ref, @@ -120,11 +129,12 @@ async def get_participant_context_view( identity=identity, audience_binding=audience_binding, idempotency_key=request.headers.get("idempotency-key", ""), - ) + ), ) if view is None: raise HTTPException(status_code=404, detail=f"Unknown participant: {participant_address}") - control_plane.record_audit( + await calls.run( + control_plane.record_audit, action="get_participant_context_view", identity=identity.identity, allowed=True, diff --git a/implementations/python/packages/raes_runtime/control_plane_security.py b/implementations/python/packages/raes_runtime/control_plane_security.py index a2f08481..7ac053a7 100644 --- a/implementations/python/packages/raes_runtime/control_plane_security.py +++ b/implementations/python/packages/raes_runtime/control_plane_security.py @@ -2,8 +2,10 @@ from __future__ import annotations +from collections.abc import Mapping from dataclasses import dataclass, field from enum import Enum +from types import MappingProxyType class ControlPlaneRole(str, Enum): @@ -62,8 +64,22 @@ class ControlPlaneSecurityConfig: identity_header: str = "x-raes-client-identity" trust_proxy_identity_headers: bool = False max_request_bytes: int = 1_000_000 - trusted_identities: dict[str, ControlPlaneIdentity] = field(default_factory=dict) - bearer_tokens: dict[str, ControlPlaneIdentity] = field(default_factory=dict) + max_pending_mutations: int = 32 + max_pending_rejection_audits: int = 8 + trusted_identities: Mapping[str, ControlPlaneIdentity] = field(default_factory=dict) + bearer_tokens: Mapping[str, ControlPlaneIdentity] = field(default_factory=dict) + + def __post_init__(self) -> None: + if self.max_pending_mutations <= 0: + raise ValueError("max_pending_mutations must be positive") + if self.max_pending_rejection_audits <= 0: + raise ValueError("max_pending_rejection_audits must be positive") + # ``frozen=True`` only blocks rebinding the attributes; a caller (or a + # later code path) could still mutate the underlying dicts and grant + # principals or tokens after construction, defeating ``strict_defaults``. + # Read-only proxies keep dict equality while blocking that mutation. + object.__setattr__(self, "trusted_identities", MappingProxyType(dict(self.trusted_identities))) + object.__setattr__(self, "bearer_tokens", MappingProxyType(dict(self.bearer_tokens))) @classmethod def strict_defaults(cls) -> ControlPlaneSecurityConfig: diff --git a/implementations/python/tests/test_issue_1093_request_rejection_offload.py b/implementations/python/tests/test_issue_1093_request_rejection_offload.py new file mode 100644 index 00000000..6eed0894 --- /dev/null +++ b/implementations/python/tests/test_issue_1093_request_rejection_offload.py @@ -0,0 +1,493 @@ +"""API-404 ASGI rejection-path audit offload tests.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Coroutine +from pathlib import Path +from threading import Event +from typing import Any, TypeVar + +import httpx +import pytest +import raes_runtime.control_plane_api_guards as api_guards +from anyio import to_thread +from raes_backend_stubs.stubs import create_stub_target +from raes_runtime.control_plane import RuntimeControlPlane +from raes_runtime.control_plane_api import create_control_plane_app +from raes_runtime.control_plane_api_guards import ( + RejectionAuditExecutor, + RequestSizeLimitMiddleware, + _declared_content_length, +) +from raes_runtime.control_plane_security import ( + ControlPlaneIdentity, + ControlPlaneRole, + ControlPlaneSecurityConfig, +) +from raes_runtime.control_plane_store_local import LocalControlPlaneStore +from starlette.types import Message, Receive, Scope, Send + +_T = TypeVar("_T") + + +def _run(coroutine: Coroutine[Any, Any, _T]) -> _T: + """Run one coroutine without replacing or closing pytest's default loop.""" + + loop = asyncio.new_event_loop() + try: + return loop.run_until_complete(coroutine) + finally: + loop.close() + + +def _accepted_app(calls: list[str]): + async def app(scope: Scope, _receive: Receive, send: Send) -> None: + calls.append(scope.get("path", "")) + await send({"type": "http.response.start", "status": 204, "headers": []}) + await send({"type": "http.response.body", "body": b""}) + + return app + + +def _http_scope() -> Scope: + return { + "type": "http", + "asgi": {"version": "3.0"}, + "http_version": "1.1", + "method": "POST", + "scheme": "http", + "path": "/accepted", + "raw_path": b"/accepted", + "query_string": b"", + "root_path": "", + "headers": [], + "client": ("127.0.0.1", 1), + "server": ("testserver", 80), + "state": {}, + } + + +def test_request_size_middleware_rejects_nonpositive_limit() -> None: + control_plane = RuntimeControlPlane(create_stub_target()) + app = _accepted_app([]) + + with pytest.raises(ValueError, match="max_request_bytes must be positive"): + RequestSizeLimitMiddleware( + app, + control_plane=control_plane, + max_request_bytes=0, + ) + + +def test_rejection_audit_executor_rejects_nonpositive_pending_bound() -> None: + control_plane = RuntimeControlPlane(create_stub_target()) + with pytest.raises(ValueError, match="max_pending rejection audits must be positive"): + RejectionAuditExecutor(control_plane, max_pending=0) + + +def test_request_size_middleware_passes_non_http_scope_through() -> None: + control_plane = RuntimeControlPlane(create_stub_target()) + calls: list[str] = [] + + async def app(scope: Scope, _receive: Receive, _send: Send) -> None: + calls.append(scope["type"]) + + async def receive() -> Message: + return {"type": "websocket.disconnect"} + + async def send(_message: Message) -> None: + return None + + middleware = RequestSizeLimitMiddleware(app, control_plane=control_plane, max_request_bytes=1) + scope: Scope = {"type": "websocket", "asgi": {"version": "3.0"}, "state": {}} + _run(middleware(scope, receive, send)) + + assert calls == ["websocket"] + + +def test_request_size_middleware_coalesces_body_then_delegates_receive() -> None: + control_plane = RuntimeControlPlane(create_stub_target()) + source_messages: list[Message] = [ + {"type": "http.request", "body": b"x", "more_body": True}, + {"type": "http.request", "body": b"y", "more_body": False}, + {"type": "http.disconnect"}, + ] + replayed: list[Message] = [] + + async def receive() -> Message: + return source_messages.pop(0) + + async def app(_scope: Scope, replay_receive: Receive, send: Send) -> None: + replayed.extend([await replay_receive(), await replay_receive()]) + await send({"type": "http.response.start", "status": 204, "headers": []}) + await send({"type": "http.response.body", "body": b""}) + + async def send(_message: Message) -> None: + return None + + middleware = RequestSizeLimitMiddleware(app, control_plane=control_plane, max_request_bytes=2) + scope = _http_scope() + _run(middleware(scope, receive, send)) + + assert replayed == [ + {"type": "http.request", "body": b"xy", "more_body": False}, + {"type": "http.disconnect"}, + ] + assert scope["state"]["raw_body"] == b"xy" + + +def test_request_size_middleware_does_not_retain_empty_transport_messages() -> None: + control_plane = RuntimeControlPlane(create_stub_target()) + source_messages: list[Message] = [ + *({"type": "http.request", "body": b"", "more_body": True} for _ in range(100)), + {"type": "http.request", "body": b"x", "more_body": False}, + ] + replayed: list[Message] = [] + + async def receive() -> Message: + return source_messages.pop(0) + + async def app(_scope: Scope, replay_receive: Receive, send: Send) -> None: + replayed.append(await replay_receive()) + await send({"type": "http.response.start", "status": 204, "headers": []}) + await send({"type": "http.response.body", "body": b""}) + + async def send(_message: Message) -> None: + return None + + middleware = RequestSizeLimitMiddleware(app, control_plane=control_plane, max_request_bytes=1) + _run(middleware(_http_scope(), receive, send)) + + assert replayed == [{"type": "http.request", "body": b"x", "more_body": False}] + assert source_messages == [] + + +def test_request_size_middleware_accepts_disconnect_before_body() -> None: + control_plane = RuntimeControlPlane(create_stub_target()) + calls: list[str] = [] + + async def receive() -> Message: + return {"type": "http.disconnect"} + + async def send(_message: Message) -> None: + return None + + middleware = RequestSizeLimitMiddleware( + _accepted_app(calls), + control_plane=control_plane, + max_request_bytes=1, + ) + scope = _http_scope() + _run(middleware(scope, receive, send)) + + assert calls == ["/accepted"] + assert scope["state"]["raw_body"] == b"" + + +def test_request_size_middleware_rejects_stream_that_exceeds_limit_without_header() -> None: + control_plane = RuntimeControlPlane(create_stub_target()) + calls: list[str] = [] + sent: list[Message] = [] + + async def receive() -> Message: + return {"type": "http.request", "body": b"xx", "more_body": False} + + async def send(message: Message) -> None: + sent.append(message) + + middleware = RequestSizeLimitMiddleware( + _accepted_app(calls), + control_plane=control_plane, + max_request_bytes=1, + ) + _run(middleware(_http_scope(), receive, send)) + + response_start = next(message for message in sent if message["type"] == "http.response.start") + assert response_start["status"] == 413 + assert calls == [] + + +def test_request_size_middleware_rejects_hostile_single_chunk_before_copy( + monkeypatch: pytest.MonkeyPatch, +) -> None: + control_plane = RuntimeControlPlane(create_stub_target()) + calls: list[str] = [] + sent: list[Message] = [] + extend_calls: list[int] = [] + real_bytearray = bytearray + + class CopyGuardBytearray(real_bytearray): + def extend(self, chunk: bytes, /) -> None: + extend_calls.append(len(chunk)) + raise AssertionError("oversized chunk reached the replay buffer") + + monkeypatch.setattr(api_guards, "bytearray", CopyGuardBytearray, raising=False) + payload = b"x" * (1024 * 1024) + + async def receive() -> Message: + return {"type": "http.request", "body": payload, "more_body": False} + + async def send(message: Message) -> None: + sent.append(message) + + middleware = RequestSizeLimitMiddleware( + _accepted_app(calls), + control_plane=control_plane, + max_request_bytes=8, + ) + _run(middleware(_http_scope(), receive, send)) + + response_start = next(message for message in sent if message["type"] == "http.response.start") + assert response_start["status"] == 413 + assert extend_calls == [] + assert calls == [] + + +@pytest.mark.parametrize( + ("chunks", "expected_status", "expected_raw_body"), + [ + ((b"ab", b"cd"), 204, b"abcd"), + ((b"ab", b"cde"), 413, None), + ], +) +def test_request_size_middleware_enforces_exact_stream_boundary( + chunks: tuple[bytes, ...], + expected_status: int, + expected_raw_body: bytes | None, +) -> None: + control_plane = RuntimeControlPlane(create_stub_target()) + calls: list[str] = [] + sent: list[Message] = [] + source_messages: list[Message] = [ + { + "type": "http.request", + "body": chunk, + "more_body": index < len(chunks) - 1, + } + for index, chunk in enumerate(chunks) + ] + + async def receive() -> Message: + return source_messages.pop(0) + + async def send(message: Message) -> None: + sent.append(message) + + middleware = RequestSizeLimitMiddleware( + _accepted_app(calls), + control_plane=control_plane, + max_request_bytes=4, + ) + scope = _http_scope() + _run(middleware(scope, receive, send)) + + response_start = next(message for message in sent if message["type"] == "http.response.start") + assert response_start["status"] == expected_status + assert calls == (["/accepted"] if expected_status == 204 else []) + assert source_messages == [] + if expected_raw_body is None: + assert "raw_body" not in scope["state"] + else: + assert scope["state"]["raw_body"] == expected_raw_body + + +@pytest.mark.parametrize( + "headers", + [ + [(b"content-length", b"1"), (b"Content-Length", b"1")], + [(b"content-length", b"-1")], + [(b"content-length", b"+1")], + [(b"content-length", b" 1")], + [(b"content-length", b"1 ")], + [(b"content-length", b"1_0")], + [(b"content-length", b"")], + ], +) +def test_declared_content_length_rejects_ambiguous_or_non_digit_values( + headers: list[tuple[bytes, bytes]], +) -> None: + with pytest.raises(ValueError, match="content-length"): + _declared_content_length(headers) + + +def test_blocked_sqlite_rejection_audit_does_not_block_event_loop( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + store = LocalControlPlaneStore(tmp_path / "control-plane") + control_plane = RuntimeControlPlane(create_stub_target(), store=store) + entered = Event() + release = Event() + real_append = store.append_audit + + def blocked_append(event: object) -> None: + entered.set() + if not release.wait(timeout=5): + raise TimeoutError("audit release was not signalled") + real_append(event) + + monkeypatch.setattr(store, "append_audit", blocked_append) + route_calls: list[str] = [] + app = RequestSizeLimitMiddleware( + _accepted_app(route_calls), + control_plane=control_plane, + max_request_bytes=1, + ) + + async def exercise() -> tuple[httpx.Response, httpx.Response]: + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://testserver", + ) as client: + rejected_task = asyncio.create_task(client.post("/oversized", content=b"xx")) + try: + assert await asyncio.to_thread(entered.wait, 2) + accepted = await asyncio.wait_for(client.get("/health"), timeout=0.5) + finally: + release.set() + return await rejected_task, accepted + + try: + rejected, accepted = _run(exercise()) + finally: + release.set() + + assert rejected.status_code == 413 + assert accepted.status_code == 204 + assert route_calls == ["/health"] + assert len(store.read_audit()) == 1 + + +def test_rejection_audit_saturation_does_not_starve_real_authenticated_route( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + store = LocalControlPlaneStore(tmp_path / "control-plane") + target = create_stub_target() + control_plane = RuntimeControlPlane(target, store=store) + entered = Event() + release = Event() + real_append = store.append_audit + + def blocked_rejection_append(event: object) -> None: + if getattr(event, "reason", "") == "request too large": + entered.set() + if not release.wait(timeout=5): + raise TimeoutError("rejection audit release was not signalled") + real_append(event) + + monkeypatch.setattr(store, "append_audit", blocked_rejection_append) + security = ControlPlaneSecurityConfig( + max_request_bytes=1, + max_pending_rejection_audits=2, + bearer_tokens={ + "auditor-token": ControlPlaneIdentity( + identity="auditor", + roles=frozenset({ControlPlaneRole.AUDITOR}), + target_name=target.name, + ) + }, + ) + app = create_control_plane_app(control_plane, security=security) + + async def exercise() -> tuple[list[httpx.Response], httpx.Response, int]: + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://testserver", + ) as client: + rejected_tasks = [ + asyncio.create_task(client.post(f"/oversized/{index}", content=b"xx")) for index in range(6) + ] + try: + assert await asyncio.to_thread(entered.wait, 2) + for _ in range(100): + if "rejection audit queue is full" in caplog.text: + break + await asyncio.sleep(0.01) + assert "rejection audit queue is full" in caplog.text + default_workers_in_use = to_thread.current_default_thread_limiter().borrowed_tokens + snapshot = await asyncio.wait_for( + client.get( + "/snapshot", + headers={"Authorization": "Bearer auditor-token"}, + ), + timeout=0.5, + ) + finally: + release.set() + return await asyncio.gather(*rejected_tasks), snapshot, default_workers_in_use + + try: + rejected, snapshot, default_workers_in_use = _run(exercise()) + finally: + release.set() + + assert {response.status_code for response in rejected} == {413} + assert snapshot.status_code == 200 + assert default_workers_in_use == 0 + rejection_audits = [event for event in store.read_audit() if event.reason == "request too large"] + assert len(rejection_audits) == 2 + + +@pytest.mark.parametrize( + ("content_length", "expected_status"), + [(b"not-a-number", 400), (b"2", 413)], +) +def test_rejection_remains_fail_closed_when_audit_persistence_fails( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + content_length: bytes, + expected_status: int, +) -> None: + store = LocalControlPlaneStore(tmp_path / f"control-plane-{expected_status}") + control_plane = RuntimeControlPlane(create_stub_target(), store=store) + + def failed_append(_event: object) -> None: + raise OSError("audit database unavailable") + + monkeypatch.setattr(store, "append_audit", failed_append) + route_calls: list[str] = [] + app = RequestSizeLimitMiddleware( + _accepted_app(route_calls), + control_plane=control_plane, + max_request_bytes=1, + ) + sent: list[Message] = [] + request_sent = False + + async def receive() -> Message: + nonlocal request_sent + if request_sent: + return {"type": "http.disconnect"} + request_sent = True + return {"type": "http.request", "body": b"xx", "more_body": False} + + async def send(message: Message) -> None: + sent.append(message) + + scope: Scope = { + "type": "http", + "asgi": {"version": "3.0"}, + "http_version": "1.1", + "method": "POST", + "scheme": "http", + "path": "/rejected", + "raw_path": b"/rejected", + "query_string": b"", + "root_path": "", + "headers": [(b"content-length", content_length)], + "client": ("127.0.0.1", 1), + "server": ("testserver", 80), + "state": {}, + } + + _run(app(scope, receive, send)) + + response_start = next(message for message in sent if message["type"] == "http.response.start") + assert response_start["status"] == expected_status + assert route_calls == [] + assert store.read_audit() == [] + assert "control-plane rejection audit persistence failed" in caplog.text diff --git a/implementations/python/tests/test_runtime_control_plane_api.py b/implementations/python/tests/test_runtime_control_plane_api.py index 6b70ce11..2a21af91 100644 --- a/implementations/python/tests/test_runtime_control_plane_api.py +++ b/implementations/python/tests/test_runtime_control_plane_api.py @@ -2,11 +2,17 @@ from __future__ import annotations +import asyncio import textwrap +from collections.abc import Coroutine from pathlib import Path +from threading import Event +from typing import Any, TypeVar +import httpx import pytest import raes_runtime.control_plane_store as control_plane_store_module +from fastapi import HTTPException from raes import parse_sdl from raes_backend_stubs.stubs import create_stub_target from raes_contracts.contracts import ( @@ -14,7 +20,8 @@ ParticipantHistoryViewModel, ParticipantStatusViewModel, ) -from raes_contracts.plan_projection import provisioning_plan_model +from raes_contracts.plan_projection import evaluation_plan_model, provisioning_plan_model +from raes_contracts.planning import ProvisioningPlan from raes_contracts.runtime_state import ( ExplicitnessClass, ExplicitnessProvenance, @@ -26,14 +33,29 @@ from raes_processor.planner import plan from raes_runtime.control_plane import RuntimeControlPlane from raes_runtime.control_plane_api import create_control_plane_app +from raes_runtime.control_plane_api._auth import _ControlPlaneApiAuth +from raes_runtime.control_plane_api._offload import _control_plane_calls, _ControlPlaneCallExecutor from raes_runtime.control_plane_security import ( ControlPlaneIdentity, ControlPlaneRole, ControlPlaneSecurityConfig, ) from raes_runtime.control_plane_store import ControlPlaneOperationRecord, LocalControlPlaneStore +from starlette.requests import Request from starlette.testclient import TestClient +_T = TypeVar("_T") + + +def _run(coroutine: Coroutine[Any, Any, _T]) -> _T: + """Run one coroutine without replacing or closing pytest's default loop.""" + + loop = asyncio.new_event_loop() + try: + return loop.run_until_complete(coroutine) + finally: + loop.close() + def _scenario(yaml_str: str): return parse_sdl(textwrap.dedent(yaml_str)) @@ -70,9 +92,15 @@ def _participant_operation_record(operation_id: str, participant_address: str) - ) -def _test_security(target_name: str, *, max_request_bytes: int = 1_000_000) -> ControlPlaneSecurityConfig: +def _test_security( + target_name: str, + *, + max_request_bytes: int = 1_000_000, + max_pending_mutations: int = 32, +) -> ControlPlaneSecurityConfig: return ControlPlaneSecurityConfig( max_request_bytes=max_request_bytes, + max_pending_mutations=max_pending_mutations, trust_proxy_identity_headers=True, trusted_identities={ "backend-service": ControlPlaneIdentity( @@ -105,6 +133,16 @@ def test_control_plane_strict_defaults_ship_without_builtin_principals(): assert security.bearer_tokens == {} +def test_control_plane_security_rejects_nonpositive_mutation_queue_bound() -> None: + with pytest.raises(ValueError, match="max_pending_mutations must be positive"): + ControlPlaneSecurityConfig(max_pending_mutations=0) + + +def test_control_plane_security_rejects_nonpositive_rejection_audit_bound() -> None: + with pytest.raises(ValueError, match="max_pending_rejection_audits must be positive"): + ControlPlaneSecurityConfig(max_pending_rejection_audits=0) + + def test_control_plane_api_default_security_does_not_trust_builtin_headers_or_tokens(): target = create_stub_target() control_plane = RuntimeControlPlane(target) @@ -158,6 +196,69 @@ def test_control_plane_api_openapi_documents_explicit_error_responses(): assert "/apparatus/operational-summary" in operation_responses +def test_control_plane_call_lookup_fails_closed_without_configured_executor() -> None: + target = create_stub_target() + app = create_control_plane_app(RuntimeControlPlane(target), security=_test_security(target.name)) + del app.state.control_plane_call_executor + request = Request({"type": "http", "app": app}) + + with pytest.raises(RuntimeError, match="executor is not configured"): + _control_plane_calls(request) + + +def test_control_plane_api_accepts_evaluation_plan() -> None: + scenario = _scenario(""" +name: evaluation-route +nodes: + vm: + type: compute + os: linux + resources: {ram: 1 gib, cpu: 1} +""") + target = create_stub_target() + execution_plan = plan(compile_runtime_model(scenario), target.manifest) + control_plane = RuntimeControlPlane(target) + app = create_control_plane_app(control_plane, security=_test_security(target.name)) + headers = { + "x-raes-client-verified": "true", + "x-raes-client-identity": "backend-service", + } + + with TestClient(app) as client: + response = client.post( + "/operations/evaluation", + json=evaluation_plan_model(execution_plan.evaluation).model_dump(mode="json", exclude_none=True), + headers=headers, + ) + + assert response.status_code == 200 + status = control_plane.get_operation(response.json()["operation_id"]) + assert status is not None + assert status.state is OperationState.SUCCEEDED + + +def test_control_plane_api_redacts_unexpected_route_errors(monkeypatch: pytest.MonkeyPatch) -> None: + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + app = create_control_plane_app(control_plane, security=_test_security(target.name)) + monkeypatch.setattr( + control_plane, + "get_snapshot", + lambda: (_ for _ in ()).throw(RuntimeError("SECRET-BACKEND-DETAIL")), + ) + + with TestClient(app, raise_server_exceptions=False) as client: + response = client.get( + "/snapshot", + headers={"authorization": "Bearer test-auditor-token"}, + ) + + assert response.status_code == 500 + assert response.json() == {"detail": "internal server error"} + assert "SECRET-BACKEND-DETAIL" not in response.text + assert control_plane.audit_log()[-1].reason == "internal-error:RuntimeError" + + def test_control_plane_api_accepts_orchestration_plan_and_exposes_snapshot(): scenario = _scenario(""" name: workflow @@ -394,6 +495,171 @@ def test_control_plane_api_supports_idempotent_retries(): assert first.json()["operation_id"] == second.json()["operation_id"] +def test_slow_backend_submission_does_not_block_unrelated_http_reads( + monkeypatch: pytest.MonkeyPatch, +) -> None: + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + app = create_control_plane_app( + control_plane, + security=_test_security(target.name), + ) + entered = Event() + release = Event() + real_submit = control_plane.submit_provisioning + + def blocking_submit( + submitted_plan: ProvisioningPlan, + *, + base_snapshot: RuntimeSnapshot | None = None, + idempotency_key: str = "", + request_fingerprint: str = "", + ) -> OperationReceipt: + entered.set() + if not release.wait(timeout=5): + raise TimeoutError("test backend was not released") + return real_submit( + submitted_plan, + base_snapshot=base_snapshot, + idempotency_key=idempotency_key, + request_fingerprint=request_fingerprint, + ) + + monkeypatch.setattr(control_plane, "submit_provisioning", blocking_submit) + headers = { + "x-raes-client-verified": "true", + "x-raes-client-identity": "backend-service", + } + + async def exercise() -> None: + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: + submission = asyncio.create_task( + client.post( + "/operations/provisioning", + json={"operations": [], "diagnostics": [], "realization_authority": []}, + headers=headers, + ) + ) + try: + assert await asyncio.to_thread(entered.wait, 2) + assert not submission.done() + snapshot = await asyncio.wait_for( + client.get("/snapshot", headers=headers), + timeout=1, + ) + assert snapshot.status_code == 200 + finally: + release.set() + response = await asyncio.wait_for(submission, timeout=2) + assert response.status_code == 200 + + _run(exercise()) + + +def test_control_plane_rejects_mutation_queue_overload( + monkeypatch: pytest.MonkeyPatch, +) -> None: + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + app = create_control_plane_app( + control_plane, + security=_test_security(target.name, max_pending_mutations=1), + ) + entered = Event() + release = Event() + real_submit = control_plane.submit_provisioning + + def blocking_submit( + submitted_plan: ProvisioningPlan, + *, + base_snapshot: RuntimeSnapshot | None = None, + idempotency_key: str = "", + request_fingerprint: str = "", + ) -> OperationReceipt: + entered.set() + if not release.wait(timeout=5): + raise TimeoutError("test backend was not released") + return real_submit( + submitted_plan, + base_snapshot=base_snapshot, + idempotency_key=idempotency_key, + request_fingerprint=request_fingerprint, + ) + + monkeypatch.setattr(control_plane, "submit_provisioning", blocking_submit) + headers = { + "x-raes-client-verified": "true", + "x-raes-client-identity": "backend-service", + } + + async def exercise() -> None: + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: + first = asyncio.create_task( + client.post( + "/operations/provisioning", + json={"operations": [], "diagnostics": [], "realization_authority": []}, + headers={**headers, "idempotency-key": "first"}, + ) + ) + try: + assert await asyncio.to_thread(entered.wait, 2) + overloaded = await asyncio.wait_for( + client.post( + "/operations/provisioning", + json={"operations": [], "diagnostics": [], "realization_authority": []}, + headers={**headers, "idempotency-key": "second"}, + ), + timeout=1, + ) + assert overloaded.status_code == 503 + assert overloaded.json() == {"detail": "control-plane mutation queue is full"} + assert overloaded.headers["retry-after"] == "1" + finally: + release.set() + assert (await asyncio.wait_for(first, timeout=2)).status_code == 200 + + _run(exercise()) + + +def test_control_plane_executor_serializes_target_mutations() -> None: + executor = _ControlPlaneCallExecutor(max_pending_mutations=2) + first_entered = Event() + release_first = Event() + execution_order: list[str] = [] + + def mutation(label: str) -> str: + execution_order.append(f"start:{label}") + if label == "first": + first_entered.set() + if not release_first.wait(timeout=5): + raise TimeoutError("first mutation was not released") + execution_order.append(f"end:{label}") + return label + + async def exercise() -> None: + first = asyncio.create_task(executor.mutate(mutation, "first")) + second: asyncio.Task[str] | None = None + try: + assert await asyncio.to_thread(first_entered.wait, 2) + second = asyncio.create_task(executor.mutate(mutation, "second")) + await asyncio.sleep(0.05) + assert execution_order == ["start:first"] + finally: + release_first.set() + assert second is not None + assert await asyncio.gather(first, second) == ["first", "second"] + + _run(exercise()) + assert execution_order == ["start:first", "end:first", "start:second", "end:second"] + + +def test_control_plane_executor_rejects_nonpositive_queue_bound() -> None: + with pytest.raises(ValueError, match="max_pending_mutations must be positive"): + _ControlPlaneCallExecutor(max_pending_mutations=0) + + def test_control_plane_api_persists_operations_and_snapshot(tmp_path: Path): scenario = _scenario(""" name: workflow @@ -571,6 +837,205 @@ def test_control_plane_api_rejects_invalid_content_length_header(): assert control_plane.audit_log()[-1].reason == "invalid content-length" +def test_control_plane_api_enforces_request_size_limit_without_content_length(): + """A chunked body (no content-length) must still be rejected with 413.""" + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + security = _test_security(target.name, max_request_bytes=32) + app = create_control_plane_app(control_plane, security=security) + headers = { + "x-raes-client-verified": "true", + "x-raes-client-identity": "backend-service", + "content-type": "application/json", + } + + def _chunked_body(): + for _ in range(100): + yield b"x" * 32 + + with TestClient(app) as client: + response = client.post( + "/operations/provisioning", + content=_chunked_body(), + headers=headers, + ) + + assert response.status_code == 413 + assert response.json() == {"detail": "request too large"} + assert control_plane.audit_log()[-1].reason == "request too large" + + +def test_control_plane_api_rejects_invalid_bearer_token_instead_of_trusting_headers(): + """An unresolvable bearer token must fail closed, not fall through to header identity.""" + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + app = create_control_plane_app( + control_plane, + security=_test_security(target.name), + ) + + with TestClient(app) as client: + response = client.get( + "/snapshot", + headers={ + "authorization": "Bearer revoked-token", + "x-raes-client-verified": "true", + "x-raes-client-identity": "backend-service", + }, + ) + + assert response.status_code == 401 + assert response.json() == {"detail": "invalid bearer token"} + assert control_plane.audit_log()[-1].reason == "invalid bearer token" + assert control_plane.audit_log()[-1].allowed is False + + +def test_control_plane_auth_rejects_non_ascii_bearer_token_as_unauthorized(): + """A non-ASCII token must be reported unauthorized, not crash the comparison. + + Starlette decodes header bytes as latin-1, so a raw request can deliver a + non-ASCII token string even though HTTP clients refuse to encode one. A + ``str``-based constant-time comparison would raise ``TypeError`` there and + surface as a 500 instead of a 401. + """ + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + auth = _ControlPlaneApiAuth(control_plane, _test_security(target.name)) + request = Request( + { + "type": "http", + "method": "GET", + "path": "/snapshot", + "headers": [(b"authorization", "Bearer token-\xf6\xe9".encode("latin-1"))], + "query_string": b"", + }, + ) + + with pytest.raises(HTTPException) as excinfo: + auth.read_identity(request) + + assert excinfo.value.status_code == 401 + assert excinfo.value.detail == "invalid bearer token" + + +def test_control_plane_api_rejects_bearer_token_bound_to_another_target(): + """A bearer token scoped to a different target must not authenticate here. + + The header-identity path already enforces this binding; the bearer path must + apply the same check rather than returning the identity unconditionally. + """ + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + security = ControlPlaneSecurityConfig( + trust_proxy_identity_headers=False, + bearer_tokens={ + "other-target-token": ControlPlaneIdentity( + identity="operator", + roles=frozenset({ControlPlaneRole.OPERATOR}), + target_name="some-other-target", + ), + }, + ) + app = create_control_plane_app(control_plane, security=security) + + with TestClient(app) as client: + response = client.get( + "/snapshot", + headers={"authorization": "Bearer other-target-token"}, + ) + + assert response.status_code == 403 + assert response.json() == {"detail": "identity is not authorized for this target"} + + +@pytest.mark.parametrize("authentication_path", ["bearer", "verified-proxy"]) +def test_control_plane_api_rejects_identity_without_target_binding(authentication_path: str) -> None: + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + identity = ControlPlaneIdentity( + identity="unbound-auditor", + roles=frozenset({ControlPlaneRole.AUDITOR}), + ) + if authentication_path == "bearer": + security = ControlPlaneSecurityConfig(bearer_tokens={"unbound-token": identity}) + headers = {"authorization": "Bearer unbound-token"} + else: + security = ControlPlaneSecurityConfig( + trust_proxy_identity_headers=True, + trusted_identities={"unbound-auditor": identity}, + ) + headers = { + "x-raes-client-verified": "true", + "x-raes-client-identity": "unbound-auditor", + } + app = create_control_plane_app(control_plane, security=security) + + with TestClient(app) as client: + response = client.get("/snapshot", headers=headers) + + assert response.status_code == 403 + assert response.json() == {"detail": "identity is not authorized for this target"} + + +def test_control_plane_security_config_mappings_cannot_be_mutated_after_construction(): + """``strict_defaults`` must stay fail-closed; frozen=True alone does not stop dict mutation.""" + security = ControlPlaneSecurityConfig.strict_defaults() + intruder = ControlPlaneIdentity(identity="intruder", roles=frozenset({ControlPlaneRole.OPERATOR})) + + with pytest.raises(TypeError): + security.bearer_tokens["stolen"] = intruder # type: ignore[index] + with pytest.raises(TypeError): + security.trusted_identities["stolen"] = intruder # type: ignore[index] + + assert security.bearer_tokens == {} + assert security.trusted_identities == {} + + +def test_request_size_guard_stops_reading_an_oversized_chunked_body(): + """The ASGI guard must stop after the first chunk that crosses the cap.""" + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + app = create_control_plane_app( + control_plane, + security=_test_security(target.name, max_request_bytes=64), + ) + chunk = b"x" * 32 + total_chunks = 1000 + delivered = 0 + + async def receive() -> dict[str, object]: + nonlocal delivered + if delivered >= total_chunks: + return {"type": "http.request", "body": b"", "more_body": False} + delivered += 1 + return {"type": "http.request", "body": chunk, "more_body": True} + + sent: list[dict[str, object]] = [] + + async def send(message: dict[str, object]) -> None: + sent.append(message) + + scope = { + "type": "http", + "asgi": {"version": "3.0"}, + "http_version": "1.1", + "method": "POST", + "scheme": "http", + "path": "/operations/provisioning", + "raw_path": b"/operations/provisioning", + "query_string": b"", + "root_path": "", + "headers": [(b"content-type", b"application/json")], + "client": ("127.0.0.1", 1), + "server": ("testserver", 80), + } + + _run(app(scope, receive, send)) + + assert any(message.get("status") == 413 for message in sent) + assert delivered <= 3 + + def test_local_control_plane_store_saves_snapshot_with_atomic_replace( tmp_path: Path, monkeypatch: pytest.MonkeyPatch,