feat(sync): deploy gate must refuse — bind switch to a signed release attestation (SP-GATE-001)

src/sourceos_syncd/cli.py

@@ -173,15 +173,27 @@ def add_katello_args(p: argparse.ArgumentParser) -> None:
         p.add_argument("--no-verify-ssl", action="store_true", help="skip TLS verification (local dev only)")
         p.add_argument("--signing-public-key", default=None, help="minisign public key (RWS...) to verify nix-cache-info before applying")
 
+    def add_attestation_args(p: argparse.ArgumentParser) -> None:
+        # The deploy gate defaults to fail-closed: without a verified release
+        # attestation binding the target version, plan() refuses to switch.
+        p.add_argument("--attestation-file", default=None,
+                       help="path to a signed release attestation (SignedArtifact JSON) for the target version")
+        p.add_argument("--attestation-public-key", default=None,
+                       help="minisign public key the release attestation is verified against")
+        p.add_argument("--no-require-attestation", dest="require_attestation", action="store_false", default=True,
+                       help="EXPLICIT opt-out of the release-attestation gate (recorded in the receipt; unsafe)")
+
     sync_plan = sync_sub.add_parser("plan", help="query Katello and emit a ContentSyncPlan (no changes)")
     add_katello_args(sync_plan)
+    add_attestation_args(sync_plan)
     add_compact(sync_plan)
 
     sync_apply = sync_sub.add_parser("apply", help="apply a ContentSyncPlan (dry-run unless --execute)")
     add_katello_args(sync_apply)
     sync_apply.add_argument("--execute", action="store_true", help="actually run nix copy + nixos-rebuild (default: dry-run)")
     sync_apply.add_argument("--store-root", default=None, help="persist receipt to this store root")
     sync_apply.add_argument("--agentplane-run-ref", default=None, help="agentplane RunArtifact URN that triggered this sync cycle (optional)")
+    add_attestation_args(sync_apply)
     add_compact(sync_apply)
 
     sync_daemon = sync_sub.add_parser("daemon", help="run the sync daemon (polls Katello; applies on new version)")
@@ -401,14 +413,25 @@ def main(argv: list[str] | None = None) -> int:
                 verify_ssl=not args.no_verify_ssl,
             )
             manifest = client.get_latest_version(args.content_view, args.lifecycle_env)
+            attestation = None
+            attestation_file = getattr(args, "attestation_file", None)
+            if attestation_file:
+                try:
+                    with open(attestation_file, encoding="utf-8") as fh:
+                        attestation = json.load(fh)
+                except (OSError, ValueError) as exc:
+                    sys.stderr.write(f"error: could not read --attestation-file {attestation_file!r}: {exc}\n")
+                    return 1
             syncer = ContentViewSyncer(
                 flake_ref=args.flake_ref,
                 locus=args.locus,
                 current_version=args.current_version,
                 signing_public_key=getattr(args, "signing_public_key", None),
                 agentplane_run_ref=getattr(args, "agentplane_run_ref", None),
+                require_attestation=getattr(args, "require_attestation", True),
+                attestation_public_key=getattr(args, "attestation_public_key", None),
             )
-            plan = syncer.plan(manifest)
+            plan = syncer.plan(manifest, attestation=attestation)
             if args.command == "plan":
                 sys.stdout.write(pretty_json(plan.to_dict(), pretty=pretty))
                 return 0 if plan.policy_gate in ("allowed", "no-op") else 2

src/sourceos_syncd/content_sync.py

@@ -7,6 +7,14 @@
 Boundary invariant: plan() is pure and side-effect-free. execute() is the
 only method that shells out — it requires an explicit caller opt-in and will
 refuse to run if the plan's policy_gate is not 'allowed'.
+
+Deploy gate (SP-GATE-001): when require_attestation is True (the default),
+plan() will only emit a `nixos-rebuild switch` when a signed release attestation
+binds the *specific* content-view version being switched to. Absent, mismatched,
+or unverifiable attestation → policy_gate="blocked" and NO switch step. This is
+what makes possession of the promote credential stop being runtime authority.
+execute() additionally aborts on the first failed step, so a verification that
+returns non-zero prevents the switch instead of being reported after the fact.
 """
 
 from __future__ import annotations
@@ -20,6 +28,7 @@
 from typing import Any
 
 from .katello_client import ContentViewManifest
+from .release_attestation import AttestationDecision, Verifier, verify_release_attestation
 
 SYNC_SCHEMA = "sourceos.content-sync-plan/v0.1"
 RECEIPT_SPEC_VERSION = "0.1.0"
@@ -41,6 +50,12 @@ class ContentSyncPlan:
     policy_gate: str
     policy_reason: str
     steps: list[str] = field(default_factory=list)
+    # Epistemic level of this plan's authority to switch: "Proved" only when a
+    # release attestation verified and bound this version; "Speculative" otherwise.
+    epistemic_level: str = "Speculative"
+    # The attestation decision (as_dict) when the gate ran; None when the gate was
+    # not required (an explicit, recorded opt-out — not a silent skip).
+    attestation: dict[str, Any] | None = None
 
     def to_dict(self) -> dict[str, Any]:
         return {
@@ -55,6 +70,8 @@ def to_dict(self) -> dict[str, Any]:
             "policy_gate": self.policy_gate,
             "policy_reason": self.policy_reason,
             "steps": self.steps,
+            "epistemic_level": self.epistemic_level,
+            "attestation": self.attestation,
         }
 
     @property
@@ -74,6 +91,13 @@ class ContentViewSyncer:
     This ensures the nix-cache-info served by Pulp was signed by the key
     embedded in the NixOS image, preventing an unauthenticated Katello from
     delivering arbitrary closures.
+
+    When require_attestation is True (default), the plan additionally REFUSES to
+    switch unless a signed release attestation binds the exact version being
+    applied — see release_attestation.verify_release_attestation. attestation_public_key
+    is the minisign key that attestation is checked against; attestation_verifier
+    is injectable for testing (default performs real minisign verification and
+    fails closed).
     """
 
     ALLOWED_LOCI = {"local", "trusted_private"}
