Qualify signed worker launch and broker boundary

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):

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):

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),

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.

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={},