feat(constation): production Lium image-attestation wire

alembic/versions/0017_image_digest_allowlist.py

@@ -0,0 +1,110 @@
+"""BASE-produced image digest allowlist + revocation denylists.
+
+Adds durable storage for mechanism 4 (digest allowlist) of prism-lium image
+attestation. Lookup rules live in ``base.compute.digest_allowlist``; these
+tables only persist registered bindings and denylist entries.
+
+Revision ID: 0017_digest_allowlist
+Revises: 0016_watcher_state
+Create Date: 2026-07-26 00:00:00.000000
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "0017_digest_allowlist"
+down_revision: str | None = "0016_watcher_state"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Apply the migration."""
+
+    op.create_table(
+        "image_digest_allowlist",
+        sa.Column("id", sa.Uuid(as_uuid=True), nullable=False),
+        sa.Column("commit_sha", sa.Text(), nullable=False),
+        sa.Column("tree_sha", sa.Text(), nullable=False),
+        sa.Column("variant", sa.Text(), nullable=False),
+        sa.Column("digest", sa.Text(), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_image_digest_allowlist")),
+        sa.UniqueConstraint("digest", name="uq_image_digest_allowlist_digest"),
+        sa.UniqueConstraint(
+            "commit_sha",
+            "tree_sha",
+            "variant",
+            name="uq_image_digest_allowlist_commit_tree_variant",
+        ),
+    )
+    op.create_index(
+        "ix_image_digest_allowlist_commit_sha",
+        "image_digest_allowlist",
+        ["commit_sha"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_image_digest_allowlist_variant",
+        "image_digest_allowlist",
+        ["variant"],
+        unique=False,
+    )
+
+    op.create_table(
+        "denied_image_digests",
+        sa.Column("digest", sa.Text(), nullable=False),
+        sa.Column("reason", sa.Text(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+        sa.PrimaryKeyConstraint("digest", name=op.f("pk_denied_image_digests")),
+    )
+
+    op.create_table(
+        "denied_image_commits",
+        sa.Column("commit_sha", sa.Text(), nullable=False),
+        sa.Column("reason", sa.Text(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.func.now(),
+            nullable=False,
+        ),
+        sa.PrimaryKeyConstraint("commit_sha", name=op.f("pk_denied_image_commits")),
+    )
+
+
+def downgrade() -> None:
+    """Revert the migration."""
+
+    op.drop_table("denied_image_commits")
+    op.drop_table("denied_image_digests")
+    op.drop_index(
+        "ix_image_digest_allowlist_variant",
+        table_name="image_digest_allowlist",
+    )
+    op.drop_index(
+        "ix_image_digest_allowlist_commit_sha",
+        table_name="image_digest_allowlist",
+    )
+    op.drop_table("image_digest_allowlist")

alembic/versions/0018_attestation_nonces.py

@@ -0,0 +1,75 @@
+"""BASE-issued single-use attestation nonces for prism-lium constation.
+
+Adds durable storage for mechanism 1 (nonce-bound attestation). Issue/consume
+rules live in ``base.compute.attestation_nonce``; this table only persists
+issued nonces and their consume timestamps (BASE clocks only).
+
+Revision ID: 0018_attestation_nonces
+Revises: 0017_digest_allowlist
+Create Date: 2026-07-26 00:00:00.000000
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "0018_attestation_nonces"
+down_revision: str | None = "0017_digest_allowlist"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Apply the migration."""
+
+    op.create_table(
+        "attestation_nonces",
+        sa.Column("nonce", sa.Text(), nullable=False),
+        sa.Column("work_unit_id", sa.Text(), nullable=False),
+        sa.Column("miner_hotkey", sa.Text(), nullable=False),
+        sa.Column("pod_id", sa.Text(), nullable=False),
+        sa.Column("issued_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("consumed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.PrimaryKeyConstraint("nonce", name=op.f("pk_attestation_nonces")),
+    )
+    op.create_index(
+        "ix_attestation_nonces_work_unit_id",
+        "attestation_nonces",
+        ["work_unit_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_attestation_nonces_miner_hotkey",
+        "attestation_nonces",
+        ["miner_hotkey"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_attestation_nonces_expires_at",
+        "attestation_nonces",
+        ["expires_at"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    """Revert the migration."""
+
+    op.drop_index(
+        "ix_attestation_nonces_expires_at",
+        table_name="attestation_nonces",
+    )
+    op.drop_index(
+        "ix_attestation_nonces_miner_hotkey",
+        table_name="attestation_nonces",
+    )
+    op.drop_index(
+        "ix_attestation_nonces_work_unit_id",
+        table_name="attestation_nonces",
+    )
+    op.drop_table("attestation_nonces")

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

@@ -39,6 +39,113 @@
 from .weights import get_weights
 
 
+def _constation_ingest_kwargs(
+    app_settings: PrismSettings,
+    result_payload: object,
+    *,
+    app: FastAPI | None = None,
+) -> dict:
+    """Resolve constation kwargs for ingest: prod wire OR insecure test seam.
+
+    Production: deserialize ``result.constation_bundle`` and attach checkers.
+    Preference:
+      1) BASE HTTP when ``constation_base_url`` + token (durable master SoT)
+      2) in-process app.state services (same issuer as local challenge routes)
+    Missing bundle returns {} so ingest fail-closes (P1). Never auto-injects
+    synthetic bundles in prod.
+    """
+    if getattr(app_settings, "allow_insecure_signatures", False):
+        return _test_constation_kwargs(app_settings, result_payload)
+
+    if not isinstance(result_payload, dict):
+        return {}
+    raw_bundle = result_payload.get("constation_bundle")
+    if raw_bundle is None:
+        return {}
+
+    from .constation import constation_bundle_from_dict
+
+    try:
+        bundle = constation_bundle_from_dict(raw_bundle)
+    except (TypeError, ValueError):
+        # Malformed embedded bundle — leave to gate as missing/invalid.
+        return {}
+
+    base_url = getattr(app_settings, "constation_base_url", None)
+    token = getattr(app_settings, "constation_internal_token", None) or ""
+    if base_url and token:
+        from .constation_checkers import BaseHttpConstationClient
+
+        client = BaseHttpConstationClient(base_url=str(base_url), token=str(token))
+        return {
+            "constation_bundle": bundle,
+            "check_allowlist": client.check_allowlist,
+            "check_nonce": client.check_nonce,
+            "verify_constation_signature": client.verify_signature,
+        }
+
+    if app is not None:
+        from .attestation_routes import make_inprocess_checkers
+
+        try:
+            checkers = make_inprocess_checkers(app)
+            return {
+                "constation_bundle": bundle,
+                "check_allowlist": checkers["check_allowlist"],
+                "check_nonce": checkers["check_nonce"],
+                "verify_constation_signature": checkers["verify_constation_signature"],
+            }
+        except Exception:
+            pass
+
+    # Bundle present but checkers not configured: still pass bundle so
+    # constation_ok fails without elevating (checkers required → fail closed).
+    return {"constation_bundle": bundle}
+
+
+def _test_constation_kwargs(app_settings: PrismSettings, result_payload: object) -> dict:
+    """Supply a valid constation bundle only under allow_insecure_signatures (unit tests).
+
+    Production (allow_insecure_signatures=False) never auto-injects — missing bundle
+    fail-closes with no score (P1 / todo 22). Result payloads may still carry an
+    explicit constation block later; this helper is the test seam only.
+    """
+    if not getattr(app_settings, "allow_insecure_signatures", False):
+        return {}
+    # If the payload already embeds constation markers, do not override.
+    if isinstance(result_payload, dict) and result_payload.get("constation_bundle") is not None:
+        return {}
+    from .constation import CheckOutcome, ConstationBundle
+
+    def _ok(**_k: object) -> CheckOutcome:
+        return CheckOutcome(ok=True, reason="ok")
+
+    man = {"route-test-harness.py": "a" * 64}
+    digest = "sha256:" + ("11" * 32)
+    bundle = ConstationBundle(
+        commit_sha="a" * 40,
+        tree_sha="b" * 40,
+        variant="cuda",
+        digest=digest,
+        work_unit_id="route-wu",
+        miner_hotkey="route-hk",
+        pod_id="route-pod",
+        nonce="route-nonce",
+        signed_attestation={"route": True},
+        expected_sealed_manifest_hashes=dict(man),
+        reported_sealed_manifest_hashes=dict(man),
+        lium_declared_digest=digest,
+        constation_gap_budget_seconds=30.0,
+        constation_observed_max_gap_seconds=1.0,
+    )
+    return {
+        "constation_bundle": bundle,
+        "check_allowlist": _ok,
+        "check_nonce": _ok,
+        "verify_constation_signature": lambda _s: _ok(),
+    }
+
+
 def create_app(
     app_settings: PrismSettings | None = None,
     *,
@@ -147,6 +254,15 @@ async def _run_raw_weight_push(app: FastAPI) -> None:
     app.state.checkpoint_publisher = publisher
     app.state.checkpoint_intake = checkpoint_intake
 
+    # Attestation constation hosts (public challenge via routes; internal check/*).
+    from .attestation_routes import (
+        build_attestation_internal_router,
+        ensure_default_constation_services,
+    )
+
+    ensure_default_constation_services(app)
+    app.include_router(build_attestation_internal_router())
+
     @app.post("/internal/v1/worker/process-next", dependencies=[Depends(authenticate_internal)])
     async def process_next() -> dict[str, str | None]:
         return {"submission_id": await worker.process_next()}
@@ -320,6 +436,7 @@ async def work_unit_result(request: Request) -> dict[str, object]:
                 result=result_payload,
                 pinned_image_digest=app_settings.worker_plane.pinned_image_digest,
                 audit_sampler=sampler,
+                **_constation_ingest_kwargs(app_settings, result_payload, app=request.app),
             )
         except ResultIngestionError as exc:
             # A transient finalization failure is retryable -> 503 so the forwarder retries; the
@@ -343,6 +460,15 @@ async def work_unit_result(request: Request) -> dict[str, object]:
                 status.HTTP_409_CONFLICT,
                 {"code": outcome.reason, "detail": "conflicting result for finalized unit"},
             )
+        if outcome.status == "rejected":
+            # P1 fail-closed: no score written; surface the miner/infra fault code.
+            raise HTTPException(
+                status.HTTP_422_UNPROCESSABLE_ENTITY,
+                {
+                    "code": outcome.reason or "constation_rejected",
+                    "detail": "constation gate refused score",
+                },
+            )
         return outcome.to_response()
 
     @app.post(