@@ -85,14 +109,24 @@ def __init__(
         current_version: str | None = None,
         signing_public_key: str | None = None,
         agentplane_run_ref: str | None = None,
+        require_attestation: bool = True,
+        attestation_public_key: str | None = None,
+        attestation_verifier: Verifier | None = None,
     ) -> None:
         self._flake_ref = flake_ref
         self._locus = locus
         self._current_version = current_version
         self._signing_public_key = signing_public_key
         self._agentplane_run_ref = agentplane_run_ref
+        self._require_attestation = require_attestation
+        self._attestation_public_key = attestation_public_key
+        self._attestation_verifier = attestation_verifier
 
-    def plan(self, manifest: ContentViewManifest) -> ContentSyncPlan:
+    def plan(
+        self,
+        manifest: ContentViewManifest,
+        attestation: dict[str, Any] | None = None,
+    ) -> ContentSyncPlan:
         """Return a non-mutating ContentSyncPlan. No I/O performed."""
 
         if self._locus not in self.ALLOWED_LOCI:
@@ -125,6 +159,37 @@ def plan(self, manifest: ContentViewManifest) -> ContentSyncPlan:
                 steps=[],
             )
 
+        # ── deploy gate: bind the switch to a signed attestation of THIS version ──
+        # A newer version being *available* is not authority to switch to it. The
+        # switch is authorized only by a release attestation that verifies and
+        # binds this exact (org, content_view, version). Fail closed.
+        decision: AttestationDecision | None = None
+        if self._require_attestation:
+            decision = verify_release_attestation(
+                attestation,
+                org=manifest.org,
+                content_view=manifest.content_view,
+                version=manifest.version,
+                trusted_key=self._attestation_public_key,
+                verifier=self._attestation_verifier,
+            )
+            if not decision.ok:
+                return ContentSyncPlan(
+                    schema=SYNC_SCHEMA,
+                    org=manifest.org,
+                    content_view=manifest.content_view,
+                    from_version=self._current_version,
+                    to_version=manifest.version,
+                    lifecycle_env=manifest.lifecycle_env,
+                    nix_cache_url=manifest.nix_cache_url,
+                    flake_ref=self._flake_ref,
+                    policy_gate="blocked",
+                    policy_reason=f"release attestation refused: {decision.reason}",
+                    steps=[],
+                    epistemic_level=decision.epistemic_level,
+                    attestation=decision.to_dict(),
+                )
+
         steps = []
 
         # Verify nix-cache-info signature before pulling any closures.
@@ -158,18 +223,26 @@ def plan(self, manifest: ContentViewManifest) -> ContentSyncPlan:
             policy_gate="allowed",
             policy_reason=f"locus '{self._locus}' permitted; new version available",
             steps=steps,
+            epistemic_level=(decision.epistemic_level if decision else "Speculative"),
+            attestation=(decision.to_dict() if decision else None),
         )
 
     def execute(self, plan: ContentSyncPlan, dry_run: bool = True) -> dict[str, Any]:
         """Execute the sync plan. dry_run=True (default) only prints steps.
 
+        Steps run in order and the loop ABORTS on the first failed or timed-out
+        step: a verification (or nix copy) that fails must PREVENT the subsequent
+        `nixos-rebuild switch`, not merely be reported after it already ran.
+
         Always emits a SyncCycleReceipt in the return dict under 'receipt'.
         """
         cycle_id = str(uuid.uuid4())
         t_start = time.monotonic()
 
         if not plan.allowed:
-            outcome = "denied" if plan.policy_gate == "denied" else "skipped"
+            outcome = "denied" if plan.policy_gate == "denied" else (
+                "blocked" if plan.policy_gate == "blocked" else "skipped"
+            )
             receipt = self._build_receipt(
                 cycle_id=cycle_id,
                 plan=plan,
@@ -185,7 +258,18 @@ def execute(self, plan: ContentSyncPlan, dry_run: bool = True) -> dict[str, Any]
             }
 
         results = []
