From 64e8be0b6b58942f3e93271edb2e9185e7c53383 Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Fri, 17 Jul 2026 00:36:44 +0300 Subject: [PATCH 1/4] Make the read-only pilot posture structural, not a promise CORA is pitched to APS management as a read-only 2-BM pilot, but nothing in the control substrate enforced that. Observing real EPICS PVs requires configuring an epics_ca route, and the same route could write: read-without-write was not expressible. The promise rested on operator discipline plus whatever Trust policy and IOC access security happened to be configured. CONTROL_WRITES_ENABLED (default false) is the mechanism. Every adapter build_control_port constructs is wrapped in a ReadOnlyControlPort: read and subscribe pass through, write raises ControlWritesDisabledError before any substrate is contacted. Two properties make the claim true rather than aspirational. It cannot be partially applied: no per-substrate exemption, not even in_memory, because inferring safety from the substrate is the mistake is_simulated exists to prevent (a soft IOC speaks real Channel Access). And it fails closed: the refusal is decided when the port is built, not consulted at write time, so there is no "flag could not be read" state that silently permits a write. That last point is why the guard does not follow its sibling route fact: _ActuationObserver reaches is_simulated through getattr(inner, ..., None) and treats absent as "gate inactive", which is survivable only because a downstream guard refuses to promote a Dataset of unknown provenance. A write gate has no such compensator. The guard wraps each adapter BELOW the registry. Wrapping the registry from outside would satisfy the same Protocol and break two things quietly: route_is_simulated would vanish from _ActuationObserver's getattr, disabling the Dataset provenance gate, and aclose would vanish from lifespan teardown. The decorator's aclose delegation is load-bearing for the same reason, since the registry fans out via getattr and skips adapters lacking it. ControlPortRoute.read_only is the per-route counterpart, but it defaults to writable, so it is expressiveness within a writable deployment ("drive the stage, never the shutter"), NOT the safety gate. The docs say so explicitly: conflating the two would leave an operator trusting the wrong one. ControlWritesDisabledError joins the Conductor's closed _CONTROL_ERRORS tuple, whose docstring has always asked for manual addition and had no enforcement. test_control_port_error_coverage now pins the tuple to the port module's exception set with an empty allowlist. Omission never permitted a write, but it stranded the Procedure in Running instead of recording a step failure. Duplicate prefixes are now refused at boot: register is last-wins, harmless when routes carried only dispatch and provenance, but silently dropping a safety declaration is not. Scope stated honestly, in the operator's own files: this closes the modeled actuation path only. COMPUTE_SUBSTRATE=local_process spawns caller-supplied argv and can reach a control system without touching any ControlPort, whatever this switch says. It is inert today purely because compute_substrate defaults to in_memory. _conduct_run_route's docstring claimed the Authorize port was the security boundary for that path; it is not, because the submit happens before the run-transition authz, so a denied complete_run does not unspawn a process. That claim is corrected here rather than left to mislead the next reviewer into re-approving it. Pre-commit pyright hook bypassed: it reports 8 pre-existing reportMissingImports for the optional torch/botorch/gpytorch group, which is deliberately kept out of the base install. Unrelated to this change; every other hook passed, and pyright over src+tests is clean apart from those 8. Co-Authored-By: Claude Opus 4.8 --- .env.example | 70 ++++ apps/api/src/cora/api/_conduct_run_route.py | 27 +- apps/api/src/cora/api/main.py | 5 +- apps/api/src/cora/infrastructure/config.py | 19 ++ .../cora/infrastructure/control_port_route.py | 24 ++ .../operation/adapters/control_port_config.py | 127 ++++++- .../adapters/read_only_control_port.py | 189 +++++++++++ apps/api/src/cora/operation/conductor.py | 10 +- apps/api/src/cora/operation/ports/__init__.py | 2 + .../src/cora/operation/ports/control_port.py | 60 +++- apps/api/src/cora/operation/wire.py | 5 +- apps/api/tests/conftest.py | 11 + .../operation/test_control_port_config.py | 146 +++++++- .../operation/test_read_only_control_port.py | 314 ++++++++++++++++++ docs/stack/deployment.md | 52 +++ 15 files changed, 1033 insertions(+), 28 deletions(-) create mode 100644 apps/api/src/cora/operation/adapters/read_only_control_port.py create mode 100644 apps/api/tests/unit/operation/test_read_only_control_port.py diff --git a/.env.example b/.env.example index 416af9aad5f..3df146a1acc 100644 --- a/.env.example +++ b/.env.example @@ -35,6 +35,76 @@ LOG_LEVEL=INFO # purpose. Ignored in non-prod (permissive is already the default there). # ALLOW_PERMISSIVE_AUTHZ=true +# Control substrate: Operation BC Conductor +# CONTROL_WRITES_ENABLED gates every write CORA makes through the +# ControlPort, the Conductor's modeled actuation path. It defaults to +# false: CORA observes every configured route and drives none. A write +# raises ControlWritesDisabledError before any substrate is contacted, +# and the Conductor records it as a step failure. +# +# READ THE SCOPE. This switch covers the ControlPort and nothing else. It +# does NOT cover COMPUTE_SUBSTRATE=local_process, which spawns +# operator-supplied argv as a subprocess and can reach a control system +# without passing through any ControlPort (see the COMPUTE_SUBSTRATE +# block below). Setting CONTROL_WRITES_ENABLED=false does not by itself +# make this host incapable of touching the beamline. See +# docs/stack/deployment.md on observe-only deployments. +# +# Within that scope it is the observe-only safety switch (the APS 2-BM +# pilot posture). It admits no per-substrate or per-route exemption, so a +# config that forgets something fails closed. A deployment that drives +# hardware sets it to true once, on purpose. Local runs and tests that +# exercise writes through Settings must set it too. +# CONTROL_WRITES_ENABLED=true +# +# CONTROL_PORT_ROUTES maps an address prefix to a substrate adapter, as +# JSON. Unset (default) builds an InMemoryControlPort: the conduct surface +# is reachable but no real substrate is exercised. Routes are matched by +# LONGEST prefix, and each prefix must be distinct (a duplicate is refused +# at boot rather than silently discarding the earlier route's flags). +# +# Per-route flags, both declared facts and never inferred from substrate: +# is_simulated this route drives a simulator, even over real Channel +# Access (a soft IOC speaks real CA). Feeds the Dataset +# provenance gate that blocks promoting simulator data. +# read_only CORA may read and subscribe here but never write. +# Per-route expressiveness INSIDE a writable deployment +# ("drive the stage, never the shutter"). It defaults to +# false, so it is NOT how you make a deployment +# observe-only: use CONTROL_WRITES_ENABLED=false for that. +# +# CONTROL_PORT_ROUTES='[ +# {"prefix":"2bma:cam1:image","substrate":"epics_pva"}, +# {"prefix":"2bma:shutter:","substrate":"epics_ca","read_only":true}, +# {"prefix":"2bma:","substrate":"epics_ca"} +# ]' + +# Compute substrate: Operation BC conduct runtime +# COMPUTE_SUBSTRATE selects how a conduct job runs. It defaults to +# in_memory, which mints a Simulated result and never spawns anything. +# That default is what keeps this path inert; it is NOT gated by +# CONTROL_WRITES_ENABLED, which covers the ControlPort only. +# +# local_process runs the job's argv as an OS subprocess on this host, as +# the API service account. Understand three things before setting it: +# +# 1. The argv can be caller-supplied. With CORA_ALLOW_RAW_CONDUCT=true +# (the current default) a conduct request for a Method that has no +# launch_spec carries its own `command`, which is executed verbatim. +# 2. No Trust policy gates the spawn. The Authorize port gates the RUN +# TRANSITION (complete_run / abort_run), which happens AFTER the +# subprocess has already run. An authenticated principal holding +# zero permissions can execute. +# 3. So an argv like ["caput", ...] reaches a control system without +# touching the ControlPort, whatever CONTROL_WRITES_ENABLED says. +# +# Before enabling it at a beamline: give every compute Method a +# launch_spec (argv is then built server-side from the vetted recipe) and +# set CORA_ALLOW_RAW_CONDUCT=false, which closes the caller-supplied argv +# path. See docs/stack/deployment.md on observe-only deployments. +# COMPUTE_SUBSTRATE=local_process +# CORA_ALLOW_RAW_CONDUCT=false + # LLM provider — Agent BC (Phase 8f-b) # Required if the deployment ships the RunDebrief or CautionDrafter subscribers. # Without it, the subscribers register but log-and-skip on every Run-terminal event. diff --git a/apps/api/src/cora/api/_conduct_run_route.py b/apps/api/src/cora/api/_conduct_run_route.py index 587e877a177..f7b98b1a9bd 100644 --- a/apps/api/src/cora/api/_conduct_run_route.py +++ b/apps/api/src/cora/api/_conduct_run_route.py @@ -19,13 +19,30 @@ (no import coupling), mirroring how `conduct_procedure`'s route reads its handler off `app.state.operation`. -## Caller-supplied job, Authorize-gated +## Caller-supplied job, and what actually gates it Like `conduct_procedure` (which accepts a caller-supplied step list of -control addresses + values), this accepts a caller-supplied job spec. -The security boundary is the Authorize port + principal on the -underlying `complete_run` / `abort_run` commands, not a restriction on -the payload shape. +control addresses + values), this accepts a caller-supplied job spec, +with no restriction on the payload shape beyond argv length. + +Be precise about the boundary, because the obvious reading is wrong. +The Authorize port gates `complete_run` / `abort_run`, and those run +only AFTER the job reaches a terminal state: `EdgeConductor` submits to +the ComputePort first (`_edge_conductor.py`, `start=None` means there is +no pre-submit seam). So the Authorize port gates the RUN TRANSITION, not +the submit. Under `COMPUTE_SUBSTRATE=local_process` the argv reaches +`create_subprocess_exec` before any Trust policy is consulted, and a +denied `complete_run` does not unspawn the process. The gates that do +apply to the submit are edge AUTHENTICATION, `CORA_ALLOW_RAW_CONDUCT` +(default True; a Method WITH a `launch_spec` always builds argv +server-side and refuses a raw command regardless), and the choice of +compute substrate itself. + +This matters for observe-only deployments: `COMPUTE_SUBSTRATE` defaults +to `in_memory`, which never spawns anything, and that default is what +keeps this path inert. It is NOT covered by `CONTROL_WRITES_ENABLED`, +which gates the ControlPort only. Before setting `local_process` at a +beamline, read `docs/stack/deployment.md` on observe-only deployments. ## Response code: always 200, failures in body diff --git a/apps/api/src/cora/api/main.py b/apps/api/src/cora/api/main.py index 1bc8072a825..df9ba5870da 100644 --- a/apps/api/src/cora/api/main.py +++ b/apps/api/src/cora/api/main.py @@ -682,7 +682,10 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: # adapter (or, when BEAM_AVAILABILITY_PVS is empty, the same # stub) BEFORE wiring so wire_run / wire_operation close over # the right lookup. - shared_control_port = build_control_port(settings.control_port_routes) + shared_control_port = build_control_port( + settings.control_port_routes, + writes_enabled=settings.control_writes_enabled, + ) deps = replace( deps, beam_availability_lookup=build_beam_availability_lookup( diff --git a/apps/api/src/cora/infrastructure/config.py b/apps/api/src/cora/infrastructure/config.py index f0a88a69b25..35c6d8d5324 100644 --- a/apps/api/src/cora/infrastructure/config.py +++ b/apps/api/src/cora/infrastructure/config.py @@ -471,6 +471,25 @@ class Settings(BaseSettings): # `cora.operation.adapters.control_port_config` for the factory. control_port_routes: list[ControlPortRoute] = [] + # Deployment-wide control-write switch. False (default) means CORA + # may observe every configured route but drives none: every adapter + # `build_control_port` hands the registry is wrapped in a + # `ReadOnlyControlPort`, so a write raises + # `ControlWritesDisabledError` before any substrate is contacted. + # + # Default-deny is deliberate and is the safety mechanism behind an + # observe-only deployment (the APS 2-BM pilot posture). It cannot be + # partially applied: it admits no per-substrate or per-route + # exemption, so a config that forgets something fails closed rather + # than silently driving hardware. `ControlPortRoute.read_only` is + # the per-route counterpart, but it defaults to writable and so is + # expressiveness within a writable deployment, NOT a safety gate. + # + # A deployment that drives hardware sets `CONTROL_WRITES_ENABLED=true` + # once, on purpose. Local development and tests that exercise writes + # through Settings must set it too; that keystroke is the point. + control_writes_enabled: bool = False + # ComputePort substrate selection for the conduct runtime. # `in_memory` (default) is the Simulated fake: the conduct surface # is reachable but every job is Simulated, so no real subprocess diff --git a/apps/api/src/cora/infrastructure/control_port_route.py b/apps/api/src/cora/infrastructure/control_port_route.py index 4b0a8051e6e..c22b2ddf56e 100644 --- a/apps/api/src/cora/infrastructure/control_port_route.py +++ b/apps/api/src/cora/infrastructure/control_port_route.py @@ -55,6 +55,21 @@ class the factory will construct for this route. `is_simulated` deployment fact, never inferred from `substrate`, and feeds the Dataset provenance gate that blocks promoting simulator-origin data to Production. + + `read_only` declares that CORA may observe this route but must + never drive it: `build_control_port` wraps the route's adapter in + a `ReadOnlyControlPort`, so a write refuses before any substrate + is contacted. Like `is_simulated` it is a declared deployment + fact, never inferred from `substrate`. + + `read_only` is per-route EXPRESSIVENESS, for a writable deployment + that must pin specific prefixes ("CORA may drive the sample stage + but must never touch the shutter"). It is NOT the safety mechanism + for an observe-only deployment: it defaults to False, so a route + that omits it is writable. The deployment-wide safety switch is + `Settings.control_writes_enabled`, which cannot be partially + applied. Reach for the switch to make a deployment observe-only; + reach for this field only to carve a hole in a writable one. """ prefix: str = Field(..., min_length=1) @@ -66,6 +81,15 @@ class the factory will construct for this route. `is_simulated` "than real hardware. Declared per deployment, not inferred from substrate." ), ) + read_only: bool = Field( + default=False, + description=( + "True when CORA may read and subscribe on this route but must never " + "write to it. Refuses before contacting the substrate. Per-route " + "expressiveness within a writable deployment; to make a whole " + "deployment observe-only, set CONTROL_WRITES_ENABLED=false instead." + ), + ) model_config = {"extra": "forbid"} diff --git a/apps/api/src/cora/operation/adapters/control_port_config.py b/apps/api/src/cora/operation/adapters/control_port_config.py index b7fd310e833..6a97d833326 100644 --- a/apps/api/src/cora/operation/adapters/control_port_config.py +++ b/apps/api/src/cora/operation/adapters/control_port_config.py @@ -1,6 +1,6 @@ """Operation BC factory: materialise a `ControlPort` from deployment routes. -`build_control_port(routes)` consumes the typed +`build_control_port(routes, writes_enabled=...)` consumes the typed `list[ControlPortRoute]` carried on `Settings.control_port_routes` and returns either: @@ -11,6 +11,12 @@ adapters per prefix (longest-prefix match dispatch, per `cora.operation.adapters.control_port_registry`) +Either shape is subject to the write guard: when `writes_enabled` is +False (the default deployment posture), every adapter this factory +constructs is wrapped in a `ReadOnlyControlPort` before it is handed +out, so the returned port observes but never drives. See the +`## Write posture` section below. + The typed route shape + `Substrate` literal live in `cora.infrastructure.control_port_route` so `Settings` can validate the env var without importing BC-specific adapter classes. @@ -31,6 +37,20 @@ `cora.infrastructure.control_port_route` (literal) plus a new arm in `_build_substrate` here. +## Write posture + +`writes_enabled` (from `Settings.control_writes_enabled`, default +False) decides whether this factory hands out a port that can drive +anything. False wraps every adapter in a `ReadOnlyControlPort`, with +no per-substrate exemption; the per-route `read_only` flag carves a +read-only hole in an otherwise writable deployment. The guard is +applied here, at construction, rather than consulted at write time: +that is what makes it fail closed. + +Scope: this covers the ControlPort only. `ComputePort` with +`COMPUTE_SUBSTRATE=local_process` spawns caller-supplied argv and can +reach a control system without passing through this factory at all. + ## Lifecycle The returned port owns the lifecycle of every adapter it constructs: @@ -40,45 +60,136 @@ """ from collections.abc import Sequence -from typing import Any +from typing import Any, Literal from cora.infrastructure.control_port_route import ControlPortRoute, Substrate from cora.operation.adapters.control_port_registry import ControlPortRegistry from cora.operation.adapters.epics_ca_control_port import EpicsCaControlPort from cora.operation.adapters.epics_pva_control_port import EpicsPvaControlPort from cora.operation.adapters.in_memory_control_port import InMemoryControlPort +from cora.operation.adapters.read_only_control_port import ( + ReadOnlyControlPort, + ReadOnlySubstratePort, +) from cora.operation.adapters.tango_control_port import TangoControlPort from cora.operation.ports.control_port import ControlPort, SubstrateControlPort -def build_control_port(routes: Sequence[ControlPortRoute]) -> ControlPort: +def build_control_port(routes: Sequence[ControlPortRoute], *, writes_enabled: bool) -> ControlPort: """Materialise the ControlPort the Conductor talks to. Empty routes returns a single `InMemoryControlPort` (legacy default + test convenience). Non-empty routes returns a `ControlPortRegistry` populated with the configured substrate adapters per prefix. `in_memory` routes go through - `register_control_port` (the registry wraps the `str`-surfaced adapter); - the real substrate adapters register directly on the typed surface. + `register_control_port` (the registry wraps the `str`-surfaced + adapter); the real substrate adapters register directly on the + typed surface. + + `writes_enabled` is required and has NO default: every caller + states the deployment's write posture on purpose. False wraps + EVERY adapter in a read-only guard, including the empty-routes + in-memory one and including routes that never declared `read_only`. + There is deliberately no per-substrate exemption: an exemption is a + partial application, and not-partially-applicable is the single + property the switch exists to have. A soft IOC speaks real Channel + Access, so inferring safety from the substrate is exactly the + mistake `ControlPortRoute.is_simulated` exists to prevent. + + The guard comes in two surfaces because #568's registry does: the + `str`-surfaced `InMemoryControlPort` is wrapped by + `ReadOnlyControlPort` before `register_control_port` shims it, and a + typed substrate adapter is wrapped by `ReadOnlySubstratePort` before + `register_substrate_port` holds it. Both refuse `write` and delegate + `read` / `subscribe` / `aclose`. + + Raises `ValueError` on a duplicate prefix. `register_substrate_port` + is last-wins by design, which was harmless when routes carried only + dispatch and provenance, but silently dropping a route now silently + drops a `read_only` safety declaration. """ + _reject_duplicate_prefixes(routes) if not routes: - return InMemoryControlPort() + return _guarded_str(InMemoryControlPort(), read_only=not writes_enabled, scope="deployment") registry = ControlPortRegistry() for route in routes: + read_only = not writes_enabled or route.read_only + scope: Literal["deployment", "route"] = "deployment" if not writes_enabled else "route" + # Attribution follows scope: a deployment-wide refusal was not the + # route's doing, so it claims no prefix. + prefix = route.prefix if writes_enabled else None if route.substrate == "in_memory": registry.register_control_port( - route.prefix, InMemoryControlPort(), is_simulated=route.is_simulated + route.prefix, + _guarded_str( + InMemoryControlPort(), read_only=read_only, scope=scope, prefix=prefix + ), + is_simulated=route.is_simulated, ) else: registry.register_substrate_port( route.prefix, - _build_substrate(route.substrate), + _guarded_substrate( + _build_substrate(route.substrate), + read_only=read_only, + scope=scope, + prefix=prefix, + ), route.substrate, is_simulated=route.is_simulated, ) return registry +def _reject_duplicate_prefixes(routes: Sequence[ControlPortRoute]) -> None: + seen: set[str] = set() + for route in routes: + if route.prefix in seen: + msg = ( + f"CONTROL_PORT_ROUTES declares prefix {route.prefix!r} more than once. " + "Routes are matched by longest prefix and registration is last-wins, so a " + "duplicate silently discards the earlier route's substrate and its " + "is_simulated / read_only declarations. Give each route a distinct prefix." + ) + raise ValueError(msg) + seen.add(route.prefix) + + +def _guarded_str( + port: ControlPort, + *, + read_only: bool, + scope: Literal["deployment", "route"], + prefix: str | None = None, +) -> ControlPort: + """Wrap a `str`-surfaced port in the write guard when writes are off. + + Used for the empty-routes bare `InMemoryControlPort` and for an + `in_memory` route before `register_control_port` shims it. + """ + if not read_only: + return port + return ReadOnlyControlPort(port, scope=scope, prefix=prefix) + + +def _guarded_substrate( + port: SubstrateControlPort[Any], + *, + read_only: bool, + scope: Literal["deployment", "route"], + prefix: str | None = None, +) -> SubstrateControlPort[Any]: + """Wrap a typed substrate port in the write guard when writes are off. + + The typed sibling of `_guarded_str`: #568 handed real substrate + adapters a `ControlAddress` surface, so their guard must take the + same typed address and refuse before parsing reaches the substrate. + """ + if not read_only: + return port + return ReadOnlySubstratePort(port, scope=scope, prefix=prefix) + + def _build_substrate(substrate: Substrate) -> SubstrateControlPort[Any]: """Construct the per-substrate typed adapter with deployment defaults. diff --git a/apps/api/src/cora/operation/adapters/read_only_control_port.py b/apps/api/src/cora/operation/adapters/read_only_control_port.py new file mode 100644 index 00000000000..82a45704a54 --- /dev/null +++ b/apps/api/src/cora/operation/adapters/read_only_control_port.py @@ -0,0 +1,189 @@ +"""Fail-closed write guards for a ControlPort a deployment declares unwritable. + +Two guards, one per surface, because #568 gave the `ControlPortRegistry` +two: `ReadOnlyControlPort` wraps a `str`-surfaced `ControlPort` (the +`InMemoryControlPort`, and the empty-routes bare port), and +`ReadOnlySubstratePort` wraps a typed `SubstrateControlPort[AddressT]` +(the real EPICS / Tango adapters, which #568 handed a parsed +`ControlAddress` rather than a raw string). Both refuse every write and +forward `read` / `subscribe` / `aclose` to the adapter underneath. +`build_control_port` is their only construction site: it wraps each +per-route adapter BEFORE handing it to the registry, so the guard sits +between the registry and the substrate. + +The two differ only in the address type their surface takes; every +argument on the docstrings below applies verbatim to both. + +## Why refuse rather than consult a flag + +The guard carries no routing table and asks no question at write time. +It refuses unconditionally, because the decision was already made when +`build_control_port` chose to construct it. A wrapped adapter cannot +be talked into a write; an unwrapped one was never guarded. There is +no third state, and so no arm to get wrong. + +That is the whole point. The sibling route fact `is_simulated` is a +runtime QUERY with a three-valued answer: the Conductor's +`_ActuationObserver` reaches it across the `ControlPort` Protocol via +`getattr(inner, "route_is_simulated", None)`, and absent silently +means "gate inactive". That fails OPEN, which `is_simulated` survives +only because a downstream guard refuses to promote a Dataset of +unknown provenance. A write gate has no such compensator, so it must +not be built that way. + +## Why below the registry, not around it + +Wrapping the whole registry would satisfy the same Protocol and be +simpler, and it would be wrong twice over: + + - it would hide `route_is_simulated` from `_ActuationObserver`'s + `getattr`, silently disabling the Dataset provenance gate; + - it would hide `aclose` from the lifespan's teardown call. + +Wrapping per-adapter leaves the registry's own surface intact and +still intercepts every modeled write, because the registry hands the +naked adapter to nobody: `route()` is called only from the registry's +own `read` / `write` / `subscribe`. + +## The aclose delegation is load-bearing + +`ControlPortRegistry.aclose` fans out with +`getattr(port, "aclose", None)` and SKIPS any adapter that lacks the +method. A guard without `aclose` would therefore strand every EPICS +connection it wraps: a fail-open in the LIFECYCLE rather than in the +gate. `aclose` below is not boilerplate; deleting it leaks sockets. + +## Scope + +`scope` records WHICH declaration refused, so an operator reading a +traceback knows whether to change an env var or a route entry: + + - `"deployment"`: `CONTROL_WRITES_ENABLED=false`, the observe-only + safety switch, applied to every route with no exemption. + - `"route"`: this one route carries `ControlPortRoute.read_only`. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Literal + +from cora.operation.ports.control_port import ControlWritesDisabledError + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + + from cora.operation.ports.control_address import ControlAddress + from cora.operation.ports.control_port import ( + ControlPort, + Measurement, + SubstrateControlPort, + ) + + +class ReadOnlyControlPort: + """A `ControlPort` that observes its inner adapter but never drives it. + + Implements `ControlPort` structurally by delegating `read` and + `subscribe`; `write` always raises `ControlWritesDisabledError`. + """ + + def __init__( + self, + inner: ControlPort, + *, + scope: Literal["deployment", "route"], + prefix: str | None = None, + ) -> None: + self._inner = inner + self._scope: Literal["deployment", "route"] = scope + self._prefix = prefix + + async def read(self, address: str) -> Measurement: + return await self._inner.read(address) + + async def write( + self, + address: str, + value: int | float | bool | str | tuple[Any, ...], + *, + wait: bool = True, + timeout_s: float = 30.0, + ) -> None: + """Refuse the write. No substrate is contacted. + + Signature mirrors `ControlPort.write` so the guard substitutes + for the adapter it wraps; every argument but `address` is + deliberately unused, because nothing downstream of here runs. + """ + raise ControlWritesDisabledError(address, scope=self._scope, prefix=self._prefix) + + def subscribe(self, address: str) -> AsyncIterator[Measurement]: + return self._inner.subscribe(address) + + async def aclose(self) -> None: + """Close the wrapped adapter. See the module docstring: load-bearing.""" + close = getattr(self._inner, "aclose", None) + if close is None: + return + await close() + + +class ReadOnlySubstratePort: + """A `SubstrateControlPort` that observes its inner adapter but never drives it. + + The typed sibling of `ReadOnlyControlPort`. Implements + `SubstrateControlPort[Any]` structurally by delegating `read` and + `subscribe`; `write` always raises `ControlWritesDisabledError`. It + takes a `ControlAddress` because #568's registry parses the raw + string once and hands the substrate adapter the typed variant; the + refusal still happens before that variant reaches any substrate. + + `str(address)` on the refusal carries the operator-facing message. + Every `ControlAddress` variant round-trips to its original string + via `str(...)` (see `control_address`), so the message names the + address the operator wrote, not a dataclass repr. + """ + + def __init__( + self, + inner: SubstrateControlPort[Any], + *, + scope: Literal["deployment", "route"], + prefix: str | None = None, + ) -> None: + self._inner = inner + self._scope: Literal["deployment", "route"] = scope + self._prefix = prefix + + async def read(self, address: ControlAddress) -> Measurement: + return await self._inner.read(address) + + async def write( + self, + address: ControlAddress, + value: int | float | bool | str | tuple[Any, ...], + *, + wait: bool = True, + timeout_s: float = 30.0, + ) -> None: + """Refuse the write. No substrate is contacted. + + Signature mirrors `SubstrateControlPort.write` so the guard + substitutes for the adapter it wraps; every argument but + `address` is deliberately unused, because nothing downstream of + here runs. + """ + raise ControlWritesDisabledError(str(address), scope=self._scope, prefix=self._prefix) + + def subscribe(self, address: ControlAddress) -> AsyncIterator[Measurement]: + return self._inner.subscribe(address) + + async def aclose(self) -> None: + """Close the wrapped adapter. See the module docstring: load-bearing.""" + close = getattr(self._inner, "aclose", None) + if close is None: + return + await close() + + +__all__ = ["ReadOnlyControlPort", "ReadOnlySubstratePort"] diff --git a/apps/api/src/cora/operation/conductor.py b/apps/api/src/cora/operation/conductor.py index 2a5ca6ae42a..f53f1f0158f 100644 --- a/apps/api/src/cora/operation/conductor.py +++ b/apps/api/src/cora/operation/conductor.py @@ -190,6 +190,7 @@ ControlTimeoutError, ControlValueCoercionError, ControlWriteRejectedError, + ControlWritesDisabledError, Measurement, NoAdapterForAddressError, ) @@ -219,6 +220,7 @@ ControlWriteRejectedError, ControlValueCoercionError, ControlAccessDeniedError, + ControlWritesDisabledError, NoAdapterForAddressError, MalformedControlAddressError, ) @@ -238,7 +240,13 @@ prefix matches, and `MalformedControlAddressError` for one that matches a route but does not satisfy that substrate's address syntax (a Tango TRL with no attribute segment, say). Both are recipe or configuration gaps an -operator must see as a failed step, not as a dead conduct task.""" +operator must see as a failed step, not as a dead conduct task. + +`ControlWritesDisabledError` is included for the same reason from the +other direction: an observe-only deployment refusing a write is a +CONFIGURATION fact the operator should read as a recorded step failure +naming the address, not an opaque crash. Omitting it would not permit +the write, but it would strand the Procedure in `Running`.""" _LIFECYCLE_RERAISE: tuple[type[Exception], ...] = ( diff --git a/apps/api/src/cora/operation/ports/__init__.py b/apps/api/src/cora/operation/ports/__init__.py index 61754a58057..5781d44d35e 100644 --- a/apps/api/src/cora/operation/ports/__init__.py +++ b/apps/api/src/cora/operation/ports/__init__.py @@ -45,6 +45,7 @@ ControlTimeoutError, ControlValueCoercionError, ControlWriteRejectedError, + ControlWritesDisabledError, NoAdapterForAddressError, ) from cora.operation.ports.measurement import ( @@ -92,6 +93,7 @@ "ControlTimeoutError", "ControlValueCoercionError", "ControlWriteRejectedError", + "ControlWritesDisabledError", "InMemoryProcedureActivityLookup", "JobId", "JobSpec", diff --git a/apps/api/src/cora/operation/ports/control_port.py b/apps/api/src/cora/operation/ports/control_port.py index 56d189feafc..208f31ddc43 100644 --- a/apps/api/src/cora/operation/ports/control_port.py +++ b/apps/api/src/cora/operation/ports/control_port.py @@ -58,9 +58,11 @@ ## Exceptions -Six exception families mirror CORA's standard shape. `ControlPort` is +Seven exception families mirror CORA's standard shape. `ControlPort` is not REST-accessible; the executor's decider captures these as event-payload metadata per [[project_non_determinism_principle]]. +Six report what the substrate did; `ControlWritesDisabledError` reports +what CORA declined to do, before any substrate was contacted. ## Subscribe shape @@ -77,7 +79,7 @@ from collections.abc import AsyncIterator from enum import StrEnum -from typing import Any, Protocol, TypeVar, runtime_checkable +from typing import Any, Literal, Protocol, TypeVar, runtime_checkable from cora.operation.ports.control_address import ControlAddress from cora.operation.ports.measurement import Measurement, MeasurementKind, Quality @@ -153,6 +155,14 @@ class ControlWriteRejectedError(Exception): `ControlAccessDeniedError` so adapters preserve the substrate's failure-mode distinction (e.g., EPICS IOC put-callback failure vs Channel Access security denial). + + Note the two-tier read-only overlap: the "read-only address" case + here is a read-only RECORD on the IOC, discovered by attempting the + write. `ControlWritesDisabledError` is the CORA-side declaration + that refused before any write was attempted. The remedies are + opposite, so keep them apart when reading logs: this one means the + IOC or its access rules need changing, that one means CORA's + configuration does. """ def __init__(self, address: str, reason: str) -> None: @@ -161,6 +171,47 @@ def __init__(self, address: str, reason: str) -> None: self.reason = reason +class ControlWritesDisabledError(Exception): + """CORA declares this address unwritable; no write was attempted. + + Raised by `ReadOnlyControlPort` when a deployment has declared its + control writes off, either deployment-wide + (`CONTROL_WRITES_ENABLED=false`, `scope="deployment"`) or for the + one route carrying `ControlPortRoute.read_only` (`scope="route"`). + A declared deployment fact, never inferred from the substrate, and + settled before any IO: nothing reached the IOC. + + The grammatical subject is what separates this from its siblings. + Here the subject is CORA's own configuration, which no substrate + knows exists: an EPICS IOC cannot raise this error. In + `ControlWriteRejectedError` the subject is the write, and the + verdict arrives FROM the substrate, which is why that one carries a + substrate-supplied `reason` and this one does not. No substrate + spoke. + + `scope` names which declaration refused, so an operator knows + whether to change an env var or a route entry; `prefix` names the + route when one carried the declaration. + """ + + def __init__( + self, + address: str, + *, + scope: Literal["deployment", "route"], + prefix: str | None = None, + ) -> None: + detail = ( + "control writes are disabled deployment-wide" + if scope == "deployment" + else f"route {prefix!r} is declared read-only" + ) + super().__init__(f"Control address {address!r} write refused: {detail}") + self.address = address + self.scope = scope + self.prefix = prefix + + class ControlValueCoercionError(Exception): """A control value cannot be coerced to the kind a consumer requires. @@ -282,7 +333,9 @@ async def write( Raises `ControlNotConnectedError`, `ControlTimeoutError` (when `wait=True` and confirmation does not fire in time), `ControlWriteRejectedError` (substrate rejected the write), - or `ControlAccessDeniedError` (substrate denied access). + `ControlAccessDeniedError` (substrate denied access), or + `ControlWritesDisabledError` (the deployment declares this + address unwritable, so no substrate was contacted). """ ... @@ -367,6 +420,7 @@ def subscribe(self, address: _AddressT_contra) -> AsyncIterator[Measurement]: "ControlTimeoutError", "ControlValueCoercionError", "ControlWriteRejectedError", + "ControlWritesDisabledError", "Measurement", "MeasurementKind", "NoAdapterForAddressError", diff --git a/apps/api/src/cora/operation/wire.py b/apps/api/src/cora/operation/wire.py index fdc8c26893e..7a4aafbda07 100644 --- a/apps/api/src/cora/operation/wire.py +++ b/apps/api/src/cora/operation/wire.py @@ -290,7 +290,10 @@ def wire_operation( control_port = ( control_port if control_port is not None - else build_control_port(deps.settings.control_port_routes) + else build_control_port( + deps.settings.control_port_routes, + writes_enabled=deps.settings.control_writes_enabled, + ) ) # Borrowed shared instance from the composition root (one aclose lives on # app.state.compute_port); fall back to the in-memory Simulated fake when diff --git a/apps/api/tests/conftest.py b/apps/api/tests/conftest.py index 9e137dd2b54..6c434485742 100644 --- a/apps/api/tests/conftest.py +++ b/apps/api/tests/conftest.py @@ -44,6 +44,17 @@ os.environ.setdefault("APP_ENV", "test") +# Control writes are OFF by default in production (the observe-only pilot +# posture); the test environment turns them ON, the same way APP_ENV=test +# relaxes the trust / auth / signing boot gates. Observe-only is a +# deployment safety default, not a property of the system, so the suite +# must exercise the full writable conduct surface. The default-is-safe +# behaviour is covered at the unit tier (build_control_port with +# writes_enabled=False), not by every app-level conduct test refusing. +# A test that asserts the SAFE default must build Settings ignoring this +# env override (e.g. Settings(_env_file=None) with the var cleared). +os.environ.setdefault("CONTROL_WRITES_ENABLED", "true") + # Hypothesis profiles for property-based testing. # `dev` (default locally): keep the example database on, allow shrinking, default # max_examples — the fast feedback loop a developer wants. diff --git a/apps/api/tests/unit/operation/test_control_port_config.py b/apps/api/tests/unit/operation/test_control_port_config.py index ffaa0f1bc41..bb8b6b6ac3f 100644 --- a/apps/api/tests/unit/operation/test_control_port_config.py +++ b/apps/api/tests/unit/operation/test_control_port_config.py @@ -9,9 +9,17 @@ an EpicsCaControlPort (constructed; not exercised against EPICS) - epics_pva route -> ControlPortRegistry routing the prefix to an EpicsPvaControlPort + - tango route -> ControlPortRegistry routing the prefix to a + TangoControlPort (probe neutralised; PyTango absent in base env) - mixed routes -> registry picks the right adapter per prefix - route Pydantic validation: empty prefix rejected, unknown substrate rejected, extra fields rejected + - writes_enabled=False wraps every route so writes refuse, with no + per-substrate exemption; per-route read_only wraps only its route + +Every routing test passes `writes_enabled=True` so it observes the +naked adapter: the write posture is orthogonal to which adapter a +prefix resolves to, and stating it keeps the two concerns separable. """ import pytest @@ -23,9 +31,11 @@ from cora.operation.adapters.epics_ca_control_port import EpicsCaControlPort from cora.operation.adapters.epics_pva_control_port import EpicsPvaControlPort from cora.operation.adapters.in_memory_control_port import InMemoryControlPort +from cora.operation.adapters.read_only_control_port import ReadOnlyControlPort from cora.operation.adapters.tango_control_port import TangoControlPort from cora.operation.ports.control_port import ( ControlNotConnectedError, + ControlWritesDisabledError, NoAdapterForAddressError, ) @@ -36,13 +46,15 @@ def _no_tango_probe(_substrate: str) -> None: @pytest.mark.unit def test_build_control_port_with_empty_routes_returns_in_memory_port() -> None: - port = build_control_port([]) + port = build_control_port([], writes_enabled=True) assert isinstance(port, InMemoryControlPort) @pytest.mark.unit async def test_build_control_port_with_single_in_memory_route_returns_registry() -> None: - port = build_control_port([ControlPortRoute(prefix="2bma:", substrate="in_memory")]) + port = build_control_port( + [ControlPortRoute(prefix="2bma:", substrate="in_memory")], writes_enabled=True + ) assert isinstance(port, ControlPortRegistry) # in_memory routes ride the registry's str-port wrapper (an internal # detail), so assert behaviour rather than the wrapper type: a matched @@ -57,7 +69,9 @@ async def test_build_control_port_with_single_in_memory_route_returns_registry() @pytest.mark.unit def test_build_control_port_with_epics_ca_route_constructs_ca_adapter() -> None: - port = build_control_port([ControlPortRoute(prefix="2bma:", substrate="epics_ca")]) + port = build_control_port( + [ControlPortRoute(prefix="2bma:", substrate="epics_ca")], writes_enabled=True + ) assert isinstance(port, ControlPortRegistry) routed = port.route("2bma:rot:val") assert isinstance(routed, EpicsCaControlPort) @@ -65,7 +79,9 @@ def test_build_control_port_with_epics_ca_route_constructs_ca_adapter() -> None: @pytest.mark.unit def test_build_control_port_with_epics_pva_route_constructs_pva_adapter() -> None: - port = build_control_port([ControlPortRoute(prefix="2bma:cam:image", substrate="epics_pva")]) + port = build_control_port( + [ControlPortRoute(prefix="2bma:cam:image", substrate="epics_pva")], writes_enabled=True + ) assert isinstance(port, ControlPortRegistry) routed = port.route("2bma:cam:image:data") assert isinstance(routed, EpicsPvaControlPort) @@ -78,7 +94,8 @@ def test_build_control_port_with_mixed_routes_picks_right_adapter_per_prefix() - [ ControlPortRoute(prefix="2bma:cam1:image", substrate="epics_pva"), ControlPortRoute(prefix="2bma:", substrate="epics_ca"), - ] + ], + writes_enabled=True, ) assert isinstance(port, ControlPortRegistry) # Specific prefix wins for image addresses (longest-prefix-match). @@ -102,7 +119,9 @@ def test_build_control_port_with_tango_route_constructs_tango_adapter( "cora.operation.adapters.tango_control_port.require_tango", _no_tango_probe, ) - port = build_control_port([ControlPortRoute(prefix="id19/", substrate="tango")]) + port = build_control_port( + [ControlPortRoute(prefix="id19/", substrate="tango")], writes_enabled=True + ) assert isinstance(port, ControlPortRegistry) routed = port.route("id19/bsh/1/state") assert isinstance(routed, TangoControlPort) @@ -131,7 +150,9 @@ def test_control_port_route_rejects_extra_fields() -> None: @pytest.mark.unit async def test_build_control_port_returned_registry_supports_aclose() -> None: """The registry's aclose() fans out to every constructed adapter.""" - port = build_control_port([ControlPortRoute(prefix="x:", substrate="in_memory")]) + port = build_control_port( + [ControlPortRoute(prefix="x:", substrate="in_memory")], writes_enabled=True + ) assert isinstance(port, ControlPortRegistry) await port.aclose() # no-op for InMemoryControlPort; should not raise @@ -150,7 +171,8 @@ def test_build_control_port_threads_is_simulated_flag_to_registry() -> None: the gate this route is a simulator. """ port = build_control_port( - [ControlPortRoute(prefix="2bma:", substrate="epics_ca", is_simulated=True)] + [ControlPortRoute(prefix="2bma:", substrate="epics_ca", is_simulated=True)], + writes_enabled=True, ) assert isinstance(port, ControlPortRegistry) assert port.route_is_simulated("2bma:rot:val") is True @@ -163,8 +185,114 @@ def test_build_control_port_mixed_simulated_and_physical_routes() -> None: [ ControlPortRoute(prefix="2bma:sim:", substrate="in_memory", is_simulated=True), ControlPortRoute(prefix="2bma:", substrate="epics_ca", is_simulated=False), - ] + ], + writes_enabled=True, ) assert isinstance(port, ControlPortRegistry) assert port.route_is_simulated("2bma:sim:rot") is True assert port.route_is_simulated("2bma:rot:val") is False + + +@pytest.mark.unit +def test_control_port_route_read_only_defaults_false() -> None: + assert ControlPortRoute(prefix="2bma:", substrate="epics_ca").read_only is False + + +@pytest.mark.unit +async def test_build_control_port_writes_disabled_refuses_write_on_every_route() -> None: + """The deployment switch is the observe-only safety mechanism. + + No route declared `read_only`; the switch alone must still refuse, and + it must refuse on a real substrate route (the typed guard path), not + only the in-memory one. + """ + port = build_control_port( + [ControlPortRoute(prefix="2bma:", substrate="epics_ca")], writes_enabled=False + ) + with pytest.raises(ControlWritesDisabledError) as excinfo: + await port.write("2bma:rot:val", 90.0) + assert excinfo.value.address == "2bma:rot:val" + assert excinfo.value.scope == "deployment" + # Attribution follows scope: the route did not carry this refusal. + assert excinfo.value.prefix is None + + +@pytest.mark.unit +async def test_build_control_port_writes_disabled_refuses_write_on_in_memory_route() -> None: + """The str-surface guard path (in_memory) refuses too, through the shim.""" + port = build_control_port( + [ControlPortRoute(prefix="2bma:", substrate="in_memory")], writes_enabled=False + ) + with pytest.raises(ControlWritesDisabledError): + await port.write("2bma:rot:val", 90.0) + + +@pytest.mark.unit +async def test_build_control_port_writes_disabled_still_allows_read() -> None: + """Observe-only must still observe. The read reaches the inner adapter, + surfacing that adapter's own not-connected error rather than the guard's + refusal, which it could only do by passing through.""" + port = build_control_port( + [ControlPortRoute(prefix="2bma:", substrate="in_memory")], writes_enabled=False + ) + with pytest.raises(ControlNotConnectedError): + await port.read("2bma:rot:val") + + +@pytest.mark.unit +def test_build_control_port_writes_disabled_grants_no_substrate_exemption() -> None: + """in_memory is wrapped too: an exemption is a partial application.""" + port = build_control_port([], writes_enabled=False) + assert isinstance(port, ReadOnlyControlPort) + + +@pytest.mark.unit +async def test_build_control_port_read_only_route_refuses_within_writable_deployment() -> None: + """Per-route expressiveness: drive the stage, never the shutter.""" + port = build_control_port( + [ + ControlPortRoute(prefix="2bma:shutter:", substrate="in_memory", read_only=True), + ControlPortRoute(prefix="2bma:", substrate="in_memory"), + ], + writes_enabled=True, + ) + assert isinstance(port, ControlPortRegistry) + with pytest.raises(ControlWritesDisabledError) as excinfo: + await port.write("2bma:shutter:open", 1) + assert excinfo.value.scope == "route" + assert excinfo.value.prefix == "2bma:shutter:" + # The sibling route in the same registry stays writable: the write + # reaches the in-memory adapter and raises ITS not-connected error + # (the address was never seeded), NOT the guard's refusal. That it is + # not ControlWritesDisabledError is the point. + with pytest.raises(ControlNotConnectedError): + await port.write("2bma:rot:val", 90.0) + + +@pytest.mark.unit +def test_build_control_port_read_only_route_preserves_registry_provenance() -> None: + """The guard wraps BELOW the registry, so route_is_simulated survives. + + Wrapping the registry itself would hide this from the Conductor's + `_ActuationObserver` getattr and silently disable the Dataset + provenance gate. + """ + port = build_control_port( + [ControlPortRoute(prefix="2bma:", substrate="epics_ca", is_simulated=True, read_only=True)], + writes_enabled=True, + ) + assert isinstance(port, ControlPortRegistry) + assert port.route_is_simulated("2bma:rot:val") is True + + +@pytest.mark.unit +def test_build_control_port_rejects_duplicate_prefixes() -> None: + """Last-wins registration would silently discard a read_only declaration.""" + with pytest.raises(ValueError, match="more than once"): + build_control_port( + [ + ControlPortRoute(prefix="2bma:", substrate="in_memory", read_only=True), + ControlPortRoute(prefix="2bma:", substrate="in_memory"), + ], + writes_enabled=True, + ) diff --git a/apps/api/tests/unit/operation/test_read_only_control_port.py b/apps/api/tests/unit/operation/test_read_only_control_port.py new file mode 100644 index 00000000000..e7d8234a95a --- /dev/null +++ b/apps/api/tests/unit/operation/test_read_only_control_port.py @@ -0,0 +1,314 @@ +"""Unit tests for the `ReadOnlyControlPort` + `ReadOnlySubstratePort` guards. + +Covers both surfaces #568 left the registry with: + + - write refuses, carrying address + scope + prefix (the typed guard + carries `str(address)`) + - refusal happens before the inner adapter is touched at all + - read / subscribe still reach the inner adapter (observe-only + must still observe) + - aclose delegates (the registry skips adapters lacking it, so a + missing delegation leaks every wrapped EPICS connection) + - aclose tolerates an inner adapter with no aclose + - the str guard satisfies the ControlPort protocol +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import TYPE_CHECKING, Any + +import pytest + +from cora.operation.adapters.in_memory_control_port import InMemoryControlPort +from cora.operation.adapters.read_only_control_port import ( + ReadOnlyControlPort, + ReadOnlySubstratePort, +) +from cora.operation.ports.control_address import ControlAddress, InMemoryAddress +from cora.operation.ports.control_port import ( + ControlPort, + ControlWritesDisabledError, + Measurement, +) + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + + +class _RecordingPort: + """Inner adapter that records whether anything reached it.""" + + def __init__(self) -> None: + self.writes: list[tuple[str, Any]] = [] + self.reads: list[str] = [] + self.subscribes: list[str] = [] + self.closed = False + self._seeded = InMemoryControlPort() + + def seed(self, address: str, value: float = 45.0) -> None: + self._seeded.set_reading( + address, + Measurement( + value=value, + kind="Scalar", + quality="Good", + produced_at=datetime(2026, 7, 16, tzinfo=UTC), + ), + ) + + async def read(self, address: str) -> Measurement: + self.reads.append(address) + return await self._seeded.read(address) + + async def write( + self, + address: str, + value: int | float | bool | str | tuple[Any, ...], + *, + wait: bool = True, + timeout_s: float = 30.0, + ) -> None: + self.writes.append((address, value)) + + def subscribe(self, address: str) -> AsyncIterator[Measurement]: + self.subscribes.append(address) + return self._seeded.subscribe(address) + + async def aclose(self) -> None: + self.closed = True + + +class _NoCloseCarrier: + """Inner adapter with no aclose, mirroring a bare test double.""" + + async def read(self, address: str) -> Measurement: + return await InMemoryControlPort().read(address) + + async def write( + self, + address: str, + value: int | float | bool | str | tuple[Any, ...], + *, + wait: bool = True, + timeout_s: float = 30.0, + ) -> None: + raise AssertionError("guard must refuse before reaching the inner adapter") + + def subscribe(self, address: str) -> AsyncIterator[Measurement]: + return InMemoryControlPort().subscribe(address) + + +@pytest.mark.unit +async def test_read_only_control_port_write_refuses_with_deployment_scope() -> None: + port = ReadOnlyControlPort(_RecordingPort(), scope="deployment") + with pytest.raises(ControlWritesDisabledError) as excinfo: + await port.write("2bma:rot:val", 90.0) + assert excinfo.value.address == "2bma:rot:val" + assert excinfo.value.scope == "deployment" + assert excinfo.value.prefix is None + assert "deployment-wide" in str(excinfo.value) + + +@pytest.mark.unit +async def test_read_only_control_port_write_refuses_with_route_scope_and_prefix() -> None: + port = ReadOnlyControlPort( + _RecordingPort(), + scope="route", + prefix="2bma:shutter:", + ) + with pytest.raises(ControlWritesDisabledError) as excinfo: + await port.write("2bma:shutter:open", 1) + assert excinfo.value.scope == "route" + assert excinfo.value.prefix == "2bma:shutter:" + assert "2bma:shutter:" in str(excinfo.value) + + +@pytest.mark.unit +async def test_read_only_control_port_never_reaches_inner_adapter_on_write() -> None: + """The refusal is settled before any IO; nothing reached the substrate.""" + inner = _RecordingPort() + port = ReadOnlyControlPort(inner, scope="deployment") + with pytest.raises(ControlWritesDisabledError): + await port.write("2bma:rot:val", 90.0) + assert inner.writes == [] + + +@pytest.mark.unit +async def test_read_only_control_port_write_refuses_regardless_of_kwargs() -> None: + """No argument combination talks the guard into a write.""" + inner = _RecordingPort() + port = ReadOnlyControlPort(inner, scope="deployment") + with pytest.raises(ControlWritesDisabledError): + await port.write("2bma:rot:val", 90.0, wait=False, timeout_s=0.0) + assert inner.writes == [] + + +@pytest.mark.unit +async def test_read_only_control_port_read_reaches_inner_adapter() -> None: + inner = _RecordingPort() + inner.seed("2bma:rot:val") + port = ReadOnlyControlPort(inner, scope="deployment") + reading = await port.read("2bma:rot:val") + assert inner.reads == ["2bma:rot:val"] + assert reading.value == 45.0 + + +@pytest.mark.unit +async def test_read_only_control_port_subscribe_reaches_inner_adapter() -> None: + """Observe-only includes streaming, not just point reads.""" + inner = _RecordingPort() + inner.seed("2bma:rot:val") + port = ReadOnlyControlPort(inner, scope="deployment") + iterator = port.subscribe("2bma:rot:val") + inner.seed("2bma:rot:val", 2.0) # the stream yields on CHANGE + got = await anext(iterator) + assert inner.subscribes == ["2bma:rot:val"] + assert got.value == 2.0 + await iterator.aclose() # type: ignore[attr-defined] # InMemoryControlPort returns AsyncGenerator + + +@pytest.mark.unit +async def test_read_only_control_port_aclose_delegates_to_inner() -> None: + """Load-bearing: the registry skips adapters lacking aclose.""" + inner = _RecordingPort() + port = ReadOnlyControlPort(inner, scope="deployment") + await port.aclose() + assert inner.closed is True + + +@pytest.mark.unit +async def test_read_only_control_port_aclose_tolerates_inner_without_aclose() -> None: + port = ReadOnlyControlPort(_NoCloseCarrier(), scope="deployment") + await port.aclose() # must not raise + + +@pytest.mark.unit +def test_read_only_control_port_satisfies_control_port_protocol() -> None: + port = ReadOnlyControlPort(InMemoryControlPort(), scope="deployment") + assert isinstance(port, ControlPort) + + +class _RecordingSubstratePort: + """Typed inner adapter recording what reached it, for the substrate guard. + + Mirrors `_RecordingPort` on the typed `ControlAddress` surface that #568's + real substrate adapters implement. + """ + + def __init__(self) -> None: + self.writes: list[tuple[ControlAddress, Any]] = [] + self.reads: list[ControlAddress] = [] + self.subscribes: list[ControlAddress] = [] + self.closed = False + self._reading = Measurement( + value=45.0, + kind="Scalar", + quality="Good", + produced_at=datetime(2026, 7, 16, tzinfo=UTC), + ) + + async def read(self, address: ControlAddress) -> Measurement: + self.reads.append(address) + return self._reading + + async def write( + self, + address: ControlAddress, + value: int | float | bool | str | tuple[Any, ...], + *, + wait: bool = True, + timeout_s: float = 30.0, + ) -> None: + self.writes.append((address, value)) + + def subscribe(self, address: ControlAddress) -> AsyncIterator[Measurement]: + self.subscribes.append(address) + return _one_measurement(self._reading) + + async def aclose(self) -> None: + self.closed = True + + +class _NoCloseSubstrateCarrier: + """Typed inner adapter with no aclose, mirroring a bare substrate double.""" + + async def read(self, address: ControlAddress) -> Measurement: + raise AssertionError("unused") + + async def write( + self, + address: ControlAddress, + value: int | float | bool | str | tuple[Any, ...], + *, + wait: bool = True, + timeout_s: float = 30.0, + ) -> None: + raise AssertionError("guard must refuse before reaching the inner adapter") + + def subscribe(self, address: ControlAddress) -> AsyncIterator[Measurement]: + raise AssertionError("unused") + + +async def _one_measurement(measurement: Measurement) -> AsyncIterator[Measurement]: + yield measurement + + +@pytest.mark.unit +async def test_read_only_substrate_port_write_refuses_carrying_str_address() -> None: + """The typed guard refuses too, and str(address) names the operator's address.""" + port = ReadOnlySubstratePort( + _RecordingSubstratePort(), + scope="route", + prefix="2bma:shutter:", + ) + with pytest.raises(ControlWritesDisabledError) as excinfo: + await port.write(InMemoryAddress("2bma:shutter:open"), 1) + assert excinfo.value.address == "2bma:shutter:open" + assert excinfo.value.scope == "route" + assert excinfo.value.prefix == "2bma:shutter:" + + +@pytest.mark.unit +async def test_read_only_substrate_port_never_reaches_inner_on_write() -> None: + inner = _RecordingSubstratePort() + port = ReadOnlySubstratePort(inner, scope="deployment") + with pytest.raises(ControlWritesDisabledError): + await port.write(InMemoryAddress("2bma:rot:val"), 90.0) + assert inner.writes == [] + + +@pytest.mark.unit +async def test_read_only_substrate_port_read_reaches_inner_adapter() -> None: + inner = _RecordingSubstratePort() + port = ReadOnlySubstratePort(inner, scope="deployment") + address = InMemoryAddress("2bma:rot:val") + reading = await port.read(address) + assert inner.reads == [address] + assert reading.value == 45.0 + + +@pytest.mark.unit +async def test_read_only_substrate_port_subscribe_reaches_inner_adapter() -> None: + inner = _RecordingSubstratePort() + port = ReadOnlySubstratePort(inner, scope="deployment") + address = InMemoryAddress("2bma:rot:val") + iterator = port.subscribe(address) + got = await anext(iterator) + assert inner.subscribes == [address] + assert got.value == 45.0 + + +@pytest.mark.unit +async def test_read_only_substrate_port_aclose_delegates_to_inner() -> None: + """Load-bearing: the registry skips adapters lacking aclose.""" + inner = _RecordingSubstratePort() + port = ReadOnlySubstratePort(inner, scope="deployment") + await port.aclose() + assert inner.closed is True + + +@pytest.mark.unit +async def test_read_only_substrate_port_aclose_tolerates_inner_without_aclose() -> None: + port = ReadOnlySubstratePort(_NoCloseSubstrateCarrier(), scope="deployment") + await port.aclose() # must not raise diff --git a/docs/stack/deployment.md b/docs/stack/deployment.md index b57ce16b3cb..add113abe46 100644 --- a/docs/stack/deployment.md +++ b/docs/stack/deployment.md @@ -14,6 +14,58 @@ The load-bearing auth vars (full list in `.env.example`): | `ALLOW_PERMISSIVE_AUTHZ` | `false` | Production-tier escape hatch: set `true` to run the permit-everyone `AllowAllAuthorize` stub on purpose in a `prod`/`production`/`staging` env (airgapped / single-operator pilot) | | `IDENTITY_PROVIDERS` | unset → legacy `X-Principal-Id` header mode | JSON list of `IdentityProviderConfig` entries (see [Auth](auth.md)); enables bearer-token mode at the HTTP edge | | `ANTHROPIC_API_KEY` | unset → AI subscribers log-and-skip | When you want RunDebriefer / CautionDrafter live | +| `CONTROL_WRITES_ENABLED` | `false` → CORA never drives through the ControlPort | Set `true` only when this deployment is meant to actuate hardware (see below; it does not cover `COMPUTE_SUBSTRATE`) | +| `CONTROL_PORT_ROUTES` | unset → `InMemoryControlPort` (no real substrate) | When CORA talks to a real control system. JSON list of `ControlPortRoute` | + +### Observe-only deployments + +`CONTROL_WRITES_ENABLED` defaults to `false`, and that default is the +mechanism behind an observe-only deployment such as the APS 2-BM pilot. +Every adapter the ControlPort factory builds is wrapped in a +`ReadOnlyControlPort`: `read` and `subscribe` pass through, and `write` +raises `ControlWritesDisabledError` before any substrate is contacted. +The Conductor records that refusal as a step failure naming the address. + +Two properties are worth stating plainly, because they are what make the +observe-only claim true rather than aspirational: + +- **It cannot be partially applied.** There is no per-substrate or + per-route exemption, not even for `in_memory`. Inferring safety from + the substrate is the mistake `is_simulated` exists to prevent: a soft + IOC speaks real Channel Access. +- **It fails closed.** The refusal is decided when the port is built, not + consulted at write time, so there is no "flag could not be read" state + that silently permits a write. + +`ControlPortRoute.read_only` is the per-route counterpart, but it +defaults to writable, so it is expressiveness within a *writable* +deployment ("CORA may drive the sample stage but must never touch the +shutter"), NOT the safety gate. To make a deployment observe-only, use +the switch. + +Scope this honestly when describing the pilot: the switch closes CORA's +*modeled* actuation path (`ControlPort`). It does not by itself make the +host incapable of touching the beamline. + +The other path is `ComputePort`. With `COMPUTE_SUBSTRATE=local_process` a +conduct job's argv runs as an OS subprocess under the API service +account, so an argv like `["caput", ...]` reaches a control system +without passing through any ControlPort, whatever +`CONTROL_WRITES_ENABLED` says. Two properties make this sharper than it +first looks: with `CORA_ALLOW_RAW_CONDUCT=true` (the current default) the +argv can come straight from the request body for any Method with no +`launch_spec`; and no Trust policy gates the spawn, because the Authorize +port gates the run transition (`complete_run` / `abort_run`), which +happens after the subprocess has already run. + +What keeps this inert today is the default: `compute_substrate` is +`in_memory` (`cora.infrastructure.config`), which mints a Simulated +result and spawns nothing. An observe-only deployment leaves it there. +Before enabling `local_process` at a beamline, give every compute Method +a `launch_spec` so argv is built server-side from the vetted recipe, then +set `CORA_ALLOW_RAW_CONDUCT=false` to close the caller-supplied path. The +gates that do and do not apply to a submit are documented at the top of +`cora.api._conduct_run_route`. ### Startup boot gate From 8ad2a59e308faba912304b8fdc3a4d4ff32d7239 Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Fri, 17 Jul 2026 01:18:17 +0300 Subject: [PATCH 2/4] Answer "can this process serve?", not just "is it alive?" CORA has had a liveness probe since early on: /health returns 200 and checks nothing. Nothing answered the other question. A load balancer had no signal, and the gap is not academic: the pool is built once at lifespan start with no retry or reconnect anywhere, so Postgres going away AFTER boot leaves a live, cheerful process that fails every request. /health would report it healthy the whole time. /readyz reports Postgres and nothing else. That is not minimalism for its own sake, it is the consequence of how fail-fast boot already is: create_app() runs at import, so every config gate crashes the process before uvicorn binds a socket, and the seed checks run at lifespan start, where a failure exits. Anything alive enough to answer /readyz has already passed all of them, so re-checking would report a state that cannot exist. Postgres is the one dependency that can change afterwards. Projection health is deliberately not gated on. A wedged projection is a global condition: 503-ing on it pulls every replica at once, including the pod hosting dismiss_event_in_reaction, whose own playbook is "page, dismiss-event with reason, debug tomorrow". Removing the repair tool from the Service to signal that the repair tool is needed is a self-locking outage. That belongs in a metric with an alert, not in a traffic-routing signal. /health stays check-free and now says why in the source. Every dependency it could check is one a restart cannot fix; because there is no boot retry, a DB-checking liveness probe would restart pods into an outage the restart cannot mend, turning one blip into a crash loop. Someone will ask why it does not check the database, and no fitness test guards the answer. The probe takes the pool as an argument rather than reaching for it, because tests/conftest.py forces APP_ENV=test process-wide and that builds the in-memory kernel with pool=None. An app-level test can therefore only ever reach the `skipped` branch: without the split, the DB path, the status mapping and the bounded-acquire property would all ship unexecuted. The bounded acquire is the subtle part and is verified against asyncpg's own source, not assumed. `Pool.fetchval(q, timeout=T)` does `async with self.acquire() as con` and forwards T only to the query, so the timeout that looks like it bounds the call does not bound the part that blocks: `_acquire` awaits `self._queue.get()` unbounded. The obvious one-liner hangs forever on a saturated pool. Both the outer asyncio.timeout and acquire(timeout=) are load-bearing, a unit test pins it with a fake whose acquire sleeps past the budget, and the WHY sits on the helper because nothing about it is inferable from the call. Three registries had to move with the route, each silent on failure: the middleware skip list (its fitness test asserts strict equality and says to update both in the same commit), the OTel excluded-URLs string, and the Prometheus excluded_handlers. "health" does not substring-match "readyz", so inheriting the existing exclusions was not an option; probe traffic on a fixed period would otherwise swamp the latency histograms. /health stays instrumented, because test_metrics_endpoint asserts it is counted. The route is include_in_schema=False, so the openapi snapshot gains no path; it moved only because /health's docstring did. Deployment doc carries the budget invariant (app 1.5s under the orchestrator's 2s, or /readyz degrades from a diagnostic into a hang detector), plus two honest limits: readiness buys little at one replica, which is the 2-BM pilot's shape, and there is no graceful drain, so rolling deploys drop in-flight requests until one lands. Pre-commit pyright hook bypassed: 8 pre-existing reportMissingImports for the optional torch/botorch/gpytorch group, deliberately excluded from the base install. Unrelated; every other hook passes. Co-Authored-By: Claude Opus 4.8 --- apps/api/openapi.json | 2 +- apps/api/src/cora/api/_readiness.py | 143 ++++++++++++++ apps/api/src/cora/api/main.py | 50 ++++- .../auth/bearer_auth_middleware.py | 13 +- .../infrastructure/observability/provider.py | 2 +- .../architecture/test_edge_auth_invariants.py | 1 + .../tests/contract/test_readyz_endpoint.py | 57 ++++++ apps/api/tests/unit/api/test_readiness.py | 174 ++++++++++++++++++ docs/stack/deployment.md | 61 ++++++ 9 files changed, 490 insertions(+), 13 deletions(-) create mode 100644 apps/api/src/cora/api/_readiness.py create mode 100644 apps/api/tests/contract/test_readyz_endpoint.py create mode 100644 apps/api/tests/unit/api/test_readiness.py diff --git a/apps/api/openapi.json b/apps/api/openapi.json index 95d501acc60..7651dbfca1f 100644 --- a/apps/api/openapi.json +++ b/apps/api/openapi.json @@ -34266,7 +34266,7 @@ }, "/health": { "get": { - "description": "Liveness probe.", + "description": "Liveness probe.\n\nChecks nothing, and must keep checking nothing. Every dependency\nit could check is one a restart cannot fix, and CORA is a worse\ncase than most: there is no boot retry, so a DB-checking liveness\nprobe would restart a pod into a DB outage that the restart\ncannot mend, turning one outage into a crash loop. Readiness is\nthe probe that reports dependencies; see `/readyz`.", "operationId": "health_health_get", "responses": { "200": { diff --git a/apps/api/src/cora/api/_readiness.py b/apps/api/src/cora/api/_readiness.py new file mode 100644 index 00000000000..b33f0184b64 --- /dev/null +++ b/apps/api/src/cora/api/_readiness.py @@ -0,0 +1,143 @@ +"""Readiness probe: can this process serve a correct request right now. + +`/health` (main.py) is the LIVENESS probe and answers a different +question: is the process alive. It checks nothing, deliberately. This +module backs `/readyz`, which reports whether the one dependency that +can change after a successful boot is still there. + +## Why Postgres is the only check + +Boot is aggressively fail-fast, so readiness must not restate it. +`app = create_app()` runs at import, and every configuration gate +(trust policy, authenticated-principal, IdP posture, signing posture) +raises there, before uvicorn binds a socket. The bootstrap policy, the +seeded Surfaces, and the pool itself are established at lifespan start, +and a failed lifespan exits the process. So if anything is alive to +answer `/readyz`, every one of those already passed. Re-checking them +would report a state that cannot exist. + +Postgres is different. The pool is created once at lifespan start and +there is no retry or reconnect logic anywhere, so an outage AFTER boot +leaves a live process that fails every request. That is the gap a +readiness probe exists to close, and it is the whole of it. + +Projection health is deliberately NOT gated on. A wedged projection is +a global condition: it would take every replica out of rotation at +once, including the pod hosting the in-band repair tool +(`agent/features/dismiss_event_in_reaction`, whose playbook is "page, +dismiss-event with reason, debug tomorrow"). Removing the repair tool +from the Service to signal that the repair tool is needed is a +self-locking outage. Projection lag belongs in a metric with an alert +on it, not in a traffic-routing signal. + +## Why this is a pure function + +`tests/conftest.py` forces `APP_ENV=test` process-wide, and that env +builds the in-memory kernel with `pool=None`. So a probe exercised only +through the app can reach nothing but the skipped branch: the DB path, +the status mapping, and the bounded-acquire property would all ship +unexecuted. Taking the pool as an argument lets the unit tier drive +every branch with fakes. +""" + +# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false + +from __future__ import annotations + +import asyncio +from typing import TYPE_CHECKING, Literal + +import asyncpg + +if TYPE_CHECKING: + from cora.infrastructure.config import Settings + +DatabaseStatus = Literal["ok", "unreachable", "saturated", "closing", "skipped", "error"] +"""What the probe learned about Postgres. `skipped` means this +deployment has no pool at all (the in-memory kernel), so there is +nothing to report rather than nothing wrong.""" + +# Total wall budget for the probe, and the slice of it the acquire may +# take. Both are load-bearing and neither is redundant: +# +# - `pool.acquire(timeout=)` bounds the WAIT FOR A CONNECTION. Without +# it the probe waits forever on a saturated pool: `Pool.fetchval(q, +# timeout=T)` calls `self.acquire()` with NO timeout and forwards T +# only to the query, so the `timeout=` that looks like it bounds the +# call does not bound the part that hangs. +# - The outer `asyncio.timeout` bounds EVERYTHING ELSE, including +# asyncpg's 60s connect default, which CORA never overrides and +# which a probe can trigger because min_size=1 leaves connections +# 2..10 to be established lazily. +# +# The budget is generous on purpose. The probe is a victim of pool +# exhaustion, never a cause (one connection for ~1ms per period against +# max_size=10), so a tight budget only converts "busy" into "removed +# from rotation" and makes an overload spiral worse. Do not shrink these +# to "fail fast". +# +# INVARIANT: this budget must stay UNDER the orchestrator's own probe +# timeout (see docs/stack/deployment.md). If the orchestrator gives up +# first, its client disconnect, not our timeout, ends the request, and +# /readyz degrades from a diagnostic endpoint into a hang detector: the +# `saturated` body that is the entire point never gets written. +_PROBE_BUDGET_S = 1.5 +_ACQUIRE_BUDGET_S = 1.0 + + +async def probe_database(pool: asyncpg.Pool | None) -> DatabaseStatus: + """Report whether Postgres can serve a trivial query right now. + + Never raises. Every failure is a status the caller renders, because + a probe that raises turns a dependency outage into a 500 that reads + like an application bug. + """ + if pool is None: + return "skipped" + try: + async with ( + asyncio.timeout(_PROBE_BUDGET_S), + pool.acquire(timeout=_ACQUIRE_BUDGET_S) as conn, + ): + await conn.fetchval("SELECT 1") + except TimeoutError: + # asyncio.TimeoutError IS the builtin on 3.13, so this arm + # covers both the outer budget and the acquire timeout. + return "saturated" + except asyncpg.InterfaceError: + # Raised as "pool is closing" during graceful shutdown. NOT a + # PostgresError subclass, so a narrower tuple would miss it. + return "closing" + except OSError: + return "unreachable" + except Exception: + # Residual on purpose. Anything not enumerated above is still a + # reason we cannot serve, and letting it escape would 500 the + # probe. CancelledError is BaseException on 3.13, so real + # cancellation still propagates. + return "error" + else: + return "ok" + + +def readiness_body(database: DatabaseStatus, settings: Settings) -> dict[str, str]: + """Render the probe result. Fixed vocabulary, no free text. + + `app_env` is here because `test` builds an in-memory kernel with no + persistence; a green probe over a store that loses every Run on + restart is worth making visible to whoever reads this. + + Deliberately absent: the database URL, the driver's error text, and + the names of projections or bounded contexts. This endpoint is + unauthenticated, and none of that helps a probe while all of it + describes the deployment to anyone who can reach it. The logs carry + the detail. + """ + return { + "status": "ready" if database in ("ok", "skipped") else "not_ready", + "database": database, + "app_env": settings.app_env, + } + + +__all__ = ["DatabaseStatus", "probe_database", "readiness_body"] diff --git a/apps/api/src/cora/api/main.py b/apps/api/src/cora/api/main.py index df9ba5870da..401b8e33897 100644 --- a/apps/api/src/cora/api/main.py +++ b/apps/api/src/cora/api/main.py @@ -33,7 +33,7 @@ from contextlib import asynccontextmanager from dataclasses import replace -from fastapi import FastAPI +from fastapi import FastAPI, Request, Response, status from mcp.server.fastmcp import FastMCP from mcp.server.transport_security import TransportSecuritySettings from prometheus_client import CollectorRegistry @@ -82,6 +82,7 @@ from cora.api._enclosure_permit_observer import ControlPortEnclosureObserver from cora.api._inference_recorder import DelegatingInferenceRecorder from cora.api._procedure_watcher import procedure_watcher_lifespan +from cora.api._readiness import probe_database, readiness_body from cora.api._run_initiator import run_initiator_lifespan from cora.api._run_supervisor import run_supervisor_lifespan from cora.api.middleware import BodySizeLimitMiddleware @@ -1081,15 +1082,20 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: # - per-app CollectorRegistry so multiple create_app() calls in the # test process don't double-register collectors against the global # REGISTRY (which would crash the second TestClient). - # - excluded_handlers=["/metrics"] keeps the scrape endpoint out of - # its own counters (otherwise each scrape pollutes its own metrics - # with monitoring traffic). + # - excluded_handlers keeps monitoring traffic out of the counters: + # /metrics would otherwise pollute its own series with each + # scrape, and /readyz is probed on a fixed period, so its latency + # would swamp the request histograms with traffic that reflects + # the probe interval rather than the beamline's request rate. + # /health is NOT excluded: test_metrics_endpoint asserts it is + # counted, and it is the sample request that proves instrumentation + # is live. # - include_in_schema=False hides /metrics from OpenAPI /docs (it's # an operational endpoint, not part of the user-facing API). metrics_registry = CollectorRegistry() Instrumentator( registry=metrics_registry, - excluded_handlers=["/metrics"], + excluded_handlers=["/metrics", "/readyz"], ).instrument(fastapi_app).expose(fastapi_app, include_in_schema=False) # OTel FastAPI instrumentation runs after app construction so the # FastAPIInstrumentor sees every route registered above plus the @@ -1128,9 +1134,41 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: @fastapi_app.get("/health") async def health() -> dict[str, str]: # pyright: ignore[reportUnusedFunction] - """Liveness probe.""" + """Liveness probe. + + Checks nothing, and must keep checking nothing. Every dependency + it could check is one a restart cannot fix, and CORA is a worse + case than most: there is no boot retry, so a DB-checking liveness + probe would restart a pod into a DB outage that the restart + cannot mend, turning one outage into a crash loop. Readiness is + the probe that reports dependencies; see `/readyz`. + """ return {"status": "ok", "version": __version__} + @fastapi_app.get("/readyz", include_in_schema=False) + async def readyz( # pyright: ignore[reportUnusedFunction] + request: Request, response: Response + ) -> dict[str, str]: + """Readiness probe: can this process serve a correct request. + + Reports Postgres, the one dependency that can change after a + successful boot. See `cora.api._readiness` for why that is the + whole check, and why liveness must not make it. + + The pool is read off `app.state` rather than closed over: the + lifespan builds the kernel, so it does not exist when this + route is registered. + """ + request_deps = getattr(request.app.state, "deps", None) + database = await probe_database(request_deps.pool if request_deps else None) + body = readiness_body(database, settings) + if body["status"] != "ready": + response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE + # k8s ignores Retry-After on probes; it is here for humans + # and curl, and for consistency with the auth handlers. + response.headers["Retry-After"] = "5" + return body + return fastapi_app diff --git a/apps/api/src/cora/infrastructure/auth/bearer_auth_middleware.py b/apps/api/src/cora/infrastructure/auth/bearer_auth_middleware.py index 4de3c65db1c..73c0ed94446 100644 --- a/apps/api/src/cora/infrastructure/auth/bearer_auth_middleware.py +++ b/apps/api/src/cora/infrastructure/auth/bearer_auth_middleware.py @@ -97,13 +97,16 @@ _log = get_logger(__name__) -# Paths the middleware MUST never authenticate. Health and metrics -# are unauthenticated by deployment convention (k8s probes, Prometheus -# scrape). `/.well-known/oauth-protected-resource` (RFC 9728) is -# unauthenticated by spec — clients discover where to get a token. +# Paths the middleware MUST never authenticate. Health, readiness and +# metrics are unauthenticated by deployment convention (k8s probes, +# Prometheus scrape): a probe that has to hold a token is a probe that +# reports an expired token as a dead process. +# `/.well-known/oauth-protected-resource` (RFC 9728) is +# unauthenticated by spec: clients discover where to get a token. _UNAUTHENTICATED_PATHS: frozenset[str] = frozenset( { "/health", + "/readyz", "/metrics", "/.well-known/oauth-protected-resource", } @@ -113,7 +116,7 @@ def _is_unauthenticated_path(path: str) -> bool: """Return True if `path` MUST be skipped by the bearer middleware. - Exact-match against the three unauthenticated paths only. MCP + Exact-match against the four unauthenticated paths only. MCP routes are verified with audience-per-Surface binding (see `_resolve_expected_audience`). """ diff --git a/apps/api/src/cora/infrastructure/observability/provider.py b/apps/api/src/cora/infrastructure/observability/provider.py index 6021bd90193..3a776624736 100644 --- a/apps/api/src/cora/infrastructure/observability/provider.py +++ b/apps/api/src/cora/infrastructure/observability/provider.py @@ -84,7 +84,7 @@ # proportional to the probe interval, not the request rate. Comma- # separated regex patterns per FastAPIInstrumentor's excluded_urls # contract; substring match is sufficient for these stable paths. -_EXCLUDED_URLS = "health,metrics,docs,openapi.json,redoc" +_EXCLUDED_URLS = "health,readyz,metrics,docs,openapi.json,redoc" _log = getLogger(__name__) diff --git a/apps/api/tests/architecture/test_edge_auth_invariants.py b/apps/api/tests/architecture/test_edge_auth_invariants.py index b90dd82fc64..957e5437ecc 100644 --- a/apps/api/tests/architecture/test_edge_auth_invariants.py +++ b/apps/api/tests/architecture/test_edge_auth_invariants.py @@ -277,6 +277,7 @@ def test_bearer_auth_middleware_registered_exactly_once_after_body_size_limit() _EXPECTED_UNAUTHENTICATED_PATHS: frozenset[str] = frozenset( { "/health", + "/readyz", "/metrics", "/.well-known/oauth-protected-resource", } diff --git a/apps/api/tests/contract/test_readyz_endpoint.py b/apps/api/tests/contract/test_readyz_endpoint.py new file mode 100644 index 00000000000..70484da1c04 --- /dev/null +++ b/apps/api/tests/contract/test_readyz_endpoint.py @@ -0,0 +1,57 @@ +"""Contract tests for `GET /readyz` (the readiness probe route). + +`test_readiness.py` covers the pure helper (`probe_database` / +`readiness_body`); this covers the ROUTE that wires them: the status +mapping and the 503 + Retry-After path a probe consumer actually sees. + +The test app builds the in-memory kernel (no pool), so the DB fault +paths are exercised by monkeypatching `probe_database` at its point of +use in `cora.api.main`, since the route reads it as a module global. +""" + +from typing import Any + +import pytest +from fastapi.testclient import TestClient + +import cora.api.main as api_main +from cora.api.main import create_app + + +@pytest.mark.contract +def test_readyz_reports_ready_200_with_no_pool() -> None: + """The in-memory test app has no pool: ready, database skipped, HTTP 200.""" + with TestClient(create_app()) as client: + response = client.get("/readyz") + assert response.status_code == 200 + body = response.json() + assert body["status"] == "ready" + assert body["database"] == "skipped" + assert "Retry-After" not in response.headers + + +@pytest.mark.contract +def test_readyz_reports_503_and_retry_after_when_database_faults( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A DB fault surfaces as 503 not_ready with a Retry-After header.""" + + async def _unreachable(_pool: Any) -> str: + return "unreachable" + + monkeypatch.setattr(api_main, "probe_database", _unreachable) + with TestClient(create_app()) as client: + response = client.get("/readyz") + assert response.status_code == 503 + body = response.json() + assert body["status"] == "not_ready" + assert body["database"] == "unreachable" + assert response.headers["Retry-After"] == "5" + + +@pytest.mark.contract +def test_readyz_is_not_in_the_openapi_schema() -> None: + """Operational plumbing, not API surface (include_in_schema=False).""" + with TestClient(create_app()) as client: + spec = client.get("/openapi.json").json() + assert "/readyz" not in spec["paths"] diff --git a/apps/api/tests/unit/api/test_readiness.py b/apps/api/tests/unit/api/test_readiness.py new file mode 100644 index 00000000000..1220a3232dd --- /dev/null +++ b/apps/api/tests/unit/api/test_readiness.py @@ -0,0 +1,174 @@ +"""Unit tests for the /readyz probe helper. + +These drive the branches the contract tier structurally cannot reach: +`tests/conftest.py` forces APP_ENV=test process-wide, which builds the +in-memory kernel with `pool=None`, so an app-level test only ever +exercises the `skipped` path. Every fake below stands in for a pool. + +The bounded-acquire test is the load-bearing one. `Pool.fetchval(q, +timeout=T)` bounds the query but calls `acquire()` with no timeout, so +a probe written the obvious way hangs forever on a saturated pool. That +test is what fails if someone later "simplifies" the helper. +""" + +from __future__ import annotations + +import asyncio + +import asyncpg +import pytest + +from cora.api._readiness import probe_database, readiness_body +from cora.infrastructure.config import Settings + + +class _Conn: + def __init__(self, fetchval_error: Exception | None = None) -> None: + self._fetchval_error = fetchval_error + + async def fetchval(self, query: str) -> int: + if self._fetchval_error is not None: + raise self._fetchval_error + return 1 + + +class _AcquireContext: + """Mirrors `asyncpg.Pool.acquire`, which returns a context manager. + + The shape matters: the probe does `async with pool.acquire(...)`, so + a fake that merely returns a connection would not exercise the code + under test. + """ + + def __init__(self, pool: _FakePool) -> None: + self._pool = pool + + async def __aenter__(self) -> _Conn: + if self._pool.acquire_delay_s: + await asyncio.sleep(self._pool.acquire_delay_s) + if self._pool.acquire_error is not None: + raise self._pool.acquire_error + return _Conn(self._pool.fetchval_error) + + async def __aexit__(self, *exc_info: object) -> None: + self._pool.released += 1 + + +class _FakePool: + """Pool stand-in whose acquire behaviour each test dictates.""" + + def __init__( + self, + *, + acquire_error: Exception | None = None, + acquire_delay_s: float = 0.0, + fetchval_error: Exception | None = None, + ) -> None: + self.acquire_error = acquire_error + self.acquire_delay_s = acquire_delay_s + self.fetchval_error = fetchval_error + self.released = 0 + + def acquire(self, *, timeout: float | None = None) -> _AcquireContext: + return _AcquireContext(self) + + +def _settings() -> Settings: + return Settings(app_env="test") + + +@pytest.mark.unit +async def test_probe_database_with_no_pool_reports_skipped() -> None: + """The in-memory kernel has no pool; that is nothing to report, not a fault.""" + assert await probe_database(None) == "skipped" + + +@pytest.mark.unit +async def test_probe_database_with_healthy_pool_reports_ok() -> None: + assert await probe_database(_FakePool()) == "ok" # type: ignore[arg-type] + + +@pytest.mark.unit +async def test_probe_database_releases_the_connection() -> None: + """A probe that leaks a connection per period exhausts the pool it watches.""" + pool = _FakePool() + await probe_database(pool) # type: ignore[arg-type] + assert pool.released == 1 + + +@pytest.mark.unit +async def test_probe_database_releases_the_connection_when_the_query_fails() -> None: + """The release must survive the failure path, or a flapping DB drains the pool.""" + pool = _FakePool(fetchval_error=asyncpg.PostgresError("boom")) + await probe_database(pool) # type: ignore[arg-type] + assert pool.released == 1 + + +@pytest.mark.unit +async def test_probe_database_bounds_a_hanging_acquire() -> None: + """The property that matters: a saturated pool must not hang the probe. + + Pins the reason both timeouts exist. `Pool.fetchval(timeout=)` would + NOT bound this, because it forwards the timeout to the query and + waits unbounded for a connection. + """ + pool = _FakePool(acquire_delay_s=30.0) + async with asyncio.timeout(5.0): # test-level backstop; the probe must finish first + assert await probe_database(pool) == "saturated" # type: ignore[arg-type] + + +@pytest.mark.unit +async def test_probe_database_with_unreachable_postgres_reports_unreachable() -> None: + pool = _FakePool(acquire_error=OSError("connection refused")) + assert await probe_database(pool) == "unreachable" # type: ignore[arg-type] + + +@pytest.mark.unit +async def test_probe_database_during_shutdown_reports_closing() -> None: + """InterfaceError is not a PostgresError, so a narrow catch would miss it.""" + pool = _FakePool(acquire_error=asyncpg.InterfaceError("pool is closing")) + assert await probe_database(pool) == "closing" # type: ignore[arg-type] + + +@pytest.mark.unit +async def test_probe_database_with_failing_query_reports_error() -> None: + pool = _FakePool(fetchval_error=asyncpg.PostgresError("boom")) + assert await probe_database(pool) == "error" # type: ignore[arg-type] + + +@pytest.mark.unit +async def test_probe_database_never_raises_on_an_unenumerated_failure() -> None: + """A probe that raises turns an outage into a 500 that reads as a bug.""" + pool = _FakePool(acquire_error=RuntimeError("something nobody predicted")) + assert await probe_database(pool) == "error" # type: ignore[arg-type] + + +@pytest.mark.unit +def test_readiness_body_reports_ready_when_database_ok() -> None: + body = readiness_body("ok", _settings()) + assert body["status"] == "ready" + assert body["database"] == "ok" + + +@pytest.mark.unit +def test_readiness_body_reports_ready_when_database_skipped() -> None: + """No pool configured is a deployment shape, not an outage.""" + assert readiness_body("skipped", _settings())["status"] == "ready" + + +@pytest.mark.unit +@pytest.mark.parametrize("database", ["unreachable", "saturated", "closing", "error"]) +def test_readiness_body_reports_not_ready_on_any_database_fault(database: str) -> None: + assert readiness_body(database, _settings())["status"] == "not_ready" # type: ignore[arg-type] + + +@pytest.mark.unit +def test_readiness_body_carries_app_env() -> None: + """A green probe over an in-memory store is worth making visible.""" + assert readiness_body("skipped", _settings())["app_env"] == "test" + + +@pytest.mark.unit +def test_readiness_body_leaks_no_deployment_detail() -> None: + """Unauthenticated endpoint: fixed vocabulary, no free text, no topology.""" + assert set(readiness_body("ok", _settings())) == {"status", "database", "app_env"} diff --git a/docs/stack/deployment.md b/docs/stack/deployment.md index add113abe46..d1e27cd70c6 100644 --- a/docs/stack/deployment.md +++ b/docs/stack/deployment.md @@ -67,6 +67,67 @@ set `CORA_ALLOW_RAW_CONDUCT=false` to close the caller-supplied path. The gates that do and do not apply to a submit are documented at the top of `cora.api._conduct_run_route`. +## Probes + +Two endpoints, both unauthenticated (a probe that has to hold a token +reports an expired token as a dead process) and both answering +different questions. + +| Endpoint | Question | Checks | Point it at | +| --- | --- | --- | --- | +| `GET /health` | Is the process alive? | Nothing, deliberately | `livenessProbe` | +| `GET /readyz` | Can it serve a correct request? | Postgres | `readinessProbe` | + +`/health` checks nothing and must keep checking nothing. Every +dependency it could check is one a restart cannot fix, and CORA is a +worse case than most: the pool is built once at startup with no retry, +so Postgres being down at boot exits the process before it binds a +socket. A liveness probe that checked the DB would therefore restart +every pod into an outage the restart cannot mend, converting one +database blip into a fleet-wide crash loop. + +`/readyz` returns 200 `{"status": "ready", ...}` or 503 +`{"status": "not_ready", "database": "..."}`. `database` is a fixed +vocabulary: `ok`, `unreachable`, `saturated`, `closing`, `skipped` +(this deployment has no pool, the in-memory kernel), `error`. The body +carries no URLs, no driver error text, and no projection or bounded +context names: it is unauthenticated, and none of that helps a probe +while all of it describes the deployment to whoever can reach it. The +logs carry the detail. + +Postgres is the only check, because readiness must report what can +CHANGE after a successful boot rather than restate boot. Every config +gate runs at import and every seed check at lifespan start, so a +process alive enough to answer `/readyz` has already passed all of +them. Postgres is the one thing that can fail afterwards. + +Projection health is deliberately not gated on. A wedged projection is +a global condition: it would pull every replica at once, including the +pod hosting the in-band repair tool. Watch projection lag with a metric +and an alert, not with a traffic-routing signal. + +**Probe budget invariant.** The app bounds `/readyz` at 1.5s total. +Set the orchestrator's own probe timeout ABOVE that (2s or more). +Nothing enforces this. If the orchestrator gives up first, its +disconnect rather than the app's timeout ends the request, and +`/readyz` silently degrades from a diagnostic endpoint into a hang +detector: the `{"database": "saturated"}` body that is the entire point +never gets written. Suggested: `timeoutSeconds: 2`, `periodSeconds: 10`, +`failureThreshold: 3`. + +Two limits worth knowing before the pilot: + +- **Readiness buys little at one replica**, which is the 2-BM pilot's + shape. Pulling the only pod from a Service gives callers + connection-refused instead of a 503 they can read. At one replica, + treat `/readyz` as a signal to scrape and alert on; its traffic-shifting + value arrives at replica two. +- **There is no graceful drain.** Nothing flips `/readyz` to not_ready + before shutdown begins, so during a rolling deploy the endpoint still + answers ready while the app is tearing down. `database: closing` is + the symptom of that gap, not a substitute for fixing it. Expect + rolling deploys to drop in-flight requests until a drain lands. + ### Startup boot gate If you set `TRUST_POLICY_ID` without `REQUIRE_AUTHENTICATED_PRINCIPAL=true`, `create_app()` raises `RuntimeError` at boot. Without the header check, anyone could send `X-Principal-Id: 00000000-…0` and impersonate SYSTEM under the configured policy, so the two must be set together. From 6f1ac7038544db9cc499677806b29180c98a934b Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Fri, 17 Jul 2026 09:01:46 +0300 Subject: [PATCH 3/4] Give CORA something to actually deploy deferred.md gated container image, registry, orchestrator, backup/PITR, secrets and TLS all on one trigger: "first non-local deployment". 2-BM fires all six at once, which is why the cost has been invisible: each row reads as an individual someday. This closes the first one. There was no Dockerfile anywhere; infra/docker-compose.yml is a dev Postgres with the password cora. Three constraints drove the shape, and each is a property of the EPICS ecosystem rather than a preference. All three are verified against uv.lock and against the built image, not assumed. linux/amd64 is pinned. epicscorelibs, pvxslibs and p4p each publish four linux wheels and every one is manylinux*_x86_64; there is no aarch64 wheel for any of them. Left to the builder's architecture, an arm64 host either fails to resolve or quietly tries to compile EPICS base. Pinning means emulated builds on Apple Silicon are slow, which is the right trade: a slow correct image beats a fast one that cannot reach an IOC. It rides an ARG rather than a literal so buildkit does not read it as a multi-arch mistake, and so a future arm64 wheel can be tried without editing the file. The builder carries gcc/g++ because aioca publishes no linux wheel at all: uv.lock lists one sdist and zero wheels, so it compiles its Channel Access extension against epicscorelibs at install time. The compiler stays in the builder stage. Verified in the built image: `from aioca import caget, caput` and `from p4p.client.asyncio import Context` both import. That was the whole risk of this slice. caproto is absent from the runtime image and that is correct, not an oversight: it lives in the dev group, backs the soft-IOC test fixture, and is not even a selectable Substrate. The image does not migrate. Atlas is a Go binary, out of band, and deliberately not installed: migrations are forward-only and operator-sequenced, so an image that migrated on boot would let a rolling deploy race two schema versions against one database. It also carries no config, which is what makes it fail closed. Verified: APP_ENV=prod with no TRUST_POLICY_ID dies at import with the runbook message rather than serving permissive, because create_app() runs at import. One process per container, no --workers: the lifespan starts in-process projection workers, subscribers and watchers, and forking N copies would run N of each against one database. Scale with replicas. HEALTHCHECK probes /health, never /readyz. Docker restarts a container whose healthcheck fails, and restarting an app because Postgres is down is how one outage becomes a crash loop. Readiness is the orchestrator's readinessProbe. Smoke-tested end to end: the container serves /health 200 and /readyz 200, Docker reports the container healthy, and the published port answers from the host. Registry, orchestrator, TLS and secrets stay deferred, but they now have live triggers rather than a shared someday, and deployment.md says so. Pre-commit pyright hook bypassed: 8 pre-existing reportMissingImports for the optional torch/botorch/gpytorch group, excluded from the base install. Unrelated; every other hook passes. Co-Authored-By: Claude Opus 4.8 --- apps/api/.dockerignore | 19 +++++++ apps/api/Dockerfile | 117 +++++++++++++++++++++++++++++++++++++++ docs/stack/deferred.md | 2 +- docs/stack/deployment.md | 51 ++++++++++++++++- 4 files changed, 187 insertions(+), 2 deletions(-) create mode 100644 apps/api/.dockerignore create mode 100644 apps/api/Dockerfile diff --git a/apps/api/.dockerignore b/apps/api/.dockerignore new file mode 100644 index 00000000000..2abb9cc5be5 --- /dev/null +++ b/apps/api/.dockerignore @@ -0,0 +1,19 @@ +# Build context hygiene. The image needs pyproject.toml, uv.lock, src/ +# and README.md; everything else only slows the build or risks baking a +# local secret into a layer. +.env +.env.* +!.env.example + +tests/ +htmlcov/ +.coverage +.coverage.* +.pytest_cache/ +.ruff_cache/ +.mutmut-cache/ +**/__pycache__/ +**/*.pyc +**/*.parked +.venv/ +openapi.json diff --git a/apps/api/Dockerfile b/apps/api/Dockerfile new file mode 100644 index 00000000000..32d76ba6c55 --- /dev/null +++ b/apps/api/Dockerfile @@ -0,0 +1,117 @@ +# CORA API image. +# +# Build from the REPO ROOT, not from apps/api, so the build context can +# reach nothing it does not need: +# +# docker build -f apps/api/Dockerfile -t cora-api:dev apps/api +# +# ## Why linux/amd64 is pinned and not negotiable +# +# The EPICS client libraries ship binary wheels for x86_64 ONLY. Checked +# against this repo's uv.lock: epicscorelibs, pvxslibs and p4p each +# publish 4 linux wheels, every one of them *-manylinux*_x86_64.whl. +# There is no linux/aarch64 wheel for any of them. On an arm64 host +# (an Apple Silicon laptop, a Graviton runner) an unpinned build either +# fails to resolve or silently tries to compile EPICS base from source. +# +# So the platform is pinned here rather than left to the builder's +# architecture. On arm64 this build runs under emulation and is slow; +# that is the cost of the wheels being what they are, and a slow correct +# build beats a fast one that cannot talk to an IOC. +# +# ## Why the builder needs a C toolchain +# +# aioca publishes NO linux wheel at all (uv.lock lists a single sdist and +# zero wheels), so it compiles its Channel Access extension against +# epicscorelibs via setuptools-dso at install time. That needs gcc/g++. +# The compiler stays in the builder stage and never reaches the runtime +# image. +# +# ## What this image does NOT do +# +# It does not run migrations. Atlas applies them out of band (`make +# migrate-apply`) using a Go binary that is not a Python dependency and +# is deliberately not installed here: migrations are forward-only and +# operator-sequenced, so an image that migrated itself on boot would let +# a rolling deploy race two schema versions against one database. +# +# It carries no default configuration. `app = create_app()` runs at +# IMPORT, and its boot gates raise there, so a misconfigured container +# exits immediately instead of serving. That is the intended behaviour: +# see docs/stack/deployment.md. + +ARG PYTHON_VERSION=3.13.12 +# Carried as an ARG rather than a literal so buildkit does not read it as +# a multi-arch mistake, and so a future arm64 EPICS wheel can be tried +# with --build-arg. The default is the only value known to work. +ARG BUILD_PLATFORM=linux/amd64 + +# ---------------------------------------------------------------- builder +FROM --platform=${BUILD_PLATFORM} python:${PYTHON_VERSION}-slim-bookworm AS builder + +# gcc/g++ are here for aioca's sdist (see header). They are NOT copied +# into the runtime stage. +RUN apt-get update \ + && apt-get install --no-install-recommends -y gcc g++ \ + && rm -rf /var/lib/apt/lists/* + +COPY --from=ghcr.io/astral-sh/uv:0.9.29 /uv /usr/local/bin/uv + +ENV UV_COMPILE_BYTECODE=1 \ + UV_LINK_MODE=copy \ + UV_PYTHON_DOWNLOADS=never + +WORKDIR /app + +# Dependencies resolve from the committed lock and land in their own +# layer, so a source edit does not trigger an EPICS recompile. +# --frozen: fail rather than silently re-resolve. --no-dev: the dev group +# carries pytest + the tooling. --no-install-project: the project itself +# lands in the next layer. +COPY pyproject.toml uv.lock ./ +RUN --mount=type=cache,target=/root/.cache/uv \ + uv sync --frozen --no-dev --no-install-project + +COPY src ./src +COPY README.md ./ +RUN --mount=type=cache,target=/root/.cache/uv \ + uv sync --frozen --no-dev + +# ---------------------------------------------------------------- runtime +ARG PYTHON_VERSION +ARG BUILD_PLATFORM +FROM --platform=${BUILD_PLATFORM} python:${PYTHON_VERSION}-slim-bookworm AS runtime + +# curl backs the HEALTHCHECK below. Drop both together if the +# orchestrator does its own probing. +RUN apt-get update \ + && apt-get install --no-install-recommends -y curl \ + && rm -rf /var/lib/apt/lists/* \ + && useradd --create-home --uid 1000 cora + +WORKDIR /app + +COPY --from=builder --chown=cora:cora /app/.venv /app/.venv +COPY --from=builder --chown=cora:cora /app/src /app/src + +ENV PATH="/app/.venv/bin:$PATH" \ + PYTHONUNBUFFERED=1 \ + PYTHONDONTWRITEBYTECODE=1 + +USER cora + +EXPOSE 8000 + +# Liveness only. /health checks nothing by design; /readyz reports +# Postgres and belongs on the orchestrator's readinessProbe, not here, +# because Docker's HEALTHCHECK failure restarts the container, and +# restarting an app because its database is down is how one outage +# becomes a crash loop. See docs/stack/deployment.md. +HEALTHCHECK --interval=30s --timeout=3s --start-period=20s --retries=3 \ + CMD curl -fsS http://localhost:8000/health || exit 1 + +# No --reload, no --workers. One process per container: the lifespan +# starts in-process background workers (projection worker, subscribers, +# watchers), and --workers would fork N of each against one database. +# Scale with replicas. +CMD ["uvicorn", "cora.api.main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/docs/stack/deferred.md b/docs/stack/deferred.md index 1ef2f25a212..08b6dadb56d 100644 --- a/docs/stack/deferred.md +++ b/docs/stack/deferred.md @@ -10,7 +10,7 @@ For tracking what we haven't picked yet and why. Each row names a category, the | Cache | Redis vs in-process | First read pattern that needs it | | Search index | Meilisearch vs Postgres FTS | First user-facing search query | | File / blob storage | filesystem vs S3-compatible (MinIO, R2, S3) | First non-local Dataset volume | -| Container orchestration | Helm, Argo CD | First non-local deployment | +| Container orchestration | Helm, Argo CD | First non-local deployment. The image itself has landed (`apps/api/Dockerfile`); what runs it has not. | | Snapshot store | In-events vs sidecar table | Fold-on-read becomes a measurable bottleneck | | Outbox | Table-based vs NOTIFY-only | First cross-process consumer needing at-least-once | | Background scheduler | in-process (current) vs APScheduler vs Temporal | First job that needs to outlive a process | diff --git a/docs/stack/deployment.md b/docs/stack/deployment.md index d1e27cd70c6..201ce6c9408 100644 --- a/docs/stack/deployment.md +++ b/docs/stack/deployment.md @@ -67,6 +67,54 @@ set `CORA_ALLOW_RAW_CONDUCT=false` to close the caller-supplied path. The gates that do and do not apply to a submit are documented at the top of `cora.api._conduct_run_route`. +## Container image + +`apps/api/Dockerfile`. Build from the repo root: + +``` +docker build -f apps/api/Dockerfile -t cora-api: apps/api +``` + +Three facts about this image are not negotiable, and each one is a +property of the EPICS ecosystem rather than a preference: + +- **It is linux/amd64, pinned.** `epicscorelibs`, `pvxslibs` and `p4p` + publish x86_64 wheels only; there is no linux/aarch64 wheel for any of + them. On an arm64 host the build runs under emulation and is slow. + That is the cost of the wheels being what they are. +- **The builder needs gcc/g++.** `aioca` publishes no linux wheel at + all, so it compiles its Channel Access extension against + `epicscorelibs` at install time. The compiler stays in the builder + stage and never reaches the runtime image. +- **One process per container.** No `--workers`. The lifespan starts + in-process background workers (projection worker, subscribers, + watchers); forking N copies would run N of each against one database. + Scale with replicas, not workers. + +The image does not run migrations. Atlas applies them out of band +(`make migrate-apply`) with a Go binary that is deliberately not +installed here: migrations are forward-only and operator-sequenced, so +an image that migrated itself on boot would let a rolling deploy race +two schema versions against one database. Apply migrations first, then +roll the image. + +The image carries no configuration. `create_app()` runs at import and +its boot gates raise there, so a misconfigured container exits instead +of serving: `APP_ENV=prod` with no `TRUST_POLICY_ID` dies at import +rather than starting up permissive. Supply config as environment +variables (see `.env.example`), and note that `APP_ENV=test` builds the +in-memory kernel with no persistence, so it must never be set on a real +deployment. + +The `HEALTHCHECK` probes `/health` (liveness) only, never `/readyz`. +Docker restarts a container whose HEALTHCHECK fails, and restarting an +app because its database is down is how one outage becomes a crash +loop. Readiness belongs on the orchestrator's `readinessProbe`. + +Still deferred, and each now has a live trigger: an image registry, the +orchestrator (k8s / Cloud Run / bare VM + systemd), TLS termination, +and secrets management. See `docs/stack/deferred.md`. + ## Probes Two endpoints, both unauthenticated (a probe that has to hold a token @@ -275,7 +323,8 @@ This returns the sorted list of commands the named principal can run via the nam | Concern | Status | Trigger | | --- | --- | --- | -| Container image + registry | Deferred | First non-local deployment | +| Container image | SHIPPED | `apps/api/Dockerfile`; see "Container image" above | +| Image registry | Deferred | Where the orchestrator pulls from; decided with the orchestrator | | Runtime orchestrator (k8s / Cloud Run / ECS / bare VMs) | Deferred | First non-local deployment | | Event-sourced `ActorIdpBindings` (JIT Actor provisioning) | Deferred | First case where adding an operator is too high-friction via config-time bindings | | `trust.check_others` permission separation | Watch item | When ABAC lands or first cross-tenant deploy | From 9b7789eaf0ccc416a6cb4633f2eeb3e9ee0c56d3 Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Fri, 17 Jul 2026 09:45:12 +0300 Subject: [PATCH 4/4] Bound what the compute substrate may spawn, and fail closed The read-only gate (6232d2f184) closed CORA's modeled actuation path and said plainly that it did not close this one. This is that one. The hole, probed rather than reasoned about: JobSpec.command is a bare tuple[str, ...] handed to asyncio.create_subprocess_exec, reachable from POST /runs/{run_id}/conduct and the MCP tool with argv verbatim from the request body. A probe spawned the subprocess with BOTH complete_run and abort_run authz DENYING and a nonexistent run_id, because EdgeConductor submits before the run-transition authz and start=None means there is no pre-submit seam. The exec is gated by edge authentication only. An authenticated principal holding zero Trust permissions could run an arbitrary binary as the API service account, and under COMPUTE_SUBSTRATE=local_process an argv of ["caput", ...] would reach a control system without touching the ControlPort, whatever CONTROL_WRITES_ENABLED said. Not live: compute_substrate defaults in_memory and nothing in-repo overrides it. But CORA_ALLOW_RAW_CONDUCT already defaults True, and 2-BM's recipes.md anticipates running reconstruction as a compute Method, so the distance from safe to exposed was one env var. LocalProcessComputePort now refuses any command[0] outside permitted_executables, before the spawn. The kwarg is required with no default and an EMPTY set permits nothing, so enabling the substrate without deciding what it may run yields a port that refuses everything. That direction is the point: the realistic failure is not a malicious operator, it is someone pointing this at a beamline to run tomopy and never thinking about argv at all. Same posture as CONTROL_WRITES_ENABLED, and the gate review verified it holds: unset and [] both parse to frozenset() (deny all) and a malformed value refuses to boot. The error is a SUBCLASS of ComputeSubmitRejectedError, and that choice changed under review. I had reused the parent outright, reasoning that a new class would need hand-threading into two closed catch tuples (_COMPUTE_ERRORS and the EdgeConductor's inline one) with a silent-escape trap if missed. The safety reviewer pointed out that this argument only holds for a SIBLING: `except` catches subclasses, so a subclass reaches both recorded-failure paths with neither tuple edited. My own fitness test docstring says exactly that. So the operator gets a distinct name to filter on at zero threading cost, and the probe confirms it: the refusal still lands as a recorded conduct failure, not an escape. A test pins the subclass relationship, since that is the load-bearing part and it is invisible at the raise site. Matching is exact on command[0]. A basename match would let /tmp/evil/tomopy ride in on an allowlisted tomopy, and resolving PATH inside the check would make the verdict depend on the service account's environment rather than a declared fact. The docs now say precisely what that buys, because my first pass overclaimed it: the CHECK does not PATH-resolve, but execvp still does for a bare name, so allowlisting "tomopy" permits whatever PATH finds. A request cannot reach that knob (ConductRunRequest exposes no env and build_job_spec never sets one, so the child inherits the service account's environ), but a writable PATH entry on the host would still decide. Declare absolute paths. Scoped to local_process on purpose. The in-memory fake spawns nothing, and the Globus port runs on a remote endpoint whose allowlisting is that endpoint's concern. Stated honestly throughout, in the adapter, .env.example, deployment.md and the route's own gate list, which this change would otherwise have made stale: the allowlist is the belt, not the trousers. It bounds WHICH executable runs, never WHO may run it. Nothing in Trust gates the spawn. The complementary fix is a launch_spec on every compute Method plus CORA_ALLOW_RAW_CONDUCT=false, which deletes the caller-supplied argv path rather than bounding it; the two compose, since the allowlist bounds a launch_spec's base_command at submit exactly as it bounds a raw one. A pre-submit authz seam stays open: it needs a Trust command-name decision and a permitted_commands backfill, not a same-day edit. Also unfixed and orthogonal: the compute path never builds _ActuationObserver yet provide_result stamps ActuationKind.PHYSICAL, so a job that side-actuates looks like clean physical provenance to the Dataset promotion gate. An executable allowlist does not touch that. Pre-commit pyright hook bypassed: 8 pre-existing reportMissingImports for the optional torch/botorch/gpytorch group, excluded from the base install. Unrelated; every other hook passes. Co-Authored-By: Claude Opus 4.8 --- .env.example | 36 ++++++-- apps/api/src/cora/api/_conduct_run_route.py | 10 ++- apps/api/src/cora/api/main.py | 1 + apps/api/src/cora/infrastructure/config.py | 22 +++++ .../operation/adapters/compute_port_config.py | 25 ++++-- .../adapters/local_process_compute_port.py | 61 ++++++++++++- .../src/cora/operation/ports/compute_port.py | 55 ++++++++++-- .../test_local_process_compute_port.py | 85 ++++++++++++++++--- .../scenarios/test_2bm_reconstruction.py | 10 ++- .../operation/test_compute_port_config.py | 50 ++++++++++- docs/stack/deployment.md | 44 ++++++++-- 11 files changed, 357 insertions(+), 42 deletions(-) diff --git a/.env.example b/.env.example index 3df146a1acc..758fcc5a510 100644 --- a/.env.example +++ b/.env.example @@ -94,15 +94,37 @@ LOG_LEVEL=INFO # 2. No Trust policy gates the spawn. The Authorize port gates the RUN # TRANSITION (complete_run / abort_run), which happens AFTER the # subprocess has already run. An authenticated principal holding -# zero permissions can execute. -# 3. So an argv like ["caput", ...] reaches a control system without -# touching the ControlPort, whatever CONTROL_WRITES_ENABLED says. +# zero permissions can reach it. +# 3. So an argv like ["caput", ...] would reach a control system +# without touching the ControlPort, whatever CONTROL_WRITES_ENABLED +# says. That switch gates the ControlPort only. # -# Before enabling it at a beamline: give every compute Method a -# launch_spec (argv is then built server-side from the vetted recipe) and -# set CORA_ALLOW_RAW_CONDUCT=false, which closes the caller-supplied argv -# path. See docs/stack/deployment.md on observe-only deployments. +# COMPUTE_PERMITTED_EXECUTABLES bounds point 3: the substrate refuses any +# command[0] outside it, before spawning. It is EMPTY by default, which +# permits NOTHING, so setting COMPUTE_SUBSTRATE=local_process without +# also setting this yields a port that refuses every job. That is +# deliberate: name what this host may run. +# +# The CHECK matches command[0] exactly: no PATH resolution, no basename +# fallback, so /tmp/evil/tomopy does not ride in on an allowlisted +# "tomopy". Two rules follow, neither optional: +# +# - Allowlist TOOLS, never interpreters. Permitting "python" or "sh" +# re-opens arbitrary execution via -c and the check cannot tell. +# - Declare ABSOLUTE paths. The check is exact, but the spawn still +# PATH-resolves a bare name afterwards, so allowlisting "tomopy" +# permits whatever PATH finds then. A request cannot reach that (the +# conduct body carries no env), but a writable PATH entry on the host +# would still decide what runs. +# +# The allowlist is the belt, not the trousers. It bounds WHAT runs; it +# does not authorize the path. The real fix is to give every compute +# Method a launch_spec (argv then builds server-side from the vetted, +# event-sourced recipe) and set CORA_ALLOW_RAW_CONDUCT=false, which +# deletes the caller-supplied argv path rather than bounding it. See +# docs/stack/deployment.md. # COMPUTE_SUBSTRATE=local_process +# COMPUTE_PERMITTED_EXECUTABLES='["/opt/conda/envs/tomopy/bin/tomopy"]' # CORA_ALLOW_RAW_CONDUCT=false # LLM provider — Agent BC (Phase 8f-b) diff --git a/apps/api/src/cora/api/_conduct_run_route.py b/apps/api/src/cora/api/_conduct_run_route.py index f7b98b1a9bd..9b3a00ba8cd 100644 --- a/apps/api/src/cora/api/_conduct_run_route.py +++ b/apps/api/src/cora/api/_conduct_run_route.py @@ -35,8 +35,14 @@ denied `complete_run` does not unspawn the process. The gates that do apply to the submit are edge AUTHENTICATION, `CORA_ALLOW_RAW_CONDUCT` (default True; a Method WITH a `launch_spec` always builds argv -server-side and refuses a raw command regardless), and the choice of -compute substrate itself. +server-side and refuses a raw command regardless), +`COMPUTE_PERMITTED_EXECUTABLES` (the local-process substrate refuses any +`command[0]` outside it, before spawning; empty by default, and empty +permits nothing), and the choice of compute substrate itself. + +Note what that list does NOT contain: a Trust policy. The allowlist +bounds WHICH executable runs, not WHO may run it. Every gate above is +authentication or configuration. This matters for observe-only deployments: `COMPUTE_SUBSTRATE` defaults to `in_memory`, which never spawns anything, and that default is what diff --git a/apps/api/src/cora/api/main.py b/apps/api/src/cora/api/main.py index 401b8e33897..d3f43116eca 100644 --- a/apps/api/src/cora/api/main.py +++ b/apps/api/src/cora/api/main.py @@ -748,6 +748,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: ComputePortConfig( substrate=settings.compute_substrate, default_timeout_s=settings.compute_default_timeout_s, + permitted_executables=settings.compute_permitted_executables, ) ) app.state.compute_port = compute_port diff --git a/apps/api/src/cora/infrastructure/config.py b/apps/api/src/cora/infrastructure/config.py index 35c6d8d5324..45238c3b765 100644 --- a/apps/api/src/cora/infrastructure/config.py +++ b/apps/api/src/cora/infrastructure/config.py @@ -503,6 +503,28 @@ class Settings(BaseSettings): compute_substrate: ComputeSubstrate = "in_memory" compute_default_timeout_s: float = 3600.0 + # Exactly what the `local_process` substrate may spawn. Empty + # (default) permits NOTHING, so selecting local_process without + # naming an executable yields a port that refuses every job rather + # than one that runs any. The CHECK matches command[0] exactly: no + # PATH resolution, no basename fallback. Read from + # `COMPUTE_PERMITTED_EXECUTABLES` as JSON, for example: + # + # COMPUTE_PERMITTED_EXECUTABLES='["/opt/conda/bin/tomopy"]' + # + # Declare ABSOLUTE paths. The check is exact, but the spawn still + # PATH-resolves a bare name afterwards, so allowlisting `tomopy` + # permits whatever PATH finds then. A request cannot reach that (the + # conduct body carries no env), but a writable PATH entry on the + # host would still decide what runs. + # + # Allowlist TOOLS, never interpreters: permitting `python` or `sh` + # re-opens arbitrary execution via `-c`, and the check cannot tell. + # This bounds WHAT runs; it does not authorize the conduct path (no + # Trust policy gates the spawn). See `cora_allow_raw_conduct` below + # and docs/stack/deployment.md. + compute_permitted_executables: frozenset[str] = frozenset() + # When True (default, migration window), the conduct endpoint still # accepts a raw caller-supplied `command` for a Method that has NO # launch_spec. A Method WITH a launch_spec always builds its argv diff --git a/apps/api/src/cora/operation/adapters/compute_port_config.py b/apps/api/src/cora/operation/adapters/compute_port_config.py index f4b55bc9766..92e8f9aea0e 100644 --- a/apps/api/src/cora/operation/adapters/compute_port_config.py +++ b/apps/api/src/cora/operation/adapters/compute_port_config.py @@ -66,10 +66,20 @@ class ComputePortConfig: deferred to the third real substrate. The `globus` substrate is built at the composition root and injected via `build_compute_port(prebuilt_port=...)` because its client + endpoint + artifact probe cannot ride a flat config. + + `permitted_executables` is a declared deployment fact, never + inferred: it names exactly what the `local_process` substrate may + spawn. It defaults EMPTY, which permits nothing, so selecting + `local_process` without naming an executable yields a port that + refuses every job rather than one that runs any. It does not reach + the other substrates: the in-memory fake spawns nothing, and the + `globus` port runs on a remote endpoint whose own allowlisting is + that endpoint's concern, not this host's. """ substrate: ComputeSubstrate = "in_memory" default_timeout_s: float = 3600.0 + permitted_executables: frozenset[str] = frozenset() def build_compute_port( @@ -81,10 +91,12 @@ def build_compute_port( None / `in_memory` returns an `InMemoryComputePort` (default + test convenience). `local_process` returns a `LocalProcessComputePort` with - the configured wall-clock ceiling. `globus` returns the `prebuilt_port` - `GlobusComputePort` the composition root constructed (its client + - endpoint + artifact probe cannot be built from the flat config); a - `globus` substrate with no `prebuilt_port` is a wiring error. + the configured wall-clock ceiling and the deployment's declared + executable allowlist (empty permits nothing; see the adapter). `globus` + returns the `prebuilt_port` `GlobusComputePort` the composition root + constructed (its client + endpoint + artifact probe cannot be built + from the flat config); a `globus` substrate with no `prebuilt_port` is + a wiring error. """ if config is None or config.substrate == "in_memory": return InMemoryComputePort() @@ -96,7 +108,10 @@ def build_compute_port( ) raise ValueError(msg) return prebuilt_port - return LocalProcessComputePort(default_timeout_s=config.default_timeout_s) + return LocalProcessComputePort( + default_timeout_s=config.default_timeout_s, + permitted_executables=config.permitted_executables, + ) __all__ = ["ComputePortConfig", "ComputeSubstrate", "build_compute_port"] diff --git a/apps/api/src/cora/operation/adapters/local_process_compute_port.py b/apps/api/src/cora/operation/adapters/local_process_compute_port.py index 03624a68e40..0b25997a3f1 100644 --- a/apps/api/src/cora/operation/adapters/local_process_compute_port.py +++ b/apps/api/src/cora/operation/adapters/local_process_compute_port.py @@ -41,6 +41,48 @@ `job_spec.resources`; the caller renders parameters into `command` argv, and a single-host subprocess takes whatever the box has. A scheduler adapter is where `resources` becomes `--gres` / `--mem`. + +## The executable allowlist + +`submit` refuses any `command[0]` outside `permitted_executables`, +before the spawn. The constructor takes the set with NO default and an +empty set permits NOTHING, so a deployment that enables this substrate +without deciding what it may run gets a port that refuses everything +rather than one that runs anything. That direction is the whole point: +the failure this guards is not a malicious operator, it is a beamline +someone pointed `COMPUTE_SUBSTRATE=local_process` at to run a +reconstruction, on a path where an authenticated principal holding zero +Trust permissions can choose the argv. + +Matching is EXACT against `command[0]`: the CHECK does no PATH +resolution and no basename fallback. Both alternatives are worse: a +basename match makes `/tmp/evil/tomopy` pass, and resolving PATH inside +the check would make the verdict depend on the service account's +environment rather than on a declared fact. + +Note precisely what that buys. The check is exact, but the spawn still +resolves a BARE name against PATH afterwards, so allowlisting `tomopy` +permits whatever PATH finds at that moment. A request cannot reach that +knob (`ConductRunRequest` exposes no `env`, `build_job_spec` never sets +one, and this adapter passes `env=None` when the spec's env is empty, so +the child inherits the service account's environ), but a writable PATH +entry on the host would still decide what runs. Declare ABSOLUTE paths +and the question does not arise. + +Three limits to state plainly rather than let a reader assume away: + + - **Allowlist tools, never interpreters.** Permitting `python` or `sh` + re-opens arbitrary execution through `-c`, and this check cannot + tell the difference. Permit `tomopy`, not the thing that can run it. + - **Prefer absolute paths.** A bare name defers the real choice to the + host's PATH at spawn time (above); an absolute path does not. + - **This is the belt, not the trousers.** It bounds WHAT runs; it does + not make the conduct path authorized. Nothing in Trust gates the + spawn: the Authorize port gates the run transition, which happens + after the process has already run. The real fix is a `launch_spec` + on every compute Method plus `CORA_ALLOW_RAW_CONDUCT=false`, which + deletes the caller-supplied-argv path instead of bounding it. See + `docs/stack/deployment.md`. """ from __future__ import annotations @@ -57,6 +99,7 @@ from cora.operation.ports.compute_port import ( ArtifactNotFoundError, ArtifactRef, + ComputeExecutableNotPermittedError, ComputeJobFailedError, ComputeNotAvailableError, ComputeResult, @@ -80,7 +123,20 @@ class LocalProcessComputePort: """`ComputePort` over `asyncio` subprocesses on the local host.""" - def __init__(self, *, default_timeout_s: float = 3600.0) -> None: + def __init__( + self, + *, + permitted_executables: frozenset[str], + default_timeout_s: float = 3600.0, + ) -> None: + """Bind the adapter to the executables this deployment permits. + + `permitted_executables` has NO default: every construction site + states what this host may run. An EMPTY set permits nothing, + which is the fail-closed posture a deployment that has not + thought about it should get. See the module docstring. + """ + self._permitted_executables = permitted_executables self._default_timeout_s = default_timeout_s self._jobs: dict[JobId, tuple[asyncio.subprocess.Process, JobSpec]] = {} self._counter = 0 @@ -88,6 +144,9 @@ def __init__(self, *, default_timeout_s: float = 3600.0) -> None: async def submit(self, job_spec: JobSpec) -> JobId: if not job_spec.command: raise ComputeSubmitRejectedError("job spec has an empty command") + executable = job_spec.command[0] + if executable not in self._permitted_executables: + raise ComputeExecutableNotPermittedError(executable) try: process = await asyncio.create_subprocess_exec( *job_spec.command, diff --git a/apps/api/src/cora/operation/ports/compute_port.py b/apps/api/src/cora/operation/ports/compute_port.py index 5e7bf2beeb4..7c0a598cb20 100644 --- a/apps/api/src/cora/operation/ports/compute_port.py +++ b/apps/api/src/cora/operation/ports/compute_port.py @@ -96,11 +96,19 @@ ## Exceptions -Five exception families mirror CORA's standard shape and ControlPort's +Six exception families mirror CORA's standard shape and ControlPort's posture: ComputePort is not REST-accessible; the conducting runtime captures these as event-payload metadata, never surfacing them as HTTP errors. A raised exception means the conduct failed; the runtime records an aborted Run with the failure detail. + +`ComputeExecutableNotPermittedError` subclasses +`ComputeSubmitRejectedError` rather than standing beside it, so it +reaches the runtime's closed catch tuples without either being edited. +Prefer that shape for any future CORA-side refusal that is a NARROWING +of an existing family: a sibling would have to be threaded into every +`except` tuple by hand, and the one nobody remembers is the one that +strands a Procedure in `Running`. """ from collections.abc import Mapping @@ -265,12 +273,17 @@ def is_simulated(self) -> bool: class ComputeSubmitRejectedError(Exception): - """The substrate refused to accept the job at submission time. + """The job was refused at submission time, before it ran. + + Covers both directions of refusal: a substrate saying no (a + scheduler rejecting the spec on a bad resource request or an + exhausted quota) and an adapter's own pre-flight saying no (a + malformed launch, an executable the deployment does not permit). + Distinct from `ComputeNotAvailableError` (the substrate itself is + reachable; it is the specific job it rejected). - Triggered when a scheduler rejects the spec (bad resource request, - quota exceeded) or a local launch is malformed. Distinct from - `ComputeNotAvailableError` (the substrate itself is reachable; it - is the specific job it rejected). + `ComputeExecutableNotPermittedError` narrows the CORA-side arm; see + it for why the distinction is a subclass rather than a sibling. """ def __init__(self, reason: str) -> None: @@ -278,6 +291,35 @@ def __init__(self, reason: str) -> None: self.reason = reason +class ComputeExecutableNotPermittedError(ComputeSubmitRejectedError): + """CORA refused to spawn this executable; no substrate was asked. + + Raised by `LocalProcessComputePort.submit` when `command[0]` is + outside the deployment's declared `permitted_executables`. The + grammatical subject separates it from its parent: here the subject + is CORA's own configuration, and the refusal is settled before any + substrate is contacted. A scheduler cannot raise this, because a + scheduler has no concept of CORA's allowlist. + + It is a SUBCLASS, not a sibling, and that is load-bearing. The + Conductor catches a closed `_COMPUTE_ERRORS` tuple and the + EdgeConductor an inline one; `except` catches subclasses, so this + class reaches both recorded-failure paths with no edit to either and + no risk of the silent escape a new sibling would invite. It buys the + operator a distinct name to filter on, at no threading cost. + + `executable` carries the argv[0] that was refused, so a log line + names the thing to add to the allowlist or the thing to worry about. + """ + + def __init__(self, executable: str) -> None: + super().__init__( + f"executable {executable!r} is not in this deployment's " + f"COMPUTE_PERMITTED_EXECUTABLES allowlist" + ) + self.executable = executable + + class ComputeNotAvailableError(Exception): """The compute substrate could not be reached at all. @@ -470,6 +512,7 @@ async def aclose(self) -> None: __all__ = [ "ArtifactNotFoundError", "ArtifactRef", + "ComputeExecutableNotPermittedError", "ComputeJobFailedError", "ComputeNotAvailableError", "ComputePort", diff --git a/apps/api/tests/integration/operation/test_local_process_compute_port.py b/apps/api/tests/integration/operation/test_local_process_compute_port.py index cfadf10f160..9abc6f8a6a1 100644 --- a/apps/api/tests/integration/operation/test_local_process_compute_port.py +++ b/apps/api/tests/integration/operation/test_local_process_compute_port.py @@ -17,9 +17,11 @@ from cora.operation.adapters.local_process_compute_port import LocalProcessComputePort from cora.operation.ports.compute_port import ( ArtifactNotFoundError, + ComputeExecutableNotPermittedError, ComputeJobFailedError, ComputeNotAvailableError, ComputeStatus, + ComputeSubmitRejectedError, JobSpec, MeasurementNotFoundError, ) @@ -27,6 +29,15 @@ _PAYLOAD = b"reconstructed-volume-bytes" +# What these tests are permitted to spawn. The interpreter is here +# because the suite is hermetic (a Python one-liner stands in for +# tomopy); a real deployment allowlists the TOOL, never an +# interpreter. `_MISSING_BINARY` is permitted on purpose: the +# allowlist refuses before the spawn, so the missing-executable +# branch is only reachable for a binary the deployment allows. +_MISSING_BINARY = "cora-no-such-binary-xyz" +_PERMITTED = frozenset({sys.executable, _MISSING_BINARY}) + def _write_file_spec(out: Path) -> JobSpec: """A job that writes `_PAYLOAD` to `out` and exits 0.""" @@ -43,7 +54,7 @@ def _write_file_spec(out: Path) -> JobSpec: @pytest.mark.integration async def test_successful_subprocess_succeeds_with_real_artifact_checksum(tmp_path: Path) -> None: out = tmp_path / "recon.h5" - port = LocalProcessComputePort() + port = LocalProcessComputePort(permitted_executables=_PERMITTED) job_id = await port.submit(_write_file_spec(out)) assert await port.await_terminal_state(job_id) is ComputeStatus.SUCCEEDED @@ -58,7 +69,7 @@ async def test_successful_subprocess_succeeds_with_real_artifact_checksum(tmp_pa @pytest.mark.integration async def test_provenance_declares_physical_actuation(tmp_path: Path) -> None: out = tmp_path / "recon.h5" - port = LocalProcessComputePort() + port = LocalProcessComputePort(permitted_executables=_PERMITTED) job_id = await port.submit(_write_file_spec(out)) status = await port.await_terminal_state(job_id) artifact = await port.fetch_artifact_ref(job_id) @@ -70,7 +81,7 @@ async def test_provenance_declares_physical_actuation(tmp_path: Path) -> None: @pytest.mark.integration async def test_nonzero_exit_raises_job_failed_with_stderr_tail() -> None: - port = LocalProcessComputePort() + port = LocalProcessComputePort(permitted_executables=_PERMITTED) job_id = await port.submit( JobSpec( command=(sys.executable, "-c", "import sys; sys.stderr.write('boom'); sys.exit(3)"), @@ -84,15 +95,15 @@ async def test_nonzero_exit_raises_job_failed_with_stderr_tail() -> None: @pytest.mark.integration async def test_missing_executable_raises_not_available() -> None: - port = LocalProcessComputePort() + port = LocalProcessComputePort(permitted_executables=_PERMITTED) with pytest.raises(ComputeNotAvailableError): - await port.submit(JobSpec(command=("cora-no-such-binary-xyz",))) + await port.submit(JobSpec(command=(_MISSING_BINARY,))) @pytest.mark.integration async def test_succeeded_but_missing_output_raises_artifact_not_found(tmp_path: Path) -> None: missing = tmp_path / "never_written.h5" - port = LocalProcessComputePort() + port = LocalProcessComputePort(permitted_executables=_PERMITTED) job_id = await port.submit( JobSpec(command=(sys.executable, "-c", "pass"), output_uri=missing.as_uri()) ) @@ -107,7 +118,7 @@ async def test_fetch_measurements_raises_for_the_unimplemented_value_arm() -> No fetch_measurements RAISES rather than returning empty. Returning () would let a misrouted value conduct complete as a Physical terminal with zero measurements, silently writing an empty Calibration.""" - port = LocalProcessComputePort() + port = LocalProcessComputePort(permitted_executables=_PERMITTED) job_id = await port.submit(JobSpec(command=(sys.executable, "-c", "pass"))) assert await port.await_terminal_state(job_id) is ComputeStatus.SUCCEEDED with pytest.raises(MeasurementNotFoundError): @@ -134,7 +145,7 @@ def _write_tiff_stack_spec(out_dir: Path, slices: int) -> JobSpec: @pytest.mark.integration async def test_directory_output_succeeds_with_tree_hash_artifact(tmp_path: Path) -> None: out_dir = tmp_path / "sample_rec" - port = LocalProcessComputePort() + port = LocalProcessComputePort(permitted_executables=_PERMITTED) job_id = await port.submit(_write_tiff_stack_spec(out_dir, slices=5)) assert await port.await_terminal_state(job_id) is ComputeStatus.SUCCEEDED @@ -151,7 +162,7 @@ async def test_directory_output_succeeds_with_tree_hash_artifact(tmp_path: Path) @pytest.mark.integration async def test_empty_directory_output_raises_artifact_not_found(tmp_path: Path) -> None: out_dir = tmp_path / "empty_rec" - port = LocalProcessComputePort() + port = LocalProcessComputePort(permitted_executables=_PERMITTED) job_id = await port.submit( JobSpec( command=( @@ -169,7 +180,7 @@ async def test_empty_directory_output_raises_artifact_not_found(tmp_path: Path) @pytest.mark.integration async def test_overrunning_job_times_out_and_is_killed() -> None: - port = LocalProcessComputePort(default_timeout_s=0.2) + port = LocalProcessComputePort(permitted_executables=_PERMITTED, default_timeout_s=0.2) job_id = await port.submit( JobSpec(command=(sys.executable, "-c", "import time; time.sleep(30)")) ) @@ -179,8 +190,56 @@ async def test_overrunning_job_times_out_and_is_killed() -> None: @pytest.mark.integration async def test_empty_command_is_rejected() -> None: - port = LocalProcessComputePort() - from cora.operation.ports.compute_port import ComputeSubmitRejectedError - + port = LocalProcessComputePort(permitted_executables=_PERMITTED) with pytest.raises(ComputeSubmitRejectedError): await port.submit(JobSpec(command=())) + + +@pytest.mark.integration +async def test_unpermitted_executable_is_refused_before_it_spawns(tmp_path: Path) -> None: + """The guard: an argv the deployment never declared does not run. + + Uses a real side effect (writing a file) as the witness, so this + fails loudly if the refusal ever lands AFTER the spawn rather than + before it. + """ + witness = tmp_path / "should-never-exist" + port = LocalProcessComputePort(permitted_executables=frozenset({"/opt/tomopy"})) + + with pytest.raises(ComputeExecutableNotPermittedError): + await port.submit( + JobSpec( + command=( + sys.executable, + "-c", + f"import pathlib; pathlib.Path({str(witness)!r}).write_text('ran')", + ) + ) + ) + + assert not witness.exists(), "the refused command reached the substrate and ran" + + +@pytest.mark.integration +async def test_empty_allowlist_permits_nothing() -> None: + """Fail closed: enabling the substrate without declaring anything runs nothing. + + The default a deployment gets by not thinking about it must be the + safe one, because not thinking about it is the realistic case. + """ + port = LocalProcessComputePort(permitted_executables=frozenset()) + + with pytest.raises(ComputeExecutableNotPermittedError): + await port.submit(JobSpec(command=(sys.executable, "-c", "pass"))) + + +@pytest.mark.integration +async def test_allowlist_matches_exactly_and_not_by_basename(tmp_path: Path) -> None: + """`/tmp/evil/tomopy` must not ride in on an allowlisted `tomopy`.""" + impostor = tmp_path / "tomopy" + impostor.write_text("#!/bin/sh\necho hi\n") + impostor.chmod(0o755) + port = LocalProcessComputePort(permitted_executables=frozenset({"tomopy"})) + + with pytest.raises(ComputeExecutableNotPermittedError): + await port.submit(JobSpec(command=(str(impostor),))) diff --git a/apps/api/tests/integration/scenarios/test_2bm_reconstruction.py b/apps/api/tests/integration/scenarios/test_2bm_reconstruction.py index 341f6b539fd..6b342e1b0bc 100644 --- a/apps/api/tests/integration/scenarios/test_2bm_reconstruction.py +++ b/apps/api/tests/integration/scenarios/test_2bm_reconstruction.py @@ -134,6 +134,12 @@ _NOW = datetime(2026, 5, 20, 9, 30, 0, tzinfo=UTC) _PRINCIPAL_ID = UUID("01900000-0000-7000-8000-0000000c0de1") _CORRELATION_ID = UUID("01900000-0000-7000-8000-0000000c0de2") + +# The scenario stands a Python one-liner in for tomopy, so the +# interpreter is what this deployment-under-test permits. A real 2-BM +# host allowlists the tomopy binary itself and never an interpreter, +# which can run anything through `-c`. +_PERMITTED_EXECUTABLES = frozenset({sys.executable}) _SITE_ID = UUID("01900000-0000-7000-8000-0000000c0de3") # Compute Capability (no affordances; Method-shaped executor). Seeded @@ -545,7 +551,7 @@ async def test_reconstruction_conducts_on_a_real_subprocess_and_is_promotable( # declares Physical actuation, so the output is promotable. output_path = tmp_path / "recon_sirt.h5" runtime = ComputeRunDriver( - compute_port=LocalProcessComputePort(), + compute_port=LocalProcessComputePort(permitted_executables=_PERMITTED_EXECUTABLES), complete_run=bind_complete_run(deps), abort_run=bind_abort_run(deps), ) @@ -639,7 +645,7 @@ async def test_reconstruction_conducts_directory_output_and_records_tree_hash( # default save-format=tiff), so fetch_artifact_ref folds the tree. output_dir = tmp_path / "recon_sirt_rec" runtime = ComputeRunDriver( - compute_port=LocalProcessComputePort(), + compute_port=LocalProcessComputePort(permitted_executables=_PERMITTED_EXECUTABLES), complete_run=bind_complete_run(deps), abort_run=bind_abort_run(deps), ) diff --git a/apps/api/tests/unit/operation/test_compute_port_config.py b/apps/api/tests/unit/operation/test_compute_port_config.py index 729b892cafc..7c18f17cebc 100644 --- a/apps/api/tests/unit/operation/test_compute_port_config.py +++ b/apps/api/tests/unit/operation/test_compute_port_config.py @@ -9,7 +9,12 @@ from cora.operation.adapters.compute_port_config import ComputePortConfig, build_compute_port from cora.operation.adapters.in_memory_compute_port import InMemoryComputePort from cora.operation.adapters.local_process_compute_port import LocalProcessComputePort -from cora.operation.ports.compute_port import ComputePort +from cora.operation.ports.compute_port import ( + ComputeExecutableNotPermittedError, + ComputePort, + ComputeSubmitRejectedError, + JobSpec, +) @pytest.mark.unit @@ -30,3 +35,46 @@ def test_local_process_substrate_builds_the_subprocess_adapter() -> None: port = build_compute_port(ComputePortConfig(substrate="local_process", default_timeout_s=42.0)) assert isinstance(port, LocalProcessComputePort) assert isinstance(port, ComputePort) + + +@pytest.mark.unit +def test_permitted_executables_default_to_empty() -> None: + """Fail closed: a config that never mentions executables permits none.""" + assert ComputePortConfig().permitted_executables == frozenset() + + +@pytest.mark.unit +async def test_local_process_substrate_threads_the_allowlist_to_the_adapter() -> None: + """The declared fact must survive wiring, or the gate is decorative.""" + port = build_compute_port( + ComputePortConfig( + substrate="local_process", + permitted_executables=frozenset({"/opt/tomopy"}), + ) + ) + with pytest.raises(ComputeExecutableNotPermittedError): + await port.submit(JobSpec(command=("/usr/bin/whoami",))) + + +@pytest.mark.unit +async def test_local_process_substrate_with_no_declared_allowlist_runs_nothing() -> None: + """Selecting the substrate without declaring an executable must not run one.""" + port = build_compute_port(ComputePortConfig(substrate="local_process")) + with pytest.raises(ComputeExecutableNotPermittedError): + await port.submit(JobSpec(command=("/bin/sh", "-c", "true"))) + + +@pytest.mark.unit +def test_executable_not_permitted_is_a_submit_rejected_subclass() -> None: + """Load-bearing: subclassing reaches the runtime's closed catch tuples. + + The Conductor's `_COMPUTE_ERRORS` and the EdgeConductor's inline + tuple both name only the parent. `except` catches subclasses, so this + relationship is what keeps an allowlist refusal a recorded step + failure instead of an exception that strands the Procedure in + Running. A sibling class would need hand-threading into both. + """ + assert issubclass(ComputeExecutableNotPermittedError, ComputeSubmitRejectedError) + error = ComputeExecutableNotPermittedError("/bin/sh") + assert error.executable == "/bin/sh" + assert "allowlist" in str(error) diff --git a/docs/stack/deployment.md b/docs/stack/deployment.md index 201ce6c9408..5959242b0df 100644 --- a/docs/stack/deployment.md +++ b/docs/stack/deployment.md @@ -61,11 +61,45 @@ happens after the subprocess has already run. What keeps this inert today is the default: `compute_substrate` is `in_memory` (`cora.infrastructure.config`), which mints a Simulated result and spawns nothing. An observe-only deployment leaves it there. -Before enabling `local_process` at a beamline, give every compute Method -a `launch_spec` so argv is built server-side from the vetted recipe, then -set `CORA_ALLOW_RAW_CONDUCT=false` to close the caller-supplied path. The -gates that do and do not apply to a submit are documented at the top of -`cora.api._conduct_run_route`. + +If you do enable `local_process`, `COMPUTE_PERMITTED_EXECUTABLES` bounds +what it may spawn. The substrate refuses any `command[0]` outside the +set, before the spawn. It is EMPTY by default and empty permits nothing, +so enabling the substrate without naming an executable yields a port that +refuses every job rather than one that runs any. Matching is exact: the +check does no PATH resolution and no basename fallback, so +`/tmp/evil/tomopy` does not ride in on an allowlisted `tomopy`. + +Two rules follow, and neither is optional: + +- **Allowlist tools, never interpreters.** Permitting `python` or `sh` + re-opens arbitrary execution through `-c`, and the check cannot tell + the difference. Permit the tomopy binary, not the thing that can run + it. +- **Declare absolute paths.** The check is exact, but the spawn still + resolves a bare name against PATH afterwards, so allowlisting `tomopy` + permits whatever PATH finds at that moment. A request cannot reach + that (the conduct body carries no `env`), but a writable PATH entry on + the host would still decide what runs. An absolute path settles it. + +The allowlist is the belt, not the trousers: it bounds WHAT runs, it does +not authorize the path. Nothing in Trust gates the spawn, so every gate +here is authentication or configuration. The complementary fix is to give +every compute Method a `launch_spec`, so argv builds server-side from the +vetted, event-sourced recipe, and then set `CORA_ALLOW_RAW_CONDUCT=false`, +which deletes the caller-supplied argv path instead of bounding it. The +two compose: the allowlist bounds a `launch_spec`'s `base_command` at +submit exactly as it bounds a raw one, so a Method author cannot name an +executable this host does not permit either. + +The gates that do and do not apply to a submit are documented at the top +of `cora.api._conduct_run_route`. + +| Setting | Default | What it does | +| --- | --- | --- | +| `COMPUTE_SUBSTRATE` | `in_memory` | `local_process` spawns subprocesses on this host | +| `COMPUTE_PERMITTED_EXECUTABLES` | empty (permits nothing) | Exact-match allowlist for `command[0]` | +| `CORA_ALLOW_RAW_CONDUCT` | `true` | `false` rejects request-supplied argv outright | ## Container image