feat(constation): wire production miner-key dial path end to end

alembic/versions/0019_allowlist_sealed_manifest_hashes.py

@@ -0,0 +1,46 @@
+"""Add sealed_manifest_hashes to image_digest_allowlist.
+
+Stores the build-time sealed surface (path → 64-hex SHA-256) alongside each
+BASE-produced digest so production can supply prism's non-empty
+``expected_sealed_manifest_hashes``. Application registration rejects empty
+maps; the column uses a non-null JSON server default of ``{}`` so existing
+rows migrate without rewrite. Legacy empty maps cannot be re-registered and
+fail closed if loaded into DigestRecord until backfilled.
+
+Revision ID: 0019_allowlist_sealed_hashes
+Revises: 0018_attestation_nonces
+Create Date: 2026-07-27 00:00:00.000000
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "0019_allowlist_sealed_hashes"
+down_revision: str | None = "0018_attestation_nonces"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Apply the migration."""
+
+    op.add_column(
+        "image_digest_allowlist",
+        sa.Column(
+            "sealed_manifest_hashes",
+            sa.JSON(),
+            server_default=sa.text("'{}'"),
+            nullable=False,
+        ),
+    )
+
+
+def downgrade() -> None:
+    """Revert the migration."""
+
+    op.drop_column("image_digest_allowlist", "sealed_manifest_hashes")

packages/challenges/prism/src/prism_challenge/attestation_routes.py

@@ -61,6 +61,7 @@ class RegisterDigestBody(BaseModel):
     tree_sha: str
     variant: str
     digest: str
+    sealed_manifest_hashes: dict[str, str]
 
 
 class CheckAllowlistBody(BaseModel):
@@ -322,12 +323,19 @@ def build_attestation_internal_router() -> APIRouter:
     async def register_digest(request: Request, body: RegisterDigestBody) -> dict[str, str]:
         ensure_default_constation_services(request.app)
         allowlist: DigestAllowlist = request.app.state.digest_allowlist
-        record = DigestRecord(
-            commit_sha=body.commit_sha,
-            tree_sha=body.tree_sha,
-            variant=ImageVariant(body.variant.strip().lower()),
-            digest=body.digest,
-        )
+        try:
+            record = DigestRecord(
+                commit_sha=body.commit_sha,
+                tree_sha=body.tree_sha,
+                variant=ImageVariant(body.variant.strip().lower()),
+                digest=body.digest,
+                sealed_manifest_hashes=body.sealed_manifest_hashes,
+            )
+        except ValueError as exc:
+            raise HTTPException(
+                status.HTTP_422_UNPROCESSABLE_ENTITY,
+                detail=str(exc),
+            ) from exc
         allowlist.register(record)
         return {"status": "registered", "digest": record.digest}
 

scripts/e2e_lium_attestation.py

@@ -208,7 +208,15 @@ async def get_pod_raw(self, pod_id: str) -> LiumPodRead:
             pod_id=pod_id,
             template_id="tmpl-e2e-1",
             docker_image_digest=digest,
-            raw={"id": pod_id, "template": {"docker_image_digest": digest}},
+            raw={
+                "id": pod_id,
+                "status": "RUNNING",
+                "executor": {
+                    "miner_hotkey": HOTKEY,
+                    "executor_ip_address": "10.0.0.1",
+                },
+                "template": {"docker_image_digest": digest},
+            },
         )
 
     async def balance(self) -> float:
@@ -374,6 +382,7 @@ async def _probe(client: Any) -> None:
             work_unit_id="wu-placeholder",  # overwritten after submission seed
             pod_id=POD_ID,
             duration_seconds=10.0 if not adversarial else 10.0,
+            required_digest=DIGEST_HONEST,
         )
     )
 
@@ -386,14 +395,14 @@ async def run_offline(mode: Mode) -> dict[str, Any]:
 
     # --- 1. Allowlist: BASE-produced digest registration (submission identity) ---
     allowlist = DigestAllowlist()
