verify exact BYOC bundle archives

Author: abrichr

docs/BYOC_CONNECTOR.md

@@ -69,13 +69,21 @@ env var (`CONTROL_PLANE_URL`, `BYOC_CONNECTOR_TOKEN`, `BYOC_ORG_ID`, ...).
 * **The bundle and report bytes** are read from / written to the **customer's own
   storage** (`--storage-root`, a local encrypted volume). Our control plane holds
   no URL to them and signs no access.
+* **The staged bundle is the exact approved ZIP archive.** `bundle_sha256` binds
+  the dispatch to those archive bytes (separately from the bundle's semantic
+  content digest). The Connector copies and hashes that file before extraction,
+  refuses a missing/different digest or an unpacked directory, then safe-extracts
+  only the verified copy. A pre-verification failure never echoes the unverified
+  digest claim in its callback.
 
 **Fail-closed everywhere.** A dispatch is refused (reported `failed`, no GUI
 touched) when it is missing the governed safety policy, missing the run-scoped
-callback token, carries an our-owned signed URL, or when the org enabled a
-grounding rung whose API key env is not set on this machine. The governed `run`
-itself refuses any bundle that is not certified, identity-armed, effect-verified,
-and encrypted — those engine gates are unchanged, so identity checks, effect
+callback token, exact archive digest, or execution substrate; carries an
+our-owned signed URL; points at archive bytes that do not match the signed
+digest; conflicts with the substrate compatibility hint; or enables a grounding
+rung whose API key env is not set on this machine. The governed `run` itself
+refuses any bundle that is not certified, identity-armed, effect-verified, and
+encrypted — those engine gates are unchanged, so identity checks, effect
 verification, and halt-don't-guess all remain intact.
 
 ## Enabling the lane (control plane)

openadapt_flow/connector/daemon.py

