fix(benchmarks): define fail-closed production trust boundary

benchmarks/codegraph_compare/production_trust.py

@@ -0,0 +1,180 @@
+"""Fail-closed production trust qualification for the NO1-002C canary.
+
+This module validates only operator-controlled configuration and immutable
+references.  It deliberately cannot sign attestations, reserve spend, collect
+evidence, or dispatch a model request; those responsibilities belong to the
+independent services described by issue #1223.
+"""
+
+from __future__ import annotations
+
+import math
+import re
+from dataclasses import dataclass
+from pathlib import Path
+
+from benchmarks.codegraph_compare.canary_evidence import canonical_sha256
+
+SCHEMA_VERSION = 1
+_SHA256 = re.compile(r"^[0-9a-f]{64}$")
+_REQUIRED_ROLES = frozenset(
+    {"anchor-custodian", "budget-gateway", "evidence-collector"}
+)
+
+
+def _nonempty(value: object, label: str) -> str:
+    if type(value) is not str or not value:
+        raise ValueError(f"{label} must be a non-empty string")
+    return value
+
+
+def _digest(value: object, label: str) -> str:
+    if type(value) is not str or _SHA256.fullmatch(value) is None:
+        raise ValueError(f"{label} must be a lowercase SHA-256 digest")
+    return value
+
+
+@dataclass(frozen=True)
+class ProductionRunSpecV1:
+    manifest_hash: str
+    cell_id: str
+    model: str
+    prompt_sha256: str
+    launch_identity_sha256: str
+    workspace_baseline_sha256: str
+    budget_ceiling_usd: float
+    token_limit: int
+    request_limit: int
+    nonce: str
+    expires_at_unix: int
+
+    @property
+    def spec_hash(self) -> str:
+        validate_production_run_spec(self)
+        return canonical_sha256({"schema_version": SCHEMA_VERSION, **self.__dict__})
+
+
+@dataclass(frozen=True)
+class OperatorTrustConfigV1:
+    """References supplied out of band; never serialized into evidence bundles."""
+
+    trust_store: Path
+    pinned_anchor: Path
+    immutable_artifact_root: Path
+    trusted_roles: frozenset[str]
+    provider_budget_enforced: bool
+    append_only_ledger: bool
+    immutable_collector: bool
+    isolated_execution: bool
+    verification_to_use_closed: bool
+    independent_judge: bool
+
+
+@dataclass(frozen=True)
+class ProductionQualification:
+    status: str
+    violations: tuple[str, ...]
+    spec_hash: str | None
+    model_callbacks_allowed: bool
+
+
+def validate_production_run_spec(spec: ProductionRunSpecV1) -> None:
+    _digest(spec.manifest_hash, "manifest_hash")
+    _nonempty(spec.cell_id, "cell_id")
+    _nonempty(spec.model, "model")
+    _digest(spec.prompt_sha256, "prompt_sha256")
+    _digest(spec.launch_identity_sha256, "launch_identity_sha256")
+    _digest(spec.workspace_baseline_sha256, "workspace_baseline_sha256")
+    if (
+        type(spec.budget_ceiling_usd) not in (int, float)
+        or not math.isfinite(spec.budget_ceiling_usd)
+        or spec.budget_ceiling_usd <= 0
+    ):
+        raise ValueError("budget_ceiling_usd must be finite and positive")
+    for label, value in (
+        ("token_limit", spec.token_limit),
+        ("request_limit", spec.request_limit),
+        ("expires_at_unix", spec.expires_at_unix),
+    ):
+        if type(value) is not int or value <= 0:
+            raise ValueError(f"{label} must be a positive integer")
+    _nonempty(spec.nonce, "nonce")
+
+
+def _trusted_external_file(path: Path, bundle_root: Path, label: str) -> str | None:
+    try:
+        resolved = path.resolve(strict=True)
+    except (OSError, RuntimeError):
+        return f"{label.upper()}_UNAVAILABLE"
+    bundle = bundle_root.resolve(strict=False)
+    if not resolved.is_file():
+        return f"{label.upper()}_NOT_FILE"
+    if resolved == bundle or bundle in resolved.parents:
+        return f"{label.upper()}_BUNDLE_CONTROLLED"
+    if path.is_symlink():
+        return f"{label.upper()}_SYMLINK"
+    return None
+
+
+def qualify_production_trust(
+    spec: ProductionRunSpecV1,
+    config: OperatorTrustConfigV1 | None,
+    *,
+    evidence_bundle_root: Path,
+    now_unix: int,
+) -> ProductionQualification:
+    """Return a fail-closed qualification without invoking production services."""
+
+    try:
+        validate_production_run_spec(spec)
+    except ValueError as error:
+        return ProductionQualification(
+            "INVALID", (f"RUN_SPEC_INVALID:{error}",), None, False
+        )
+    spec_hash = spec.spec_hash
+    if config is None:
+        return ProductionQualification(
+            "NOT_EVALUATED", ("OPERATOR_TRUST_CONFIG_UNAVAILABLE",), spec_hash, False
+        )
+
+    violations: list[str] = []
+    for path, label in (
+        (config.trust_store, "trust_store"),
+        (config.pinned_anchor, "pinned_anchor"),
+    ):
+        violation = _trusted_external_file(path, evidence_bundle_root, label)
+        if violation is not None:
+            violations.append(violation)
+    artifact_root = config.immutable_artifact_root.resolve(strict=False)
+    bundle = evidence_bundle_root.resolve(strict=False)
+    if artifact_root == bundle or bundle in artifact_root.parents:
+        violations.append("ARTIFACT_ROOT_BUNDLE_CONTROLLED")
+    if config.immutable_artifact_root.exists():
+        violations.append("ARTIFACT_ROOT_PREEXISTS")
+    if config.trusted_roles != _REQUIRED_ROLES:
+        violations.append("TRUST_ROLES_INCOMPLETE")
+    for enabled, violation in (
+        (config.provider_budget_enforced, "PROVIDER_BUDGET_GATEWAY_UNAVAILABLE"),
+        (config.append_only_ledger, "APPEND_ONLY_LEDGER_UNAVAILABLE"),
+        (config.immutable_collector, "IMMUTABLE_COLLECTOR_UNAVAILABLE"),
+        (config.isolated_execution, "ISOLATED_EXECUTION_UNAVAILABLE"),
+        (config.verification_to_use_closed, "VERIFICATION_TO_USE_OPEN"),
+        (config.independent_judge, "INDEPENDENT_JUDGE_UNAVAILABLE"),
+    ):
+        if enabled is not True:
+            violations.append(violation)
+    if type(now_unix) is not int or now_unix < 0:
+        violations.append("TRUSTED_CLOCK_INVALID")
+    elif spec.expires_at_unix <= now_unix:
+        violations.append("RUN_SPEC_EXPIRED")
+
+    if violations:
+        return ProductionQualification(
+            "NOT_EVALUATED", tuple(violations), spec_hash, False
+        )
+    return ProductionQualification(
+        "NOT_EVALUATED",
+        ("SIGNED_ATTESTATIONS_AND_JUDGE_VERDICT_REQUIRED",),
+        spec_hash,
+        False,
+    )

