From c808752c3694ea3a384b1800f49a66082fcf3f86 Mon Sep 17 00:00:00 2001 From: Matthew Wesney Date: Thu, 23 Jul 2026 13:11:41 -0400 Subject: [PATCH] Add fixed recipe worker protocol boundary --- .../execution/worker_protocol.py | 424 ++++++++++++++++++ docs/adr/0001-phase2-evidence.md | 11 +- docs/adr/0001-phase2-worker-protocol.md | 76 ++++ packaging/build_recipe_worker.ps1 | 29 ++ packaging/recipe_worker/README.md | 26 ++ packaging/recipe_worker/recipe_worker.py | 24 + packaging/recipe_worker/recipe_worker.spec | 51 +++ tests/test_phase2_worker_protocol.py | 245 ++++++++++ tools/execution_spikes/README.md | 14 + 9 files changed, 896 insertions(+), 4 deletions(-) create mode 100644 backend/cortex_backend/execution/worker_protocol.py create mode 100644 docs/adr/0001-phase2-worker-protocol.md create mode 100644 packaging/build_recipe_worker.ps1 create mode 100644 packaging/recipe_worker/README.md create mode 100644 packaging/recipe_worker/recipe_worker.py create mode 100644 packaging/recipe_worker/recipe_worker.spec create mode 100644 tests/test_phase2_worker_protocol.py diff --git a/backend/cortex_backend/execution/worker_protocol.py b/backend/cortex_backend/execution/worker_protocol.py new file mode 100644 index 0000000..ec519af --- /dev/null +++ b/backend/cortex_backend/execution/worker_protocol.py @@ -0,0 +1,424 @@ +"""Strict, transport-neutral contract for the fixed recipe worker. + +The native broker owns framing, authentication, peer identity, and job ownership. +This module validates the bounded messages carried inside that broker session. It +does not open a pipe, read a path, or launch a process. The worker receives image +bytes in authenticated chunks and returns only bounded, content-addressed output +metadata and chunks. +""" + +from __future__ import annotations + +import base64 +import binascii +import hashlib +import json +import re +from typing import Any, Final, Literal + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +from .recipe_provider import ( + MAX_INPUT_BYTES, + MAX_OUTPUT_BYTES, + RecipeProviderError, + RecipeProviderResult, +) +from .recipes import ImageTransformPlan + + +MAX_WORKER_CHUNK_BYTES: Final[int] = 48 * 1024 +MAX_WORKER_CHUNK_TEXT: Final[int] = 4 * ((MAX_WORKER_CHUNK_BYTES + 2) // 3) +MAX_WORKER_INPUT_BYTES: Final[int] = MAX_INPUT_BYTES +MAX_WORKER_OUTPUT_BYTES: Final[int] = MAX_OUTPUT_BYTES +_HASH = re.compile(r"^[0-9a-f]{64}$") +_SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$") +_SAFE_CODE = re.compile(r"^[a-z][a-z0-9_]{0,63}$") + + +class WorkerProtocolError(ValueError): + """Stable, non-sensitive worker protocol failure category.""" + + def __init__(self, code: str) -> None: + if _SAFE_CODE.fullmatch(code) is None: + raise ValueError("invalid worker protocol code") + self.code = code + super().__init__("The recipe worker rejected the request safely.") + + +class _WorkerModel(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) + + def canonical_json(self) -> bytes: + try: + return json.dumps( + self.model_dump(mode="json"), + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("ascii") + except (TypeError, ValueError, OverflowError, UnicodeEncodeError): + raise WorkerProtocolError("message_invalid") from None + + +def _safe_id(value: str) -> str: + if _SAFE_ID.fullmatch(value) is None: + raise ValueError("worker identifier is invalid") + return value + + +def _safe_hash(value: str) -> str: + if _HASH.fullmatch(value) is None: + raise ValueError("worker digest is invalid") + return value + + +def _safe_code(value: str) -> str: + if _SAFE_CODE.fullmatch(value) is None: + raise ValueError("worker error code is invalid") + return value + + +def _decode_chunk(value: str) -> bytes: + if not isinstance(value, str) or not 1 <= len(value) <= MAX_WORKER_CHUNK_TEXT: + raise WorkerProtocolError("chunk_invalid") + if not re.fullmatch(r"[A-Za-z0-9_-]+={0,2}", value): + raise WorkerProtocolError("chunk_invalid") + padded = value + "=" * (-len(value) % 4) + try: + decoded = base64.b64decode(padded, altchars=b"-_", validate=True) + except (ValueError, binascii.Error): + raise WorkerProtocolError("chunk_invalid") from None + if not decoded or len(decoded) > MAX_WORKER_CHUNK_BYTES: + raise WorkerProtocolError("chunk_invalid") + canonical = base64.urlsafe_b64encode(decoded).decode("ascii").rstrip("=") + if canonical != value.rstrip("="): + raise WorkerProtocolError("chunk_noncanonical") + return decoded + + +class WorkerPrepare(_WorkerModel): + """Begin one immutable image-transform request.""" + + schema_version: Literal["recipe.worker.prepare.v1"] + request_id: str + job_id: str + plan: ImageTransformPlan + input_size: int = Field(strict=True, ge=1, le=MAX_WORKER_INPUT_BYTES) + input_sha256: str + input_mime_type: Literal["image/png", "image/jpeg", "image/webp"] + + _request_id = field_validator("request_id")(_safe_id) + _job_id = field_validator("job_id")(_safe_id) + _input_sha256 = field_validator("input_sha256")(_safe_hash) + + +class WorkerInputChunk(_WorkerModel): + """Append one in-order, independently hashed input chunk.""" + + schema_version: Literal["recipe.worker.input_chunk.v1"] + request_id: str + job_id: str + offset: int = Field(strict=True, ge=0, le=MAX_WORKER_INPUT_BYTES) + data: str = Field(min_length=1, max_length=MAX_WORKER_CHUNK_TEXT) + sha256: str + + _request_id = field_validator("request_id")(_safe_id) + _job_id = field_validator("job_id")(_safe_id) + _sha256 = field_validator("sha256")(_safe_hash) + + def decoded(self) -> bytes: + content = _decode_chunk(self.data) + if hashlib.sha256(content).hexdigest() != self.sha256: + raise WorkerProtocolError("chunk_hash_mismatch") + return content + + +class WorkerInputComplete(_WorkerModel): + """Commit the input stream only after size and whole-stream hash match.""" + + schema_version: Literal["recipe.worker.input_complete.v1"] + request_id: str + job_id: str + input_size: int = Field(strict=True, ge=1, le=MAX_WORKER_INPUT_BYTES) + input_sha256: str + + _request_id = field_validator("request_id")(_safe_id) + _job_id = field_validator("job_id")(_safe_id) + _input_sha256 = field_validator("input_sha256")(_safe_hash) + + +class WorkerCancel(_WorkerModel): + """Request bounded cancellation of the current transform.""" + + schema_version: Literal["recipe.worker.cancel.v1"] + request_id: str + job_id: str + reason: Literal["user", "timeout", "shutdown"] + + _request_id = field_validator("request_id")(_safe_id) + _job_id = field_validator("job_id")(_safe_id) + + +class WorkerCollect(_WorkerModel): + """Read one bounded output chunk after a successful transform.""" + + schema_version: Literal["recipe.worker.collect.v1"] + request_id: str + job_id: str + offset: int = Field(strict=True, ge=0, le=MAX_WORKER_OUTPUT_BYTES) + max_bytes: int = Field(strict=True, ge=1, le=MAX_WORKER_CHUNK_BYTES) + + _request_id = field_validator("request_id")(_safe_id) + _job_id = field_validator("job_id")(_safe_id) + + +class WorkerOutputChunk(_WorkerModel): + """A bounded, content-addressed output chunk.""" + + schema_version: Literal["recipe.worker.output_chunk.v1"] + request_id: str + job_id: str + offset: int = Field(strict=True, ge=0, le=MAX_WORKER_OUTPUT_BYTES) + data: str = Field(min_length=1, max_length=MAX_WORKER_CHUNK_TEXT) + sha256: str + final: bool + + _request_id = field_validator("request_id")(_safe_id) + _job_id = field_validator("job_id")(_safe_id) + _sha256 = field_validator("sha256")(_safe_hash) + + def decoded(self) -> bytes: + content = _decode_chunk(self.data) + if hashlib.sha256(content).hexdigest() != self.sha256: + raise WorkerProtocolError("chunk_hash_mismatch") + return content + + +class WorkerResult(_WorkerModel): + """Private metadata for the complete output, before host publication.""" + + schema_version: Literal["recipe.worker.result.v1"] + request_id: str + job_id: str + mime_type: Literal["image/png", "image/jpeg", "image/webp"] + format: Literal["PNG", "JPEG", "WEBP"] + width: int = Field(strict=True, ge=1, le=16_384) + height: int = Field(strict=True, ge=1, le=16_384) + output_size: int = Field(strict=True, ge=1, le=MAX_WORKER_OUTPUT_BYTES) + output_sha256: str + + _request_id = field_validator("request_id")(_safe_id) + _job_id = field_validator("job_id")(_safe_id) + _output_sha256 = field_validator("output_sha256")(_safe_hash) + + @staticmethod + def from_provider( + result: RecipeProviderResult, + *, + request_id: str, + job_id: str, + ) -> "WorkerResult": + return WorkerResult( + schema_version="recipe.worker.result.v1", + request_id=request_id, + job_id=job_id, + mime_type=result.mime_type, + format=result.format, + width=result.width, + height=result.height, + output_size=len(result.content), + output_sha256=result.sha256, + ) + + +class WorkerError(_WorkerModel): + """Redacted worker failure; paths, decoder text, and payloads never leave it.""" + + schema_version: Literal["recipe.worker.error.v1"] + request_id: str + job_id: str + code: str + + _request_id = field_validator("request_id")(_safe_id) + _job_id = field_validator("job_id")(_safe_id) + _code = field_validator("code")(_safe_code) + + +class RecipeWorkerSession: + """Bounded state machine used by the future native worker loop. + + The session has no filesystem or process capabilities. A native launcher must + provide the already-started provider and authenticated transport; this class + only enforces request ordering, byte limits, hashes, cancellation, and output + chunking. + """ + + def __init__(self, provider: Any) -> None: + if not hasattr(provider, "transform"): + raise TypeError("worker provider must expose transform") + self._provider = provider + self._state = "idle" + self._prepare: WorkerPrepare | None = None + self._input = bytearray() + self._result: RecipeProviderResult | None = None + self._cancelled = False + + @property + def state(self) -> Literal["idle", "receiving", "running", "complete", "cancelled", "failed"]: + return self._state # type: ignore[return-value] + + def prepare(self, message: WorkerPrepare) -> None: + if self._state != "idle": + raise WorkerProtocolError("request_state_invalid") + self._prepare = message + self._input = bytearray() + self._result = None + self._cancelled = False + self._state = "receiving" + + def input_chunk(self, message: WorkerInputChunk) -> None: + if self._state != "receiving" or self._prepare is None: + raise WorkerProtocolError("request_state_invalid") + if message.request_id != self._prepare.request_id or message.job_id != self._prepare.job_id: + raise WorkerProtocolError("request_identity_mismatch") + if message.offset != len(self._input): + raise WorkerProtocolError("chunk_out_of_order") + content = message.decoded() + if len(self._input) + len(content) > self._prepare.input_size: + raise WorkerProtocolError("input_size_exceeded") + self._input.extend(content) + + def cancel(self, message: WorkerCancel) -> None: + if self._prepare is None or message.request_id != self._prepare.request_id: + raise WorkerProtocolError("request_identity_mismatch") + if message.job_id != self._prepare.job_id: + raise WorkerProtocolError("request_identity_mismatch") + if self._state in {"complete", "failed"}: + raise WorkerProtocolError("request_state_invalid") + self._cancelled = True + self._state = "cancelled" + self._input.clear() + self._result = None + + def complete_input(self, message: WorkerInputComplete) -> WorkerResult: + if self._state != "receiving" or self._prepare is None: + raise WorkerProtocolError("request_state_invalid") + if message.request_id != self._prepare.request_id or message.job_id != self._prepare.job_id: + raise WorkerProtocolError("request_identity_mismatch") + if message.input_size != self._prepare.input_size or message.input_sha256 != self._prepare.input_sha256: + raise WorkerProtocolError("input_claim_mismatch") + if len(self._input) != self._prepare.input_size: + raise WorkerProtocolError("input_size_mismatch") + if hashlib.sha256(self._input).hexdigest() != self._prepare.input_sha256: + raise WorkerProtocolError("input_hash_mismatch") + self._state = "running" + try: + result = self._provider.transform( + self._prepare.plan, + bytes(self._input), + cancel_check=lambda: self._cancelled, + ) + except RecipeProviderError as error: + self._state = "cancelled" if error.code == "cancelled" else "failed" + self._input.clear() + raise WorkerProtocolError(error.code) from None + except Exception: + self._state = "failed" + self._input.clear() + raise WorkerProtocolError("provider_failed") from None + if self._cancelled: + self._state = "cancelled" + self._input.clear() + raise WorkerProtocolError("cancelled") + self._input.clear() + self._result = result + self._state = "complete" + return WorkerResult.from_provider( + result, + request_id=self._prepare.request_id, + job_id=self._prepare.job_id, + ) + + def collect(self, message: WorkerCollect) -> WorkerOutputChunk: + if self._state != "complete" or self._prepare is None or self._result is None: + raise WorkerProtocolError("request_state_invalid") + if message.request_id != self._prepare.request_id or message.job_id != self._prepare.job_id: + raise WorkerProtocolError("request_identity_mismatch") + output = self._result.content + if message.offset > len(output): + raise WorkerProtocolError("output_offset_invalid") + end = min(message.offset + message.max_bytes, len(output)) + chunk = output[message.offset:end] + if not chunk: + raise WorkerProtocolError("output_offset_invalid") + encoded = base64.urlsafe_b64encode(chunk).decode("ascii").rstrip("=") + return WorkerOutputChunk( + schema_version="recipe.worker.output_chunk.v1", + request_id=message.request_id, + job_id=message.job_id, + offset=message.offset, + data=encoded, + sha256=hashlib.sha256(chunk).hexdigest(), + final=end == len(output), + ) + + +class RecipeWorkerDispatcher: + """Decode one broker operation and dispatch it to a bounded session. + + The caller must authenticate and authorize the enclosing broker frame before + calling this method. Unknown operations and malformed bodies fail closed; + this dispatcher never interprets a string as a command or filesystem path. + """ + + _MODELS = { + "prepare": WorkerPrepare, + "input_chunk": WorkerInputChunk, + "input_complete": WorkerInputComplete, + "cancel": WorkerCancel, + "collect": WorkerCollect, + } + + def __init__(self, session: RecipeWorkerSession) -> None: + self._session = session + + def dispatch(self, operation: str, body: dict[str, Any]) -> WorkerResult | WorkerOutputChunk | None: + model_type = self._MODELS.get(operation) + if model_type is None or not isinstance(body, dict): + raise WorkerProtocolError("operation_invalid") + try: + message = model_type.model_validate(body) + except (TypeError, ValueError): + raise WorkerProtocolError("message_invalid") from None + if operation == "prepare": + self._session.prepare(message) + return None + if operation == "input_chunk": + self._session.input_chunk(message) + return None + if operation == "input_complete": + return self._session.complete_input(message) + if operation == "cancel": + self._session.cancel(message) + return None + return self._session.collect(message) + + +__all__ = [ + "MAX_WORKER_CHUNK_BYTES", + "MAX_WORKER_INPUT_BYTES", + "MAX_WORKER_OUTPUT_BYTES", + "RecipeWorkerDispatcher", + "RecipeWorkerSession", + "WorkerCancel", + "WorkerCollect", + "WorkerError", + "WorkerInputChunk", + "WorkerInputComplete", + "WorkerOutputChunk", + "WorkerPrepare", + "WorkerProtocolError", + "WorkerResult", +] diff --git a/docs/adr/0001-phase2-evidence.md b/docs/adr/0001-phase2-evidence.md index 0a7207a..8ec1398 100644 --- a/docs/adr/0001-phase2-evidence.md +++ b/docs/adr/0001-phase2-evidence.md @@ -5,6 +5,7 @@ - **Scope:** Provider-independent contracts plus a qualification-only fixed-function core - **Source decision:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md) - **Contract ADR:** [Phase 2 typed recipe and primitive contract](0001-phase2-recipe-contract.md) +- **Worker ADR:** [Phase 2 fixed recipe worker protocol and package boundary](0001-phase2-worker-protocol.md) ## Stage checklist @@ -17,13 +18,14 @@ | Signed recipe manifest | **Complete (verification only)** | Ed25519 signature verification uses a pinned key-id allowlist; every declared bundle entry is path-, size-, and SHA-256-verified; monotonic updates and explicit rollback authorization are enforced. | | Signed bundle installation/update | **Complete (storage-only)** | Digest-named immutable generations, exclusive staging, atomic activation state, chained keyring rotation, explicit rollback authorization, and previous-generation recovery are covered by installer tests. No provider is loaded. | | Signed worker provenance binding | **Complete (storage-only)** | `verify_active_worker()` rechecks the active signed generation, binds exactly one `image_transform` role to `recipe_worker.exe`, revalidates byte identity, and rejects missing/ambiguous/mismatched/tampered/reparse entries without launching. | +| Fixed worker protocol and package closure | **Complete (qualification-only)** | `worker_protocol.py` enforces bounded prepare/chunk/complete/cancel/collect state, canonical per-chunk hashes, whole-input claims, redacted output metadata, and no-capability bodies. `packaging/recipe_worker/recipe_worker.spec` builds the fixed `recipe_worker.exe`; the entrypoint exits `78` until native broker wiring exists. | | Authenticated broker contract | **Complete (transport-neutral)** | Bounded versioned frames, direction-specific HMAC keys, canonical messages, peer ACL/integrity policy, and owner-scoped authorization are covered by adversarial tests. | | Native named-pipe adapter/DACL/peer-token binding | **Complete (transport-only)** | Protected local pipe, expected PID, OS token identity, X25519/HKDF handshake, direction keys, and close-on-error lifecycle are covered by native broker tests. | | User-artifact copy-in, output validation, and publication | **Complete (boundary only)** | Explicit owner/turn grants, bounded stable snapshots, link/reparse/hardlink/sparse/ADS rejection, byte-derived MIME policy, exact output claims, quarantine, hash/size limits, atomic repository publication, rollback, and cleanup categories are covered by `tests/test_phase2_artifact_boundary.py`. | | Fixed-function image provider core | **Complete (qualification-only)** | `RecipeImageProvider` validates allowlisted PNG/JPEG/WebP bytes, verifies/loads one frame with Pillow bomb/resource limits, applies only parsed steps, strips metadata, revalidates encoded output, checks cancellation, and remains disabled until external sandbox health passes. | | Windows recipe sandbox qualification harness | **Complete (qualification harness; worker gate blocked)** | `recipe_sandbox_qualification.py` composes out-of-process AppContainer isolation and Job Object cancellation with a fixed decoder corpus, then fails closed because the signed `recipe_worker.exe` bundle and trust-root launch verification are not shipped. | | Suspended native launcher/resource policy | **Complete (disposable control spike)** | `native_launcher_qualification.py` creates a fixed suspended AppContainer child, applies and queries active-process, kill-on-close, CPU-time, memory, and no-breakaway Job Object policy before resume; real worker enforcement and broker binding remain blocked. | -| OS sandbox provider and provider-produced image outputs | **Blocked / release gate** | The actual provider worker still needs signed provenance, LPAC/AppContainer policy, Job Object resource limits/accounting, broker PID/token binding, watchdog, hostile decoder execution inside the sandbox, external review, and lifecycle wiring. | +| OS sandbox provider and provider-produced image outputs | **Blocked / release gate** | The package/protocol boundary is qualified, but the actual provider worker still needs a signed installed generation, native broker loop, LPAC/AppContainer policy, Job Object resource limits/accounting, live broker PID/token binding, watchdog, hostile decoder execution inside the sandbox, external review, and lifecycle wiring. | ## Security invariants @@ -109,13 +111,14 @@ npm.cmd run build --prefix frontend npm.cmd test --prefix frontend -- --run ``` -**Validation result (2026-07-21):** 16 Phase 2 contract tests, 9 signed-manifest tests, +**Validation result (2026-07-23):** 16 Phase 2 contract tests, 9 signed-manifest tests, 7 broker-contract tests, 9 native-broker tests, 7 bundle-installer tests, 16 -artifact-boundary tests, 17 recipe-provider tests, 6 worker-provenance tests, 4 +artifact-boundary tests, 17 recipe-provider tests, 6 worker-provenance tests, 7 +worker-protocol tests, 4 native-launcher tests, and 5 sandbox-qualification tests passed; the full Python suite -passed (219 tests total) with one +passed (226 tests total) with one native-platform skip and one pre-existing `pytest-asyncio` deprecation warning. Frontend lint, typecheck, production build, and all 39 frontend tests passed. Contract generation, compileall, and `git diff --check` passed. No production execution diff --git a/docs/adr/0001-phase2-worker-protocol.md b/docs/adr/0001-phase2-worker-protocol.md new file mode 100644 index 0000000..8336034 --- /dev/null +++ b/docs/adr/0001-phase2-worker-protocol.md @@ -0,0 +1,76 @@ +# ADR-0001 Phase 2 fixed recipe worker protocol and package boundary + +- **Status:** Protocol and packaging qualification complete; native execution blocked +- **Phase:** 2 - fixed-function image provider +- **Parent:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md) +- **Depends on:** [Signed worker provenance](0001-phase2-worker-provenance.md), [native broker adapter](0001-phase2-native-broker.md), and [native launcher](0001-phase2-native-launcher.md) +- **Scope:** Bounded worker messages, in-order byte streaming, fixed provider session state, and reproducible Windows package closure + +## Decision + +The worker protocol is transport-neutral and runs only inside an authenticated +native broker session. The broker remains responsible for framing, HMAC direction +keys, peer PID/token identity, installation principal, job ownership, and lifecycle +shutdown. The worker protocol validates the operation body after those checks. + +One request follows this sequence: + +1. `recipe.worker.prepare.v1` declares one parsed `ImageTransformPlan`, an opaque + artifact identifier, expected byte count, expected SHA-256, and an allowlisted + image MIME type. +2. One or more `recipe.worker.input_chunk.v1` messages append canonical base64 + chunks in strict offset order. Each chunk has its own SHA-256 and is at most + 48 KiB decoded. +3. `recipe.worker.input_complete.v1` repeats the whole-input size and digest. The + worker decodes only after both claims match the received bytes. +4. The fixed provider runs with the worker's cancellation callback and returns + only redacted metadata in `recipe.worker.result.v1`. +5. `recipe.worker.collect.v1` reads bounded output chunks. Each output chunk has a + digest and final marker; the host artifact boundary remains responsible for + quarantine, validation, hashing, and publication. + +Cancellation is terminal for the request. Unknown operations, malformed bodies, +replayed offsets, identity mismatches, size/digest mismatches, provider failures, +and output offsets fail closed with stable categories. No worker message accepts a +filesystem path, shell command, executable, network target, token, or model source. + +The package entrypoint is intentionally launch-refusing (exit code `78`) until the +reviewed native broker adapter is wired in. This prevents an unsigned or directly +started package from becoming a stdio, shell, or host-process fallback. The +PyInstaller definition and Windows build script qualify dependency closure and the +fixed `recipe_worker.exe` path only; they do not sign, install, or authorize the +worker. + +## Security invariants + +- The worker never opens a path or creates a transport. +- Input and output bytes are bounded by the provider ceilings; chunks are bounded + below the broker frame ceiling. +- Input chunks are canonical, contiguous, independently hashed, and committed only + after whole-stream size and digest verification. +- The session state machine permits one request at a time and never resumes after + cancellation or terminal failure. +- Output metadata is private and content-addressed; publication remains outside the + worker session. +- A package build is not a signed bundle. The release pipeline must create the + exact `recipe.manifest.v1` entry for `image_transform`/`recipe_worker.exe`, sign + it with the pinned key, install it through `SignedBundleInstaller`, and re-run + `verify_active_worker()`. + +## Evidence + +`tests/test_phase2_worker_protocol.py` covers successful in-order streaming and +collection, malformed operations, replay/order failure, claim mismatch, +cancellation terminality, and chunk tampering. On the controlled Windows host +(2026-07-23), the PyInstaller build produced +`dist/recipe-runtime/recipe_worker.exe` (11,320,824 bytes); direct launch returned +exit code `78` as required. + +## Remaining blockers + +This ADR does not authorize provider execution. The next stage must replace the +launch-refusing entrypoint with the reviewed native broker loop, launch the verified +worker suspended under AppContainer/LPAC and Job Object policy, bind the live broker +session to the actual worker PID and AppContainer token, and run the hostile decoder +and cancellation corpus inside that process. Lifecycle/UI enablement remains behind +those gates and external security review. diff --git a/packaging/build_recipe_worker.ps1 b/packaging/build_recipe_worker.ps1 new file mode 100644 index 0000000..807c088 --- /dev/null +++ b/packaging/build_recipe_worker.ps1 @@ -0,0 +1,29 @@ +param( + [switch]$SkipDependencyInstall +) + +$ErrorActionPreference = "Stop" +$RepositoryRoot = Split-Path -Parent $PSScriptRoot +Set-Location $RepositoryRoot + +if (-not $IsWindows -and $PSVersionTable.PSEdition -eq "Core") { + throw "The recipe worker must be packaged on Windows." +} + +if (-not $SkipDependencyInstall) { + python -m pip install --disable-pip-version-check -r requirements.txt + python -m pip install --disable-pip-version-check "pyinstaller>=6.14,<7" +} + +Remove-Item -LiteralPath (Join-Path $RepositoryRoot "build\recipe_worker") -Recurse -Force -ErrorAction SilentlyContinue +Remove-Item -LiteralPath (Join-Path $RepositoryRoot "dist\recipe-runtime") -Recurse -Force -ErrorAction SilentlyContinue + +python -m PyInstaller --noconfirm --clean packaging/recipe_worker/recipe_worker.spec + +$Executable = Join-Path $RepositoryRoot "dist\recipe-runtime\recipe_worker.exe" +if (-not (Test-Path -LiteralPath $Executable -PathType Leaf)) { + throw "The recipe worker package did not produce recipe_worker.exe." +} + +Write-Host "Recipe worker contract package ready: $Executable" +Write-Host "This output is intentionally unsigned and not launch-authorized." diff --git a/packaging/recipe_worker/README.md b/packaging/recipe_worker/README.md new file mode 100644 index 0000000..0822449 --- /dev/null +++ b/packaging/recipe_worker/README.md @@ -0,0 +1,26 @@ +# Recipe worker package + +This directory defines the fixed-function worker package boundary. The package +entrypoint deliberately exits with status `78` until the reviewed native broker +adapter is added. That refusal is a safety property: a package build must not +silently become a host-process or stdio execution fallback. + +Build it on a supported Windows machine with: + +```powershell +powershell -ExecutionPolicy Bypass -File packaging/build_recipe_worker.ps1 +``` + +The output is `dist/recipe-runtime/recipe_worker.exe`. It is a dependency-closure +artifact for qualification only and is **not signed or launch-authorized** by this +repository. Release packaging must, outside source control: + +1. hash every immutable package entry; +2. create the exact `recipe.manifest.v1` entry with role `image_transform` and + path `recipe_worker.exe`; +3. sign the canonical manifest with the pinned release key; and +4. install it through `SignedBundleInstaller` before `verify_active_worker()` can + be considered. + +The native launcher and live broker PID/AppContainer-token binding remain required +before this package can perform any transform. diff --git a/packaging/recipe_worker/recipe_worker.py b/packaging/recipe_worker/recipe_worker.py new file mode 100644 index 0000000..f2f5ec5 --- /dev/null +++ b/packaging/recipe_worker/recipe_worker.py @@ -0,0 +1,24 @@ +"""Fixed recipe worker package entrypoint. + +This entrypoint intentionally refuses to run until the native broker adapter is +provided by the reviewed launcher stage. It exists so packaging, dependency +closure, and signed-bundle provenance can be qualified without creating an +unsafe stdio, shell, or host-process fallback. +""" + +from __future__ import annotations + +import sys + + +def main(argv: list[str] | None = None) -> int: + """Refuse every launch until the native broker contract is wired in.""" + + del argv + # A non-zero deterministic exit is important: a packaged but unbound worker + # must never look healthy to a caller that accidentally starts it directly. + return 78 + + +if __name__ == "__main__": # pragma: no cover - exercised by the packaged exe. + raise SystemExit(main(sys.argv[1:])) diff --git a/packaging/recipe_worker/recipe_worker.spec b/packaging/recipe_worker/recipe_worker.spec new file mode 100644 index 0000000..de5d3be --- /dev/null +++ b/packaging/recipe_worker/recipe_worker.spec @@ -0,0 +1,51 @@ +"""PyInstaller one-folder definition for the fixed recipe worker contract.""" + +from pathlib import Path + + +ROOT = Path(SPECPATH).resolve().parents[1] + +a = Analysis( + [str(ROOT / "packaging" / "recipe_worker" / "recipe_worker.py")], + pathex=[str(ROOT), str(ROOT / "backend")], + binaries=[], + datas=[], + hiddenimports=[ + "cortex_backend.execution.worker_protocol", + "cortex_backend.execution.recipe_provider", + "PIL", + "PIL.Image", + "PIL.ImageEnhance", + "PIL.ImageFile", + ], + hookspath=[], + hooksconfig={}, + runtime_hooks=[], + excludes=[], + noarchive=False, + optimize=0, +) + +pyz = PYZ(a.pure) + +exe = EXE( + pyz, + a.scripts, + [], + name="recipe_worker", + debug=False, + bootloader_ignore_signals=False, + strip=False, + upx=False, + console=False, + exclude_binaries=True, +) + +coll = COLLECT( + exe, + a.binaries, + a.datas, + strip=False, + upx=False, + name="recipe-runtime", +) diff --git a/tests/test_phase2_worker_protocol.py b/tests/test_phase2_worker_protocol.py new file mode 100644 index 0000000..f08bec7 --- /dev/null +++ b/tests/test_phase2_worker_protocol.py @@ -0,0 +1,245 @@ +"""Qualification tests for the bounded fixed recipe worker contract.""" + +from __future__ import annotations + +import base64 +from hashlib import sha256 +from io import BytesIO +from threading import Event, Thread + +import pytest +from PIL import Image + +from cortex_backend.execution.lifecycle import RuntimeHealth +from cortex_backend.execution.recipe_provider import RecipeImageProvider, RecipeProviderResult +from cortex_backend.execution.recipes import parse_image_transform +from cortex_backend.execution.worker_protocol import ( + RecipeWorkerDispatcher, + RecipeWorkerSession, + WorkerCancel, + WorkerCollect, + WorkerInputChunk, + WorkerInputComplete, + WorkerPrepare, + WorkerProtocolError, +) + + +def _image_bytes() -> bytes: + image = Image.new("RGB", (4, 3), (120, 80, 40)) + try: + with BytesIO() as stream: + image.save(stream, format="PNG") + return stream.getvalue() + finally: + image.close() + + +def _plan(): + return parse_image_transform( + { + "schema_version": "artifact.transform.v1", + "input_artifact_id": "artifact-1", + "steps": [{"op": "grayscale"}], + "output_format": "png", + } + ) + + +def _provider() -> RecipeImageProvider: + provider = RecipeImageProvider() + assert provider.start(RuntimeHealth.ready("worker protocol test")) + return provider + + +def _prepare(content: bytes) -> WorkerPrepare: + return WorkerPrepare( + schema_version="recipe.worker.prepare.v1", + request_id="request-1", + job_id="job-1", + plan=_plan(), + input_size=len(content), + input_sha256=sha256(content).hexdigest(), + input_mime_type="image/png", + ) + + +def _chunk(content: bytes, *, offset: int = 0) -> WorkerInputChunk: + encoded = base64.urlsafe_b64encode(content).decode("ascii").rstrip("=") + return WorkerInputChunk( + schema_version="recipe.worker.input_chunk.v1", + request_id="request-1", + job_id="job-1", + offset=offset, + data=encoded, + sha256=sha256(content).hexdigest(), + ) + + +def test_worker_session_requires_in_order_hashed_input_and_collects_output(): + content = _image_bytes() + session = RecipeWorkerSession(_provider()) + prepare = _prepare(content) + session.prepare(prepare) + session.input_chunk(_chunk(content[:8])) + session.input_chunk(_chunk(content[8:], offset=8)) + + result = session.complete_input( + WorkerInputComplete( + schema_version="recipe.worker.input_complete.v1", + request_id="request-1", + job_id="job-1", + input_size=len(content), + input_sha256=sha256(content).hexdigest(), + ) + ) + + assert session.state == "complete" + assert result.output_size > 0 + output = session.collect( + WorkerCollect( + schema_version="recipe.worker.collect.v1", + request_id="request-1", + job_id="job-1", + offset=0, + max_bytes=1024, + ) + ) + assert output.decoded() + assert output.final + + +def test_dispatcher_rejects_unknown_or_malformed_operations(): + dispatcher = RecipeWorkerDispatcher(RecipeWorkerSession(_provider())) + with pytest.raises(WorkerProtocolError) as unknown: + dispatcher.dispatch("shell", {}) + assert unknown.value.code == "operation_invalid" + + with pytest.raises(WorkerProtocolError) as malformed: + dispatcher.dispatch("prepare", {"schema_version": "recipe.worker.prepare.v1"}) + assert malformed.value.code == "message_invalid" + + +def test_worker_session_rejects_replay_order_and_claim_mismatch(): + content = _image_bytes() + session = RecipeWorkerSession(_provider()) + session.prepare(_prepare(content)) + with pytest.raises(WorkerProtocolError) as order: + session.input_chunk(_chunk(content[:4], offset=1)) + assert order.value.code == "chunk_out_of_order" + + with pytest.raises(WorkerProtocolError) as claim: + session.complete_input( + WorkerInputComplete( + schema_version="recipe.worker.input_complete.v1", + request_id="request-1", + job_id="job-1", + input_size=len(content), + input_sha256="0" * 64, + ) + ) + assert claim.value.code == "input_claim_mismatch" + + +def test_worker_session_cancellation_is_terminal_and_redacted(): + content = _image_bytes() + session = RecipeWorkerSession(_provider()) + session.prepare(_prepare(content)) + session.cancel( + WorkerCancel( + schema_version="recipe.worker.cancel.v1", + request_id="request-1", + job_id="job-1", + reason="user", + ) + ) + assert session.state == "cancelled" + with pytest.raises(WorkerProtocolError) as after: + session.complete_input( + WorkerInputComplete( + schema_version="recipe.worker.input_complete.v1", + request_id="request-1", + job_id="job-1", + input_size=len(content), + input_sha256=sha256(content).hexdigest(), + ) + ) + assert after.value.code == "request_state_invalid" + + +def test_cancellation_wins_if_requested_while_provider_is_running(): + content = _image_bytes() + entered = Event() + release = Event() + + class _SlowProvider: + def transform(self, plan, payload, *, cancel_check): + del plan, payload, cancel_check + entered.set() + release.wait(timeout=5) + return RecipeProviderResult( + content=b"output", + mime_type="image/png", + width=1, + height=1, + format="PNG", + sha256=sha256(b"output").hexdigest(), + ) + + session = RecipeWorkerSession(_SlowProvider()) + session.prepare(_prepare(content)) + session.input_chunk(_chunk(content)) + complete_error: list[WorkerProtocolError] = [] + + def complete() -> None: + try: + session.complete_input( + WorkerInputComplete( + schema_version="recipe.worker.input_complete.v1", + request_id="request-1", + job_id="job-1", + input_size=len(content), + input_sha256=sha256(content).hexdigest(), + ) + ) + except WorkerProtocolError as error: + complete_error.append(error) + + thread = Thread(target=complete) + thread.start() + assert entered.wait(timeout=5) + session.cancel( + WorkerCancel( + schema_version="recipe.worker.cancel.v1", + request_id="request-1", + job_id="job-1", + reason="user", + ) + ) + release.set() + thread.join(timeout=5) + assert not thread.is_alive() + assert [error.code for error in complete_error] == ["cancelled"] + assert session.state == "cancelled" + + +@pytest.mark.parametrize( + "mutator", + [ + lambda payload: {**payload, "sha256": "0" * 64}, + lambda payload: {**payload, "data": "%%%"}, + ], +) +def test_worker_input_chunk_rejects_tampering(mutator): + content = b"worker input" + payload = { + "schema_version": "recipe.worker.input_chunk.v1", + "request_id": "request-1", + "job_id": "job-1", + "offset": 0, + "data": base64.urlsafe_b64encode(content).decode("ascii").rstrip("="), + "sha256": sha256(content).hexdigest(), + } + message = WorkerInputChunk.model_validate(mutator(payload)) + with pytest.raises(WorkerProtocolError): + message.decoded() diff --git a/tools/execution_spikes/README.md b/tools/execution_spikes/README.md index 5cb1f26..08aa7fb 100644 --- a/tools/execution_spikes/README.md +++ b/tools/execution_spikes/README.md @@ -48,6 +48,17 @@ the fixed suspended child receives and reports Job Object CPU/memory/active-proc limits with no breakaway flags. Its overall exit remains blocked until the signed worker package and broker PID/token binding exist. +The fixed worker protocol/package boundary can be qualified separately on Windows: + +```powershell +powershell -ExecutionPolicy Bypass -File packaging/build_recipe_worker.ps1 +``` + +This produces `dist/recipe-runtime/recipe_worker.exe` and verifies dependency +closure only. The entrypoint exits with status `78` until the native broker loop is +implemented; the output is unsigned and must not be installed or launched as a +provider. The package contract is covered by `tests/test_phase2_worker_protocol.py`. + ## What the probes prove - `environment`: supported Windows host and interpreter metadata. @@ -78,6 +89,9 @@ worker package and broker PID/token binding exist. - `native_launcher_qualification`: creates only a fixed suspended `findstr.exe` child, applies and queries Job Object resource policy before resume, and reports the signed-worker and broker-binding blockers without launching either. +- `worker_protocol`: validates the future worker's bounded request state machine, + in-order hashed chunks, cancellation, and redacted output contract; it has no + filesystem, process, or transport capability. - `security_review`: records the conditional Phase 0 spike review and residual blockers. - `pyinstaller_package_preconditions`: all currently known one-folder package