diff --git a/backend/cortex_backend/api/routes.py b/backend/cortex_backend/api/routes.py index e035058..0ed9d26 100644 --- a/backend/cortex_backend/api/routes.py +++ b/backend/cortex_backend/api/routes.py @@ -6,6 +6,8 @@ from dataclasses import asdict from datetime import datetime, timezone import asyncio +import base64 +import binascii import hmac import json import logging @@ -29,6 +31,10 @@ from cortex_backend.services.models import ModelPullProgress from cortex_backend.execution.coordinator import DurableFakeCoordinator from cortex_backend.execution.fake import FakeExecutionPlan +from cortex_backend.execution.attachment_staging import ( + AttachmentStagingError, + AttachmentStagingService, +) from cortex_backend.execution.models import ExecutionJob, ExecutionEvent, TerminalExecutionStatus from cortex_backend.execution.recipe_coordinator import ( RECIPE_IMAGE_PROFILE, @@ -52,6 +58,8 @@ ClearMemoryRequest, CreateChatRequest, DiagnosticsResponse, + AttachmentStageAccepted, + AttachmentStageRequest, ExecutionAccepted, ExecutionApprovalDecisionRequest, ExecutionPreviewRequest, @@ -481,6 +489,57 @@ def start_recipe_image_transform( sequence=job.sequence, ) + @router.post( + "/execution/attachments", + response_model=AttachmentStageAccepted, + status_code=status.HTTP_201_CREATED, + ) + def stage_attachment( + request: Request, + payload: AttachmentStageRequest, + principal: SessionPrincipal = Depends(require_session), + ) -> AttachmentStageAccepted: + """Stage one bounded attachment for a qualified recipe request. + + The route accepts only a bounded base64 envelope. Decoded bytes pass + through the same owner-scoped artifact boundary used by recipe output; + no caller path, filename, or executable instruction is accepted. + """ + + coordinator = _recipe_coordinator(request) + try: + content = base64.b64decode(payload.content_base64, validate=True) + except (ValueError, binascii.Error): + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="Attachment payload is invalid.", + ) from None + try: + staged = AttachmentStagingService( + coordinator.repository, + coordinator.artifact_boundary, + ).stage( + owner=_execution_owner(principal), + request_id=payload.request_id, + content=content, + retention_seconds=payload.retention_seconds, + ) + except AttachmentStagingError as exc: + _raise_attachment_staging_error(exc) + artifact = staged.artifact + return AttachmentStageAccepted( + job_id=staged.job.job_id, + request_id=staged.job.request_id, + profile="attachment.stage.v1", + status="succeeded", + sequence=staged.job.sequence, + artifact_id=artifact.artifact_id, + mime_type=artifact.mime_type, + size=artifact.size, + sha256=artifact.sha256, + expires_at=datetime.fromisoformat(artifact.expires_at), + ) + @router.get("/execution/tasks", response_model=ExecutionTaskListResponse) def execution_tasks( request: Request, @@ -1505,6 +1564,30 @@ def _raise_recipe_request_error(exc: RecipeExecutionError) -> None: ) from exc +def _raise_attachment_staging_error(exc: AttachmentStagingError) -> None: + """Map attachment stage categories to stable, non-sensitive responses.""" + + if exc.code == "request_conflict": + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Attachment request conflicts with an existing request.", + ) from exc + if exc.code == "attachment_in_progress": + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Attachment request is already in progress.", + ) from exc + if exc.code in {"attachment_artifact_unavailable", "attachment_failed"}: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Attachment request is no longer available.", + ) from exc + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="Attachment could not be staged safely.", + ) from exc + + def _execution_repository(request: Request): return _execution_runtime(request).repository diff --git a/backend/cortex_backend/api/schemas.py b/backend/cortex_backend/api/schemas.py index 41b3715..353da1e 100644 --- a/backend/cortex_backend/api/schemas.py +++ b/backend/cortex_backend/api/schemas.py @@ -14,6 +14,10 @@ DEFAULT_RECIPE_RETENTION_SECONDS, MAX_RECIPE_RETENTION_SECONDS, ) +from cortex_backend.execution.attachment_staging import ( + DEFAULT_ATTACHMENT_RETENTION_SECONDS, + MAX_ATTACHMENT_RETENTION_SECONDS, +) from cortex_backend.execution.recipes import ( ImageTransformPlan, RecipeValidationError, @@ -254,6 +258,37 @@ def _bind_plan_to_source(self) -> "RecipeImageTransformRequest": return self +# The default repository ceiling is 10 MiB. This request ceiling includes +# base64 overhead while the service applies the configured byte ceiling again. +MAX_ATTACHMENT_BASE64_LENGTH = 14 * 1024 * 1024 + + +class AttachmentStageRequest(APIModel): + """Bounded base64 envelope for a user attachment. + + The service decodes and MIME-sniffs the bytes before an artifact is + published. The encoded payload is never written to the durable job. + """ + + request_id: str = Field( + min_length=1, + max_length=128, + pattern=r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$", + strict=True, + ) + content_base64: str = Field( + min_length=4, + max_length=MAX_ATTACHMENT_BASE64_LENGTH, + strict=True, + ) + retention_seconds: int = Field( + default=DEFAULT_ATTACHMENT_RETENTION_SECONDS, + ge=1, + le=MAX_ATTACHMENT_RETENTION_SECONDS, + strict=True, + ) + + class ExecutionAccepted(APIModel): job_id: str request_id: str @@ -270,6 +305,19 @@ class RecipeImageTransformAccepted(APIModel): sequence: int +class AttachmentStageAccepted(APIModel): + job_id: str + request_id: str + profile: Literal["attachment.stage.v1"] + status: Literal["succeeded"] + sequence: int + artifact_id: str + mime_type: str + size: int + sha256: str + expires_at: datetime + + class ExecutionApprovalDecisionRequest(APIModel): decision: Literal["approved", "denied"] diff --git a/backend/cortex_backend/execution/__init__.py b/backend/cortex_backend/execution/__init__.py index 7587c42..8fb93d0 100644 --- a/backend/cortex_backend/execution/__init__.py +++ b/backend/cortex_backend/execution/__init__.py @@ -32,6 +32,16 @@ PublishedArtifact, sniff_artifact_mime, ) +from .attachment_staging import ( + ATTACHMENT_PAYLOAD_SCHEMA, + ATTACHMENT_RESULT_SCHEMA, + ATTACHMENT_STAGE_PROFILE, + AttachmentStageResult, + AttachmentStagingError, + AttachmentStagingService, + DEFAULT_ATTACHMENT_RETENTION_SECONDS, + MAX_ATTACHMENT_RETENTION_SECONDS, +) from .fake import FakeExecutionPlan, FakeExecutionProvider from .bundle_installer import ( BundleInstallError, @@ -51,6 +61,7 @@ QualificationLifecycleConfig, QualificationProfileError, build_execution_lifecycle, + build_native_recipe_coordinator_factory, build_recipe_coordinator_factory, parse_execution_profile, ) @@ -89,6 +100,11 @@ NativeWorkerLauncher, NativeWorkerPolicy, ) +from .native_recipe_attempt import ( + NativeRecipeWorkerAttempt, + NativeRecipeWorkerAttemptFactory, + build_native_recipe_worker_attempt_factory, +) from .native_win32 import ( NativeWin32Error, NativeWin32ProcessFactory, @@ -186,6 +202,14 @@ "ArtifactBoundary", "ArtifactBoundaryError", "ArtifactSourceGrant", + "ATTACHMENT_PAYLOAD_SCHEMA", + "ATTACHMENT_RESULT_SCHEMA", + "ATTACHMENT_STAGE_PROFILE", + "AttachmentStageResult", + "AttachmentStagingError", + "AttachmentStagingService", + "DEFAULT_ATTACHMENT_RETENTION_SECONDS", + "MAX_ATTACHMENT_RETENTION_SECONDS", "ApprovalPolicyError", "ApprovalTransitionError", "BrokerAclPolicy", @@ -220,6 +244,7 @@ "ExecutionProfile", "CoordinatorFactory", "build_recipe_coordinator_factory", + "build_native_recipe_coordinator_factory", "CalculatorPlan", "CheckPlan", "FakeExecutionPlan", @@ -250,6 +275,9 @@ "NativeWorkerLaunchPlan", "NativeWorkerLauncher", "NativeWorkerPolicy", + "NativeRecipeWorkerAttempt", + "NativeRecipeWorkerAttemptFactory", + "build_native_recipe_worker_attempt_factory", "NativeWin32Error", "NativeWin32ProcessFactory", "Win32SuspendedWorker", diff --git a/backend/cortex_backend/execution/artifact_boundary.py b/backend/cortex_backend/execution/artifact_boundary.py index ac4446f..e994380 100644 --- a/backend/cortex_backend/execution/artifact_boundary.py +++ b/backend/cortex_backend/execution/artifact_boundary.py @@ -328,27 +328,27 @@ def _authorize_job(self, job_id: str, owner: str) -> None: def _generated_name(digest: str) -> str: return f"artifact-{digest[:32]}" - def copy_in( + def _publish_bytes( self, - grant: ArtifactSourceGrant, + job_id: str, + owner: str, + content: bytes, *, - retention_seconds: int = 86_400, + retention_seconds: int, ) -> ExecutionArtifact: - """Copy one explicitly granted source into the owner-scoped artifact store.""" + """Validate and publish one bounded byte payload for an owned job.""" - if not isinstance(grant, ArtifactSourceGrant): - raise ArtifactBoundaryError("artifact_grant_invalid") self._validate_retention(retention_seconds) - self._authorize_job(grant.job_id, grant.owner) - source = grant.source_path - _validate_absolute_path(source) - _validate_components(source) - content = _read_stable(source, self.max_input_bytes) + self._authorize_job(job_id, owner) + if not isinstance(content, bytes) or not content: + raise ArtifactBoundaryError("artifact_content_invalid") + if len(content) > self.max_input_bytes: + raise ArtifactBoundaryError("artifact_too_large") mime_type = sniff_artifact_mime(content) digest = sha256(content).hexdigest() try: return self.repository.publish_artifact( - grant.job_id, + job_id, name=self._generated_name(digest), content=content, mime_type=mime_type, @@ -357,6 +357,46 @@ def copy_in( except ExecutionRepositoryError: raise ArtifactBoundaryError("artifact_publish_failed") from None + def stage_bytes( + self, + job_id: str, + owner: str, + content: bytes, + *, + retention_seconds: int = 86_400, + ) -> ExecutionArtifact: + """Stage one bounded attachment payload without accepting a source path.""" + + return self._publish_bytes( + job_id, + owner, + content, + retention_seconds=retention_seconds, + ) + + def copy_in( + self, + grant: ArtifactSourceGrant, + *, + retention_seconds: int = 86_400, + ) -> ExecutionArtifact: + """Copy one explicitly granted source into the owner-scoped artifact store.""" + + if not isinstance(grant, ArtifactSourceGrant): + raise ArtifactBoundaryError("artifact_grant_invalid") + self._validate_retention(retention_seconds) + self._authorize_job(grant.job_id, grant.owner) + source = grant.source_path + _validate_absolute_path(source) + _validate_components(source) + content = _read_stable(source, self.max_input_bytes) + return self._publish_bytes( + grant.job_id, + grant.owner, + content, + retention_seconds=retention_seconds, + ) + @staticmethod def _validate_retention(retention_seconds: int) -> None: if ( diff --git a/backend/cortex_backend/execution/attachment_staging.py b/backend/cortex_backend/execution/attachment_staging.py new file mode 100644 index 0000000..4fb9ce4 --- /dev/null +++ b/backend/cortex_backend/execution/attachment_staging.py @@ -0,0 +1,259 @@ +"""Durable, owner-scoped staging for user attachment bytes. + +The attachment boundary is intentionally a small capability: it accepts one +bounded in-memory payload, validates the bytes through :class:`ArtifactBoundary`, +and returns only an opaque artifact identifier and derived metadata. Raw paths, +filenames, shell text, and executable authority never enter the job payload. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from hashlib import sha256 +import json +import re +from typing import Any +from uuid import uuid4 + +from .artifact_boundary import ArtifactBoundary, ArtifactBoundaryError, sniff_artifact_mime +from .models import ExecutionArtifact, ExecutionJob +from .repository import ExecutionRepository, ExecutionRepositoryError + + +ATTACHMENT_STAGE_PROFILE = "attachment.stage.v1" +ATTACHMENT_PAYLOAD_SCHEMA = "attachment.stage.v1" +ATTACHMENT_RESULT_SCHEMA = "attachment.result.v1" +DEFAULT_ATTACHMENT_RETENTION_SECONDS = 86_400 +MAX_ATTACHMENT_RETENTION_SECONDS = 30 * 86_400 +_SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$") +_SAFE_CODE = re.compile(r"^[a-z][a-z0-9_]{0,63}$") + + +class AttachmentStagingError(RuntimeError): + """Stable, redacted attachment staging failure category.""" + + def __init__(self, code: str) -> None: + if _SAFE_CODE.fullmatch(code) is None: + raise ValueError("invalid attachment staging error code") + self.code = code + super().__init__("The attachment could not be staged safely.") + + +@dataclass(frozen=True, slots=True) +class AttachmentStageResult: + """Opaque result returned after one durable attachment stage operation.""" + + job: ExecutionJob + artifact: ExecutionArtifact + + +def _safe_id(value: Any, field: str) -> str: + if not isinstance(value, str) or _SAFE_ID.fullmatch(value) is None: + raise ValueError(f"{field} must be a bounded opaque identifier") + return value + + +def _canonical(value: Mapping[str, Any]) -> str: + return json.dumps( + dict(value), + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ) + + +class AttachmentStagingService: + """Create and complete owner-scoped ``attachment.stage.v1`` jobs.""" + + def __init__(self, repository: ExecutionRepository, boundary: ArtifactBoundary) -> None: + if not isinstance(repository, ExecutionRepository): + raise TypeError("repository must be an ExecutionRepository") + if not isinstance(boundary, ArtifactBoundary): + raise TypeError("boundary must be an ArtifactBoundary") + if boundary.repository is not repository: + raise ValueError("attachment boundary repository mismatch") + self.repository = repository + self.boundary = boundary + + def stage( + self, + *, + owner: str, + request_id: str, + content: bytes, + retention_seconds: int = DEFAULT_ATTACHMENT_RETENTION_SECONDS, + ) -> AttachmentStageResult: + """Validate, persist, and return one idempotent attachment stage.""" + + _safe_id(owner, "owner") + _safe_id(request_id, "request_id") + if not isinstance(content, bytes) or not content: + raise AttachmentStagingError("attachment_content_invalid") + if len(content) > self.boundary.max_input_bytes: + raise AttachmentStagingError("attachment_too_large") + if ( + isinstance(retention_seconds, bool) + or not isinstance(retention_seconds, int) + or not 1 <= retention_seconds <= MAX_ATTACHMENT_RETENTION_SECONDS + ): + raise AttachmentStagingError("attachment_retention_invalid") + try: + mime_type = sniff_artifact_mime(content) + except ArtifactBoundaryError: + raise AttachmentStagingError("attachment_invalid") from None + digest = sha256(content).hexdigest() + payload = { + "schema_version": ATTACHMENT_PAYLOAD_SCHEMA, + "sha256": digest, + "size": len(content), + "mime_type": mime_type, + "retention_seconds": retention_seconds, + } + try: + job, created = self.repository.create_job( + job_id=uuid4().hex, + owner=owner, + request_id=request_id, + profile=ATTACHMENT_STAGE_PROFILE, + payload=payload, + ) + except ExecutionRepositoryError: + raise AttachmentStagingError("attachment_persist_failed") from None + if not created: + return self._existing(job, payload) + + try: + artifact = self.boundary.stage_bytes( + job.job_id, + owner, + content, + retention_seconds=retention_seconds, + ) + except ArtifactBoundaryError as exc: + code = self._boundary_code(exc.code) + self._fail(job.job_id, code) + raise AttachmentStagingError(code) from None + result = self._result_payload(artifact, digest=digest, mime_type=mime_type) + try: + completed = self.repository.transition( + job.job_id, + status="succeeded", + event="completed", + phase="completed", + data={"message": "Attachment staged."}, + result=result, + ) + except Exception: + try: + self.repository.delete_artifact(artifact.artifact_id) + except Exception: + self._fail(job.job_id, "attachment_cleanup_pending") + raise AttachmentStagingError("attachment_cleanup_pending") from None + self._fail(job.job_id, "attachment_persist_failed") + raise AttachmentStagingError("attachment_persist_failed") from None + return AttachmentStageResult(job=completed, artifact=artifact) + + def _existing( + self, + job: ExecutionJob, + payload: Mapping[str, Any], + ) -> AttachmentStageResult: + try: + payload_matches = _canonical(job.payload) == _canonical(payload) + except (TypeError, ValueError): + payload_matches = False + if job.profile != ATTACHMENT_STAGE_PROFILE or not payload_matches: + raise AttachmentStagingError("request_conflict") + if job.status != "succeeded": + if job.status in {"queued", "running"}: + raise AttachmentStagingError("attachment_in_progress") + raise AttachmentStagingError("attachment_failed") + result = job.result + if not isinstance(result, Mapping): + raise AttachmentStagingError("attachment_result_invalid") + artifact_id = result.get("artifact_id") + if not isinstance(artifact_id, str) or _SAFE_ID.fullmatch(artifact_id) is None: + raise AttachmentStagingError("attachment_result_invalid") + artifact = self.repository.get_artifact(artifact_id, owner=job.owner) + if artifact is None: + raise AttachmentStagingError("attachment_artifact_unavailable") + if ( + set(result) != { + "schema_version", + "artifact_id", + "mime_type", + "size", + "sha256", + "expires_at", + } + or result.get("schema_version") != ATTACHMENT_RESULT_SCHEMA + or result.get("sha256") != artifact.sha256 + or result.get("size") != artifact.size + or result.get("mime_type") != artifact.mime_type + or result.get("expires_at") != artifact.expires_at + or artifact.sha256 != payload.get("sha256") + or artifact.size != payload.get("size") + or artifact.mime_type != payload.get("mime_type") + ): + raise AttachmentStagingError("attachment_result_invalid") + try: + content = self.repository.read_artifact(artifact.artifact_id) + if sha256(content).hexdigest() != artifact.sha256 or sniff_artifact_mime(content) != artifact.mime_type: + raise AttachmentStagingError("attachment_artifact_invalid") + except AttachmentStagingError: + raise + except (ExecutionRepositoryError, ArtifactBoundaryError): + raise AttachmentStagingError("attachment_artifact_unavailable") from None + return AttachmentStageResult(job=job, artifact=artifact) + + @staticmethod + def _result_payload( + artifact: ExecutionArtifact, + *, + digest: str, + mime_type: str, + ) -> Mapping[str, Any]: + return { + "schema_version": ATTACHMENT_RESULT_SCHEMA, + "artifact_id": artifact.artifact_id, + "mime_type": mime_type, + "size": artifact.size, + "sha256": digest, + "expires_at": artifact.expires_at, + } + + @staticmethod + def _boundary_code(code: str) -> str: + return { + "artifact_too_large": "attachment_too_large", + "artifact_content_invalid": "attachment_content_invalid", + "artifact_retention_invalid": "attachment_retention_invalid", + "artifact_publish_failed": "attachment_publish_failed", + }.get(code, "attachment_invalid") + + def _fail(self, job_id: str, code: str) -> None: + try: + self.repository.transition( + job_id, + status="failed", + event="failed", + phase="failed", + data={"message": "Attachment staging failed safely."}, + error=code, + ) + except Exception: + pass + + +__all__ = [ + "ATTACHMENT_PAYLOAD_SCHEMA", + "ATTACHMENT_RESULT_SCHEMA", + "ATTACHMENT_STAGE_PROFILE", + "AttachmentStageResult", + "AttachmentStagingError", + "AttachmentStagingService", + "DEFAULT_ATTACHMENT_RETENTION_SECONDS", + "MAX_ATTACHMENT_RETENTION_SECONDS", +] diff --git a/backend/cortex_backend/execution/native_recipe_attempt.py b/backend/cortex_backend/execution/native_recipe_attempt.py new file mode 100644 index 0000000..823235d --- /dev/null +++ b/backend/cortex_backend/execution/native_recipe_attempt.py @@ -0,0 +1,316 @@ +"""Per-job composition of the signed native recipe worker attempt. + +This module is the missing runtime binding between the durable recipe +coordinator and the already-reviewed Windows launcher/broker adapters. Every +attempt owns a fresh broker binder, pipe identity, suspended worker, and +authenticated protocol client. Failure at any step tears down the native +handles and is reduced to a stable coordinator error. There is no host +process, subprocess, or alternate transport fallback. +""" + +from __future__ import annotations + +from collections.abc import Callable +from queue import Empty, Queue +import os +import re +from threading import Thread +from typing import Any +from uuid import uuid4 + +from .bundle_installer import SignedBundleInstaller +from .models import ExecutionJob +from .native_launcher import ( + BrokerWorkerBinding, + NativeBrokerIdentityBinder, + NativeLauncherError, + NativeProcessFactory, + NativeSuspendedWorker, + NativeWorkerLauncher, + NativeWorkerPolicy, +) +from .recipe_coordinator import ( + DEFAULT_CANCEL_GRACE_SECONDS, + DEFAULT_WORKER_TIMEOUT_SECONDS, + RecipeExecutionError, + RecipeWorkerAttempt, + RecipeWorkerAttemptFactory, + RecipeWorkerClient, + RecipeWorkerOutput, +) + + +_PRINCIPAL = re.compile(r"^[0-9a-f]{64}$") +_SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$") +_SID = re.compile(r"^S-1-[0-9]+(?:-[0-9]+)+$") +_DEFAULT_ACCEPT_TIMEOUT_SECONDS = 15.0 + + +def _bounded_seconds(value: float, *, minimum: float, maximum: float, field: str) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError(f"{field} is invalid") + converted = float(value) + if not minimum <= converted <= maximum: + raise ValueError(f"{field} is invalid") + return converted + + +class NativeRecipeWorkerAttempt: + """Recipe worker client plus the native resources that keep it trusted.""" + + def __init__( + self, + client: RecipeWorkerClient, + *, + binder: NativeBrokerIdentityBinder, + worker: NativeSuspendedWorker, + accept_thread: Thread, + ) -> None: + if not isinstance(client, RecipeWorkerClient): + raise TypeError("client must be a RecipeWorkerClient") + self._client = client + self._binder = binder + self._worker = worker + self._accept_thread = accept_thread + self._closed = False + + def transform(self, *args: Any, **kwargs: Any) -> RecipeWorkerOutput: + return self._client.transform(*args, **kwargs) + + def cancel(self, reason: str = "user") -> None: + self._client.cancel(reason) + + def close(self) -> None: + if self._closed: + return + self._closed = True + try: + self._client.close() + finally: + try: + self._binder.close_binding() + finally: + try: + self._worker.close() + finally: + self._accept_thread.join(timeout=0.5) + + +class NativeRecipeWorkerAttemptFactory: + """Build one fresh signed/native attempt for each durable recipe job.""" + + def __init__( + self, + installer: SignedBundleInstaller, + *, + allowed_user_sids: frozenset[str], + process_factory_factory: Callable[[], NativeProcessFactory], + policy: NativeWorkerPolicy | None = None, + accept_timeout_seconds: float = _DEFAULT_ACCEPT_TIMEOUT_SECONDS, + worker_timeout_seconds: float = DEFAULT_WORKER_TIMEOUT_SECONDS, + cancel_grace_seconds: float = DEFAULT_CANCEL_GRACE_SECONDS, + binder_factory: Callable[..., NativeBrokerIdentityBinder] = NativeBrokerIdentityBinder, + launcher_factory: Callable[..., NativeWorkerLauncher] = NativeWorkerLauncher, + ) -> None: + if not isinstance(installer, SignedBundleInstaller): + raise TypeError("installer must be a SignedBundleInstaller") + if not isinstance(allowed_user_sids, frozenset) or not allowed_user_sids: + raise ValueError("allowed_user_sids must be a non-empty frozenset") + if any(not isinstance(sid, str) or _SID.fullmatch(sid) is None for sid in allowed_user_sids): + raise ValueError("allowed_user_sids are invalid") + if not callable(process_factory_factory): + raise TypeError("process_factory_factory must be callable") + if not callable(binder_factory) or not callable(launcher_factory): + raise TypeError("native adapter factories must be callable") + if policy is not None and not isinstance(policy, NativeWorkerPolicy): + raise TypeError("policy must be a NativeWorkerPolicy") + self._installer = installer + self._allowed_user_sids = allowed_user_sids + self._process_factory_factory = process_factory_factory + self._policy = policy or NativeWorkerPolicy() + self._accept_timeout = _bounded_seconds( + accept_timeout_seconds, + minimum=0.1, + maximum=120.0, + field="accept_timeout_seconds", + ) + self._worker_timeout = _bounded_seconds( + worker_timeout_seconds, + minimum=0.1, + maximum=600.0, + field="worker_timeout_seconds", + ) + self._cancel_grace = _bounded_seconds( + cancel_grace_seconds, + minimum=0.1, + maximum=30.0, + field="cancel_grace_seconds", + ) + self._binder_factory = binder_factory + self._launcher_factory = launcher_factory + + def __call__(self, job: ExecutionJob) -> RecipeWorkerAttempt: + if not isinstance(job, ExecutionJob): + raise RecipeExecutionError("worker_job_invalid") + if _PRINCIPAL.fullmatch(job.owner) is None: + raise RecipeExecutionError("worker_principal_invalid") + if _SAFE_ID.fullmatch(job.job_id) is None: + raise RecipeExecutionError("worker_job_invalid") + + binder: NativeBrokerIdentityBinder | None = None + worker: NativeSuspendedWorker | None = None + accept_thread: Thread | None = None + connection: Any | None = None + accepted: Queue[object] | None = None + try: + binder = self._binder_factory(allowed_user_sids=self._allowed_user_sids) + if not all( + callable(getattr(binder, name, None)) + for name in ("accept", "close_binding") + ): + raise RecipeExecutionError("worker_broker_invalid") + process_factory = self._process_factory_factory() + if not callable(getattr(process_factory, "create_suspended", None)): + raise RecipeExecutionError("worker_process_factory_invalid") + launcher = self._launcher_factory( + self._installer, + process_factory=process_factory, + broker_binder=binder, + ) + if not callable(getattr(launcher, "launch", None)): + raise RecipeExecutionError("worker_launcher_invalid") + binding = BrokerWorkerBinding( + pipe_name=rf"\\.\pipe\cortex-recipe-{uuid4().hex}", + broker_process_id=os.getpid(), + installation_principal_id=job.owner, + job_id=job.job_id, + ) + worker = launcher.launch(binding, self._policy) + if not callable(getattr(worker, "close", None)): + raise RecipeExecutionError("worker_handle_invalid") + + accepted = Queue(maxsize=1) + + def accept() -> None: + try: + connection = binder.accept( + owner_for_job=lambda requested: ( + job.owner if requested == job.job_id else None + ) + ) + accepted.put(connection) + except Exception as error: # pragma: no cover - native failure path. + accepted.put(error) + + accept_thread = Thread( + target=accept, + name=f"cortex-native-broker-{job.job_id}", + daemon=True, + ) + accept_thread.start() + try: + value = accepted.get(timeout=self._accept_timeout) + except Empty: + raise RecipeExecutionError("worker_broker_timeout") from None + if isinstance(value, Exception): + raise RecipeExecutionError("worker_broker_failed") from None + if not all(callable(getattr(value, name, None)) for name in ("send_message", "receive_message", "close")): + raise RecipeExecutionError("worker_connection_invalid") + connection = value + client = RecipeWorkerClient( + connection, + installation_principal_id=job.owner, + timeout_seconds=self._worker_timeout, + cancel_grace_seconds=self._cancel_grace, + ) + return NativeRecipeWorkerAttempt( + client, + binder=binder, + worker=worker, + accept_thread=accept_thread, + ) + except RecipeExecutionError: + if connection is not None: + try: + connection.close() + except Exception: + pass + self._cleanup(binder, worker, accept_thread=accept_thread, accepted=accepted) + raise + except NativeLauncherError: + if connection is not None: + try: + connection.close() + except Exception: + pass + self._cleanup(binder, worker, accept_thread=accept_thread, accepted=accepted) + raise RecipeExecutionError("worker_launch_failed") from None + except Exception: + if connection is not None: + try: + connection.close() + except Exception: + pass + self._cleanup(binder, worker, accept_thread=accept_thread, accepted=accepted) + raise RecipeExecutionError("worker_launch_failed") from None + + @staticmethod + def _cleanup( + binder: NativeBrokerIdentityBinder | None, + worker: NativeSuspendedWorker | None, + *, + accept_thread: Thread | None = None, + accepted: Queue[object] | None = None, + ) -> None: + if binder is not None: + try: + binder.close_binding() + except Exception: + pass + if worker is not None: + try: + worker.close() + except Exception: + pass + if accept_thread is not None: + accept_thread.join(timeout=0.5) + if accepted is not None: + try: + value = accepted.get_nowait() + except Empty: + return + if callable(getattr(value, "close", None)): + try: + value.close() + except Exception: + pass + + +def build_native_recipe_worker_attempt_factory( + installer: SignedBundleInstaller, + *, + allowed_user_sids: frozenset[str], + process_factory_factory: Callable[[], NativeProcessFactory], + policy: NativeWorkerPolicy | None = None, + accept_timeout_seconds: float = _DEFAULT_ACCEPT_TIMEOUT_SECONDS, + worker_timeout_seconds: float = DEFAULT_WORKER_TIMEOUT_SECONDS, + cancel_grace_seconds: float = DEFAULT_CANCEL_GRACE_SECONDS, +) -> RecipeWorkerAttemptFactory: + """Return the explicit signed/native worker factory for qualification wiring.""" + + return NativeRecipeWorkerAttemptFactory( + installer, + allowed_user_sids=allowed_user_sids, + process_factory_factory=process_factory_factory, + policy=policy, + accept_timeout_seconds=accept_timeout_seconds, + worker_timeout_seconds=worker_timeout_seconds, + cancel_grace_seconds=cancel_grace_seconds, + ) + + +__all__ = [ + "NativeRecipeWorkerAttempt", + "NativeRecipeWorkerAttemptFactory", + "build_native_recipe_worker_attempt_factory", +] diff --git a/backend/cortex_backend/execution/qualification.py b/backend/cortex_backend/execution/qualification.py index 60c3027..e71eab4 100644 --- a/backend/cortex_backend/execution/qualification.py +++ b/backend/cortex_backend/execution/qualification.py @@ -30,6 +30,9 @@ RecipeExecutionCoordinator, RecipeWorkerAttemptFactory, ) +from .native_launcher import NativeProcessFactory, NativeWorkerPolicy +from .native_recipe_attempt import build_native_recipe_worker_attempt_factory +from .bundle_installer import SignedBundleInstaller from .release_gate import RecipeRuntimeReleaseGate from .repository import ExecutionRepository @@ -96,6 +99,43 @@ def factory(repository: ExecutionRepository) -> LifecycleCoordinator: return factory +def build_native_recipe_coordinator_factory( + installer: SignedBundleInstaller, + *, + allowed_user_sids: frozenset[str], + process_factory_factory: Callable[[], NativeProcessFactory], + policy: NativeWorkerPolicy | None = None, + accept_timeout_seconds: float = 15.0, + worker_timeout_seconds: float = 120.0, + cancel_grace_seconds: float = 5.0, + artifact_boundary_factory: Callable[[ExecutionRepository], ArtifactBoundary] | None = None, + lease_seconds: float = 30.0, + supervisor_lease_seconds: float = 30.0, +) -> CoordinatorFactory: + """Bind the signed/native attempt factory to the lifecycle repository. + + The process factory is deliberately required as an explicit composition + input. This helper creates no process while configuring the lifecycle and + has no host-process or alternate-transport fallback. + """ + + worker_factory = build_native_recipe_worker_attempt_factory( + installer, + allowed_user_sids=allowed_user_sids, + process_factory_factory=process_factory_factory, + policy=policy, + accept_timeout_seconds=accept_timeout_seconds, + worker_timeout_seconds=worker_timeout_seconds, + cancel_grace_seconds=cancel_grace_seconds, + ) + return build_recipe_coordinator_factory( + worker_factory, + artifact_boundary_factory=artifact_boundary_factory, + lease_seconds=lease_seconds, + supervisor_lease_seconds=supervisor_lease_seconds, + ) + + @dataclass(frozen=True, slots=True) class QualificationLifecycleConfig: """Controls required before the local qualification lifecycle can start.""" @@ -229,6 +269,7 @@ def build_execution_lifecycle( "QualificationLifecycleConfig", "QualificationProfileError", "build_execution_lifecycle", + "build_native_recipe_coordinator_factory", "build_recipe_coordinator_factory", "parse_execution_profile", ] diff --git a/contracts/cortex-api.ts b/contracts/cortex-api.ts index 970ce71..d5c192e 100644 --- a/contracts/cortex-api.ts +++ b/contracts/cortex-api.ts @@ -16,6 +16,25 @@ export interface AppearanceSettings { theme?: "light" | "dark" | "system"; } +export interface AttachmentStageAccepted { + job_id: string; + request_id: string; + profile: "attachment.stage.v1"; + status: "succeeded"; + sequence: number; + artifact_id: string; + mime_type: string; + size: number; + sha256: string; + expires_at: string; +} + +export interface AttachmentStageRequest { + request_id: string; + content_base64: string; + retention_seconds?: number; +} + export interface BrightnessStep { op: "brightness"; factor: number | string; diff --git a/contracts/openapi.json b/contracts/openapi.json index a37024c..82f7378 100644 --- a/contracts/openapi.json +++ b/contracts/openapi.json @@ -84,6 +84,100 @@ "title": "AppearanceSettings", "type": "object" }, + "AttachmentStageAccepted": { + "additionalProperties": false, + "properties": { + "artifact_id": { + "title": "Artifact Id", + "type": "string" + }, + "expires_at": { + "format": "date-time", + "title": "Expires At", + "type": "string" + }, + "job_id": { + "title": "Job Id", + "type": "string" + }, + "mime_type": { + "title": "Mime Type", + "type": "string" + }, + "profile": { + "const": "attachment.stage.v1", + "title": "Profile", + "type": "string" + }, + "request_id": { + "title": "Request Id", + "type": "string" + }, + "sequence": { + "title": "Sequence", + "type": "integer" + }, + "sha256": { + "title": "Sha256", + "type": "string" + }, + "size": { + "title": "Size", + "type": "integer" + }, + "status": { + "const": "succeeded", + "title": "Status", + "type": "string" + } + }, + "required": [ + "job_id", + "request_id", + "profile", + "status", + "sequence", + "artifact_id", + "mime_type", + "size", + "sha256", + "expires_at" + ], + "title": "AttachmentStageAccepted", + "type": "object" + }, + "AttachmentStageRequest": { + "additionalProperties": false, + "description": "Bounded base64 envelope for a user attachment.\n\nThe service decodes and MIME-sniffs the bytes before an artifact is\npublished. The encoded payload is never written to the durable job.", + "properties": { + "content_base64": { + "maxLength": 14680064, + "minLength": 4, + "title": "Content Base64", + "type": "string" + }, + "request_id": { + "maxLength": 128, + "minLength": 1, + "pattern": "^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$", + "title": "Request Id", + "type": "string" + }, + "retention_seconds": { + "default": 86400, + "maximum": 2592000.0, + "minimum": 1.0, + "title": "Retention Seconds", + "type": "integer" + } + }, + "required": [ + "request_id", + "content_base64" + ], + "title": "AttachmentStageRequest", + "type": "object" + }, "BrightnessStep": { "additionalProperties": false, "properties": { @@ -2525,6 +2619,45 @@ "summary": "Diagnostics" } }, + "/api/v1/execution/attachments": { + "post": { + "description": "Stage one bounded attachment for a qualified recipe request.\n\nThe route accepts only a bounded base64 envelope. Decoded bytes pass\nthrough the same owner-scoped artifact boundary used by recipe output;\nno caller path, filename, or executable instruction is accepted.", + "operationId": "stage_attachment_api_v1_execution_attachments_post", + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/AttachmentStageRequest" + } + } + }, + "required": true + }, + "responses": { + "201": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/AttachmentStageAccepted" + } + } + }, + "description": "Successful Response" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "summary": "Stage Attachment" + } + }, "/api/v1/execution/preview/fake": { "post": { "operationId": "start_fake_execution_api_v1_execution_preview_fake_post", diff --git a/docs/adr/0001-capability-tiered-agentic-execution-harness.md b/docs/adr/0001-capability-tiered-agentic-execution-harness.md index fc8be3c..c00cdb5 100644 --- a/docs/adr/0001-capability-tiered-agentic-execution-harness.md +++ b/docs/adr/0001-capability-tiered-agentic-execution-harness.md @@ -1000,11 +1000,16 @@ timeouts; this is the open-source qualification evidence and does not need production trust material. Explicit lifecycle composition for `release_profile="qualification"` is now available through the documented local/CI-only builder; the application remains default-off. The durable recipe -coordinator, typed `POST /api/v1/execution/recipe/image` request surface, and -generated TypeScript client method are available only after that lifecycle is -ready and consume an opaque artifact ID staged by a separate trusted boundary. -The next core slice is trusted attachment staging and binding the real -signed/native broker worker into the injected attempt factory. The separate +coordinator, typed `POST /api/v1/execution/recipe/image` request surface, +qualification-only `POST /api/v1/execution/attachments` staging surface, and +generated TypeScript client methods are available only after that lifecycle is +ready. Attachment staging stores only bounded, MIME-sniffed bytes behind an +opaque artifact ID. `build_native_recipe_coordinator_factory` composes a fresh +signed/native launcher, broker binder, protected pipe identity, and authenticated +client for each recipe job; it requires an explicit reviewed process-factory +injection and never falls back to a host process. The next core slice is an +end-to-end durable-coordinator run through the packaged signed worker, including +cancellation, retention, atomic publication, and native cleanup. The separate `recipe.release-review.v1` verifier is available as optional release hardening; it remains observation-only and no approval record or production trust material is committed. diff --git a/docs/adr/0001-phase2-artifact-boundary.md b/docs/adr/0001-phase2-artifact-boundary.md index cac20e6..6296465 100644 --- a/docs/adr/0001-phase2-artifact-boundary.md +++ b/docs/adr/0001-phase2-artifact-boundary.md @@ -28,7 +28,11 @@ and [MoveFileEx](https://learn.microsoft.com/en-us/windows/win32/api/winbase/nf- `ArtifactBoundary` is the only Phase 2 API allowed to copy a user source into the artifact store or publish provider output. It is owner-scoped, fail-closed, and -provider-independent. +provider-independent. In addition to explicit path grants, it exposes +`stage_bytes(job_id, owner, content, retention_seconds)` for a trusted caller +that already holds bounded attachment bytes. This method shares the same owner +check, byte limit, MIME sniffer, digest-derived name, retention, and atomic +repository publication as path copy-in; it never accepts a filename or path. ### 1. Explicit source grants and immutable snapshots @@ -55,6 +59,13 @@ stored as `application/octet-stream`; this is metadata safety, not permission to or execute them. Production image decoding remains behind the separate sandboxed [recipe provider qualification gate](0001-phase2-recipe-provider.md). +The qualification-only `AttachmentStagingService` wraps this byte capability in +an idempotent `attachment.stage.v1` job. Its durable payload contains only the +schema, digest, size, derived MIME, and retention. A matching retry revalidates +the stored artifact; a changed payload with the same owner/request key is a +stable conflict. The API returns an opaque artifact ID and safe metadata, never +the encoded payload or repository path. + ### 3. Private output staging and exact claims The provider receives a private staging directory and returns a finite list of @@ -88,7 +99,10 @@ The boundary exposes only categories such as `artifact_owner_mismatch`, `artifact_source_changed`, `artifact_too_large`, `invalid_artifact`, `artifact_unclaimed_output`, `artifact_mime_mismatch`, `artifact_output_limit`, `artifact_publish_failed`, and `artifact_cleanup_pending`. Raw paths and operating -system details do not cross the application boundary. +system details do not cross the application boundary. Byte staging additionally +uses `attachment_content_invalid`, `attachment_too_large`, +`attachment_retention_invalid`, and `attachment_cleanup_pending` at its service +boundary. ## Verification and evidence @@ -96,7 +110,10 @@ system details do not cross the application boundary. ADS/link/reparse rejection, source mutation, sparse-file rejection, exact claims, quarantine, active/archive rejection, executable rejection, finite-number checks, MIME mismatch, aggregate limits, all-or-nothing publication rollback, and stored-size -integrity. The deterministic `tools/execution_spikes/artifact_security_review.py` +integrity. `tests/test_phase2_attachment_staging.py` additionally covers bounded +byte input, idempotency, active/archive rejection, retention, and terminal +artifact revalidation. The deterministic +`tools/execution_spikes/artifact_security_review.py` probe repeats the fixed 12-case corpus in a disposable temporary root and reports `blocked` when a required host link primitive is unavailable. The complete matrix is recorded in the [Phase 2 evidence log](0001-phase2-evidence.md). diff --git a/docs/adr/0001-phase2-evidence.md b/docs/adr/0001-phase2-evidence.md index 185df0d..5cf5de6 100644 --- a/docs/adr/0001-phase2-evidence.md +++ b/docs/adr/0001-phase2-evidence.md @@ -26,7 +26,7 @@ | Fixed worker protocol and package closure | **Complete (qualification-only)** | `worker_protocol.py` and `worker_runtime.py` enforce bounded prepare/chunk/complete/cancel/collect state, authenticated envelope identity, concurrent cancellation, redacted output/errors, and no-capability bodies. `packaging/recipe_worker/recipe_worker.spec` builds the fixed `recipe_worker.exe` (Windows build verified 2026-07-23); the entrypoint accepts only the fixed native-broker identity arguments and returns `78` on direct or failed launches. | | Authenticated broker contract | **Complete (transport-neutral)** | Bounded versioned frames, direction-specific HMAC keys, canonical messages, peer ACL/integrity policy, and owner-scoped authorization are covered by adversarial tests. | | Native named-pipe adapter/DACL/peer-token binding | **Complete (transport-only)** | Protected local pipe, expected PID, OS token identity, X25519/HKDF handshake, direction keys, and close-on-error lifecycle are covered by native broker tests. | -| User-artifact copy-in, output validation, and publication | **Complete (boundary only)** | Explicit owner/turn grants, bounded stable snapshots, link/reparse/hardlink/sparse/ADS rejection, byte-derived MIME policy, exact output claims, quarantine, hash/size limits, atomic repository publication, rollback, and cleanup categories are covered by `tests/test_phase2_artifact_boundary.py`. | +| User-artifact copy-in, byte staging, output validation, and publication | **Complete (qualification boundary)** | Explicit owner/turn grants and bounded in-memory attachment staging, 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` and `tests/test_phase2_attachment_staging.py`. | | Deterministic artifact security review | **Complete (qualification-only)** | `artifact_security_review.py --json --strict` passed the fixed 12-case disposable corpus (`artifact-boundary-review.v1`, digest `a748cc9f0a514c8d`): owner/path binding, link/hardlink rejection, active/non-finite content, exact claims/quarantine, rollback, and repository size integrity. Missing link primitives remain blocked. | | Deterministic resource/watchdog accounting | **Complete (qualification-only)** | `resource_watchdog_qualification.py --json --strict` passed the fixed `resource-watchdog.v1` corpus (digest `5eac03e2b4981543`): immutable ADR budgets, wall/idle watchdogs, clock and cumulative-sample regression, stable CPU/memory/input/output/console/observation/message limit precedence, missing-memory fail-closed behavior, actual Windows Job Object accounting, and kill-on-close process-tree reaping. The explicit local/qualification lifecycle composition now consumes this evidence; external review is not required for this open-source path. | | Release/lifecycle health preflight | **Complete (official + qualification profiles)** | `RecipeRuntimeReleaseGate` defaults to `release_profile="official"` and requires the reviewed native process factory, live broker binder, and external-review result. An explicit `release_profile="qualification"` records `qualification_profile` and omits only the optional outside-review requirement for local/CI development; it performs no launch, broker bind, provider load, or lifecycle mutation. | @@ -35,9 +35,9 @@ | 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, and checks cancellation. The provider remains separate from the API; the typed coordinator route is qualification-only and default-off. | | Windows recipe sandbox qualification harness | **Complete (signed launch/broker/hostile/cancellation/resource/watchdog)** | `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 fixed PNG transform, truncated-PNG decoder rejection, active-SVG decoder rejection, and in-flight cancellation corpus. `resource_watchdog_qualification.py` separately proves immutable budgets, actual Job Object accounting, and kill-on-close tree reaping. The native broker uses bounded availability polling before reads so the cancellation reader remains live while the packaged provider transforms. | | 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 | **Complete (qualification; default-off in the app)** | Signed installation, provenance, AppContainer/job identity, broker handshake, `prepare`, `input_chunk`, `input_complete`, `collect_output`, hostile decoder rejection, in-flight cancellation acknowledgement, artifact security review, resource accounting, and watchdog tree reaping are qualified. The explicit lifecycle composition, durable recipe coordinator/publication path, and typed qualification-only API are wired behind injection; attachment staging and the real attempt-factory binding remain the next gate. Official-release review/signing remains optional hardening. | +| OS sandbox provider and provider-produced image outputs | **Complete (qualification; default-off in the app)** | Signed installation, provenance, AppContainer/job identity, broker handshake, `prepare`, `input_chunk`, `input_complete`, `collect_output`, hostile decoder rejection, in-flight cancellation acknowledgement, artifact security review, resource accounting, and watchdog tree reaping are qualified. The explicit lifecycle composition, durable recipe coordinator/publication path, typed qualification-only API, trusted attachment staging, and per-job signed/native attempt factory are wired behind injection; the next gate is a durable-coordinator run through the packaged worker. Official-release review/signing remains optional hardening. | | Explicit qualification-profile lifecycle wiring | **Complete (local/CI composition; default-off in the app)** | `build_execution_lifecycle()` accepts only exact `disabled`/`qualification` selection, requires a qualification release gate, coordinator factory, and provider-health probe, and composes them in a fail-closed order. `Cortex_Preview.build_preview_app` exposes this only as an explicit injection; the normal app supplies no profile or controls. | -| Durable recipe coordinator/request and artifact publication | **Complete (qualification-only API; default-off in the app)** | `RecipeExecutionCoordinator` persists only opaque artifact IDs and canonical plan digests, enforces owner-scoped input reads and idempotency conflicts, leases/recoveries/cancels attempts, validates worker envelopes/chunks, and publishes exactly one output through `ArtifactBoundary.collect_outputs`. `POST /api/v1/execution/recipe/image` accepts only a typed plan plus an owner-scoped, already-staged artifact ID after a ready `qualification` lifecycle. `build_recipe_coordinator_factory` binds an injected worker-attempt factory without host fallback. `tests/test_phase2_recipe_coordinator.py`, `tests/test_phase2_recipe_api.py`, and `tests/test_phase2_qualification_lifecycle.py` cover the hostile, cancellation, API, and lifecycle paths. | +| Durable recipe coordinator/request, attachment staging, and artifact publication | **Complete (qualification-only API; default-off in the app)** | `RecipeExecutionCoordinator` persists only opaque artifact IDs and canonical plan digests, enforces owner-scoped input reads and idempotency conflicts, leases/recoveries/cancels attempts, validates worker envelopes/chunks, and publishes exactly one output through `ArtifactBoundary.collect_outputs`. `POST /api/v1/execution/attachments` creates an idempotent owner-scoped `attachment.stage.v1` record through `ArtifactBoundary.stage_bytes`; `POST /api/v1/execution/recipe/image` consumes its opaque artifact ID after a ready `qualification` lifecycle. `build_native_recipe_coordinator_factory` binds a fresh signed/native launcher/broker/client attempt per job without host fallback. `tests/test_phase2_recipe_coordinator.py`, `tests/test_phase2_recipe_api.py`, `tests/test_phase2_attachment_staging.py`, `tests/test_phase2_native_recipe_attempt.py`, and `tests/test_phase2_qualification_lifecycle.py` cover the hostile, cancellation, API, native-composition, and lifecycle paths. | ## Security invariants @@ -83,41 +83,44 @@ while quarantine/cleanup failures surface for supervisor recovery. 18. Artifact records are opaque IDs; repository read/delete/purge operations remain confined to the configured artifact root and verify the stored SHA-256. -19. The fixed-function provider accepts only immutable bytes and parsed plans, uses an +19. Attachment staging accepts only bounded bytes through the trusted boundary; + durable records contain no base64 payload, source path, filename, or command. + Matching retries revalidate the artifact and changed retries conflict. +20. The fixed-function provider accepts only immutable bytes and parsed plans, uses an independent format allowlist, treats decoder warnings as errors, rejects multiple frames, enforces hard byte/pixel/dimension/memory/step caps, and revalidates output. -20. Provider startup requires an external available sandbox health result; dependency +21. Provider startup requires an external available sandbox health result; dependency or codec failure, cancellation, decoder failure, and output metadata/size failure leave the provider disabled and return stable categories only. -21. The sandbox qualification harness never authorizes a provider launch from a +22. The sandbox qualification harness never authorizes a provider launch from a missing, unsigned, or merely present worker directory; it reports `blocked` and never falls back to host-process decoding. -22. Worker provenance is storage-only: only an installer-validated immutable +23. Worker provenance is storage-only: only an installer-validated immutable generation with one exact `image_transform`/`recipe_worker.exe` role and stable byte identity can proceed to a future launcher; no executable is loaded here. -23. The disposable launcher applies all required Job Object policy before resume, +24. The disposable launcher applies all required Job Object policy before resume, queries configured limits plus actual CPU, memory, process, and I/O accounting, never grants breakaway, and reports the absent worker/broker gates as blocking. -24. Release signing reads an external raw private key only for the signing operation, +25. 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 +26. 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. -26. Native broker reads poll bounded pipe availability before synchronous reads so a +27. Native broker reads poll bounded pipe availability before synchronous reads so a cancellation reader cannot starve the provider transform; pipe errors still fail closed and do not bypass framing, authentication, or sequence checks. -27. Hostile decoder bytes are rejected inside the signed worker, and cancellation is +28. Hostile decoder bytes are rejected inside the signed worker, and cancellation is sent only after `input_complete` over the authenticated broker; a missing or ambiguous terminal response remains a blocked qualification result. -28. Typed parser fuzzing uses a fixed seed, bounded iteration count, bounded payload +29. Typed parser fuzzing uses a fixed seed, bounded iteration count, bounded payload depth, and stable redacted error categories; an unexpected exception or budget violation is a blocked qualification result. -29. Artifact security qualification uses fixed bytes and a disposable temporary root; +30. Artifact security qualification uses fixed bytes and a disposable temporary root; it checks owner/path binding, active-content and numeric safety, link rejection, exact claims, quarantine, rollback, and stored-size integrity. The probe accepts no user/model input and reports an unavailable required link primitive as blocked. -30. Resource budgets are immutable and bounded; watchdog clocks must be finite and +31. Resource budgets are immutable and bounded; watchdog clocks must be finite and monotonic; cumulative CPU, memory, byte, and message samples cannot regress; limit precedence is stable; and a terminal result without required accounting remains unavailable rather than being reported as a green qualification. @@ -131,10 +134,12 @@ python -m pytest tests/test_phase2_broker.py -q python -m pytest tests/test_phase2_native_broker.py -q python -m pytest tests/test_phase2_bundle_installer.py -q python -m pytest tests/test_phase2_artifact_boundary.py -q +python -m pytest tests/test_phase2_attachment_staging.py -q python -m pytest tests/test_phase2_recipe_provider.py -q python -m pytest tests/test_phase2_worker_provenance.py -q python -m pytest tests/test_phase2_worker_release.py -q python -m pytest tests/test_native_launcher_qualification.py -q +python -m pytest tests/test_phase2_native_recipe_attempt.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 @@ -312,5 +317,19 @@ production build passed, generated OpenAPI/TypeScript contracts were refreshed, `compileall` passed, and `git diff --check` passed. The new `build_recipe_coordinator_factory` binds only an injected worker-attempt factory to the lifecycle-owned repository; it creates no process, broker, provider, or -host fallback. Attachment staging and real signed/native attempt binding remain -the next planned slice. +host fallback. + +**Attachment/native composition stage verification (2026-07-29):** The trusted +byte boundary now exposes `ArtifactBoundary.stage_bytes`, and the qualification +API adds an idempotent owner-scoped `attachment.stage.v1` route with a bounded +base64 envelope. The stage service persists only digest/size/MIME/retention, +revalidates matching retries, rejects changed retries, and never returns a path +or payload. The native composition adds a fresh signed launcher, broker binder, +pipe identity, authenticated client, and bounded cleanup scope per recipe job; +the explicit coordinator helper requires a reviewed process-factory injection +and creates no process during configuration. Focused verification passed 52 +tests across artifact boundary, attachment staging, native composition, API, +and launcher suites; the qualification route remains unavailable in the normal +default-off app. The next gate is an end-to-end durable-coordinator run through +the packaged signed worker, including cancellation, retention, publication, and +native cleanup. diff --git a/docs/adr/0001-phase2-qualification-lifecycle.md b/docs/adr/0001-phase2-qualification-lifecycle.md index 03294f3..8a8e97c 100644 --- a/docs/adr/0001-phase2-qualification-lifecycle.md +++ b/docs/adr/0001-phase2-qualification-lifecycle.md @@ -34,10 +34,15 @@ source application remain `disabled` by default. The helper `build_recipe_coordinator_factory` now binds an injected, already-qualified worker-attempt factory to the lifecycle-owned repository; it does not discover a package, launch a process, bind a broker, or provide a host fallback. The -qualification-only recipe API route consumes an opaque, already-staged artifact -ID and remains unavailable unless this lifecycle is ready. Attachment staging, -automatic model tool selection, worker-package discovery, and persistent -signing/trust material remain separate gates. +qualification-only recipe API route consumes an opaque artifact ID and remains +unavailable unless this lifecycle is ready. The companion +`POST /api/v1/execution/attachments` route uses the same ready lifecycle and +stages bounded bytes through the trusted artifact boundary; it does not discover +paths or accept executable content. `build_native_recipe_coordinator_factory` +provides the explicit signed/native attempt composition when callers supply the +installer, allowed user SIDs, and reviewed process-factory factory. Automatic +model tool selection, worker-package discovery, and persistent signing/trust +material remain separate gates. ## Failure and recovery behavior @@ -59,12 +64,14 @@ qualification code without an outside reviewer, production signing key, or trusted release root. The normal application cannot be enabled accidentally by importing the provider or setting an implicit default. -The recipe-specific coordinator/request and artifact-publication path plus its -typed qualification-only API surface are now implemented behind this lifecycle +The recipe-specific coordinator/request, trusted attachment staging, native +attempt composition, and artifact-publication path plus their typed +qualification-only API surfaces are now implemented behind this lifecycle boundary. It preserves the same release gate, native launcher, broker identity, -resource/watchdog, and trusted artifact controls. The next slice is trusted -attachment staging and binding the real signed/native broker worker into the -attempt factory; it must remain default-off in the application. +resource/watchdog, and trusted artifact controls. The next slice is an +end-to-end durable-coordinator run against the packaged signed worker, including +cancellation, retention, publication, and native cleanup; it must remain +default-off in the application. ## Verification @@ -72,5 +79,6 @@ attempt factory; it must remain default-off in the application. default-off behavior, missing configuration, official-gate rejection, blocked health ordering, provider-health composition, coordinator startup, profile diagnostics, clean stop, and repository-bound worker-attempt factory wiring. -`tests/test_phase2_recipe_api.py` covers the route's ready-lifecycle gate and -owner/idempotency/error behavior. +`tests/test_phase2_recipe_api.py` covers both routes' ready-lifecycle gates and +owner/idempotency/error behavior. `tests/test_phase2_native_recipe_attempt.py` +covers explicit native composition without launch during configuration. diff --git a/docs/adr/0001-phase2-recipe-coordinator.md b/docs/adr/0001-phase2-recipe-coordinator.md index 77c85b2..c382c75 100644 --- a/docs/adr/0001-phase2-recipe-coordinator.md +++ b/docs/adr/0001-phase2-recipe-coordinator.md @@ -1,15 +1,17 @@ # ADR-0001 Phase 2 durable recipe coordinator and artifact publication - **Status:** Implemented and verified behind an explicit qualification-only - API boundary; the normal application remains default-off + API boundary, including trusted attachment staging and signed/native attempt + composition; the normal application remains default-off - **Phase:** 2 - fixed-function image recipe - **Parent:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md) - **Depends on:** [typed recipe contract](0001-phase2-recipe-contract.md), [trusted artifact boundary](0001-phase2-artifact-boundary.md), [worker protocol](0001-phase2-worker-protocol.md), and [qualification lifecycle](0001-phase2-qualification-lifecycle.md) -- **Scope:** Durable request/attempt coordination, cancellation and recovery, - and all-or-nothing publication for the fixed image recipe +- **Scope:** Durable attachment staging, signed/native request-attempt + composition, cancellation and recovery, and all-or-nothing publication for + the fixed image recipe ## Decision @@ -28,12 +30,26 @@ identifier must match the request identifier exactly. No source path, filename, command, model name, shell text, network target, or executable authority is accepted or written to durable job state. +`POST /api/v1/execution/attachments` is the qualification-only input boundary. +It accepts a bounded base64 envelope, decodes it in memory, and passes the bytes +through `ArtifactBoundary.stage_bytes`; the durable `attachment.stage.v1` record +stores only the content digest, size, derived MIME, and bounded retention. Its +response contains an opaque artifact ID and safe metadata, never a path or the +encoded payload. + The worker seam is `RecipeWorkerAttempt`. A factory receives the durable job and -returns one already-authenticated attempt. The attempt receives immutable input +returns one fresh, already-authenticated attempt. The native implementation +`NativeRecipeWorkerAttemptFactory` creates a new `NativeBrokerIdentityBinder`, +`NativeWorkerLauncher`, protected pipe binding, suspended AppContainer worker, +and `RecipeWorkerClient` per job. It requires an injected reviewed process +factory and signed installer, binds the broker PID/principal/job identity before +resume, waits only within a bounded accept window, and closes the client, +broker, and process tree on every failure or cancellation. It has no host +process or alternate transport fallback. The attempt receives immutable input bytes and a typed plan, and returns `RecipeWorkerOutput` with byte-derived MIME, format, dimensions, and SHA-256. The concrete `RecipeWorkerClient` speaks only -the existing authenticated broker/worker protocol; it does not open a process, -choose a provider, or invent a fallback transport. +the existing authenticated broker/worker protocol; it does not choose a +provider or invent a fallback transport. ## Durable lifecycle @@ -50,15 +66,18 @@ choose a provider, or invent a fallback transport. payload, and resumes only the known recipe profile. Invalid recovery metadata becomes `recovery_invalid_payload`; it is never interpreted as executable input. -4. Cancellation sets a durable `cancelling` state, signals the worker attempt, +4. Attachment staging is idempotent on `(owner, request_id)`. A duplicate + matching payload revalidates the stored artifact bytes and returns the same + opaque artifact; a different payload is a stable `request_conflict`. +5. Cancellation sets a durable `cancelling` state, signals the worker attempt, and waits only within the worker's bounded cancellation path. A cancelled attempt cannot publish a successful result. -5. A valid output is written to a private temporary staging directory under the +6. A valid output is written to a private temporary staging directory under the artifact root. `ArtifactBoundary.collect_outputs` requires the exact single `output` claim, re-sniffs bytes, enforces size/link/reparse/hash limits, and publishes atomically. Any publication failure rolls back records and quarantines or reports cleanup failure through a stable category. -6. The terminal result contains only the published artifact ID, safe MIME/format, +7. The terminal result contains only the published artifact ID, safe MIME/format, size, digest, dimensions, and plan digest. It never returns the staging or repository path. A cancellation race after publication deletes the unpublished records before recording `cancelled`. @@ -84,20 +103,21 @@ tokens, and stack traces do not enter job events or results. Terminal repository state is immutable, so late worker callbacks cannot overwrite a validated result. The worker factory is deliberately injected. The qualification helper -`build_recipe_coordinator_factory` binds that already-qualified attempt factory -to the lifecycle-owned repository without creating a process, binding a broker, -loading a provider, or falling back to host execution. This stage does not +`build_recipe_coordinator_factory` remains the generic seam, while +`build_native_recipe_coordinator_factory` composes the signed/native attempt +factory when a caller supplies the installer, allowed user SIDs, and reviewed +process-factory factory. Configuration creates no process. This stage does not pretend that an unsigned package, missing broker identity binding, or absent production trust root is a usable runtime; those remain release/qualification composition inputs owned by the existing lifecycle and launcher gates. ## Explicit non-goals -This ADR does not add an attachment-upload/staging route, automatic model tool -selection, arbitrary Python/WASI execution, source-path access, direct user-file -mutation, network access, application-exit persistence, production signing, or -external-review requirements. It also does not make the qualification profile -the application default. +This ADR does not add automatic model tool selection, arbitrary Python/WASI +execution, source-path access, direct user-file mutation, network access, +application-exit persistence, production signing, or external-review +requirements. It also does not make the qualification profile the application +default or claim that the normal app discovers a worker package implicitly. ## Verification @@ -108,15 +128,21 @@ runtime round trip including in-flight cancellation. `tests/test_phase2_recipe_a covers default-off/blocked exposure, typed JSON plan parsing, owner-scoped artifacts, idempotency/conflict handling, and the shared status/task surface. `tests/test_phase2_qualification_lifecycle.py` covers the explicit coordinator -factory seam. The generated OpenAPI and TypeScript contracts include the typed -request/acceptance envelope, and `frontend/src/api/client.ts` exposes the -qualification route without enabling it in the UI by default. +factory seam. `tests/test_phase2_attachment_staging.py` covers byte limits, +active/archive rejection, owner/idempotency binding, result revalidation, and +retention. `tests/test_phase2_native_recipe_attempt.py` covers fresh per-job +identity/resource scopes, principal validation, explicit process-factory +composition, and bounded cleanup. The generated OpenAPI and TypeScript +contracts include both typed envelopes, and `frontend/src/api/client.ts` +exposes the qualification routes without enabling them in the UI by default. ## Next stage -The next implementation decision is trusted attachment staging plus binding the -real signed/native broker worker into the injected attempt factory. That stage -must preserve this API/coordinator contract, keep the normal application -default-off, and prove cancellation, ownership, retention, and publication with -the real qualified attempt. No external reviewer or production key is required -to continue the open-source qualification path. +The next implementation decision is an end-to-end qualification run through the +durable coordinator using the packaged signed worker: stage an attachment, +launch/bind the native attempt, exercise transform and cancellation, and verify +retention, owner isolation, atomic publication, and complete native cleanup. +That run must preserve this API/coordinator contract and keep the normal +application default-off. No external reviewer or production key is required to +continue the open-source qualification path; production trust remains optional +release hardening. diff --git a/docs/adr/0001-phase2-recipe-provider.md b/docs/adr/0001-phase2-recipe-provider.md index 39f96ae..6e7585b 100644 --- a/docs/adr/0001-phase2-recipe-provider.md +++ b/docs/adr/0001-phase2-recipe-provider.md @@ -118,9 +118,11 @@ worker provenance, disposable [Windows sandbox qualification harness](0001-phase native broker identity, watchdog, and accounting controls through an injected health-gated builder. The durable recipe coordinator binds that worker contract to owner-scoped staged artifacts and atomic publication, and the typed API route -is available only after the lifecycle is ready. The next gate is trusted -attachment staging plus binding the real native broker worker into the attempt -factory; the provider must still launch out of process under the bounded worker -contract and must never fall back to a host process or in-process execution. +is available only after the lifecycle is ready. Trusted attachment staging and +signed/native attempt composition are now present behind that same lifecycle. +The next gate is an end-to-end durable coordinator run through the packaged +worker: the provider must launch out of process, complete a staged transform, +cancel in flight, and prove retention, publication, and native cleanup. It must +never fall back to a host process or in-process execution. External review and production signing remain optional official-release hardening. diff --git a/docs/adr/0001-phase2-worker-protocol.md b/docs/adr/0001-phase2-worker-protocol.md index 6732952..8980cb1 100644 --- a/docs/adr/0001-phase2-worker-protocol.md +++ b/docs/adr/0001-phase2-worker-protocol.md @@ -84,12 +84,19 @@ On the controlled Windows host (2026-07-23), the PyInstaller build produced `dist/recipe-runtime/recipe_worker.exe`; direct launch without the exact native broker arguments returned exit code `78` as required. -## Remaining blockers +## Remaining qualification gate -This ADR does not authorize provider execution. The remaining stage must install a -real signed generation, launch it through the reviewed suspended AppContainer/Job -Object factory, bind the live broker session to the actual worker PID and -AppContainer token, and run the hostile decoder/cancellation corpus through that -packaged process with the qualified resource/watchdog and artifact-boundary -controls. Lifecycle/UI enablement remains behind those gates and external security -review. +The durable coordinator now has an explicit native attempt factory that installs +no fallback transport: each job creates a fresh pipe/broker identity, launches a +verified signed generation through the reviewed suspended AppContainer/Job Object +factory, binds the live session to the actual worker PID and AppContainer token, +and closes the client, broker, and process tree on failure or cancellation. +Attachment staging is likewise available only through the owner-scoped artifact +boundary and qualification lifecycle. + +This ADR still does not authorize the normal application to execute providers. +The next gate must run the full hostile decoder/cancellation corpus through the +durable coordinator using that packaged process and verify retention, atomic +publication, and cleanup. Lifecycle/UI enablement remains behind those gates; +external security review and production trust remain optional official-release +hardening. diff --git a/frontend/src/api/client.test.ts b/frontend/src/api/client.test.ts index 940f684..2c8cb43 100644 --- a/frontend/src/api/client.test.ts +++ b/frontend/src/api/client.test.ts @@ -97,4 +97,34 @@ describe("CortexApi", () => { const request = fetcher.mock.calls[0]?.[1] as RequestInit; expect(new Headers(request.headers).get("Authorization")).toBe("Bearer session-1"); }); + + it("stages a bounded attachment through the qualification route", async () => { + const fetcher = vi.fn().mockResolvedValue(new Response(JSON.stringify({ + job_id: "attachment-job", + request_id: "attachment-request", + profile: "attachment.stage.v1", + status: "succeeded", + sequence: 1, + artifact_id: "artifact-1", + mime_type: "image/png", + size: 4, + sha256: "a".repeat(64), + expires_at: "2026-07-20T00:00:00Z", + }), { status: 201, headers: { "Content-Type": "application/json" } })); + const api = new CortexApi("/api/v1", fetcher); + window.sessionStorage.setItem("cortex.session.token", "session-1"); + + await api.stageAttachment({ + request_id: "attachment-request", + content_base64: "iVBORw==", + }); + + expect(fetcher).toHaveBeenCalledWith( + "/api/v1/execution/attachments", + expect.objectContaining({ + method: "POST", + body: JSON.stringify({ request_id: "attachment-request", content_base64: "iVBORw==" }), + }), + ); + }); }); diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index 055af51..269c6bc 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -1,5 +1,7 @@ import type { AddMemoryRequest, + AttachmentStageAccepted, + AttachmentStageRequest, ChatResponse, ChatSummary, CreateChatRequest, @@ -255,6 +257,13 @@ export class CortexApi { }); } + stageAttachment(payload: AttachmentStageRequest): Promise { + return this.request("/execution/attachments", { + method: "POST", + body: JSON.stringify(payload), + }); + } + cancelExecution(jobId: string): Promise { return this.request( `/execution/${encodeURIComponent(jobId)}/cancel`, diff --git a/tests/test_phase2_attachment_staging.py b/tests/test_phase2_attachment_staging.py new file mode 100644 index 0000000..2e8b33e --- /dev/null +++ b/tests/test_phase2_attachment_staging.py @@ -0,0 +1,105 @@ +"""Adversarial tests for durable trusted attachment staging.""" + +from __future__ import annotations + +import base64 +from io import BytesIO +from pathlib import Path + +import pytest +from PIL import Image + +from cortex_backend.execution import ( + ArtifactBoundary, + AttachmentStagingError, + AttachmentStagingService, + ExecutionRepository, +) + + +OWNER = "a" * 64 + + +def _image_bytes() -> bytes: + image = Image.new("RGB", (4, 3), (120, 80, 40)) + try: + with BytesIO() as stream: + image.save(stream, format="PNG") + return stream.getvalue() + finally: + image.close() + + +def _service(tmp_path: Path, *, maximum: int = 2 * 1024 * 1024): + repository = ExecutionRepository( + tmp_path / "execution.sqlite", + tmp_path / "artifacts", + max_artifact_bytes=maximum, + ) + boundary = ArtifactBoundary(repository, max_input_bytes=maximum) + return repository, AttachmentStagingService(repository, boundary) + + +def test_stage_bytes_is_owner_scoped_idempotent_and_never_persists_payload(tmp_path: Path): + repository, service = _service(tmp_path) + content = _image_bytes() + + first = service.stage(owner=OWNER, request_id="attach-1", content=content) + duplicate = service.stage(owner=OWNER, request_id="attach-1", content=content) + + assert first.artifact.artifact_id == duplicate.artifact.artifact_id + assert first.artifact.mime_type == "image/png" + assert first.job.status == "succeeded" + assert first.job.result is not None + assert "path" not in str(first.job.result).lower() + assert base64.b64encode(content).decode("ascii") not in str(first.job.payload) + assert repository.read_artifact(first.artifact.artifact_id) == content + + with pytest.raises(AttachmentStagingError) as conflict: + service.stage(owner=OWNER, request_id="attach-1", content=content + b"x") + assert conflict.value.code == "request_conflict" + + +@pytest.mark.parametrize( + ("content", "code"), + [ + (b"", "attachment_content_invalid"), + (b"PK\x03\x04not-an-image", "attachment_invalid"), + (b"", "attachment_invalid"), + ], +) +def test_stage_bytes_rejects_empty_archive_and_active_payloads( + tmp_path: Path, + content: bytes, + code: str, +): + _repository, service = _service(tmp_path) + with pytest.raises(AttachmentStagingError) as error: + service.stage(owner=OWNER, request_id="attach-invalid", content=content) + assert error.value.code == code + + +def test_stage_bytes_enforces_configured_byte_and_retention_limits(tmp_path: Path): + _repository, service = _service(tmp_path, maximum=64) + with pytest.raises(AttachmentStagingError) as too_large: + service.stage(owner=OWNER, request_id="attach-large", content=b"x" * 65) + assert too_large.value.code == "attachment_too_large" + with pytest.raises(AttachmentStagingError) as retention: + service.stage( + owner=OWNER, + request_id="attach-retention", + content=b"x", + retention_seconds=0, + ) + assert retention.value.code == "attachment_retention_invalid" + + +def test_stage_bytes_duplicate_terminal_result_is_revalidated(tmp_path: Path): + repository, service = _service(tmp_path) + staged = service.stage(owner=OWNER, request_id="attach-integrity", content=_image_bytes()) + artifact_path = Path(staged.artifact.path) + artifact_path.write_bytes(b"tampered") + + with pytest.raises(AttachmentStagingError) as error: + service.stage(owner=OWNER, request_id="attach-integrity", content=_image_bytes()) + assert error.value.code in {"attachment_artifact_unavailable", "attachment_artifact_invalid"} diff --git a/tests/test_phase2_native_recipe_attempt.py b/tests/test_phase2_native_recipe_attempt.py new file mode 100644 index 0000000..2ac948a --- /dev/null +++ b/tests/test_phase2_native_recipe_attempt.py @@ -0,0 +1,220 @@ +"""Adversarial tests for per-job signed/native recipe attempt composition.""" + +from __future__ import annotations + +import base64 +from hashlib import sha256 +import json +from pathlib import Path +from typing import Any + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from cortex_backend.execution import ( + BrokerWorkerBinding, + NativeRecipeWorkerAttemptFactory, + NativeWorkerLaunchPlan, + NativeWorkerPolicy, + SignedBundleInstaller, + TrustedRecipeKeys, +) +from cortex_backend.execution.models import ExecutionJob + + +OWNER = "a" * 64 + + +def _canonical(payload: dict[str, Any]) -> bytes: + return json.dumps(payload, ensure_ascii=True, sort_keys=True, separators=(",", ":")).encode("ascii") + + +def _installer(tmp_path: Path) -> SignedBundleInstaller: + signer = Ed25519PrivateKey.generate() + public = signer.public_key().public_bytes( + serialization.Encoding.Raw, + serialization.PublicFormat.Raw, + ) + content = b"verified recipe worker fixture" + source = tmp_path / "incoming" + source.mkdir() + (source / "recipe_worker.exe").write_bytes(content) + unsigned = { + "schema_version": "recipe.manifest.v1", + "key_id": "release-1", + "sequence": 1, + "bundle_version": "1.0.0", + "rollback_of": None, + "entries": [ + { + "recipe_id": "image-transform", + "bundle_path": "recipe_worker.exe", + "entrypoint": "image_transform", + "version": "1.0.0", + "size": len(content), + "sha256": sha256(content).hexdigest(), + } + ], + } + signed = { + **unsigned, + "signature": base64.urlsafe_b64encode(signer.sign(_canonical(unsigned))) + .decode("ascii") + .rstrip("="), + } + installer = SignedBundleInstaller( + tmp_path / "store", + TrustedRecipeKeys({"release-1": public}), + ) + installer.install(signed, source) + return installer + + +def _job(job_id: str = "job-1") -> ExecutionJob: + return ExecutionJob( + job_id=job_id, + owner=OWNER, + request_id=f"request-{job_id}", + profile="recipe.image.v1", + status="running", + sequence=1, + created_at="2026-01-01T00:00:00+00:00", + updated_at="2026-01-01T00:00:00+00:00", + ) + + +class _Connection: + def __init__(self) -> None: + self.closed = False + + def send_message(self, _message: Any) -> None: + return None + + def receive_message(self) -> Any: + raise AssertionError("transform is not part of this composition test") + + def close(self) -> None: + self.closed = True + + +class _Binder: + def __init__(self, connection: _Connection, record: dict[str, Any]) -> None: + self.connection = connection + self.record = record + self.closed = False + + def accept(self, *, owner_for_job): + assert owner_for_job(self.record["binding"].job_id) == OWNER + assert owner_for_job("foreign-job") is None + return self.connection + + def close_binding(self) -> None: + self.closed = True + + +class _Worker: + process_id = 442 + app_container_sid = "S-1-15-2-123-456" + + def __init__(self, record: dict[str, Any]) -> None: + self.record = record + self.closed = False + + def close(self) -> None: + self.closed = True + + +class _ProcessFactory: + def create_suspended(self, _plan: NativeWorkerLaunchPlan): + raise AssertionError("launcher fake owns process creation in this test") + + +class _Launcher: + def __init__(self, _installer, *, process_factory, broker_binder): + assert callable(getattr(process_factory, "create_suspended", None)) + self.binder = broker_binder + + def launch(self, binding: BrokerWorkerBinding, _policy: NativeWorkerPolicy): + self.binder.record["binding"] = binding + worker = _Worker(self.binder.record) + self.binder.record["worker"] = worker + return worker + + +def test_factory_creates_fresh_pipe_binder_and_worker_scope_per_job(tmp_path: Path): + installer = _installer(tmp_path) + records: list[dict[str, Any]] = [] + + def binder_factory(**_kwargs): + record: dict[str, Any] = {"connection": _Connection()} + binder = _Binder(record["connection"], record) + record["binder"] = binder + records.append(record) + return binder + + factory = NativeRecipeWorkerAttemptFactory( + installer, + allowed_user_sids=frozenset({"S-1-5-21-1-2-3-4"}), + process_factory_factory=_ProcessFactory, + binder_factory=binder_factory, + launcher_factory=_Launcher, + ) + + first = factory(_job("job-1")) + second = factory(_job("job-2")) + assert len(records) == 2 + assert records[0]["binding"].pipe_name != records[1]["binding"].pipe_name + assert records[0]["binding"].installation_principal_id == OWNER + assert records[0]["binding"].broker_process_id > 0 + assert records[0]["binding"].job_id == "job-1" + assert records[1]["binding"].job_id == "job-2" + + first.close() + second.close() + for record in records: + assert record["connection"].closed + assert record["binder"].closed + assert record["worker"].closed + + +def test_factory_rejects_non_principal_job_before_native_adapters(tmp_path: Path): + installer = _installer(tmp_path) + called = [] + factory = NativeRecipeWorkerAttemptFactory( + installer, + allowed_user_sids=frozenset({"S-1-5-21-1-2-3-4"}), + process_factory_factory=lambda: called.append(True), + ) + invalid = _job() + invalid = ExecutionJob( + job_id=invalid.job_id, + owner="not-a-principal", + request_id=invalid.request_id, + profile=invalid.profile, + status=invalid.status, + sequence=invalid.sequence, + created_at=invalid.created_at, + updated_at=invalid.updated_at, + ) + with pytest.raises(Exception) as error: + factory(invalid) + assert getattr(error.value, "code", None) == "worker_principal_invalid" + assert called == [] + + +def test_factory_requires_explicit_process_factory_and_bounded_options(tmp_path: Path): + installer = _installer(tmp_path) + with pytest.raises(TypeError): + NativeRecipeWorkerAttemptFactory( + installer, + allowed_user_sids=frozenset({"S-1-5-21-1-2-3-4"}), + process_factory_factory=None, # type: ignore[arg-type] + ) + with pytest.raises(ValueError): + NativeRecipeWorkerAttemptFactory( + installer, + allowed_user_sids=frozenset({"S-1-5-21-1-2-3-4"}), + process_factory_factory=_ProcessFactory, + accept_timeout_seconds=120.1, + ) diff --git a/tests/test_phase2_qualification_lifecycle.py b/tests/test_phase2_qualification_lifecycle.py index 4c8966c..40a2be8 100644 --- a/tests/test_phase2_qualification_lifecycle.py +++ b/tests/test_phase2_qualification_lifecycle.py @@ -10,6 +10,7 @@ QualificationLifecycleConfig, QualificationProfileError, build_execution_lifecycle, + build_native_recipe_coordinator_factory, build_recipe_coordinator_factory, parse_execution_profile, ) @@ -161,3 +162,22 @@ def worker_attempt_factory(job): assert coordinator.artifact_boundary.repository is repository coordinator.shutdown() assert worker_calls == [] + + +def test_native_recipe_factory_is_explicit_and_does_not_launch_during_composition(tmp_path): + repository = ExecutionRepository(tmp_path / "execution.sqlite", tmp_path / "artifacts") + installer = SignedBundleInstaller( + tmp_path / "native-store", + TrustedRecipeKeys({"qualification": b"q" * 32}), + ) + process_calls: list[bool] = [] + factory = build_native_recipe_coordinator_factory( + installer, + allowed_user_sids=frozenset({"S-1-5-21-1-2-3-4"}), + process_factory_factory=lambda: process_calls.append(True), + ) + + coordinator = factory(repository) + assert coordinator.repository is repository + assert process_calls == [] + coordinator.shutdown() diff --git a/tests/test_phase2_recipe_api.py b/tests/test_phase2_recipe_api.py index b874bb7..18e2e70 100644 --- a/tests/test_phase2_recipe_api.py +++ b/tests/test_phase2_recipe_api.py @@ -3,6 +3,7 @@ from __future__ import annotations from io import BytesIO +import base64 import hashlib from threading import Event import time @@ -134,6 +135,15 @@ def test_recipe_route_requires_ready_qualification_and_preserves_default_off(tmp ) assert response.status_code == 404 assert "path" not in response.text.lower() + attachment = client.post( + "/api/v1/execution/attachments", + headers=headers, + json={ + "request_id": "api-attachment", + "content_base64": base64.b64encode(_image_bytes()).decode("ascii"), + }, + ) + assert attachment.status_code == 404 def test_recipe_route_is_owner_scoped_idempotent_and_reaches_existing_execution_surface(tmp_path): @@ -190,6 +200,48 @@ def test_recipe_route_is_owner_scoped_idempotent_and_reaches_existing_execution_ assert tasks.json()["tasks"][0]["profile"] == "recipe.image.v1" +def test_attachment_stage_is_bounded_idempotent_and_returns_opaque_artifact(tmp_path): + app, _artifact_id = _app(tmp_path) + encoded = base64.b64encode(_image_bytes()).decode("ascii") + with TestClient(app) as client: + headers = _session(client, app) + accepted = client.post( + "/api/v1/execution/attachments", + headers=headers, + json={"request_id": "api-attachment", "content_base64": encoded}, + ) + assert accepted.status_code == 201 + body = accepted.json() + assert body["profile"] == "attachment.stage.v1" + assert body["status"] == "succeeded" + assert body["mime_type"] == "image/png" + assert body["artifact_id"] + assert "path" not in accepted.text.lower() + assert encoded not in str(app.state.execution_lifecycle.repository.get_job(body["job_id"]).payload) + + duplicate = client.post( + "/api/v1/execution/attachments", + headers=headers, + json={"request_id": "api-attachment", "content_base64": encoded}, + ) + assert duplicate.status_code == 201 + assert duplicate.json()["artifact_id"] == body["artifact_id"] + + conflict = client.post( + "/api/v1/execution/attachments", + headers=headers, + json={"request_id": "api-attachment", "content_base64": base64.b64encode(b"other").decode("ascii")}, + ) + assert conflict.status_code == 409 + + malformed = client.post( + "/api/v1/execution/attachments", + headers=headers, + json={"request_id": "api-malformed", "content_base64": "not-base64!"}, + ) + assert malformed.status_code == 422 + + def test_recipe_route_rejects_mismatched_plan_and_foreign_artifact_without_leaks(tmp_path): app, artifact_id = _app(tmp_path) repository = app.state.execution_lifecycle.repository