tests/unit/test_production_trust.py

@@ -0,0 +1,127 @@
+"""Behavioral tests for the NO1-002D production trust boundary."""
+
+from __future__ import annotations
+
+from dataclasses import replace
+from pathlib import Path
+
+from benchmarks.codegraph_compare.production_trust import (
+    OperatorTrustConfigV1,
+    ProductionRunSpecV1,
+    qualify_production_trust,
+)
+
+
+def _spec() -> ProductionRunSpecV1:
+    return ProductionRunSpecV1(
+        manifest_hash="a" * 64,
+        cell_id="tsa-warm-canary",
+        model="gpt-production",
+        prompt_sha256="b" * 64,
+        launch_identity_sha256="c" * 64,
+        workspace_baseline_sha256="d" * 64,
+        budget_ceiling_usd=3.0,
+        token_limit=100_000,
+        request_limit=2,
+        nonce="judge-issued-nonce",
+        expires_at_unix=2_000_000_000,
+    )
+
+
+def _config(tmp_path: Path, bundle: Path) -> OperatorTrustConfigV1:
+    operator = tmp_path / "operator"
+    operator.mkdir()
+    trust_store = operator / "trust-store.json"
+    anchor = operator / "anchor.json"
+    trust_store.write_text("{}\n", encoding="utf-8")
+    anchor.write_text("{}\n", encoding="utf-8")
+    return OperatorTrustConfigV1(
+        trust_store=trust_store,
+        pinned_anchor=anchor,
+        immutable_artifact_root=tmp_path / "collector" / "new-run",
+        trusted_roles=frozenset(
+            {"anchor-custodian", "budget-gateway", "evidence-collector"}
+        ),
+        provider_budget_enforced=True,
+        append_only_ledger=True,
+        immutable_collector=True,
+        isolated_execution=True,
+        verification_to_use_closed=True,
+        independent_judge=True,
+    )
+
+
+def test_missing_operator_configuration_blocks_every_model_callback(tmp_path: Path):
+    # Issue #1223: bundle-only qualification must never enable production.
+    result = qualify_production_trust(
+        _spec(), None, evidence_bundle_root=tmp_path, now_unix=1_900_000_000
+    )
+
+    assert result.status == "NOT_EVALUATED"
+    assert result.violations == ("OPERATOR_TRUST_CONFIG_UNAVAILABLE",)
+    assert result.model_callbacks_allowed is False
+
+
+def test_bundle_provided_trust_store_is_rejected(tmp_path: Path):
+    # Issue #1223: bundled keys and TOFU are forbidden.
+    bundle = tmp_path / "bundle"
+    bundle.mkdir()
+    config = _config(tmp_path, bundle)
+    bundled_store = bundle / "trust-store.json"
+    bundled_store.write_text("{}\n", encoding="utf-8")
+
+    result = qualify_production_trust(
+        _spec(),
+        replace(config, trust_store=bundled_store),
+        evidence_bundle_root=bundle,
+        now_unix=1_900_000_000,
+    )
+
+    assert result.violations == ("TRUST_STORE_BUNDLE_CONTROLLED",)
+    assert result.model_callbacks_allowed is False
+
+
+def test_expired_spec_is_rejected_by_trusted_clock(tmp_path: Path):
+    # Issue #1223: expired attestations must fail closed.
+    bundle = tmp_path / "bundle"
+    bundle.mkdir()
+    config = _config(tmp_path, bundle)
+
+    result = qualify_production_trust(
+        _spec(), config, evidence_bundle_root=bundle, now_unix=2_000_000_000
+    )
+
+    assert result.violations == ("RUN_SPEC_EXPIRED",)
+    assert result.model_callbacks_allowed is False
+
+
+def test_precreated_artifact_root_is_rejected(tmp_path: Path):
+    # Issue #1223: collectors must create a fresh externally controlled root.
+    bundle = tmp_path / "bundle"
+    bundle.mkdir()
+    config = _config(tmp_path, bundle)
+    config.immutable_artifact_root.mkdir(parents=True)
+
+    result = qualify_production_trust(
+        _spec(), config, evidence_bundle_root=bundle, now_unix=1_900_000_000
+    )
+
+    assert result.violations == ("ARTIFACT_ROOT_PREEXISTS",)
+    assert result.model_callbacks_allowed is False
+
+
+def test_complete_external_configuration_still_requires_signed_judge_evidence(
+    tmp_path: Path,
+):
+    # Issue #1223: local configuration alone cannot self-qualify production.
+    bundle = tmp_path / "bundle"
+    bundle.mkdir()
+    config = _config(tmp_path, bundle)
+
+    result = qualify_production_trust(
+        _spec(), config, evidence_bundle_root=bundle, now_unix=1_900_000_000
+    )
+
+    assert result.status == "NOT_EVALUATED"
+    assert result.violations == ("SIGNED_ATTESTATIONS_AND_JUDGE_VERDICT_REQUIRED",)
+    assert result.model_callbacks_allowed is False