-    allowlist.register(
-        DigestRecord(
-            commit_sha=COMMIT_SHA,
-            tree_sha=TREE_SHA,
-            variant=VARIANT,
-            digest=DIGEST_HONEST,
-        )
+    honest_record = DigestRecord(
+        commit_sha=COMMIT_SHA,
+        tree_sha=TREE_SHA,
+        variant=VARIANT,
+        digest=DIGEST_HONEST,
+        sealed_manifest_hashes=dict(SEALED_MANIFEST),
     )
+    allowlist.register(honest_record)
 
     # --- 2. Continuous constation runner/poller (fixture Lium, real runner) ---
     # First pass uses placeholder work_unit; we re-bind nonce to real submission id.
@@ -467,7 +476,7 @@ def verify_signature(_signed: object) -> Any:
         pod_id=POD_ID,
         nonce=issued.nonce,
         signed_attestation={"sig": "fixture-ok"},
-        expected_sealed_manifest_hashes=dict(SEALED_MANIFEST),
+        expected_sealed_manifest_hashes=dict(honest_record.sealed_manifest_hashes),
         reported_sealed_manifest_hashes=dict(SEALED_MANIFEST),
         lium_declared_digest=lium_for_bundle,
         constation_gap_budget_seconds=gap_budget,
@@ -584,12 +593,12 @@ def _assert_honest(bag: dict[str, Any]) -> list[str]:
 
 def _assert_adversarial(bag: dict[str, Any]) -> list[str]:
     errors: list[str] = []
-    # Runner must fail closed on mid-run swap (corroboration mismatch).
+    # Runner must fail closed on mid-run swap (triangle: sidecar != required).
     if bag["run_record_ok"]:
         errors.append("runner expected fail on mid-run digest swap")
-    if bag["run_record_reason"] != ConstationFailCode.CORROBORATION_MISMATCH.value:
+    if bag["run_record_reason"] != ConstationFailCode.REQUIRED_DIGEST_MISMATCH.value:
         errors.append(
-            f"runner reason expected corroboration_mismatch got {bag['run_record_reason']}"  # noqa: E501
+            f"runner reason expected required_digest_mismatch got {bag['run_record_reason']}"  # noqa: E501
         )
     if bag["run_record_fault_class"] != FaultClass.MINER.value:
         errors.append(

scripts/prism_lium_attestation_e2e.py

@@ -183,16 +183,27 @@ async def attest(self, *, pod_id: str, phase: str) -> str:
 @dataclass
 class ScriptedLium:
     digests: list[str | None]
+    miner_hotkey: str = HOTKEY
+    status: str = "RUNNING"
     calls: int = 0
 
     async def get_pod_raw(self, pod_id: str) -> LiumPodRead:
         self.calls += 1
         idx = min(self.calls - 1, len(self.digests) - 1)
+        digest = self.digests[idx]
         return LiumPodRead(
             pod_id=pod_id,
             template_id="tmpl-fixture",
-            docker_image_digest=self.digests[idx],
-            raw={},
+            docker_image_digest=digest,
+            raw={
+                "id": pod_id,
+                "status": self.status,
+                "executor": {
+                    "miner_hotkey": self.miner_hotkey,
+                    "executor_ip_address": "10.0.0.1",
+                },
+                "template": {"docker_image_digest": digest},
+            },
         )
 
     async def balance(self) -> float:
@@ -264,6 +275,7 @@ def _register_allowlist(allowlist: DigestAllowlist, digest: str) -> None:
             tree_sha=TREE,
             variant=ImageVariant.CUDA,
             digest=digest,
+            sealed_manifest_hashes=dict(SEALED),
         )
     )
 
@@ -447,7 +459,7 @@ async def sleep_fn(seconds: float) -> None:
             work_unit_id=work_unit_id,
             pod_id=pod_id,
             duration_seconds=duration,
-            expected_sidecar_digest=sidecar.digest,
+            required_digest=sidecar.digest,
         )
     )
 

scripts/surface_constation.sh