+        aborted = False
         for step in plan.steps:
+            if aborted:
+                # Fail closed: once a step fails we refuse to run anything after
+                # it — above all the nixos-rebuild switch.
+                results.append({
+                    "step": step,
+                    "status": "not-run",
+                    "reason": "aborted: a prior step failed — refusing to proceed to nixos-rebuild",
+                })
+                continue
+
             if dry_run:
                 results.append({"step": step, "status": "dry_run", "reason": "dry_run=True"})
                 continue
@@ -201,19 +285,23 @@ def execute(self, plan: ContentSyncPlan, dry_run: bool = True) -> dict[str, Any]
                 proc = subprocess.run(
                     step, shell=True, capture_output=True, text=True, timeout=600
                 )
+                status = "ok" if proc.returncode == 0 else "failed"
                 results.append({
                     "step": step,
-                    "status": "ok" if proc.returncode == 0 else "failed",
+                    "status": status,
                     "returncode": proc.returncode,
                     "stdout": proc.stdout.strip()[:500],
                     "stderr": proc.stderr.strip()[:500],
                 })
+                if status == "failed":
+                    aborted = True
             except subprocess.TimeoutExpired:
                 results.append({"step": step, "status": "timeout"})
+                aborted = True
 
         duration_ms = int((time.monotonic() - t_start) * 1000)
         outcome = "dry_run" if dry_run else (
-            "applied" if all(r.get("status") in ("ok", "dry_run", "skipped") for r in results)
+            "applied" if (not aborted and all(r.get("status") in ("ok", "dry_run", "skipped") for r in results))
             else "failed"
         )
         receipt = self._build_receipt(
@@ -256,6 +344,10 @@ def _build_receipt(
             "outcome": outcome,
             "policyGate": plan.policy_gate,
             "policyReason": plan.policy_reason,
+            # The switch's epistemic level travels with the receipt: unattested
+            # switches are visibly Speculative downstream, never laundered to Proved.
+            "epistemicLevel": plan.epistemic_level,
+            "attestation": plan.attestation,
             "steps": steps,
             "nixCacheUrl": plan.nix_cache_url,
             "flakeRef": plan.flake_ref,

src/sourceos_syncd/daemon.py

@@ -53,6 +53,9 @@ def __init__(
         verify_ssl: bool = True,
         signing_public_key: str | None = None,
         agentplane_run_ref: str | None = None,
+        require_attestation: bool = True,
+        attestation_public_key: str | None = None,
+        attestation_dir: str | None = None,
     ) -> None:
         self._client = KatelloContentClient(
             base_url=katello_url,
@@ -67,6 +70,10 @@ def __init__(
         self._flake_ref = flake_ref
         self._signing_public_key = signing_public_key
         self._agentplane_run_ref = agentplane_run_ref
+        self._require_attestation = require_attestation
+        self._attestation_public_key = attestation_public_key
+        self._attestation_dir = attestation_dir
+        self._attestation_verifier = None  # injectable in tests; default = real minisign
         self._poll_interval_s = poll_interval_s
         self._store = ReceiptStore(root=store_root or "/var/lib/sourceos-syncd")
         self._running = True
@@ -111,6 +118,23 @@ def run(self) -> int:
         log.info("daemon stopped")
         return 0
 
+    def _load_attestation(self, manifest: ContentViewManifest) -> dict[str, Any] | None:
+        """Load the release attestation for this version, if one is available.
+
+        Reads ``{attestation_dir}/{content_view}-{version}.json``. Returns None when
+        no attestation_dir is configured or no file exists — and None means the
+        gate refuses (fail closed), which is the intended posture until the paired
+        producer WO publishes attestations.
+        """
+        if not self._attestation_dir:
+            return None
+        path = os.path.join(self._attestation_dir, f"{manifest.content_view}-{manifest.version}.json")
+        try:
+            with open(path, encoding="utf-8") as fh:
+                return json.load(fh)
+        except (OSError, ValueError):
+            return None
+
     def _poll_once(self) -> None:
         current_version = self._store.read_current_version()
         manifest = self._client.get_latest_version(self._content_view, self._lifecycle_env)
@@ -121,8 +145,11 @@ def _poll_once(self) -> None:
             current_version=current_version,
             signing_public_key=self._signing_public_key,
             agentplane_run_ref=self._agentplane_run_ref,
+            require_attestation=self._require_attestation,
+            attestation_public_key=self._attestation_public_key,
+            attestation_verifier=self._attestation_verifier,
         )
-        plan = syncer.plan(manifest)
+        plan = syncer.plan(manifest, attestation=self._load_attestation(manifest))
 
         if plan.policy_gate == "no-op":
             log.debug("already at %s — no sync needed", manifest.version)