@@ -40,7 +40,6 @@ def phi_free_callback_body(job: ByocJob, result: ExecutionResult) -> dict[str, A
         "workflow_id": job.workflow_id,
         "bundle_version_id": job.bundle_version_id,
         "runtime_validation_id": job.runtime_validation_id,
-        "bundle_sha256": job.bundle_sha256,
         "mode": job.mode,
         "status": result.status,
         "report_path": result.report_ref,
@@ -49,6 +48,10 @@ def phi_free_callback_body(job: ByocJob, result: ExecutionResult) -> dict[str, A
         # name internal paths). error_code is the exact enum run-callback accepts.
         "error_code": "runner_failure" if result.status == "failed" else None,
     }
+    if result.verified_bundle_sha256 is not None:
+        # Never echo the job's untrusted digest claim. This field exists only
+        # after the connector copied, hashed, and matched the exact archive.
+        body["bundle_sha256"] = result.verified_bundle_sha256
     if result.halt is not None:
         # run-callback reads only halt.present; the structured block is additive.
         body["halt"] = {"present": True}

openadapt_flow/connector/executor.py

@@ -9,8 +9,10 @@
 
 Fail-closed application of the control-plane-delivered policy:
 
-* :meth:`ByocJob.ensure_governed` must pass (policy present, callback token
-  present, bundle resolvable) or the job is refused before any GUI is touched;
+* :meth:`ByocJob.ensure_governed` must pass (policy, callback token, closed
+  substrate, and exact archive SHA present) or the job is refused before any
+  GUI is touched; customer storage must then return the exact approved ZIP
+  bytes, whose digest is checked before safe extraction;
 * if the org enabled a grounding rung (``grounding_model.enabled``) whose
   ``api_key_env`` is NOT set in the Connector's own environment, the job is
   refused — the org required a governed control this machine cannot honor, so we
@@ -31,6 +33,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import json
 import os
 import subprocess
@@ -42,7 +45,7 @@
 
 from openadapt_flow.connector.config import ConnectorSettings
 from openadapt_flow.connector.protocol import ByocGovernanceError, ByocJob
-from openadapt_flow.connector.storage import CustomerStorage
+from openadapt_flow.connector.storage import CustomerStorage, extract_bundle_archive
 
 #: A run-gate refusal (fail-closed admission denied) exits 2 before the replay
 #: creates report.json.
@@ -69,6 +72,7 @@ class ExecutionResult:
     halt: Optional[dict[str, Any]]
     report_ref: Optional[str]
     error: Optional[str] = None
+    verified_bundle_sha256: Optional[str] = None
 
 
 def _grounding_env_available(job: ByocJob) -> bool:
@@ -82,6 +86,14 @@ def _grounding_env_available(job: ByocJob) -> bool:
     return bool(os.environ.get(gm.api_key_env))
 
 
+def _sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
 def build_run_argv(
     job: ByocJob,
     settings: ConnectorSettings,
@@ -248,18 +260,27 @@ def execute_job(
 
     with tempfile.TemporaryDirectory(prefix="oa-byoc-") as tmp:
         tmp_path = Path(tmp)
+        bundle_archive = tmp_path / "approved-bundle.zip"
         bundle_scratch = tmp_path / "bundle"
         run_dir = tmp_path / "run"
         bundle_scratch.mkdir(parents=True, exist_ok=True)
         run_dir.mkdir(parents=True, exist_ok=True)
 
         _write_policy_audit(job, run_dir)
 
-        # 2. Resolve the bundle from the CUSTOMER'S OWN storage (never our URL).
+        # 2. Copy and verify the EXACT approved archive bytes from the customer's
+        # storage before extraction. The content digest inside workflow.json is
+        # a separate semantic binding; neither can substitute for this signed
+        # archive-byte identity.
         try:
-            bundle_dir = storage.fetch_bundle(
-                job.storage.bundle_ref if job.storage else None, bundle_scratch
+            fetched_archive = storage.fetch_bundle_archive(
+                job.storage.bundle_ref if job.storage else None, bundle_archive
             )
+            observed_sha256 = _sha256_file(fetched_archive)
+            if observed_sha256 != job.bundle_sha256:
+                raise RuntimeError("bundle archive SHA-256 does not match dispatch")
+            extract_bundle_archive(fetched_archive, bundle_scratch)
+            bundle_dir = bundle_scratch
         except Exception as exc:  # storage failure — fail closed, PHI-free msg
             return ExecutionResult(
                 "failed",
@@ -269,11 +290,21 @@ def execute_job(
                 f"customer-storage bundle fetch failed: {type(exc).__name__}",
             )
 
-        params_file = _write_params_file(job.params, run_dir)
+        try:
+            params_file = _write_params_file(job.params, run_dir)
 
-        # 3. The governed, fail-closed child invocation.
-        argv = build_run_argv(job, settings, Path(bundle_dir), run_dir, params_file)
-        outcome = runner(argv, run_dir)
+            # 3. The governed, fail-closed child invocation.
+            argv = build_run_argv(job, settings, Path(bundle_dir), run_dir, params_file)
+            outcome = runner(argv, run_dir)
+        except Exception as exc:
+            return ExecutionResult(
+                "failed",
+                {},
+                None,
+                job.report_ref(),
+                f"governed child invocation failed: {type(exc).__name__}",
+                verified_bundle_sha256=observed_sha256,
+            )
 
         report = outcome.report or {}
         status = status_from_report(outcome.returncode, report)
@@ -291,13 +322,14 @@ def execute_job(
         report_ref = job.report_ref()
         try:
             storage.write_report(report_ref, report)
-        except Exception as exc:  # pragma: no cover - report persist best-effort
+        except Exception as exc:  # no durable report means the run is not complete
             return ExecutionResult(
-                status,
+                "failed",
                 metrics_from_report(report),
                 halt_object(report),
                 report_ref,
                 f"customer-storage report write failed: {type(exc).__name__}",
+                verified_bundle_sha256=observed_sha256,
             )
 
         return ExecutionResult(
@@ -306,4 +338,5 @@ def execute_job(
             halt=halt_object(report),
             report_ref=report_ref,
             error=outcome_error,
+            verified_bundle_sha256=observed_sha256,
         )

openadapt_flow/connector/protocol.py

@@ -105,7 +105,10 @@ class ByocJob(BaseModel):
     run_token: Optional[str] = None
     bundle_version_id: Optional[str] = None
     runtime_validation_id: Optional[str] = None
-    bundle_sha256: Optional[str] = None
+    #: SHA-256 of the exact approved sanitized derivative ZIP bytes staged at
+    #: ``storage.bundle_ref``. Distinct from the bundle's semantic content
+    #: digest; required so a valid but different archive cannot be executed.
+    bundle_sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
 
     # --- Governed policy delivery (fail-closed) --------------------------------
     #: The org's resolved (baseline-filled) Tier-3 safety block. Always fully
@@ -159,6 +162,11 @@ def ensure_governed(self, *, require_run_token: bool = True) -> None:
                 "byoc dispatch is missing a run-scoped callback token; refusing "
                 "to run a job whose outcome we could not report (fail closed)"
             )
+        if not self.bundle_version_id or not self.runtime_validation_id:
+            raise ByocGovernanceError(
+                "byoc dispatch is missing its immutable bundle-version or "
+                "runtime-validation binding (fail closed)"
+            )
         storage_ref = self.storage.bundle_ref if self.storage else None
         if not storage_ref:
             raise ByocGovernanceError(

openadapt_flow/connector/storage.py

@@ -11,9 +11,9 @@
 
 * :class:`LocalCustomerStorage` — REAL: refs resolve under a customer directory
   (ideally a full-disk-encrypted volume). The on-prem clinic posture.
-* :class:`InMemoryCustomerStorage` — tests/dry-run: pretends the bundle is
-  present and CAPTURES the report in memory, proving the report bytes never
-  leave for the control plane, with zero infra.
+* :class:`InMemoryCustomerStorage` — tests/dry-run: stages exact archive bytes
+  and CAPTURES the report in memory, proving the report bytes never leave for
+  the control plane, with zero infra.
 
 S3 / Azure Blob backends are documented as production follow-ups (they need a
 customer cloud to exercise); the operator reference agent in openadapt-cloud
@@ -24,6 +24,7 @@
 
 import json
 import os
+import shutil
 import zipfile
 from pathlib import Path
 from typing import Any, Optional, Protocol
@@ -34,16 +35,16 @@ class CustomerStorage(Protocol):
 
     kind: str
 
-    def fetch_bundle(self, ref: Optional[str], dest_dir: Path) -> Path:
-        """Resolve ``ref`` to an unpacked bundle directory under ``dest_dir``."""
+    def fetch_bundle_archive(self, ref: Optional[str], dest_file: Path) -> Path:
+        """Copy the exact referenced ZIP bytes to ``dest_file``."""
         ...
 
     def write_report(self, ref: Optional[str], report: dict[str, Any]) -> Optional[str]:
         """Persist the PHI-bearing report to the customer store; return its key."""
         ...
 
 
-def _safe_extract(zip_path: Path, dest: Path) -> None:
+def extract_bundle_archive(zip_path: Path, dest: Path) -> None:
     """Extract a zip, refusing any entry that escapes ``dest`` (zip-slip)."""
     dest = dest.resolve()
     with zipfile.ZipFile(zip_path) as zf:
@@ -62,16 +63,25 @@ class LocalCustomerStorage:
     def __init__(self, root: str) -> None:
         self.root = Path(root)
 
-    def fetch_bundle(self, ref: Optional[str], dest_dir: Path) -> Path:
+    def fetch_bundle_archive(self, ref: Optional[str], dest_file: Path) -> Path:
         if not ref:
             raise RuntimeError("byoc job has no storage.bundle_ref to read")
-        src = self.root / ref
+        root = self.root.resolve()
+        src = (root / ref).resolve()
+        if src != root and root not in src.parents:
+            raise RuntimeError(
+                "bundle ref escapes the configured customer storage root"
+            )
         if not src.exists():
             raise RuntimeError(f"bundle not found in customer storage: {src}")
-        if src.is_dir():
-            return src  # already an unpacked bundle directory
-        _safe_extract(src, dest_dir)
-        return dest_dir
+        if not src.is_file():
+            raise RuntimeError(
+                "byoc bundle_ref must resolve to the exact approved ZIP archive, "
+                "not an unpacked directory"
+            )
+        dest_file.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copyfile(src, dest_file)
+        return dest_file
 
     def write_report(self, ref: Optional[str], report: dict[str, Any]) -> Optional[str]:
         if not ref:
@@ -87,20 +97,24 @@ def write_report(self, ref: Optional[str], report: dict[str, Any]) -> Optional[s
 
 
 class InMemoryCustomerStorage:
-    """Tests/dry-run backend: a pre-staged bundle dir + an in-memory report sink.
+    """Tests/dry-run backend: exact bundle bytes + an in-memory report sink.
 
     Proves the report bytes are written to the CUSTOMER side and never returned
     to the control plane, with zero infra.
     """
 
     kind = "memory"
 
-    def __init__(self, bundle_dir: Optional[Path] = None) -> None:
-        self.bundle_dir = bundle_dir
+    def __init__(self, bundle_bytes: Optional[bytes] = None) -> None:
+        self.bundle_bytes = bundle_bytes
         self.written: dict[str, dict[str, Any]] = {}
 
-    def fetch_bundle(self, ref: Optional[str], dest_dir: Path) -> Path:
-        return self.bundle_dir if self.bundle_dir is not None else dest_dir
+    def fetch_bundle_archive(self, ref: Optional[str], dest_file: Path) -> Path:
+        if self.bundle_bytes is None:
+            raise RuntimeError("in-memory customer storage has no bundle archive")
+        dest_file.parent.mkdir(parents=True, exist_ok=True)
+        dest_file.write_bytes(self.bundle_bytes)
+        return dest_file
 
     def write_report(self, ref: Optional[str], report: dict[str, Any]) -> Optional[str]:
         self.written[ref or "?"] = report