@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+# T12: production constation surface scenarios S1–S8 + B1s/B2s.
+# Exit non-zero on any failure. No live Lium.
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$ROOT"
+
+export UV_CACHE_DIR="${UV_CACHE_DIR:-/var/tmp/uv-cache}"
+# Prism optional surface pieces (S3/S6) + local surface helpers.
+export PYTHONPATH="${ROOT}/packages/challenges/prism/src:${ROOT}/tests/surface${PYTHONPATH:+:${PYTHONPATH}}"
+
+SURFACE_NODEIDS=(
+  "tests/surface/test_constation_production_surface.py::test_s1_honest_path_seal_store_forward_nests_bundle"
+  "tests/surface/test_constation_production_surface.py::test_s2_adversarial_triangle_fail_no_ok_bundle_put"
+  "tests/surface/test_constation_production_surface_prism_http.py::test_s3_missing_bundle_prism_behavior"
+  "tests/surface/test_constation_production_surface.py::test_s4_nonce_issue_seal_consume_once_ok"
+  "tests/surface/test_constation_production_surface.py::test_s5_allowlist_sealed_hashes_required"
+  "tests/surface/test_constation_production_surface_prism_http.py::test_s6_legacy_gate_no_elevation_without_constation"
+  "tests/surface/test_constation_production_surface.py::test_s7_forwarder_embed_from_lookup"
+  "tests/surface/test_constation_production_surface_prism_http.py::test_s8_challenge_route_smoke"
+  "tests/surface/test_constation_production_surface.py::test_b1s_verify_key_wiring_smoke"
+  "tests/surface/test_constation_production_surface.py::test_b2s_orchestrator_source_never_calls_consume"
+)
+
+echo "==> surface_constation: UV_CACHE_DIR=${UV_CACHE_DIR}"
+echo "==> running ${#SURFACE_NODEIDS[@]} node ids"
+
+exec uv run pytest -q --tb=short "${SURFACE_NODEIDS[@]}"

src/base/cli_app/main.py

