diff --git a/scripts/release_candidate_driver.py b/scripts/release_candidate_driver.py index 8cf1412b3..a4e4b8e23 100644 --- a/scripts/release_candidate_driver.py +++ b/scripts/release_candidate_driver.py @@ -2416,7 +2416,17 @@ def _retained_authority_binding( return authority_bytes, failures authority = nvattest.get("authority") authority_sha256 = nvattest.get("authority_sha256") - if isinstance(authority, Mapping) and authority_bytes is not None: + if not isinstance(authority, Mapping): + failures.append( + _failure( + "retained nvattest authority is invalid", + expected="ledger nvattest authority object", + actual=type(authority).__name__, + repair="bash scripts/release.sh --recover", + ) + ) + return authority_bytes, failures + if authority_bytes is not None: canonical = _canonical_nvattest_authority_bytes(authority) if canonical != authority_bytes: failures.append( @@ -2439,6 +2449,125 @@ def _retained_authority_binding( return authority_bytes, failures +def _validate_retained_nvattest_binding( + *, + evidence_dir: Path, + ledger: Mapping[str, Any], + digest: str, + ledger_sha256: str, + release_dir: Path, + version: str, +) -> dict[str, str]: + failures: list[Failure] = [] + 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", + ) + ] + ) + + authority_bytes, authority_failures = _retained_authority_binding( + release_dir=release_dir, + ledger=ledger, + ) + failures.extend(authority_failures) + challenge = nvattest.get("challenge") + if not isinstance(challenge, str) or not CHALLENGE_RE.fullmatch(challenge): + failures.append( + _failure( + "retained nvattest challenge is invalid", + expected="64 lowercase hexadecimal characters", + actual=repr(challenge), + repair="bash scripts/release.sh --recover", + ) + ) + + support_distributions = nvattest.get("support_distributions") + support_declaration_failures = validate_support_declarations( + support_distributions, + repair="bash scripts/release.sh --recover", + ) + failures.extend(support_declaration_failures) + if not support_declaration_failures: + failures.extend( + _retained_support_binding_failures(evidence_dir=evidence_dir, ledger=ledger) + ) + if support_declaration_failures or authority_bytes is None: + raise DriverError(failures) + + assert isinstance(support_distributions, Sequence) + hashes: dict[str, str] = {} + for target in PROOF_TARGETS: + path = evidence_dir / "nvattest" / f"{target}.json" + if not path.is_file() or path.is_symlink(): + failures.append( + _failure( + "release nvattest receipt is missing", + expected=f"{target} nvattest receipt", + actual="missing", + repair="bash scripts/release.sh --recover", + ) + ) + continue + data = path.read_bytes() + failures.extend( + validate_nvattest_proof_bytes( + data, + expected_challenge=challenge if isinstance(challenge, str) else "", + 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] + ) + ) + try: + receipt = json.loads(data.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + receipt = None + installed_authority = ( + receipt.get("installed_authority") if isinstance(receipt, Mapping) else None + ) + if isinstance(installed_authority, Mapping) and installed_authority.get( + "size_bytes" + ) != len(authority_bytes): + failures.append( + _failure( + "nvattest receipt installed authority size disagrees with candidate wheels", + expected=str(len(authority_bytes)), + actual=str(installed_authority.get("size_bytes")), + repair="bash scripts/release.sh --recover", + ) + ) + hashes[target] = file_sha256_size(path)[0] + extras = sorted( + path.name + for path in (evidence_dir / "nvattest").glob("*.json") + if path.stem not in PROOF_TARGETS + ) + if extras: + failures.append( + _failure( + "release nvattest set has extra targets", + expected=", ".join(PROOF_TARGETS), + actual=", ".join(extras), + repair="bash scripts/release.sh --recover", + ) + ) + if failures: + raise DriverError(failures) + return hashes + + def _ledger_candidate_file_names( ledger: Mapping[str, Any], ) -> tuple[frozenset[str], list[Failure]]: @@ -3404,15 +3533,6 @@ 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)) return failures @@ -3559,89 +3679,6 @@ 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, @@ -3720,8 +3757,8 @@ def _report( release_dir=release_dir, version=version, ) - nvattest_hashes = _nvattest_hashes( - evidence_dir / "nvattest", + nvattest_hashes = _validate_retained_nvattest_binding( + evidence_dir=evidence_dir, ledger=ledger, digest=digest, ledger_sha256=ledger_sha256, diff --git a/scripts/release_publish.py b/scripts/release_publish.py index 8e5858e1c..554e1b026 100644 --- a/scripts/release_publish.py +++ b/scripts/release_publish.py @@ -13,6 +13,7 @@ tag readiness, and it never invokes git or gh. from __future__ import annotations import argparse +import hashlib import json import logging import os @@ -37,7 +38,12 @@ from scripts.check_rust_release_manifest import ( # noqa: E402 rust_artifact_targets, validate_core_unsupported_tombstone_record, ) -from scripts.release_candidate_driver import CandidateReport, DriverError # noqa: E402 +from scripts.build_nvattest_authority import render_nvattest_authority_json # noqa: E402 +from scripts.release_candidate_driver import ( # noqa: E402 + CandidateReport, + DriverError, + _retained_authority_binding, +) from scripts.release_ledger import read_retained_ledger # noqa: E402 from scripts.transparency_core import ( # noqa: E402 fail_closed, @@ -481,6 +487,49 @@ def _verify_recover(config: PublishConfig) -> CandidateReport: return report +def _assert_nvattest_authority_matches_checkout( + config: PublishConfig, + report: CandidateReport, +) -> None: + authority_bytes, failures = _retained_authority_binding( + release_dir=report.release_dir, + ledger=config.retained_ledger, + ) + if authority_bytes is None and not failures: + failures.append( + failure( + "release publish retained nvattest authority is missing", + expected="retained candidate root wheel authority bytes", + actual="", + repair=( + "bash scripts/release.sh --recover " + f"{config.version} {config.source_commit}" + ), + ) + ) + if failures: + raise DriverError(failures) + + checkout_authority_bytes = render_nvattest_authority_json().encode("utf-8") + if checkout_authority_bytes != authority_bytes: + raise DriverError( + [ + failure( + ( + "release publish checkout nvattest authority does not match " + "retained candidate" + ), + expected=hashlib.sha256(authority_bytes).hexdigest(), + actual=hashlib.sha256(checkout_authority_bytes).hexdigest(), + repair=( + "run the publisher from the release checkout at " + f"{config.source_commit}" + ), + ) + ] + ) + + def _verify_core_unsupported_tombstone_prerequisite(config: PublishConfig) -> None: if config.mode != "production": return @@ -1061,7 +1110,8 @@ def publish_release( max_verify_attempts: int = DEFAULT_VERIFY_ATTEMPTS, verify_sleep_seconds: float = DEFAULT_VERIFY_SLEEP_SECONDS, ) -> PublishResult: - _verify_recover(config) + report = _verify_recover(config) + _assert_nvattest_authority_matches_checkout(config, report) if config.mode == "production": _ensure_source_commit_exists(config, git_runner) _verify_core_unsupported_tombstone_prerequisite(config) diff --git a/scripts/transparency_core.py b/scripts/transparency_core.py index d0a538665..f50b0eb65 100644 --- a/scripts/transparency_core.py +++ b/scripts/transparency_core.py @@ -142,6 +142,10 @@ def version_object_key(product: str, version: str, name: str) -> str: return f"{version_prefix(product, version)}{name}" +def nvattest_public_receipt_name(target: str) -> str: + return f"{target}.nvattest.json" + + def product_prefix(product: str) -> str: return f"releases/{product}/" @@ -985,8 +989,31 @@ def collect_candidate_parts(report: CandidateReport) -> CandidateTransparencyPar ) proofs.append(NamedDigest(name=name, sha256=digest)) version_files[name] = path + nvattest_dir = report.evidence_dir / "nvattest" + for target, digest in sorted(report.nvattest_sha256.items()): + name = nvattest_public_receipt_name(target) + path = nvattest_dir / f"{target}.json" + try: + actual_digest, _actual_bytes = file_sha256_size(path) + except OSError as exc: + fail_closed( + "retained nvattest receipt could not be read", + expected=f"nvattest/{target}.json", + actual=type(exc).__name__, + repair="bash scripts/release.sh --recover ", + ) + if actual_digest != digest: + fail_closed( + "retained nvattest digest does not match candidate report", + expected=digest, + actual=actual_digest, + repair="bash scripts/release.sh --recover ", + ) + proofs.append(NamedDigest(name=name, sha256=digest)) + version_files[name] = path validate_retained_source_clean(report) validate_retained_proof_versions(report, ledger) + validate_retained_nvattest_versions(report, ledger) return CandidateTransparencyParts( artifacts=tuple(sorted(artifacts, key=lambda item: item.name)), manifests=tuple(sorted(manifests, key=lambda item: item.name)), @@ -1075,6 +1102,55 @@ def validate_retained_proof_versions( raise DriverError(failures) +def validate_retained_nvattest_versions( + report: CandidateReport, + ledger: Mapping[str, Any], +) -> None: + expected_targets = ( + ledger.get("proofs", {}).get("expected_targets") + if isinstance(ledger.get("proofs"), Mapping) + else None + ) + expected_target_set = ( + set(expected_targets) if isinstance(expected_targets, list) else set() + ) + failures: list[Failure] = [] + for path in sorted((report.evidence_dir / "nvattest").glob("*.json")): + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + failures.append( + failure( + "retained nvattest receipt could not be read", + expected=path.name, + actual=type(exc).__name__, + repair="bash scripts/release.sh --recover ", + ) + ) + continue + target = payload.get("target") + if target not in expected_target_set: + failures.append( + failure( + "retained nvattest target is stale or unexpected", + expected=", ".join(sorted(expected_target_set)), + actual=repr(target), + repair="bash scripts/release.sh --recover ", + ) + ) + if payload.get("version") != report.version: + failures.append( + failure( + "retained nvattest version is stale", + expected=report.version, + actual=repr(payload.get("version")), + repair="bash scripts/release.sh --recover ", + ) + ) + if failures: + raise DriverError(failures) + + def atomic_write(path: Path, data: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True) temp_path = path.with_name(f".{path.name}.{os.getpid()}.tmp") diff --git a/tests/helpers/release_candidate_fixtures.py b/tests/helpers/release_candidate_fixtures.py index e6dc463f7..21cec5c50 100644 --- a/tests/helpers/release_candidate_fixtures.py +++ b/tests/helpers/release_candidate_fixtures.py @@ -678,6 +678,7 @@ def _nvattest_services( candidate_paths: Sequence[Path], support_paths: Sequence[Path], canonical_authority_bytes: bytes, + call_counts: dict[str, int] | None = None, ) -> nvattest_proof.NvattestProofServices: env_root = root / "nvattest-env" / target policy_os, policy_arch = TARGET_POLICY[target] @@ -687,7 +688,12 @@ def _nvattest_services( for entry in expected_distribution_entries(candidate_paths) ) + def count(name: str) -> None: + if call_counts is not None: + call_counts[name] = call_counts.get(name, 0) + 1 + def create_environment(_target: str) -> Path: + count("nvattest_create_environment") (env_root / "bin").mkdir(parents=True, exist_ok=True) python = env_root / "bin" / "python" python.write_text( @@ -702,6 +708,7 @@ def _nvattest_services( candidate_wheels: Sequence[Path], support_wheels: Sequence[Path], ) -> CommandResult: + count("nvattest_install_wheels") assert tuple(candidate_wheels) == tuple(candidate_paths) assert tuple(support_wheels) == tuple(support_paths) site_root = env_root / "lib" / "python3.13" / "site-packages" @@ -732,6 +739,7 @@ def _nvattest_services( ) def fetch(label: str, url: str, dest: Path) -> nvattest_proof.FetchObservation: + count("nvattest_fetch") dest.parent.mkdir(parents=True, exist_ok=True) if label == "archive": artifact = authority_target["artifact"] @@ -758,6 +766,7 @@ def _nvattest_services( target_key: str, journal_path: Path, ) -> nvattest_proof.DriverObservation: + count("nvattest_run_package_install") assert target_key == _nvattest_target_key(target) payload = _nvattest_driver_payload( target=target, @@ -781,14 +790,17 @@ def _nvattest_services( ) def run_smoke(nvattest_bin: Path) -> CommandResult: + count("nvattest_run_smoke") 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: { + def integrity_recheck( + _journal: Path, + _target_key: str, + _fetches: Mapping[str, nvattest_proof.FetchObservation], + driver_observation: nvattest_proof.DriverObservation, + ) -> dict[str, Any]: + count("nvattest_integrity_recheck") + return { "members": driver_observation.payload["members"], "sidecar": driver_observation.payload["sidecar"], "sidecar_path": driver_observation.payload["sidecar_path"], @@ -797,20 +809,43 @@ def _nvattest_services( "tree_fingerprint_sha256": driver_observation.payload[ "tree_fingerprint_sha256" ], - }, + } + + def clock() -> datetime: + count("nvattest_clock") + return datetime(2026, 7, 20, 12, 35, tzinfo=UTC) + + def cleanup(path: Path) -> None: + count("nvattest_cleanup") + shutil.rmtree(path) + + def observe_host() -> nvattest_proof.HostObservation: + count("nvattest_observe_host") + return nvattest_proof.HostObservation(os=policy_os, arch=policy_arch) + + return nvattest_proof.NvattestProofServices( + create_environment=create_environment, + install_wheels=install_wheels, + fetch=fetch, + run_package_install=run_package_install, + integrity_recheck=integrity_recheck, 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 - ), + clock=clock, + cleanup=cleanup, + observe_host=observe_host, ) def services( root: Path, *, native_mutation: str | None = None ) -> driver.CandidateServices: + call_counts: dict[str, int] = {} + + def count(name: str) -> None: + call_counts[name] = call_counts.get(name, 0) + 1 + def clean_outputs(repo_root: Path, version: str) -> None: + count("clean_outputs") for relative in ( "build", "dist", @@ -825,6 +860,7 @@ def services( path.unlink() def build_local_dist(repo_root: Path, include_models: bool) -> None: + count("build_local_dist") dist = repo_root / "dist" dist.mkdir(parents=True, exist_ok=True) for name in driver._expected_local_dist_names(include_models=include_models): @@ -847,6 +883,7 @@ def services( def create_source_bundle( _repo: Path, commit: str, output_path: Path ) -> SourceBundle: + count("create_source_bundle") assert commit == SOURCE_COMMIT output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_bytes(b"bundle") @@ -860,6 +897,7 @@ def services( def build_host( source_bundle: SourceBundle, commit: str, output_dir: Path ) -> BuildHostResult: + count("build_host") assert source_bundle.path.read_bytes() == b"bundle" assert source_bundle.source_commit == SOURCE_COMMIT assert source_bundle.sha256 == hashlib.sha256(b"bundle").hexdigest() @@ -868,6 +906,7 @@ def services( return write_macos_host_outputs(output_dir, mutate=native_mutation) def materialize_support_wheels(destination: Path) -> tuple[Path, ...]: + count("materialize_support_wheels") entries = read_support_lock_entries(root / "uv.lock") return tuple( write_support_wheel(destination, name=entry.name, version=entry.version) @@ -875,6 +914,7 @@ def services( ) def run_target_proofs(**kwargs: Any) -> TargetProofPaths: + count("run_target_proofs") output_path = Path(kwargs["output_path"]) target = str(kwargs["target"]) install_paths = target_install_paths_from_ledger( @@ -938,37 +978,73 @@ def services( candidate_paths=install_paths, support_paths=support_paths, canonical_authority_bytes=NVATTEST_AUTHORITY_BYTES, + call_counts=call_counts, ), canonical_authority_bytes=NVATTEST_AUTHORITY_BYTES, ) return TargetProofPaths(install=install_receipt, nvattest=nvattest_receipt) def cleanup(paths: Sequence[Path]) -> None: + count("cleanup_transients") for path in paths: if path.is_dir(): shutil.rmtree(path) elif path.exists() or path.is_symlink(): path.unlink() - return driver.CandidateServices( - git_head=lambda _repo: SOURCE_COMMIT, - git_status=lambda _repo: "", - core_lock_sha256=lambda _repo: LOCK_SHA, - clean_outputs=clean_outputs, - build_local_dist=build_local_dist, - prepare_policy=lambda _repo, _env: _policy(), - coordinator_tool_evidence=lambda: { + def git_head(_repo: Path) -> str: + count("git_head") + return SOURCE_COMMIT + + def git_status(_repo: Path) -> str: + count("git_status") + return "" + + def core_lock_sha256(_repo: Path) -> str: + count("core_lock_sha256") + return LOCK_SHA + + def prepare_policy(_repo: Path, _env: Mapping[str, str]) -> PolicyRun: + count("prepare_policy") + return _policy() + + def coordinator_tool_evidence() -> Mapping[str, Mapping[str, str]]: + count("coordinator_tool_evidence") + return { lane: pins.fixture_lane_tool_evidence(lane) for lane in ("source", "linux-x86_64-musl", "linux-aarch64-musl") - }, + } + + def challenge_factory() -> str: + count("challenge_factory") + return hashlib.sha256(str(root).encode("utf-8")).hexdigest() + + def transaction_hook(_point: str) -> None: + count("transaction_hook") + + def reset_call_counts() -> None: + for name in tuple(call_counts): + call_counts[name] = 0 + + service = driver.CandidateServices( + git_head=git_head, + git_status=git_status, + core_lock_sha256=core_lock_sha256, + clean_outputs=clean_outputs, + build_local_dist=build_local_dist, + prepare_policy=prepare_policy, + coordinator_tool_evidence=coordinator_tool_evidence, create_source_bundle=create_source_bundle, build_host=build_host, cleanup_transients=cleanup, - challenge_factory=lambda: hashlib.sha256(str(root).encode("utf-8")).hexdigest(), + challenge_factory=challenge_factory, materialize_support_wheels=materialize_support_wheels, run_target_proofs=run_target_proofs, - transaction_hook=lambda _point: None, + transaction_hook=transaction_hook, ) + object.__setattr__(service, "call_counts", call_counts) + object.__setattr__(service, "reset_call_counts", reset_call_counts) + return service def recover(root: Path) -> driver.CandidateReport: diff --git a/tests/test_release_candidate_driver.py b/tests/test_release_candidate_driver.py index 6639037de..79ed61152 100644 --- a/tests/test_release_candidate_driver.py +++ b/tests/test_release_candidate_driver.py @@ -3,6 +3,7 @@ from __future__ import annotations +import copy import hashlib import inspect import json @@ -415,6 +416,161 @@ def _real_candidate(tmp_path: Path) -> tuple[Path, driver.CandidateReport]: return root, driver.run_candidate(root, _env(), _services(root)) +def _read_json(path: Path) -> dict[str, Any]: + payload = json.loads(path.read_text(encoding="utf-8")) + assert isinstance(payload, dict) + return payload + + +def _write_json(path: Path, payload: Mapping[str, Any]) -> None: + path.write_bytes(checker.canonical_json_bytes(payload)) + + +def _ledger_path(report: driver.CandidateReport) -> Path: + return report.evidence_dir / "ledger.json" + + +def _write_ledger(report: driver.CandidateReport, payload: Mapping[str, Any]) -> str: + path = _ledger_path(report) + _write_json(path, payload) + return driver.file_sha256_size(path)[0] + + +def _nvattest_receipt_path(report: driver.CandidateReport, target: str) -> Path: + return report.evidence_dir / "nvattest" / f"{target}.json" + + +def _proof_receipt_path(report: driver.CandidateReport, target: str) -> Path: + return report.evidence_dir / "proofs" / f"{target}.json" + + +def _update_retained_receipt_ledger_sha( + report: driver.CandidateReport, ledger_sha256: str +) -> None: + for target in driver.PROOF_TARGETS: + for path in ( + _proof_receipt_path(report, target), + _nvattest_receipt_path(report, target), + ): + payload = _read_json(path) + payload["ledger_sha256"] = ledger_sha256 + _write_json(path, payload) + + +def _assert_fails_with_error(root: Path, expected_error: str) -> None: + with pytest.raises(driver.DriverError) as exc: + _recover(root) + + errors = [failure.error for failure in exc.value.failures] + assert expected_error in errors, errors + + +def _reset_service_call_counts(services: driver.CandidateServices) -> None: + services.reset_call_counts() + + +def _assert_service_call_counts_zero(services: driver.CandidateServices) -> None: + counts = dict(services.call_counts) + assert counts + assert counts == {name: 0 for name in counts} + + +def _assert_recovery_failure_preserves_retained_tree( + root: Path, + report: driver.CandidateReport, + expected_error: str, + services: driver.CandidateServices, +) -> None: + before_payload = _structural_snapshot(report.release_dir) + before_evidence = _structural_snapshot(report.evidence_dir) + _reset_service_call_counts(services) + + _assert_fails_with_error(root, expected_error) + + assert _structural_snapshot(report.release_dir) == before_payload + assert _structural_snapshot(report.evidence_dir) == before_evidence + _assert_service_call_counts_zero(services) + + +def _retained_support_declarations( + report: driver.CandidateReport, +) -> list[dict[str, Any]]: + support_paths = tuple( + path.resolve() + for path in sorted((report.evidence_dir / "support").glob("*.whl")) + ) + declarations = driver.support_distribution_entries(support_paths) + assert ( + {entry["name"] for entry in declarations} + == driver.SUPPORT_DISTRIBUTION_NAMES + ) + return declarations + + +def _sync_nvattest_receipt_with_ledger( + receipt: dict[str, Any], + *, + nvattest: Mapping[str, Any], + ledger_sha256: str, +) -> None: + authority = nvattest["authority"] + assert isinstance(authority, Mapping) + authority_bytes = driver._canonical_nvattest_authority_bytes(authority) + host = receipt["host"] + assert isinstance(host, Mapping) + target_key = host["authority_target_key"] + authority_targets = authority["targets"] + assert isinstance(authority_targets, Mapping) + authority_target = authority_targets[target_key] + assert isinstance(authority_target, Mapping) + source = authority_target["source"] + artifact = authority_target["artifact"] + companion_manifest = authority_target["companion_manifest"] + assert isinstance(source, Mapping) + assert isinstance(artifact, Mapping) + assert isinstance(companion_manifest, Mapping) + + receipt["challenge"] = nvattest["challenge"] + receipt["ledger_sha256"] = ledger_sha256 + receipt["support_distributions"] = copy.deepcopy(nvattest["support_distributions"]) + receipt["installed_authority"]["sha256"] = nvattest["authority_sha256"] + receipt["installed_authority"]["size_bytes"] = len(authority_bytes) + receipt["nvattest"] = { + "artifact": { + "name": artifact["name"], + "sha256": artifact["sha256"], + "size_bytes": artifact["size_bytes"], + "url": artifact["url"], + }, + "companion_manifest": { + "name": companion_manifest["name"], + "sha256": companion_manifest["sha256"], + "url": companion_manifest["url"], + }, + "source": { + "fork_commit": source["fork_commit"], + "upstream_base": source["upstream_base"], + "version": source["version"], + }, + "target_key": target_key, + } + receipt["archive_fetch"] = { + "sha256": artifact["sha256"], + "size_bytes": artifact["size_bytes"], + "url": artifact["url"], + } + receipt["manifest_fetch"]["sha256"] = companion_manifest["sha256"] + receipt["manifest_fetch"]["url"] = companion_manifest["url"] + receipt["companion_manifest"]["sha256"] = companion_manifest["sha256"] + receipt["companion_manifest"]["target_key"] = target_key + receipt["integrity"]["sidecar"]["artifact"] = dict(artifact) + receipt["integrity"]["sidecar"]["target_key"] = target_key + receipt["integrity"]["sidecar"]["version"] = source["version"] + sidecar_bytes = checker.canonical_json_bytes(receipt["integrity"]["sidecar"]) + receipt["integrity"]["sidecar_sha256"] = hashlib.sha256(sidecar_bytes).hexdigest() + receipt["integrity"]["sidecar_size_bytes"] = len(sidecar_bytes) + + def _write_directory_sentinel(path: Path) -> None: marker = path / "inside" / "marker.txt" marker.parent.mkdir(parents=True, exist_ok=True) @@ -2968,3 +3124,203 @@ def test_recovery_rejects_self_consistent_native_member_forgery( exc.value.failures[0].error == "retained ledger native_members do not match finalized wheels" ) + + +def test_recovery_rejects_self_consistent_nvattest_authority_forgery( + tmp_path: Path, +) -> None: + root = _repo(tmp_path) + report = driver.run_candidate(root, _env(), _services(root)) + before_payload = _structural_snapshot(report.release_dir) + ledger = _read_json(_ledger_path(report)) + nvattest = ledger["nvattest"] + forged_authority = copy.deepcopy(nvattest["authority"]) + assert isinstance(forged_authority, dict) + for target_key, authority_target in sorted(forged_authority["targets"].items()): + authority_target["artifact"]["sha256"] = hashlib.sha256( + f"forged authority {target_key}".encode("utf-8") + ).hexdigest() + forged_authority_bytes = driver._canonical_nvattest_authority_bytes( + forged_authority + ) + forged_challenge = hashlib.sha256( + f"forged challenge {report.version}".encode("utf-8") + ).hexdigest() + assert CHALLENGE_RE.fullmatch(forged_challenge) + nvattest["challenge"] = forged_challenge + nvattest["authority"] = forged_authority + nvattest["authority_sha256"] = hashlib.sha256(forged_authority_bytes).hexdigest() + nvattest["support_distributions"] = _retained_support_declarations(report) + forged_ledger_sha = _write_ledger(report, ledger) + _update_retained_receipt_ledger_sha(report, forged_ledger_sha) + + for target in driver.PROOF_TARGETS: + receipt_path = _nvattest_receipt_path(report, target) + receipt = _read_json(receipt_path) + _sync_nvattest_receipt_with_ledger( + receipt, + nvattest=nvattest, + ledger_sha256=forged_ledger_sha, + ) + _write_json(receipt_path, receipt) + + assert _structural_snapshot(report.release_dir) == before_payload + _assert_fails_with_error( + root, + "retained nvattest authority disagrees with candidate wheels", + ) + + +def test_recovery_success_preserves_retained_tree_and_uses_no_seams( + tmp_path: Path, +) -> None: + root = _repo(tmp_path) + services = _services(root) + report = driver.run_candidate(root, _env(), services) + before_payload = _structural_snapshot(report.release_dir) + before_evidence = _structural_snapshot(report.evidence_dir) + _reset_service_call_counts(services) + + recovered = _recover(root) + + assert recovered.heading == "retained-candidate-valid" + assert _structural_snapshot(report.release_dir) == before_payload + assert _structural_snapshot(report.evidence_dir) == before_evidence + _assert_service_call_counts_zero(services) + + +def test_recovery_failure_preserves_retained_tree_and_uses_no_seams( + tmp_path: Path, +) -> None: + root = _repo(tmp_path) + services = _services(root) + report = driver.run_candidate(root, _env(), services) + _nvattest_receipt_path(report, driver.PROOF_TARGETS[0]).unlink() + + _assert_recovery_failure_preserves_retained_tree( + root, + report, + "release nvattest inventory is not exact", + services, + ) + + +@pytest.mark.parametrize( + ("mutation", "expected_error"), + [ + ("extra", "release nvattest inventory is not exact"), + ("duplicate", "nvattest proof target is not bound to expected input"), + ("swapped", "nvattest proof target is not bound to expected input"), + ("stale", "nvattest proof version is not bound to expected input"), + ("replayed", "nvattest proof challenge is not bound to expected input"), + ("mutated", "nvattest proof kind is invalid"), + ("noncanonical", "nvattest proof bytes are not canonical"), + ], +) +def test_recovery_rejects_retained_nvattest_receipt_mutations( + tmp_path: Path, + mutation: str, + expected_error: str, +) -> None: + root = _repo(tmp_path) + report = driver.run_candidate(root, _env(), _services(root)) + targets = tuple(driver.PROOF_TARGETS) + first_target = targets[0] + second_target = targets[1] + first_path = _nvattest_receipt_path(report, first_target) + second_path = _nvattest_receipt_path(report, second_target) + + if mutation == "extra": + extra_path = report.evidence_dir / "nvattest" / "extra.json" + extra_path.write_bytes(first_path.read_bytes()) + elif mutation == "duplicate": + second_path.write_bytes(first_path.read_bytes()) + elif mutation == "swapped": + first_bytes = first_path.read_bytes() + second_bytes = second_path.read_bytes() + first_path.write_bytes(second_bytes) + second_path.write_bytes(first_bytes) + elif mutation == "stale": + receipt = _read_json(first_path) + receipt["version"] = f"{report.version}+stale" + _write_json(first_path, receipt) + elif mutation == "replayed": + replay_root = _repo(tmp_path / "replayed") + replay_report = driver.run_candidate( + replay_root, _env(), _services(replay_root) + ) + replayed_challenge = _read_json(_ledger_path(replay_report))["nvattest"][ + "challenge" + ] + assert CHALLENGE_RE.fullmatch(replayed_challenge) + assert replayed_challenge != _read_json(_ledger_path(report))["nvattest"][ + "challenge" + ] + receipt = _read_json(first_path) + receipt["challenge"] = replayed_challenge + _write_json(first_path, receipt) + elif mutation == "mutated": + receipt = _read_json(first_path) + receipt["kind"] = "mutated-nvattest-receipt" + _write_json(first_path, receipt) + elif mutation == "noncanonical": + receipt = _read_json(first_path) + first_path.write_text( + json.dumps(receipt, indent=2, sort_keys=True), encoding="utf-8" + ) + else: + raise AssertionError(f"unknown mutation {mutation}") + + _assert_fails_with_error(root, expected_error) + + +def test_recovery_rejects_retained_nvattest_wrong_challenge( + tmp_path: Path, +) -> None: + root = _repo(tmp_path) + report = driver.run_candidate(root, _env(), _services(root)) + ledger = _read_json(_ledger_path(report)) + invalid_challenge = "not-a-valid-nvattest-challenge" + assert not CHALLENGE_RE.fullmatch(invalid_challenge) + ledger["nvattest"]["challenge"] = invalid_challenge + ledger_sha = _write_ledger(report, ledger) + _update_retained_receipt_ledger_sha(report, ledger_sha) + + _assert_fails_with_error(root, "retained ledger nvattest challenge is invalid") + + +def test_recovery_rejects_retained_nvattest_support_wheel_byte_mutation( + tmp_path: Path, +) -> None: + root = _repo(tmp_path) + report = driver.run_candidate(root, _env(), _services(root)) + support_path = sorted((report.evidence_dir / "support").glob("*.whl"))[0] + support_path.write_bytes(support_path.read_bytes() + b"\nmutated support bytes\n") + + _assert_fails_with_error( + root, + "retained nvattest support bytes disagree with ledger", + ) + + +def test_recovery_rejects_retained_nvattest_support_declaration_mutation( + tmp_path: Path, +) -> None: + root = _repo(tmp_path) + report = driver.run_candidate(root, _env(), _services(root)) + ledger = _read_json(_ledger_path(report)) + declarations = ledger["nvattest"]["support_distributions"] + assert ( + {entry["name"] for entry in declarations} + == driver.SUPPORT_DISTRIBUTION_NAMES + ) + declarations[0]["sha256"] = hashlib.sha256( + f"forged support declaration {declarations[0]['name']}".encode("utf-8") + ).hexdigest() + ledger_sha = _write_ledger(report, ledger) + _update_retained_receipt_ledger_sha(report, ledger_sha) + + _assert_fails_with_error( + root, + "retained nvattest support bytes disagree with ledger", + ) diff --git a/tests/test_release_publish.py b/tests/test_release_publish.py index 8d74efcf7..c7630b808 100644 --- a/tests/test_release_publish.py +++ b/tests/test_release_publish.py @@ -8,8 +8,9 @@ import json import logging import subprocess import sys +import zipfile from collections.abc import Callable, Mapping, Sequence -from dataclasses import replace +from dataclasses import dataclass, replace from pathlib import Path from typing import Any @@ -18,6 +19,7 @@ import pytest import scripts.check_rust_release_manifest as manifest import scripts.check_wheel_contents as wheel_checker import scripts.release_publish as publisher +from scripts.build_nvattest_authority import render_nvattest_authority_json from scripts.release_candidate_driver import ( CandidateReport, DriverError, @@ -25,6 +27,7 @@ from scripts.release_candidate_driver import ( from scripts.release_candidate_driver import ( run_candidate as run_release_candidate, ) +from scripts.release_install_smoke import PROOF_TARGETS from scripts.transparency_core import failure from tests.helpers.release_candidate_fixtures import ( MACOS_ONNXRUNTIME, @@ -46,11 +49,6 @@ from tests.helpers.release_candidate_fixtures import ( SOURCE_COMMIT = "0123456789abcdef0123456789abcdef01234567" OTHER_COMMIT = "fedcba9876543210fedcba9876543210fedcba98" TOKEN = "pypi-canary-token" -PROOF_TARGETS = ( - "linux-x86_64-musl", - "linux-aarch64-musl", - "macos-arm64", -) @pytest.fixture(autouse=True) @@ -119,6 +117,24 @@ def _manifest_names() -> list[str]: ] +def _checkout_authority_bytes() -> bytes: + return render_nvattest_authority_json().encode("utf-8") + + +def _write_publish_fixture_file(path: Path, content: bytes) -> None: + if path.name.startswith("solstone-") and path.name.endswith(".whl"): + info = zipfile.ZipInfo( + wheel_checker.NVATTEST_AUTHORITY_MEMBER, + (2026, 7, 20, 12, 0, 0), + ) + info.create_system = 3 + info.external_attr = 0o644 << 16 + with zipfile.ZipFile(path, "w") as wheel: + wheel.writestr(info, _checkout_authority_bytes()) + return + path.write_bytes(content) + + def _candidate( root: Path, *, @@ -131,6 +147,7 @@ def _candidate( evidence_dir = root / "target" / "release-evidence" / candidate_version release_dir.mkdir(parents=True, exist_ok=True) (evidence_dir / "proofs").mkdir(parents=True, exist_ok=True) + (evidence_dir / "nvattest").mkdir(parents=True, exist_ok=True) names = [ *publisher.expected_package_names(include_models=include_models), @@ -139,12 +156,17 @@ def _candidate( if unknown_name is not None: names.append(unknown_name) for name in names: - (release_dir / name).write_bytes(f"retained bytes for {name}\n".encode()) + _write_publish_fixture_file( + release_dir / name, + f"retained bytes for {name}\n".encode(), + ) files: list[dict[str, Any]] = [] for path in sorted(release_dir.iterdir(), key=lambda item: item.name): digest, byte_count = _sha(path) files.append({"bytes": byte_count, "name": path.name, "sha256": digest}) + authority_bytes = _checkout_authority_bytes() + authority = json.loads(authority_bytes.decode("utf-8")) ledger = { "candidate": { @@ -161,6 +183,14 @@ def _candidate( "decision": "include" if include_models else "exclude", "package_version": _models_version(), }, + "nvattest": { + "authority": authority, + "authority_sha256": hashlib.sha256(authority_bytes).hexdigest(), + "challenge": hashlib.sha256( + f"publish fixture challenge {candidate_version}".encode("utf-8") + ).hexdigest(), + "support_distributions": [], + }, "product": "solstone", "proofs": {"expected_targets": list(PROOF_TARGETS)}, "source_commit": SOURCE_COMMIT, @@ -180,7 +210,12 @@ def _candidate( encoding="utf-8", ) proof_hashes[target] = _sha(path)[0] - nvattest_hashes[target] = "0" * 64 + nvattest_path = evidence_dir / "nvattest" / f"{target}.json" + nvattest_path.write_text( + json.dumps({"target": target, "version": candidate_version}), + encoding="utf-8", + ) + nvattest_hashes[target] = _sha(nvattest_path)[0] return CandidateReport( heading="retained-candidate-valid", @@ -216,6 +251,20 @@ def _write_ledger(report: CandidateReport, ledger: Mapping[str, Any]) -> None: ) +def _mutate_first_authority_target(authority: Mapping[str, Any], label: str) -> None: + targets = authority["targets"] + first_key = sorted(targets)[0] + targets[first_key]["artifact"]["sha256"] = hashlib.sha256( + f"{label} {first_key}".encode("utf-8") + ).hexdigest() + + +def _divergent_checkout_authority_json() -> str: + authority = json.loads(render_nvattest_authority_json()) + _mutate_first_authority_target(authority, "checkout authority divergence") + return json.dumps(authority, indent=2, sort_keys=True) + "\n" + + def _config( root: Path, report: CandidateReport, @@ -502,6 +551,93 @@ def _gh_runner(calls: list[str], *, fail: bool = False) -> publisher.ProcessRunn return run +@dataclass +class PublishSeamCounters: + index_calls: int = 0 + upload_calls: int = 0 + git_calls: int = 0 + gh_calls: int = 0 + + def index_client( + self, + _base_url: str, + projects: Sequence[publisher.ProjectExpectation], + ) -> Mapping[tuple[str, str], Mapping[str, Any] | None]: + self.index_calls += 1 + return _empty_snapshot(projects) + + def upload_runner( + self, + argv: Sequence[str], + *, + cwd: Path, + env: Mapping[str, str], + capture_output: bool, + text: bool, + check: bool, + ) -> subprocess.CompletedProcess[str]: + self.upload_calls += 1 + return subprocess.CompletedProcess(list(argv), 0, stdout="", stderr="") + + def git_runner( + self, + argv: Sequence[str], + *, + cwd: Path, + capture_output: bool, + text: bool, + check: bool, + ) -> subprocess.CompletedProcess[str]: + self.git_calls += 1 + return subprocess.CompletedProcess( + list(argv), + 0, + stdout=SOURCE_COMMIT, + stderr="", + ) + + def gh_runner( + self, + argv: Sequence[str], + *, + cwd: Path, + capture_output: bool, + text: bool, + check: bool, + ) -> subprocess.CompletedProcess[str]: + self.gh_calls += 1 + return subprocess.CompletedProcess(list(argv), 0, stdout="", stderr="") + + def assert_zero(self) -> None: + assert self.index_calls == 0 + assert self.upload_calls == 0 + assert self.git_calls == 0 + assert self.gh_calls == 0 + + +def _patch_late_publish_steps_forbidden( + monkeypatch: pytest.MonkeyPatch, + calls: list[str], +) -> None: + def forbidden(name: str) -> Callable[..., Any]: + def run(*_args: Any, **_kwargs: Any) -> Any: + calls.append(name) + raise AssertionError(f"{name} must not be invoked") + + return run + + for name in ( + "classify_candidate_artifacts", + "_read_changelog_block", + "_index_matches", + "_run_twine_upload", + "_verify_uploaded_index", + "publish_git_tag", + "record_github_release_witness", + ): + monkeypatch.setattr(publisher, name, forbidden(name)) + + def _forbidden_runner(label: str, calls: list[str]) -> publisher.ProcessRunner: def run(*_args: Any, **_kwargs: Any) -> subprocess.CompletedProcess[str]: calls.append(label) @@ -532,6 +668,22 @@ def _run_publish( ) +def _run_publish_with_counted_seams( + config: publisher.PublishConfig, + seams: PublishSeamCounters, +) -> publisher.PublishResult: + return publisher.publish_release( + config=config, + index_client=seams.index_client, + upload_runner=seams.upload_runner, + git_runner=seams.git_runner, + gh_runner=seams.gh_runner, + sleep=lambda _seconds: None, + max_verify_attempts=1, + verify_sleep_seconds=0, + ) + + def _first_failure(error: DriverError) -> str: return error.failures[0].error @@ -977,6 +1129,98 @@ def test_recover_heading_must_be_retained_valid( assert calls == [] +@pytest.mark.parametrize("mode", ("production", "test")) +def test_recovery_failure_short_circuits_before_publisher_seams( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mode: publisher.Mode, +) -> None: + report = _candidate(tmp_path) + config = _config(tmp_path, report, mode=mode) + late_calls: list[str] = [] + _patch_late_publish_steps_forbidden(monkeypatch, late_calls) + + def recover(_root: Path, *, version: str, source_commit: str) -> CandidateReport: + raise DriverError( + [ + failure( + "release publish recovery failed", + expected=version, + actual=source_commit, + repair="bash scripts/release.sh --recover", + ) + ] + ) + + monkeypatch.setattr(publisher, "recover_candidate", recover) + seams = PublishSeamCounters() + + with pytest.raises(DriverError) as excinfo: + _run_publish_with_counted_seams(config, seams) + + assert _first_failure(excinfo.value) == "release publish recovery failed" + seams.assert_zero() + assert late_calls == [] + + +@pytest.mark.parametrize("mode", ("production", "test")) +def test_retained_authority_binding_failure_short_circuits_before_publisher_seams( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mode: publisher.Mode, +) -> None: + report = _candidate(tmp_path) + ledger = _ledger(report) + _mutate_first_authority_target( + ledger["nvattest"]["authority"], + "retained authority divergence", + ) + _write_ledger(report, ledger) + _patch_recover(monkeypatch, report, rehash_payloads=True) + late_calls: list[str] = [] + _patch_late_publish_steps_forbidden(monkeypatch, late_calls) + seams = PublishSeamCounters() + + with pytest.raises(DriverError) as excinfo: + _run_publish_with_counted_seams( + _config(tmp_path, report, mode=mode, ledger=ledger), + seams, + ) + + assert _first_failure(excinfo.value) == ( + "retained nvattest authority disagrees with candidate wheels" + ) + seams.assert_zero() + assert late_calls == [] + + +@pytest.mark.parametrize("mode", ("production", "test")) +def test_checkout_authority_divergence_short_circuits_before_publisher_seams( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + mode: publisher.Mode, +) -> None: + report = _candidate(tmp_path) + _patch_recover(monkeypatch, report, rehash_payloads=True) + monkeypatch.setattr( + publisher, + "render_nvattest_authority_json", + _divergent_checkout_authority_json, + ) + late_calls: list[str] = [] + _patch_late_publish_steps_forbidden(monkeypatch, late_calls) + seams = PublishSeamCounters() + + with pytest.raises(DriverError) as excinfo: + _run_publish_with_counted_seams(_config(tmp_path, report, mode=mode), seams) + + assert _first_failure(excinfo.value) == ( + "release publish checkout nvattest authority does not match retained candidate" + ) + seams.assert_zero() + assert late_calls == [] + + def test_unknown_asset_class_prevents_transport( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/test_transparency_publish.py b/tests/test_transparency_publish.py index 8942d4c2d..84965639f 100644 --- a/tests/test_transparency_publish.py +++ b/tests/test_transparency_publish.py @@ -4,6 +4,7 @@ import hashlib import json import logging import subprocess +from dataclasses import replace from datetime import UTC, datetime from pathlib import Path from typing import Any @@ -12,6 +13,8 @@ import pytest import scripts.transparency_publish as publisher from scripts.release_candidate_driver import CandidateReport, DriverError +from scripts.release_install_smoke import PROOF_TARGETS +from scripts.release_public_evidence import validate_public_evidence_tree from scripts.transparency_core import ( ENTRY_OBJECT_NAME, ENTRY_SIGNATURE_NAME, @@ -28,11 +31,13 @@ from scripts.transparency_core import ( build_latest_pointer, build_ledger_entry, canonical_json_bytes, + collect_candidate_parts, entry_trusted_comment, latest_key, latest_signature_key, latest_trusted_comment, ledger_key, + nvattest_public_receipt_name, parse_latest_bytes, parse_ledger_entry_bytes, sha256_bytes, @@ -42,7 +47,6 @@ from scripts.transparency_head_log import HeadLogRow, append_head_row from scripts.transparency_signing import FakeTransparencySigner from scripts.transparency_transport import DirectoryTransparencyTransport, HttpResult -PROOF_TARGETS = ("linux-aarch64-musl", "linux-x86_64-musl", "macos-arm64") SOURCE_COMMIT = "a" * 40 FIXTURE_DIR = Path(__file__).parent / "fixtures" / "transparency" STAGING_MANIFEST_SHA256 = ( @@ -70,12 +74,14 @@ def _candidate( *, version: str = "0.9.1", proof_version: str | None = None, + nvattest_version: str | None = None, dirty: bool = False, ) -> CandidateReport: release_dir = root / "dist" / "release-candidate" / version evidence_dir = root / "target" / "release-evidence" / version release_dir.mkdir(parents=True, exist_ok=True) (evidence_dir / "proofs").mkdir(parents=True, exist_ok=True) + (evidence_dir / "nvattest").mkdir(parents=True, exist_ok=True) package_names = [f"package-{index}.whl" for index in range(11)] manifest_names = [ "solstone-core.rust-release-manifest.json", @@ -125,7 +131,13 @@ def _candidate( 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 + nvattest = {"target": target, "version": nvattest_version or version} + nvattest_path = evidence_dir / "nvattest" / f"{target}.json" + nvattest_path.write_text( + json.dumps(nvattest, sort_keys=True), + encoding="utf-8", + ) + nvattest_hashes[target] = _sha(nvattest_path)[0] return CandidateReport( heading="retained-candidate-valid", version=version, @@ -140,6 +152,13 @@ def _candidate( ) +def _refresh_nvattest_digest(report: CandidateReport, target: str) -> CandidateReport: + nvattest_sha256 = dict(report.nvattest_sha256) + receipt = report.evidence_dir / "nvattest" / f"{target}.json" + nvattest_sha256[target] = _sha(receipt)[0] + return replace(report, nvattest_sha256=nvattest_sha256) + + def _patch_recover(monkeypatch: pytest.MonkeyPatch, *, version: str = "0.9.1") -> None: def recover(root: Path, *, version: str, source_commit: str) -> CandidateReport: assert source_commit == SOURCE_COMMIT @@ -511,17 +530,18 @@ def test_staging_payload_is_archive_superset( artifact_names = {path.name for path in (stage.payload_dir / "artifacts").iterdir()} assert artifact_names == {f"package-{index}.whl" for index in range(11)} evidence_names = {path.name for path in stage.version_dir.iterdir()} - assert { + expected_receipts = {f"{target}.json" for target in PROOF_TARGETS} | { + nvattest_public_receipt_name(target) for target in PROOF_TARGETS + } + expected_evidence = { ENTRY_OBJECT_NAME, ENTRY_SIGNATURE_NAME, - "linux-aarch64-musl.json", - "linux-x86_64-musl.json", - "macos-arm64.json", "solstone-core.rust-release-manifest.json", "solstone-journal-cuda.rust-release-manifest.json", "solstone-journal-models.rust-release-manifest.json", "solstone-journal.rust-release-manifest.json", - }.issubset(evidence_names) + } | expected_receipts + assert expected_evidence <= evidence_names assert (stage.payload_dir / LEDGER_OBJECT_NAME).is_file() assert (stage.payload_dir / LATEST_OBJECT_NAME).is_file() assert (stage.payload_dir / LATEST_SIGNATURE_NAME).is_file() @@ -552,9 +572,36 @@ def test_built_entry_inventory_matches_retained_rail_ledger( for item in candidate_files if item["name"].endswith(".rust-release-manifest.json") } - assert {(item["name"], item["sha256"]) for item in entry["proofs"]} == { + expected_proofs = { (f"{target}.json", digest) for target, digest in report.proof_sha256.items() + } | { + (nvattest_public_receipt_name(target), digest) + for target, digest in report.nvattest_sha256.items() } + assert {(item["name"], item["sha256"]) for item in entry["proofs"]} == ( + expected_proofs + ) + assert len(entry["proofs"]) == len(PROOF_TARGETS) * 2 + + +def test_install_proof_glob_stays_scoped_to_install_receipts(tmp_path: Path) -> None: + report = _candidate(tmp_path) + install_receipts = {f"{target}.json" for target in PROOF_TARGETS} + nvattest_receipts = { + nvattest_public_receipt_name(target) for target in PROOF_TARGETS + } + + assert { + path.name for path in (report.evidence_dir / "proofs").glob("*.json") + } == install_receipts + assert { + path.name for path in (report.evidence_dir / "nvattest").glob("*.json") + } == install_receipts + + parts = collect_candidate_parts(report) + assert {item.name for item in parts.proofs} == ( + install_receipts | nvattest_receipts + ) def test_excluded_models_package_names_are_not_transparency_artifacts( @@ -637,15 +684,17 @@ def test_publish_genesis_uploads_fixed_layout_and_order( in result.public_urls ) calls = [(call["plane"], call["op"], call["key"]) for call in transport.call_log] + receipt_names = sorted( + {f"{target}.json" for target in PROOF_TARGETS} + | {nvattest_public_receipt_name(target) for target in PROOF_TARGETS} + ) immutable = [ version_object_key(PRODUCT, "0.9.1", ENTRY_OBJECT_NAME), version_object_key(PRODUCT, "0.9.1", ENTRY_SIGNATURE_NAME), *[ version_object_key(PRODUCT, "0.9.1", name) for name in ( - "linux-aarch64-musl.json", - "linux-x86_64-musl.json", - "macos-arm64.json", + *receipt_names, "solstone-core.rust-release-manifest.json", "solstone-journal-cuda.rust-release-manifest.json", "solstone-journal-models.rust-release-manifest.json", @@ -1507,6 +1556,98 @@ def test_stale_proofs_fail_closed_before_signing( assert error.value.failures[0].error == "retained proof version is stale" +def test_missing_nvattest_receipt_fails_closed(tmp_path: Path) -> None: + report = _candidate(tmp_path) + target = PROOF_TARGETS[0] + (report.evidence_dir / "nvattest" / f"{target}.json").unlink() + + with pytest.raises(DriverError) as error: + collect_candidate_parts(report) + + assert error.value.failures[0].error == ( + "retained nvattest receipt could not be read" + ) + + +def test_extra_nvattest_receipt_fails_closed(tmp_path: Path) -> None: + report = _candidate(tmp_path) + extra = report.evidence_dir / "nvattest" / "unexpected.json" + extra.write_text( + json.dumps( + { + "target": "not-a-proof-target", + "version": report.version, + }, + sort_keys=True, + ), + encoding="utf-8", + ) + + with pytest.raises(DriverError) as error: + collect_candidate_parts(report) + + assert error.value.failures[0].error == ( + "retained nvattest target is stale or unexpected" + ) + + +def test_stale_nvattest_receipt_fails_closed(tmp_path: Path) -> None: + report = _candidate(tmp_path) + target = PROOF_TARGETS[0] + receipt = report.evidence_dir / "nvattest" / f"{target}.json" + receipt.write_text( + json.dumps( + { + "target": target, + "version": f"{report.version}.stale", + }, + sort_keys=True, + ), + encoding="utf-8", + ) + report = _refresh_nvattest_digest(report, target) + + with pytest.raises(DriverError) as error: + collect_candidate_parts(report) + + assert error.value.failures[0].error == "retained nvattest version is stale" + + +def test_staged_nvattest_public_evidence_contains_no_private_reach_details( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + report = _candidate(tmp_path) + _patch_recover(monkeypatch) + stage = _stage_candidate(tmp_path, monkeypatch) + forbidden_fragments = ( + str(tmp_path), + "response_body", + "credential", + "secret", + "token", + "password", + "bearer", + ) + + for target in PROOF_TARGETS: + name = nvattest_public_receipt_name(target) + path = stage.version_dir / name + payload = json.loads(path.read_text(encoding="utf-8")) + evidence_text = json.dumps(payload, sort_keys=True).lower() + + assert _sha(path)[0] == report.nvattest_sha256[target] + assert validate_public_evidence_tree("nvattest_proof", payload) == [] + assert all( + fragment.lower() not in evidence_text for fragment in forbidden_fragments + ) + assert all( + not value.startswith("/") + for value in payload.values() + if isinstance(value, str) + ) + + def test_dirty_retained_manifest_fails_closed_before_signing( tmp_path: Path, monkeypatch: pytest.MonkeyPatch,