diff --git a/backend/cortex_backend/execution/__init__.py b/backend/cortex_backend/execution/__init__.py index 577e7ff..91a9dec 100644 --- a/backend/cortex_backend/execution/__init__.py +++ b/backend/cortex_backend/execution/__init__.py @@ -70,6 +70,7 @@ from .native_launcher import ( BrokerWorkerBinding, BrokerWorkerBinder, + NativeBrokerIdentityBinder, NativeLauncherError, NativeProcessFactory, NativeSuspendedWorker, @@ -77,6 +78,11 @@ NativeWorkerLauncher, NativeWorkerPolicy, ) +from .native_win32 import ( + NativeWin32Error, + NativeWin32ProcessFactory, + Win32SuspendedWorker, +) from .worker_provenance import ( EXPECTED_WORKER_PATH, VerifiedRecipeWorker, @@ -149,12 +155,16 @@ "NativeBrokerServerConfig", "BrokerWorkerBinding", "BrokerWorkerBinder", + "NativeBrokerIdentityBinder", "NativeLauncherError", "NativeProcessFactory", "NativeSuspendedWorker", "NativeWorkerLaunchPlan", "NativeWorkerLauncher", "NativeWorkerPolicy", + "NativeWin32Error", + "NativeWin32ProcessFactory", + "Win32SuspendedWorker", "PrimitiveEvaluationError", "OutputClaim", "PublishedArtifact", diff --git a/backend/cortex_backend/execution/native_launcher.py b/backend/cortex_backend/execution/native_launcher.py index 6d9f668..49d6302 100644 --- a/backend/cortex_backend/execution/native_launcher.py +++ b/backend/cortex_backend/execution/native_launcher.py @@ -9,6 +9,7 @@ from __future__ import annotations from dataclasses import dataclass +from collections.abc import Callable from hashlib import sha256 import os from pathlib import Path @@ -16,7 +17,13 @@ import subprocess from typing import Final, Protocol +from .broker import BrokerAclPolicy, BrokerPeerPolicy from .bundle_installer import SignedBundleInstaller +from .native_broker import ( + NativeBrokerConnection, + NativeBrokerServer, + NativeBrokerServerConfig, +) from .worker_provenance import ( EXPECTED_WORKER_PATH, VerifiedRecipeWorker, @@ -28,6 +35,8 @@ _SAFE_CODE = re.compile(r"^[a-z][a-z0-9_]{0,63}$") _SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$") _PRINCIPAL = re.compile(r"^[0-9a-f]{64}$") +_SID = re.compile(r"^S-1-[0-9]+(?:-[0-9]+)+$") +_APPCONTAINER_SID = re.compile(r"^S-1-15-2-(?:[0-9]+-)+[0-9]+$") _PIPE = re.compile(r"^\\\\\.\\pipe\\cortex-[A-Za-z0-9._-]{1,200}$") _MAX_WORKER_BYTES = 128 * 1024 * 1024 _JOB_OBJECT_LIMIT_PROCESS_TIME = 0x00000002 @@ -174,6 +183,97 @@ def bind_worker( ) -> None: """Bind expected PID, token identity, principal, and job ownership.""" + def close_binding(self) -> None: + """Close the per-worker broker endpoint during cancellation or failure.""" + + +class NativeBrokerIdentityBinder: + """Bind one live named-pipe server to a suspended worker identity.""" + + def __init__( + self, + *, + allowed_user_sids: frozenset[str], + server_factory: Callable[[NativeBrokerServerConfig], NativeBrokerServer] = NativeBrokerServer, + ) -> None: + if not allowed_user_sids or any(_SID.fullmatch(sid) is None for sid in allowed_user_sids): + raise ValueError("allowed user SIDs are invalid") + self._allowed_user_sids = frozenset(allowed_user_sids) + self._server_factory = server_factory + self._server: NativeBrokerServer | None = None + self._binding: BrokerWorkerBinding | None = None + + @property + def bound(self) -> bool: + return self._server is not None and self._binding is not None + + @property + def binding(self) -> BrokerWorkerBinding | None: + return self._binding + + def bind_worker( + self, + *, + process_id: int, + app_container_sid: str, + binding: BrokerWorkerBinding, + ) -> None: + if self.bound: + raise NativeLauncherError("native_broker_already_bound") + if type(process_id) is not int or process_id <= 0: + raise NativeLauncherError("native_worker_identity_invalid") + if _APPCONTAINER_SID.fullmatch(app_container_sid) is None: + raise NativeLauncherError("native_appcontainer_sid_invalid") + if binding.broker_process_id != os.getpid(): + raise NativeLauncherError("native_broker_process_mismatch") + acl = BrokerAclPolicy( + allowed_user_sids=self._allowed_user_sids, + allowed_app_container_sids=frozenset({app_container_sid}), + ) + peer_policy = BrokerPeerPolicy( + acl=acl, + expected_process_id=process_id, + maximum_integrity="low", + ) + config = NativeBrokerServerConfig(pipe_name=binding.pipe_name, peer_policy=peer_policy) + try: + server = self._server_factory(config) + server.open() + except NativeLauncherError: + raise + except Exception: + try: + if "server" in locals(): + server.close() + except Exception: + pass + raise NativeLauncherError("native_broker_open_failed") from None + self._server = server + self._binding = binding + + def accept( + self, + *, + owner_for_job: Callable[[str], str | None], + ) -> NativeBrokerConnection: + if self._server is None or self._binding is None: + raise NativeLauncherError("native_broker_not_bound") + try: + return self._server.accept( + expected_principal_id=self._binding.installation_principal_id, + owner_for_job=owner_for_job, + ) + except NativeLauncherError: + raise + except Exception: + raise NativeLauncherError("native_broker_accept_failed") from None + + def close_binding(self) -> None: + server, self._server = self._server, None + self._binding = None + if server is not None: + server.close() + def _revalidate_worker(worker: VerifiedRecipeWorker) -> Path: root = worker.bundle_root @@ -291,6 +391,7 @@ def launch( raise NativeLauncherError("native_process_factory_required") plan = self.prepare(binding, policy) worker: NativeSuspendedWorker | None = None + broker_bound = False try: worker = self._process_factory.create_suspended(plan) if type(worker.process_id) is not int or worker.process_id <= 0: @@ -303,13 +404,22 @@ def launch( app_container_sid=worker.app_container_sid, binding=plan.broker, ) + broker_bound = True worker.resume() return worker except NativeLauncherError: + if broker_bound: + close_binding = getattr(self._broker_binder, "close_binding", None) + if close_binding is not None: + close_binding() if worker is not None: worker.close() raise except Exception: + if broker_bound: + close_binding = getattr(self._broker_binder, "close_binding", None) + if close_binding is not None: + close_binding() if worker is not None: worker.close() raise NativeLauncherError("native_launch_failed") from None @@ -318,6 +428,7 @@ def launch( __all__ = [ "BrokerWorkerBinding", "BrokerWorkerBinder", + "NativeBrokerIdentityBinder", "NativeLauncherError", "NativeProcessFactory", "NativeSuspendedWorker", diff --git a/backend/cortex_backend/execution/native_win32.py b/backend/cortex_backend/execution/native_win32.py new file mode 100644 index 0000000..faa773b --- /dev/null +++ b/backend/cortex_backend/execution/native_win32.py @@ -0,0 +1,528 @@ +"""Reviewed Win32 process factory for the fixed recipe worker. + +The factory is intentionally opt-in: callers must inject it into +``NativeWorkerLauncher`` together with a live broker binder. It creates a worker +with a zero-capability AppContainer token suspended, applies and queries the Job +Object limits before resume, and kills/cleans every handle on failure. It never +accepts a path, command, environment, or shell value from a model or user. +""" + +from __future__ import annotations + +import ctypes +from ctypes import wintypes +from dataclasses import dataclass +import sys +from typing import Any +from uuid import uuid4 + +from .native_launcher import ( + NativeLauncherError, + NativeSuspendedWorker, + NativeWorkerLaunchPlan, + NativeWorkerPolicy, +) + + +_EXTENDED_STARTUPINFO_PRESENT = 0x00080000 +_CREATE_SUSPENDED = 0x00000004 +_CREATE_UNICODE_ENVIRONMENT = 0x00000400 +_CREATE_NO_WINDOW = 0x08000000 +_PROC_THREAD_ATTRIBUTE_SECURITY_CAPABILITIES = 0x00020009 +_JOB_OBJECT_EXTENDED_LIMIT_INFORMATION = 9 +_JOB_OBJECT_LIMIT_BREAKAWAY_OK = 0x00000800 +_JOB_OBJECT_LIMIT_SILENT_BREAKAWAY_OK = 0x00001000 +_TOKEN_IS_APPCONTAINER = 29 +_TOKEN_APPCONTAINER_SID = 31 +_TOKEN_QUERY = 0x0008 +_WAIT_OBJECT_0 = 0 +_WAIT_TIMEOUT = 0x102 +_INFINITE = 0xFFFFFFFF + + +class NativeWin32Error(NativeLauncherError): + """Stable native adapter failure with no OS details exposed.""" + + +class _LargeInteger(ctypes.Union): + _fields_ = [("quad_part", ctypes.c_longlong)] + + +class _IoCounters(ctypes.Structure): + _fields_ = [ + ("read_operations", ctypes.c_ulonglong), + ("write_operations", ctypes.c_ulonglong), + ("other_operations", ctypes.c_ulonglong), + ("read_transfers", ctypes.c_ulonglong), + ("write_transfers", ctypes.c_ulonglong), + ("other_transfers", ctypes.c_ulonglong), + ] + + +class _BasicLimitInformation(ctypes.Structure): + _fields_ = [ + ("per_process_user_time", _LargeInteger), + ("per_job_user_time", _LargeInteger), + ("limit_flags", wintypes.DWORD), + ("minimum_working_set_size", ctypes.c_size_t), + ("maximum_working_set_size", ctypes.c_size_t), + ("active_process_limit", wintypes.DWORD), + ("affinity", ctypes.c_size_t), + ("priority_class", wintypes.DWORD), + ("scheduling_class", wintypes.DWORD), + ] + + +class _ExtendedLimitInformation(ctypes.Structure): + _fields_ = [ + ("basic_limit_information", _BasicLimitInformation), + ("io_info", _IoCounters), + ("process_memory_limit", ctypes.c_size_t), + ("job_memory_limit", ctypes.c_size_t), + ("peak_process_memory_used", ctypes.c_size_t), + ("peak_job_memory_used", ctypes.c_size_t), + ] + + +class _StartupInfo(ctypes.Structure): + _fields_ = [ + ("cb", wintypes.DWORD), + ("reserved", wintypes.LPWSTR), + ("desktop", wintypes.LPWSTR), + ("title", wintypes.LPWSTR), + ("x", wintypes.DWORD), + ("y", wintypes.DWORD), + ("x_size", wintypes.DWORD), + ("y_size", wintypes.DWORD), + ("x_count_chars", wintypes.DWORD), + ("y_count_chars", wintypes.DWORD), + ("fill_attribute", wintypes.DWORD), + ("flags", wintypes.DWORD), + ("show_window", wintypes.WORD), + ("reserved2", wintypes.WORD), + ("reserved2_ptr", ctypes.POINTER(ctypes.c_ubyte)), + ("std_input", wintypes.HANDLE), + ("std_output", wintypes.HANDLE), + ("std_error", wintypes.HANDLE), + ] + + +class _StartupInfoEx(ctypes.Structure): + _fields_ = [("startup_info", _StartupInfo), ("attribute_list", wintypes.LPVOID)] + + +class _ProcessInformation(ctypes.Structure): + _fields_ = [ + ("process", wintypes.HANDLE), + ("thread", wintypes.HANDLE), + ("process_id", wintypes.DWORD), + ("thread_id", wintypes.DWORD), + ] + + +class _SecurityCapabilities(ctypes.Structure): + _fields_ = [ + ("app_container_sid", wintypes.LPVOID), + ("capabilities", wintypes.LPVOID), + ("capability_count", wintypes.DWORD), + ("reserved", wintypes.DWORD), + ] + + +@dataclass(slots=True) +class _Win32: + kernel32: Any + userenv: Any + advapi32: Any + + +def _require_windows() -> None: + if sys.platform != "win32": + raise NativeWin32Error("native_windows_required") + + +def _invalid_handle(handle: Any) -> bool: + return not handle or handle == ctypes.c_void_p(-1).value + + +def _raise(code: str) -> None: + raise NativeWin32Error(code) + + +def _configure() -> _Win32: + _require_windows() + kernel32 = ctypes.WinDLL("kernel32.dll", use_last_error=True) + userenv = ctypes.WinDLL("userenv.dll", use_last_error=True) + advapi32 = ctypes.WinDLL("advapi32.dll", use_last_error=True) + + userenv.CreateAppContainerProfile.argtypes = [ + wintypes.LPCWSTR, + wintypes.LPCWSTR, + wintypes.LPCWSTR, + wintypes.LPVOID, + wintypes.DWORD, + ctypes.POINTER(wintypes.LPVOID), + ] + # HRESULT is a signed 32-bit value; wintypes.HRESULT is not exposed by all + # supported CPython Windows builds (notably Python 3.11 on CI). + userenv.CreateAppContainerProfile.restype = ctypes.c_long + userenv.DeleteAppContainerProfile.argtypes = [wintypes.LPCWSTR] + userenv.DeleteAppContainerProfile.restype = ctypes.c_long + advapi32.OpenProcessToken.argtypes = [ + wintypes.HANDLE, + wintypes.DWORD, + ctypes.POINTER(wintypes.HANDLE), + ] + advapi32.OpenProcessToken.restype = wintypes.BOOL + advapi32.GetTokenInformation.argtypes = [ + wintypes.HANDLE, + wintypes.INT, + wintypes.LPVOID, + wintypes.DWORD, + ctypes.POINTER(wintypes.DWORD), + ] + advapi32.GetTokenInformation.restype = wintypes.BOOL + advapi32.ConvertSidToStringSidW.argtypes = [ + wintypes.LPVOID, + ctypes.POINTER(wintypes.LPWSTR), + ] + advapi32.ConvertSidToStringSidW.restype = wintypes.BOOL + + kernel32.InitializeProcThreadAttributeList.argtypes = [ + wintypes.LPVOID, + wintypes.DWORD, + wintypes.DWORD, + ctypes.POINTER(ctypes.c_size_t), + ] + kernel32.InitializeProcThreadAttributeList.restype = wintypes.BOOL + kernel32.UpdateProcThreadAttribute.argtypes = [ + wintypes.LPVOID, + wintypes.DWORD, + ctypes.c_size_t, + wintypes.LPVOID, + ctypes.c_size_t, + wintypes.LPVOID, + ctypes.POINTER(ctypes.c_size_t), + ] + kernel32.UpdateProcThreadAttribute.restype = wintypes.BOOL + kernel32.DeleteProcThreadAttributeList.argtypes = [wintypes.LPVOID] + kernel32.DeleteProcThreadAttributeList.restype = None + kernel32.CreateProcessW.argtypes = [ + wintypes.LPCWSTR, + wintypes.LPWSTR, + wintypes.LPVOID, + wintypes.LPVOID, + wintypes.BOOL, + wintypes.DWORD, + wintypes.LPVOID, + wintypes.LPCWSTR, + wintypes.LPVOID, + ctypes.POINTER(_ProcessInformation), + ] + kernel32.CreateProcessW.restype = wintypes.BOOL + kernel32.CreateJobObjectW.argtypes = [wintypes.LPVOID, wintypes.LPCWSTR] + kernel32.CreateJobObjectW.restype = wintypes.HANDLE + kernel32.SetInformationJobObject.argtypes = [ + wintypes.HANDLE, + wintypes.INT, + wintypes.LPVOID, + wintypes.DWORD, + ] + kernel32.SetInformationJobObject.restype = wintypes.BOOL + kernel32.QueryInformationJobObject.argtypes = [ + wintypes.HANDLE, + wintypes.INT, + wintypes.LPVOID, + wintypes.DWORD, + ctypes.POINTER(wintypes.DWORD), + ] + kernel32.QueryInformationJobObject.restype = wintypes.BOOL + kernel32.AssignProcessToJobObject.argtypes = [wintypes.HANDLE, wintypes.HANDLE] + kernel32.AssignProcessToJobObject.restype = wintypes.BOOL + kernel32.ResumeThread.argtypes = [wintypes.HANDLE] + kernel32.ResumeThread.restype = wintypes.DWORD + kernel32.TerminateProcess.argtypes = [wintypes.HANDLE, wintypes.UINT] + kernel32.TerminateProcess.restype = wintypes.BOOL + kernel32.WaitForSingleObject.argtypes = [wintypes.HANDLE, wintypes.DWORD] + kernel32.WaitForSingleObject.restype = wintypes.DWORD + kernel32.CloseHandle.argtypes = [wintypes.HANDLE] + kernel32.CloseHandle.restype = wintypes.BOOL + kernel32.LocalFree.argtypes = [wintypes.HLOCAL] + kernel32.LocalFree.restype = wintypes.HLOCAL + return _Win32(kernel32=kernel32, userenv=userenv, advapi32=advapi32) + + +def _close(win: _Win32, handle: Any) -> None: + if not _invalid_handle(handle): + win.kernel32.CloseHandle(handle) + + +def _app_container_sid(win: _Win32, process: Any) -> str: + token = wintypes.HANDLE() + if not win.advapi32.OpenProcessToken(process, _TOKEN_QUERY, ctypes.byref(token)): + _raise("native_token_open_failed") + try: + is_container = wintypes.DWORD() + returned = wintypes.DWORD() + if not win.advapi32.GetTokenInformation( + token, + _TOKEN_IS_APPCONTAINER, + ctypes.byref(is_container), + ctypes.sizeof(is_container), + ctypes.byref(returned), + ): + _raise("native_token_query_failed") + if not is_container.value: + _raise("native_appcontainer_missing") + required = wintypes.DWORD() + win.advapi32.GetTokenInformation( + token, + _TOKEN_APPCONTAINER_SID, + None, + 0, + ctypes.byref(required), + ) + if not required.value: + _raise("native_appcontainer_sid_missing") + buffer = ctypes.create_string_buffer(required.value) + if not win.advapi32.GetTokenInformation( + token, + _TOKEN_APPCONTAINER_SID, + buffer, + required.value, + ctypes.byref(required), + ): + _raise("native_appcontainer_sid_missing") + sid_ptr = ctypes.cast(buffer, ctypes.POINTER(ctypes.c_void_p))[0] + sid_text = wintypes.LPWSTR() + if not win.advapi32.ConvertSidToStringSidW(sid_ptr, ctypes.byref(sid_text)): + _raise("native_appcontainer_sid_invalid") + try: + return sid_text.value or "" + finally: + win.kernel32.LocalFree(sid_text) + finally: + _close(win, token) + + +class Win32SuspendedWorker: + """Own a suspended worker and its kill-on-close Job Object.""" + + def __init__( + self, + win: _Win32, + *, + profile_name: str, + process: Any, + thread: Any, + process_id: int, + app_container_sid: str, + ) -> None: + self._win = win + self._profile_name = profile_name + self._process = process + self._thread = thread + self._job: Any = None + self._closed = False + self._resumed = False + self.process_id = process_id + self.app_container_sid = app_container_sid + + def apply_job_policy(self, policy: NativeWorkerPolicy) -> None: + if self._closed or self._resumed or self._job is not None: + _raise("native_job_state_invalid") + job = self._win.kernel32.CreateJobObjectW(None, None) + if _invalid_handle(job): + _raise("native_job_create_failed") + try: + limits = _ExtendedLimitInformation() + limits.basic_limit_information.limit_flags = policy.required_limit_flags + limits.basic_limit_information.active_process_limit = policy.active_process_limit + limits.basic_limit_information.per_process_user_time.quad_part = ( + policy.process_user_time_100ns + ) + limits.basic_limit_information.per_job_user_time.quad_part = policy.job_user_time_100ns + limits.process_memory_limit = policy.process_memory_limit_bytes + limits.job_memory_limit = policy.job_memory_limit_bytes + if not self._win.kernel32.SetInformationJobObject( + job, + _JOB_OBJECT_EXTENDED_LIMIT_INFORMATION, + ctypes.byref(limits), + ctypes.sizeof(limits), + ): + _raise("native_job_configure_failed") + if not self._win.kernel32.AssignProcessToJobObject(job, self._process): + _raise("native_job_assign_failed") + self._verify_job_policy(job, policy) + self._job = job + except Exception: + _close(self._win, job) + raise + + def _verify_job_policy(self, job: Any, policy: NativeWorkerPolicy) -> None: + info = _ExtendedLimitInformation() + returned = wintypes.DWORD() + if not self._win.kernel32.QueryInformationJobObject( + job, + _JOB_OBJECT_EXTENDED_LIMIT_INFORMATION, + ctypes.byref(info), + ctypes.sizeof(info), + ctypes.byref(returned), + ): + _raise("native_job_query_failed") + flags = int(info.basic_limit_information.limit_flags) + if flags != policy.required_limit_flags or flags & ( + _JOB_OBJECT_LIMIT_BREAKAWAY_OK | _JOB_OBJECT_LIMIT_SILENT_BREAKAWAY_OK + ): + _raise("native_job_policy_mismatch") + if int(info.basic_limit_information.active_process_limit) != policy.active_process_limit: + _raise("native_job_policy_mismatch") + if int(info.process_memory_limit) != policy.process_memory_limit_bytes: + _raise("native_job_policy_mismatch") + if int(info.job_memory_limit) != policy.job_memory_limit_bytes: + _raise("native_job_policy_mismatch") + if int(info.basic_limit_information.per_process_user_time.quad_part) != policy.process_user_time_100ns: + _raise("native_job_policy_mismatch") + if int(info.basic_limit_information.per_job_user_time.quad_part) != policy.job_user_time_100ns: + _raise("native_job_policy_mismatch") + + def resume(self) -> None: + if self._closed or self._resumed or self._job is None: + _raise("native_resume_state_invalid") + if self._win.kernel32.ResumeThread(self._thread) == 0xFFFFFFFF: + _raise("native_resume_failed") + self._resumed = True + + def close(self) -> None: + if self._closed: + return + self._closed = True + if self._job is None and not _invalid_handle(self._process): + self._win.kernel32.TerminateProcess(self._process, 1) + if not _invalid_handle(self._job): + _close(self._win, self._job) + self._job = None + if not _invalid_handle(self._process): + self._win.kernel32.WaitForSingleObject(self._process, 3000) + _close(self._win, self._thread) + _close(self._win, self._process) + self._thread = None + self._process = None + self._win.userenv.DeleteAppContainerProfile(self._profile_name) + + def __enter__(self) -> "Win32SuspendedWorker": + return self + + def __exit__(self, *_exc: object) -> None: + self.close() + + +class NativeWin32ProcessFactory: + """Create a fixed worker in a zero-capability AppContainer, suspended.""" + + def __init__(self, *, profile_prefix: str = "CortexRecipe") -> None: + if not isinstance(profile_prefix, str) or not profile_prefix.isidentifier(): + raise ValueError("profile prefix is invalid") + self._profile_prefix = profile_prefix + + def create_suspended(self, plan: NativeWorkerLaunchPlan) -> NativeSuspendedWorker: + _require_windows() + profile_name = f"{self._profile_prefix}{uuid4().hex}" + win = _configure() + profile_sid = wintypes.LPVOID() + process_info = _ProcessInformation() + attribute_list: wintypes.LPVOID | None = None + attribute_buffer: ctypes.Array[ctypes.c_char] | None = None + profile_created = False + try: + hr = win.userenv.CreateAppContainerProfile( + profile_name, + "Cortex recipe worker", + "Zero-capability fixed image recipe worker", + None, + 0, + ctypes.byref(profile_sid), + ) + if hr not in (0, 0x800700B7): + _raise("native_profile_create_failed") + profile_created = True + required = ctypes.c_size_t() + win.kernel32.InitializeProcThreadAttributeList(None, 1, 0, ctypes.byref(required)) + if not required.value: + _raise("native_attribute_size_failed") + attribute_buffer = ctypes.create_string_buffer(required.value) + attribute_list = ctypes.cast(attribute_buffer, wintypes.LPVOID) + if not win.kernel32.InitializeProcThreadAttributeList( + attribute_list, 1, 0, ctypes.byref(required) + ): + _raise("native_attribute_init_failed") + capabilities = _SecurityCapabilities( + app_container_sid=profile_sid, + capabilities=None, + capability_count=0, + reserved=0, + ) + if not win.kernel32.UpdateProcThreadAttribute( + attribute_list, + 0, + _PROC_THREAD_ATTRIBUTE_SECURITY_CAPABILITIES, + ctypes.byref(capabilities), + ctypes.sizeof(capabilities), + None, + None, + ): + _raise("native_attribute_security_failed") + startup = _StartupInfoEx() + startup.startup_info.cb = ctypes.sizeof(startup) + startup.attribute_list = attribute_list + command_buffer = ctypes.create_unicode_buffer(plan.command_line) + flags = ( + _EXTENDED_STARTUPINFO_PRESENT + | _CREATE_SUSPENDED + | _CREATE_UNICODE_ENVIRONMENT + | _CREATE_NO_WINDOW + ) + if not win.kernel32.CreateProcessW( + str(plan.executable), + command_buffer, + None, + None, + False, + flags, + None, + None, + ctypes.byref(startup), + ctypes.byref(process_info), + ): + _raise("native_process_create_failed") + sid = _app_container_sid(win, process_info.process) + if not sid.startswith("S-1-15-2-"): + _raise("native_appcontainer_sid_invalid") + if attribute_list is not None: + win.kernel32.DeleteProcThreadAttributeList(attribute_list) + attribute_list = None + return Win32SuspendedWorker( + win, + profile_name=profile_name, + process=process_info.process, + thread=process_info.thread, + process_id=int(process_info.process_id), + app_container_sid=sid, + ) + except Exception: + if attribute_list is not None: + win.kernel32.DeleteProcThreadAttributeList(attribute_list) + if not _invalid_handle(process_info.thread): + _close(win, process_info.thread) + if not _invalid_handle(process_info.process): + win.kernel32.TerminateProcess(process_info.process, 1) + win.kernel32.WaitForSingleObject(process_info.process, 3000) + _close(win, process_info.process) + if profile_created: + win.userenv.DeleteAppContainerProfile(profile_name) + if isinstance(sys.exc_info()[1], NativeLauncherError): + raise + raise NativeWin32Error("native_process_create_failed") from None + + +__all__ = ["NativeWin32Error", "NativeWin32ProcessFactory", "Win32SuspendedWorker"] diff --git a/docs/adr/0001-capability-tiered-agentic-execution-harness.md b/docs/adr/0001-capability-tiered-agentic-execution-harness.md index 03599c4..1725ab1 100644 --- a/docs/adr/0001-capability-tiered-agentic-execution-harness.md +++ b/docs/adr/0001-capability-tiered-agentic-execution-harness.md @@ -947,9 +947,10 @@ pass without executing code. composes AppContainer/Job Object controls and a fixed decoder corpus. Storage-only signed worker provenance now binds exactly one image-transform role to the fixed worker entrypoint, and the disposable launcher now qualifies pre-resume Job Object - resource policy. Native worker launch and broker PID/token binding remain blocked; - collect opt-in aggregate reliability metrics, never content, only after the - provider is sandbox-qualified. + resource policy. The reviewed Win32 suspended factory and launcher-side broker + PID/AppContainer-token binder are now qualified; signed-worker end-to-end broker + execution remains blocked. Collect opt-in aggregate reliability metrics, never + content, only after the provider is sandbox-qualified. **Implementation gate:** parser, artifact-boundary, qualification-provider, worker-provenance, sandbox-qualification, and native-launcher-policy regression diff --git a/docs/adr/0001-phase2-evidence.md b/docs/adr/0001-phase2-evidence.md index 2cf1269..5dd3cc4 100644 --- a/docs/adr/0001-phase2-evidence.md +++ b/docs/adr/0001-phase2-evidence.md @@ -24,8 +24,8 @@ | 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 (boundary + disposable control spike)** | `NativeWorkerLauncher` revalidates the signed worker, fixes the broker-only command line, requires both native factory and broker binder, and enforces policy-before-bind-before-resume. `native_launcher_qualification.py` separately proves the queried Job Object policy with a fixed suspended child; concrete worker enforcement and live broker binding remain blocked. | -| OS sandbox provider and provider-produced image outputs | **Blocked / release gate** | The package/protocol/launch boundary is qualified, but the actual provider worker still needs a signed installed generation, concrete Win32 process factory, 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. | +| 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 is qualified, but the actual provider worker still needs a signed installed generation, worker-side broker loop, end-to-end authenticated input/output, watchdog, hostile decoder execution inside the sandbox, external review, and lifecycle wiring. | ## Security invariants @@ -114,11 +114,11 @@ npm.cmd test --prefix frontend -- --run **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, 7 -worker-protocol tests, 11 native-launcher-boundary tests, 4 +worker-protocol tests, 16 native-launcher/factory tests, 4 native-launcher tests, and 5 sandbox-qualification tests passed; the full Python suite -passed (237 tests total) with one +passed (242 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-native-broker.md b/docs/adr/0001-phase2-native-broker.md index 384b87b..a7fca54 100644 --- a/docs/adr/0001-phase2-native-broker.md +++ b/docs/adr/0001-phase2-native-broker.md @@ -1,6 +1,6 @@ # ADR-0001 Phase 2 native broker adapter -- **Status:** Implemented and verified; provider enablement remains blocked +- **Status:** Transport and launcher identity binder implemented and verified; signed-worker end-to-end enablement remains blocked - **Parent:** [Phase 2 authenticated broker contract](0001-phase2-broker-contract.md) - **Scope:** Windows named-pipe transport, protected DACL, OS peer identity, and authenticated session-key establishment @@ -39,6 +39,14 @@ identity; it binds responses only to the expected installation and job owner. Any framing, identity, handshake, direction, or authorization failure closes the connection and returns only a stable category. +The launcher-side `NativeBrokerIdentityBinder` creates this server only after the +worker has been created suspended. It sets `expected_process_id` to the actual +worker PID and allows exactly the worker's AppContainer SID in the protected DACL; +the broker PID is required to equal the current server process. The binder pins the +installation principal and durable job for `accept()`, and closes the endpoint on +resume or cancellation failure. This is the live identity binding seam; the +signed worker package and worker-side client loop remain separate release gates. + ## Lifecycle and failure contract The adapter is a blocking transport primitive intended to run in a coordinator diff --git a/docs/adr/0001-phase2-native-launcher.md b/docs/adr/0001-phase2-native-launcher.md index 1dae651..8b22ba5 100644 --- a/docs/adr/0001-phase2-native-launcher.md +++ b/docs/adr/0001-phase2-native-launcher.md @@ -1,6 +1,6 @@ # ADR-0001 Phase 2 disposable native launcher qualification -- **Status:** Launch-plan boundary and policy contract complete; concrete worker launch and live broker binding blocked +- **Status:** Win32 suspended factory and broker identity binder qualified; signed-worker end-to-end execution remains blocked - **Phase:** 2 - fixed-function image provider - **Parent:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md) - **Depends on:** [Windows sandbox qualification](0001-phase2-sandbox-qualification.md) and [signed worker provenance](0001-phase2-worker-provenance.md) @@ -23,6 +23,20 @@ accepts only an installer-verified worker and a trusted `BrokerWorkerBinding`: 5. enforce the order `create suspended -> apply Job policy -> bind worker PID and AppContainer SID -> resume`. +`NativeWin32ProcessFactory` now implements the reviewed factory: it creates a +unique zero-capability AppContainer profile, starts the fixed executable suspended, +verifies the process token is an AppContainer with the expected SID shape, and +returns a handle that applies/query-verifies the active-process, CPU, memory, +kill-on-close, and no-breakaway Job Object policy before resume. Cleanup terminates +unassigned processes, closes the kill-on-close Job, waits for reaping, closes all +handles, and deletes the disposable profile. + +`NativeBrokerIdentityBinder` now creates a protected `NativeBrokerServer` whose +peer policy contains the actual suspended worker PID and AppContainer SID. It also +requires the broker PID to equal the current server process, pins the installation +principal/job, and exposes accept/close ownership to the coordinator. The launcher +closes the endpoint if binding or resume fails. + The disposable qualification helper still exercises the lower-level construction sequence with a fixed benign executable: @@ -42,9 +56,10 @@ remain green after this change. This is qualification evidence for control application, not approval to launch a recipe worker. The fixed probe reports `provider_launch_authorized=false` until -the signed worker package, concrete native process factory, and broker identity -gates pass. The production launcher boundary also fails closed when either adapter -is absent; it never falls back to `subprocess`, a shell, stdio, or a weaker sandbox. +the signed worker package is installed and the worker loop completes an end-to-end +authenticated broker session. The production launcher boundary also fails closed +when either adapter is absent; it never falls back to `subprocess`, a shell, stdio, +or a weaker sandbox. ## Evidence and limits @@ -65,11 +80,12 @@ behavior across supported Windows versions, or external launcher review. - The fixed signed `recipe_worker.exe` package is not shipped at the immutable installer generation used by the launcher. -- The reviewed Win32 process factory that creates the worker AppContainer token, - assigns and verifies the Job Object, and exposes a suspended process handle is - not implemented yet. -- The native broker transport is not yet bound to the launched worker PID and OS - token by a single reviewed launcher transaction. +- The fixed signed `recipe_worker.exe` bundle is not installed in the immutable + generation used by the launcher, so no real worker has completed the broker + handshake yet. +- The packaged worker loop still exits with its launch-refusing status; the + end-to-end authenticated input/output, watchdog, and cancellation path must be + wired before provider execution is authorized. - Watchdog progress, output framing, staging ACLs, hostile decoder execution, and lifecycle health-gated wiring remain separate release gates. @@ -81,8 +97,11 @@ there is no host-process or weaker-sandbox fallback. `tests/test_phase2_native_launcher.py` covers bounded policy values, no-breakaway flags, trusted binding validation, worker revalidation, fixed command-line construction, refusal before process creation without a binder, policy/bind/resume -ordering, cleanup on binding failure, tamper rejection, and non-Windows blocking. -`tests/test_native_launcher_qualification.py` covers the disposable probe's +ordering, cleanup on binding/resume failure, broker PID/AppContainer binding, and +tamper rejection. `tests/test_phase2_native_win32.py` creates a fixed suspended +AppContainer child on Windows, verifies its token, applies/query-verifies Job Object +policy, and closes it without resuming. `tests/test_native_launcher_qualification.py` +covers the disposable probe's non-Windows blocking, report fail-closed behavior, and no-breakaway invariant. The full repository suite and the existing AppContainer/cancellation corpus are required before this boundary can be merged. diff --git a/docs/adr/0001-phase2-sandbox-qualification.md b/docs/adr/0001-phase2-sandbox-qualification.md index 38b99f3..a0360f9 100644 --- a/docs/adr/0001-phase2-sandbox-qualification.md +++ b/docs/adr/0001-phase2-sandbox-qualification.md @@ -22,8 +22,9 @@ The harness is deliberately fail-closed and has independent checks for: packaging location; 5. it reports per-worker CPU/memory/breakaway/accounting controls as blocked until the native launcher applies and verifies them; and -6. it reports broker PID/token binding as blocked until the launcher binds the - qualified transport to the actual worker. +6. it reports end-to-end broker execution as blocked until the launcher binds the + qualified transport to a signed worker's actual PID/AppContainer token and the + worker completes the authenticated client handshake. The fourth check is intentionally blocked in this stage. A directory, executable, self-reported digest, or unverified manifest cannot authorize a launch. The future diff --git a/tests/test_phase2_native_launcher.py b/tests/test_phase2_native_launcher.py index 697397c..d69bc5b 100644 --- a/tests/test_phase2_native_launcher.py +++ b/tests/test_phase2_native_launcher.py @@ -8,6 +8,7 @@ import os from pathlib import Path import subprocess +from uuid import uuid4 import pytest from cryptography.hazmat.primitives import serialization @@ -17,6 +18,7 @@ from cortex_backend.execution.bundle_installer import SignedBundleInstaller from cortex_backend.execution.native_launcher import ( BrokerWorkerBinding, + NativeBrokerIdentityBinder, NativeLauncherError, NativeWorkerLaunchPlan, NativeWorkerLauncher, @@ -229,6 +231,47 @@ def bind_worker(self, **kwargs): assert events == ["create", "policy", "bind", "close"] +def test_launcher_closes_live_broker_binding_if_resume_fails(tmp_path: Path): + installer = _installer(tmp_path) + events: list[str] = [] + + class _Worker: + process_id = 446 + app_container_sid = "S-1-15-2-123-456" + + def apply_job_policy(self, policy: NativeWorkerPolicy) -> None: + events.append("policy") + + def resume(self) -> None: + events.append("resume") + raise NativeLauncherError("native_resume_failed") + + def close(self) -> None: + events.append("worker-close") + + class _Factory: + def create_suspended(self, plan: NativeWorkerLaunchPlan): + events.append("create") + return _Worker() + + class _Binder: + def bind_worker(self, **kwargs): + del kwargs + events.append("bind") + + def close_binding(self): + events.append("broker-close") + + with pytest.raises(NativeLauncherError) as error: + NativeWorkerLauncher( + installer, + process_factory=_Factory(), + broker_binder=_Binder(), + ).launch(_binding()) + assert error.value.code == "native_resume_failed" + assert events == ["create", "policy", "bind", "resume", "broker-close", "worker-close"] + + def test_tampered_worker_is_rejected_at_launch_plan_boundary(tmp_path: Path): installer = _installer(tmp_path) installed = installer.status() @@ -240,6 +283,111 @@ def test_tampered_worker_is_rejected_at_launch_plan_boundary(tmp_path: Path): assert error.value.code == "worker_bundle_integrity_failed" +def test_native_broker_identity_binder_pins_worker_pid_and_appcontainer_sid( + monkeypatch: pytest.MonkeyPatch, +): + servers = [] + accepted = object() + + class _Server: + def __init__(self, config): + self.config = config + self.opened = False + self.closed = False + + def open(self): + self.opened = True + + def accept(self, *, expected_principal_id, owner_for_job): + assert expected_principal_id == "a" * 64 + assert owner_for_job("job-1") == "owner" + return accepted + + def close(self): + self.closed = True + + def factory(config): + server = _Server(config) + servers.append(server) + return server + + monkeypatch.setattr(launcher_module.os, "getpid", lambda: 321) + binder = NativeBrokerIdentityBinder( + allowed_user_sids=frozenset({"S-1-5-21-1-2-3-4"}), + server_factory=factory, + ) + binding = _binding() + binder.bind_worker( + process_id=446, + app_container_sid="S-1-15-2-123-456", + binding=binding, + ) + + assert binder.bound + assert binder.binding == binding + assert servers[0].config.peer_policy.expected_process_id == 446 + assert servers[0].config.peer_policy.acl.allowed_app_container_sids == frozenset( + {"S-1-15-2-123-456"} + ) + assert binder.accept(owner_for_job=lambda job_id: "owner" if job_id == "job-1" else None) is accepted + binder.close_binding() + assert not binder.bound + assert servers[0].closed + + +def test_native_broker_identity_binder_rejects_pid_or_token_mismatch( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setattr(launcher_module.os, "getpid", lambda: 321) + binder = NativeBrokerIdentityBinder( + allowed_user_sids=frozenset({"S-1-5-21-1-2-3-4"}), + server_factory=lambda config: object(), + ) + with pytest.raises(NativeLauncherError) as pid_error: + binder.bind_worker( + process_id=446, + app_container_sid="S-1-15-2-123-456", + binding=BrokerWorkerBinding( + pipe_name=_binding().pipe_name, + broker_process_id=322, + installation_principal_id="a" * 64, + job_id="job-1", + ), + ) + assert pid_error.value.code == "native_broker_process_mismatch" + with pytest.raises(NativeLauncherError) as sid_error: + binder.bind_worker( + process_id=446, + app_container_sid="S-1-5-21-1-2-3-4", + binding=_binding(), + ) + assert sid_error.value.code == "native_appcontainer_sid_invalid" + + +@pytest.mark.skipif(os.name != "nt", reason="native broker server requires Windows") +def test_native_broker_identity_binder_opens_real_protected_server(): + binding = BrokerWorkerBinding( + pipe_name=rf"\\.\pipe\cortex-binder-{uuid4().hex}", + broker_process_id=os.getpid(), + installation_principal_id="a" * 64, + job_id="job-real-binder", + ) + binder = NativeBrokerIdentityBinder( + allowed_user_sids=frozenset({"S-1-5-21-100-200-300-400"}), + ) + binder.bind_worker( + process_id=os.getpid(), + app_container_sid="S-1-15-2-100-200-300-400", + binding=binding, + ) + try: + assert binder.bound + assert binder.binding == binding + finally: + binder.close_binding() + assert not binder.bound + + def test_non_windows_launch_is_blocked_before_adapters(monkeypatch: pytest.MonkeyPatch, tmp_path: Path): installer = _installer(tmp_path) monkeypatch.setattr(launcher_module.os, "name", "posix") diff --git a/tests/test_phase2_native_win32.py b/tests/test_phase2_native_win32.py new file mode 100644 index 0000000..d65826b --- /dev/null +++ b/tests/test_phase2_native_win32.py @@ -0,0 +1,62 @@ +"""Controlled Windows evidence for the concrete suspended worker factory.""" + +from __future__ import annotations + +import os +from pathlib import Path +import subprocess + +import pytest + +from cortex_backend.execution.native_launcher import ( + BrokerWorkerBinding, + NativeWorkerLaunchPlan, + NativeWorkerPolicy, +) +from cortex_backend.execution.native_win32 import NativeWin32ProcessFactory +from cortex_backend.execution.worker_provenance import VerifiedRecipeWorker + + +@pytest.mark.skipif(os.name != "nt", reason="Win32 process factory requires Windows") +def test_win32_factory_creates_suspended_appcontainer_and_job_policy(): + system_root = Path(os.environ.get("WINDIR", r"C:\Windows")) + executable = system_root / "System32" / "findstr.exe" + if not executable.is_file(): + pytest.skip("findstr.exe is unavailable on this Windows host") + binding = BrokerWorkerBinding( + pipe_name=r"\\.\pipe\cortex-worker-factory-test", + broker_process_id=os.getpid(), + installation_principal_id="a" * 64, + job_id="job-factory-test", + ) + plan = NativeWorkerLaunchPlan( + worker=VerifiedRecipeWorker( + bundle_root=system_root, + bundle_digest="0" * 64, + key_id="release-1", + worker_path="recipe_worker.exe", + worker_sha256="0" * 64, + worker_size=1, + recipe_version="1.0.0", + ), + executable=executable, + command_line=subprocess.list2cmdline( + [ + str(executable), + "--native-broker", + "--broker-pipe", + binding.pipe_name, + "--broker-pid", + str(binding.broker_process_id), + ] + ), + broker=binding, + policy=NativeWorkerPolicy(), + ) + worker = NativeWin32ProcessFactory().create_suspended(plan) + try: + assert worker.process_id > 0 + assert worker.app_container_sid.startswith("S-1-15-2-") + worker.apply_job_policy(plan.policy) + finally: + worker.close() diff --git a/tools/execution_spikes/README.md b/tools/execution_spikes/README.md index bb71b83..2031b37 100644 --- a/tools/execution_spikes/README.md +++ b/tools/execution_spikes/README.md @@ -46,7 +46,8 @@ resource/watchdog enforcement are implemented. The native launcher qualification prints a passing resource-policy subcheck when the fixed suspended child receives and reports Job Object CPU/memory/active-process limits with no breakaway flags. Its overall exit remains blocked until the signed -worker package and broker PID/token binding exist. +worker package is installed and the worker completes the live authenticated broker +handshake; launcher-side PID/AppContainer binding is now implemented separately. The fixed worker protocol/package boundary can be qualified separately on Windows: @@ -93,6 +94,9 @@ provider. The package contract is covered by `tests/test_phase2_worker_protocol. launch-plan boundary that revalidates the signed worker and refuses process creation until a reviewed native process factory and live broker binder are supplied. It does not provide a fallback launcher. +- `backend/cortex_backend/execution/native_win32.py`: reviewed Windows factory + that creates a suspended zero-capability AppContainer child, verifies its token, + applies/query-verifies Job Object policy, and cleans up every handle on failure. - `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.