@@ -550,6 +550,7 @@ def _master_orchestration_driver(
     worker_service: WorkerCoordinationService | None = None,
     worker_assignment_service: WorkerAssignmentService | None = None,
     bundle_store: Any | None = None,
+    constation_hook: Any | None = None,
 ) -> MasterOrchestrationDriver:
     """Build the live master orchestration driver (architecture.md sec 4).
 
@@ -602,6 +603,7 @@ def _bundle_lookup(work_unit_id: str):
                 retries=settings.master.challenge_retries,
                 bundle_lookup=_bundle_lookup if bundle_store is not None else None,
             ),
+            constation_hook=constation_hook,
         )
     return MasterOrchestrationDriver(
         assignment_service=assignment_service,
@@ -1147,7 +1149,12 @@ def master_proxy(config: Path = typer.Option(Path("config/master.example.yaml"))
     from datetime import timedelta
 
     from base.master.constation.allowlist_repository import DigestAllowlistRepository
+    from base.master.constation.attestation_keys import load_attestation_verify_key
     from base.master.constation.bundle_store import ConstationBundleStore
+    from base.master.constation.custody_keys import (
+        build_constation_runtime,
+        make_constation_pre_forward_hook,
+    )
     from base.master.constation.nonce_repository import DurableAttestationNonceService
     from base.master.constation.routes import build_constation_router
 
@@ -1156,6 +1163,19 @@ def master_proxy(config: Path = typer.Option(Path("config/master.example.yaml"))
     constation_nonce_service = DurableAttestationNonceService(
         session_factory, ttl=timedelta(hours=2)
     )
+    # Custody + MinerPodBinding + ProductionConstationOrchestrator when enabled
+    # and custody master key is present (fail-closed otherwise; master still boots).
+    constation_runtime = build_constation_runtime(
+        settings,
+        nonce_service=constation_nonce_service,
+        bundle_store=constation_bundle_store,
+    )
+    constation_pod_binding = constation_runtime.pod_binding
+    constation_orchestrator = constation_runtime.orchestrator
+    constation_hook = make_constation_pre_forward_hook(
+        constation_orchestrator,
+        duration_seconds=float(settings.constation.duration_seconds),
+    )
     # Internal token for master constation check_*/issue/register/bundle.
     # Reuse admin token when present so ops and prism can share one secret.
     constation_internal_token = (
@@ -1170,6 +1190,8 @@ def master_proxy(config: Path = typer.Option(Path("config/master.example.yaml"))
         nonce_service=constation_nonce_service,
         bundle_store=constation_bundle_store,
         internal_token=str(constation_internal_token),
+        attestation_verify_key=load_attestation_verify_key(settings),
+        pod_binding=constation_pod_binding,
     )
 
     orchestration_driver = _master_orchestration_driver(
@@ -1180,6 +1202,7 @@ def master_proxy(config: Path = typer.Option(Path("config/master.example.yaml"))
         worker_service=worker_service,
         worker_assignment_service=worker_assignment_service,
         bundle_store=constation_bundle_store,
+        constation_hook=constation_hook,
     )
     # Registry-driven challenge deploy (architecture.md sec 4 + sec 9.2): the
     # master reconcile loop turns every ACTIVE registry challenge into a running
@@ -1256,6 +1279,11 @@ def master_proxy(config: Path = typer.Option(Path("config/master.example.yaml"))
         ),
         constation_router=constation_router,
     )
+    # Expose constation services for worker-path / ops reachability (optional).
+    if constation_pod_binding is not None:
+        proxy.state.constation_pod_binding = constation_pod_binding
+    if constation_orchestrator is not None:
+        proxy.state.constation_orchestrator = constation_orchestrator
     endpoint = f"{settings.master.proxy_host}:{settings.master.proxy_port}"
     typer.echo(f"Starting proxy API on {endpoint}")
     uvicorn.run(proxy, host=settings.master.proxy_host, port=settings.master.proxy_port)

src/base/compute/__init__.py

@@ -24,6 +24,11 @@
     LiumKeyCustody,
     generate_custody_master_key,
 )
+from base.compute.constation_pod import (
+    assert_pod_bound,
+    extract_miner_hotkey,
+    pod_is_running,
+)
 from base.compute.constation_poller import (
     ContinuousConstationPoller,
     PollerConfig,
@@ -33,6 +38,22 @@
     ConstationRunner,
     ConstationRunRequest,
 )
+from base.compute.constation_sidecar_client import (
+    HttpSidecarAttestor,
+    SidecarAttestHit,
+    SidecarAttestMiss,
+    SidecarAttestResult,
+)
+from base.compute.constation_sidecar_endpoint import (
+    SidecarEndpointHit,
+    SidecarEndpointMiss,
+    SidecarEndpointMissReason,
+    resolve_sidecar_base_url,
+)
+from base.compute.constation_triangle import (
+    DigestTriangleResult,
+    evaluate_digest_triangle,
+)
 from base.compute.constation_types import (
     ConstationFailCode,
     ConstationRunRecord,
@@ -146,5 +167,18 @@
     "PollerConfig",
     "PollerResult",
     "evaluate_corroboration",
+    "DigestTriangleResult",
+    "evaluate_digest_triangle",
     "generate_custody_master_key",
+    "HttpSidecarAttestor",
+    "SidecarAttestHit",
+    "SidecarAttestMiss",
+    "SidecarAttestResult",
+    "SidecarEndpointHit",
+    "SidecarEndpointMiss",
+    "SidecarEndpointMissReason",
+    "resolve_sidecar_base_url",
+    "assert_pod_bound",
+    "extract_miner_hotkey",
+    "pod_is_running",
 ]

src/base/compute/constation_custody.py

@@ -119,6 +119,18 @@ async def register(self, *, miner_hotkey: str, api_key: str) -> ConstationVerdic
         logger.info("lium key registered (encrypted) for miner_hotkey=%s", hotkey)
         return ConstationVerdict(ok=True, reason=ConstationFailCode.OK)
 
+    def store_probed_key(self, *, miner_hotkey: str, api_key: str) -> None:
+        """Encrypt-at-rest after the caller has already validated the key.
+
+        No network I/O. Used by pod binding after probe + assert_pod_bound so a
+        failed bind never leaves a half-registered key.
+        """
+        hotkey = _require_nonblank("miner_hotkey", miner_hotkey)
+        key = _require_nonblank("api_key", api_key)
+        token = self._fernet.encrypt(key.encode("utf-8"))
+        self._by_hotkey[hotkey] = token
+        logger.info("lium key stored (encrypted) for miner_hotkey=%s", hotkey)
+
     def unlock_api_key(self, miner_hotkey: str) -> str:
         """Decrypt the stored key for in-process use. Raises if missing/corrupt."""
         hotkey = _require_nonblank("miner_hotkey", miner_hotkey)