diff --git a/implementations/python/packages/raes_processor/planner/ordering.py b/implementations/python/packages/raes_processor/planner/ordering.py index 52e4e1cf..7ef6cf23 100644 --- a/implementations/python/packages/raes_processor/planner/ordering.py +++ b/implementations/python/packages/raes_processor/planner/ordering.py @@ -2,8 +2,6 @@ from ..models import Diagnostic, PlannedResource, RuntimeDomain, SnapshotEntry from ..semantics.planner import ( - DependencyKind, - dependency_graph_for_resources, resource_delete_order, resource_dependency_cycles, resource_topological_order, @@ -12,10 +10,6 @@ from ..semantics.realization_snapshot_sanitization import realization_payloads_match -def _ordering_graph(resources: dict[str, PlannedResource]) -> dict[str, tuple[str, ...]]: - return dependency_graph_for_resources(resources, kind=DependencyKind.ORDERING) - - def _ordering_cycles(resources: dict[str, PlannedResource]) -> list[tuple[str, ...]]: return resource_dependency_cycles(resources) diff --git a/implementations/python/packages/raes_processor/semantics/planner.py b/implementations/python/packages/raes_processor/semantics/planner.py index b246a16d..26ff9f6e 100644 --- a/implementations/python/packages/raes_processor/semantics/planner.py +++ b/implementations/python/packages/raes_processor/semantics/planner.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections import deque -from collections.abc import Callable, Iterable, Mapping +from collections.abc import Callable, Iterable, Iterator, Mapping from dataclasses import dataclass from enum import Enum from typing import Protocol, TypeVar @@ -123,7 +123,7 @@ def dependency_cycles( on_stack: set[str] = set() cycles: list[tuple[str, ...]] = [] - def strongconnect(node: str) -> None: + def visit(node: str) -> None: nonlocal index indices[node] = index lowlinks[node] = index @@ -131,16 +131,7 @@ def strongconnect(node: str) -> None: stack.append(node) on_stack.add(node) - for dependency in graph[node]: - if dependency not in indices: - strongconnect(dependency) - lowlinks[node] = min(lowlinks[node], lowlinks[dependency]) - elif dependency in on_stack: - lowlinks[node] = min(lowlinks[node], indices[dependency]) - - if lowlinks[node] != indices[node]: - return - + def close_component(node: str) -> None: component: list[str] = [] while stack: member = stack.pop() @@ -153,6 +144,33 @@ def strongconnect(node: str) -> None: if len(component) > 1 or component[0] in graph[component[0]]: cycles.append(tuple(component)) + # Tarjan is driven from an explicit stack rather than recursion: DFS depth + # reaches the longest dependency path, so a recursive walk raises + # RecursionError on scenarios with more than roughly a thousand chained + # resources, aborting planning instead of reporting cycles. + def strongconnect(root: str) -> None: + visit(root) + frames: list[tuple[str, Iterator[str]]] = [(root, iter(graph[root]))] + while frames: + node, dependencies = frames[-1] + descended = False + for dependency in dependencies: + if dependency not in indices: + visit(dependency) + frames.append((dependency, iter(graph[dependency]))) + descended = True + break + if dependency in on_stack: + lowlinks[node] = min(lowlinks[node], indices[dependency]) + if descended: + continue + frames.pop() + if frames: + parent = frames[-1][0] + lowlinks[parent] = min(lowlinks[parent], lowlinks[node]) + if lowlinks[node] == indices[node]: + close_component(node) + for node in sorted(graph, key=canonical_resource_identity): if node not in indices: strongconnect(node) 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..de41f910 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,6 +72,24 @@ 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") + 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: + """Reject an identity scoped to a different target than this control plane.""" + if identity.target_name and 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_guards.py b/implementations/python/packages/raes_runtime/control_plane_api_guards.py index ced8ebf3..fe21eb71 100644 --- a/implementations/python/packages/raes_runtime/control_plane_api_guards.py +++ b/implementations/python/packages/raes_runtime/control_plane_api_guards.py @@ -56,10 +56,27 @@ async def _body_size_guard_response( *, 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 + # Accumulate the body incrementally and stop as soon as the running total + # exceeds the limit. Buffering via ``request.body()`` would read the whole + # payload first, so a request without a declared ``content-length`` (e.g. + # ``Transfer-Encoding: chunked``) bypasses ``_content_length_guard_response`` + # and could exhaust memory before any size check runs. + body = bytearray() + async for chunk in request.stream(): + # Measure before copying: a single oversized chunk would otherwise be + # appended in full before the limit is consulted. + if len(body) + len(chunk) > max_request_bytes: + return _request_too_large_response(control_plane, request) + body.extend(chunk) + accepted_body = bytes(body) + # Streaming consumes the receive channel, so seed Starlette's body cache the + # way ``Request.body()`` would. Route handlers and FastAPI's own body parsing + # then still see the payload instead of an exhausted stream: Starlette's own + # ``_CachedRequest.wrapped_receive`` replays ``_body`` to the inner app, which + # is the framework's hook for middleware that consumes the body, and there is + # no public equivalent. + request._body = accepted_body # NOSONAR - documented Starlette body-cache hook, no public equivalent + request.state.raw_body = accepted_body return None diff --git a/implementations/python/packages/raes_runtime/control_plane_security.py b/implementations/python/packages/raes_runtime/control_plane_security.py index a2f08481..100aed53 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,16 @@ 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) + trusted_identities: Mapping[str, ControlPlaneIdentity] = field(default_factory=dict) + bearer_tokens: Mapping[str, ControlPlaneIdentity] = field(default_factory=dict) + + def __post_init__(self) -> None: + # ``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/packages/raes_runtime/control_plane_timeouts.py b/implementations/python/packages/raes_runtime/control_plane_timeouts.py index 6d9f0dfc..24114cc8 100644 --- a/implementations/python/packages/raes_runtime/control_plane_timeouts.py +++ b/implementations/python/packages/raes_runtime/control_plane_timeouts.py @@ -14,6 +14,9 @@ from .control_plane_workflows import maybe_apply_compensation, parse_timestamp +TIMED_OUT_REASON = "workflow timed out" +UNPARSEABLE_START_REASON = "workflow timed out: started_at could not be parsed" + def workflow_timeout_update( snapshot: RuntimeSnapshot, @@ -26,11 +29,10 @@ def workflow_timeout_update( update = None timeout_seconds = _eligible_workflow_timeout_seconds(entry) normalized = _running_workflow_result(orchestration_results.get(workflow_address)) - if ( - timeout_seconds is not None - and normalized is not None - and _workflow_has_timed_out(normalized, timeout_seconds, submitted_at) - ): + terminal_reason = None + if timeout_seconds is not None and normalized is not None: + terminal_reason = _workflow_timeout_reason(normalized, timeout_seconds, submitted_at) + if terminal_reason is not None: update = _timed_out_workflow_update( snapshot, workflow_address, @@ -38,6 +40,7 @@ def workflow_timeout_update( timeout_seconds, orchestration_history, submitted_at, + terminal_reason, ) return update @@ -77,17 +80,30 @@ def _coerce_timeout_seconds(raw: object) -> int | None: return timeout -def _workflow_has_timed_out( +def _workflow_timeout_reason( normalized: WorkflowExecutionState, timeout_seconds: int, submitted_at: str, -) -> bool: +) -> str | None: + """Return the terminal reason when the workflow must time out, else ``None``. + + ``submitted_at`` is the caller's reconciliation clock and governs the whole + pass, so an unusable value is raised rather than quietly disabling every + timeout. A running workflow whose own ``started_at`` cannot be parsed has no + derivable deadline; reporting "not timed out" would pin it in RUNNING + forever, so it is reclaimed under a distinct reason instead. + """ + + current = parse_timestamp(submitted_at) try: - deadline = parse_timestamp(normalized.started_at).timestamp() + timeout_seconds - current = parse_timestamp(submitted_at).timestamp() - except Exception: - return False - return current >= deadline + started = parse_timestamp(normalized.started_at) + except (TypeError, ValueError): + return UNPARSEABLE_START_REASON + # Elapsed time is compared against the timeout rather than added to the start + # instant: `timeout_seconds` has no declared upper bound, and folding a very + # large one into a float timestamp or a timedelta overflows. + elapsed_seconds = (current - started).total_seconds() + return TIMED_OUT_REASON if elapsed_seconds >= timeout_seconds else None def _timed_out_workflow_update( @@ -97,8 +113,9 @@ def _timed_out_workflow_update( timeout_seconds: int, orchestration_history: dict[str, list[dict[str, object]]], submitted_at: str, + terminal_reason: str, ) -> tuple[dict[str, object], list[dict[str, object]]]: - timed_out_state = _timed_out_workflow_state(normalized, submitted_at) + timed_out_state = _timed_out_workflow_state(normalized, submitted_at, terminal_reason) history = orchestration_history.setdefault(workflow_address, []) history.append( WorkflowHistoryEvent( @@ -119,6 +136,7 @@ def _timed_out_workflow_update( def _timed_out_workflow_state( normalized: WorkflowExecutionState, submitted_at: str, + terminal_reason: str, ) -> WorkflowExecutionState: return WorkflowExecutionState( state_schema_version=normalized.state_schema_version, @@ -126,7 +144,7 @@ def _timed_out_workflow_state( run_id=normalized.run_id, started_at=normalized.started_at, updated_at=submitted_at, - terminal_reason="workflow timed out", + terminal_reason=terminal_reason, compensation_status=WorkflowCompensationStatus.NOT_REQUIRED, compensation_started_at=None, compensation_updated_at=None, diff --git a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py index 901cac39..f50f0303 100644 --- a/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py +++ b/implementations/python/packages/raes_runtime/participant_scheduler_concurrency.py @@ -138,6 +138,48 @@ def _reserve_concurrent_actions( ) +def _restore_pre_batch_snapshot( + run: SchedulerRunState, + pre_batch: RuntimeSnapshot, +) -> None: + """Undo ``_reserve_concurrent_actions`` when the batch produced no results. + + Reservations are taken before the backend batch call, and only + ``_commit_concurrent_result`` clears a participant's ``in_flight``. Abandoning + the batch without undoing them would leave participants in-flight and the + service non-quiescent with nothing to complete them. + + The pre-batch snapshot is reinstated wholesale rather than adjusted field by + field. Without per-action results there is no basis for a failed-action + transition (``next_tick``, ``next_action_index``, lifecycle), and arithmetic + on the counters has to agree with pre-existing in-flight work that + ``_due_contexts`` does not exclude. Reinstating is exact for both. + """ + + run.working = pre_batch + + +def _abandon_concurrent_batch( + batch: _ConcurrentBatch, + pre_batch: RuntimeSnapshot, + *, + code: str, + message: str, +) -> None: + """Report a backend batch that produced no usable results and undo its reservations.""" + + _restore_pre_batch_snapshot(batch.run, pre_batch) + _set_concurrent_failure( + batch.run, + Diagnostic( + code=code, + domain="participant", + address=batch.policy.address, + message=message, + ), + ) + + def _finish_concurrent_service_state( run: SchedulerRunState, policy_address: str, @@ -348,11 +390,35 @@ def _execute_concurrent_batch(batch: _ConcurrentBatch) -> None: _bound_action_request(context, batch.run.working, state) for context, state in zip(selected_contexts, selected_states, strict=True) ) + pre_batch = batch.run.working _reserve_concurrent_actions(batch.run, selected_contexts) base = batch.run.working - results = batch_method(requests, base, len(requests)) - if len(results) != len(requests): - raise ValueError("concurrent participant result count must match requests") + # The batch method is backend-supplied. A raising or miscounting backend is a + # conformance failure to report, not an exception to leak through the + # scheduler, and either way the reservations taken above must be released. + try: + results = batch_method(requests, base, len(requests)) + result_count = len(results) + except Exception as exc: # NOSONAR - backend trust boundary; any failure becomes a diagnostic + # Only the exception type crosses the boundary, matching + # `_backend_call_failed`: backend messages can carry host paths, + # credentials, or participant data that must not enter a portable + # diagnostic. + _abandon_concurrent_batch( + batch, + pre_batch, + code="runtime.participant-concurrent-batch-failed", + message=f"Backend concurrent participant batch did not complete ({type(exc).__name__}).", + ) + return + if result_count != len(requests): + _abandon_concurrent_batch( + batch, + pre_batch, + code="runtime.participant-concurrent-result-count-invalid", + message=f"Backend returned {result_count} concurrent participant results for {len(requests)} requests.", + ) + return for context, state, request, result in zip(selected_contexts, selected_states, requests, results, strict=True): _commit_concurrent_result(context, state, request, result, base, batch.run) if batch.run.failure is not None: diff --git a/implementations/python/tests/test_mcp_server.py b/implementations/python/tests/test_mcp_server.py index 57c73669..89f409ba 100644 --- a/implementations/python/tests/test_mcp_server.py +++ b/implementations/python/tests/test_mcp_server.py @@ -31,7 +31,7 @@ def _text(result) -> str: def _call(server, tool: str, args: dict | None = None) -> str: """Synchronously call a tool and return its text.""" - return asyncio.get_event_loop().run_until_complete(_async_call(server, tool, args or {})) + return asyncio.run(_async_call(server, tool, args or {})) async def _async_call(server, tool: str, args: dict) -> str: @@ -897,7 +897,7 @@ def test_server_has_all_tools(self): # Using the real registration surface (rather than a hand-copied # literal) means a drift between what the server exposes and what # raes_tool_surface advertises cannot pass silently. - registered = asyncio.get_event_loop().run_until_complete(server.list_tools()) + registered = asyncio.run(server.list_tools()) registered_names = {tool.name for tool in registered} assert registered_names, "server registered no tools" diff --git a/implementations/python/tests/test_participant_concurrent_batch_reservations.py b/implementations/python/tests/test_participant_concurrent_batch_reservations.py new file mode 100644 index 00000000..aa0c4471 --- /dev/null +++ b/implementations/python/tests/test_participant_concurrent_batch_reservations.py @@ -0,0 +1,204 @@ +"""Reservation release when a backend concurrent participant batch misbehaves.""" + +from __future__ import annotations + +from dataclasses import dataclass +from types import SimpleNamespace + +import pytest +import raes_runtime.participant_scheduler_operations as scheduler_operations +from raes_contracts.contracts import ParticipantAutonomousExecutionStateModel +from raes_contracts.contracts.participant_execution import ParticipantExecutionServiceStateModel +from raes_contracts.runtime_state import RuntimeSnapshot +from raes_runtime.participant_scheduler_concurrency import _execute_concurrent_batch +from raes_runtime.participant_scheduler_types import SchedulerRunState + +_POLICY_ADDRESS = "participant.autonomous-execution.green-users" +_IMPLEMENTATION_REF = "participant-implementation-manifests.green-worker.v1" +_PARTICIPANTS = ( + "participant.behavior.green-user-a", + "participant.behavior.green-user-b", +) + + +@dataclass(frozen=True) +class _StubContext: + """Minimal stand-in for `_DueActionContext` for the batch-failure paths.""" + + policy: object + participant_address: str + key: str + + +def _execution_state( + participant_address: str, + *, + in_flight: int = 0, +) -> ParticipantAutonomousExecutionStateModel: + return ParticipantAutonomousExecutionStateModel( + policy_address=_POLICY_ADDRESS, + policy_digest="sha256:" + "0" * 64, + participant_address=participant_address, + episode_id=f"{participant_address}-autonomous-0", + participant_implementation_ref=_IMPLEMENTATION_REF, + clock_address="time.clock.scenario-clock", + time_segment=0, + lifecycle_state="running", + next_tick=0, + next_action_index=0, + # attempted_actions == succeeded + failed + in_flight is a snapshot invariant. + attempted_actions=in_flight, + succeeded_actions=0, + failed_actions=0, + in_flight=in_flight, + ) + + +def _service_state(*, in_flight: int = 0) -> ParticipantExecutionServiceStateModel: + digest = "sha256:" + "0" * 64 + return ParticipantExecutionServiceStateModel( + execution_scope_ref="participant.execution-scope.green", + policy_address=_POLICY_ADDRESS, + desired_lifecycle="running", + observed_lifecycle="running", + generation=1, + observed_generation=1, + health="healthy", + readiness="ready", + accepting_new_work=True, + draining=False, + quiescent=in_flight == 0, + resources_released=False, + policy_digest=digest, + binding_digest=digest, + time_declaration_digest=digest, + capacity=2, + reserved=0, + in_flight=in_flight, + last_transition_ref=f"operation:{_POLICY_ADDRESS}:start:generation-1", + evidence_refs=("evidence.green-login.native-action",), + ) + + +def _state_key(participant_address: str) -> str: + return f"{_POLICY_ADDRESS}.state.{participant_address}" + + +def _batch(participant_runtime: object, *, in_flight: int = 0) -> SimpleNamespace: + policy = SimpleNamespace(address=_POLICY_ADDRESS, max_in_flight=2) + states = [_execution_state(address, in_flight=in_flight) for address in _PARTICIPANTS] + snapshot = RuntimeSnapshot( + participant_autonomous_execution_states={ + _state_key(address): state.model_dump(mode="json") + for address, state in zip(_PARTICIPANTS, states, strict=True) + }, + participant_execution_services={_POLICY_ADDRESS: _service_state(in_flight=in_flight).model_dump(mode="json")}, + ) + run = SchedulerRunState(working=snapshot, diagnostics=[], changed=[]) + contexts = [ + _StubContext(policy=policy, participant_address=address, key=_state_key(address)) for address in _PARTICIPANTS + ] + return SimpleNamespace( + policy=policy, + time_model=None, + participant_runtime=participant_runtime, + current_tick=0, + cadence_ticks=1, + run=run, + contexts=contexts, + states=states, + ) + + +@pytest.fixture(autouse=True) +def _stub_request_binding(monkeypatch: pytest.MonkeyPatch) -> None: + """Request construction is irrelevant to the batch-failure paths under test.""" + + monkeypatch.setattr( + scheduler_operations, + "_bound_action_request", + lambda context, working, state: SimpleNamespace(participant_address=context.participant_address), + ) + + +def _assert_reservations_released(run: SchedulerRunState, before: RuntimeSnapshot) -> None: + """The failure snapshot must equal the pre-batch snapshot exactly. + + A partly-applied occurrence (a recorded failure whose next_tick and + next_action_index never advanced) could be serviced again at the same tick, + and adjusting counters by hand has to stay consistent with in-flight work + from an earlier batch. + """ + + assert run.working == before + + +def test_miscounted_backend_batch_is_reported_and_releases_reservations(): + """A wrong result count previously raised, leaking every reservation. + + Reservations are taken before the backend call and only cleared per + committed result, so escaping here left participants permanently in-flight + and the service non-quiescent, blocking all later occurrences. + """ + batch = _batch(SimpleNamespace(admit_actions_concurrently=lambda requests, snapshot, workers: ())) + before = batch.run.working + + _execute_concurrent_batch(batch) + + assert batch.run.failure is not None + codes = [diagnostic.code for diagnostic in batch.run.failure.diagnostics] + assert "runtime.participant-concurrent-result-count-invalid" in codes + _assert_reservations_released(batch.run, before) + + +def test_raising_backend_batch_is_reported_and_releases_reservations(): + """A raising backend is a conformance failure, not an exception to leak.""" + + def _explode(requests, snapshot, workers): + raise RuntimeError("backend exploded") + + batch = _batch(SimpleNamespace(admit_actions_concurrently=_explode)) + before = batch.run.working + + _execute_concurrent_batch(batch) + + assert batch.run.failure is not None + codes = [diagnostic.code for diagnostic in batch.run.failure.diagnostics] + assert "runtime.participant-concurrent-batch-failed" in codes + # Only the exception type may cross the backend boundary. + message = next( + d.message for d in batch.run.failure.diagnostics if d.code == "runtime.participant-concurrent-batch-failed" + ) + assert "backend exploded" not in message + assert "RuntimeError" in message + _assert_reservations_released(batch.run, before) + + +def test_rollback_withdraws_only_this_batch_reservation(): + """Pre-existing in-flight work must survive a failed batch rollback. + + `_due_contexts` does not require `in_flight == 0`, so a participant can be + due while an earlier action is still outstanding. Clearing the aggregate + would erase that earlier work from the counters. + """ + batch = _batch( + SimpleNamespace(admit_actions_concurrently=lambda requests, snapshot, workers: ()), + in_flight=1, + ) + before = batch.run.working + + _execute_concurrent_batch(batch) + + assert batch.run.failure is not None + for address in _PARTICIPANTS: + state = ParticipantAutonomousExecutionStateModel.model_validate( + batch.run.working.participant_autonomous_execution_states[_state_key(address)] + ) + assert (state.in_flight, state.attempted_actions) == (1, 1) + assert state.attempted_actions == state.succeeded_actions + state.failed_actions + state.in_flight + # Service readback must not claim zero in-flight while earlier work is live. + service = ParticipantExecutionServiceStateModel.model_validate( + batch.run.working.participant_execution_services[_POLICY_ADDRESS] + ) + assert (service.in_flight, service.quiescent) == (1, False) + assert batch.run.working == before diff --git a/implementations/python/tests/test_runtime_control_plane_api.py b/implementations/python/tests/test_runtime_control_plane_api.py index f973d2b0..6b92970f 100644 --- a/implementations/python/tests/test_runtime_control_plane_api.py +++ b/implementations/python/tests/test_runtime_control_plane_api.py @@ -2,11 +2,13 @@ from __future__ import annotations +import asyncio import textwrap from pathlib import Path 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 ( @@ -26,12 +28,15 @@ 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_guards import request_size_guard_response 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 @@ -574,6 +579,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"} + + +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 body guard must cap while streaming, not buffer the whole payload first. + + A request without ``content-length`` (e.g. ``Transfer-Encoding: chunked``) is + invisible to the content-length guard, so the body guard is the only limit. If + it buffers the full body before measuring it, an unbounded chunked upload can + exhaust memory even though the response is still 413. + """ + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + 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} + + request = Request( + { + "type": "http", + "method": "POST", + "path": "/operations/provisioning", + "headers": [(b"content-type", b"application/json")], + "query_string": b"", + }, + receive=receive, + ) + + response = asyncio.run( + request_size_guard_response(control_plane, request, max_request_bytes=64), + ) + + assert response is not None + assert response.status_code == 413 + # Rejected after crossing the cap, not after draining all 1000 chunks. + assert delivered <= 3 + + +def test_request_size_guard_rejects_a_single_oversized_chunk_without_buffering_it(): + """One chunk larger than the cap must be refused before it is copied.""" + target = create_stub_target() + control_plane = RuntimeControlPlane(target) + oversized = b"x" * 4096 + delivered = 0 + + async def receive() -> dict[str, object]: + nonlocal delivered + delivered += 1 + return {"type": "http.request", "body": oversized, "more_body": False} + + request = Request( + { + "type": "http", + "method": "POST", + "path": "/operations/provisioning", + "headers": [(b"content-type", b"application/json")], + "query_string": b"", + }, + receive=receive, + ) + + response = asyncio.run( + request_size_guard_response(control_plane, request, max_request_bytes=64), + ) + + assert response is not None + assert response.status_code == 413 + assert not hasattr(request.state, "raw_body") + + def test_local_control_plane_store_saves_snapshot_with_atomic_replace( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py new file mode 100644 index 00000000..f5dcbf1a --- /dev/null +++ b/implementations/python/tests/test_runtime_workflow_timeout_reconciliation.py @@ -0,0 +1,119 @@ +"""Workflow timeout reconciliation edge cases for the runtime control plane.""" + +from __future__ import annotations + +import pytest +from raes_contracts.planning import RuntimeDomain +from raes_contracts.runtime_state import RuntimeSnapshot, SnapshotEntry +from raes_contracts.workflow import WorkflowExecutionState, WorkflowStatus +from raes_runtime.control_plane_timeouts import ( + TIMED_OUT_REASON, + UNPARSEABLE_START_REASON, + workflow_timeout_update, +) + +_WORKFLOW_ADDRESS = "orchestration.workflow.response" + + +def _workflow_entry_with_timeout(timeout_seconds: int) -> SnapshotEntry: + return SnapshotEntry( + address=_WORKFLOW_ADDRESS, + domain=RuntimeDomain.ORCHESTRATION, + resource_type="workflow", + payload={"execution_contract": {"timeout_seconds": timeout_seconds}}, + ) + + +def _workflow_entry(timeout_seconds: int = 1) -> SnapshotEntry: + return SnapshotEntry( + address=_WORKFLOW_ADDRESS, + domain=RuntimeDomain.ORCHESTRATION, + resource_type="workflow", + payload={"execution_contract": {"timeout_seconds": timeout_seconds}}, + ) + + +def _running_result(started_at: str) -> dict[str, object]: + """Build a persisted RUNNING workflow payload with ``started_at`` as recorded. + + The payload is edited after construction because the model rejects values it + considers unusable, while reconciliation reads snapshots back through + ``WorkflowExecutionState.from_payload`` and must cope with whatever the store + actually holds. + """ + + payload = WorkflowExecutionState( + workflow_status=WorkflowStatus.RUNNING, + run_id="run-1", + started_at="2000-01-01T00:00:00Z", + updated_at="2000-01-01T00:00:00Z", + ).to_payload() + payload["started_at"] = started_at + return payload + + +def _reconcile(started_at: str, submitted_at: str) -> tuple[dict[str, object], list[dict[str, object]]] | None: + return workflow_timeout_update( + RuntimeSnapshot(), + _WORKFLOW_ADDRESS, + _workflow_entry(), + {_WORKFLOW_ADDRESS: _running_result(started_at)}, + {}, + submitted_at, + ) + + +def test_expired_workflow_is_timed_out(): + update = _reconcile("2000-01-01T00:00:00Z", "2000-01-01T00:01:00Z") + + assert update is not None + assert update[0]["workflow_status"] == WorkflowStatus.TIMED_OUT.value + assert update[0]["terminal_reason"] == TIMED_OUT_REASON + + +def test_workflow_inside_its_deadline_is_left_running(): + assert _reconcile("2000-01-01T00:00:00Z", "2000-01-01T00:00:00Z") is None + + +@pytest.mark.parametrize("started_at", ["None", "not-a-timestamp", "2000-13-45T99:99:99Z"]) +def test_workflow_with_unparseable_started_at_is_reclaimed(started_at: str): + """A running workflow with no derivable deadline must not stay RUNNING forever. + + Swallowing the parse failure and reporting "not timed out" pinned such a + workflow in RUNNING for the lifetime of the control plane, so reconciliation + could never reclaim it. + """ + update = _reconcile(started_at, "2000-01-01T00:01:00Z") + + assert update is not None + assert update[0]["workflow_status"] == WorkflowStatus.TIMED_OUT.value + assert update[0]["terminal_reason"] == UNPARSEABLE_START_REASON + + +def test_enormous_timeout_reports_not_timed_out_instead_of_overflowing(): + """`timeout_seconds` has no declared upper bound, so it must not overflow. + + Folding a very large timeout into a float timestamp or a timedelta raises + `OverflowError`, which would abort the whole reconciliation pass and surface + as a 500 from the HTTP adapter. + """ + update = workflow_timeout_update( + RuntimeSnapshot(), + _WORKFLOW_ADDRESS, + _workflow_entry_with_timeout(10**400), + {_WORKFLOW_ADDRESS: _running_result("2000-01-01T00:00:00Z")}, + {}, + "2030-01-01T00:00:00Z", + ) + + assert update is None + + +def test_unparseable_reconciliation_clock_is_raised_not_swallowed(): + """A bad caller-supplied ``now`` governs every workflow, so it must surface. + + Reported as ``ValueError``; the HTTP adapter maps that to 409 rather than + silently disabling timeouts for the whole pass. + """ + with pytest.raises(ValueError): + _reconcile("2000-01-01T00:00:00Z", "not-a-timestamp") diff --git a/implementations/python/tests/test_semantics_planner.py b/implementations/python/tests/test_semantics_planner.py index 4f74ba0d..1da53068 100644 --- a/implementations/python/tests/test_semantics_planner.py +++ b/implementations/python/tests/test_semantics_planner.py @@ -2,16 +2,21 @@ from __future__ import annotations +import itertools from types import SimpleNamespace -from hypothesis import given +from hypothesis import given, settings from hypothesis import strategies as st from raes_processor.semantics.planner import ( DependencyKind, + canonical_resource_identity, + dependency_cycles, dependency_edges, + dependency_graph, refresh_impacted_nodes, resource_delete_order, resource_topological_order, + topological_dependency_order, ) @@ -81,6 +86,78 @@ def _dag_resources_with_change_sets(draw): return resources, subset_a, subset_b +def _dependency_graphs() -> st.SearchStrategy[dict[str, tuple[str, ...]]]: + """Small graphs that freely admit self-loops and multi-node cycles.""" + + def _build(size: int, choices: list[list[int]]) -> dict[str, tuple[str, ...]]: + nodes = [f"nodes.host-{index}" for index in range(size)] + return { + node: tuple(nodes[target % size] for target in targets) + for node, targets in zip(nodes, choices, strict=True) + } + + # Deliberately larger and denser than the DAG strategies above: the explicit + # frame stack has to resume a partially consumed iterator, so the graphs that + # matter are the ones where a node still has unvisited dependencies left when + # a descent happens and some of those are already on the stack. + return st.integers(min_value=1, max_value=18).flatmap( + lambda size: st.lists( + st.lists(st.integers(min_value=0, max_value=17), max_size=6), + min_size=size, + max_size=size, + ).map(lambda choices: _build(size, choices)) + ) + + +def _reference_dependency_cycles( + dependencies_by_node: dict[str, tuple[str, ...]], +) -> list[tuple[str, ...]]: + """Recursive Tarjan reference for differential comparison. + + Kept deliberately naive: it mirrors the textbook recursion the production + walk replaced, so any behavioural drift in the explicit-stack version shows + up as a mismatch rather than as a silently different plan order. + """ + + graph = dependency_graph(dependencies_by_node) + if not graph: + return [] + counter = itertools.count() + indices: dict[str, int] = {} + lowlinks: dict[str, int] = {} + stack: list[str] = [] + on_stack: set[str] = set() + cycles: list[tuple[str, ...]] = [] + + def strongconnect(node: str) -> None: + indices[node] = lowlinks[node] = next(counter) + stack.append(node) + on_stack.add(node) + for dependency in graph[node]: + if dependency not in indices: + strongconnect(dependency) + lowlinks[node] = min(lowlinks[node], lowlinks[dependency]) + elif dependency in on_stack: + lowlinks[node] = min(lowlinks[node], indices[dependency]) + if lowlinks[node] != indices[node]: + return + component: list[str] = [] + while stack: + member = stack.pop() + on_stack.remove(member) + component.append(member) + if member == node: + break + component = sorted(component) + if len(component) > 1 or component[0] in graph[component[0]]: + cycles.append(tuple(component)) + + for node in sorted(graph, key=canonical_resource_identity): + if node not in indices: + strongconnect(node) + return sorted(cycles, key=lambda cycle: tuple(canonical_resource_identity(node) for node in cycle)) + + class TestPlannerSemantics: def test_dependency_edges_preserve_kinds(self): resources = { @@ -134,3 +211,40 @@ def test_refresh_propagation_is_monotonic(self, payload): impacted_b = subset_b | set(refresh_impacted_nodes(resources, subset_b)) assert impacted_a <= impacted_b + + +class TestDependencyCycleScale: + """Cycle detection must survive dependency chains longer than the recursion limit.""" + + _DEEP = 5000 + + def _chain(self, size: int) -> dict[str, tuple[str, ...]]: + graph: dict[str, tuple[str, ...]] = { + f"nodes.host-{index:05d}": (f"nodes.host-{index + 1:05d}",) for index in range(size) + } + graph[f"nodes.host-{size:05d}"] = () + return graph + + def test_deep_acyclic_chain_reports_no_cycles(self): + assert dependency_cycles(self._chain(self._DEEP)) == [] + + def test_deep_cycle_is_still_detected(self): + size = 3000 + graph = {f"nodes.host-{index:05d}": (f"nodes.host-{(index + 1) % size:05d}",) for index in range(size)} + + cycles = dependency_cycles(graph) + + assert len(cycles) == 1 + assert len(cycles[0]) == size + + def test_deep_chain_topological_order_is_complete(self): + graph = self._chain(self._DEEP) + + assert len(topological_dependency_order(graph)) == len(graph) + + @settings(max_examples=400) + @given(_dependency_graphs()) + def test_cycle_detection_matches_a_reference_walk(self, graph): + """Guards the explicit-stack walk against the recursive semantics it replaced.""" + + assert dependency_cycles(graph) == _reference_dependency_cycles(graph)