Add trusted attachment staging and signed native worker binding

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
 

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"]
 

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

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 (