diff --git a/changelog.d/3131.added.md b/changelog.d/3131.added.md new file mode 100644 index 000000000..111535108 --- /dev/null +++ b/changelog.d/3131.added.md @@ -0,0 +1,14 @@ +- **`[sandbox].mode = "isolated"`: the subprocess sandbox worker, with `db_lookup` and + `fhir_lookup` brokered through the engine.** A Handler's lookup is sent to the engine over the + worker pipe and answered by the same lookup runner `mode = "off"` uses, so it meets the same + checks. The engine refuses a brokered lookup bound to a Router or an `accepts=` predicate, and one + whose `[egress].allowed_db` or `[egress].allowed_http` list is empty, even under the audited + opt-out `[security].block_unlisted_outbound = false`. `wall_seconds` bounds the whole call, + the time the engine spends serving its lookups included. + This mode adds no operating-system confinement yet: an `isolated` worker can open files and + sockets exactly as a `subprocess` worker can. See [CONFIGURATION.md](../docs/CONFIGURATION.md), + section `[sandbox]`. (vault `BACKLOG #3131`) +- **Every sandbox worker mode now disables `[pipeline].fuse_thread_hops`,** not only + `subprocess`. Under `isolated`, a runner whose graph has no config directory refuses at + construction, rather than run its Handlers in the engine process with unbrokered lookups. + (vault `BACKLOG #3131`) diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index dee9f7a47..9409aba3d 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -1311,6 +1311,51 @@ matters most in the other direction: setting `mode="off"` because one Handler ne takes every other Router and Handler out of the sandbox too. To isolate some feeds and not others you need separate engine processes. +**`mode="isolated"` is `subprocess` plus a lookup broker** ([ADR +0147](adr/0147-hardened-runtime-isolation-for-router-handler-code-ipc-brokered-sandbox-extends-adr-0087.md), +vault BACKLOG #3131). It runs the same per-inbound worker, under the same caps and the same codec. +It differs from `subprocess` in at least these ways: + +- **A Handler's `db_lookup`/`fhir_lookup` works.** The worker sends the call to the engine over the + worker pipe, and the engine answers it with the same lookup runner it uses at `mode="off"`, against + the same `DatabaseLookup`/`FhirLookup` connections. So a lookup there meets every check it meets + in-process, at least the `[egress]` host check made when the connection is built, the read-only + statement test, `max_rows` and the result timeout. Its result, or its error, reaches the Handler as + it would at `mode="off"`, so a Handler that catches `DbLookupError` or `FhirLookupError` still can. + A `Decimal` column value arrives as a `Decimal`. A value outside the worker pipe's types, such as a + `UUID`, fails that lookup with an error that says so. +- **A lookup whose `[egress]` list is empty is refused,** `[egress].allowed_db` for `db_lookup` and + `[egress].allowed_http` for `fhir_lookup`. **That holds under the audited opt-out + `[security].block_unlisted_outbound = false` too,** where an empty list would otherwise mean + unrestricted, so the opt-out cannot open a lookup from the worker to an unlisted host. The first + refusal per inbound and list logs a `WARNING`; each refused call fails like any lookup error. + A call made where no lookup runner exists fails with the `mode="off"` error first, so a graph + that declares no `DatabaseLookup` is never told to edit `[egress]`. +- **It is still a Handler-only call.** A Router or an `accepts=` predicate that calls either lookup + fails as it does at `mode="off"`, and the engine refuses a brokered call bound to either of them. + The dry run has no lookup runner, so it fails there as in every mode. +- **`wall_seconds` counts the lookups.** The cap bounds the whole call, including the time the engine + spends serving its lookups, so raise it for a feed whose lookups are slow. When the cap runs out + during a lookup, the call fails at once and its worker is recycled. A lookup that has already + reached its runner runs on to its own timeout in the engine, and its result is discarded. One that + has not reached it is normally not started. One that was a moment from its runner call when the + cap ran out can still start; it is then recorded and holds the connection like any other. While a + lookup runs on this way, a later lookup on the same connection from that inbound waits for it, + until a quarter of its own call's cap (at most 2 seconds) is left. It is refused if it is still + waiting then, which leaves that quarter for the Handler to handle the refusal. This rule holds + per inbound, within one sandbox session: the engine starts an inbound's session afresh at a + reload or a stop, and a lookup still running from before then is not waited for. Two inbounds can + each run a lookup on one connection. It does not bound the database either: a `db_lookup` its + runner gave up on can still be running there. +- **It needs a config directory.** A graph built in code with none cannot start a worker. Under + `subprocess` such a graph quietly runs in-process; under `isolated` the runner refuses to be built + instead, because in-process its lookups would not go through the broker. + +**What `isolated` does not do yet:** confine the worker at the operating-system level. ADR 0147's +Landlock and seccomp confinement on Linux and its AppContainer token on Windows are not built, so an +`isolated` worker can still open files and sockets as the service account, exactly as a `subprocess` +worker can. *What this boundary does NOT cover*, below, holds for both modes. + **What turning it on costs you.** At least these things change relative to `mode="off"`: | | At `mode="subprocess"` | @@ -1367,8 +1412,8 @@ under `serve`. And config code there sees the full environment, not the worker's of these. The dry run has no lookup runner, so the call raises there in every mode. Unless the Handler catches that error, the fixture records `ERROR`. The gate then fails unless its `.expect` file declares `ERROR`. Under `subprocess` that matches `serve`, which refuses the lookup; at -`mode="off"` it does not, because `serve` runs the lookup there. And -`[pipeline].fuse_thread_hops` is **hard-disabled** whenever `mode="subprocess"`: fusion runs +`mode="off"` and `mode="isolated"` it does not, because `serve` runs the lookup there. And +`[pipeline].fuse_thread_hops` is **hard-disabled** whenever `mode` is not `off`: fusion runs Router/Handler code in-process on an executor hop, so honouring both would silently unsandbox the code you asked to isolate. The runner fails closed to the async sandboxed path and logs it. To get fusion you must leave `mode="off"`. @@ -1425,7 +1470,7 @@ on a live feed; a Router that only reads is unaffected. Grep your config for `os | Key | Type | Default | Notes | |---|---|---|---| -| `mode` | enum | `off` | `off` (in-process, byte-identical, no subprocess) or `subprocess` (persistent per-inbound worker child). | +| `mode` | enum | `off` | `off` (in-process, byte-identical, no subprocess), `subprocess` (persistent per-inbound worker child), or `isolated` (the same worker, with `db_lookup`/`fhir_lookup` brokered through the engine; see *`mode="isolated"`* above). Any other value refuses to load. | | `wall_seconds` | float (>0) | 5.0 | **Authoritative** wall-clock cap per Router/Handler call on **every** platform — the parent kills a worker that overruns it, so a pathological busy-loop can't wedge intake. The worker sets no `RLIMIT_CPU`, so CPU that admin code spends outside a call, such as a thread left running or a grandchild that leaves the worker's process group, has no bound at all ([ADR 0087](adr/0087-sandbox-subprocess-isolation.md), amendment of 2026-10-07). | | `mem_mb` | int (≥1) or null | 512 | POSIX-only memory cap (MiB) inside the child (no-op on Windows). `null` disables it. **On Linux it is headroom above the loaded graph:** the worker measures its own `VmData` once `load_config` has built the graph, and sets `RLIMIT_DATA` to that plus `mem_mb`. So a graph bigger than `mem_mb` still starts, and address space a thread only reserves does not count. The worker also starts with `MALLOC_ARENA_MAX=2` unless you pass your own value through `pass_environment`, so threads cannot use up the headroom for the worker's whole life. On other POSIX systems it is an absolute `RLIMIT_AS` cap on the whole worker, and the worker logs a line saying so. A call that runs out of memory dead-letters with an error naming the memory cap, and its worker exits so the next message gets a fresh one ([ADR 0087](adr/0087-sandbox-subprocess-isolation.md), amendment of 2026-10-11). | | `pass_environment` | list of names | `[]` | Extra environment variable **names** the worker is given, beyond its allowlist. Use it for a variable your config code reads. Names only; the value comes from the engine's own environment at worker start. **Refused at load** if a name is one of the engine's own: any `MEFOR_*` name, and `VAULT_TOKEN` or `PGPASSWORD`. Give a variable your config reads a name outside `MEFOR_`. Whatever you list is readable by every Router and Handler, so do not list a secret of your own either. Env: `MEFOR_SANDBOX_PASS_ENVIRONMENT`, comma-separated. | diff --git a/docs/adr/0147-hardened-runtime-isolation-for-router-handler-code-ipc-brokered-sandbox-extends-adr-0087.md b/docs/adr/0147-hardened-runtime-isolation-for-router-handler-code-ipc-brokered-sandbox-extends-adr-0087.md index 47fac4be8..0fecbb2f2 100644 --- a/docs/adr/0147-hardened-runtime-isolation-for-router-handler-code-ipc-brokered-sandbox-extends-adr-0087.md +++ b/docs/adr/0147-hardened-runtime-isolation-for-router-handler-code-ipc-brokered-sandbox-extends-adr-0087.md @@ -174,8 +174,32 @@ unresearched. - [ ] The Landlock **network-allowlist kernel floor** (≥6.7 for net rules) + the seccomp profile; the degrade path on an older kernel (FS-only Landlock, or refuse `mode=isolated`). - [ ] A **throughput benchmark** of the broker round-trip + confinement vs the ADR 0087 subprocess baseline. -- [ ] Whether this ships as `[sandbox].mode=isolated` (a third mode above `subprocess`) or a sub-flag on - `subprocess`. +- [x] Whether this ships as `[sandbox].mode=isolated` (a third mode above `subprocess`) or a sub-flag on + `subprocess`. **Settled 2026-10-27: a third mode, `isolated`.** Decided by a Manager seat under + the owner's delegation, with two companion rulings: every check meaning "runs in a sandbox + worker" reads `is not SandboxMode.OFF`, and a brokered lookup refuses while its `[egress]` list + is empty, even under `[security].block_unlisted_outbound = false` (Amendment A's open AC-2 + question). See *Build status* below. - [ ] (Added 2026-10-07, Amendment A.) How the AppContainer spawn fits the worker's suspended start and kill-on-close job (vault BACKLOG #3065). - [ ] A **WASI / CPython-on-WASM feasibility spike** (C-extension deps, throughput) before re-scoring option 3. + +## Build status (2026-10-28, vault BACKLOG #3131 slice A) + +**Built:** the `isolated` mode and decision (1), the parent-held broker. A Handler's +`db_lookup`/`fhir_lookup` in the worker writes a `lookup` frame in the closed tag set. The frame is +bound to the dispatch's `(id, phase, name)` triple and a per-dispatch sequence number (AC-6, for a +per-inbound worker, whose lane the session already is). The parent answers it on a helper thread, in +the dispatch's context, through the same accessor and runner the in-process path uses (AC-1, AC-2). +It refuses one bound to a Router or `accepts=` dispatch, and one whose `[egress].allowed_db` or +`[egress].allowed_http` is empty. A refused or failed lookup is the Handler's error, routed to +`ERROR`/dead-letter post-ACK (AC-5, for the broker). `wall_seconds` bounds the whole dispatch, the +brokered lookups and the writes of their replies included. What happens to a lookup the cap cuts +off, and to a later lookup on the same connection, is stated once, with the limits of that bound, +in [CONFIGURATION.md, `[sandbox]`](../CONFIGURATION.md#sandbox), under `mode="isolated"`. + +**Not built:** decision (2), the OS confinement: AC-3 (Landlock and seccomp), AC-4 (the AppContainer +token), and AC-7 (no opening or signalling the parent or a sibling). The first two open boxes above, +the #3065 box and the throughput benchmark still gate them. AC-6's scoping for a pooled worker +(ADR 0187) is not built either, because no pooled worker exists yet. Until AC-3 and AC-4 land, an +`isolated` worker is confined exactly as a `subprocess` worker is. diff --git a/messagefoundry/config/db_lookup.py b/messagefoundry/config/db_lookup.py index ecfc79263..5aa783f42 100644 --- a/messagefoundry/config/db_lookup.py +++ b/messagefoundry/config/db_lookup.py @@ -40,6 +40,7 @@ "DbLookupError", "LookupRunner", "db_lookup", + "is_available", "set_active", "reset", "activated", @@ -93,6 +94,13 @@ def activated(runner: LookupRunner | None) -> Iterator[None]: _active.reset(token) +def is_available() -> bool: + """Whether a lookup runner is published here, so :func:`db_lookup` would run rather than raise + its "unavailable here" error. The ``mode=isolated`` broker asks this before any refusal of its + own, so a call that would fail that way at ``mode=off`` fails that way there too (ADR 0147).""" + return _active.get() is not None + + def db_lookup( connection: str, statement: str, params: Mapping[str, Any] | None = None ) -> list[dict[str, Any]]: diff --git a/messagefoundry/config/fhir_lookup.py b/messagefoundry/config/fhir_lookup.py index fa052f5ea..224c16837 100644 --- a/messagefoundry/config/fhir_lookup.py +++ b/messagefoundry/config/fhir_lookup.py @@ -53,6 +53,7 @@ "FhirLookupError", "FhirLookupRunner", "fhir_lookup", + "is_available", "set_active", "reset", "activated", @@ -115,6 +116,13 @@ def activated(runner: FhirLookupRunner | None) -> Iterator[None]: _active.reset(token) +def is_available() -> bool: + """Whether a FHIR-lookup runner is published here, so :func:`fhir_lookup` would run rather than + raise its "unavailable here" error. The ``mode=isolated`` broker asks this before any refusal of + its own (ADR 0147).""" + return _active.get() is not None + + def fhir_lookup( connection: str, query: str, diff --git a/messagefoundry/config/settings.py b/messagefoundry/config/settings.py index 32c5f8d87..f78f56f73 100644 --- a/messagefoundry/config/settings.py +++ b/messagefoundry/config/settings.py @@ -2498,7 +2498,7 @@ class PipelineSettings(_Section): # NOT re-read it — restart to change, exactly like claim_mode). Harness A/B via # MEFOR_PIPELINE_FUSE_THREAD_HOPS. # ONE OTHER SETTING CAN CANCEL THIS ONE, and you must be told here rather than in that knob's - # docs: the runner HARD-DISABLES fusion whenever [sandbox].mode is subprocess (it ships off, so + # docs: the runner HARD-DISABLES fusion whenever [sandbox].mode is not off (it ships off, so # this bites only a site that turned the sandbox on). Fusion runs Router/Handler/accepts= code # in-process on an executor hop, so honouring both would silently run unsandboxed the code a # config asked to isolate; the runner fails CLOSED to the async sandboxed path and logs it. To @@ -2595,13 +2595,15 @@ class SandboxSettings(_Section): * **Live enrichment is refused.** ``db_lookup``/``fhir_lookup`` re-enter the event loop, which a process boundary breaks, so they fail closed inside the child. **A Handler needing either must - run ``mode="off"``** — that escape is supported and is not going away. ``[sandbox]`` is a - single **engine-wide** section: :meth:`messagefoundry.pipeline.engine.Engine.add_registry` + run ``mode="off"``, or ``mode="isolated"``** — the off escape is supported and is not going + away. ``mode="isolated"`` (ADR 0147, vault BACKLOG #3131) is the same worker with both lookups + brokered through the engine; ``docs/CONFIGURATION.md``, section ``[sandbox]``, says what it + does and does not do. ``[sandbox]`` is a single **engine-wide** section: :meth:`messagefoundry.pipeline.engine.Engine.add_registry` renders ONE ``SandboxPolicy`` for the whole graph and a connection carries no per-connection sandbox field, so ``mode="off"`` set for one Handler runs **every** Router and Handler in the process in-process, not just that one. * **``wall_seconds`` starts being enforced.** At ``mode="off"`` there is no timeout at all; at - ``mode="subprocess"`` the parent kills a worker that overruns and dead-letters that message + ``mode="subprocess"`` or ``mode="isolated"`` the parent kills a worker that overruns and dead-letters that message post-ACK. A busy-loop can no longer wedge intake, **and** a legitimately slow Handler that used to finish now dead-letters. ``startup_seconds`` and the POSIX ``mem_mb`` arm with it. * **Throughput** ~0.19 ms per dispatch with no reference view; a 20k-entry crosswalk ~4.5 ms @@ -2620,9 +2622,10 @@ class SandboxSettings(_Section): **restart to change**, exactly like ``claim_mode``).""" # off (default, byte-identical, no subprocess — and the supported escape for a Handler needing - # live enrichment) | subprocess (persistent per-inbound worker child). ENGINE-WIDE, not per - # connection: one policy is rendered for the whole graph. - mode: Literal["off", "subprocess"] = Field(default="off") + # live enrichment) | subprocess (persistent per-inbound worker child) | isolated (that worker, + # with db_lookup/fhir_lookup brokered through the engine; ADR 0147). Any other value refuses to + # load. ENGINE-WIDE, not per connection: one policy is rendered for the whole graph. + mode: Literal["off", "subprocess", "isolated"] = Field(default="off") # Authoritative wall-clock cap (seconds) per Router/Handler call on EVERY platform: the parent # kills a worker that overruns it, so a pathological busy-loop can never wedge intake. Floor > 0. # There is no RLIMIT_CPU, so CPU spent outside a call has no bound (ADR 0087, 2026-10-07). diff --git a/messagefoundry/fhirsearch.py b/messagefoundry/fhirsearch.py index ef411dcd8..ea358cfb8 100644 --- a/messagefoundry/fhirsearch.py +++ b/messagefoundry/fhirsearch.py @@ -137,6 +137,19 @@ class FhirRaw: value: str +@final +@dataclass(frozen=True, slots=True) +class ForeignSearchValue: + """A search value of a kind :func:`fhir_lookup` does not take, carried by its type NAME only. + + Not a kind an author writes, and not in ``__all__``: it is the sandbox codec's carrier. The + ``mode=isolated`` sandbox (ADR 0147) sends a Handler's ``params`` to the engine, and a value of + any other type cannot cross the pipe, so it crosses as this. :func:`resolve_search_pairs` refuses it with the words it gives the original value, in the + engine and after the engine's own checks, so the Handler sees what ``mode=off`` shows it.""" + + type_name: str + + #: One search value: data (``str``), a token whose halves are declared (``FhirToken``), or author #: syntax (``FhirRaw``). FhirSearchValue = str | FhirToken | FhirRaw @@ -163,7 +176,7 @@ def _screen(key: str, value: str, *, half: str) -> str: ) -def _resolve_one(key: str, value: FhirSearchValue) -> str: +def _resolve_one(key: str, value: object) -> str: """One ``params`` value to the string that goes on the URL, before percent-encoding.""" if isinstance(value, str): return _screen(key, value, half="value") @@ -173,10 +186,11 @@ def _resolve_one(key: str, value: FhirSearchValue) -> str: return f"{value.system}|{_screen(key, value.code, half='code half of the FhirToken')}" if isinstance(value, FhirRaw): return value.value + got = value.type_name if isinstance(value, ForeignSearchValue) else type(value).__name__ raise ValueError( f"FHIR search parameter {key!r}: a search value must be a str (data), a " f"FhirToken(system, code), or a FhirRaw(...) author-syntax value -- got " - f"{type(value).__name__}. A number is not a kind here: wrap it, str(value). " + f"{got}. A number is not a kind here: wrap it, str(value). " f"(The value is withheld: it may carry PHI.)" ) diff --git a/messagefoundry/pipeline/_sandbox_codec.py b/messagefoundry/pipeline/_sandbox_codec.py index 48607c82c..9ee5153b7 100644 --- a/messagefoundry/pipeline/_sandbox_codec.py +++ b/messagefoundry/pipeline/_sandbox_codec.py @@ -55,6 +55,7 @@ import struct from collections.abc import Mapping, Sequence from dataclasses import dataclass +from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any, Final @@ -71,6 +72,7 @@ handler_item_fault, handler_result_items, ) +from messagefoundry.fhirsearch import FhirRaw, FhirToken, ForeignSearchValue from messagefoundry.parsing.message import Message, RawMessage from messagefoundry.redaction import safe_exc @@ -79,19 +81,30 @@ "SandboxError", "SandboxCodecError", "GraphShape", + "LOOKUP_DB", + "LOOKUP_FHIR", + "LookupRequest", + "PreparedLookup", "Request", "Response", "build_payload", "decode_boot", + "decode_answer", "decode_boot_reply", "decode_frame", + "decode_lookup_reply", "decode_request", "decode_response", "encode_bootfail", "encode_boot", "encode_fail", "encode_frame", + "encode_lookup", + "encode_prepared_lookup", + "encode_lookup_fail", + "encode_lookup_ok", "encode_ok", + "prepare_lookup", "encode_ready", "encode_request", "enc_value", @@ -102,7 +115,7 @@ class SandboxError(RuntimeError): """A sandboxed Router/Handler was denied, timed out, crashed, or could not be marshalled. - Raised only on the ``subprocess`` path (``off`` never raises this). The caller treats it exactly + Raised only in a worker mode, ``subprocess`` or ``isolated`` (``off`` never raises this). The caller treats it exactly like any other post-ACK router/transform failure: ``ERROR``/dead-letter, no NAK. Defined **here**, not in :mod:`messagefoundry.pipeline.sandbox`, only because @@ -301,6 +314,11 @@ def enc_value(value: object, blobs: _Blobs, _depth: int = 0) -> Any: return blobs.ref(value) if isinstance(value, bytes | bytearray): return {"b": base64.b64encode(bytes(value)).decode("ascii")} + if isinstance(value, Decimal): + # A NUMERIC/DECIMAL/MONEY column comes back from the driver as a Decimal, so a brokered + # db_lookup row carries one wherever mode=off would hand the Handler one (ADR 0147). Its str() + # round-trips exactly, NaN and Infinity included. + return {"N": str(value)} # datetime BEFORE date — datetime is a date subclass, and tomllib produces all three natively, so a # .toml code set with a date value is a legitimate shape plain JSON refuses. if isinstance(value, _dt.datetime): @@ -368,6 +386,13 @@ def dec_value(node: Any, reader: _Reader, _depth: int = 0) -> Any: return base64.b64decode(arg, validate=True) except (ValueError, TypeError) as exc: raise SandboxCodecError(f"sandbox bytes tag is not valid base64: {exc}") from exc + if tag == "N": + if not isinstance(arg, str): + raise SandboxCodecError("sandbox decimal tag must carry a string") + try: + return Decimal(arg) + except InvalidOperation as exc: + raise SandboxCodecError("sandbox decimal tag is not a decimal number") from exc if tag == "f": if not isinstance(arg, str) or arg not in _FLOAT_TAGS: raise SandboxCodecError(f"unknown sandbox float tag {arg!r}") @@ -1032,6 +1057,7 @@ def encode_boot( forbidden: Sequence[str], mem_mb: int | None, code_sets: object, + broker: bool = False, ) -> bytes: """The one-per-spawn bootstrap frame. @@ -1039,7 +1065,12 @@ def encode_boot( from those bytes and reads nothing from disk (vault BACKLOG #2596). ``env`` is deliberately NOT marshalled: the worker's load takes only the snapshot and the worker - never read it — a dead field on the wire is a field nobody validates.""" + never read it — a dead field on the wire is a field nobody validates. + + ``broker`` tells the worker to ask the parent for a Handler's ``db_lookup``/``fhir_lookup`` + (``mode=isolated``, ADR 0147) rather than refuse it in the child. It is a request for a protocol, + not a grant: the parent answers a ``lookup`` frame only on a session whose own policy is + isolated, so a worker that claims it on any other session is dropped.""" blobs = _Blobs() header = _envelope( { @@ -1049,6 +1080,7 @@ def encode_boot( "forbidden": list(forbidden), "mem_mb": None if mem_mb is None else int(mem_mb), "code_sets": enc_code_sets(code_sets, blobs), + "broker": bool(broker), }, "boot", ) @@ -1063,6 +1095,7 @@ class Boot: forbidden: tuple[str, ...] mem_mb: int | None code_sets: dict[str, CodeSet] | None + broker: bool = False def decode_boot(body: bytes) -> Boot: @@ -1074,12 +1107,14 @@ def decode_boot(body: bytes) -> Boot: ) mem_mb = None if header.get("mem_mb") is None else _req_int(header.get("mem_mb"), "mem_mb") code_sets = dec_code_sets(header.get("code_sets"), reader) + broker = _req_bool(header.get("broker"), "boot broker") reader.finish() return Boot( snapshot=snapshot, forbidden=forbidden, mem_mb=mem_mb, code_sets=code_sets, + broker=broker, ) @@ -1267,14 +1302,14 @@ def decode_response(body: bytes, *, request_id: str, phase: str, name: str) -> R :meth:`sandbox.SandboxSession.dispatch` supplies the other half — an unsolicited frame queued before or after a dispatch is fatal to the worker — so a pre-staged answer has no window at all.""" header, reader, kind = _open(body, ("ok", "fail")) - if _req_str(header.get("id"), "response id") != request_id: - raise SandboxCodecError( - f"sandbox response correlation mismatch (expected request {request_id!r})" - ) - if _req_str(header.get("phase"), "response phase") != phase: - raise SandboxCodecError(f"sandbox response phase mismatch (expected {phase!r})") - if _req_str(header.get("name"), "response name") != name: - raise SandboxCodecError(f"sandbox response name mismatch (expected {name!r})") + return _answer(header, reader, kind, request_id=request_id, phase=phase, name=name) + + +def _answer( + header: dict[str, Any], reader: _Reader, kind: str, *, request_id: str, phase: str, name: str +) -> Response: + """The ``ok``/``fail`` body of :func:`decode_response`, on a frame already opened.""" + _check_binding(header, request_id, phase, name) if kind == "ok": if "result" not in header: raise SandboxCodecError("sandbox 'ok' frame carries no result") @@ -1287,3 +1322,265 @@ def decode_response(body: bytes, *, request_id: str, phase: str, name: str) -> R error = reader.text(header.get("error")) reader.finish() return Response(ok=False, result=None, kind=fail_kind, error=error) + + +def _check_binding(header: dict[str, Any], request_id: str, phase: str, name: str) -> None: + """Refuse a frame whose ``(id, phase, name)`` is not the dispatch's (see :func:`decode_response`).""" + if _req_str(header.get("id"), "response id") != request_id: + raise SandboxCodecError( + f"sandbox response correlation mismatch (expected request {request_id!r})" + ) + if _req_str(header.get("phase"), "response phase") != phase: + raise SandboxCodecError(f"sandbox response phase mismatch (expected {phase!r})") + if _req_str(header.get("name"), "response name") != name: + raise SandboxCodecError(f"sandbox response name mismatch (expected {name!r})") + + +# --- the lookup broker (mode=isolated, ADR 0147) ------------------------------ +# While a Handler runs at mode=isolated, the worker may write ``lookup`` frames before its answer: one +# per db_lookup/fhir_lookup call, each answered by one ``lookupok``/``lookupfail`` frame from the +# parent before the worker writes anything else. Each carries the dispatch's ``(id, phase, name)`` +# triple, so it is bound to the dispatch in flight (ADR 0147 AC-6), and a ``seq`` that starts at 1 per +# dispatch and rises by one, so a reply cannot be taken for another call's. + +#: The two lookup kinds a ``lookup`` frame may name. +LOOKUP_DB: Final = "db" +LOOKUP_FHIR: Final = "fhir" + + +@dataclass(frozen=True) +class LookupRequest: + """One decoded ``lookup`` frame (parent side). ``target`` is the statement of a db lookup or the + query of a fhir one; ``params`` is rebuilt as the in-process call would have received it.""" + + seq: int + kind: str + connection: str + target: str + params: Any + + +def _enc_fhir_params(params: object, blobs: _Blobs) -> Any: + """A ``fhir_lookup`` ``params`` mapping in its own small grammar: ``[key, is_list, items]`` per + key, each item ``{"s": str}``, ``{"k": [system, code]}``, ``{"r": raw}``, or ``{"x": type name}`` + for a value of any other kind, which arrives as a + :class:`~messagefoundry.fhirsearch.ForeignSearchValue`. Values are not screened here: the + parent's executor judges each one, after its own "unavailable here" and connection checks. A + key that is not a ``str``, and ``params`` that is not a mapping, are still refused here, in the + worker, before those checks.""" + if params is None: + return None + if not isinstance(params, Mapping): + raise SandboxCodecError( + f"fhir_lookup params must be a mapping, not {type(params).__name__}" + ) + out: list[Any] = [] + for key, value in params.items(): + if not isinstance(key, str): + raise SandboxCodecError("a fhir_lookup params key must be a str") + is_list = isinstance(value, list) + items: list[Any] = [] + for one in value if is_list else [value]: + if isinstance(one, str): + items.append({"s": blobs.ref(one)}) + elif isinstance(one, FhirToken): + items.append({"k": [blobs.ref(one.system), blobs.ref(one.code)]}) + elif isinstance(one, FhirRaw): + items.append({"r": blobs.ref(one.value)}) + else: + # Any other kind crosses by its type name alone, never its value, so the engine's + # executor refuses it with the words it uses at mode=off, after its own checks. + items.append({"x": blobs.ref(type(one).__name__)}) + out.append([key, is_list, items]) + return out + + +def _dec_fhir_params(node: Any, reader: _Reader) -> dict[str, Any] | None: + if node is None: + return None + out: dict[str, Any] = {} + for entry in _req_list(node, "fhir_lookup params"): + if len(_req_list(entry, "fhir_lookup param")) != 3: + raise SandboxCodecError("a fhir_lookup param must be [key, is_list, items]") + key = _req_str(entry[0], "fhir_lookup param key") + if key in out: + raise SandboxCodecError(f"fhir_lookup param {key!r} appears twice") + is_list = _req_bool(entry[1], "fhir_lookup param is_list") + items: list[Any] = [] + for item in _req_list(entry[2], "fhir_lookup param items"): + obj = _req_obj(item, "fhir_lookup param item") + if len(obj) != 1: + raise SandboxCodecError("a fhir_lookup param item must carry exactly one tag") + ((tag, arg),) = obj.items() + if tag == "s": + items.append(reader.text(arg)) + elif tag == "k": + pair = _req_list(arg, "FhirToken") + if len(pair) != 2: + raise SandboxCodecError("a FhirToken must be [system, code]") + items.append(FhirToken(reader.text(pair[0]), reader.text(pair[1]))) + elif tag == "r": + items.append(FhirRaw(reader.text(arg))) + elif tag == "x": + items.append(ForeignSearchValue(reader.text(arg))) + else: + raise SandboxCodecError(f"unknown fhir_lookup param tag {tag!r}") + if not is_list and len(items) != 1: + raise SandboxCodecError("a single fhir_lookup param must carry exactly one item") + out[key] = items if is_list else items[0] + return out + + +@dataclass(frozen=True) +class PreparedLookup: + """A ``lookup`` frame's arguments, described but not yet framed. Describing them runs the + Handler's own code (a ``Mapping`` it passed is read), so the worker does it before it takes its + broker lock; :func:`encode_prepared_lookup` then frames it with no user code at all.""" + + kind: str + slots: dict[str, Any] + blobs: tuple[str, ...] + + +def prepare_lookup( + *, kind: str, connection: object, target: object, params: object +) -> PreparedLookup: + """Describe one lookup's arguments. ``target`` is the statement of a db lookup or the query of + a fhir one. Raises :class:`SandboxCodecError` for an argument outside the grammar, which the + worker turns into the lookup's own error type, as the in-process call would fail on a bad + argument.""" + blobs = _Blobs() + # Segment order IS walk order (see _Reader): connection, then target, then params, the order + # decode_answer reads them in. Building params first would number its segments ahead of the + # other two, and any lookup with a long params value and a long statement would not decode. + conn_ref = blobs.ref(connection) + target_ref = blobs.ref(target) + if kind == LOOKUP_DB: + if params is not None and not isinstance(params, Mapping): + raise SandboxCodecError( + f"db_lookup params must be a mapping or None, not {type(params).__name__}" + ) + enc_params = None if params is None else enc_value(dict(params), blobs) + elif kind == LOOKUP_FHIR: + enc_params = _enc_fhir_params(params, blobs) + else: + raise SandboxCodecError(f"unknown lookup kind {kind!r}") + return PreparedLookup( + kind=kind, + slots={"connection": conn_ref, "target": target_ref, "params": enc_params}, + blobs=tuple(blobs.items), + ) + + +def encode_prepared_lookup( + prepared: PreparedLookup, *, request_id: str, phase: str, name: str, seq: int +) -> bytes: + """Frame a :class:`PreparedLookup` for the dispatch in flight (child side).""" + header = _envelope( + { + "id": request_id, + "phase": phase, + "name": name, + "seq": seq, + "kind": prepared.kind, + **prepared.slots, + }, + "lookup", + ) + return encode_frame(header, prepared.blobs) + + +def encode_lookup( + *, + request_id: str, + phase: str, + name: str, + seq: int, + kind: str, + connection: object, + target: object, + params: object, +) -> bytes: + """:func:`prepare_lookup` and :func:`encode_prepared_lookup` in one call.""" + prepared = prepare_lookup(kind=kind, connection=connection, target=target, params=params) + return encode_prepared_lookup(prepared, request_id=request_id, phase=phase, name=name, seq=seq) + + +def decode_answer( + body: bytes, *, request_id: str, phase: str, name: str +) -> Response | LookupRequest: + """:func:`decode_response`, plus the ``lookup`` frame a worker may write mid-dispatch at + ``mode=isolated``. Only an isolated session calls this; every other one calls + :func:`decode_response`, which refuses a ``lookup`` frame as an unexpected frame type, so no other + mode can be talked into brokering a lookup.""" + header, reader, kind = _open(body, ("ok", "fail", "lookup")) + if kind != "lookup": + return _answer(header, reader, kind, request_id=request_id, phase=phase, name=name) + _check_binding(header, request_id, phase, name) + seq = _req_int(header.get("seq"), "lookup seq") + lookup_kind = header.get("kind") + if lookup_kind not in (LOOKUP_DB, LOOKUP_FHIR): + raise SandboxCodecError(f"unknown lookup kind {lookup_kind!r}") + # Walk order matches encode_lookup: connection, target, then params. + connection = reader.text(header.get("connection")) + target = reader.text(header.get("target")) + params: Any + if lookup_kind == LOOKUP_DB: + node = header.get("params") + params = None if node is None else dec_value(node, reader) + if params is not None and not isinstance(params, dict): + raise SandboxCodecError("db_lookup params must decode to a mapping") + else: + params = _dec_fhir_params(header.get("params"), reader) + reader.finish() + return LookupRequest( + seq=seq, kind=lookup_kind, connection=connection, target=target, params=params + ) + + +def encode_lookup_ok(*, request_id: str, phase: str, name: str, seq: int, result: object) -> bytes: + """The parent's answer to one ``lookup`` frame: the rows of a db lookup or the dict of a fhir one, + in the closed value grammar. Raises :class:`SandboxCodecError` for a value outside it.""" + blobs = _Blobs() + header = _envelope( + { + "id": request_id, + "phase": phase, + "name": name, + "seq": seq, + "result": enc_value(result, blobs), + }, + "lookupok", + ) + return encode_frame(header, blobs.items) + + +def encode_lookup_fail(*, request_id: str, phase: str, name: str, seq: int, error: str) -> bytes: + """The parent's refusal of one ``lookup`` frame. The worker raises it in the Handler as the + lookup's own error type, so a Handler that catches the error sees what ``mode=off`` shows it.""" + blobs = _Blobs() + header = _envelope( + {"id": request_id, "phase": phase, "name": name, "seq": seq, "error": blobs.ref(error)}, + "lookupfail", + ) + return encode_frame(header, blobs.items) + + +def decode_lookup_reply( + body: bytes, *, request_id: str, phase: str, name: str, seq: int +) -> tuple[bool, object, str]: + """``(ok, result, error)`` for one ``lookupok``/``lookupfail`` frame (child side), bound to the + dispatch and to the ``seq`` of the lookup it answers.""" + header, reader, kind = _open(body, ("lookupok", "lookupfail")) + _check_binding(header, request_id, phase, name) + if _req_int(header.get("seq"), "lookup reply seq") != seq: + raise SandboxCodecError(f"sandbox lookup reply is out of sequence (expected {seq})") + if kind == "lookupfail": + error = reader.text(header.get("error")) + reader.finish() + return False, None, error + if "result" not in header: + raise SandboxCodecError("sandbox 'lookupok' frame carries no result") + result = dec_value(header["result"], reader) + reader.finish() + return True, result, "" diff --git a/messagefoundry/pipeline/_sandbox_worker.py b/messagefoundry/pipeline/_sandbox_worker.py index 54a16a2e2..1a09915d0 100644 --- a/messagefoundry/pipeline/_sandbox_worker.py +++ b/messagefoundry/pipeline/_sandbox_worker.py @@ -9,7 +9,7 @@ decode path cannot name a type, import a module, or reach ``__reduce__``. Nothing is pickled in either direction, in either process. -1. **Bootstrap** — reads one ``boot`` frame ``{snapshot, forbidden, mem_mb, code_sets}``, loads +1. **Bootstrap** — reads one ``boot`` frame ``{snapshot, forbidden, mem_mb, code_sets, broker}``, loads the message :class:`~messagefoundry.config.wiring.Registry` from the engine's config snapshot (``load_snapshot``, the loader the engine uses, over the bytes the engine loaded and checked; the worker never reads the config directory, vault BACKLOG #2596), @@ -22,7 +22,9 @@ re-establishes the run-scoped context providers for the phase (substituting the boot-carried code sets, which the per-dispatch frame deliberately does not repeat), runs the function, and **describes** the result back as ``ok`` — or ``fail`` with ``kind`` ``denied`` (forbidden import, a - live ``db_lookup``/``fhir_lookup``, an undescribable result), ``error`` (a plain handler raise) or + live ``db_lookup``/``fhir_lookup`` at ``mode=subprocess``), ``error`` (a plain handler raise, an + undescribable result, or at ``mode=isolated`` a lookup error, the lookup having been asked of the + parent through :class:`_Broker` while the Handler ran) or ``memory`` (a ``MemoryError``, after which the worker exits so the parent starts a fresh one). The reply echoes the request's ``id``, ``phase`` and ``name`` so the parent can prove the frame answers the call it made. @@ -43,6 +45,7 @@ import logging import os import sys +import threading from dataclasses import replace from typing import Any @@ -205,8 +208,136 @@ def _install_import_guard(forbidden: tuple[str, ...]) -> None: sys.meta_path.insert(0, _ForbiddenImportFinder(forbidden)) -def _run_one(registry: Any, req: Any, code_sets: Any) -> tuple[bool, object, str, str]: - """Execute one router/handler request; return ``(ok, result, kind, error)``. Never raises.""" +class _Broker: + """The worker's half of the ``mode=isolated`` lookup broker (ADR 0147). + + :meth:`runners` hands out a dispatch's two runners, which stand in for the engine's while its + Handler runs: each writes one ``lookup`` frame to the parent and blocks on the parent's one reply, + so a ``db_lookup``/``fhir_lookup`` call returns what the parent's own runner returned, or raises the + lookup's own error type with the parent's reason. The parent decides every call; this side only + asks. + + Calls are serialized under one lock, because the pipe carries one exchange at a time. A runner + serves only the dispatch it was handed out for. A call from anywhere else, such as a thread a + Handler left running that holds an earlier dispatch's runner, is refused rather than written: + between dispatches the parent would read the frame as unsolicited and drop the worker, and during + another dispatch the lookup would be charged to that message. A desynchronized or closed pipe marks + the broker broken, and :func:`main` then exits without answering, so the message whose exchange + broke is the one that fails and the parent starts a fresh worker for the next.""" + + def __init__(self, stdin: Any, stdout: Any) -> None: + self._stdin = stdin + self._stdout = stdout + self._lock = threading.Lock() + self._req: Any = None + self._seq = 0 + self.broken = False + + def begin(self, req: Any) -> None: + with self._lock: + self._req, self._seq = req, 0 + + def end(self) -> None: + """Close the dispatch. Taken under the lock, so a lookup still in flight on another thread + finishes, inside the dispatch, before the answer frame is written.""" + with self._lock: + self._req = None + + def runners(self, req: Any) -> tuple[Any, Any]: + """The ``(db_lookup, fhir_lookup)`` runners for dispatch ``req`` alone.""" + from messagefoundry.config.db_lookup import DbLookupError + from messagefoundry.config.fhir_lookup import FhirLookupError + from messagefoundry.pipeline._sandbox_codec import LOOKUP_DB, LOOKUP_FHIR + + def db(connection: str, statement: str, params: Any = None) -> Any: + return self._ask( + req, LOOKUP_DB, DbLookupError, "db_lookup", connection, statement, params + ) + + def fhir(connection: str, query: str, params: Any = None) -> Any: + # The engine's executor screens each search value, as at mode=off, and after its own + # "unavailable here" check, so that is the error a call with no runner gets. + return self._ask( + req, LOOKUP_FHIR, FhirLookupError, "fhir_lookup", connection, query, params + ) + + return db, fhir + + def _ask( + self, + req: Any, + kind: str, + error: type[Exception], + fn: str, + connection: object, + target: object, + params: object, + ) -> Any: + from messagefoundry.pipeline import _sandbox_codec as codec + from messagefoundry.pipeline._sandbox_codec import MAX_FRAME, SandboxError + from messagefoundry.pipeline.sandbox import _read_frame_bytes, _write_frame + + try: + # Before the lock: describing the arguments reads the Handler's own Mapping, whose code + # could call a lookup itself, and the lock is not re-entrant. + prepared = codec.prepare_lookup( + kind=kind, connection=connection, target=target, params=params + ) + except SandboxError as exc: + # A bad argument fails as the call would in-process: the lookup's own error. Nothing is + # written, so the pipe stays in step. + raise error(f"{fn}: its arguments cannot cross the sandbox: {exc}") from None + with self._lock: + if self.broken: + raise error(f"{fn}: an earlier lookup in this dispatch broke the sandbox pipe") + if self._req is not req: + raise error(f"{fn}: called outside the Handler dispatch it was published for") + try: + frame = codec.encode_prepared_lookup( + prepared, + request_id=req.request_id, + phase=req.phase, + name=req.name, + seq=self._seq + 1, + ) + except SandboxError as exc: + # A frame the codec refuses (over a header or segment cap) fails as a bad argument + # does, as the lookup's own error, as at mode=off: the Handler can catch it, and + # nothing was written, so the pipe stays in step. + raise error(f"{fn}: its arguments cannot cross the sandbox: {exc}") from None + if len(frame) > MAX_FRAME: + raise error( + f"{fn}: its arguments are too large for the sandbox ({len(frame)} bytes)" + ) + self._seq += 1 + try: + _write_frame(self._stdout, frame) + body = _read_frame_bytes(self._stdin) + if body is None: + raise SandboxError("the engine closed the pipe") + ok, result, reason = codec.decode_lookup_reply( + body, request_id=req.request_id, phase=req.phase, name=req.name, seq=self._seq + ) + except (OSError, SandboxError) as exc: + # A pipe the parent abandoned, or a reply that does not answer this call: the pipe no + # longer lines up with the protocol. + self.broken = True + raise error(f"{fn}: the sandbox lookup broker failed: {exc}") from None + if not ok: + raise error(reason) + return result + + +def _run_one( + registry: Any, req: Any, code_sets: Any, broker: _Broker | None = None +) -> tuple[bool, object, str, str]: + """Execute one router/handler request; return ``(ok, result, kind, error)``. Never raises. + + ``broker`` is set at ``mode=isolated`` (ADR 0147): a Handler's ``db_lookup``/``fhir_lookup`` + is then asked of the parent, and an error it raises is that Handler's error, as at ``mode=off``. + Without it, both lookups are refused in the worker, as ADR 0087 built them.""" + from messagefoundry.config import db_lookup as db_lookup_mod + from messagefoundry.config import fhir_lookup as fhir_lookup_mod from messagefoundry.config.db_lookup import DbLookupError from messagefoundry.config.fhir_lookup import FhirLookupError from messagefoundry.config.run_context import run_contexts @@ -243,8 +374,17 @@ def _run_one(registry: Any, req: Any, code_sets: Any) -> tuple[bool, object, str if fn is None: return False, None, "error", f"no such {phase} {name!r} in registry" + # The brokered runners exist only for a Handler: a Router or an accepts= predicate finds no + # runner, so its lookup raises the "unavailable here" error it raises at mode=off. + db_runner, fhir_runner = ( + broker.runners(req) if broker is not None and phase == "transform" else (None, None) + ) try: - with run_contexts(run_context, phase=phase_key): + with ( + run_contexts(run_context, phase=phase_key), + db_lookup_mod.activated(db_runner), + fhir_lookup_mod.activated(fhir_runner), + ): result = fn(req.payload) if phase == "transform": # Materialise a CONTAINER return here, INSIDE the run context (BACKLOG #341). A @@ -269,8 +409,13 @@ def _run_one(registry: Any, req: Any, code_sets: Any) -> tuple[bool, object, str # ``[sandbox].mode`` never change the routing decision (ADR 0087). result = bool(result) except (DbLookupError, FhirLookupError) as exc: + if broker is not None: + # mode=isolated: the lookup ran, or was refused, through the parent, so its error is the + # Handler's error exactly as at mode=off. A Router's lookup lands here too, with the + # "unavailable here" text mode=off gives it. + return False, None, "error", _error_text(exc) # db_lookup/fhir_lookup bridge back onto the engine event loop (run_coroutine_threadsafe), - # which a subprocess boundary breaks — forbidden + fail-closed for this PR (ADR 0087). + # which a subprocess boundary breaks — forbidden + fail-closed at mode=subprocess (ADR 0087). return ( False, None, @@ -303,13 +448,13 @@ def _error_text(exc: BaseException) -> str: return type(exc).__name__ -def _respond(registry: Any, req: Any, code_sets: Any) -> bytes: +def _respond(registry: Any, req: Any, code_sets: Any, broker: _Broker | None = None) -> bytes: """Build the response frame for one request. Never raises — an undescribable result is reported — except ``MemoryError``, which :func:`main` answers.""" from messagefoundry.pipeline import _sandbox_codec as codec from messagefoundry.pipeline._sandbox_codec import SandboxError - ok, result, kind, error = _run_one(registry, req, code_sets) + ok, result, kind, error = _run_one(registry, req, code_sets, broker) if ok: try: return codec.encode_ok( @@ -377,6 +522,8 @@ def main() -> int: except (OSError, SandboxError): return 1 + # mode=isolated (ADR 0147): a Handler's lookups are asked of the parent over this same pipe. + broker = _Broker(stdin, stdout) if boot.broker else None while True: frame = _read_frame_bytes(stdin) if frame is None: @@ -389,14 +536,23 @@ def main() -> int: # dead-letter) and respawns a clean child, rather than hanging to the wall cap. log.error("sandbox worker: undecodable request frame (%s)", exc) return 1 + if broker is not None: + broker.begin(req) try: - resp: bytes | None = _respond(registry, req, boot.code_sets) + resp: bytes | None = _respond(registry, req, boot.code_sets, broker) except MemoryError as exc: # Caught here rather than around the Handler alone, so a result too big to encode is # answered the same way. Only the text is kept: the reply is built AFTER this block, # because until it ends the traceback keeps the Handler's frames, and the memory their # locals hold, alive. resp, detail = None, _error_text(exc) + if broker is not None: + broker.end() # before the answer: no lookup frame may follow it + if broker.broken: + # A lookup exchange went out of step. Exit WITHOUT answering, so the parent reads EOF + # and fails THIS message, rather than answer and leave the next message to find a + # worker on its way out. + return 1 if resp is None: # A call that ran out of memory may leave memory held (a module global, a cache, a # fragmented heap), and the next innocent message would pay for it. Answer, then exit so diff --git a/messagefoundry/pipeline/dryrun.py b/messagefoundry/pipeline/dryrun.py index 73e9ba0dd..8020cd146 100644 --- a/messagefoundry/pipeline/dryrun.py +++ b/messagefoundry/pipeline/dryrun.py @@ -340,7 +340,7 @@ def _accepted( if pred is None: kept.append(hname) continue - if sandbox is not None and sandbox.mode is SandboxMode.SUBPROCESS: + if sandbox is not None and sandbox.mode is not SandboxMode.OFF: # ADR 0087 parity: the predicate is user code and must run under the SAME isolation as the # Router/Handler it sits beside. Evaluating it here in the parent would let a predicate # reach the imports and resources the sandbox exists to deny. @@ -431,7 +431,7 @@ def route_only( # its "not even an allocation" promise rather than paying for a seam it never crosses. origin: tuple[str | bytes, str] | None = None raw_result: list[str] | str | None - if sandbox is not None and sandbox.mode is SandboxMode.SUBPROCESS: + if sandbox is not None and sandbox.mode is not SandboxMode.OFF: if derived: origin = (raw, ic.content_type.value) raw_result = cast( @@ -525,7 +525,7 @@ def transform_one( if payload is None: payload = _payload(raw, content_type) raw_result: HandlerResult - if sandbox is not None and sandbox.mode is SandboxMode.SUBPROCESS: + if sandbox is not None and sandbox.mode is not SandboxMode.OFF: raw_result = cast( "HandlerResult", run_sandboxed( diff --git a/messagefoundry/pipeline/engine.py b/messagefoundry/pipeline/engine.py index ab52e60b5..e3bfebb8e 100644 --- a/messagefoundry/pipeline/engine.py +++ b/messagefoundry/pipeline/engine.py @@ -1480,20 +1480,14 @@ def add_registry(self, registry: Registry) -> RegistryRunner: """Run a code-first Connection/Router/Handler graph (one runner for the whole graph).""" # ADR 0087 (#197): render [sandbox] into the runner's policy + the (config_dir, env) source a # subprocess worker uses to re-load the graph. mode=off (the default) makes _sandbox_for a - # no-op — byte-identical, no child ever spawned; mode=subprocess spawns one persistent worker - # child per inbound that dispatches. ONE policy for the WHOLE graph: [sandbox] is engine-wide + # no-op — byte-identical, no child ever spawned; mode=subprocess or mode=isolated spawns one + # persistent worker child per inbound that dispatches. ONE policy for the WHOLE graph: [sandbox] is engine-wide # and no connection carries a per-connection sandbox field, so the mode chosen here governs # every Router and Handler in this process. - from messagefoundry.pipeline.sandbox import SandboxMode, SandboxPolicy - - _sb = self._sandbox_settings - sandbox_policy = SandboxPolicy( - mode=SandboxMode(_sb.mode), - wall_seconds=_sb.wall_seconds, - mem_mb=_sb.mem_mb, - startup_seconds=_sb.startup_seconds, - pass_environment=_sb.pass_environment, - ) + from messagefoundry.pipeline.sandbox import policy_from_settings + + # mode=isolated also reads [egress]: its lookup broker refuses a lookup whose list is empty. + sandbox_policy = policy_from_settings(self._sandbox_settings, self._egress_settings) sandbox_config_source = ( str(self.config_dir) if self.config_dir is not None else None, self._active_environment, diff --git a/messagefoundry/pipeline/sandbox.py b/messagefoundry/pipeline/sandbox.py index aba52222a..523225be8 100644 --- a/messagefoundry/pipeline/sandbox.py +++ b/messagefoundry/pipeline/sandbox.py @@ -32,6 +32,13 @@ child's own re-read of ``codesets/`` is never authoritative), runs the function, and describes its result back. The worker is *long-lived* (one child per inbound), never a per-message fork — a fork per message would destroy the throughput target. +* ``isolated`` (ADR 0147, vault BACKLOG #3131) — the ``subprocess`` worker, with a Handler's + ``db_lookup``/``fhir_lookup`` brokered through the parent instead of refused: the worker writes a + ``lookup`` frame mid-dispatch and the parent answers it with the runner the in-process path uses + (:meth:`SandboxSession._serve_lookup`), refusing what :meth:`SandboxSession._lookup_refusal` + lists. **No OS-level confinement yet:** ADR 0147's Landlock/seccomp and AppContainer limbs are + unbuilt, so this mode confines exactly what ``subprocess`` does. Every check that means "runs in a + worker" reads ``is not SandboxMode.OFF``, so this mode takes every worker branch. **Both legs of the pipe are untrusted by contract.** The child runs exactly the code the sandbox exists to distrust: while it lives, a grandchild it spawns inherits fd 1 (the response pipe) and can @@ -61,8 +68,8 @@ On top of that address-space boundary the child adds defence-in-depth: a forbidden-import guard (``socket``/store/crypto), a wall-clock cap enforced by the parent (plus a POSIX memory cap when available), and a fail-closed refusal of the live -``db_lookup``/``fhir_lookup`` bridges (they re-enter the event loop, which a process boundary -breaks — forwarding them over IPC is a documented next-phase residual). The import guard is +``db_lookup``/``fhir_lookup`` bridges at ``mode=subprocess`` (they re-enter the event loop, which a +process boundary breaks; ``mode=isolated`` forwards them over the pipe instead). The import guard is **defence-in-depth only** — a module imported before it goes up keeps a live reference, so the address-space boundary and the codec are the load-bearing controls. @@ -85,6 +92,7 @@ from __future__ import annotations +import contextvars import enum import logging import queue @@ -102,15 +110,22 @@ from messagefoundry import proctree from messagefoundry.childenv import python_child_argv, worker_environment from messagefoundry.config.code_sets import CodeSet +from messagefoundry.config.db_lookup import DbLookupError, db_lookup +from messagefoundry.config.db_lookup import is_available as db_lookup_available +from messagefoundry.config.fhir_lookup import FhirLookupError, fhir_lookup +from messagefoundry.config.fhir_lookup import is_available as fhir_lookup_available from messagefoundry.config.run_context import RunContext from messagefoundry.config.settings import ( INSECURE_CONFIG_SOURCE_ESCAPE_ENV, + EgressSettings, + SandboxSettings, security_enforcement_env_names, ) from messagefoundry.config.snapshot import ConfigSnapshot from messagefoundry.controlchars import scrub_control_chars from messagefoundry.pipeline import _sandbox_codec as codec from messagefoundry.pipeline._sandbox_codec import GraphShape, SandboxCodecError, SandboxError +from messagefoundry.redaction import safe_exc __all__ = [ "SandboxMode", @@ -124,6 +139,7 @@ "graph_shape", "graph_differences", "run_sandboxed", + "policy_from_settings", "DEFAULT_FORBIDDEN_MODULES", "WORKER_MODULE", ] @@ -161,6 +177,10 @@ class SandboxMode(str, enum.Enum): # noqa: UP042 # docstring's mode bullets; [sandbox].mode is engine-wide, so that escape is too). OFF = "off" # in-process, byte-identical, zero overhead (default) SUBPROCESS = "subprocess" # persistent per-inbound worker child + # The SUBPROCESS worker, with db_lookup/fhir_lookup brokered through the parent (ADR 0147). Every + # check that means "runs in a worker" is `is not SandboxMode.OFF`, so this mode can never take an + # in-process branch that only SUBPROCESS was excluded from (vault BACKLOG #3131). + ISOLATED = "isolated" @dataclass(frozen=True) @@ -175,8 +195,8 @@ class SandboxPolicy: ``mode=off`` (the shipped default) is the zero-overhead, byte-identical parity mode. ``wall_seconds`` is the **authoritative** cap on every platform — the parent kills a worker that overruns it (so a - pathological busy-loop Router/Handler can never wedge intake), and it is enforced **only** at - ``mode=subprocess``: the in-process path has no timeout at all. There is no ``RLIMIT_CPU``, so + pathological busy-loop Router/Handler can never wedge intake), and it is enforced **only** in + a worker mode (``subprocess``, ``isolated``): the in-process path has no timeout at all. There is no ``RLIMIT_CPU``, so CPU that admin code spends outside a call has no bound (ADR 0087, amendment of 2026-10-07). ``mem_mb`` adds a POSIX memory backstop *inside* the child where the ``resource`` module exists (a no-op on Windows). On Linux it is headroom above the loaded graph: ``RLIMIT_DATA`` at the @@ -193,6 +213,30 @@ class SandboxPolicy: #: ``[sandbox].pass_environment``: the extra variable NAMES the worker is given. Settings load #: refuses the engine's own; this dataclass carries what it let through. pass_environment: tuple[str, ...] = () + #: ``[egress].allowed_db`` and ``[egress].allowed_http``, read only by the parent's lookup broker + #: at ``mode=isolated`` (ADR 0147), which refuses a lookup while its list is empty + #: (:meth:`SandboxSession._lookup_refusal`). The hosts themselves are checked where the in-process + #: path checks them, when the lookup executors are built. Empty by default, so a policy built + #: without them brokers nothing. Never sent to the worker. + lookup_allowed_db: tuple[str, ...] = () + lookup_allowed_http: tuple[str, ...] = () + + +def policy_from_settings(settings: SandboxSettings, egress: EgressSettings) -> SandboxPolicy: + """The :class:`SandboxPolicy` the engine renders from ``[sandbox]`` and ``[egress]``. + + ``SandboxMode(settings.mode)`` raises ``ValueError`` on a mode it does not know, so a value + that reached this point without the settings model's own check still refuses to start rather + than being read as some other mode.""" + return SandboxPolicy( + mode=SandboxMode(settings.mode), + wall_seconds=settings.wall_seconds, + mem_mb=settings.mem_mb, + startup_seconds=settings.startup_seconds, + pass_environment=settings.pass_environment, + lookup_allowed_db=tuple(egress.allowed_db), + lookup_allowed_http=tuple(egress.allowed_http), + ) def graph_shape(registry: Any) -> GraphShape: @@ -211,6 +255,23 @@ def graph_shape(registry: Any) -> GraphShape: #: How a refusal names a difference in the config modules' bytes alone. _CONTENT_DIFFERS: Final = "config module content" +#: The accessor each lookup kind names, for the broker's refusals and log lines. +_LOOKUP_FN: Final = {codec.LOOKUP_DB: "db_lookup", codec.LOOKUP_FHIR: "fhir_lookup"} + +#: Longest error text the broker sends back to a worker. A lookup error can quote the connection +#: name the worker sent, and the worker chooses its length. +_LOOKUP_ERROR_CAP: Final = 2048 + + +def _shown_name(name: str) -> str: + """A connection name a worker sent, scrubbed and cut, for a refusal or a log line.""" + return scrub_control_chars(name)[:80] + + +def _bounded(text: str) -> str: + return text if len(text) <= _LOOKUP_ERROR_CAP else text[:_LOOKUP_ERROR_CAP] + " [cut]" + + #: How many names a refusal lists per side of one part of the graph. _NAMES_SHOWN: Final = 3 @@ -218,7 +279,7 @@ def graph_shape(registry: Any) -> GraphShape: def _shown(names: set[str] | frozenset[str]) -> str: """Up to :data:`_NAMES_SHOWN` names for an error text. Half of them come from the worker, so each is scrubbed and cut before it reaches a log line.""" - listed = [scrub_control_chars(name)[:80] for name in sorted(names)[:_NAMES_SHOWN]] + listed = [_shown_name(name) for name in sorted(names)[:_NAMES_SHOWN]] more = len(names) - len(listed) return ", ".join(listed) + (f" and {more} more" if more > 0 else "") @@ -455,11 +516,23 @@ def _write_frame(stream: Any, body: bytes) -> None: pipe; the caller maps either to :class:`SandboxError`.""" if len(body) > _MAX_FRAME: raise SandboxError(f"sandbox frame too large: {len(body)} bytes") - stream.write(_LEN.pack(len(body))) - stream.write(body) + _write_all(stream, _LEN.pack(len(body))) + _write_all(stream, body) stream.flush() +def _write_all(stream: Any, data: bytes) -> None: + """Write every byte of ``data``. The parent's end of the worker pipe is unbuffered + (``bufsize=0``), and a raw ``write`` may return a short count, as a large one interrupted by a + signal can; dropping the remainder would hand the worker a torn frame.""" + view = memoryview(data) + while view: + written = stream.write(view) + if not written: # None is a non-blocking raw stream with no room; the pipes here block + raise OSError("sandbox pipe accepted no bytes") + view = view[written:] + + def _read_exact(stream: Any, n: int) -> bytes | None: """Read exactly ``n`` bytes, or ``None`` on EOF (a closed/crashed peer).""" chunks: list[bytes] = [] @@ -649,6 +722,12 @@ def __init__( self._generation = 0 self._lock = threading.Lock() self._closed = False + # The [egress] lists whose empty-list refusal this session has logged. + self._lookup_refusals_logged: set[str] = set() + # Per (kind, connection), the completion event of the runner call running on it, removed by + # that call when it returns: see _serve_lookup. + self._running_lookups: dict[tuple[str, str], threading.Event] = {} + self._running_lock = threading.Lock() @property def mode(self) -> SandboxMode: @@ -677,6 +756,7 @@ def _spawn(self) -> None: forbidden=self.policy.forbidden_modules, mem_mb=self.policy.mem_mb, code_sets=self._code_sets, + broker=self.policy.mode is SandboxMode.ISOLATED, ) except (OSError, ValueError, SandboxError) as exc: # WiringError is a ValueError raise SandboxError(f"failed to bootstrap sandbox worker: {exc}") from exc @@ -952,6 +1032,298 @@ def _live_worker(self) -> subprocess.Popen[bytes]: assert proc is not None return proc + # -- the lookup broker (mode=isolated, ADR 0147) ---------------------------- + + def _await_answer( + self, proc: subprocess.Popen[bytes], request_id: str, phase: str, name: str + ) -> codec.Response: + """The dispatch's ``ok``/``fail`` frame, after serving every ``lookup`` frame the worker + writes before it at ``mode=isolated``. Any other mode decodes with + :func:`codec.decode_response`, which refuses a ``lookup`` frame, so only an isolated session + ever brokers a lookup and no worker can ask another session to. + + ``wall_seconds`` bounds the whole dispatch, the lookups the parent serves for it included + (:meth:`_serve_bounded`), so the cap stays the bound it is at ``mode=subprocess`` and a feed + whose lookups are slow needs a larger cap.""" + brokered = self.policy.mode is SandboxMode.ISOLATED + deadline = time.monotonic() + self.policy.wall_seconds + expected_seq = 1 + while True: + try: + frame = self._responses.get(timeout=max(0.0, deadline - time.monotonic())) + except queue.Empty: + # Wall cap exceeded — the authoritative resource bound on every platform. Kill the + # runaway child (a busy-loop can't wedge intake) and fail closed. + self._kill(proc) + raise SandboxError( + f"sandbox {phase} {name!r} exceeded the {self.policy.wall_seconds}s wall cap" + ) from None + if frame is _EOF: + self._kill(proc) + served = expected_seq - 1 + after = f", after {served} brokered lookup(s)" if served else "" + raise SandboxError(f"sandbox worker crashed while running {phase} {name!r}{after}") + try: + # Decoding happens HERE, on the dispatch thread, inside this try — not on the daemon + # reader thread, where a rejection would be a silent reader death and a wall-cap hang. + answer = ( + codec.decode_answer(frame, request_id=request_id, phase=phase, name=name) + if brokered + else codec.decode_response(frame, request_id=request_id, phase=phase, name=name) + ) + if isinstance(answer, codec.LookupRequest) and answer.seq != expected_seq: + raise SandboxCodecError( + f"sandbox lookup is out of sequence (expected {expected_seq})" + ) + except SandboxError as exc: + # A desynchronized or forged frame: the worker's stream can no longer be trusted to + # line up with our requests, so drop it and respawn on the next message. + self._kill(proc) + raise SandboxError( + f"sandbox {phase} {name!r} returned an invalid frame: {exc}" + ) from exc + if isinstance(answer, codec.Response): + return answer + expected_seq += 1 + if not self._serve_bounded(proc, request_id, phase, name, answer, deadline): + self._kill(proc) + raise SandboxError( + f"sandbox {phase} {name!r} exceeded the {self.policy.wall_seconds}s wall cap, " + "which counts the lookups the engine served for it" + ) + + def _serve_bounded( + self, + proc: subprocess.Popen[bytes], + request_id: str, + phase: str, + name: str, + lookup: codec.LookupRequest, + deadline: float, + ) -> bool: + """Serve one lookup and write its reply, within the dispatch's wall cap; ``False`` if the + cap ran out first, and the caller then kills the worker. + + Both halves run on a helper thread, because either can block past the cap while this thread + holds the session lock, and a reload's ``close`` waits on that lock. The lookup can wait on + its runner for the runner's own result timeout, longer than the default cap. The reply can + be larger than the pipe's buffer, and a worker that sent a ``lookup`` frame and then stopped + reading stdin would hold the write for good. The thread runs in a copy of this thread's + context, so it sees the lookup runner the transform worker published. Killing the worker + normally ends a blocked write. A broken pipe raises :class:`SandboxError` after the worker is + killed. + + **Each lookup records itself per connection while its runner runs** (:meth:`_serve_lookup` + registers it, and says what a later lookup on that connection then does). The record is the + lookup half's completion, not the thread's: a reply write that never ends cannot keep the + connection marked busy. + + **An unexpected fault in the engine fails the dispatch at once,** with a fixed cause and the + worker killed, rather than leave the worker waiting for a reply until the cap. The helper logs + it by exception type only, never its text, which could carry message content, and it does so + even when the cap has already cut the helper off.""" + assert proc.stdin is not None + stdin = proc.stdin + failure: list[BaseException] = [] + fault: list[BaseException] = [] + # Set by this thread the moment it gives up on the helper. The helper checks it, with the + # deadline, before it registers and calls the runner, and checks it alone before it writes. + abandoned = threading.Event() + + def report(exc: BaseException) -> None: + fault.append(exc) + log.error( + "inbound %r: the engine failed while serving a brokered lookup for %s %r (%s)", + self._inbound, + phase, + name, + type(exc).__name__, + ) + + def serve() -> None: + try: + reply = self._lookup_reply(request_id, phase, name, lookup, deadline, abandoned) + except Exception as exc: # noqa: BLE001 — logged by type, failed by the dispatch thread + report(exc) + return + if abandoned.is_set(): + return # the worker is being killed; nothing reads this reply + try: + _write_frame(stdin, reply) + except (OSError, ValueError, SandboxError) as exc: # ValueError: a pipe closed under it + failure.append(exc) + except Exception as exc: # noqa: BLE001 — logged by type, failed by the dispatch thread + report(exc) + + context = contextvars.copy_context() + helper = threading.Thread( + target=context.run, + args=(serve,), + name=f"mf-sandbox-lookup-{self._inbound}", + daemon=True, + ) + helper.start() + helper.join(max(0.0, deadline - time.monotonic())) + if helper.is_alive(): + abandoned.set() + return False + if fault: + self._kill(proc) + raise SandboxError( + f"sandbox {phase} {name!r} failed while serving its worker's lookup: an " + "unexpected engine error, logged by type" + ) + if failure: + self._kill(proc) + raise SandboxError(f"could not answer the sandbox worker's lookup: {failure[0]}") + return True + + def _lookup_reply( + self, + request_id: str, + phase: str, + name: str, + lookup: codec.LookupRequest, + deadline: float, + abandoned: threading.Event, + ) -> bytes: + """The ``lookupok``/``lookupfail`` frame answering one ``lookup``. A result that cannot cross + is a ``lookupfail`` the worker can report.""" + ok, result, error = self._serve_lookup(phase, lookup, deadline, abandoned) + fn = _LOOKUP_FN[lookup.kind] + if ok: + try: + body = codec.encode_lookup_ok( + request_id=request_id, phase=phase, name=name, seq=lookup.seq, result=result + ) + except SandboxError as exc: + # At least a column type outside the closed grammar, such as a UUID. + error = f"{fn} result cannot cross the sandbox: {exc}" + else: + if len(body) <= _MAX_FRAME: + return body + error = f"{fn} result is larger than a sandbox frame ({len(body)} bytes)" + return codec.encode_lookup_fail( + request_id=request_id, phase=phase, name=name, seq=lookup.seq, error=error + ) + + def _serve_lookup( + self, + phase: str, + lookup: codec.LookupRequest, + deadline: float, + abandoned: threading.Event, + ) -> tuple[bool, object, str]: + """Answer one brokered lookup in the parent: ``(ok, result, error)``. + + The answer comes from the same ``db_lookup``/``fhir_lookup`` accessor a Handler calls at + ``mode=off``, in this dispatch's context, so it resolves the runner the transform worker + published for this transform and nothing else (ADR 0147 AC-2). Where none is published, as on + a Router, in a dry run, on the inline fast path or in a graph that declares no lookup, the call + raises exactly what it raises at ``mode=off``, and it is made before any refusal of the + broker's own, so that is the error the Handler sees. The executor behind the runner was built + against ``[egress]`` and runs its own statement test and row ceiling, so each applies here as + in-process. The slice-B seam: a broker-only check of a lookup's own arguments goes between the + refusals below and the call. + + **One runner call at a time per connection.** Under ``_running_lock``, a lookup registers its + own completion event for its connection, then calls the runner, and removes the record when + the runner returns. A later lookup on that connection can only meet one the wall cap cut off, + since a session serves one dispatch at a time. It waits on that record by the rule + ``docs/CONFIGURATION.md`` states (section ``[sandbox]``, *``mode="isolated"``*), and is + refused if it still runs then. Every runner call is recorded, so a connection never has two in + one session (a reload replaces the session; ``docs/CONFIGURATION.md`` states the limits). + **A helper does not call the runner once its dispatch is over:** it checks ``abandoned`` and + the deadline just before it registers. That is a check, not a fence. A helper that passed it + an instant before the dispatch gave up still calls the runner, and that call is recorded and + holds the connection like any other.""" + is_db = lookup.kind == codec.LOOKUP_DB + available = db_lookup_available() if is_db else fhir_lookup_available() + refusal = self._lookup_refusal(phase, lookup) if available else None + if refusal is not None: + return False, None, refusal + key = (lookup.kind, lookup.connection) + done = threading.Event() + if available: + margin = min(2.0, self.policy.wall_seconds / 4) + while True: + with self._running_lock: + if abandoned.is_set() or time.monotonic() >= deadline: + # The dispatch is over: its worker is being killed and nobody waits. + return ( + False, + None, + f"{_LOOKUP_FN[lookup.kind]}: the call's wall cap ran out", + ) + current = self._running_lookups.get(key) + if current is None: + self._running_lookups[key] = done + break + if not current.wait(max(0.0, deadline - margin - time.monotonic())): + return False, None, self._busy_refusal(lookup) + try: + if is_db: + return True, db_lookup(lookup.connection, lookup.target, lookup.params), "" + return True, fhir_lookup(lookup.connection, lookup.target, lookup.params), "" + except (DbLookupError, FhirLookupError) as exc: + # PHI-free by both types' contract: a connection name and a routing-safe reason. + return False, None, _bounded(str(exc)) + except Exception as exc: # noqa: BLE001 — the worker waits on this reply; report, never raise + return False, None, f"{_LOOKUP_FN[lookup.kind]} failed: {safe_exc(exc)}" + finally: + if available: + with self._running_lock: + if self._running_lookups.get(key) is done: + del self._running_lookups[key] + done.set() + + @staticmethod + def _busy_refusal(lookup: codec.LookupRequest) -> str: + fn = _LOOKUP_FN[lookup.kind] + return ( + # Short, scope first: a dead-letter keeps about the first 200 characters of this. + f"{fn}({_shown_name(lookup.connection)!r}) refused: this inbound's sandbox session " + "runs one lookup per connection, and one the wall cap cut off is still running " + "(ADR 0147)" + ) + + def _lookup_refusal(self, phase: str, lookup: codec.LookupRequest) -> str | None: + """Why the parent refuses a brokered lookup outright, or ``None`` to serve it. + + A lookup is answered only for a Handler, so one bound to a Router or an ``accepts=`` dispatch + is refused even if a worker sends it; the child has no lookup runner in those phases either. + And one whose ``[egress]`` list is empty is refused, under the ruling ``docs/CONFIGURATION.md`` + states in section ``[sandbox]``, *``mode="isolated"``* (vault BACKLOG #3131). The first such + refusal per session and list logs a WARNING.""" + fn = _LOOKUP_FN[lookup.kind] + shown = _shown_name(lookup.connection) + if phase != "transform": + return ( + f"{fn}({shown!r}) is unavailable here — the engine answers a brokered lookup only " + "for a Handler, never for a Router or an accepts= predicate (ADR 0147)" + ) + if lookup.kind == codec.LOOKUP_DB: + allowed, key = self.policy.lookup_allowed_db, "allowed_db" + else: + allowed, key = self.policy.lookup_allowed_http, "allowed_http" + if allowed: + return None + if key not in self._lookup_refusals_logged: + # Keyed by the list, whatever connection a worker names, so it holds at most two entries. + self._lookup_refusals_logged.add(key) + log.warning( + "inbound %r: a brokered %s on %r was refused: [egress].%s is empty " + "([sandbox].mode=isolated, ADR 0147)", + self._inbound, + fn, + shown, + key, + ) + return ( + f"{fn}({shown!r}) refused: [egress].{key} is empty, and [sandbox].mode=isolated " + "brokers a lookup only to a host that list names (ADR 0147)" + ) + # -- dispatch ------------------------------------------------------------- def dispatch( @@ -1001,29 +1373,7 @@ def dispatch( # reset the worker. self._kill(proc) raise SandboxError(f"failed to marshal sandbox {phase} {name!r}: {exc}") from exc - try: - frame = self._responses.get(timeout=self.policy.wall_seconds) - except queue.Empty: - # Wall cap exceeded — the authoritative resource bound on every platform. Kill the - # runaway child (a busy-loop can't wedge intake) and fail closed. - self._kill(proc) - raise SandboxError( - f"sandbox {phase} {name!r} exceeded the {self.policy.wall_seconds}s wall cap" - ) from None - if frame is _EOF: - self._kill(proc) - raise SandboxError(f"sandbox worker crashed while running {phase} {name!r}") - try: - # Decoding happens HERE, on the dispatch thread, inside this try — not on the daemon - # reader thread, where a rejection would be a silent reader death and a wall-cap hang. - resp = codec.decode_response(frame, request_id=request_id, phase=phase, name=name) - except SandboxError as exc: - # A desynchronized or forged frame: the worker's stream can no longer be trusted to - # line up with our requests, so drop it and respawn on the next message. - self._kill(proc) - raise SandboxError( - f"sandbox {phase} {name!r} returned an invalid frame: {exc}" - ) from exc + resp = self._await_answer(proc, request_id, phase, name) # One request, one frame: a second FRAME already queued means the worker is writing frames # nobody asked for, so this answer is not trustworthy either. (A queued EOF is only a dead # peer — it reaps the worker and leaves this proven answer standing.) @@ -1058,9 +1408,9 @@ def run_sandboxed( With ``session is None`` or ``session.mode is OFF`` this is exactly ``fn(payload)`` — in-process, byte-identical, zero overhead (the parity default); nothing above that branch touches ``payload``, - ``origin``, or the codec. With ``session.mode is SUBPROCESS`` the call is marshalled to the - persistent worker via :meth:`SandboxSession.dispatch`, enforcing the forbidden-import / resource - caps and raising :class:`SandboxError` on any violation. + ``origin``, or the codec. In every other mode (``subprocess``, ``isolated``) the call is + marshalled to the persistent worker via :meth:`SandboxSession.dispatch`, enforcing the + forbidden-import / resource caps and raising :class:`SandboxError` on any violation. ``origin`` is the ``(raw, content_type)`` the caller built ``payload`` from — pass it **only** when the caller derived the payload itself, so the child can rebuild it byte-faithfully; ``None`` makes diff --git a/messagefoundry/pipeline/wiring_runner.py b/messagefoundry/pipeline/wiring_runner.py index 62049d5af..f163788b6 100644 --- a/messagefoundry/pipeline/wiring_runner.py +++ b/messagefoundry/pipeline/wiring_runner.py @@ -186,6 +186,7 @@ from messagefoundry.pipeline.reply_wait import ReplyRendezvous from messagefoundry.pipeline.sandbox import ( GraphShape, + SandboxError, SandboxMode, SandboxPolicy, SandboxSession, @@ -1341,6 +1342,18 @@ def __init__( # construction (a /config/reload does NOT re-read it — restart to change, like claim_mode). self._sandbox_policy = sandbox_policy self._sandbox_config_source = sandbox_config_source + if ( + sandbox_policy is not None + and sandbox_policy.mode is SandboxMode.ISOLATED + and registry.source_dir is None + and (sandbox_config_source is None or sandbox_config_source[0] is None) + ): + # Refused here, before a listener ACKs anything, rather than once per message by + # _sandbox_for, whose refusal stays as the backstop for a reload (vault BACKLOG #3131). + raise ValueError( + "[sandbox].mode=isolated needs a config directory for its workers to load, and " + "this graph has none: it never runs a Router or Handler in the engine process" + ) self._sandbox_sessions: dict[str, SandboxSession] = {} self._sandbox_graph_shape: tuple[Registry, GraphShape] | None = None # Sessions read that cache from their worker threads. Two building it at once would each @@ -5648,7 +5661,7 @@ def _engine_graph_shape(self) -> GraphShape: async def _close_sandbox_sessions(self) -> None: """Drop every sandbox session and close its worker, off the loop: each ``close()`` waits on - a process. A no-op unless ``[sandbox].mode=subprocess`` spawned any. The next dispatch on an + a process. A no-op unless a worker sandbox mode spawned any. The next dispatch on an inbound makes a fresh session, whose worker loads the config as it is then. A dispatch already holding a closed session is retried or re-pended by :meth:`_off_loop_sandboxed`, never dead-lettered as a Router or Handler fault.""" @@ -5666,8 +5679,10 @@ def close_all() -> None: def _sandbox_for(self, name: str) -> SandboxSession | None: """The persistent sandbox worker for inbound ``name`` (ADR 0087), or ``None`` to run in-process. - Returns ``None`` — the byte-identical in-process path — unless ``[sandbox].mode=subprocess`` - AND a config dir is known. The child re-loads the graph to look a function up by name, from + Returns ``None`` — the byte-identical in-process path — unless ``[sandbox].mode`` is + ``subprocess`` or ``isolated`` AND a config dir is known. With no config dir, ``isolated`` + raises :class:`SandboxError` rather than degrade, because the in-process path would run its + lookups unbrokered (vault BACKLOG #3131); the rest of this paragraph is about ``subprocess``. The child re-loads the graph to look a function up by name, from ``Registry.config_snapshot``, the bytes the served graph was loaded from, and never reads the directory (vault BACKLOG #2596). A graph with no snapshot hands the session ``Registry.source_dir``, else the constructor's ``sandbox_config_source``, which the session @@ -5688,6 +5703,15 @@ def _sandbox_for(self, name: str) -> SandboxSession | None: self._sandbox_config_source[0] if self._sandbox_config_source else None ) if cfg_dir is None: + if policy.mode is SandboxMode.ISOLATED: + # FAIL CLOSED. The in-process path would run this inbound's Handlers with the + # engine's own db_lookup/fhir_lookup runners, unbrokered, under a setting that + # asked for the broker (vault BACKLOG #3131). The message dead-letters instead. + raise SandboxError( + f"[sandbox].mode=isolated cannot run inbound {name!r}: its graph has no " + "config directory for a worker to load, and this mode never runs a Router " + "or Handler in the engine process" + ) return None env = self._sandbox_config_source[1] if self._sandbox_config_source else None # The engine's code-set tables travel once per spawn in the boot frame (not per dispatch), @@ -6670,11 +6694,14 @@ async def _activate_fusion(self) -> bool: # Rather than run user config code unsandboxed under a config that asked for a sandbox, fail # CLOSED to the async (sandboxed) path. An IPC round-trip inside a fused hop would negate the # fusion anyway, so degrading is the right trade — never a silent isolation bypass. - if self._sandbox_policy is not None and self._sandbox_policy.mode is SandboxMode.SUBPROCESS: + # `is not OFF`, never `is SUBPROCESS`: every worker mode is isolation this hop would bypass + # (mode=isolated included, vault BACKLOG #3131). + if self._sandbox_policy is not None and self._sandbox_policy.mode is not SandboxMode.OFF: log.warning( - "fuse_thread_hops is set but [sandbox].mode=subprocess: fusion runs Router/Handler/" + "fuse_thread_hops is set but [sandbox].mode=%s: fusion runs Router/Handler/" "accepts= code in-process (unsandboxed), so it is DISABLED — running the async sandboxed " - "pipeline path. Turn the sandbox off to use fusion, or leave fusion off to keep isolation." + "pipeline path. Turn the sandbox off to use fusion, or leave fusion off to keep isolation.", + self._sandbox_policy.mode.value, ) return False backend = self.store.backend @@ -10323,10 +10350,11 @@ async def _prepare_routed( lookup_stack.enter_context(db_lookup_activated(self._run_lookup)) if self._fhir_lookup_executor is not None: lookup_stack.enter_context(fhir_lookup_activated(self._run_fhir_lookup)) - # ADR 0087 (#197): sandbox=subprocess marshals the Handler to the per-inbound - # worker (transform_rc travels with it); a db_lookup/fhir_lookup inside a - # sandboxed Handler fails closed there (they can't bridge across the process - # boundary in this PR). sandbox=None is the byte-identical in-process path. + # ADR 0087 (#197): a worker sandbox mode marshals the Handler to the per-inbound + # worker (transform_rc travels with it). A db_lookup/fhir_lookup inside it fails + # closed at mode=subprocess, and at mode=isolated is brokered back to this + # thread, which is why the runners above are published either way (ADR 0147). + # sandbox=None is the byte-identical in-process path. outcome = await self._off_loop_sandboxed( name, transform_one, diff --git a/tests/test_checks.py b/tests/test_checks.py index 7ab8a9b5c..69ae493ff 100644 --- a/tests/test_checks.py +++ b/tests/test_checks.py @@ -653,7 +653,7 @@ def test_no_expect_keeps_default_semantics(tmp_path: Path) -> None: @pytest.mark.parametrize("bridge", sorted(_LOOKUP_CALLS)) -@pytest.mark.parametrize("sandbox_mode", ["off", "subprocess"]) +@pytest.mark.parametrize("sandbox_mode", ["off", "subprocess", "isolated"]) def test_a_lookup_handler_fails_the_gate_in_every_sandbox_mode( tmp_path: Path, bridge: str, sandbox_mode: str ) -> None: diff --git a/tests/test_from_none_is_not_redaction.py b/tests/test_from_none_is_not_redaction.py index 1fe60d0ac..8d648b623 100644 --- a/tests/test_from_none_is_not_redaction.py +++ b/tests/test_from_none_is_not_redaction.py @@ -221,9 +221,21 @@ class _Allowed: _SAFE + "queue.Empty carries nothing", ), _Allowed( - "messagefoundry/pipeline/sandbox.py::SandboxSession.dispatch::Empty::SandboxError", + "messagefoundry/pipeline/sandbox.py::SandboxSession._await_answer::Empty::SandboxError", _SAFE + "queue.Empty carries nothing", ), + _Allowed( + "messagefoundry/pipeline/_sandbox_worker.py::_Broker._ask::SandboxError::error", + _SAFE + "the lookup error quotes the codec refusal, which names a slot, a type or a size " + "and never a value; the cause is dropped so the Handler sees the mode=off error type alone. " + "Two sites: describing the arguments, and framing them", + count=2, + ), + _Allowed( + "messagefoundry/pipeline/_sandbox_worker.py::_Broker._ask::(OSError, SandboxError)::error", + _SAFE + "the lookup error quotes the pipe or frame refusal, which names a frame and never " + "its content (decode_frame keeps only safe_exc text)", + ), _Allowed( "messagefoundry/store/store.py::parse_audit_anchor::ValueError::AuditAnchorError", _SAFE + "int() quotes the anchor's count text, and the new error carries the whole anchor", diff --git a/tests/test_sandbox_isolated.py b/tests/test_sandbox_isolated.py new file mode 100644 index 000000000..90a1aaf42 --- /dev/null +++ b/tests/test_sandbox_isolated.py @@ -0,0 +1,1140 @@ +# SPDX-License-Identifier: AGPL-3.0-or-later +# Copyright (C) 2026 MessageFoundry Foundation, LLC and contributors +"""``[sandbox].mode = "isolated"`` (ADR 0147, vault BACKLOG #3131). + +Two properties, each pinned per pipeline phase and per mode: + +1. **Where the code runs.** Every check that means "runs in a sandbox worker" reads ``is not OFF``, + so the isolated mode takes the worker branch of the Router, the ``accepts=`` predicate and the + Handler, exactly as ``subprocess`` does. Before the conversion those checks read ``is + SUBPROCESS``, and an isolated session ran each phase in the engine process. +2. **The brokered lookups.** In the isolated mode a Handler's ``db_lookup``/``fhir_lookup`` is + asked of the parent over the closed-tag codec, and the parent answers it with the runner the + in-process path uses. The parent refuses a request outside the Handler phase, and refuses one + whose ``[egress]`` list is empty, whatever ``[security].block_unlisted_outbound`` says (owner + ruling of 2026-10-27, made under delegation). ``subprocess`` still refuses both lookups in the + worker, and ``off`` runs them in-process. + +Synthetic HL7 only.""" + +from __future__ import annotations + +import asyncio +import os +import threading +import time +from decimal import Decimal +from pathlib import Path +from typing import Any + +import pytest + +from messagefoundry.config import db_lookup as db_lookup_mod +from messagefoundry.config import fhir_lookup as fhir_lookup_mod +from messagefoundry.config.run_context import RunContext +from messagefoundry.config.settings import EgressSettings, SandboxSettings +from messagefoundry.config.wiring import Registry, load_config +from messagefoundry.fhirsearch import FhirRaw, FhirToken +from messagefoundry.pipeline import _sandbox_codec as codec +from messagefoundry.pipeline.dryrun import route_only, transform_one +from messagefoundry.pipeline.sandbox import ( + SandboxError, + SandboxMode, + SandboxPolicy, + SandboxSession, + policy_from_settings, +) +from messagefoundry.store import MessageStatus + +RAW = "MSH|^~\\&|SEND|F|RECV|F|20240101120000||ADT^A01|MSG00001|P|2.3\rPID|1||900001||DOE^JANE\r" + +_GRAPH = """ +import os +from collections.abc import Mapping + +from messagefoundry import ( + inbound, outbound, router, handler, MLLP, Send, db_lookup, fhir_lookup, FhirToken, FhirRaw, +) +from messagefoundry.config.db_lookup import DbLookupError + +PARENT = __PARENT__ + +inbound("IB_I", MLLP(port=19411), router="r_where") +inbound("IB_RL", MLLP(port=19413), router="r_lookup") +outbound("OB_I", MLLP(host="127.0.0.1", port=19412)) + + +def _where(): + return "parent" if os.getpid() == PARENT else "child" + + +@router("r_where") +def r_where(msg): + # Names the handler for the process it ran in, so the routing decision says where it ran. + return "h_" + _where() + + +@router("r_lookup") +def r_lookup(msg): + db_lookup("LK", "SELECT 1 AS v", {}) # unavailable on a Router, in every mode + return "h_child" + + +def _in_child(msg): + return _where() == "child" + + +@handler("h_child", accepts=_in_child) +def h_child(msg): + return Send("OB_I", _where()) + + +@handler("h_parent") +def h_parent(msg): + return Send("OB_I", _where()) + + +@handler("h_db") +def h_db(msg): + rows = db_lookup("LK", "SELECT v, d FROM t WHERE k = :k", {"k": msg.field("PID-3"), "n": 7}) + return Send("OB_I", f"{rows[0]['v']}|{type(rows[0]['d']).__name__}|{rows[0]['d']}") + + +@handler("h_db_caught") +def h_db_caught(msg): + try: + db_lookup("LK", "SELECT 1 AS v", {}) + except DbLookupError as exc: + return Send("OB_I", "caught:" + str(exc)) + return Send("OB_I", "not refused") + + +class _LazyParams(Mapping): + # A Mapping whose values come from another lookup, read while the outer call describes it. + def __iter__(self): + return iter(["k"]) + + def __len__(self): + return 1 + + def __getitem__(self, key): + return db_lookup("LK", "SELECT inner", {})[0]["v"] + + +@handler("h_db_nested") +def h_db_nested(msg): + rows = db_lookup("LK", "SELECT outer", _LazyParams()) + return Send("OB_I", rows[0]["v"]) + + +@handler("h_fhir_badkind") +def h_fhir_badkind(msg): + from messagefoundry.config.fhir_lookup import FhirLookupError + + try: + fhir_lookup("FH", "Patient", {"_count": 10, "x": ["ok", 2.5]}) + except FhirLookupError as exc: + return Send("OB_I", str(exc)) + return Send("OB_I", "not refused") + + +@handler("h_fhir") +def h_fhir(msg): + bundle = fhir_lookup( + "FH", "Patient", {"identifier": FhirToken("MRN", "900001"), "_sort": [FhirRaw("-date"), "x"]} + ) + return Send("OB_I", bundle["id"]) +""" + + +@pytest.fixture +def graph(tmp_path: Path) -> tuple[Registry, str]: + (tmp_path / "graph.py").write_text( + _GRAPH.replace("__PARENT__", str(os.getpid())), encoding="utf-8" + ) + return load_config(tmp_path), str(tmp_path) + + +def _session( + config_dir: str, + mode: SandboxMode, + *, + allowed_db: tuple[str, ...] = ("db.example",), + allowed_http: tuple[str, ...] = ("fhir.example",), +) -> SandboxSession: + return SandboxSession( + SandboxPolicy( + mode=mode, + wall_seconds=30.0, + lookup_allowed_db=allowed_db, + lookup_allowed_http=allowed_http, + ), + inbound="IB_I", + config_dir=config_dir, + env=None, + graph=None, + ) + + +def _bodies(registry: Registry, hname: str, session: SandboxSession | None) -> list[str]: + deliveries, _, _, _ = transform_one( + registry, hname, RAW, sandbox=session, run_context=RunContext() + ) + return [str(d.payload) for d in deliveries] + + +# --- 1. where each phase runs, per mode ---------------------------------------------------------- + + +@pytest.mark.parametrize( + ("mode", "where"), + [ + (SandboxMode.OFF, "parent"), + (SandboxMode.SUBPROCESS, "child"), + (SandboxMode.ISOLATED, "child"), + ], +) +def test_router_accepts_and_handler_run_where_the_mode_says( + graph: tuple[Registry, str], mode: SandboxMode, where: str +) -> None: + registry, config_dir = graph + session = _session(config_dir, mode) + try: + # The Router names the handler for the process it ran in. h_child's accepts= predicate + # declines anywhere but a worker, so a Router run in a worker beside a predicate run in the + # engine routes nowhere: each of the two phases is pinned on its own. + names = route_only( + registry, registry.inbound["IB_I"], RAW, sandbox=session, run_context=RunContext() + ) + assert names == ["h_" + where] + # The Handler reports the process it ran in. + assert _bodies(registry, "h_parent", session) == [where] + finally: + session.close() + + +# --- 2. the brokered lookups, per mode ----------------------------------------------------------- + + +class _Runners: + """Stand-ins for the engine's lookup runners, published the way the transform worker publishes + the real ones: on the thread that runs ``transform_one``, which is where the broker answers.""" + + def __init__(self) -> None: + self.db_calls: list[tuple[str, str, Any]] = [] + self.fhir_calls: list[tuple[str, str, Any]] = [] + + def db(self, connection: str, statement: str, params: Any) -> list[dict[str, Any]]: + self.db_calls.append((connection, statement, params)) + return [{"v": "VAL", "d": Decimal("12.50")}] + + def fhir(self, connection: str, query: str, params: Any) -> dict[str, Any]: + self.fhir_calls.append((connection, query, params)) + return {"resourceType": "Bundle", "id": "BUNDLE-1", "total": 1} + + +@pytest.fixture +def runners() -> Any: + r = _Runners() + with db_lookup_mod.activated(r.db), fhir_lookup_mod.activated(r.fhir): + yield r + + +def test_isolated_brokers_db_lookup_through_the_parent( + graph: tuple[Registry, str], runners: _Runners +) -> None: + registry, config_dir = graph + session = _session(config_dir, SandboxMode.ISOLATED) + try: + bodies = _bodies(registry, "h_db", session) + finally: + session.close() + # The row the parent's runner returned reached the Handler, a Decimal still a Decimal. + assert bodies == ["VAL|Decimal|12.50"] + assert runners.db_calls == [("LK", "SELECT v, d FROM t WHERE k = :k", {"k": "900001", "n": 7})] + + +def test_isolated_brokers_fhir_lookup_with_its_search_kinds( + graph: tuple[Registry, str], runners: _Runners +) -> None: + registry, config_dir = graph + session = _session(config_dir, SandboxMode.ISOLATED) + try: + bodies = _bodies(registry, "h_fhir", session) + finally: + session.close() + assert bodies == ["BUNDLE-1"] + # Each value arrives as the kind the Handler wrote, so the parent's executor screens it as it + # would in-process: a FhirToken's code half, never a FhirRaw. + assert runners.fhir_calls == [ + ( + "FH", + "Patient", + {"identifier": FhirToken("MRN", "900001"), "_sort": [FhirRaw("-date"), "x"]}, + ) + ] + + +def test_mode_off_runs_the_lookup_in_process( + graph: tuple[Registry, str], runners: _Runners +) -> None: + registry, _ = graph + assert _bodies(registry, "h_db", None) == ["VAL|Decimal|12.50"] + assert len(runners.db_calls) == 1 + + +def test_subprocess_still_refuses_both_lookups_in_the_worker( + graph: tuple[Registry, str], runners: _Runners +) -> None: + registry, config_dir = graph + session = _session(config_dir, SandboxMode.SUBPROCESS) + try: + for hname in ("h_db", "h_fhir"): + with pytest.raises(SandboxError, match="db_lookup/fhir_lookup is forbidden"): + _bodies(registry, hname, session) + finally: + session.close() + assert runners.db_calls == [] and runners.fhir_calls == [] + + +@pytest.mark.parametrize( + ("hname", "empty", "key"), + [ + ("h_db", {"allowed_db": ()}, "allowed_db"), + ("h_fhir", {"allowed_http": ()}, "allowed_http"), + ], +) +def test_isolated_refuses_a_lookup_whose_egress_list_is_empty( + graph: tuple[Registry, str], + runners: _Runners, + caplog: pytest.LogCaptureFixture, + hname: str, + empty: dict[str, tuple[str, ...]], + key: str, +) -> None: + registry, config_dir = graph + session = _session(config_dir, SandboxMode.ISOLATED, **empty) + try: + with caplog.at_level("WARNING", logger="messagefoundry.pipeline.sandbox"): + for _ in range(2): + with pytest.raises(SandboxError, match=rf"\[egress\]\.{key} is empty"): + _bodies(registry, hname, session) + finally: + session.close() + # Refused in the parent, before any runner ran: the published runner saw nothing. + assert runners.db_calls == [] and runners.fhir_calls == [] + # Logged once for the session, not once per message. + refusals = [r for r in caplog.records if "was refused" in r.getMessage()] + assert len(refusals) == 1 and key in refusals[0].getMessage() + + +def test_a_refusal_reaches_the_handler_as_its_lookup_error( + graph: tuple[Registry, str], runners: _Runners +) -> None: + # A Handler that catches DbLookupError sees the parent's refusal as that error, as at mode=off. + registry, config_dir = graph + session = _session(config_dir, SandboxMode.ISOLATED, allowed_db=()) + try: + (body,) = _bodies(registry, "h_db_caught", session) + finally: + session.close() + assert body.startswith("caught:db_lookup('LK') refused: [egress].allowed_db is empty") + + +def test_a_lookup_with_no_runner_in_the_parent_fails_as_at_mode_off( + graph: tuple[Registry, str], +) -> None: + # No runner published: a dry run, the inline fast path, or a graph with no lookup connection. + registry, config_dir = graph + session = _session(config_dir, SandboxMode.ISOLATED) + try: + with pytest.raises(SandboxError, match="unavailable here"): + _bodies(registry, "h_db", session) + finally: + session.close() + + +@pytest.mark.parametrize("mode", [SandboxMode.SUBPROCESS, SandboxMode.ISOLATED]) +def test_a_router_lookup_is_refused_in_every_worker_mode( + graph: tuple[Registry, str], runners: _Runners, mode: SandboxMode +) -> None: + registry, config_dir = graph + session = _session(config_dir, mode) + try: + with pytest.raises(SandboxError): + route_only( + registry, registry.inbound["IB_RL"], RAW, sandbox=session, run_context=RunContext() + ) + finally: + session.close() + assert runners.db_calls == [] + + +def test_the_parent_refuses_a_lookup_bound_to_a_router_dispatch(tmp_path: Path) -> None: + # The worker has no runner on a Router, so it never asks; the parent refuses one anyway. + session = _session(str(tmp_path), SandboxMode.ISOLATED) + lookup = codec.LookupRequest( + seq=1, kind=codec.LOOKUP_DB, connection="LK", target="SELECT 1", params=None + ) + for phase in ("router", "accepts"): + refusal = session._lookup_refusal(phase, lookup) + assert refusal is not None and "only for a Handler" in refusal + assert session._lookup_refusal("transform", lookup) is None + + +def test_a_lookup_frame_drops_a_subprocess_worker() -> None: + # Only an isolated session decodes a lookup frame; any other refuses it as a protocol violation. + frame = codec.encode_lookup( + request_id="r1", + phase="transform", + name="h", + seq=1, + kind=codec.LOOKUP_DB, + connection="LK", + target="SELECT 1", + params=None, + ) + with pytest.raises(codec.SandboxCodecError, match="unexpected sandbox frame type 'lookup'"): + codec.decode_response(frame, request_id="r1", phase="transform", name="h") + answer = codec.decode_answer(frame, request_id="r1", phase="transform", name="h") + assert isinstance(answer, codec.LookupRequest) and answer.target == "SELECT 1" + # Bound to the dispatch in flight (ADR 0147 AC-6): another dispatch's id is refused. + with pytest.raises(codec.SandboxCodecError, match="correlation mismatch"): + codec.decode_answer(frame, request_id="r2", phase="transform", name="h") + + +def test_a_lookup_reply_is_bound_to_its_sequence() -> None: + reply = codec.encode_lookup_ok( + request_id="r1", phase="transform", name="h", seq=2, result=[{"a": Decimal("1")}] + ) + assert codec.decode_lookup_reply( + reply, request_id="r1", phase="transform", name="h", seq=2 + ) == (True, [{"a": Decimal("1")}], "") + with pytest.raises(codec.SandboxCodecError, match="out of sequence"): + codec.decode_lookup_reply(reply, request_id="r1", phase="transform", name="h", seq=1) + + +# --- 3. the mode setting fails closed -------------------------------------------------------------- + + +def test_policy_reads_the_egress_lists_for_the_broker() -> None: + policy = policy_from_settings( + SandboxSettings(mode="isolated"), + EgressSettings(deny_by_default=False, allowed_db=["db.example"], allowed_http=[]), + ) + assert policy.mode is SandboxMode.ISOLATED + assert policy.lookup_allowed_db == ("db.example",) and policy.lookup_allowed_http == () + + +def test_an_unknown_mode_refuses() -> None: + with pytest.raises(ValueError, match="mode"): + SandboxSettings(mode="isolate") # type: ignore[arg-type] + # Past the model's own check too: the engine's rendering refuses rather than guess a mode. + smuggled = SandboxSettings.model_construct(mode="bogus") + with pytest.raises(ValueError, match="bogus"): + policy_from_settings(smuggled, EgressSettings()) + + +# --- 4. the running engine: the split transform worker, per mode ---------------------------------- + +_RUNNER_GRAPH = """ +from messagefoundry import inbound, outbound, router, handler, File, Send, DatabaseLookup, db_lookup + +inbound("IB_L", File(directory={inbox!r}, poll_seconds=0.05), router="r") +inbound("IB_W", File(directory={inbox_w!r}, poll_seconds=0.05), router="r_w") +outbound("OB_L", File(directory={outbox!r})) +DatabaseLookup("LK", server="db.example", database="D") + + +@router("r") +def r(msg): + return "h" + + +@router("r_w") +def r_w(msg): + return "h_write" + + +@handler("h") +def h(msg): + rows = db_lookup("LK", "SELECT v FROM t WHERE k = :k", {{"k": msg.field("PID-3")}}) + return Send("OB_L", rows[0]["v"]) + + +@handler("h_write") +def h_write(msg): + db_lookup("LK", "INSERT INTO t (k) VALUES (:k)", {{"k": "1"}}) + return Send("OB_L", "wrote") +""" + + +def _runner_dirs(tmp_path: Path) -> dict[str, str]: + dirs = {key: tmp_path / key for key in ("inbox", "inbox_w", "outbox", "cfg")} + for path in dirs.values(): + path.mkdir() + return {key: str(path) for key, path in dirs.items()} + + +async def _run_one_message( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mode: str, + egress: EgressSettings, + *, + inbound_name: str = "IB_L", +) -> tuple[str, list[str], Any]: + """Start a runner on the graph, push one message through ``inbound_name``, and return its final + status, what was delivered, and the fake pool, whose cursor records any statement that ran.""" + from messagefoundry.pipeline.wiring_runner import RegistryRunner + from messagefoundry.store import MessageStore, Stage + from tests.test_db_lookup import _patch_pool + + pool = _patch_pool(monkeypatch, rows=[("FROM-DB",)], columns=["v"]) + dirs = _runner_dirs(tmp_path) + (Path(dirs["cfg"]) / "graph.py").write_text(_RUNNER_GRAPH.format(**dirs), encoding="utf-8") + egress = egress.model_copy(update={"allowed_file_dirs": [dirs["outbox"]]}) + store = await MessageStore.open(tmp_path / "isolated.db") + runner = RegistryRunner( + load_config(dirs["cfg"]), + store, + sandbox_policy=policy_from_settings( + SandboxSettings.model_validate({"mode": mode, "wall_seconds": 60.0}), egress + ), + sandbox_config_source=(dirs["cfg"], None), + egress=egress, + ) + await runner.start() + try: + mid = await store.enqueue_ingress( + channel_id=inbound_name, raw=RAW, control_id="MSG00001", message_type="ADT^A01" + ) + runner._wake_lane(Stage.INGRESS, inbound_name) + terminal = {MessageStatus.PROCESSED.value, MessageStatus.ERROR.value} + status = "" + for _ in range(1200): + msg = await store.get_message(mid) + assert msg is not None + status = str(msg["status"]) + if status in terminal: + break + await asyncio.sleep(0.05) + finally: + await runner.stop() + await store.close() + delivered = [p.read_text(encoding="utf-8") for p in Path(dirs["outbox"]).iterdir()] + return status, delivered, pool + + +_LISTED = EgressSettings(allowed_db=["db.example"]) +_PROCESSED = MessageStatus.PROCESSED.value +_ERROR = MessageStatus.ERROR.value + + +@pytest.mark.parametrize( + ("mode", "expect", "ran"), + [("off", _PROCESSED, True), ("subprocess", _ERROR, False), ("isolated", _PROCESSED, True)], +) +async def test_the_transform_worker_runs_a_lookup_per_mode( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, mode: str, expect: str, ran: bool +) -> None: + status, delivered, pool = await _run_one_message(tmp_path, monkeypatch, mode, _LISTED) + assert status == expect + assert delivered == (["FROM-DB"] if ran else []) + assert (pool.cursor_obj.executed is not None) is ran + + +async def test_isolated_refuses_an_empty_allowed_db_under_the_audited_opt_out( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # block_unlisted_outbound = false makes an empty allowed_db unrestricted: the executor builds and + # mode=off would dial. The broker refuses anyway (owner ruling 2026-10-27, under delegation). + opt_out = EgressSettings(deny_by_default=False) + status, delivered, pool = await _run_one_message(tmp_path, monkeypatch, "isolated", opt_out) + assert (status, delivered) == (_ERROR, []) + assert pool.cursor_obj.executed is None + + +async def test_a_brokered_write_is_refused_by_the_executors_read_only_test( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # The broker answers through the same executor as mode=off, so its statement test applies. + status, delivered, pool = await _run_one_message( + tmp_path, monkeypatch, "isolated", _LISTED, inbound_name="IB_W" + ) + assert (status, delivered) == (_ERROR, []) + assert pool.cursor_obj.executed is None + + +def test_isolated_with_no_config_dir_refuses_rather_than_run_in_process(tmp_path: Path) -> None: + from messagefoundry.pipeline.wiring_runner import RegistryRunner + + def runner_for(mode: SandboxMode) -> RegistryRunner: + return RegistryRunner( + Registry(), # a graph built in code: no source_dir, no snapshot + object(), # type: ignore[arg-type] + sandbox_policy=SandboxPolicy(mode=mode), + sandbox_config_source=(None, None), + egress=EgressSettings(deny_by_default=False), + ) + + # Refused at construction, before anything is received. + with pytest.raises(ValueError, match="never runs a Router or Handler in the engine"): + runner_for(SandboxMode.ISOLATED) + # mode=subprocess keeps its documented degradation to in-process for such a graph. + runner = runner_for(SandboxMode.SUBPROCESS) + assert runner._sandbox_for("IB_ANY") is None + # And the per-message backstop, for a policy that reaches the runner by another route. + runner._sandbox_policy = SandboxPolicy(mode=SandboxMode.ISOLATED) + with pytest.raises(SandboxError, match="never runs a Router or Handler in the engine"): + runner._sandbox_for("IB_ANY") + + +@pytest.mark.parametrize("mode", [SandboxMode.SUBPROCESS, SandboxMode.ISOLATED]) +async def test_fusion_is_disabled_in_every_worker_mode( + mode: SandboxMode, caplog: pytest.LogCaptureFixture, tmp_path: Path +) -> None: + from messagefoundry.pipeline.wiring_runner import RegistryRunner + + runner = RegistryRunner( + Registry(), + object(), # type: ignore[arg-type] + sandbox_policy=SandboxPolicy(mode=mode), + sandbox_config_source=(str(tmp_path), None), + egress=EgressSettings(deny_by_default=False), + ) + runner._fuse_thread_hops = True + with caplog.at_level("WARNING"): + assert await runner._activate_fusion() is False + assert any(f"[sandbox].mode={mode.value}" in r.getMessage() for r in caplog.records) + + +# --- 5. round-one review fixes --------------------------------------------------------------------- + + +@pytest.mark.parametrize("kind", [codec.LOOKUP_DB, codec.LOOKUP_FHIR]) +def test_a_lookup_with_long_arguments_round_trips(kind: str) -> None: + # Every slot past the inline threshold rides a segment, and segments must be numbered in the + # order the decoder reads them: connection, target, then params. + long = "x" * 5000 + params: Any = {"k": long} if kind == codec.LOOKUP_DB else {"identifier": FhirToken("S", long)} + frame = codec.encode_lookup( + request_id="r1", + phase="transform", + name="h", + seq=1, + kind=kind, + connection="C" * 5000, + target="SELECT " + long if kind == codec.LOOKUP_DB else "Patient" + long, + params=params, + ) + answer = codec.decode_answer(frame, request_id="r1", phase="transform", name="h") + assert isinstance(answer, codec.LookupRequest) + assert answer.connection == "C" * 5000 and answer.params == params + + +def _new_lookup_helpers(earlier: set[threading.Thread]) -> list[threading.Thread]: + """The broker's lookup helper threads started since ``earlier`` was taken.""" + return [ + thread + for thread in set(threading.enumerate()) - earlier + if thread.name.startswith("mf-sandbox-lookup-") + ] + + +def _worker_broker() -> Any: + """The worker's ``_Broker``. Importing the worker installs a stderr root handler, which is right + in its child process and wrong here, so the root logger is restored after the import, as the + ``sandbox_run_one`` fixture in tests/test_db_lookup.py does.""" + import logging + + root = logging.getLogger() + handlers, level = list(root.handlers), root.level + try: + from messagefoundry.pipeline import _sandbox_worker + finally: + for h in list(root.handlers): + root.removeHandler(h) + for h in handlers: + root.addHandler(h) + root.setLevel(level) + return _sandbox_worker._Broker + + +def test_a_runner_serves_only_the_dispatch_it_was_published_for() -> None: + import io + from types import SimpleNamespace + + from messagefoundry.config.db_lookup import DbLookupError + + broker_cls = _worker_broker() + out = io.BytesIO() + broker = broker_cls(io.BytesIO(), out) + first = SimpleNamespace(request_id="a", phase="transform", name="h") + second = SimpleNamespace(request_id="b", phase="transform", name="h") + broker.begin(first) + stale_db, _ = broker.runners(first) + broker.end() + broker.begin(second) + # A thread left over from the first message still holds its runner: refused, nothing written. + with pytest.raises(DbLookupError, match="outside the Handler dispatch"): + stale_db("LK", "SELECT 1", None) + assert out.getvalue() == b"" + assert not broker.broken + + +def test_an_oversized_lookup_fails_the_call_and_keeps_the_pipe_in_step( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import io + from types import SimpleNamespace + + from messagefoundry.config.db_lookup import DbLookupError + + monkeypatch.setattr(codec, "MAX_FRAME", 1024) + broker_cls = _worker_broker() + out = io.BytesIO() + broker = broker_cls(io.BytesIO(), out) + req = SimpleNamespace(request_id="a", phase="transform", name="h") + broker.begin(req) + db, _ = broker.runners(req) + with pytest.raises(DbLookupError, match="too large for the sandbox"): + db("LK", "SELECT 1", {"k": "x" * 4096}) + assert out.getvalue() == b"" and not broker.broken + + +def test_a_worker_that_stops_reading_cannot_hold_the_dispatch_past_the_cap( + graph: tuple[Registry, str], +) -> None: + # A Handler forges a lookup whose reply is far larger than the pipe buffer, then sleeps instead + # of reading it. Unbounded, the parent's write of the reply blocks with the session lock held, + # and the wall cap never fires. The cap must still end the dispatch. + def big(connection: str, statement: str, params: Any) -> list[dict[str, Any]]: + return [{"v": "x" * 2_000_000}] + + _, config_dir = graph + source = (Path(config_dir) / "graph.py").read_text(encoding="utf-8") + _FORGER + (Path(config_dir) / "graph.py").write_text(source, encoding="utf-8") + registry = load_config(config_dir) + session = SandboxSession( + SandboxPolicy( + mode=SandboxMode.ISOLATED, wall_seconds=3.0, lookup_allowed_db=("db.example",) + ), + inbound="IB_I", + config_dir=config_dir, + env=None, + graph=None, + ) + try: + started = time.monotonic() + with db_lookup_mod.activated(big), pytest.raises(SandboxError, match="wall cap"): + _bodies(registry, "h_forge_lookup", session) + assert time.monotonic() - started < 30 + finally: + session.close() + + +_FORGER = """ + +@handler("h_forge_lookup") +def h_forge_lookup(msg): + import os, sys, time + + from messagefoundry.pipeline import _sandbox_codec as codec + + # The dispatch's request object is a local of the worker's main loop. + frame = sys._getframe() + while frame is not None and "req" not in frame.f_locals: + frame = frame.f_back + req = frame.f_locals["req"] + body = codec.encode_lookup( + request_id=req.request_id, phase=req.phase, name=req.name, seq=1, kind="db", + connection="LK", target="SELECT 1", params=None, + ) + os.write(1, len(body).to_bytes(4, "big") + body) + time.sleep(60) # never reads the reply + return Send("OB_I", "unreachable") +""" + + +# --- 6. round-two review fixes --------------------------------------------------------------------- + + +def test_a_lookup_with_no_runner_is_not_told_to_edit_egress( + graph: tuple[Registry, str], caplog: pytest.LogCaptureFixture +) -> None: + # No runner and an empty list, as a site with no database has: the mode=off error, no WARNING. + registry, config_dir = graph + session = _session(config_dir, SandboxMode.ISOLATED, allowed_db=()) + try: + with ( + caplog.at_level("WARNING", logger="messagefoundry.pipeline.sandbox"), + pytest.raises(SandboxError, match="unavailable here") as raised, + ): + _bodies(registry, "h_db", session) + finally: + session.close() + assert "allowed_db" not in str(raised.value) + assert not [r for r in caplog.records if "was refused" in r.getMessage()] + + +def test_a_slow_lookup_is_cut_off_by_the_wall_cap(graph: tuple[Registry, str]) -> None: + def slow(connection: str, statement: str, params: Any) -> list[dict[str, Any]]: + time.sleep(20) + return [{"v": "late", "d": Decimal(0)}] + + registry, config_dir = graph + session = SandboxSession( + SandboxPolicy(mode=SandboxMode.ISOLATED, wall_seconds=2.0, lookup_allowed_db=("db",)), + inbound="IB_I", + config_dir=config_dir, + env=None, + graph=None, + ) + try: + # Warm the worker first, so the timing below measures the cap rather than a cold start. + _bodies(registry, "h_parent", session) + started = time.monotonic() + with db_lookup_mod.activated(slow), pytest.raises(SandboxError, match="wall cap"): + _bodies(registry, "h_db", session) + assert time.monotonic() - started < 15 + finally: + session.close() + + +def test_a_lookup_made_while_describing_another_does_not_deadlock( + graph: tuple[Registry, str], +) -> None: + calls: list[str] = [] + + def runner(connection: str, statement: str, params: Any) -> list[dict[str, Any]]: + calls.append(statement) + return [{"v": "INNER" if statement == "SELECT inner" else f"OUTER:{params['k']}"}] + + registry, config_dir = graph + session = _session(config_dir, SandboxMode.ISOLATED) + try: + with db_lookup_mod.activated(runner): + assert _bodies(registry, "h_db_nested", session) == ["OUTER:INNER"] + finally: + session.close() + assert calls == ["SELECT inner", "SELECT outer"] + + +def test_a_short_write_is_completed() -> None: + from messagefoundry.pipeline.sandbox import _write_frame + + class Trickle: + def __init__(self) -> None: + self.data = bytearray() + + def write(self, chunk: Any) -> int: + self.data.extend(bytes(chunk)[:3]) + return min(3, len(chunk)) + + def flush(self) -> None: + pass + + out = Trickle() + _write_frame(out, b"0123456789") + assert bytes(out.data) == (10).to_bytes(4, "big") + b"0123456789" + + +# --- 7. the Lander's covering review of 415353d7 (round-472dd562) ---------------------------------- + + +def test_a_lookup_the_cap_cut_off_is_the_only_one_outstanding( + graph: tuple[Registry, str], runners: _Runners +) -> None: + # A hung backend: every lookup blocks until the test releases it. Each dispatch hits the cap. + # Unbounded, each later dispatch starts another lookup beside the abandoned ones. + release = threading.Event() + lock = threading.Lock() + state = {"calls": 0, "in_flight": 0, "peak": 0} + + def hung(connection: str, statement: str, params: Any) -> list[dict[str, Any]]: + with lock: + state["calls"] += 1 + state["in_flight"] += 1 + state["peak"] = max(state["peak"], state["in_flight"]) + try: + release.wait(60) + return [{"v": "late", "d": Decimal(0)}] + finally: + with lock: + state["in_flight"] -= 1 + + registry, config_dir = graph + session = SandboxSession( + SandboxPolicy( + mode=SandboxMode.ISOLATED, + wall_seconds=4.0, + lookup_allowed_db=("db",), + lookup_allowed_http=("fhir",), + ), + inbound="IB_I", + config_dir=config_dir, + env=None, + graph=None, + ) + try: + with db_lookup_mod.activated(hung): + with pytest.raises(SandboxError, match="wall cap"): + _bodies(registry, "h_db", session) + # The next dispatch waits for the stalled lookup, then is refused its own rather than + # start a second one beside it. + started = time.monotonic() + with pytest.raises(SandboxError, match="still running"): + _bodies(registry, "h_db", session) + assert time.monotonic() - started < 15 + # Another connection is not held up by the stalled one. + with fhir_lookup_mod.activated(runners.fhir): + assert _bodies(registry, "h_fhir", session) == ["BUNDLE-1"] + assert (state["calls"], state["peak"]) == (1, 1) + # Once the abandoned lookup ends, the session serves lookups on it again: the next one waits + # for the stale lookup's completion rather than poll the runner's own counter. + release.set() + with db_lookup_mod.activated(hung): + assert _bodies(registry, "h_db", session)[0].startswith("late|") + finally: + release.set() + session.close() + + +def test_a_codec_refusal_while_framing_is_the_lookups_own_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import io + from types import SimpleNamespace + + from messagefoundry.config.db_lookup import DbLookupError + + # Framing refuses here: one long param is one segment, over a cap of none. + monkeypatch.setattr(codec, "_MAX_BLOBS", 0) + broker = _worker_broker()(io.BytesIO(), io.BytesIO()) + req = SimpleNamespace(request_id="a", phase="transform", name="h") + broker.begin(req) + db, _ = broker.runners(req) + with pytest.raises(DbLookupError, match="cannot cross the sandbox"): + db("LK", "SELECT 1", {"k": "x" * 5000}) + assert not broker.broken + + +def _executor_like(connection: str, query: str, params: Any) -> dict[str, Any]: + """The FHIR executor's screening, as mode=off meets it (transports/fhir.py ``read``).""" + from messagefoundry.config.fhir_lookup import FhirLookupError + from messagefoundry.fhirsearch import resolve_search_pairs + + try: + resolve_search_pairs(params or {}) + except ValueError as exc: + raise FhirLookupError(f"fhir_lookup on {connection!r}: {exc}") from exc + return {"resourceType": "Bundle", "id": "B"} + + +def test_a_wrong_kind_search_value_fails_as_at_mode_off(graph: tuple[Registry, str]) -> None: + registry, config_dir = graph + session = _session(config_dir, SandboxMode.ISOLATED) + try: + with fhir_lookup_mod.activated(_executor_like): + off = _bodies(registry, "h_fhir_badkind", None) + isolated = _bodies(registry, "h_fhir_badkind", session) + # No runner: the "unavailable here" error, in both modes, before any value is judged. + off_bare = _bodies(registry, "h_fhir_badkind", None) + isolated_bare = _bodies(registry, "h_fhir_badkind", session) + finally: + session.close() + assert "got int" in off[0] + assert isolated == off + assert "unavailable here" in off_bare[0] + assert isolated_bare == off_bare + + +def test_an_unexpected_engine_fault_fails_the_dispatch_promptly( + graph: tuple[Registry, str], + runners: _Runners, + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + def explode(self: Any, *args: Any) -> bytes: + raise RuntimeError("FAULT-SENTINEL-7f3a") + + monkeypatch.setattr(SandboxSession, "_lookup_reply", explode) + registry, config_dir = graph + session = SandboxSession( + SandboxPolicy(mode=SandboxMode.ISOLATED, wall_seconds=20.0, lookup_allowed_db=("db",)), + inbound="IB_I", + config_dir=config_dir, + env=None, + graph=None, + ) + try: + _bodies(registry, "h_parent", session) # warm the worker + started = time.monotonic() + with ( + caplog.at_level("DEBUG"), + pytest.raises(SandboxError, match="failed while serving") as raised, + ): + _bodies(registry, "h_db", session) + assert time.monotonic() - started < 10 + finally: + session.close() + assert "FAULT-SENTINEL-7f3a" not in str(raised.value) + assert not [r for r in caplog.records if "FAULT-SENTINEL-7f3a" in r.getMessage()] + assert any("RuntimeError" in r.getMessage() for r in caplog.records) + + +def test_cut_off_waiters_never_start_a_second_call_on_a_hung_connection( + graph: tuple[Registry, str], monkeypatch: pytest.MonkeyPatch +) -> None: + # A's lookup hangs and stays recorded. Dispatch B's helper is still building its refusal when + # B's cap fires, and C then finds A's record too. Neither may start a second call on it. + release = threading.Event() + calls: list[str] = [] + hung_helpers: list[threading.Thread] = [] + refusing_helpers: list[threading.Thread] = [] + + def hung(connection: str, statement: str, params: Any) -> list[dict[str, Any]]: + calls.append(statement) + hung_helpers.append(threading.current_thread()) + release.wait(60) + return [{"v": "late", "d": Decimal(0)}] + + real_refusal = SandboxSession._busy_refusal + + def slow_refusal(lookup: codec.LookupRequest) -> str: + refusing_helpers.append(threading.current_thread()) + time.sleep(2.0) # past the 1 s margin at wall_seconds=4 + return real_refusal(lookup) + + monkeypatch.setattr(SandboxSession, "_busy_refusal", staticmethod(slow_refusal)) + registry, config_dir = graph + session = SandboxSession( + SandboxPolicy(mode=SandboxMode.ISOLATED, wall_seconds=4.0, lookup_allowed_db=("db",)), + inbound="IB_I", + config_dir=config_dir, + env=None, + graph=None, + ) + try: + with db_lookup_mod.activated(hung): + for _ in range(3): # A hangs; B and C each find the connection busy + with pytest.raises(SandboxError, match="exceeded the 4.0s wall cap"): + _bodies(registry, "h_db", session) + # B's and C's helpers must be over before the count is read, or a late call could land + # after it. Each named itself in the refusal it was building when its cap fired. + assert len(refusing_helpers) == 2 + for thread in refusing_helpers: + thread.join(10) + assert not thread.is_alive() + assert len(calls) == 1 + assert hung_helpers[0].is_alive() # control: A really is still inside the runner + finally: + release.set() + session.close() + + +def test_a_helper_past_its_deadline_never_calls_the_runner( + graph: tuple[Registry, str], monkeypatch: pytest.MonkeyPatch +) -> None: + # Dispatch B's helper starts so late that B's cap has already ended, and by then the hung lookup + # A has finished. It must not call the runner: nobody waits for that call. The guard is the call + # count; peak alone would miss it, since such a call is recorded and C would wait for it. + release_a, release_all = threading.Event(), threading.Event() + lock = threading.Lock() + state = {"calls": 0, "in_flight": 0, "peak": 0} + + def hung(connection: str, statement: str, params: Any) -> list[dict[str, Any]]: + with lock: + state["calls"] += 1 + first = state["calls"] == 1 + state["in_flight"] += 1 + state["peak"] = max(state["peak"], state["in_flight"]) + try: + (release_a if first else release_all).wait(60) + return [{"v": "late", "d": Decimal(0)}] + finally: + with lock: + state["in_flight"] -= 1 + + go = threading.Event() + go.set() + real = SandboxSession._lookup_refusal + + def gated(self: SandboxSession, phase: str, lookup: codec.LookupRequest) -> str | None: + go.wait(30) # stands in for a helper thread scheduled late, past its dispatch's cap + return real(self, phase, lookup) + + monkeypatch.setattr(SandboxSession, "_lookup_refusal", gated) + registry, config_dir = graph + session = SandboxSession( + SandboxPolicy(mode=SandboxMode.ISOLATED, wall_seconds=2.0, lookup_allowed_db=("db",)), + inbound="IB_I", + config_dir=config_dir, + env=None, + graph=None, + ) + try: + with db_lookup_mod.activated(hung): + earlier = set(threading.enumerate()) + with pytest.raises(SandboxError, match="wall cap"): + _bodies(registry, "h_db", session) # A: hangs, cut off + # Exactly one helper each for A and B: the sync points below depend on finding them. + (a_helper,) = _new_lookup_helpers(earlier) + go.clear() + earlier = set(threading.enumerate()) + with pytest.raises(SandboxError, match="wall cap"): + _bodies(registry, "h_db", session) # B: its helper is held before the runner + (b_helper,) = _new_lookup_helpers(earlier) + release_a.set() + # Wait for A's own helper to finish, not the runner's counter: only then is the connection + # free, so B's helper can only be stopped by its dispatch being over. + a_helper.join(10) + assert not a_helper.is_alive() + go.set() # B's helper goes on, past B's deadline, and finds A finished + b_helper.join(10) + assert not b_helper.is_alive() + release_all.set() + assert _bodies(registry, "h_db", session)[0].startswith("late|") # C + assert (state["calls"], state["peak"]) == (2, 1) # A and C; never B + finally: + release_a.set() + release_all.set() + go.set() + session.close() + + +@pytest.mark.parametrize("over", ["deadline", "abandoned"]) +def test_each_late_runner_check_stops_the_call_alone(tmp_path: Path, over: str) -> None: + # The test above meets both checks at once, so it passes with either one deleted. Here only one + # says the dispatch is over: a deadline already past with nobody gone, or an abandoned dispatch + # with time left. The window between the deadline and abandoned.set() is the first case. + calls: list[str] = [] + + def runner(connection: str, statement: str, params: Any) -> list[dict[str, Any]]: + calls.append(statement) + return [] + + session = _session(str(tmp_path), SandboxMode.ISOLATED) + lookup = codec.LookupRequest( + seq=1, kind=codec.LOOKUP_DB, connection="LK", target="SELECT 1", params=None + ) + abandoned = threading.Event() + if over == "abandoned": + abandoned.set() + deadline = time.monotonic() + 30 + else: + deadline = time.monotonic() - 0.01 + try: + with db_lookup_mod.activated(runner): + ok, _, error = session._serve_lookup("transform", lookup, deadline, abandoned) + assert (ok, calls) == (False, []) + assert "wall cap ran out" in error + # Control: the same lookup, with time left and its dispatch still waiting, is served. + ok, _, _ = session._serve_lookup( + "transform", lookup, time.monotonic() + 30, threading.Event() + ) + assert (ok, calls) == (True, ["SELECT 1"]) + assert session._running_lookups == {} + finally: + session.close()