From bc7bcdbc1c724fb8f9816bad03c035b161480075 Mon Sep 17 00:00:00 2001 From: Matthew Wesney Date: Fri, 24 Jul 2026 13:22:18 -0400 Subject: [PATCH] Qualify signed worker launch and broker boundary Repair AppContainer package read/execute access with reversible per-profile ACLs, make worker startup broker reads deterministic, constrain Pillow codec loading, and add a disposable signed end-to-end qualification gate. Document the live evidence and keep the provider transform release gate blocked on timeout. --- .../cortex_backend/execution/native_win32.py | 170 ++++++++ .../execution/recipe_provider.py | 13 +- .../execution/worker_runtime.py | 24 +- docs/adr/0001-phase2-evidence.md | 22 +- packaging/recipe_worker/recipe_worker.spec | 3 + tests/test_phase2_native_win32.py | 12 +- tests/test_phase2_recipe_provider.py | 25 +- tools/execution_spikes/README.md | 27 ++ .../recipe_worker_e2e_qualification.py | 406 ++++++++++++++++++ 9 files changed, 687 insertions(+), 15 deletions(-) create mode 100644 tools/execution_spikes/recipe_worker_e2e_qualification.py diff --git a/backend/cortex_backend/execution/native_win32.py b/backend/cortex_backend/execution/native_win32.py index faa773b..0c3a1fa 100644 --- a/backend/cortex_backend/execution/native_win32.py +++ b/backend/cortex_backend/execution/native_win32.py @@ -12,6 +12,7 @@ import ctypes from ctypes import wintypes from dataclasses import dataclass +from pathlib import Path import sys from typing import Any from uuid import uuid4 @@ -35,6 +36,15 @@ _TOKEN_IS_APPCONTAINER = 29 _TOKEN_APPCONTAINER_SID = 31 _TOKEN_QUERY = 0x0008 +_SE_FILE_OBJECT = 1 +_DACL_SECURITY_INFORMATION = 0x00000004 +_TRUSTEE_IS_SID = 0 +_TRUSTEE_IS_UNKNOWN = 0 +_GRANT_ACCESS = 1 +_REVOKE_ACCESS = 4 +_OBJECT_INHERIT_ACE = 0x00000001 +_CONTAINER_INHERIT_ACE = 0x00000002 +_FILE_GENERIC_READ_EXECUTE = 0x001200A9 _WAIT_OBJECT_0 = 0 _WAIT_TIMEOUT = 0x102 _INFINITE = 0xFFFFFFFF @@ -129,6 +139,25 @@ class _SecurityCapabilities(ctypes.Structure): ] +class _TrusteeW(ctypes.Structure): + _fields_ = [ + ("multiple_trustee", ctypes.c_void_p), + ("multiple_trustee_operation", wintypes.DWORD), + ("trustee_form", wintypes.DWORD), + ("trustee_type", wintypes.DWORD), + ("name", ctypes.c_void_p), + ] + + +class _ExplicitAccessW(ctypes.Structure): + _fields_ = [ + ("access_permissions", wintypes.DWORD), + ("access_mode", wintypes.DWORD), + ("inheritance", wintypes.DWORD), + ("trustee", _TrusteeW), + ] + + @dataclass(slots=True) class _Win32: kernel32: Any @@ -187,6 +216,34 @@ def _configure() -> _Win32: ctypes.POINTER(wintypes.LPWSTR), ] advapi32.ConvertSidToStringSidW.restype = wintypes.BOOL + advapi32.GetNamedSecurityInfoW.argtypes = [ + wintypes.LPCWSTR, + wintypes.DWORD, + wintypes.DWORD, + ctypes.POINTER(ctypes.c_void_p), + ctypes.POINTER(ctypes.c_void_p), + ctypes.POINTER(ctypes.c_void_p), + ctypes.POINTER(ctypes.c_void_p), + ctypes.POINTER(ctypes.c_void_p), + ] + advapi32.GetNamedSecurityInfoW.restype = wintypes.DWORD + advapi32.SetEntriesInAclW.argtypes = [ + wintypes.DWORD, + ctypes.POINTER(_ExplicitAccessW), + ctypes.c_void_p, + ctypes.POINTER(ctypes.c_void_p), + ] + advapi32.SetEntriesInAclW.restype = wintypes.DWORD + advapi32.SetNamedSecurityInfoW.argtypes = [ + wintypes.LPWSTR, + wintypes.DWORD, + wintypes.DWORD, + ctypes.c_void_p, + ctypes.c_void_p, + ctypes.c_void_p, + ctypes.c_void_p, + ] + advapi32.SetNamedSecurityInfoW.restype = wintypes.DWORD kernel32.InitializeProcThreadAttributeList.argtypes = [ wintypes.LPVOID, @@ -305,6 +362,98 @@ def _app_container_sid(win: _Win32, process: Any) -> str: _close(win, token) +def _update_package_read_execute_acl( + win: _Win32, + package_root: Path, + profile_sid: Any, + *, + access_mode: int, +) -> None: + """Add or remove one fresh AppContainer read/execute ACE. + + User-owned application-data directories are not automatically readable by an + AppContainer. The immutable worker generation therefore receives a narrowly + scoped ACE for the newly created profile SID before CreateProcessW runs. The + existing DACL is merged rather than replaced, and inheritance covers the + one-folder dependency closure without granting write, delete, or network + capabilities. The ACE is removed when the worker closes. + """ + + if not package_root.is_dir() or package_root.is_symlink(): + _raise("native_package_root_invalid") + old_dacl = ctypes.c_void_p() + security_descriptor = ctypes.c_void_p() + new_dacl = ctypes.c_void_p() + trustee = _TrusteeW( + multiple_trustee=None, + multiple_trustee_operation=0, + trustee_form=_TRUSTEE_IS_SID, + trustee_type=_TRUSTEE_IS_UNKNOWN, + name=ctypes.cast(profile_sid, ctypes.c_void_p), + ) + explicit = _ExplicitAccessW( + access_permissions=_FILE_GENERIC_READ_EXECUTE if access_mode == _GRANT_ACCESS else 0, + access_mode=access_mode, + inheritance=_OBJECT_INHERIT_ACE | _CONTAINER_INHERIT_ACE, + trustee=trustee, + ) + try: + status = win.advapi32.GetNamedSecurityInfoW( + str(package_root), + _SE_FILE_OBJECT, + _DACL_SECURITY_INFORMATION, + None, + None, + ctypes.byref(old_dacl), + None, + ctypes.byref(security_descriptor), + ) + if status != 0: + _raise("native_package_acl_read_failed") + status = win.advapi32.SetEntriesInAclW( + 1, + ctypes.byref(explicit), + old_dacl, + ctypes.byref(new_dacl), + ) + if status != 0: + _raise("native_package_acl_build_failed") + status = win.advapi32.SetNamedSecurityInfoW( + str(package_root), + _SE_FILE_OBJECT, + _DACL_SECURITY_INFORMATION, + None, + None, + new_dacl, + None, + ) + if status != 0: + _raise("native_package_acl_write_failed") + finally: + if new_dacl.value: + win.kernel32.LocalFree(new_dacl) + if security_descriptor.value: + win.kernel32.LocalFree(security_descriptor) + + +def _grant_package_read_execute(win: _Win32, package_root: Path, profile_sid: Any) -> None: + _update_package_read_execute_acl( + win, + package_root, + profile_sid, + access_mode=_GRANT_ACCESS, + ) + + +def _revoke_package_read_execute(win: _Win32, package_root: Path, profile_sid: Any) -> None: + _update_package_read_execute_acl( + win, + package_root, + profile_sid, + access_mode=_REVOKE_ACCESS, + ) + + class Win32SuspendedWorker: """Own a suspended worker and its kill-on-close Job Object.""" @@ -317,6 +466,8 @@ def __init__( thread: Any, process_id: int, app_container_sid: str, + package_root: Path, + profile_sid: Any, ) -> None: self._win = win self._profile_name = profile_name @@ -325,6 +476,8 @@ def __init__( self._job: Any = None self._closed = False self._resumed = False + self._package_root = package_root + self._profile_sid = profile_sid self.process_id = process_id self.app_container_sid = app_container_sid @@ -408,6 +561,12 @@ def close(self) -> None: _close(self._win, self._process) self._thread = None self._process = None + try: + _revoke_package_read_execute(self._win, self._package_root, self._profile_sid) + except Exception: + # The process is already closed; ACL cleanup must not mask the + # lifecycle result or prevent profile deletion. + pass self._win.userenv.DeleteAppContainerProfile(self._profile_name) def __enter__(self) -> "Win32SuspendedWorker": @@ -434,6 +593,7 @@ def create_suspended(self, plan: NativeWorkerLaunchPlan) -> NativeSuspendedWorke attribute_list: wintypes.LPVOID | None = None attribute_buffer: ctypes.Array[ctypes.c_char] | None = None profile_created = False + package_acl_granted = False try: hr = win.userenv.CreateAppContainerProfile( profile_name, @@ -446,6 +606,9 @@ def create_suspended(self, plan: NativeWorkerLaunchPlan) -> NativeSuspendedWorke if hr not in (0, 0x800700B7): _raise("native_profile_create_failed") profile_created = True + package_root = plan.executable.parent + _grant_package_read_execute(win, package_root, profile_sid) + package_acl_granted = True required = ctypes.c_size_t() win.kernel32.InitializeProcThreadAttributeList(None, 1, 0, ctypes.byref(required)) if not required.value: @@ -508,6 +671,8 @@ def create_suspended(self, plan: NativeWorkerLaunchPlan) -> NativeSuspendedWorke thread=process_info.thread, process_id=int(process_info.process_id), app_container_sid=sid, + package_root=package_root, + profile_sid=profile_sid, ) except Exception: if attribute_list is not None: @@ -518,6 +683,11 @@ def create_suspended(self, plan: NativeWorkerLaunchPlan) -> NativeSuspendedWorke win.kernel32.TerminateProcess(process_info.process, 1) win.kernel32.WaitForSingleObject(process_info.process, 3000) _close(win, process_info.process) + if package_acl_granted: + try: + _revoke_package_read_execute(win, plan.executable.parent, profile_sid) + except Exception: + pass if profile_created: win.userenv.DeleteAppContainerProfile(profile_name) if isinstance(sys.exc_info()[1], NativeLauncherError): diff --git a/backend/cortex_backend/execution/recipe_provider.py b/backend/cortex_backend/execution/recipe_provider.py index 7e2574a..9b065a6 100644 --- a/backend/cortex_backend/execution/recipe_provider.py +++ b/backend/cortex_backend/execution/recipe_provider.py @@ -14,6 +14,7 @@ from dataclasses import dataclass from hashlib import sha256 from io import BytesIO +from importlib import import_module import math import re from threading import RLock @@ -140,8 +141,16 @@ def _pillow_health() -> tuple[bool, str, str]: if Image is None or ImageEnhance is None or ImageFile is None or PIL is None: return False, "recipe_dependency_missing", "The image recipe dependency is unavailable." try: - Image.init() - supported = set(Image.registered_extensions().values()) + # Avoid Pillow's global ``Image.init``: it imports every optional codec + # (including heavyweight/stub integrations) and can block indefinitely + # inside the low-capability worker. Load only the fixed formats. + for plugin in ("PngImagePlugin", "JpegImagePlugin", "WebPImagePlugin"): + import_module(f"PIL.{plugin}") + # The explicit imports populate the extension table. Mark Pillow's + # plugin registry initialized so Image.open() cannot fall back to its + # broad optional-plugin scan during the first request. + supported = set(Image.EXTENSION.values()) + Image._initialized = 2 except Exception: return False, "recipe_codec_unavailable", "The required image codecs are unavailable." if not {"PNG", "JPEG", "WEBP"}.issubset(supported): diff --git a/backend/cortex_backend/execution/worker_runtime.py b/backend/cortex_backend/execution/worker_runtime.py index e8eb6bb..094263b 100644 --- a/backend/cortex_backend/execution/worker_runtime.py +++ b/backend/cortex_backend/execution/worker_runtime.py @@ -254,18 +254,22 @@ def run(self) -> WorkerRuntimeReport: if not isinstance(health, RuntimeHealth) or not health.available: raise WorkerRuntimeError("provider_unavailable") dispatcher = RecipeWorkerDispatcher(RecipeWorkerSession(provider)) - receiver = Thread( - target=self._receive_loop, - args=(self._connection, incoming, receiver_stop), - name="cortex-worker-broker-reader", - daemon=True, - ) - receiver.start() pending_collect: BrokerMessage | None = None cancel_acknowledged = False active_request: BrokerMessage | None = None transform_started_at = 0.0 + def start_receiver() -> None: + nonlocal receiver + if receiver is None: + receiver = Thread( + target=self._receive_loop, + args=(self._connection, incoming, receiver_stop), + name="cortex-worker-broker-reader", + daemon=True, + ) + receiver.start() + while processed < self._max_messages: if ( active_request is not None @@ -306,6 +310,11 @@ def run(self) -> WorkerRuntimeReport: ) terminal = "failed" return WorkerRuntimeReport(processed, terminal) + if receiver is None: + try: + incoming.put(self._connection.receive_message()) + except Exception: + incoming.put(WorkerRuntimeError("broker_receive_failed")) try: request = incoming.get(timeout=0.05) except Empty: @@ -398,6 +407,7 @@ def run(self) -> WorkerRuntimeReport: self._validate_envelope(request) active_request = request transform_started_at = monotonic() + start_receiver() completion_thread = Thread( target=self._dispatch_completion, args=(dispatcher, request, completion), diff --git a/docs/adr/0001-phase2-evidence.md b/docs/adr/0001-phase2-evidence.md index 2ac5c8b..737e21f 100644 --- a/docs/adr/0001-phase2-evidence.md +++ b/docs/adr/0001-phase2-evidence.md @@ -1,7 +1,7 @@ # ADR-0001 Phase 2 evidence log - **Phase:** 2 — signed image recipes and calculator/check primitives -- **Status:** Typed contract, signed-manifest verification, native broker transport, authenticated worker loop, signed bundle installation, trusted artifact boundary, and qualification-only provider core complete; OS sandbox/provider and release gates remain open +- **Status:** Typed contract, signed-manifest verification, native broker transport, authenticated worker loop, signed bundle installation, trusted artifact boundary, signed worker launch/broker qualification, and qualification-only provider core complete; the packaged provider transform and OS sandbox release gates remain open - **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) @@ -24,9 +24,9 @@ | 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 (factory + binder + disposable control spike)** | `NativeWin32ProcessFactory` creates a suspended zero-capability AppContainer child and verifies Job Object policy before resume. `NativeBrokerIdentityBinder` pins the live server to the worker PID/AppContainer SID and launcher cleanup closes it on failure. The fixed qualification helper remains separate evidence. | -| OS sandbox provider and provider-produced image outputs | **Blocked / release gate** | The package/protocol/launch boundary and worker-side broker loop are qualified, but the actual provider worker still needs a signed installed generation, end-to-end authenticated input/output through the suspended process, watchdog, hostile decoder execution inside the sandbox, external review, and lifecycle wiring. | +| Windows recipe sandbox qualification harness | **Complete (signed launch/broker; provider transform blocked)** | `recipe_worker_e2e_qualification.py` signs a disposable package with an in-memory key, installs/verifies one immutable generation, binds the live AppContainer identity to the broker, and exercises the packaged protocol. The fixed corpus passes through `input_chunk`; the provider `input_complete` transform times out and the harness fails closed. | +| Suspended native launcher/resource policy | **Complete (factory + binder + ACL cleanup + qualification evidence)** | `NativeWin32ProcessFactory` grants only inherited read/execute access to the fresh AppContainer SID on the verified package root, applies and verifies Job Object policy before resume, and removes the per-launch ACE during cleanup. `NativeBrokerIdentityBinder` pins the live server to the worker PID/AppContainer SID and launcher cleanup closes it on failure. | +| OS sandbox provider and provider-produced image outputs | **Blocked / release gate** | Signed installation, provenance, AppContainer/job identity, broker handshake, `prepare`, and `input_chunk` are live-qualified. The actual provider transform still needs to complete inside the packaged worker, followed by watchdog/hostile-decoder evidence, external review, and lifecycle wiring. | ## Security invariants @@ -90,6 +90,9 @@ 24. Release signing reads an external raw private key only for the signing operation, self-verifies the canonical manifest, rejects reparse/hardlink/mutable package inputs, and never treats a generated manifest as launch authorization. +25. The native launcher grants only a per-profile inherited read/execute ACE on the + verified package root, removes that ACE during worker cleanup, and never grants + package write/delete access; a qualification timeout is always a blocked result. ## Re-run target @@ -107,6 +110,7 @@ python -m pytest tests/test_native_launcher_qualification.py -q python -m pytest tests/test_recipe_sandbox_qualification.py -q python tools/execution_spikes/native_launcher_qualification.py python tools/execution_spikes/recipe_sandbox_qualification.py --json --strict +python tools/execution_spikes/recipe_worker_e2e_qualification.py --json --strict python -m compileall -q backend\cortex_backend\execution tests python -m pytest -q python tools/generate_contracts.py @@ -131,3 +135,13 @@ fail-closed `blocked` status because the signed worker bundle is not shipped. Th Windows PyInstaller package built successfully, and an external-key smoke signed and verified its complete 822-file closure (one `image_transform` role plus 821 inert `resource` entries); no key or signed artifact was retained. + +**Signed worker qualification result (2026-07-24):** The disposable +`recipe_worker_e2e_qualification.py --json --timeout-seconds 5` run passed +ephemeral signing/installation, active provenance verification, AppContainer/job +policy and identity binding, authenticated broker handshake, `prepare`, and +`input_chunk`. It blocked at `input_complete` with `worker_response_timeout` while +the packaged provider transform remained inside the worker. Bounded cleanup closed +the broker, binder, worker, and profile; no `recipe_worker.exe` process remained. +This is evidence for the launch/protocol boundary only and does not close the +provider or production lifecycle release gate. diff --git a/packaging/recipe_worker/recipe_worker.spec b/packaging/recipe_worker/recipe_worker.spec index 5892429..9300b8c 100644 --- a/packaging/recipe_worker/recipe_worker.spec +++ b/packaging/recipe_worker/recipe_worker.spec @@ -19,6 +19,9 @@ a = Analysis( "PIL.Image", "PIL.ImageEnhance", "PIL.ImageFile", + "PIL.PngImagePlugin", + "PIL.JpegImagePlugin", + "PIL.WebPImagePlugin", ], hookspath=[], hooksconfig={}, diff --git a/tests/test_phase2_native_win32.py b/tests/test_phase2_native_win32.py index d65826b..c3276d8 100644 --- a/tests/test_phase2_native_win32.py +++ b/tests/test_phase2_native_win32.py @@ -4,7 +4,9 @@ import os from pathlib import Path +import shutil import subprocess +from tempfile import TemporaryDirectory import pytest @@ -23,6 +25,14 @@ def test_win32_factory_creates_suspended_appcontainer_and_job_policy(): executable = system_root / "System32" / "findstr.exe" if not executable.is_file(): pytest.skip("findstr.exe is unavailable on this Windows host") + with TemporaryDirectory(prefix="cortex-native-factory-") as temp_root: + package_root = Path(temp_root) + test_executable = package_root / "findstr.exe" + shutil.copy2(executable, test_executable) + _run_factory_probe(test_executable) + + +def _run_factory_probe(executable: Path) -> None: binding = BrokerWorkerBinding( pipe_name=r"\\.\pipe\cortex-worker-factory-test", broker_process_id=os.getpid(), @@ -31,7 +41,7 @@ def test_win32_factory_creates_suspended_appcontainer_and_job_policy(): ) plan = NativeWorkerLaunchPlan( worker=VerifiedRecipeWorker( - bundle_root=system_root, + bundle_root=executable.parent, bundle_digest="0" * 64, key_id="release-1", worker_path="recipe_worker.exe", diff --git a/tests/test_phase2_recipe_provider.py b/tests/test_phase2_recipe_provider.py index a9d9bfc..88db8e0 100644 --- a/tests/test_phase2_recipe_provider.py +++ b/tests/test_phase2_recipe_provider.py @@ -205,13 +205,36 @@ def test_provider_health_blocks_missing_dependency_or_codec(monkeypatch): assert not provider.enabled monkeypatch.undo() - monkeypatch.setattr(provider_module.Image, "registered_extensions", lambda: {".png": "PNG"}) + monkeypatch.setattr(provider_module.Image, "EXTENSION", {".png": "PNG"}) unavailable = provider.start(RuntimeHealth.ready("test sandbox attestation")) assert not unavailable.available assert unavailable.code == "recipe_codec_unavailable" assert not provider.enabled +def test_provider_health_loads_only_allowlisted_codec_plugins(monkeypatch): + imported: list[str] = [] + + def import_plugin(name: str): + imported.append(name) + return object() + + monkeypatch.setattr(provider_module, "import_module", import_plugin) + monkeypatch.setattr( + provider_module.Image, + "EXTENSION", + {".png": "PNG", ".jpg": "JPEG", ".webp": "WEBP"}, + ) + available, code, _message = provider_module._pillow_health() + assert available is True + assert code == "ready" + assert imported == [ + "PIL.PngImagePlugin", + "PIL.JpegImagePlugin", + "PIL.WebPImagePlugin", + ] + + def test_provider_rejects_input_before_decoder_when_encoded_limit_is_exceeded(): provider = _started_provider(RecipeProviderLimits(max_input_bytes=8)) with pytest.raises(RecipeProviderError) as error: diff --git a/tools/execution_spikes/README.md b/tools/execution_spikes/README.md index 4b3fa44..37845bc 100644 --- a/tools/execution_spikes/README.md +++ b/tools/execution_spikes/README.md @@ -62,6 +62,29 @@ unsigned and must not be installed or launched as a provider. The package and runtime contracts are covered by `tests/test_phase2_worker_protocol.py` and `tests/test_phase2_worker_runtime.py`. +The signed worker/AppContainer/broker qualification gate is a separate, +disposable end-to-end check: + +```powershell +python tools/execution_spikes/recipe_worker_e2e_qualification.py --json +python tools/execution_spikes/recipe_worker_e2e_qualification.py --json --strict +``` + +It creates an in-memory ephemeral Ed25519 trust root, signs the already-built +one-folder package, installs one immutable generation, verifies provenance, +launches the worker through the native AppContainer/job-policy factory, and +exercises only the fixed 4x3 PNG grayscale corpus over the authenticated broker. +It accepts no user files, model text, commands, or production trust material. +`--strict` returns exit code `2` unless every stage passes. Full package closure +verification can take several minutes on Windows; the protocol timeout applies +after launch and is fail-closed. + +The current evidence result is intentionally blocked at the provider transform: +signed installation, provenance, AppContainer/job identity binding, broker +handshake, `prepare`, and `input_chunk` pass, while `input_complete` times out +inside the packaged provider. The harness reports `worker_response_timeout`, +cleans up boundedly, and never reports a green result for this partial run. + ## What the probes prove - `environment`: supported Windows host and interpreter metadata. @@ -92,6 +115,10 @@ runtime contracts are covered by `tests/test_phase2_worker_protocol.py` and - `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. +- `recipe_worker_e2e_qualification`: signs and installs a disposable worker + generation, proves the live native broker identity boundary, and runs the fixed + packaged-worker protocol corpus; any provider or cleanup timeout remains + blocking. - `backend/cortex_backend/execution/native_launcher.py`: production-facing launch-plan boundary that revalidates the signed worker and refuses process creation until a reviewed native process factory and live broker binder are diff --git a/tools/execution_spikes/recipe_worker_e2e_qualification.py b/tools/execution_spikes/recipe_worker_e2e_qualification.py new file mode 100644 index 0000000..ceb61b5 --- /dev/null +++ b/tools/execution_spikes/recipe_worker_e2e_qualification.py @@ -0,0 +1,406 @@ +"""Disposable signed-worker/AppContainer/broker qualification gate. + +This probe is deliberately separate from Cortex runtime wiring. It signs the +already-built fixed worker with an in-memory ephemeral Ed25519 key, installs it +into a disposable store, launches it through the reviewed AppContainer factory, +and exercises only the fixed PNG grayscale protocol corpus. No user files, +model text, commands, paths, or production trust roots are accepted. +""" + +from __future__ import annotations + +import argparse +import base64 +import ctypes +from hashlib import sha256 +from io import BytesIO +import json +import os +from pathlib import Path +import shutil +import sys +import subprocess +from ctypes import wintypes +from queue import Queue +from tempfile import mkdtemp +from threading import Thread +from typing import Any +from uuid import uuid4 + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "backend")) + +from cryptography.hazmat.primitives import serialization # noqa: E402 +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey # noqa: E402 +from PIL import Image # noqa: E402 + +from cortex_backend.execution.broker import BrokerMessage # noqa: E402 +from cortex_backend.execution.bundle_installer import SignedBundleInstaller # noqa: E402 +from cortex_backend.execution.manifest import TrustedRecipeKeys # noqa: E402 +from cortex_backend.execution.native_broker import NativeBrokerConnection # noqa: E402 +from cortex_backend.execution.native_launcher import ( # noqa: E402 + BrokerWorkerBinding, + NativeBrokerIdentityBinder, + NativeWorkerLauncher, +) +from cortex_backend.execution.native_win32 import NativeWin32ProcessFactory # noqa: E402 +from cortex_backend.execution.recipes import parse_image_transform # noqa: E402 +from cortex_backend.execution.worker_protocol import ( # noqa: E402 + WorkerCollect, + WorkerInputChunk, + WorkerInputComplete, + WorkerPrepare, +) +from cortex_backend.execution.worker_provenance import verify_active_worker # noqa: E402 +from cortex_backend.execution.worker_release import build_signed_worker_manifest # noqa: E402 + + +DEFAULT_TIMEOUT_SECONDS = 20.0 +_TOKEN_QUERY = 0x0008 +_TOKEN_USER = 1 +_PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 + + +class _TokenUser(ctypes.Structure): + _fields_ = [("sid", ctypes.c_void_p), ("attributes", wintypes.DWORD)] + + +def _current_user_sid() -> str: + if os.name != "nt": + raise RuntimeError("windows_required") + kernel32 = ctypes.WinDLL("kernel32.dll", use_last_error=True) + advapi32 = ctypes.WinDLL("advapi32.dll", use_last_error=True) + kernel32.GetCurrentProcess.restype = wintypes.HANDLE + advapi32.OpenProcessToken.argtypes = [ + wintypes.HANDLE, + wintypes.DWORD, + ctypes.POINTER(wintypes.HANDLE), + ] + advapi32.OpenProcessToken.restype = wintypes.BOOL + advapi32.GetTokenInformation.argtypes = [ + wintypes.HANDLE, + wintypes.INT, + ctypes.c_void_p, + wintypes.DWORD, + ctypes.POINTER(wintypes.DWORD), + ] + advapi32.GetTokenInformation.restype = wintypes.BOOL + advapi32.ConvertSidToStringSidW.argtypes = [ + ctypes.c_void_p, + ctypes.POINTER(wintypes.LPWSTR), + ] + advapi32.ConvertSidToStringSidW.restype = wintypes.BOOL + kernel32.CloseHandle.argtypes = [wintypes.HANDLE] + kernel32.LocalFree.argtypes = [wintypes.HLOCAL] + token = wintypes.HANDLE() + if not advapi32.OpenProcessToken(kernel32.GetCurrentProcess(), _TOKEN_QUERY, ctypes.byref(token)): + raise RuntimeError("user_sid_unavailable") + try: + required = wintypes.DWORD() + advapi32.GetTokenInformation(token, _TOKEN_USER, None, 0, ctypes.byref(required)) + if not required.value: + raise RuntimeError("user_sid_unavailable") + buffer = ctypes.create_string_buffer(required.value) + if not advapi32.GetTokenInformation( + token, _TOKEN_USER, buffer, required, ctypes.byref(required) + ): + raise RuntimeError("user_sid_unavailable") + user = _TokenUser.from_buffer_copy(buffer) + sid_text = wintypes.LPWSTR() + if not advapi32.ConvertSidToStringSidW(user.sid, ctypes.byref(sid_text)): + raise RuntimeError("user_sid_unavailable") + try: + if not sid_text.value: + raise RuntimeError("user_sid_unavailable") + return sid_text.value + finally: + kernel32.LocalFree(sid_text) + finally: + kernel32.CloseHandle(token) + + +def _process_executable(process_id: int) -> Path | None: + """Return a process image path without adding a host dependency.""" + + if os.name != "nt": + return None + kernel32 = ctypes.WinDLL("kernel32.dll", use_last_error=True) + kernel32.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD] + kernel32.OpenProcess.restype = wintypes.HANDLE + kernel32.QueryFullProcessImageNameW.argtypes = [ + wintypes.HANDLE, + wintypes.DWORD, + wintypes.LPWSTR, + ctypes.POINTER(wintypes.DWORD), + ] + kernel32.QueryFullProcessImageNameW.restype = wintypes.BOOL + kernel32.CloseHandle.argtypes = [wintypes.HANDLE] + process = kernel32.OpenProcess(_PROCESS_QUERY_LIMITED_INFORMATION, False, process_id) + if not process: + return None + try: + buffer = ctypes.create_unicode_buffer(32768) + size = wintypes.DWORD(len(buffer)) + if not kernel32.QueryFullProcessImageNameW(process, 0, buffer, ctypes.byref(size)): + return None + return Path(buffer.value).resolve() + finally: + kernel32.CloseHandle(process) + + +def _fixed_png() -> 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 _receive_with_timeout( + connection: NativeBrokerConnection, + timeout_seconds: float, +) -> BrokerMessage: + result: Queue[object] = Queue(maxsize=1) + + def receive() -> None: + try: + result.put(connection.receive_message()) + except Exception as error: # pragma: no cover - native failure path. + result.put(error) + + Thread(target=receive, name="cortex-qualification-reader", daemon=True).start() + try: + value = result.get(timeout=timeout_seconds) + except Exception: + raise TimeoutError("broker response timeout") from None + if isinstance(value, Exception): + raise value + return value + + +def _message(operation: str, model: Any, *, principal: str, job_id: str) -> BrokerMessage: + return BrokerMessage( + schema_version="broker.message.v1", + direction="to_executor", + operation=operation, + request_id=model.request_id, + job_id=job_id, + installation_principal_id=principal, + body=model.model_dump(mode="json"), + ) + + +def _response_code(message: BrokerMessage, operation: str) -> str: + if message.operation != operation or message.body.get("request_id") != message.request_id: + raise ValueError("broker response identity mismatch") + return str(message.body.get("schema_version", "")) + + +def _bounded_cleanup(action: Any, timeout_seconds: float = 5.0) -> None: + finished = Queue(maxsize=1) + + def run() -> None: + try: + action() + finally: + finished.put(True) + + thread = Thread(target=run, name="cortex-qualification-cleanup", daemon=True) + thread.start() + try: + finished.get(timeout=timeout_seconds) + except Exception: + pass + + +def _install_ephemeral(source_root: Path, store_root: Path) -> SignedBundleInstaller: + signer = Ed25519PrivateKey.generate() + private_bytes = signer.private_bytes( + serialization.Encoding.Raw, + serialization.PrivateFormat.Raw, + serialization.NoEncryption(), + ) + public_bytes = signer.public_key().public_bytes( + serialization.Encoding.Raw, + serialization.PublicFormat.Raw, + ) + release = build_signed_worker_manifest( + source_root, + private_key_bytes=private_bytes, + key_id="qualification-ephemeral", + bundle_version="1.0.0", + sequence=1, + ) + installer = SignedBundleInstaller( + store_root, + TrustedRecipeKeys({"qualification-ephemeral": public_bytes}), + ) + installer.install(release.manifest, source_root) + return installer + + +def qualify(source_root: Path, *, timeout_seconds: float) -> dict[str, Any]: + if not source_root.is_dir(): + return {"status": "blocked", "code": "package_missing", "stages": []} + workspace = Path(mkdtemp(prefix="cortex-worker-e2e-qualification-")) + stages: list[str] = [] + connection: NativeBrokerConnection | None = None + worker = None + binder: NativeBrokerIdentityBinder | None = None + try: + installer = _install_ephemeral(source_root, workspace / "store") + stages.extend(["signed_ephemeral_manifest", "installed_immutable_generation"]) + verify_active_worker(installer) + stages.append("provenance_verified") + principal = sha256(os.urandom(32)).hexdigest() + job_id = f"qualification-{uuid4().hex[:12]}" + binding = BrokerWorkerBinding( + pipe_name=rf"\\.\pipe\cortex-qualification-{uuid4().hex}", + broker_process_id=os.getpid(), + installation_principal_id=principal, + job_id=job_id, + ) + binder = NativeBrokerIdentityBinder(allowed_user_sids=frozenset({_current_user_sid()})) + launcher = NativeWorkerLauncher( + installer, + process_factory=NativeWin32ProcessFactory(), + broker_binder=binder, + ) + worker = launcher.launch(binding) + stages.append("appcontainer_job_policy_and_identity_bound") + accepted: Queue[object] = Queue(maxsize=1) + + def accept() -> None: + try: + accepted.put( + binder.accept( + owner_for_job=lambda value: principal if value == job_id else None + ) + ) + except Exception as error: # pragma: no cover - native failure path. + accepted.put(error) + + Thread(target=accept, name="cortex-qualification-accept", daemon=True).start() + try: + value = accepted.get(timeout=timeout_seconds) + except Exception: + raise TimeoutError("broker handshake timeout") from None + if isinstance(value, Exception): + raise value + connection = value + stages.append("authenticated_broker_handshake") + + content = _fixed_png() + request_id = "qualification-request" + plan = parse_image_transform( + { + "schema_version": "artifact.transform.v1", + "input_artifact_id": "qualification-artifact", + "steps": [{"op": "grayscale"}], + "output_format": "png", + } + ) + prepare = WorkerPrepare( + schema_version="recipe.worker.prepare.v1", + request_id=request_id, + job_id=job_id, + plan=plan, + input_size=len(content), + input_sha256=sha256(content).hexdigest(), + input_mime_type="image/png", + ) + connection.send_message(_message("prepare", prepare, principal=principal, job_id=job_id)) + response = _receive_with_timeout(connection, timeout_seconds) + if _response_code(response, "prepare") != "recipe.worker.ack.v1": + raise ValueError("prepare acknowledgement invalid") + stages.append("prepare_ack") + + chunk = WorkerInputChunk( + schema_version="recipe.worker.input_chunk.v1", + request_id=request_id, + job_id=job_id, + offset=0, + data=base64.urlsafe_b64encode(content).decode("ascii").rstrip("="), + sha256=sha256(content).hexdigest(), + ) + connection.send_message(_message("input_chunk", chunk, principal=principal, job_id=job_id)) + response = _receive_with_timeout(connection, timeout_seconds) + if _response_code(response, "input_chunk") != "recipe.worker.ack.v1": + raise ValueError("input acknowledgement invalid") + stages.append("input_chunk_ack") + + complete = WorkerInputComplete( + schema_version="recipe.worker.input_complete.v1", + request_id=request_id, + job_id=job_id, + input_size=len(content), + input_sha256=sha256(content).hexdigest(), + ) + connection.send_message(_message("input_complete", complete, principal=principal, job_id=job_id)) + response = _receive_with_timeout(connection, timeout_seconds) + if response.operation != "input_complete": + raise ValueError("input completion response invalid") + stages.append("input_complete_result") + + collect = WorkerCollect( + schema_version="recipe.worker.collect.v1", + request_id=request_id, + job_id=job_id, + offset=0, + max_bytes=48 * 1024, + ) + connection.send_message(_message("collect", collect, principal=principal, job_id=job_id)) + response = _receive_with_timeout(connection, timeout_seconds) + if _response_code(response, "collect") != "recipe.worker.output_chunk.v1": + raise ValueError("output response invalid") + stages.append("collect_output") + return {"status": "passed", "stages": stages} + except TimeoutError: + return {"status": "blocked", "code": "worker_response_timeout", "stages": stages} + except Exception: + return {"status": "blocked", "code": "qualification_failed_closed", "stages": stages} + finally: + if connection is not None: + _bounded_cleanup(connection.close) + if binder is not None: + _bounded_cleanup(binder.close_binding) + if worker is not None: + _bounded_cleanup(worker.close) + try: + executable = _process_executable(worker.process_id) + if executable is not None and executable.is_relative_to(workspace.resolve()): + subprocess.run( + ["taskkill", "/PID", str(worker.process_id), "/T", "/F"], + check=False, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + except Exception: + pass + try: + shutil.rmtree(workspace) + except OSError: + # The disposable AppContainer generation is never reused; cleanup + # failure remains visible in the local process but is not a pass. + pass + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--source-root", type=Path, default=ROOT / "dist" / "recipe-runtime") + parser.add_argument("--timeout-seconds", type=float, default=DEFAULT_TIMEOUT_SECONDS) + parser.add_argument("--json", action="store_true") + parser.add_argument("--strict", action="store_true") + args = parser.parse_args(argv) + if not 1 <= args.timeout_seconds <= 120: + parser.error("--timeout-seconds must be between 1 and 120") + result = qualify(args.source_root.resolve(), timeout_seconds=args.timeout_seconds) + print(json.dumps(result, sort_keys=True, separators=(",", ":") if args.json else None)) + return 2 if args.strict and result["status"] != "passed" else 0 + + +if __name__ == "__main__": + raise SystemExit(main())