diff --git a/.env.example b/.env.example index 416af9aad5f..758fcc5a510 100644 --- a/.env.example +++ b/.env.example @@ -35,6 +35,98 @@ 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 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. +# +# 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) # 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/.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/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/_conduct_run_route.py b/apps/api/src/cora/api/_conduct_run_route.py index 587e877a177..9b3a00ba8cd 100644 --- a/apps/api/src/cora/api/_conduct_run_route.py +++ b/apps/api/src/cora/api/_conduct_run_route.py @@ -19,13 +19,36 @@ (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), +`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 +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/_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 1bc8072a825..d3f43116eca 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 @@ -682,7 +683,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( @@ -744,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 @@ -1078,15 +1083,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 @@ -1125,9 +1135,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/config.py b/apps/api/src/cora/infrastructure/config.py index f0a88a69b25..45238c3b765 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 @@ -484,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/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/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/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/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/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/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/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/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/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/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/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/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/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/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/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/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 b57ce16b3cb..5959242b0df 100644 --- a/docs/stack/deployment.md +++ b/docs/stack/deployment.md @@ -14,6 +14,201 @@ 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. + +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 + +`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 +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 @@ -162,7 +357,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 |