diff --git a/scripts/channel_adapters/proof_host.py b/scripts/channel_adapters/proof_host.py index a4d5c454c..959699ed8 100644 --- a/scripts/channel_adapters/proof_host.py +++ b/scripts/channel_adapters/proof_host.py @@ -11,6 +11,7 @@ import shlex import sys import tempfile from pathlib import Path +from typing import Any ROOT = Path(__file__).resolve().parents[2] if str(ROOT) not in sys.path: @@ -47,14 +48,19 @@ if sys.version_info < (3, 11): raise SystemExit(7) sys.path.insert(0, {clone_root!r}) from scripts.release_digest import file_sha256_size -from scripts.release_install_smoke import run_install_proof +from scripts.release_install_smoke import run_install_proof as run_install_smoke_proof +from scripts.release_nvattest_proof import run_nvattest_proof reqdir = Path({reqdir!r}) req = json.loads((reqdir / "request.json").read_text()) ledger = json.loads((reqdir / "ledger.json").read_text()) -candidate_dir = reqdir / "candidate" +candidate_dir = reqdir / req["paths"]["candidate_dir"] +support_dir = reqdir / req["paths"]["support_dir"] +authority_path = reqdir / req["paths"]["authority_file"] candidate_paths = [candidate_dir / f["basename"] for f in req["candidate_files"]] -proof_path = reqdir / "output" / "proof.json" -run_install_proof( +support_paths = [support_dir / f["basename"] for f in req["support_files"]] +install_proof_path = reqdir / req["paths"]["install_proof"] +nvattest_proof_path = reqdir / req["paths"]["nvattest_proof"] +run_install_smoke_proof( target=req["target"], version=req["version"], source_commit=req["source_commit"], @@ -64,14 +70,33 @@ run_install_proof( candidate_dir=candidate_dir, candidate_paths=candidate_paths, ledger_payload=ledger, - output_path=proof_path, + output_path=install_proof_path, ) -proof_sha256, proof_bytes = file_sha256_size(proof_path) -print("PROOF_OK " + json.dumps({{"sha256": proof_sha256, "bytes": proof_bytes}}, sort_keys=True)) +run_nvattest_proof( + target=req["target"], + version=req["version"], + source_commit=req["source_commit"], + core_lock_sha256=req["core_lock_sha256"], + candidate_digest=req["candidate_digest"], + ledger_sha256=req["ledger_sha256"], + challenge=req["challenge"], + candidate_dir=candidate_dir, + candidate_paths=candidate_paths, + support_wheel_paths=support_paths, + output_path=nvattest_proof_path, + canonical_authority_bytes=authority_path.read_bytes(), +) +def descriptor(path): + sha256, byte_count = file_sha256_size(path) + return {{"sha256": sha256, "bytes": byte_count}} +print("PROOF_OK " + json.dumps({{ + "install_proof": descriptor(install_proof_path), + "nvattest_proof": descriptor(nvattest_proof_path), +}}, sort_keys=True)) """ -def _proof_status(stdout: str) -> tuple[str, int]: +def _proof_status(stdout: str) -> dict[str, dict[str, Any]]: for line in stdout.splitlines(): if not line.startswith(f"{PROOF_TOKEN} "): continue @@ -80,11 +105,19 @@ def _proof_status(stdout: str) -> tuple[str, int]: except json.JSONDecodeError: break if isinstance(payload, dict): - sha256 = payload.get("sha256") - byte_count = payload.get("bytes") - if isinstance(sha256, str) and isinstance(byte_count, int): - return sha256, byte_count - die("proof run did not report proof digest/size", detail=stdout) + descriptors: dict[str, dict[str, Any]] = {} + for key in ("install_proof", "nvattest_proof"): + descriptor = payload.get(key) + if not isinstance(descriptor, dict): + break + sha256 = descriptor.get("sha256") + byte_count = descriptor.get("bytes") + if not isinstance(sha256, str) or not isinstance(byte_count, int): + break + descriptors[key] = {"sha256": sha256, "bytes": byte_count} + if set(descriptors) == {"install_proof", "nvattest_proof"}: + return descriptors + die("proof run did not report proof digests/sizes", detail=stdout) def _verify_host(target: str, lane: LaneConfig) -> None: @@ -107,8 +140,11 @@ def prove(target: str, lane: LaneConfig, request_path: Path) -> None: request_dir = request_path.resolve().parent ledger_path = request_dir / req["paths"]["ledger"] candidate_dir = request_dir / req["paths"]["candidate_dir"] - out_proof = request_dir / req["paths"]["proof"] - out_proof.parent.mkdir(parents=True, exist_ok=True) + support_dir = request_dir / req["paths"]["support_dir"] + authority_file = request_dir / req["paths"]["authority_file"] + out_install_proof = request_dir / req["paths"]["install_proof"] + out_nvattest_proof = request_dir / req["paths"]["nvattest_proof"] + out_install_proof.parent.mkdir(parents=True, exist_ok=True) _verify_host(target, lane) @@ -119,7 +155,7 @@ def prove(target: str, lane: LaneConfig, request_path: Path) -> None: } result = run([sys.executable, "-c", harness], check=False, env=clean_env) require_success_token(result, PROOF_TOKEN, "local proof run") - expected_sha256, expected_bytes = _proof_status(result.stdout or "") + proof_status = _proof_status(result.stdout or "") else: work = _remote_work(lane, cohort) rreq = f"{work}/reqdir" @@ -128,7 +164,8 @@ def prove(target: str, lane: LaneConfig, request_path: Path) -> None: ssh_run( lane, f"set -e; rm -rf {quoted_work}; " - f"mkdir -p {quoted_rreq}/candidate {quoted_rreq}/output {quoted_work}/clone", + f"mkdir -p {quoted_rreq}/candidate {quoted_rreq}/support " + f"{quoted_rreq}/authority {quoted_rreq}/output {quoted_work}/clone", ) fd, bundle_name = tempfile.mkstemp( prefix=f"proof-src-{cohort}-", @@ -147,6 +184,10 @@ def prove(target: str, lane: LaneConfig, request_path: Path) -> None: for candidate in req["candidate_files"]: name = candidate["basename"] scp_to(lane, candidate_dir / name, f"{rreq}/candidate/{name}") + for support in req["support_files"]: + name = support["basename"] + scp_to(lane, support_dir / name, f"{rreq}/support/{name}") + scp_to(lane, authority_file, f"{rreq}/{req['paths']['authority_file']}") harness = _harness(f"{work}/clone", rreq) hpath = f"{work}/harness.py" ssh_run( @@ -156,15 +197,20 @@ def prove(target: str, lane: LaneConfig, request_path: Path) -> None: lane, f"{shlex.quote(lane.remote_python)} {shlex.quote(hpath)}", check=False ) require_success_token(result, PROOF_TOKEN, f"remote proof run on {target}") - expected_sha256, expected_bytes = _proof_status(result.stdout or "") - scp_from(lane, f"{rreq}/output/proof.json", out_proof) - - verify_retrieved_file( - out_proof, - expected_sha256=expected_sha256, - expected_bytes=expected_bytes, - label="proof.json", - ) + proof_status = _proof_status(result.stdout or "") + scp_from(lane, f"{rreq}/{req['paths']['install_proof']}", out_install_proof) + scp_from(lane, f"{rreq}/{req['paths']['nvattest_proof']}", out_nvattest_proof) + + for key, path, label in ( + ("install_proof", out_install_proof, "install-proof.json"), + ("nvattest_proof", out_nvattest_proof, "nvattest-proof.json"), + ): + verify_retrieved_file( + path, + expected_sha256=str(proof_status[key]["sha256"]), + expected_bytes=int(proof_status[key]["bytes"]), + label=label, + ) exp_os, exp_arch = TARGET_POLICY[target] response = { "schema_version": 1, @@ -175,10 +221,15 @@ def prove(target: str, lane: LaneConfig, request_path: Path) -> None: "candidate_digest": req["candidate_digest"], "ledger_sha256": req["ledger_sha256"], }, - "proof": { - "path": req["paths"]["proof"], - "sha256": expected_sha256, - "bytes": expected_bytes, + "install_proof": { + "path": req["paths"]["install_proof"], + "sha256": proof_status["install_proof"]["sha256"], + "bytes": proof_status["install_proof"]["bytes"], + }, + "nvattest_proof": { + "path": req["paths"]["nvattest_proof"], + "sha256": proof_status["nvattest_proof"]["sha256"], + "bytes": proof_status["nvattest_proof"]["bytes"], }, } write_json(request_dir / req["paths"]["response"], response) diff --git a/scripts/check_rust_release_manifest.py b/scripts/check_rust_release_manifest.py index 32ef0272b..25a3fbc01 100644 --- a/scripts/check_rust_release_manifest.py +++ b/scripts/check_rust_release_manifest.py @@ -2378,6 +2378,8 @@ def run_fixtures_mode() -> list[Failure]: target_install_paths_from_ledger, write_install_proof, ) + from scripts.build_nvattest_authority import render_nvattest_authority_json + from scripts.release_nvattest_proof import SUPPORT_DISTRIBUTION_NAMES from scripts.release_ledger import read_retained_ledger, write_ledger except ImportError as exc: return [ @@ -2408,6 +2410,26 @@ def run_fixtures_mode() -> list[Failure]: policy_checked_at="2026-07-20T12:00:00Z", result="pass", ) + authority_bytes = render_nvattest_authority_json().encode("utf-8") + nvattest = { + "authority": json.loads(authority_bytes.decode("utf-8")), + "authority_sha256": hashlib.sha256(authority_bytes).hexdigest(), + "challenge": hashlib.sha256(b"fixture challenge").hexdigest(), + "support_distributions": [ + { + "bytes": len(name.encode("utf-8")), + "filename": ( + f"{name.replace('-', '_')}-0.0.{index}-py3-none-any.whl" + ), + "name": name, + "sha256": hashlib.sha256(name.encode("utf-8")).hexdigest(), + "version": f"0.0.{index}", + } + for index, name in enumerate( + sorted(SUPPORT_DISTRIBUTION_NAMES), start=1 + ) + ], + } fixture_root_wheel = next( name for name in expected_package_names(include_models=False) @@ -2553,6 +2575,7 @@ def run_fixtures_mode() -> list[Failure]: if name.endswith(".tar.gz") ), }, + nvattest=nvattest, ) ledger_payload = read_retained_ledger(ledger_path) ledger_sha256 = file_sha256_size(ledger_path)[0] diff --git a/scripts/release_candidate_driver.py b/scripts/release_candidate_driver.py index a4f2d4885..8cf1412b3 100644 --- a/scripts/release_candidate_driver.py +++ b/scripts/release_candidate_driver.py @@ -7,14 +7,18 @@ from __future__ import annotations import argparse +import hashlib import json import os +import secrets import shlex import shutil import stat import subprocess import sys import tomllib +import urllib.request +import zipfile from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass from pathlib import Path @@ -45,6 +49,7 @@ from scripts.check_rust_release_manifest import ( from scripts.check_wheel_contents import ( EXPECTED_MODEL_SHA256, MAX_BASE_WHEEL_BYTES, + NVATTEST_AUTHORITY_MEMBER, check_dist, ) from scripts.normalize_maturin_sdist import ( @@ -72,6 +77,7 @@ from scripts.release_install_smoke import ( RETAINED_PROOF_REPAIR, InstallProofError, _expected_install_members, + _select_names_for_target, candidate_file_entries, target_install_paths_from_ledger, validate_install_proof_bytes, @@ -82,10 +88,26 @@ from scripts.release_ledger import ( validate_native_members_against_release_dir, write_ledger, ) +from scripts.release_nvattest_proof import ( + CHALLENGE_RE, + SUPPORT_DISTRIBUTION_NAMES, + NvattestProofError, + support_distribution_entries, + validate_nvattest_proof_bytes, +) +from scripts.release_nvattest_support import ( + SupportLockEntry, + SupportLockError, + read_support_lock_entries, + support_declarations_from_lock, + validate_support_declarations, + verify_support_wheels_against_lock, +) from scripts.release_proof_host import ( ProofHostError, + TargetProofPaths, proof_channels_from_env, - run_install_proof_with_channels, + run_target_proofs_with_channels, ) from scripts.release_public_evidence import validate_public_evidence_tree from scripts.release_tool_pins import ( @@ -146,6 +168,7 @@ class CandidateReport: candidate_digest: str ledger_sha256: str proof_sha256: Mapping[str, str] + nvattest_sha256: Mapping[str, str] bundle_digest: str @@ -170,7 +193,9 @@ class CandidateServices: create_source_bundle: Callable[[Path, str, Path], SourceBundle] build_host: Callable[[SourceBundle, str, Path], BuildHostResult] cleanup_transients: Callable[[Sequence[Path]], None] - run_install_proof: Callable[..., Path] + challenge_factory: Callable[[], str] + materialize_support_wheels: Callable[[Path], Sequence[Path]] + run_target_proofs: Callable[..., TargetProofPaths] transaction_hook: Callable[[str], None] @@ -1010,6 +1035,39 @@ def _default_cleanup_transients(paths: Sequence[Path]) -> None: ) from None +def _default_materialize_support_wheels(destination: Path) -> tuple[Path, ...]: + try: + root = destination.parents[3] + except IndexError: + raise DriverError( + [ + _failure( + "nvattest support materialization failed", + expected="evidence support directory under release checkout", + actual=destination.name, + repair="bash scripts/release.sh --candidate", + ) + ] + ) from None + try: + entries = read_support_lock_entries(root / "uv.lock") + except SupportLockError as exc: + raise DriverError(_driver_failures_from_support_error(exc)) from None + destination.mkdir(parents=True, exist_ok=True) + paths: list[Path] = [] + for entry in entries: + output_path = destination / entry.filename + temp_path = output_path.with_name(f".{output_path.name}.tmp") + try: + with urllib.request.urlopen(entry.url, timeout=30) as response: + temp_path.write_bytes(response.read()) + os.rename(temp_path, output_path) + finally: + temp_path.unlink(missing_ok=True) + paths.append(output_path) + return tuple(paths) + + def default_services(env: Mapping[str, str] | None = None) -> CandidateServices: build_host = ( _default_build_host_from_env(env) if env is not None else _missing_build_host @@ -1028,7 +1086,9 @@ def default_services(env: Mapping[str, str] | None = None) -> CandidateServices: create_source_bundle=_default_create_source_bundle, build_host=build_host, cleanup_transients=_default_cleanup_transients, - run_install_proof=proof_runner, + challenge_factory=lambda: secrets.token_hex(32), + materialize_support_wheels=_default_materialize_support_wheels, + run_target_proofs=proof_runner, transaction_hook=lambda _point: None, ) @@ -1048,22 +1108,24 @@ def _missing_build_host( ) -def _default_proof_hosts_from_env(env: Mapping[str, str]) -> Callable[..., Path]: +def _default_proof_hosts_from_env( + env: Mapping[str, str], +) -> Callable[..., TargetProofPaths]: try: channels = proof_channels_from_env(env) except ProofHostError as exc: raise DriverError(exc.failures) from None - def proof_runner(**kwargs: Any) -> Path: + def proof_runner(**kwargs: Any) -> TargetProofPaths: try: - return run_install_proof_with_channels(channels, **kwargs) + return run_target_proofs_with_channels(channels, **kwargs) except ProofHostError as exc: raise DriverError(exc.failures) from None return proof_runner -def _missing_proof_host(**_kwargs: Any) -> Path: +def _missing_proof_host(**_kwargs: Any) -> TargetProofPaths: raise DriverError( [ _failure( @@ -1990,6 +2052,393 @@ def _safe_retained_basename(value: Any) -> bool: ) +def _canonical_nvattest_authority_bytes(payload: Mapping[str, Any]) -> bytes: + return (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode("utf-8") + + +def _driver_failures_from_support_error(exc: SupportLockError) -> list[Failure]: + return [ + _failure( + failure.error, + expected=failure.expected, + actual=failure.actual, + repair=failure.repair, + ) + for failure in exc.failures + ] + + +def _driver_failures_from_nvattest_error(exc: NvattestProofError) -> list[Failure]: + return [ + _failure( + failure.error, + expected=failure.expected, + actual=failure.actual, + repair=failure.repair, + ) + for failure in exc.failures + ] + + +def _validate_nvattest_challenge(value: str) -> list[Failure]: + if CHALLENGE_RE.fullmatch(value): + return [] + return [ + _failure( + "nvattest challenge is invalid", + expected="64 lowercase hexadecimal characters", + actual=repr(value), + repair="bash scripts/release.sh --candidate", + ) + ] + + +def _support_destination_failures( + support_dir: Path, + paths: Sequence[Path], + expected_entries: Sequence[SupportLockEntry], +) -> list[Failure]: + failures: list[Failure] = [] + expected_names = {entry.filename for entry in expected_entries} + try: + support_root = support_dir.resolve() + except OSError: + support_root = support_dir + for raw_path in paths: + path = Path(raw_path) + try: + path.resolve().relative_to(support_root) + except (OSError, ValueError): + failures.append( + _failure( + "nvattest support materialization escaped destination", + expected="support wheel path under evidence support directory", + actual=path.name, + repair="bash scripts/release.sh --candidate", + ) + ) + try: + support_entries = {path.name: path for path in support_dir.iterdir()} + except FileNotFoundError: + return [ + _failure( + "nvattest support inventory is not exact", + expected=", ".join(sorted(expected_names)), + actual="", + repair="bash scripts/release.sh --candidate", + ) + ] + if set(support_entries) != expected_names: + failures.append( + _failure( + "nvattest support inventory is not exact", + expected=", ".join(sorted(expected_names)), + actual=", ".join(sorted(support_entries)) or "", + repair="bash scripts/release.sh --candidate", + ) + ) + for name, path in sorted(support_entries.items()): + try: + entry = path.lstat() + except OSError as exc: + failures.append( + _failure( + "nvattest support wheel could not be inspected", + expected=f"regular support wheel {name}", + actual=type(exc).__name__, + repair="bash scripts/release.sh --candidate", + ) + ) + continue + if stat.S_ISLNK(entry.st_mode) or not stat.S_ISREG(entry.st_mode): + failures.append( + _failure( + "nvattest support wheel is not a regular file", + expected=f"regular support wheel {name}", + actual="non-regular", + repair="bash scripts/release.sh --candidate", + ) + ) + return failures + + +def _materialize_verified_support_wheels( + *, + root: Path, + support_dir: Path, + services: CandidateServices, +) -> tuple[tuple[dict[str, object], ...], tuple[Path, ...]]: + try: + expected_entries = read_support_lock_entries(root / "uv.lock") + expected_support = support_declarations_from_lock(expected_entries) + except SupportLockError as exc: + raise DriverError(_driver_failures_from_support_error(exc)) from None + try: + materialized = tuple( + Path(path) for path in services.materialize_support_wheels(support_dir) + ) + except Exception as exc: + raise DriverError( + [ + _failure( + "nvattest support materialization failed", + expected="support wheels materialized into evidence support directory", + actual=type(exc).__name__, + repair="bash scripts/release.sh --candidate", + ) + ] + ) from None + materialization_failures = _support_destination_failures( + support_dir, materialized, expected_entries + ) + if materialization_failures: + raise DriverError(materialization_failures) + support_paths = tuple( + path.resolve() + for path in sorted(support_dir.iterdir(), key=lambda path: path.name) + ) + try: + observed = verify_support_wheels_against_lock(support_paths, expected_entries) + except SupportLockError as exc: + raise DriverError(_driver_failures_from_support_error(exc)) from None + if observed != expected_support: + raise DriverError( + [ + _failure( + "nvattest support declaration is not bound to release lock", + expected=repr(expected_support), + actual=repr(observed), + repair="bash scripts/release.sh --candidate", + ) + ] + ) + return expected_support, support_paths + + +def _root_wheel_name_for_target( + target: str, + names: Sequence[str], +) -> tuple[str | None, list[Failure]]: + selected = _select_names_for_target(target, names) + root_wheels = [name for name in selected if name.startswith("solstone-")] + if len(root_wheels) == 1: + return root_wheels[0], [] + return None, [ + _failure( + "candidate nvattest root wheel selection is invalid", + expected=f"exactly one retained root wheel for {target}", + actual=", ".join(root_wheels) or "", + repair="bash scripts/release.sh --recover", + ) + ] + + +def _extract_nvattest_authority_bytes( + release_dir: Path, + names: Sequence[str], +) -> tuple[bytes | None, list[Failure]]: + failures: list[Failure] = [] + by_target: dict[str, bytes] = {} + for target in PROOF_TARGETS: + root_name, target_failures = _root_wheel_name_for_target(target, names) + failures.extend(target_failures) + if root_name is None: + continue + wheel_path = release_dir / root_name + try: + with zipfile.ZipFile(wheel_path) as wheel: + by_target[target] = wheel.read(NVATTEST_AUTHORITY_MEMBER) + except KeyError: + failures.append( + _failure( + "candidate nvattest authority member is missing", + expected=f"{root_name}:{NVATTEST_AUTHORITY_MEMBER}", + actual="", + repair="bash scripts/release.sh --recover", + ) + ) + except (OSError, zipfile.BadZipFile) as exc: + failures.append( + _failure( + "candidate nvattest authority member could not be read", + expected=f"readable {NVATTEST_AUTHORITY_MEMBER}", + actual=type(exc).__name__, + repair="bash scripts/release.sh --recover", + ) + ) + distinct = {bytes_value for bytes_value in by_target.values()} + if len(distinct) > 1: + failures.append( + _failure( + "candidate nvattest authority members differ by target", + expected="byte-identical authority member in every root wheel", + actual=", ".join(sorted(by_target)), + repair="bash scripts/release.sh --recover", + ) + ) + return (next(iter(distinct)) if len(distinct) == 1 else None), failures + + +def _candidate_names_from_dir(release_dir: Path) -> tuple[str, ...]: + return tuple(sorted(path.name for path in release_dir.iterdir() if path.is_file())) + + +def _nvattest_ledger_payload( + *, + challenge: str, + authority_bytes: bytes, + support_distributions: Sequence[Mapping[str, object]], +) -> dict[str, object]: + failures = _validate_nvattest_challenge(challenge) + try: + authority = json.loads(authority_bytes.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + failures.append( + _failure( + "candidate nvattest authority member is not valid JSON", + expected="canonical authority JSON object", + actual=str(exc), + repair="bash scripts/release.sh --candidate", + ) + ) + authority = {} + if not isinstance(authority, Mapping): + failures.append( + _failure( + "candidate nvattest authority member is not an object", + expected="canonical authority JSON object", + actual=type(authority).__name__, + repair="bash scripts/release.sh --candidate", + ) + ) + authority = {} + elif _canonical_nvattest_authority_bytes(authority) != authority_bytes: + failures.append( + _failure( + "candidate nvattest authority member is not canonical", + expected="canonical sorted authority JSON bytes", + actual="authority bytes differ", + repair="bash scripts/release.sh --candidate", + ) + ) + if failures: + raise DriverError(failures) + return { + "authority": dict(authority), + "authority_sha256": hashlib.sha256(authority_bytes).hexdigest(), + "challenge": challenge, + "support_distributions": [dict(entry) for entry in support_distributions], + } + + +def _retained_support_binding_failures( + *, + evidence_dir: Path, + ledger: Mapping[str, Any], +) -> list[Failure]: + failures: list[Failure] = [] + nvattest = ledger.get("nvattest") + if not isinstance(nvattest, Mapping): + return [ + _failure( + "retained ledger nvattest binding is invalid", + expected="nvattest object", + actual=type(nvattest).__name__, + repair="bash scripts/release.sh --recover", + ) + ] + support_distributions = nvattest.get("support_distributions") + declaration_failures = validate_support_declarations( + support_distributions, + repair="bash scripts/release.sh --recover", + ) + if declaration_failures: + return declaration_failures + assert isinstance(support_distributions, Sequence) + expected = [ + dict(entry) for entry in support_distributions if isinstance(entry, Mapping) + ] + expected_names = {str(entry["filename"]) for entry in expected} + support_dir = evidence_dir / "support" + actual_paths = {path.name: path for path in support_dir.iterdir()} + if set(actual_paths) != expected_names: + failures.append( + _failure( + "retained nvattest support inventory disagrees with ledger", + expected=", ".join(sorted(expected_names)), + actual=", ".join(sorted(actual_paths)) or "", + repair="bash scripts/release.sh --recover", + ) + ) + support_paths = tuple( + actual_paths[name].resolve() + for name in sorted(set(actual_paths) & expected_names) + ) + if support_paths: + try: + observed = support_distribution_entries(support_paths) + except NvattestProofError as exc: + failures.extend(_driver_failures_from_nvattest_error(exc)) + observed = [] + if observed != expected: + failures.append( + _failure( + "retained nvattest support bytes disagree with ledger", + expected=repr(expected), + actual=repr(observed), + repair="bash scripts/release.sh --recover", + ) + ) + return failures + + +def _retained_authority_binding( + *, + release_dir: Path, + ledger: Mapping[str, Any], +) -> tuple[bytes | None, list[Failure]]: + names, name_failures = _ledger_candidate_file_names(ledger) + if name_failures: + return None, name_failures + authority_bytes, failures = _extract_nvattest_authority_bytes( + release_dir, sorted(names) + ) + nvattest = ledger.get("nvattest") + if not isinstance(nvattest, Mapping): + failures.append( + _failure( + "retained ledger nvattest binding is invalid", + expected="nvattest object", + actual=type(nvattest).__name__, + repair="bash scripts/release.sh --recover", + ) + ) + return authority_bytes, failures + authority = nvattest.get("authority") + authority_sha256 = nvattest.get("authority_sha256") + if isinstance(authority, Mapping) and authority_bytes is not None: + canonical = _canonical_nvattest_authority_bytes(authority) + if canonical != authority_bytes: + failures.append( + _failure( + "retained nvattest authority disagrees with candidate wheels", + expected="ledger authority canonical bytes match retained root wheels", + actual="authority bytes differ", + repair="bash scripts/release.sh --recover", + ) + ) + if hashlib.sha256(authority_bytes).hexdigest() != authority_sha256: + failures.append( + _failure( + "retained nvattest authority digest disagrees with candidate wheels", + expected=str(authority_sha256), + actual=hashlib.sha256(authority_bytes).hexdigest(), + repair="bash scripts/release.sh --recover", + ) + ) + return authority_bytes, failures + + def _ledger_candidate_file_names( ledger: Mapping[str, Any], ) -> tuple[frozenset[str], list[Failure]]: @@ -2456,7 +2905,7 @@ def _validate_evidence_inventory( return [ _failure( "release evidence directory is missing", - expected="ledger.json and proofs directory", + expected="ledger.json, proofs, nvattest, and support", actual="missing", repair="bash scripts/release.sh --recover", ) @@ -2471,9 +2920,9 @@ def _validate_evidence_inventory( ) ] entries = {path.name: path for path in evidence_dir.iterdir()} - required_entries = {"ledger.json", "proofs"} + required_entries = {"ledger.json", "proofs", "nvattest", "support"} accepted_entries = set(required_entries) - expected_inventory = "ledger.json, proofs" + expected_inventory = "ledger.json, nvattest, proofs, support" if publication_prerequisite_version is not None: accepted_entries.add(CORE_UNSUPPORTED_TOMBSTONE_RECORD) expected_inventory = ( @@ -2502,41 +2951,125 @@ def _validate_evidence_inventory( ) ) proofs_dir = entries.get("proofs") - if proofs_dir is None: - return failures - proofs_entry = proofs_dir.lstat() - if stat.S_ISLNK(proofs_entry.st_mode) or not stat.S_ISDIR(proofs_entry.st_mode): - failures.append( - _failure( - "release proofs entry is not an owned directory", - expected="non-symlink proofs directory", - actual="non-directory", - repair="bash scripts/release.sh --recover", + if proofs_dir is not None: + proofs_entry = proofs_dir.lstat() + if stat.S_ISLNK(proofs_entry.st_mode) or not stat.S_ISDIR(proofs_entry.st_mode): + failures.append( + _failure( + "release proofs entry is not an owned directory", + expected="non-symlink proofs directory", + actual="non-directory", + repair="bash scripts/release.sh --recover", + ) ) - ) - return failures - proof_entries = {path.name: path for path in proofs_dir.iterdir()} - expected_proofs = {f"{target}.json" for target in PROOF_TARGETS} - if set(proof_entries) != expected_proofs: - failures.append( - _failure( - "release proof inventory is not exact", - expected=", ".join(sorted(expected_proofs)), - actual=", ".join(sorted(proof_entries)) or "", - repair="bash scripts/release.sh --recover", + else: + proof_entries = {path.name: path for path in proofs_dir.iterdir()} + expected_proofs = {f"{target}.json" for target in PROOF_TARGETS} + if set(proof_entries) != expected_proofs: + failures.append( + _failure( + "release proof inventory is not exact", + expected=", ".join(sorted(expected_proofs)), + actual=", ".join(sorted(proof_entries)) or "", + repair="bash scripts/release.sh --recover", + ) + ) + for proof_name, proof_path in sorted(proof_entries.items()): + proof_entry = proof_path.lstat() + if not stat.S_ISREG(proof_entry.st_mode): + failures.append( + _failure( + "release proof is not a regular file", + expected=f"regular proof file {proof_name}", + actual="non-regular", + repair="bash scripts/release.sh --recover", + ) + ) + nvattest_dir = entries.get("nvattest") + if nvattest_dir is not None: + nvattest_entry = nvattest_dir.lstat() + if stat.S_ISLNK(nvattest_entry.st_mode) or not stat.S_ISDIR( + nvattest_entry.st_mode + ): + failures.append( + _failure( + "release nvattest entry is not an owned directory", + expected="non-symlink nvattest directory", + actual="non-directory", + repair="bash scripts/release.sh --recover", + ) ) - ) - for proof_name, proof_path in sorted(proof_entries.items()): - proof_entry = proof_path.lstat() - if not stat.S_ISREG(proof_entry.st_mode): + else: + nvattest_entries = {path.name: path for path in nvattest_dir.iterdir()} + expected_nvattest = {f"{target}.json" for target in PROOF_TARGETS} + if set(nvattest_entries) != expected_nvattest: + failures.append( + _failure( + "release nvattest inventory is not exact", + expected=", ".join(sorted(expected_nvattest)), + actual=", ".join(sorted(nvattest_entries)) or "", + repair="bash scripts/release.sh --recover", + ) + ) + for receipt_name, receipt_path in sorted(nvattest_entries.items()): + receipt_entry = receipt_path.lstat() + if not stat.S_ISREG(receipt_entry.st_mode): + failures.append( + _failure( + "release nvattest receipt is not a regular file", + expected=f"regular nvattest receipt {receipt_name}", + actual="non-regular", + repair="bash scripts/release.sh --recover", + ) + ) + support_dir = entries.get("support") + if support_dir is not None: + support_entry = support_dir.lstat() + if stat.S_ISLNK(support_entry.st_mode) or not stat.S_ISDIR( + support_entry.st_mode + ): failures.append( _failure( - "release proof is not a regular file", - expected=f"regular proof file {proof_name}", - actual="non-regular", + "release support entry is not an owned directory", + expected="non-symlink support directory", + actual="non-directory", repair="bash scripts/release.sh --recover", ) ) + else: + support_entries = {path.name: path for path in support_dir.iterdir()} + if len(support_entries) != len(SUPPORT_DISTRIBUTION_NAMES): + failures.append( + _failure( + "release support inventory is not structurally exact", + expected=( + f"{len(SUPPORT_DISTRIBUTION_NAMES)} retained support wheels" + ), + actual=str(len(support_entries)), + repair="bash scripts/release.sh --recover", + ) + ) + for wheel_name, wheel_path in sorted(support_entries.items()): + wheel_entry = wheel_path.lstat() + if ( + stat.S_ISLNK(wheel_entry.st_mode) + or not stat.S_ISREG(wheel_entry.st_mode) + or not wheel_name.endswith(".whl") + ): + actual = ( + "non-.whl" + if stat.S_ISREG(wheel_entry.st_mode) + and not wheel_name.endswith(".whl") + else "non-file" + ) + failures.append( + _failure( + "release support wheel is not a regular wheel file", + expected=f"regular .whl support file {wheel_name}", + actual=actual, + repair="bash scripts/release.sh --recover", + ) + ) prerequisite_path = entries.get(CORE_UNSUPPORTED_TOMBSTONE_RECORD) if publication_prerequisite_version is not None and prerequisite_path is not None: failures.extend( @@ -2871,8 +3404,16 @@ def _validate_deep_ledger_binding( ) failures.extend(native_member_failures) failures.extend(_validate_native_summary(release_dir, ledger)) + authority_bytes, authority_failures = _retained_authority_binding( + release_dir=release_dir, + ledger=ledger, + ) + failures.extend(authority_failures) + failures.extend( + _retained_support_binding_failures(evidence_dir=evidence_dir, ledger=ledger) + ) + _ = authority_bytes failures.extend(validate_public_evidence_tree("ledger", ledger)) - _ = evidence_dir return failures @@ -3018,6 +3559,89 @@ def _proof_hashes( return hashes +def _nvattest_hashes( + nvattest_dir: Path, + *, + ledger: Mapping[str, Any], + digest: str, + ledger_sha256: str, + release_dir: Path, + version: str, +) -> dict[str, str]: + authority_bytes, authority_failures = _retained_authority_binding( + release_dir=release_dir, + ledger=ledger, + ) + if authority_failures or authority_bytes is None: + raise DriverError(authority_failures) + nvattest = ledger.get("nvattest") + if not isinstance(nvattest, Mapping): + raise DriverError( + [ + _failure( + "retained ledger nvattest binding is invalid", + expected="nvattest object", + actual=type(nvattest).__name__, + repair="bash scripts/release.sh --recover", + ) + ] + ) + support_distributions = nvattest.get("support_distributions") + support_failures = validate_support_declarations( + support_distributions, + repair="bash scripts/release.sh --recover", + ) + if support_failures: + raise DriverError(support_failures) + hashes: dict[str, str] = {} + for target in PROOF_TARGETS: + path = nvattest_dir / f"{target}.json" + if not path.is_file() or path.is_symlink(): + raise DriverError( + [ + _failure( + "release nvattest receipt is missing", + expected=f"{target} nvattest receipt", + actual="missing", + repair="bash scripts/release.sh --recover", + ) + ] + ) + data = path.read_bytes() + failures = validate_nvattest_proof_bytes( + data, + expected_challenge=str(nvattest.get("challenge")), + target=target, + version=version, + source_commit=str(ledger.get("source_commit")), + core_lock_sha256=str(ledger.get("core_lock_sha256")), + candidate_digest=digest, + ledger_sha256=ledger_sha256, + canonical_authority_bytes=authority_bytes, + expected_support_distributions=support_distributions, # type: ignore[arg-type] + ) + if failures: + raise DriverError(failures) + hashes[target] = file_sha256_size(path)[0] + extras = sorted( + path.name + for path in nvattest_dir.glob("*.json") + if path.stem not in PROOF_TARGETS + ) + if extras: + raise DriverError( + [ + _failure( + "release nvattest set has extra targets", + expected=", ".join(PROOF_TARGETS), + actual=", ".join(extras), + repair="bash scripts/release.sh --recover", + ) + ] + ) + return hashes + + def _report( *, heading: str, @@ -3096,6 +3720,14 @@ def _report( release_dir=release_dir, version=version, ) + nvattest_hashes = _nvattest_hashes( + evidence_dir / "nvattest", + ledger=ledger, + digest=digest, + ledger_sha256=ledger_sha256, + release_dir=release_dir, + version=version, + ) return CandidateReport( heading=heading, version=version, @@ -3105,7 +3737,13 @@ def _report( candidate_digest=digest, ledger_sha256=ledger_sha256, proof_sha256=proof_hashes, - bundle_digest=bundle_digest(digest, ledger_sha256, proof_hashes), + nvattest_sha256=nvattest_hashes, + bundle_digest=bundle_digest( + digest, + ledger_sha256, + proof_hashes, + nvattest_hashes, + ), ) @@ -3131,7 +3769,17 @@ def _evidence_report_inventory(evidence_dir: Path) -> list[dict[str, Any]]: _inventory_entry(path, name=f"proofs/{path.name}") for path in sorted(proofs_dir.iterdir(), key=lambda item: item.name) ) - return entries + nvattest_dir = evidence_dir / "nvattest" + entries.extend( + _inventory_entry(path, name=f"nvattest/{path.name}") + for path in sorted(nvattest_dir.iterdir(), key=lambda item: item.name) + ) + support_dir = evidence_dir / "support" + entries.extend( + _inventory_entry(path, name=f"support/{path.name}") + for path in sorted(support_dir.iterdir(), key=lambda item: item.name) + ) + return sorted(entries, key=lambda item: item["name"]) def _proof_report_inventory(evidence_dir: Path) -> dict[str, dict[str, Any]]: @@ -3142,6 +3790,22 @@ def _proof_report_inventory(evidence_dir: Path) -> dict[str, dict[str, Any]]: } +def _nvattest_report_inventory(evidence_dir: Path) -> dict[str, dict[str, Any]]: + nvattest_dir = evidence_dir / "nvattest" + return { + target: _inventory_entry(nvattest_dir / f"{target}.json", name=f"{target}.json") + for target in sorted(PROOF_TARGETS) + } + + +def _support_report_inventory(evidence_dir: Path) -> list[dict[str, Any]]: + support_dir = evidence_dir / "support" + return [ + _inventory_entry(path, name=path.name) + for path in sorted(support_dir.iterdir(), key=lambda item: item.name) + ] + + def _publication_prerequisite_report_inventory( evidence_dir: Path, ) -> list[dict[str, Any]]: @@ -3169,10 +3833,16 @@ def format_report(report: CandidateReport) -> str: "payload_inventory": _payload_report_inventory(report.release_dir), "evidence_inventory": _evidence_report_inventory(report.evidence_dir), "proof_inventory": _proof_report_inventory(report.evidence_dir), + "nvattest_inventory": _nvattest_report_inventory(report.evidence_dir), + "support_inventory": _support_report_inventory(report.evidence_dir), "proof_sha256": { target: report.proof_sha256[target] for target in sorted(report.proof_sha256) }, + "nvattest_sha256": { + target: report.nvattest_sha256[target] + for target in sorted(report.nvattest_sha256) + }, "publication_prerequisite_inventory": ( _publication_prerequisite_report_inventory(report.evidence_dir) ), @@ -3380,6 +4050,26 @@ def run_candidate( source_commit=expected_commit, core_lock_sha256=expected_lock, ) + support_distributions, support_paths = _materialize_verified_support_wheels( + root=root, + support_dir=evidence_staging / "support", + services=svc, + ) + challenge = svc.challenge_factory() + challenge_failures = _validate_nvattest_challenge(challenge) + if challenge_failures: + raise DriverError(challenge_failures) + authority_bytes, authority_failures = _extract_nvattest_authority_bytes( + release_dir, + _candidate_names_from_dir(release_dir), + ) + if authority_failures or authority_bytes is None: + raise DriverError(authority_failures) + nvattest_payload = _nvattest_ledger_payload( + challenge=challenge, + authority_bytes=authority_bytes, + support_distributions=support_distributions, + ) ledger_path = write_ledger( evidence_root=evidence_root, version=version, @@ -3395,6 +4085,7 @@ def run_candidate( "decision": models_decision, "package_version": models_package_version, }, + nvattest=nvattest_payload, output_dir=evidence_staging, ) ledger_sha256 = file_sha256_size(ledger_path)[0] @@ -3404,7 +4095,7 @@ def run_candidate( ) proofs_dir = evidence_staging / "proofs" for target in PROOF_TARGETS: - proof_paths[target] = svc.run_install_proof( + target_proofs = svc.run_target_proofs( target=target, version=version, source_commit=expected_commit, @@ -3414,8 +4105,15 @@ def run_candidate( candidate_dir=release_dir, candidate_paths=candidate_paths, ledger_payload=ledger_payload, + challenge=challenge, + support_wheel_paths=support_paths, + canonical_authority_bytes=authority_bytes, output_path=proofs_dir / f"{target}.json", + nvattest_output_path=( + evidence_staging / "nvattest" / f"{target}.json" + ), ) + proof_paths[target] = target_proofs.install failures = validate_public_evidence_tree("ledger", ledger_payload) if failures: raise DriverError(failures) diff --git a/scripts/release_digest.py b/scripts/release_digest.py index a34caea41..2fee557ed 100644 --- a/scripts/release_digest.py +++ b/scripts/release_digest.py @@ -42,10 +42,15 @@ def bundle_digest( candidate_digest: str, ledger_sha256: str, proof_sha256_by_target: Mapping[str, str], + nvattest_sha256_by_target: Mapping[str, str], ) -> str: payload = { "candidate_digest": candidate_digest, "ledger_sha256": ledger_sha256, + "nvattest_sha256": { + target: nvattest_sha256_by_target[target] + for target in sorted(nvattest_sha256_by_target) + }, "proof_sha256": { target: proof_sha256_by_target[target] for target in sorted(proof_sha256_by_target) diff --git a/scripts/release_ledger.py b/scripts/release_ledger.py index 2e1e63f03..bc126330a 100644 --- a/scripts/release_ledger.py +++ b/scripts/release_ledger.py @@ -30,6 +30,8 @@ from scripts.check_wheel_contents import ( from scripts.release_advisory_policy import PolicyRun, validate_snapshot_identity from scripts.release_digest import candidate_digest, file_sha256_size from scripts.release_install_smoke import CANDIDATE, PROOF_TARGETS +from scripts.release_nvattest_proof import CHALLENGE_RE +from scripts.release_nvattest_support import validate_support_declarations from scripts.release_public_evidence import validate_public_evidence_tree TOP_LEVEL_KEYS = frozenset( @@ -49,10 +51,14 @@ TOP_LEVEL_KEYS = frozenset( "policy_run", "native_summary", "proofs", + "nvattest", "redaction", ) ) MODELS_KEYS = frozenset(("decision", "package_version")) +NVATTEST_KEYS = frozenset( + ("challenge", "authority_sha256", "authority", "support_distributions") +) POLICY_RUN_KEYS = frozenset( ( "advisory_source_id", @@ -328,6 +334,85 @@ def validate_models_payload(value: Any, candidate: Any | None = None) -> list[Fa return failures +def _canonical_nvattest_authority_bytes(payload: Mapping[str, Any]) -> bytes: + return (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode("utf-8") + + +def validate_nvattest_payload(value: Any) -> list[Failure]: + failures: list[Failure] = [] + if not isinstance(value, Mapping): + return [ + _failure( + "retained ledger nvattest binding is invalid", + expected="nvattest object", + actual=type(value).__name__, + repair="python3 scripts/check_rust_release_manifest.py", + ) + ] + if set(value) != NVATTEST_KEYS: + failures.append( + _failure( + "retained ledger nvattest key set is invalid", + expected=", ".join(sorted(NVATTEST_KEYS)), + actual=", ".join(sorted(str(key) for key in value)) or "", + repair="python3 scripts/check_rust_release_manifest.py", + ) + ) + challenge = value.get("challenge") + if not isinstance(challenge, str) or not CHALLENGE_RE.fullmatch(challenge): + failures.append( + _failure( + "retained ledger nvattest challenge is invalid", + expected="64 lowercase hexadecimal characters", + actual=repr(challenge), + repair="python3 scripts/check_rust_release_manifest.py", + ) + ) + authority_sha256 = value.get("authority_sha256") + if not isinstance(authority_sha256, str) or not SHA256_RE.fullmatch( + authority_sha256 + ): + failures.append( + _failure( + "retained ledger nvattest authority_sha256 is invalid", + expected="lowercase SHA-256", + actual=repr(authority_sha256), + repair="python3 scripts/check_rust_release_manifest.py", + ) + ) + authority = value.get("authority") + if not isinstance(authority, Mapping): + failures.append( + _failure( + "retained ledger nvattest authority is invalid", + expected="parsed authority JSON object", + actual=type(authority).__name__, + repair="python3 scripts/check_rust_release_manifest.py", + ) + ) + elif isinstance(authority_sha256, str) and SHA256_RE.fullmatch(authority_sha256): + digest = hashlib.sha256( + _canonical_nvattest_authority_bytes(authority) + ).hexdigest() + if digest != authority_sha256: + failures.append( + _failure( + "retained ledger nvattest authority digest is invalid", + expected=authority_sha256, + actual=digest, + repair="python3 scripts/check_rust_release_manifest.py", + ) + ) + failures.extend( + validate_support_declarations( + value.get("support_distributions"), + repair="python3 scripts/check_rust_release_manifest.py", + ) + ) + failures.extend(validate_public_evidence_tree("ledger.nvattest", value)) + return failures + + def validate_retained_ledger(payload: Mapping[str, Any]) -> list[Failure]: failures: list[Failure] = [] if set(payload) != TOP_LEVEL_KEYS: @@ -382,6 +467,7 @@ def validate_retained_ledger(payload: Mapping[str, Any]) -> list[Failure]: failures.extend( validate_models_payload(payload.get("models"), payload.get("candidate")) ) + failures.extend(validate_nvattest_payload(payload.get("nvattest"))) failures.extend(validate_native_members(payload.get("native_members"))) failures.extend(validate_public_evidence_tree("ledger", payload)) return failures @@ -780,6 +866,7 @@ def build_ledger( policy_run: PolicyRun, native_records: Sequence[Mapping[str, Any]], models: Mapping[str, str], + nvattest: Mapping[str, Any], ) -> dict[str, Any]: failures: list[Failure] = [] if not SOURCE_COMMIT_RE.fullmatch(source_commit): @@ -818,6 +905,7 @@ def build_ledger( failures.extend(exc.failures) native_members = {} failures.extend(validate_models_payload(models, candidate)) + failures.extend(validate_nvattest_payload(nvattest)) failures.extend(_validate_tool_evidence(tool_evidence)) if failures: raise LedgerError(failures) @@ -840,6 +928,14 @@ def build_ledger( "policy_run": _policy_run_payload(policy_run), "native_summary": native_summary, "proofs": {"expected_targets": list(PROOF_TARGETS)}, + "nvattest": { + "authority": nvattest["authority"], + "authority_sha256": nvattest["authority_sha256"], + "challenge": nvattest["challenge"], + "support_distributions": [ + dict(entry) for entry in nvattest["support_distributions"] + ], + }, "redaction": {"validator": "recursive-key-value-public-evidence"}, } if set(ledger) != TOP_LEVEL_KEYS: @@ -861,6 +957,7 @@ def write_ledger( policy_run: PolicyRun, native_records: Sequence[Mapping[str, Any]], models: Mapping[str, str], + nvattest: Mapping[str, Any], output_dir: Path | None = None, ) -> Path: ledger = build_ledger( @@ -872,6 +969,7 @@ def write_ledger( policy_run=policy_run, native_records=native_records, models=models, + nvattest=nvattest, ) resolved_output_dir = output_dir or (evidence_root / version) resolved_output_dir.mkdir(parents=True, exist_ok=True) diff --git a/scripts/release_nvattest_support.py b/scripts/release_nvattest_support.py new file mode 100644 index 000000000..157777b86 --- /dev/null +++ b/scripts/release_nvattest_support.py @@ -0,0 +1,447 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Locked support-wheel helpers for nvattest release evidence.""" + +from __future__ import annotations + +import re +import tomllib +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Any +from urllib.parse import unquote, urlparse + +from scripts.check_rust_release_manifest import SHA256_RE, Failure +from scripts.release_nvattest_proof import ( + SUPPORT_DISTRIBUTION_NAMES, + NvattestProofError, + support_distribution_entries, +) + +REPAIR = "bash scripts/release.sh --candidate" +SUPPORT_ENTRY_KEYS = frozenset(("bytes", "filename", "name", "sha256", "version")) + + +@dataclass(frozen=True) +class SupportLockEntry: + name: str + version: str + filename: str + bytes: int + sha256: str + url: str + + def declaration(self) -> dict[str, object]: + return { + "bytes": self.bytes, + "filename": self.filename, + "name": self.name, + "sha256": self.sha256, + "version": self.version, + } + + +class SupportLockError(RuntimeError): + def __init__(self, failures: Sequence[Failure]) -> None: + self.failures = tuple(failures) + super().__init__("; ".join(failure.error for failure in self.failures)) + + +def _failure( + error: str, + *, + expected: str, + actual: str, + repair: str = REPAIR, +) -> Failure: + return Failure(error=error, expected=expected, actual=actual, repair=repair) + + +def _normalize_distribution_name(value: str) -> str: + return re.sub(r"[-_.]+", "-", value).lower() + + +def _safe_basename(value: object) -> bool: + return ( + isinstance(value, str) + and bool(value) + and Path(value).name == value + and value not in {".", ".."} + and "/" not in value + and "\\" not in value + ) + + +def _url_basename(url: object) -> str | None: + if not isinstance(url, str) or not url: + return None + parsed = urlparse(url) + basename = Path(unquote(parsed.path)).name + return basename if _safe_basename(basename) else None + + +def _read_lock_payload(lock_path: Path) -> Mapping[str, Any]: + try: + with lock_path.open("rb") as handle: + payload = tomllib.load(handle) + except (OSError, tomllib.TOMLDecodeError) as exc: + raise SupportLockError( + [ + _failure( + "nvattest support lock could not be read", + expected="valid uv.lock TOML", + actual=f"{type(exc).__name__}: {exc}", + ) + ] + ) from None + if not isinstance(payload, Mapping): + raise SupportLockError( + [ + _failure( + "nvattest support lock could not be read", + expected="valid uv.lock TOML object", + actual=type(payload).__name__, + ) + ] + ) + return payload + + +def read_support_lock_entries(lock_path: Path) -> tuple[SupportLockEntry, ...]: + payload = _read_lock_payload(lock_path) + packages = payload.get("package") + if not isinstance(packages, list): + raise SupportLockError( + [ + _failure( + "nvattest support lock package is not exact", + expected="uv.lock package list", + actual=type(packages).__name__, + ) + ] + ) + + failures: list[Failure] = [] + entries: list[SupportLockEntry] = [] + for name in sorted(SUPPORT_DISTRIBUTION_NAMES): + matches = [ + package + for package in packages + if isinstance(package, Mapping) + and _normalize_distribution_name(str(package.get("name", ""))) == name + ] + if len(matches) != 1: + failures.append( + _failure( + "nvattest support lock package is not exact", + expected=f"exactly one [[package]] for {name}", + actual=str(len(matches)), + ) + ) + continue + + package = matches[0] + version = package.get("version") + wheels = package.get("wheels") + if not isinstance(version, str) or not version: + failures.append( + _failure( + "nvattest support lock wheel metadata is invalid", + expected=f"{name} non-empty version", + actual=repr(version), + ) + ) + continue + if not isinstance(wheels, list): + failures.append( + _failure( + "nvattest support lock wheel entry is not exact", + expected=f"exactly one py3-none-any wheel for {name}", + actual=type(wheels).__name__, + ) + ) + continue + + wheel_matches = [ + wheel + for wheel in wheels + if isinstance(wheel, Mapping) + and (_url_basename(wheel.get("url")) or "").endswith("-py3-none-any.whl") + ] + if len(wheel_matches) != 1: + actual = [ + _url_basename(wheel.get("url")) or repr(wheel.get("url")) + for wheel in wheels + if isinstance(wheel, Mapping) + ] + failures.append( + _failure( + "nvattest support lock wheel entry is not exact", + expected=f"exactly one py3-none-any wheel for {name}", + actual=", ".join(sorted(actual)) or "", + ) + ) + continue + + wheel = wheel_matches[0] + filename = _url_basename(wheel.get("url")) + raw_hash = wheel.get("hash") + size = wheel.get("size") + if ( + filename is None + or not filename.endswith(".whl") + or not isinstance(raw_hash, str) + or not raw_hash.startswith("sha256:") + or not SHA256_RE.fullmatch(raw_hash.removeprefix("sha256:")) + or not isinstance(size, int) + or size <= 0 + ): + failures.append( + _failure( + "nvattest support lock wheel metadata is invalid", + expected="url basename, sha256:, positive size, version", + actual=repr(wheel), + ) + ) + continue + entries.append( + SupportLockEntry( + name=name, + version=version, + filename=filename, + bytes=size, + sha256=raw_hash.removeprefix("sha256:"), + url=str(wheel["url"]), + ) + ) + + if failures: + raise SupportLockError(failures) + return tuple(sorted(entries, key=lambda entry: (entry.name, entry.version))) + + +def support_declarations_from_lock( + entries: Sequence[SupportLockEntry], +) -> tuple[dict[str, object], ...]: + declarations = tuple(entry.declaration() for entry in entries) + failures = validate_support_declarations(declarations, repair=REPAIR) + if failures: + raise SupportLockError(failures) + return declarations + + +def validate_support_declarations( + value: Any, + *, + expected: Sequence[Mapping[str, Any]] | None = None, + repair: str, +) -> list[Failure]: + failures: list[Failure] = [] + if not isinstance(value, list | tuple): + return [ + _failure( + "nvattest support declaration is invalid", + expected="canonical support distribution list", + actual=type(value).__name__, + repair=repair, + ) + ] + actual = [dict(entry) for entry in value if isinstance(entry, Mapping)] + if len(actual) != len(value): + failures.append( + _failure( + "nvattest support declaration contains non-object entries", + expected="support distribution objects", + actual=repr(value), + repair=repair, + ) + ) + names: list[str] = [] + for entry in actual: + if set(entry) != SUPPORT_ENTRY_KEYS: + failures.append( + _failure( + "nvattest support distribution entry is invalid", + expected=", ".join(sorted(SUPPORT_ENTRY_KEYS)), + actual=", ".join(sorted(str(key) for key in entry)) or "", + repair=repair, + ) + ) + continue + name = entry["name"] + names.append(str(name)) + if ( + not isinstance(name, str) + or _normalize_distribution_name(name) != name + or name not in SUPPORT_DISTRIBUTION_NAMES + ): + failures.append( + _failure( + "nvattest support distribution name is invalid", + expected=", ".join(sorted(SUPPORT_DISTRIBUTION_NAMES)), + actual=repr(name), + repair=repair, + ) + ) + if not _safe_basename(entry["filename"]) or not str(entry["filename"]).endswith( + ".whl" + ): + failures.append( + _failure( + "nvattest support wheel filename is invalid", + expected="safe wheel basename", + actual=repr(entry["filename"]), + repair=repair, + ) + ) + if not isinstance(entry["version"], str) or not entry["version"]: + failures.append( + _failure( + "nvattest support version is invalid", + expected="non-empty version string", + actual=repr(entry["version"]), + repair=repair, + ) + ) + if not isinstance(entry["bytes"], int) or entry["bytes"] <= 0: + failures.append( + _failure( + "nvattest support wheel byte count is invalid", + expected="positive integer", + actual=repr(entry["bytes"]), + repair=repair, + ) + ) + if not isinstance(entry["sha256"], str) or not SHA256_RE.fullmatch( + entry["sha256"] + ): + failures.append( + _failure( + "nvattest support wheel sha256 is invalid", + expected="lowercase SHA-256", + actual=repr(entry["sha256"]), + repair=repair, + ) + ) + if len(names) != len(set(names)): + failures.append( + _failure( + "nvattest support distribution is duplicated", + expected="unique support distribution names", + actual=", ".join(names), + repair=repair, + ) + ) + if set(names) != SUPPORT_DISTRIBUTION_NAMES: + failures.append( + _failure( + "nvattest support distribution set is not exact", + expected=", ".join(sorted(SUPPORT_DISTRIBUTION_NAMES)), + actual=", ".join(sorted(set(names))) or "", + repair=repair, + ) + ) + canonical = sorted(actual, key=lambda item: (item["name"], item["version"])) + if actual != canonical: + failures.append( + _failure( + "nvattest support declaration is not canonical", + expected="support distributions sorted by normalized name and version", + actual=repr(actual), + repair=repair, + ) + ) + if expected is not None and actual != [dict(entry) for entry in expected]: + failures.append( + _failure( + "nvattest support declaration is not bound to expected wheels", + expected=repr([dict(entry) for entry in expected]), + actual=repr(actual), + repair=repair, + ) + ) + return failures + + +def verify_support_wheels_against_lock( + paths: Sequence[Path], + entries: Sequence[SupportLockEntry], +) -> tuple[dict[str, object], ...]: + expected = support_declarations_from_lock(entries) + failures: list[Failure] = [] + for raw_path in paths: + path = Path(raw_path) + if ( + not path.is_absolute() + or path.is_symlink() + or not path.is_file() + or path.suffix != ".whl" + ): + failures.append( + _failure( + "nvattest support wheel path is invalid", + expected="absolute regular local wheel file path", + actual=str(raw_path), + ) + ) + if failures: + raise SupportLockError(failures) + try: + observed = tuple(support_distribution_entries(paths)) + except NvattestProofError as exc: + raise SupportLockError( + [ + _failure( + failure.error, + expected=failure.expected, + actual=failure.actual, + repair=REPAIR, + ) + for failure in exc.failures + ] + ) from None + by_filename = {str(entry["filename"]): entry for entry in observed} + expected_by_filename = {entry.filename: entry for entry in entries} + if set(by_filename) != set(expected_by_filename): + failures.append( + _failure( + "nvattest support inventory is not exact", + expected=", ".join(sorted(expected_by_filename)), + actual=", ".join(sorted(by_filename)) or "", + ) + ) + for filename in sorted(set(by_filename) & set(expected_by_filename)): + lock_entry = expected_by_filename[filename] + observed_entry = by_filename[filename] + if ( + observed_entry.get("sha256") != lock_entry.sha256 + or observed_entry.get("bytes") != lock_entry.bytes + ): + failures.append( + _failure( + "nvattest support wheel does not match uv.lock", + expected=f"{filename} {lock_entry.sha256}/{lock_entry.bytes}", + actual=( + f"{observed_entry.get('sha256')}/{observed_entry.get('bytes')}" + ), + ) + ) + if ( + observed_entry.get("name") != lock_entry.name + or observed_entry.get("version") != lock_entry.version + ): + failures.append( + _failure( + "nvattest support wheel METADATA does not match uv.lock", + expected=f"{lock_entry.name}=={lock_entry.version}", + actual=( + f"{observed_entry.get('name')}=={observed_entry.get('version')}" + ), + ) + ) + if failures: + raise SupportLockError(failures) + return expected diff --git a/scripts/release_proof_host.py b/scripts/release_proof_host.py index d85bf5311..9307d7477 100644 --- a/scripts/release_proof_host.py +++ b/scripts/release_proof_host.py @@ -6,6 +6,7 @@ from __future__ import annotations +import hashlib import json import os import shlex @@ -47,6 +48,7 @@ TARGET_ENV_KEYS: Mapping[str, str] = { "linux-aarch64-musl": "RELEASE_PROOF_HOST_LINUX_AARCH64_MUSL_CHANNEL", "macos-arm64": "RELEASE_PROOF_HOST_MACOS_ARM64_CHANNEL", } +# `authority` is a descriptor (sha256 + bytes); `authority_file` is the staged relative path. REQUEST_KEYS = frozenset( ( "schema_version", @@ -58,18 +60,46 @@ REQUEST_KEYS = frozenset( "ledger_sha256", "core_lock_sha256", "candidate_files", + "support_files", + "authority", + "challenge", "paths", "expected_host", ) ) REQUEST_FILE_KEYS = frozenset(("basename", "bytes", "sha256", "path")) -REQUEST_PATH_KEYS = frozenset(("candidate_dir", "ledger", "response", "proof")) +REQUEST_AUTHORITY_KEYS = frozenset(("sha256", "bytes")) +REQUEST_PATH_KEYS = frozenset( + ( + "candidate_dir", + "support_dir", + "authority_file", + "ledger", + "response", + "install_proof", + "nvattest_proof", + ) +) REQUEST_HOST_KEYS = frozenset(("os", "arch")) -RESPONSE_KEYS = frozenset(("schema_version", "cohort_id", "attestation", "proof")) +RESPONSE_KEYS = frozenset( + ( + "schema_version", + "cohort_id", + "attestation", + "install_proof", + "nvattest_proof", + ) +) ATTESTATION_KEYS = frozenset(("os", "arch", "candidate_digest", "ledger_sha256")) PROOF_FILE_KEYS = frozenset(("path", "sha256", "bytes")) +@dataclass(frozen=True) +class TargetProofPaths: + install: Path + nvattest: Path + + @dataclass(frozen=True) class DirectoryIdentity: path: Path @@ -80,7 +110,7 @@ class DirectoryIdentity: class ProofHostChannel(Protocol): - def run_install_proof( + def run_target_proofs( self, *, target: str, @@ -92,8 +122,12 @@ class ProofHostChannel(Protocol): candidate_dir: Path, candidate_paths: Sequence[Path], ledger_payload: Mapping[str, Any], + challenge: str, + support_wheel_paths: Sequence[Path], + canonical_authority_bytes: bytes, output_path: Path, - ) -> Path: ... + nvattest_output_path: Path, + ) -> TargetProofPaths: ... class ProofHostError(RuntimeError): @@ -280,6 +314,32 @@ def _validate_regular_file(path: Path, *, label: str) -> None: ) +def _request_file_entries( + paths: Sequence[Path], + *, + directory: str, +) -> list[dict[str, object]]: + entries: list[dict[str, object]] = [] + for path in paths: + sha256, byte_count = file_sha256_size(path) + entries.append( + { + "basename": path.name, + "bytes": byte_count, + "sha256": sha256, + "path": f"{directory}/{path.name}", + } + ) + return entries + + +def _authority_descriptor(data: bytes) -> dict[str, object]: + return { + "bytes": len(data), + "sha256": hashlib.sha256(data).hexdigest(), + } + + def _validate_fresh_directory_path(path: Path, *, label: str) -> None: try: path.lstat() @@ -323,7 +383,10 @@ def _validate_scalars( core_lock_sha256: str, candidate_digest: str, ledger_sha256: str, + challenge: str, ) -> None: + from scripts.release_nvattest_proof import CHALLENGE_RE + failures: list[Failure] = [] if target not in TARGET_POLICY: failures.append( @@ -357,6 +420,15 @@ def _validate_scalars( repair="bash scripts/release.sh --candidate", ) ) + if not CHALLENGE_RE.fullmatch(challenge): + failures.append( + _failure( + "proof-host nvattest challenge is invalid", + expected="64 lowercase hexadecimal characters", + actual=challenge, + repair="bash scripts/release.sh --candidate", + ) + ) if failures: raise ProofHostError(failures) @@ -532,6 +604,61 @@ class ExternalProofHostChannel: ) return ProofHostError(failures) if failures else None + def _copy_verified_files( + self, + paths: Sequence[Path], + *, + destination_dir: Path, + request_directory: str, + label: str, + identities: Sequence[DirectoryIdentity | None], + ) -> list[dict[str, object]]: + seen: set[str] = set() + for path in paths: + if _safe_basename(path.name) is None: + raise ProofHostError( + [ + _failure( + f"proof-host {label} filename is unsafe", + expected=f"safe {label} basename", + actual=path.name, + repair="bash scripts/release.sh --candidate", + ) + ] + ) + if path.name in seen: + raise ProofHostError( + [ + _failure( + f"proof-host {label} filename is duplicated", + expected=f"unique {label} basenames", + actual=path.name, + repair="bash scripts/release.sh --candidate", + ) + ] + ) + seen.add(path.name) + _validate_regular_file(path, label=label) + source_sha256, source_bytes = file_sha256_size(path) + _validate_directory_identities(identities) + self._file_copier(path, destination_dir / path.name) + _validate_directory_identities(identities) + _validate_regular_file(destination_dir / path.name, label=f"request {label}") + copied_sha256, copied_bytes = file_sha256_size(destination_dir / path.name) + _validate_directory_identities(identities) + if copied_sha256 != source_sha256 or copied_bytes != source_bytes: + raise ProofHostError( + [ + _failure( + f"proof-host copied {label} changed bytes", + expected=f"{source_sha256}/{source_bytes}", + actual=f"{copied_sha256}/{copied_bytes}", + repair="bash scripts/release.sh --candidate", + ) + ] + ) + return _request_file_entries(paths, directory=request_directory) + def _request_payload( self, *, @@ -543,6 +670,9 @@ class ExternalProofHostChannel: candidate_digest: str, ledger_sha256: str, install_paths: Sequence[Path], + challenge: str, + support_wheel_paths: Sequence[Path], + canonical_authority_bytes: bytes, ) -> dict[str, object]: os_name, arch = TARGET_POLICY[target] files = [ @@ -552,6 +682,8 @@ class ExternalProofHostChannel: } for entry in candidate_file_entries(install_paths) ] + support_files = _request_file_entries(support_wheel_paths, directory="support") + authority = _authority_descriptor(canonical_authority_bytes) payload: dict[str, object] = { "schema_version": 1, "cohort_id": cohort_id, @@ -562,11 +694,17 @@ class ExternalProofHostChannel: "ledger_sha256": ledger_sha256, "core_lock_sha256": core_lock_sha256, "candidate_files": files, + "support_files": support_files, + "authority": authority, + "challenge": challenge, "paths": { "candidate_dir": "candidate", + "support_dir": "support", + "authority_file": "authority/nvattest_authority_v1.json", "ledger": "ledger.json", "response": "response.json", - "proof": "output/proof.json", + "install_proof": "output/install-proof.json", + "nvattest_proof": "output/nvattest-proof.json", }, "expected_host": {"os": os_name, "arch": arch}, } @@ -575,6 +713,11 @@ class ExternalProofHostChannel: for entry in files: if set(entry) != REQUEST_FILE_KEYS: raise AssertionError("proof-host request file key set drifted") + for entry in support_files: + if set(entry) != REQUEST_FILE_KEYS: + raise AssertionError("proof-host request support file key set drifted") + if set(authority) != REQUEST_AUTHORITY_KEYS: + raise AssertionError("proof-host request authority key set drifted") if set(payload["paths"]) != REQUEST_PATH_KEYS: # type: ignore[arg-type] raise AssertionError("proof-host request path key set drifted") if set(payload["expected_host"]) != REQUEST_HOST_KEYS: # type: ignore[arg-type] @@ -653,28 +796,31 @@ class ExternalProofHostChannel: ] if failures: raise ProofHostError(failures) - proof = payload.get("proof") - if not isinstance(proof, Mapping): - raise ProofHostError( - [ - _failure( - "proof-host response proof descriptor is invalid", - expected="proof file descriptor", - actual=type(proof).__name__, - repair="bash scripts/release.sh --candidate", - ) - ] + descriptors: dict[str, Mapping[str, object]] = {} + for key in ("install_proof", "nvattest_proof"): + descriptor = payload.get(key) + if not isinstance(descriptor, Mapping): + raise ProofHostError( + [ + _failure( + f"proof-host response {key} descriptor is invalid", + expected="proof file descriptor", + actual=type(descriptor).__name__, + repair="bash scripts/release.sh --candidate", + ) + ] + ) + failures = _key_set_failures( + f"proof-host response {key} descriptor", + descriptor, + PROOF_FILE_KEYS, ) - failures = _key_set_failures( - "proof-host response proof descriptor", - proof, - PROOF_FILE_KEYS, - ) - if failures: - raise ProofHostError(failures) - return proof + if failures: + raise ProofHostError(failures) + descriptors[key] = descriptor + return descriptors - def run_install_proof( + def run_target_proofs( self, *, target: str, @@ -686,8 +832,12 @@ class ExternalProofHostChannel: candidate_dir: Path, candidate_paths: Sequence[Path], ledger_payload: Mapping[str, Any], + challenge: str, + support_wheel_paths: Sequence[Path], + canonical_authority_bytes: bytes, output_path: Path, - ) -> Path: + nvattest_output_path: Path, + ) -> TargetProofPaths: if target != self._target: raise ProofHostError( [ @@ -706,90 +856,110 @@ class ExternalProofHostChannel: core_lock_sha256=core_lock_sha256, candidate_digest=candidate_digest, ledger_sha256=ledger_sha256, + challenge=challenge, ) install_paths = target_install_paths_from_ledger( ledger_payload, target=target, candidate_dir=candidate_dir, ) + from scripts.release_nvattest_proof import ( + support_distribution_entries, + validate_nvattest_proof_bytes, + ) + + support_distributions = support_distribution_entries(support_wheel_paths) cohort_id = self._cohort_id_factory() _validate_cohort_id(cohort_id) request_dir = output_path.parent / f".{target}.proof-request-{cohort_id}" request_candidate_dir = request_dir / "candidate" + request_support_dir = request_dir / "support" + request_authority_dir = request_dir / "authority" request_output_dir = request_dir / "output" response_path = request_dir / "response.json" request_path = request_dir / "request.json" - proof_path = request_dir / "output" / "proof.json" + authority_path = request_authority_dir / "nvattest_authority_v1.json" + install_proof_path = request_output_dir / "install-proof.json" + nvattest_proof_path = request_output_dir / "nvattest-proof.json" request_identity: DirectoryIdentity | None = None candidate_identity: DirectoryIdentity | None = None + support_identity: DirectoryIdentity | None = None + authority_identity: DirectoryIdentity | None = None output_identity: DirectoryIdentity | None = None remote_started = False primary: ProofHostError | None = None completed = False try: _validate_fresh_directory_path(request_dir, label="request") - if output_path.exists() or output_path.is_symlink(): - raise ProofHostError( - [ - _failure( - "proof-host output proof already exists", - expected="fresh proof output path", - actual=output_path.name, - repair="bash scripts/release.sh --candidate", - ) - ] - ) + for proof_output in (output_path, nvattest_output_path): + if proof_output.exists() or proof_output.is_symlink(): + raise ProofHostError( + [ + _failure( + "proof-host output proof already exists", + expected="fresh proof output path", + actual=proof_output.name, + repair="bash scripts/release.sh --candidate", + ) + ] + ) request_candidate_dir.mkdir(parents=True) + request_support_dir.mkdir() + request_authority_dir.mkdir() request_output_dir.mkdir() request_identity = _capture_directory_identity(request_dir, label="request") candidate_identity = _capture_directory_identity( request_candidate_dir, label="candidate" ) + support_identity = _capture_directory_identity( + request_support_dir, label="support" + ) + authority_identity = _capture_directory_identity( + request_authority_dir, label="authority" + ) output_identity = _capture_directory_identity( request_output_dir, label="output" ) - for path in install_paths: - if _safe_basename(path.name) is None: - raise ProofHostError( - [ - _failure( - "proof-host candidate filename is unsafe", - expected="safe candidate wheel basename", - actual=path.name, - repair="bash scripts/release.sh --candidate", - ) - ] - ) - _validate_regular_file(path, label="candidate wheel") - source_sha256, source_bytes = file_sha256_size(path) - _validate_directory_identities( - (request_identity, candidate_identity, output_identity) - ) - self._file_copier(path, request_candidate_dir / path.name) - _validate_directory_identities( - (request_identity, candidate_identity, output_identity) - ) - _validate_regular_file( - request_candidate_dir / path.name, - label="request candidate wheel", - ) - copied_sha256, copied_bytes = file_sha256_size( - request_candidate_dir / path.name - ) - _validate_directory_identities( - (request_identity, candidate_identity, output_identity) + identities = ( + request_identity, + candidate_identity, + support_identity, + authority_identity, + output_identity, + ) + self._copy_verified_files( + install_paths, + destination_dir=request_candidate_dir, + request_directory="candidate", + label="candidate wheel", + identities=identities, + ) + self._copy_verified_files( + support_wheel_paths, + destination_dir=request_support_dir, + request_directory="support", + label="support wheel", + identities=identities, + ) + _validate_directory_identities(identities) + authority_path.write_bytes(canonical_authority_bytes) + _validate_directory_identities(identities) + _validate_regular_file(authority_path, label="request authority file") + authority_sha256, authority_bytes = file_sha256_size(authority_path) + if _authority_descriptor(canonical_authority_bytes) != { + "sha256": authority_sha256, + "bytes": authority_bytes, + }: + raise ProofHostError( + [ + _failure( + "proof-host staged authority changed bytes", + expected="canonical authority descriptor", + actual=f"{authority_sha256}/{authority_bytes}", + repair="bash scripts/release.sh --candidate", + ) + ] ) - if copied_sha256 != source_sha256 or copied_bytes != source_bytes: - raise ProofHostError( - [ - _failure( - "proof-host copied candidate wheel changed bytes", - expected=f"{source_sha256}/{source_bytes}", - actual=f"{copied_sha256}/{copied_bytes}", - repair="bash scripts/release.sh --candidate", - ) - ] - ) (request_dir / "ledger.json").write_bytes( canonical_json_bytes(ledger_payload) ) @@ -802,6 +972,9 @@ class ExternalProofHostChannel: candidate_digest=candidate_digest, ledger_sha256=ledger_sha256, install_paths=install_paths, + challenge=challenge, + support_wheel_paths=support_wheel_paths, + canonical_authority_bytes=canonical_authority_bytes, ) request_path.write_bytes(canonical_json_bytes(request_payload)) public_failures = validate_public_evidence_tree( @@ -829,66 +1002,99 @@ class ExternalProofHostChannel: ) ] ) - _validate_directory_identities( - (request_identity, candidate_identity, output_identity) - ) + _validate_directory_identities(identities) _validate_regular_file(response_path, label="response") response = _json_object(response_path) - proof_descriptor = self._validate_response( + proof_descriptors = self._validate_response( response, cohort_id=cohort_id, target=target, candidate_digest=candidate_digest, ledger_sha256=ledger_sha256, ) - proof_name = proof_descriptor.get("path") - if proof_name != "output/proof.json": + install_descriptor = proof_descriptors["install_proof"] + nvattest_descriptor = proof_descriptors["nvattest_proof"] + install_proof_name = install_descriptor.get("path") + if install_proof_name != "output/install-proof.json": raise ProofHostError( [ _failure( - "proof-host proof path is invalid", - expected="output/proof.json", - actual=repr(proof_name), + "proof-host install proof path is invalid", + expected="output/install-proof.json", + actual=repr(install_proof_name), repair="bash scripts/release.sh --candidate", ) ] ) - _validate_directory_identities( - (request_identity, candidate_identity, output_identity) - ) - _validate_regular_file(proof_path, label="proof") - proof_sha256, proof_bytes_count = file_sha256_size(proof_path) - if proof_descriptor.get("sha256") != proof_sha256: + nvattest_proof_name = nvattest_descriptor.get("path") + if nvattest_proof_name != "output/nvattest-proof.json": raise ProofHostError( [ _failure( - "proof-host proof SHA-256 is wrong", - expected=proof_sha256, - actual=repr(proof_descriptor.get("sha256")), + "proof-host nvattest proof path is invalid", + expected="output/nvattest-proof.json", + actual=repr(nvattest_proof_name), repair="bash scripts/release.sh --candidate", ) ] ) - if proof_descriptor.get("bytes") != proof_bytes_count: + _validate_directory_identities(identities) + _validate_regular_file(install_proof_path, label="install proof") + _validate_regular_file(nvattest_proof_path, label="nvattest proof") + install_sha256, install_bytes_count = file_sha256_size(install_proof_path) + if install_descriptor.get("sha256") != install_sha256: raise ProofHostError( [ _failure( - "proof-host proof byte count is wrong", - expected=str(proof_bytes_count), - actual=repr(proof_descriptor.get("bytes")), + "proof-host install proof SHA-256 is wrong", + expected=install_sha256, + actual=repr(install_descriptor.get("sha256")), repair="bash scripts/release.sh --candidate", ) ] ) - _validate_directory_identities( - (request_identity, candidate_identity, output_identity) - ) - proof_bytes = proof_path.read_bytes() - _validate_directory_identities( - (request_identity, candidate_identity, output_identity) + if install_descriptor.get("bytes") != install_bytes_count: + raise ProofHostError( + [ + _failure( + "proof-host install proof byte count is wrong", + expected=str(install_bytes_count), + actual=repr(install_descriptor.get("bytes")), + repair="bash scripts/release.sh --candidate", + ) + ] + ) + nvattest_sha256, nvattest_bytes_count = file_sha256_size( + nvattest_proof_path ) + if nvattest_descriptor.get("sha256") != nvattest_sha256: + raise ProofHostError( + [ + _failure( + "proof-host nvattest proof SHA-256 is wrong", + expected=nvattest_sha256, + actual=repr(nvattest_descriptor.get("sha256")), + repair="bash scripts/release.sh --candidate", + ) + ] + ) + if nvattest_descriptor.get("bytes") != nvattest_bytes_count: + raise ProofHostError( + [ + _failure( + "proof-host nvattest proof byte count is wrong", + expected=str(nvattest_bytes_count), + actual=repr(nvattest_descriptor.get("bytes")), + repair="bash scripts/release.sh --candidate", + ) + ] + ) + _validate_directory_identities(identities) + install_proof_bytes = install_proof_path.read_bytes() + nvattest_proof_bytes = nvattest_proof_path.read_bytes() + _validate_directory_identities(identities) proof_failures = validate_install_proof_bytes( - proof_bytes, + install_proof_bytes, target=target, version=version, source_commit=source_commit, @@ -900,15 +1106,33 @@ class ExternalProofHostChannel: ) if proof_failures: raise ProofHostError(proof_failures) - output_path.parent.mkdir(parents=True, exist_ok=True) - temp_path = output_path.with_name(f".{output_path.name}.tmp") - try: - temp_path.write_bytes(proof_bytes) - os.rename(temp_path, output_path) - finally: - temp_path.unlink(missing_ok=True) + nvattest_failures = validate_nvattest_proof_bytes( + nvattest_proof_bytes, + expected_challenge=challenge, + target=target, + version=version, + source_commit=source_commit, + core_lock_sha256=core_lock_sha256, + candidate_digest=candidate_digest, + ledger_sha256=ledger_sha256, + canonical_authority_bytes=canonical_authority_bytes, + expected_support_distributions=support_distributions, + ) + if nvattest_failures: + raise ProofHostError(nvattest_failures) + for final_path, proof_bytes in ( + (output_path, install_proof_bytes), + (nvattest_output_path, nvattest_proof_bytes), + ): + final_path.parent.mkdir(parents=True, exist_ok=True) + temp_path = final_path.with_name(f".{final_path.name}.tmp") + try: + temp_path.write_bytes(proof_bytes) + os.rename(temp_path, final_path) + finally: + temp_path.unlink(missing_ok=True) completed = True - return output_path + return TargetProofPaths(install=output_path, nvattest=nvattest_output_path) except BaseException as exc: if isinstance(exc, ProofHostError): primary = exc @@ -937,6 +1161,8 @@ class ExternalProofHostChannel: local_cleanup = self._cleanup_local( ( (request_candidate_dir, candidate_identity), + (request_support_dir, support_identity), + (request_authority_dir, authority_identity), (request_output_dir, output_identity), (request_dir, request_identity), ) @@ -961,10 +1187,10 @@ def proof_channels_from_env( } -def run_install_proof_with_channels( +def run_target_proofs_with_channels( channels: Mapping[str, ProofHostChannel], **kwargs: Any, -) -> Path: +) -> TargetProofPaths: target = str(kwargs.get("target", "")) channel = channels.get(target) if channel is None: @@ -978,4 +1204,4 @@ def run_install_proof_with_channels( ) ] ) - return channel.run_install_proof(**kwargs) + return channel.run_target_proofs(**kwargs) diff --git a/tests/helpers/release_candidate_fixtures.py b/tests/helpers/release_candidate_fixtures.py index e58926a23..e6dc463f7 100644 --- a/tests/helpers/release_candidate_fixtures.py +++ b/tests/helpers/release_candidate_fixtures.py @@ -20,6 +20,7 @@ from typing import Any, Literal import scripts.check_rust_release_manifest as checker import scripts.record_macos_native_wheel as native import scripts.release_candidate_driver as driver +import scripts.release_nvattest_proof as nvattest_proof import scripts.release_tool_pins as pins from scripts.check_wheel_contents import ( CORE_REQUIRED_SDIST_MEMBERS, @@ -44,9 +45,17 @@ from scripts.release_install_smoke import ( target_install_paths_from_ledger, write_install_proof, ) +from scripts.release_nvattest_proof import SUPPORT_DISTRIBUTION_NAMES +from scripts.release_nvattest_support import read_support_lock_entries +from scripts.release_proof_host import TARGET_POLICY, TargetProofPaths from solstone.think.probe import ( SOLSTONE_CORE_SPEAKERS_ANALYZE_PLATFORM_TAGS, ) +from solstone.think.providers.nvattest_authority import ( + authority_payload, + nvattest_target_key, +) +from solstone.think.providers.nvattest_install import SIDECAR_SCHEMA_VERSION SPEAKERS_ANALYZE_LINUX_X86_64_TAG = SOLSTONE_CORE_SPEAKERS_ANALYZE_PLATFORM_TAGS[ ("linux", "x86_64") @@ -61,6 +70,7 @@ from tests.helpers.release_wheel_fixtures import ( write_core_wheel, write_platform_base_wheel, write_speakers_analyze_wheel, + write_support_wheel, ) SOURCE_COMMIT = "a" * 40 @@ -112,6 +122,7 @@ def repo(tmp_path: Path) -> Path: '[project]\nname = "solstone-journal-models"\nversion = "1.0.0"\n', encoding="utf-8", ) + _write_fixture_support_lock(root) return root @@ -142,6 +153,41 @@ def _policy() -> PolicyRun: ) +def _support_version(index: int) -> str: + return f"0.0.{index}" + + +def _write_fixture_support_wheels(path: Path) -> tuple[Path, ...]: + return tuple( + write_support_wheel(path, name=name, version=_support_version(index)) + for index, name in enumerate(sorted(SUPPORT_DISTRIBUTION_NAMES), start=1) + ) + + +def _write_fixture_support_lock(root: Path) -> None: + wheels = _write_fixture_support_wheels(root / "fixture-support-lock") + by_name = {wheel.name.split("-", 1)[0].replace("_", "-"): wheel for wheel in wheels} + blocks = ["version = 1\n"] + for index, name in enumerate(sorted(SUPPORT_DISTRIBUTION_NAMES), start=1): + wheel = by_name[name] + sha256, size = driver.file_sha256_size(wheel) + blocks.extend( + [ + "[[package]]\n", + f'name = "{name}"\n', + f'version = "{_support_version(index)}"\n', + "wheels = [\n", + " { " + f'url = "wheels/{wheel.name}", ' + f'hash = "sha256:{sha256}", ' + f"size = {size} " + "},\n", + "]\n\n", + ] + ) + (root / "uv.lock").write_text("".join(blocks), encoding="utf-8") + + def _wheel_metadata(name: str) -> tuple[str, str]: parts = name.removesuffix(".whl").split("-") distribution = parts[0] @@ -508,6 +554,259 @@ def _proof_observation( ) +def _nvattest_target_key(target: str) -> str: + policy_os, policy_arch = TARGET_POLICY[target] + target_key = nvattest_target_key(policy_os.lower(), policy_arch) + assert target_key is not None + return target_key + + +def _nvattest_authority_target(target: str) -> Mapping[str, Any]: + target_key = _nvattest_target_key(target) + targets = authority_payload()["targets"] + return targets[target_key] + + +def _nvattest_fixture_path(name: str) -> Path: + return Path(__file__).resolve().parents[1] / "fixtures" / "nvattest" / name + + +def _nvattest_member_facts( + authority_target: Mapping[str, Any], +) -> list[dict[str, Any]]: + return [ + { + "content_sha256": ( + hashlib.sha256(str(member["relpath"]).encode("utf-8")).hexdigest() + if member["kind"] == "regular" + else None + ), + "executable": member["executable"], + "kind": member["kind"], + "relpath": member["relpath"], + "symlink_target": member["symlink_target"], + } + for member in sorted( + authority_target["inventory"], + key=lambda item: item["relpath"], + ) + ] + + +def _nvattest_driver_payload( + *, + target: str, + env_root: Path, + journal_path: Path, + canonical_authority_bytes: bytes, +) -> dict[str, Any]: + target_key = _nvattest_target_key(target) + authority_target = _nvattest_authority_target(target) + site_root = env_root / "lib" / "python3.13" / "site-packages" + authority_path = ( + site_root / "solstone" / "think" / "providers" / "nvattest_authority_v1.json" + ) + cache_root = journal_path / "cache" / "providers" / "nvattest" + sidecar_path = cache_root / ".nvattest-install.json" + fingerprint = hashlib.sha256(target_key.encode("utf-8")).hexdigest() + sidecar = { + "artifact": dict(authority_target["artifact"]), + "schema_version": SIDECAR_SCHEMA_VERSION, + "target_key": target_key, + "tree_fingerprint_sha256": fingerprint, + "version": authority_target["source"]["version"], + } + sidecar_bytes = checker.canonical_json_bytes(sidecar) + return { + "authority_module_file": str( + site_root / "solstone" / "think" / "providers" / "nvattest_authority.py" + ), + "authority_origin": str( + site_root / "solstone" / "think" / "providers" / "nvattest_authority.py" + ), + "authority_path": str(authority_path), + "authority_sha256": hashlib.sha256(canonical_authority_bytes).hexdigest(), + "authority_size_bytes": len(canonical_authority_bytes), + "cache_root": str(cache_root), + "dist_info": [ + { + "dist_info_path": str( + site_root / f"solstone-{checker._current_version()}.dist-info" + ), + "name": "solstone", + "version": checker._current_version(), + } + ], + "journal_path": str(journal_path), + "members": _nvattest_member_facts(authority_target), + "module_file": str( + site_root / "solstone" / "think" / "providers" / "nvattest_install.py" + ), + "module_origin": str( + site_root / "solstone" / "think" / "providers" / "nvattest_install.py" + ), + "sidecar": sidecar, + "sidecar_path": str(sidecar_path), + "sidecar_sha256": hashlib.sha256(sidecar_bytes).hexdigest(), + "sidecar_size_bytes": len(sidecar_bytes), + "site_packages": [str(site_root)], + "solstone_journal_present": False, + "spp_nvattest_dir_present": False, + "tree_fingerprint_sha256": fingerprint, + } + + +def _nvattest_command_result( + argv: Sequence[str], + *, + stdout: str = "", + exit_code: int = 0, +) -> CommandResult: + return CommandResult( + argv=tuple(argv), + exit_code=exit_code, + stdout=stdout, + stderr="", + env=SCRUBBED_COMMAND_ENV, + ) + + +def _nvattest_services( + *, + root: Path, + target: str, + candidate_paths: Sequence[Path], + support_paths: Sequence[Path], + canonical_authority_bytes: bytes, +) -> nvattest_proof.NvattestProofServices: + env_root = root / "nvattest-env" / target + policy_os, policy_arch = TARGET_POLICY[target] + authority_target = _nvattest_authority_target(target) + expected_distribution_output = "".join( + f"{entry['name']}=={entry['version']}\n" + for entry in expected_distribution_entries(candidate_paths) + ) + + def create_environment(_target: str) -> Path: + (env_root / "bin").mkdir(parents=True, exist_ok=True) + python = env_root / "bin" / "python" + python.write_text( + f"#!/bin/sh\ncat <<'EOF'\n{expected_distribution_output}EOF\n", + encoding="utf-8", + ) + python.chmod(0o755) + return env_root + + def install_wheels( + env_python: Path, + candidate_wheels: Sequence[Path], + support_wheels: Sequence[Path], + ) -> CommandResult: + assert tuple(candidate_wheels) == tuple(candidate_paths) + assert tuple(support_wheels) == tuple(support_paths) + site_root = env_root / "lib" / "python3.13" / "site-packages" + authority_path = ( + site_root + / "solstone" + / "think" + / "providers" + / "nvattest_authority_v1.json" + ) + authority_path.parent.mkdir(parents=True, exist_ok=True) + authority_path.write_bytes(canonical_authority_bytes) + (site_root / f"solstone-{checker._current_version()}.dist-info").mkdir( + parents=True, exist_ok=True + ) + return _nvattest_command_result( + ( + str(env_python), + "-m", + "pip", + "install", + "--no-index", + "--no-deps", + *(str(path) for path in candidate_wheels), + *(str(path) for path in support_wheels), + ), + stdout="installed", + ) + + def fetch(label: str, url: str, dest: Path) -> nvattest_proof.FetchObservation: + dest.parent.mkdir(parents=True, exist_ok=True) + if label == "archive": + artifact = authority_target["artifact"] + dest.write_bytes(b"fixture archive bytes are not retained\n") + return nvattest_proof.FetchObservation( + label=label, + url=url, + path=dest, + sha256=str(artifact["sha256"]), + size_bytes=int(artifact["size_bytes"]), + ) + source = _nvattest_fixture_path( + str(authority_target["companion_manifest"]["name"]) + ) + shutil.copy2(source, dest) + sha256, size_bytes = driver.file_sha256_size(dest) + return nvattest_proof.FetchObservation( + label=label, url=url, path=dest, sha256=sha256, size_bytes=size_bytes + ) + + def run_package_install( + env_python: Path, + driver_path: Path, + target_key: str, + journal_path: Path, + ) -> nvattest_proof.DriverObservation: + assert target_key == _nvattest_target_key(target) + payload = _nvattest_driver_payload( + target=target, + env_root=env_root, + journal_path=journal_path, + canonical_authority_bytes=canonical_authority_bytes, + ) + return nvattest_proof.DriverObservation( + command=_nvattest_command_result( + ( + str(env_python), + str(driver_path), + "--target-key", + target_key, + "--journal-path", + str(journal_path), + ), + stdout=json.dumps(payload, sort_keys=True), + ), + payload=payload, + ) + + def run_smoke(nvattest_bin: Path) -> CommandResult: + return _nvattest_command_result((str(nvattest_bin), "--help"), stdout="usage\n") + + return nvattest_proof.NvattestProofServices( + create_environment=create_environment, + install_wheels=install_wheels, + fetch=fetch, + run_package_install=run_package_install, + integrity_recheck=lambda _journal, _target_key, _fetches, driver_observation: { + "members": driver_observation.payload["members"], + "sidecar": driver_observation.payload["sidecar"], + "sidecar_path": driver_observation.payload["sidecar_path"], + "sidecar_sha256": driver_observation.payload["sidecar_sha256"], + "sidecar_size_bytes": driver_observation.payload["sidecar_size_bytes"], + "tree_fingerprint_sha256": driver_observation.payload[ + "tree_fingerprint_sha256" + ], + }, + run_smoke=run_smoke, + clock=lambda: datetime(2026, 7, 20, 12, 35, tzinfo=UTC), + cleanup=lambda path: shutil.rmtree(path), + observe_host=lambda: nvattest_proof.HostObservation( + os=policy_os, arch=policy_arch + ), + ) + + def services( root: Path, *, native_mutation: str | None = None ) -> driver.CandidateServices: @@ -568,7 +867,14 @@ def services( assert commit == SOURCE_COMMIT return write_macos_host_outputs(output_dir, mutate=native_mutation) - def run_proof(**kwargs: Any) -> Path: + def materialize_support_wheels(destination: Path) -> tuple[Path, ...]: + entries = read_support_lock_entries(root / "uv.lock") + return tuple( + write_support_wheel(destination, name=entry.name, version=entry.version) + for entry in entries + ) + + def run_target_proofs(**kwargs: Any) -> TargetProofPaths: output_path = Path(kwargs["output_path"]) target = str(kwargs["target"]) install_paths = target_install_paths_from_ledger( @@ -576,8 +882,19 @@ def services( target=target, candidate_dir=Path(kwargs["candidate_dir"]), ) + install_excluded_keys = { + "canonical_authority_bytes", + "challenge", + "nvattest_output_path", + "output_path", + "support_wheel_paths", + } proof = build_install_proof( - **{key: value for key, value in kwargs.items() if key != "output_path"}, + **{ + key: value + for key, value in kwargs.items() + if key not in install_excluded_keys + }, observation=_proof_observation( target, env_root=root / "env" / target, @@ -587,7 +904,7 @@ def services( ), recorded_at=datetime(2026, 7, 20, 12, 30, tzinfo=UTC), ) - return write_install_proof( + install_receipt = write_install_proof( output_path, proof, target=target, @@ -599,6 +916,32 @@ def services( candidate_dir=Path(kwargs["candidate_dir"]), ledger_payload=kwargs["ledger_payload"], ) + evidence_staging = output_path.parents[1] + support_paths = tuple( + sorted((evidence_staging / "support").iterdir(), key=lambda path: path.name) + ) + nvattest_receipt = nvattest_proof.run_nvattest_proof( + target=target, + version=str(kwargs["version"]), + source_commit=str(kwargs["source_commit"]), + core_lock_sha256=str(kwargs["core_lock_sha256"]), + candidate_digest=str(kwargs["candidate_digest"]), + ledger_sha256=str(kwargs["ledger_sha256"]), + challenge=str(kwargs["ledger_payload"]["nvattest"]["challenge"]), + candidate_dir=Path(kwargs["candidate_dir"]), + candidate_paths=install_paths, + support_wheel_paths=support_paths, + output_path=Path(kwargs["nvattest_output_path"]), + services=_nvattest_services( + root=root, + target=target, + candidate_paths=install_paths, + support_paths=support_paths, + canonical_authority_bytes=NVATTEST_AUTHORITY_BYTES, + ), + canonical_authority_bytes=NVATTEST_AUTHORITY_BYTES, + ) + return TargetProofPaths(install=install_receipt, nvattest=nvattest_receipt) def cleanup(paths: Sequence[Path]) -> None: for path in paths: @@ -621,7 +964,9 @@ def services( create_source_bundle=create_source_bundle, build_host=build_host, cleanup_transients=cleanup, - run_install_proof=run_proof, + challenge_factory=lambda: hashlib.sha256(str(root).encode("utf-8")).hexdigest(), + materialize_support_wheels=materialize_support_wheels, + run_target_proofs=run_target_proofs, transaction_hook=lambda _point: None, ) diff --git a/tests/helpers/release_wheel_fixtures.py b/tests/helpers/release_wheel_fixtures.py index 8a7159c09..02b82b7cf 100644 --- a/tests/helpers/release_wheel_fixtures.py +++ b/tests/helpers/release_wheel_fixtures.py @@ -7,12 +7,14 @@ from __future__ import annotations import base64 import hashlib +import re import struct import zipfile from collections.abc import Mapping, Sequence from pathlib import Path import scripts.check_wheel_contents as checker +from scripts.build_nvattest_authority import render_nvattest_authority_json from solstone.think.probe import ( SOLSTONE_CORE_SPEAKERS_ANALYZE_PLATFORM_TAGS, ) @@ -26,7 +28,7 @@ SPEAKERS_ANALYZE_DEFAULT_TAG = SOLSTONE_CORE_SPEAKERS_ANALYZE_PLATFORM_TAGS[ ELF_HEADER_SIZE = 64 ELF_PROGRAM_HEADER_SIZE = 56 -NVATTEST_AUTHORITY_BYTES = b'{"schema_version":1}\n' +NVATTEST_AUTHORITY_BYTES = render_nvattest_authority_json().encode("utf-8") ROOT_LAUNCHER_BYTES = { name: f"#!/bin/sh\necho fixture {name}\n".encode("utf-8") for name in checker.ROOT_LAUNCHER_NAMES @@ -51,6 +53,27 @@ def _write_member( wheel.writestr(info, content) +def write_support_wheel(path: Path, *, name: str, version: str) -> Path: + path.mkdir(parents=True, exist_ok=True) + distribution = re.sub(r"[-.]+", "_", name) + wheel_path = path / f"{distribution}-{version}-py3-none-any.whl" + dist_info = f"{distribution}-{version}.dist-info" + members = { + f"{dist_info}/METADATA": f"Name: {name}\nVersion: {version}\n".encode("utf-8"), + f"{dist_info}/WHEEL": b"Wheel-Version: 1.0\n", + } + rows = [ + f"{member},{record_hash(content)},{len(content)}" + for member, content in members.items() + ] + rows.append(f"{dist_info}/RECORD,,") + with zipfile.ZipFile(wheel_path, "w") as wheel: + for member, content in members.items(): + _write_member(wheel, member, content) + _write_member(wheel, f"{dist_info}/RECORD", "\n".join(rows).encode("utf-8")) + return wheel_path + + def minimal_elf( machine: int, *, program_type: int = 1, dynamic_needed: bool = False ) -> bytes: diff --git a/tests/test_channel_adapters.py b/tests/test_channel_adapters.py index 62e03a2ec..28123567f 100644 --- a/tests/test_channel_adapters.py +++ b/tests/test_channel_adapters.py @@ -26,7 +26,12 @@ from scripts.release_tool_pins import ( MACOS_SWIFT_FLATTENED_BANNER, UV_MACOS_FIXTURE_BANNER, ) -from tests.helpers.release_wheel_fixtures import ROOT_LAUNCHER_BYTES, record_hash +from tests.helpers import release_candidate_fixtures as candidate_fixtures +from tests.helpers.release_wheel_fixtures import ( + NVATTEST_AUTHORITY_BYTES, + ROOT_LAUNCHER_BYTES, + record_hash, +) def _completed( @@ -178,8 +183,12 @@ def _write_build_request(tmp_path: Path) -> tuple[Path, build_rail.SourceBundle, def _write_proof_request(tmp_path: Path) -> tuple[Path, dict[str, Any], dict[str, Any]]: request_dir = tmp_path / "proof-request" candidate_dir = request_dir / "candidate" + support_dir = request_dir / "support" + authority_dir = request_dir / "authority" output_dir = request_dir / "output" candidate_dir.mkdir(parents=True) + support_dir.mkdir() + authority_dir.mkdir() output_dir.mkdir() for name in ( "solstone-1.0.0-py3-none-any.whl", @@ -216,6 +225,12 @@ def _write_proof_request(tmp_path: Path) -> tuple[Path, dict[str, Any], dict[str target="linux-x86_64-musl", candidate_dir=candidate_dir, ) + support_paths = candidate_fixtures._write_fixture_support_wheels( # noqa: SLF001 + support_dir + ) + (authority_dir / "nvattest_authority_v1.json").write_bytes( + NVATTEST_AUTHORITY_BYTES + ) channel = proof_rail.ExternalProofHostChannel( "linux-x86_64-musl", ["adapter"], @@ -229,6 +244,9 @@ def _write_proof_request(tmp_path: Path) -> tuple[Path, dict[str, Any], dict[str candidate_digest=digest, ledger_sha256=ledger_sha, install_paths=install_paths, + challenge="e" * 64, + support_wheel_paths=support_paths, + canonical_authority_bytes=NVATTEST_AUTHORITY_BYTES, ) request_path = request_dir / "request.json" request_path.write_text(json.dumps(payload), encoding="utf-8") @@ -437,7 +455,8 @@ def test_proof_request_response_round_trip_through_rail_parser( monkeypatch: pytest.MonkeyPatch, ) -> None: request_path, request_payload, ledger = _write_proof_request(tmp_path) - proof_path = request_path.parent / "output" / "proof.json" + proof_path = request_path.parent / request_payload["paths"]["install_proof"] + nvattest_path = request_path.parent / request_payload["paths"]["nvattest_proof"] def fake_runner(argv, **kwargs): if argv == ["uname", "-s"]: @@ -450,9 +469,25 @@ def test_proof_request_response_round_trip_through_rail_parser( request_payload=request_payload, ledger=ledger, ) + nvattest_path.write_bytes(b"nvattest proof\n") sha256, byte_count = common.sha256_size(proof_path) + nvattest_sha256, nvattest_byte_count = common.sha256_size(nvattest_path) return _completed( - f'{proof_host.PROOF_TOKEN} {{"bytes": {byte_count}, "sha256": "{sha256}"}}\n' + f"{proof_host.PROOF_TOKEN} " + + json.dumps( + { + "install_proof": { + "bytes": byte_count, + "sha256": sha256, + }, + "nvattest_proof": { + "bytes": nvattest_byte_count, + "sha256": nvattest_sha256, + }, + }, + sort_keys=True, + ) + + "\n" ) raise AssertionError(argv) @@ -462,14 +497,15 @@ def test_proof_request_response_round_trip_through_rail_parser( response = json.loads((request_path.parent / "response.json").read_text()) channel = proof_rail.ExternalProofHostChannel("linux-x86_64-musl", ["adapter"]) - proof_descriptor = channel._validate_response( + proof_descriptors = channel._validate_response( response, cohort_id="cohort", target="linux-x86_64-musl", candidate_digest=request_payload["candidate_digest"], ledger_sha256=request_payload["ledger_sha256"], ) - assert proof_descriptor["path"] == "output/proof.json" + assert proof_descriptors["install_proof"]["path"] == "output/install-proof.json" + assert proof_descriptors["nvattest_proof"]["path"] == "output/nvattest-proof.json" proof_failures = smoke.validate_install_proof_bytes( proof_path.read_bytes(), target=request_payload["target"], diff --git a/tests/test_release_candidate_driver.py b/tests/test_release_candidate_driver.py index 1e2f77e05..6639037de 100644 --- a/tests/test_release_candidate_driver.py +++ b/tests/test_release_candidate_driver.py @@ -14,7 +14,7 @@ import sys import tarfile import tomllib import zipfile -from collections.abc import Sequence +from collections.abc import Mapping, Sequence from dataclasses import dataclass, replace from io import BytesIO from pathlib import Path @@ -33,6 +33,7 @@ from scripts.check_wheel_contents import ( core_wheel_script_members, ) from scripts.release_build_host import BuildHostResult, SourceBundle +from scripts.release_nvattest_proof import CHALLENGE_RE from scripts.transparency_core import collect_candidate_parts, snapshot_candidate from tests.helpers.release_candidate_fixtures import ( LOCK_SHA, @@ -521,6 +522,25 @@ def _assert_dist_preflight_failure( assert exc.value.failures[0].repair == "bash scripts/release.sh --candidate" +def _differing_payload_paths( + first: object, + second: object, + path: tuple[str, ...] = (), +) -> set[tuple[str, ...]]: + if isinstance(first, Mapping) and isinstance(second, Mapping): + keys = set(first) | set(second) + return { + diff + for key in keys + for diff in _differing_payload_paths( + first.get(key), + second.get(key), + (*path, str(key)), + ) + } + return set() if first == second else {path} + + def test_fake_all_host_candidate_and_recovery_are_deterministic( tmp_path: Path, ) -> None: @@ -542,11 +562,16 @@ def test_fake_all_host_candidate_and_recovery_are_deterministic( / f".{checker._current_version()}.source.bundle" ).exists() assert first.candidate_digest == second.candidate_digest - assert first.bundle_digest == second.bundle_digest - assert ( - first.evidence_dir.joinpath("ledger.json").read_bytes() - == second.evidence_dir.joinpath("ledger.json").read_bytes() - ) + first_ledger = json.loads(first.evidence_dir.joinpath("ledger.json").read_text()) + second_ledger = json.loads(second.evidence_dir.joinpath("ledger.json").read_text()) + assert _differing_payload_paths(first_ledger, second_ledger) == { + ("nvattest", "challenge") + } + first_challenge = first_ledger["nvattest"]["challenge"] + second_challenge = second_ledger["nvattest"]["challenge"] + assert CHALLENGE_RE.fullmatch(first_challenge) + assert CHALLENGE_RE.fullmatch(second_challenge) + assert first_challenge != second_challenge assert sorted(path.name for path in first.release_dir.iterdir()) == sorted( path.name for path in second.release_dir.iterdir() ) @@ -644,16 +669,16 @@ def test_candidate_pair_promote_rejects_publication_prerequisite_in_staging( root = _repo(tmp_path) base_services = _services(root) - def run_install_proof(**kwargs: Any) -> Path: - output_path = base_services.run_install_proof(**kwargs) - evidence_staging = output_path.parents[1] + def run_target_proofs(**kwargs: Any) -> Any: + target_proofs = base_services.run_target_proofs(**kwargs) + evidence_staging = target_proofs.install.parents[1] write_core_unsupported_tombstone_record( evidence_staging, checker._current_version(), ) - return output_path + return target_proofs - services = replace(base_services, run_install_proof=run_install_proof) + services = replace(base_services, run_target_proofs=run_target_proofs) with pytest.raises(driver.DriverError) as exc: driver.run_candidate(root, _env(), services) @@ -940,6 +965,7 @@ def test_recovery_identity_is_invariant_across_absent_and_present_prerequisite( "candidate_digest", "ledger_sha256", "proof_sha256", + "nvattest_sha256", "bundle_digest", "payload_files", ): @@ -948,6 +974,8 @@ def test_recovery_identity_is_invariant_across_absent_and_present_prerequisite( "payload_inventory", "evidence_inventory", "proof_inventory", + "nvattest_inventory", + "support_inventory", ): assert absent_payload[key] == present_payload[key] assert absent_payload.pop("publication_prerequisite_inventory") == [] @@ -1117,6 +1145,8 @@ def test_machine_report_is_canonical_sorted_and_not_publication_authorization( ) for target, entry in payload["proof_inventory"].items(): assert entry["sha256"] == payload["proof_sha256"][target] + for target, entry in payload["nvattest_inventory"].items(): + assert entry["sha256"] == payload["nvattest_sha256"][target] def test_candidate_cleanup_receives_release_zig_cache_root(tmp_path: Path) -> None: diff --git a/tests/test_release_digest.py b/tests/test_release_digest.py index b50c50c8d..38d29c1b0 100644 --- a/tests/test_release_digest.py +++ b/tests/test_release_digest.py @@ -6,6 +6,7 @@ from __future__ import annotations import hashlib import scripts.release_digest as digest +from scripts.check_rust_release_manifest import canonical_json_bytes def test_candidate_digest_uses_basename_sorted_two_space_lf_stream(tmp_path) -> None: @@ -50,13 +51,31 @@ def test_bundle_digest_excludes_self_hash_and_is_canonical() -> None: "linux-x86_64-musl": "3" * 64, "linux-aarch64-musl": "2" * 64, } + nvattest_hashes = { + "macos-arm64": "7" * 64, + "linux-x86_64-musl": "6" * 64, + "linux-aarch64-musl": "5" * 64, + } + expected_payload = { + "candidate_digest": candidate, + "ledger_sha256": ledger, + "nvattest_sha256": { + target: nvattest_hashes[target] for target in sorted(nvattest_hashes) + }, + "proof_sha256": { + target: proof_hashes[target] for target in sorted(proof_hashes) + }, + } + expected = hashlib.sha256(canonical_json_bytes(expected_payload)).hexdigest() - assert digest.bundle_digest(candidate, ledger, proof_hashes) == ( - "ccce5e9d1a416579b9a07c3e00a60acfd24e78d7fc38fd75f8a16962d27a1021" + assert ( + digest.bundle_digest(candidate, ledger, proof_hashes, nvattest_hashes) + == expected ) with_extra_self_hash = dict(proof_hashes) with_extra_self_hash["bundle_digest"] = "5" * 64 - assert digest.bundle_digest(candidate, ledger, with_extra_self_hash) != ( - "ccce5e9d1a416579b9a07c3e00a60acfd24e78d7fc38fd75f8a16962d27a1021" + assert ( + digest.bundle_digest(candidate, ledger, with_extra_self_hash, nvattest_hashes) + != expected ) diff --git a/tests/test_release_ledger.py b/tests/test_release_ledger.py index 151dadc83..82c7c10ef 100644 --- a/tests/test_release_ledger.py +++ b/tests/test_release_ledger.py @@ -3,6 +3,7 @@ from __future__ import annotations +import hashlib import json import zipfile from pathlib import Path @@ -13,6 +14,7 @@ import scripts.check_release_preflight as preflight import scripts.check_rust_release_manifest as checker import scripts.release_ledger as ledger import scripts.release_tool_pins as pins +from scripts.build_nvattest_authority import render_nvattest_authority_json from scripts.check_rust_release_manifest import canonical_json_bytes from scripts.check_wheel_contents import ( CORE_SCRIPT_NAMES, @@ -21,6 +23,7 @@ from scripts.check_wheel_contents import ( SPEAKERS_ANALYZE_TARGETS, ) from scripts.release_advisory_policy import PolicyRun +from scripts.release_nvattest_proof import SUPPORT_DISTRIBUTION_NAMES SOURCE_COMMIT = "a" * 40 MALFORMED_DB_COMMIT_CASES = ( @@ -233,6 +236,25 @@ def _models(decision: str = "exclude") -> dict[str, str]: return {"decision": decision, "package_version": "1.0.0"} +def _nvattest(challenge: str | None = None) -> dict[str, object]: + authority_bytes = render_nvattest_authority_json().encode("utf-8") + return { + "authority": json.loads(authority_bytes.decode("utf-8")), + "authority_sha256": hashlib.sha256(authority_bytes).hexdigest(), + "challenge": challenge or hashlib.sha256(b"challenge").hexdigest(), + "support_distributions": [ + { + "bytes": len(name.encode("utf-8")), + "filename": f"{name.replace('-', '_')}-0.0.{index}-py3-none-any.whl", + "name": name, + "sha256": hashlib.sha256(name.encode("utf-8")).hexdigest(), + "version": f"0.0.{index}", + } + for index, name in enumerate(sorted(SUPPORT_DISTRIBUTION_NAMES), start=1) + ], + } + + def _ledger_path(root: Path) -> Path: return ledger.write_ledger( evidence_root=root / "target" / "release-evidence", @@ -244,6 +266,7 @@ def _ledger_path(root: Path) -> Path: policy_run=_policy(), native_records=_native_records(), models=_models(), + nvattest=_nvattest(), ) @@ -264,6 +287,7 @@ def test_ledger_is_byte_deterministic_for_fixed_inputs(tmp_path: Path) -> None: policy_run=_policy(), native_records=_native_records(), models=_models(), + nvattest=_nvattest(), ) second = ledger.write_ledger( evidence_root=tmp_path / "two" / "target" / "release-evidence", @@ -275,6 +299,7 @@ def test_ledger_is_byte_deterministic_for_fixed_inputs(tmp_path: Path) -> None: policy_run=_policy(), native_records=_native_records(), models=_models(), + nvattest=_nvattest(), ) assert first.read_bytes() == second.read_bytes() @@ -291,6 +316,7 @@ def test_ledger_key_set_excludes_transport_and_bundle_state(tmp_path: Path) -> N policy_run=_policy(), native_records=_native_records(), models=_models(), + nvattest=_nvattest(), ) payload = json.loads(path.read_text(encoding="utf-8")) text = path.read_text(encoding="utf-8") @@ -326,6 +352,7 @@ def test_ledger_rejects_raw_signer_team_and_uuid_evidence(tmp_path: Path) -> Non policy_run=_policy(), native_records=records, models=_models(), + nvattest=_nvattest(), ) records = _native_records() @@ -340,6 +367,7 @@ def test_ledger_rejects_raw_signer_team_and_uuid_evidence(tmp_path: Path) -> Non policy_run=_policy(), native_records=records, models=_models(), + nvattest=_nvattest(), ) @@ -354,6 +382,7 @@ def test_ledger_requires_exactly_three_native_records(tmp_path: Path) -> None: policy_run=_policy(), native_records=_native_records()[:1], models=_models(), + nvattest=_nvattest(), ) assert exc.value.failures[0].error == "macOS native record set is incomplete" @@ -375,6 +404,7 @@ def test_ledger_requires_full_tool_cohort_per_lane(tmp_path: Path, lane: str) -> policy_run=_policy(), native_records=_native_records(), models=_models(), + nvattest=_nvattest(), ) assert any("tool" in failure.error for failure in exc.value.failures) diff --git a/tests/test_release_nvattest_support.py b/tests/test_release_nvattest_support.py new file mode 100644 index 000000000..4ad2a27b3 --- /dev/null +++ b/tests/test_release_nvattest_support.py @@ -0,0 +1,101 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from scripts.release_digest import file_sha256_size +from scripts.release_nvattest_proof import SUPPORT_DISTRIBUTION_NAMES +from scripts.release_nvattest_support import ( + SupportLockError, + read_support_lock_entries, + support_declarations_from_lock, + verify_support_wheels_against_lock, +) +from tests.helpers.release_wheel_fixtures import write_support_wheel + + +def _support_version(index: int) -> str: + return f"0.0.{index}" + + +def _write_support_wheels(path: Path) -> tuple[Path, ...]: + return tuple( + write_support_wheel(path, name=name, version=_support_version(index)) + for index, name in enumerate(sorted(SUPPORT_DISTRIBUTION_NAMES), start=1) + ) + + +def _write_lock( + path: Path, wheels: tuple[Path, ...], *, omit: str | None = None +) -> None: + blocks = ["version = 1\n"] + by_name = {wheel.name.split("-", 1)[0].replace("_", "-"): wheel for wheel in wheels} + for index, name in enumerate(sorted(SUPPORT_DISTRIBUTION_NAMES), start=1): + if name == omit: + continue + wheel = by_name[name] + sha256, size = file_sha256_size(wheel) + blocks.extend( + [ + "[[package]]\n", + f'name = "{name}"\n', + f'version = "{_support_version(index)}"\n', + "wheels = [\n", + " { " + f'url = "wheels/{wheel.name}", ' + f'hash = "sha256:{sha256}", ' + f"size = {size} " + "},\n", + "]\n\n", + ] + ) + path.write_text("".join(blocks), encoding="utf-8") + + +def test_support_lock_entries_verify_materialized_wheels(tmp_path: Path) -> None: + wheels = _write_support_wheels(tmp_path / "wheels") + lock = tmp_path / "uv.lock" + _write_lock(lock, wheels) + + entries = read_support_lock_entries(lock) + declarations = support_declarations_from_lock(entries) + + assert {entry.name for entry in entries} == SUPPORT_DISTRIBUTION_NAMES + assert verify_support_wheels_against_lock(wheels, entries) == declarations + + +def test_support_lock_requires_exact_support_packages(tmp_path: Path) -> None: + wheels = _write_support_wheels(tmp_path / "wheels") + lock = tmp_path / "uv.lock" + missing = sorted(SUPPORT_DISTRIBUTION_NAMES)[0] + _write_lock(lock, wheels, omit=missing) + + with pytest.raises(SupportLockError) as exc: + read_support_lock_entries(lock) + + assert any( + failure.error == "nvattest support lock package is not exact" + for failure in exc.value.failures + ) + + +def test_support_wheel_verification_rejects_lock_byte_mismatch( + tmp_path: Path, +) -> None: + wheels = _write_support_wheels(tmp_path / "wheels") + lock = tmp_path / "uv.lock" + _write_lock(lock, wheels) + entries = read_support_lock_entries(lock) + wheels[0].write_bytes(wheels[0].read_bytes() + b"changed") + + with pytest.raises(SupportLockError) as exc: + verify_support_wheels_against_lock(wheels, entries) + + assert any( + failure.error == "nvattest support wheel does not match uv.lock" + for failure in exc.value.failures + ) diff --git a/tests/test_release_proof_host.py b/tests/test_release_proof_host.py index d3909e8e4..c450e36a4 100644 --- a/tests/test_release_proof_host.py +++ b/tests/test_release_proof_host.py @@ -17,15 +17,22 @@ import pytest import scripts.check_rust_release_manifest as checker import scripts.release_install_smoke as smoke +import scripts.release_nvattest_proof as nvattest_proof import scripts.release_proof_host as proof_host from scripts.check_rust_release_manifest import canonical_json_bytes from scripts.release_digest import candidate_digest -from tests.helpers.release_wheel_fixtures import ROOT_LAUNCHER_BYTES, record_hash +from tests.helpers import release_candidate_fixtures as candidate_fixtures +from tests.helpers.release_wheel_fixtures import ( + NVATTEST_AUTHORITY_BYTES, + ROOT_LAUNCHER_BYTES, + record_hash, +) SOURCE_COMMIT = "a" * 40 CORE_LOCK = "b" * 64 LEDGER_SHA = "c" * 64 COHORT = "d" * 32 +CHALLENGE = "e" * 64 def _wheel_metadata(name: str) -> tuple[str, str]: @@ -92,6 +99,12 @@ def _candidate(tmp_path: Path) -> Path: return candidate +def _support_wheels(root: Path) -> tuple[Path, ...]: + return candidate_fixtures._write_fixture_support_wheels( # noqa: SLF001 + root / "support-wheels" + ) + + def _file_entry(path: Path) -> dict[str, Any]: data = path.read_bytes() return { @@ -286,10 +299,19 @@ def _runner( request = json.loads((cwd / "request.json").read_text(encoding="utf-8")) assert request["target"] == target assert request["paths"]["candidate_dir"] == "candidate" + assert request["paths"]["support_dir"] == "support" + assert ( + request["paths"]["authority_file"] + == "authority/nvattest_authority_v1.json" + ) assert all( str(entry["path"]).startswith("candidate/") for entry in request["candidate_files"] ) + assert all( + str(entry["path"]).startswith("support/") + for entry in request["support_files"] + ) proof_bytes = _proof_bytes( target=target, version=version, @@ -298,8 +320,47 @@ def _runner( mutate_proof=proof_mutation, mutate_observation=observation_mutation, ) - proof_path = cwd / "output" / "proof.json" - proof_path.write_bytes(proof_bytes) + install_proof_path = cwd / request["paths"]["install_proof"] + install_proof_path.write_bytes(proof_bytes) + request_candidate_dir = cwd / request["paths"]["candidate_dir"] + request_support_dir = cwd / request["paths"]["support_dir"] + candidate_paths = [ + request_candidate_dir / entry["basename"] + for entry in request["candidate_files"] + ] + support_paths = [ + request_support_dir / entry["basename"] + for entry in request["support_files"] + ] + authority_bytes = (cwd / request["paths"]["authority_file"]).read_bytes() + nvattest_proof_path = cwd / request["paths"]["nvattest_proof"] + nvattest_proof.run_nvattest_proof( + target=target, + version=version, + source_commit=SOURCE_COMMIT, + core_lock_sha256=CORE_LOCK, + candidate_digest=ledger["candidate"]["candidate_digest"], + ledger_sha256=LEDGER_SHA, + challenge=request["challenge"], + candidate_dir=request_candidate_dir, + candidate_paths=candidate_paths, + support_wheel_paths=support_paths, + output_path=nvattest_proof_path, + services=candidate_fixtures._nvattest_services( # noqa: SLF001 + root=cwd, + target=target, + candidate_paths=candidate_paths, + support_paths=support_paths, + canonical_authority_bytes=authority_bytes, + ), + canonical_authority_bytes=authority_bytes, + ) + install_descriptor = { + "path": request["paths"]["install_proof"], + "sha256": hashlib.sha256(proof_bytes).hexdigest(), + "bytes": len(proof_bytes), + } + nvattest_bytes = nvattest_proof_path.read_bytes() response: dict[str, Any] = { "schema_version": 1, "cohort_id": request["cohort_id"], @@ -309,10 +370,11 @@ def _runner( "candidate_digest": ledger["candidate"]["candidate_digest"], "ledger_sha256": LEDGER_SHA, }, - "proof": { - "path": "output/proof.json", - "sha256": hashlib.sha256(proof_bytes).hexdigest(), - "bytes": len(proof_bytes), + "install_proof": install_descriptor, + "nvattest_proof": { + "path": request["paths"]["nvattest_proof"], + "sha256": hashlib.sha256(nvattest_bytes).hexdigest(), + "bytes": len(nvattest_bytes), }, } if response_mutation is not None: @@ -340,6 +402,32 @@ def _channel( ) +def _target_proof_kwargs( + *, + target: str, + candidate: Path, + ledger: Mapping[str, Any], + output_path: Path, +) -> dict[str, Any]: + nvattest_output_path = output_path.parent / "nvattest" / f"{target}.json" + return { + "target": target, + "version": "1.0.0", + "source_commit": SOURCE_COMMIT, + "core_lock_sha256": CORE_LOCK, + "candidate_digest": ledger["candidate"]["candidate_digest"], + "ledger_sha256": LEDGER_SHA, + "candidate_dir": candidate, + "candidate_paths": tuple(candidate.iterdir()), + "ledger_payload": ledger, + "challenge": CHALLENGE, + "support_wheel_paths": _support_wheels(output_path.parent), + "canonical_authority_bytes": NVATTEST_AUTHORITY_BYTES, + "output_path": output_path, + "nvattest_output_path": nvattest_output_path, + } + + def test_proof_host_transfers_target_install_set_and_accepts_valid_proof( tmp_path: Path, ) -> None: @@ -353,21 +441,22 @@ def test_proof_host_transfers_target_install_set_and_accepts_valid_proof( _runner(target=target, candidate=candidate, ledger=ledger, calls=calls), ) - result = channel.run_install_proof( - target=target, - version="1.0.0", - source_commit=SOURCE_COMMIT, - core_lock_sha256=CORE_LOCK, - candidate_digest=ledger["candidate"]["candidate_digest"], - ledger_sha256=LEDGER_SHA, - candidate_dir=candidate, - candidate_paths=tuple(candidate.iterdir()), - ledger_payload=ledger, - output_path=output, + result = channel.run_target_proofs( + **_target_proof_kwargs( + target=target, + candidate=candidate, + ledger=ledger, + output_path=output, + ) ) - assert result == output + expected_nvattest = output.parent / "nvattest" / f"{target}.json" + assert result == proof_host.TargetProofPaths( + install=output, + nvattest=expected_nvattest, + ) payload = json.loads(output.read_text(encoding="utf-8")) + assert expected_nvattest.is_file() expected_paths = smoke.target_install_paths_from_ledger( ledger, target=target, @@ -403,17 +492,13 @@ def test_proof_host_rejects_candidate_copy_mutation_before_adapter( ) with pytest.raises(proof_host.ProofHostError) as exc: - channel.run_install_proof( - target=target, - version="1.0.0", - source_commit=SOURCE_COMMIT, - core_lock_sha256=CORE_LOCK, - candidate_digest=ledger["candidate"]["candidate_digest"], - ledger_sha256=LEDGER_SHA, - candidate_dir=candidate, - candidate_paths=tuple(candidate.iterdir()), - ledger_payload=ledger, - output_path=tmp_path / "proofs" / f"{target}.json", + channel.run_target_proofs( + **_target_proof_kwargs( + target=target, + candidate=candidate, + ledger=ledger, + output_path=tmp_path / "proofs" / f"{target}.json", + ) ) assert ( @@ -441,8 +526,8 @@ def test_missing_proof_host_channel_fails_closed() -> None: "proof-host response attestation key set is invalid", ), ( - lambda response: response["proof"].pop("bytes"), - "proof-host response proof descriptor key set is invalid", + lambda response: response["install_proof"].pop("bytes"), + "proof-host response install_proof descriptor key set is invalid", ), ( lambda response: response["attestation"].__setitem__("os", "Darwin"), @@ -475,17 +560,13 @@ def test_proof_host_rejects_closed_schema_and_attestation_mutations( ) with pytest.raises(proof_host.ProofHostError) as exc: - channel.run_install_proof( - target=target, - version="1.0.0", - source_commit=SOURCE_COMMIT, - core_lock_sha256=CORE_LOCK, - candidate_digest=ledger["candidate"]["candidate_digest"], - ledger_sha256=LEDGER_SHA, - candidate_dir=candidate, - candidate_paths=tuple(candidate.iterdir()), - ledger_payload=ledger, - output_path=tmp_path / "proof.json", + channel.run_target_proofs( + **_target_proof_kwargs( + target=target, + candidate=candidate, + ledger=ledger, + output_path=tmp_path / "proof.json", + ) ) assert any(failure.error == error for failure in exc.value.failures) @@ -564,17 +645,13 @@ def test_proof_host_rejects_semantic_proof_mutations( ) with pytest.raises(proof_host.ProofHostError) as exc: - channel.run_install_proof( - target=target, - version="1.0.0", - source_commit=SOURCE_COMMIT, - core_lock_sha256=CORE_LOCK, - candidate_digest=ledger["candidate"]["candidate_digest"], - ledger_sha256=LEDGER_SHA, - candidate_dir=candidate, - candidate_paths=tuple(candidate.iterdir()), - ledger_payload=ledger, - output_path=tmp_path / "proof.json", + channel.run_target_proofs( + **_target_proof_kwargs( + target=target, + candidate=candidate, + ledger=ledger, + output_path=tmp_path / "proof.json", + ) ) assert any(failure.error == error for failure in exc.value.failures) @@ -589,17 +666,16 @@ def test_proof_host_rejects_channel_target_swap(tmp_path: Path) -> None: ) with pytest.raises(proof_host.ProofHostError) as exc: - channel.run_install_proof( - target="linux-aarch64-musl", - version="1.0.0", - source_commit=SOURCE_COMMIT, - core_lock_sha256=CORE_LOCK, - candidate_digest=ledger["candidate"]["candidate_digest"], - ledger_sha256=LEDGER_SHA, - candidate_dir=candidate, - candidate_paths=tuple(candidate.iterdir()), - ledger_payload=ledger, - output_path=tmp_path / "proof.json", + channel.run_target_proofs( + **{ + **_target_proof_kwargs( + target="linux-aarch64-musl", + candidate=candidate, + ledger=ledger, + output_path=tmp_path / "proof.json", + ), + "target": "linux-aarch64-musl", + } ) assert exc.value.failures[0].error == "proof-host channel target mismatch" @@ -626,17 +702,13 @@ def test_proof_host_cleanup_runs_for_baseexception( ) with pytest.raises(proof_host.ProofHostError) as raised: - channel.run_install_proof( - target=target, - version="1.0.0", - source_commit=SOURCE_COMMIT, - core_lock_sha256=CORE_LOCK, - candidate_digest=ledger["candidate"]["candidate_digest"], - ledger_sha256=LEDGER_SHA, - candidate_dir=candidate, - candidate_paths=tuple(candidate.iterdir()), - ledger_payload=ledger, - output_path=tmp_path / "proof.json", + channel.run_target_proofs( + **_target_proof_kwargs( + target=target, + candidate=candidate, + ledger=ledger, + output_path=tmp_path / "proof.json", + ) ) assert any( diff --git a/tests/test_release_publish.py b/tests/test_release_publish.py index 6d44774a8..8d74efcf7 100644 --- a/tests/test_release_publish.py +++ b/tests/test_release_publish.py @@ -172,6 +172,7 @@ def _candidate( write_core_unsupported_tombstone_record(evidence_dir, candidate_version) proof_hashes: dict[str, str] = {} + nvattest_hashes: dict[str, str] = {} for target in PROOF_TARGETS: path = evidence_dir / "proofs" / f"{target}.json" path.write_text( @@ -179,6 +180,7 @@ def _candidate( encoding="utf-8", ) proof_hashes[target] = _sha(path)[0] + nvattest_hashes[target] = "0" * 64 return CandidateReport( heading="retained-candidate-valid", @@ -189,6 +191,7 @@ def _candidate( candidate_digest="b" * 64, ledger_sha256=_sha(evidence_dir / "ledger.json")[0], proof_sha256=proof_hashes, + nvattest_sha256=nvattest_hashes, bundle_digest="c" * 64, ) diff --git a/tests/test_transparency_publish.py b/tests/test_transparency_publish.py index 6e9d63d66..8942d4c2d 100644 --- a/tests/test_transparency_publish.py +++ b/tests/test_transparency_publish.py @@ -119,11 +119,13 @@ def _candidate( encoding="utf-8", ) proof_hashes: dict[str, str] = {} + nvattest_hashes: dict[str, str] = {} for target in PROOF_TARGETS: proof = {"target": target, "version": proof_version or version} path = evidence_dir / "proofs" / f"{target}.json" path.write_text(json.dumps(proof, sort_keys=True), encoding="utf-8") proof_hashes[target] = _sha(path)[0] + nvattest_hashes[target] = "0" * 64 return CandidateReport( heading="retained-candidate-valid", version=version, @@ -133,6 +135,7 @@ def _candidate( candidate_digest="b" * 64, ledger_sha256=_sha(evidence_dir / "ledger.json")[0], proof_sha256=proof_hashes, + nvattest_sha256=nvattest_hashes, bundle_digest="c" * 64, )