Something went wrong. Try again.
personal memory agent
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256#!/usr/bin/env python3# SPDX-License-Identifier: AGPL-3.0-only# Copyright (c) 2026 sol pbc
"""Provider-neutral proof-host channels for release candidate install proofs."""
from __future__ import annotations
import hashlibimport jsonimport osimport shleximport shutilimport statimport subprocessimport uuidfrom collections.abc import Callable, Mapping, Sequencefrom dataclasses import dataclassfrom pathlib import Pathfrom typing import Any, Protocol
from scripts.check_rust_release_manifest import ( SHA256_RE, SOURCE_COMMIT_RE, Failure, canonical_json_bytes,)from scripts.release_digest import file_sha256_sizefrom scripts.release_install_smoke import ( CURRENT_PROOF_SCHEMA_VERSION, PROOF_TARGETS, candidate_file_entries, target_install_paths_from_ledger, validate_install_proof_bytes,)from scripts.release_nvattest_proof import ( CHALLENGE_RE, NvattestProofError, candidate_wheel_entries, support_distribution_entries_with_metadata, validate_nvattest_proof_bytes,)from scripts.release_public_evidence import validate_public_evidence_treefrom scripts.release_target_policy import TARGET_ENV_KEYS, TARGET_POLICY
Runner = Callable[..., subprocess.CompletedProcess[str]]IdFactory = Callable[[], str]FileCopier = Callable[[Path, Path], object]
# `authority` is a descriptor (sha256 + bytes); `authority_file` is the staged relative path.REQUEST_KEYS = frozenset( ( "schema_version", "cohort_id", "target", "version", "source_commit", "candidate_digest", "ledger_sha256", "core_lock_sha256", "candidate_files", "support_files", "authority", "challenge", "paths", "expected_host", ))REQUEST_FILE_KEYS = frozenset(("basename", "bytes", "sha256", "path"))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", "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 label: str st_dev: int st_ino: int mode: int
class ProofHostChannel(Protocol): def run_target_proofs( self, *, target: str, version: str, source_commit: str, core_lock_sha256: str, candidate_digest: str, ledger_sha256: str, 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, nvattest_output_path: Path, ) -> TargetProofPaths: ...
class ProofHostError(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) -> Failure: return Failure(error=error, expected=expected, actual=actual, repair=repair)
def _run( runner: Runner, argv: Sequence[str], *, cwd: Path | None = None,) -> subprocess.CompletedProcess[str]: return runner( list(argv), cwd=cwd, capture_output=True, text=True, check=False, )
def _key_set_failures( label: str, payload: Mapping[str, object], expected: frozenset[str],) -> list[Failure]: actual = frozenset(str(key) for key in payload) if actual == expected: return [] return [ _failure( f"{label} key set is invalid", expected=", ".join(sorted(expected)), actual=", ".join(sorted(actual)) or "<empty>", repair="bash scripts/release.sh --candidate", ) ]
def _validate_cohort_id(cohort_id: str) -> None: if ( len(cohort_id) != 32 or cohort_id.lower() != cohort_id or any(char not in "0123456789abcdef" for char in cohort_id) ): raise ProofHostError( [ _failure( "proof-host cohort id is invalid", expected="32 lowercase hexadecimal characters", actual=cohort_id, repair="bash scripts/release.sh --candidate", ) ] )
def _safe_basename(value: object) -> str | None: if not isinstance(value, str): return None if not value or value in {".", ".."}: return None if Path(value).name != value or "/" in value or "\\" in value: return None return value
def _capture_directory_identity(path: Path, *, label: str) -> DirectoryIdentity: try: current = path.lstat() except OSError as exc: raise ProofHostError( [ _failure( f"proof-host {label} directory identity could not be captured", expected="owned non-symlink directory", actual=type(exc).__name__, repair="bash scripts/release.sh --candidate", ) ] ) from None if stat.S_ISLNK(current.st_mode): actual = "symlink" elif not stat.S_ISDIR(current.st_mode): actual = "non-directory" else: return DirectoryIdentity( path=path, label=label, st_dev=current.st_dev, st_ino=current.st_ino, mode=current.st_mode, ) raise ProofHostError( [ _failure( f"proof-host {label} directory identity could not be captured", expected="owned non-symlink directory", actual=actual, repair="bash scripts/release.sh --candidate", ) ] )
def _directory_identity_failure(identity: DirectoryIdentity) -> Failure | None: try: current = identity.path.lstat() except OSError as exc: return _failure( f"proof-host {identity.label} directory identity changed", expected="same owned non-symlink directory", actual=type(exc).__name__, repair="bash scripts/release.sh --candidate", ) if stat.S_ISLNK(current.st_mode): actual = "symlink" elif not stat.S_ISDIR(current.st_mode): actual = "non-directory" elif ( current.st_dev != identity.st_dev or current.st_ino != identity.st_ino or current.st_mode != identity.mode ): actual = "different directory" else: return None return _failure( f"proof-host {identity.label} directory identity changed", expected="same owned non-symlink directory", actual=actual, repair="bash scripts/release.sh --candidate", )
def _validate_directory_identities( identities: Sequence[DirectoryIdentity | None],) -> None: failures = [ failure for identity in identities if identity is not None if (failure := _directory_identity_failure(identity)) is not None ] if failures: raise ProofHostError(failures)
def _validate_regular_file(path: Path, *, label: str) -> None: try: mode = path.lstat().st_mode except OSError as exc: raise ProofHostError( [ _failure( f"proof-host {label} is not a regular file", expected="regular non-symlink file", actual=type(exc).__name__, repair="bash scripts/release.sh --candidate", ) ] ) from None if stat.S_ISLNK(mode): actual = "symlink" elif not stat.S_ISREG(mode): actual = "non-regular" else: return raise ProofHostError( [ _failure( f"proof-host {label} is not a regular file", expected="regular non-symlink file", actual=actual, repair="bash scripts/release.sh --candidate", ) ] )
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() except FileNotFoundError: return if path.is_symlink(): actual = "symlink" elif not path.is_dir(): actual = "non-directory" elif any(path.iterdir()): actual = "pre-existing entries" else: actual = "empty directory" raise ProofHostError( [ _failure( f"proof-host {label} directory is unsafe", expected="fresh directory path", actual=actual, repair="bash scripts/release.sh --candidate", ) ] )
def _combine_failures( primary: ProofHostError | None, cleanup: ProofHostError | None) -> ProofHostError | None: failures: list[Failure] = [] if primary is not None: failures.extend(primary.failures) if cleanup is not None: failures.extend(cleanup.failures) return ProofHostError(failures) if failures else None
def _validate_scalars( *, target: str, source_commit: str, core_lock_sha256: str, candidate_digest: str, ledger_sha256: str, challenge: str,) -> None: failures: list[Failure] = [] if target not in TARGET_POLICY: failures.append( _failure( "proof-host target is invalid", expected=", ".join(PROOF_TARGETS), actual=target, repair="bash scripts/release.sh --candidate", ) ) if not SOURCE_COMMIT_RE.fullmatch(source_commit): failures.append( _failure( "proof-host source commit is invalid", expected="40 or 64 lowercase hexadecimal characters", actual=source_commit, repair="bash scripts/release.sh --candidate", ) ) for label, value in ( ("core lock", core_lock_sha256), ("candidate digest", candidate_digest), ("ledger sha256", ledger_sha256), ): if not SHA256_RE.fullmatch(value): failures.append( _failure( f"proof-host {label} is invalid", expected="lowercase SHA-256", actual=value, 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)
def _json_object(path: Path) -> Mapping[str, object]: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise ProofHostError( [ _failure( "proof-host response is not valid JSON", expected="JSON object", actual=type(exc).__name__, repair="bash scripts/release.sh --candidate", ) ] ) from None if not isinstance(payload, Mapping): raise ProofHostError( [ _failure( "proof-host response is not an object", expected="JSON object", actual=type(payload).__name__, repair="bash scripts/release.sh --candidate", ) ] ) return payload
class ExternalProofHostChannel: """External command adapter; command and credentials live outside source."""
def __init__( self, target: str, command: Sequence[str], *, runner: Runner = subprocess.run, cohort_id_factory: IdFactory | None = None, file_copier: FileCopier = shutil.copyfile, ) -> None: if target not in TARGET_POLICY: raise ValueError("proof target is invalid") if not command: raise ValueError("proof-host command is required") self._target = target self._command = tuple(command) self._runner = runner self._cohort_id_factory = cohort_id_factory or (lambda: uuid.uuid4().hex) self._file_copier = file_copier
@classmethod def from_env( cls, target: str, env: Mapping[str, str], *, runner: Runner = subprocess.run, cohort_id_factory: IdFactory | None = None, file_copier: FileCopier = shutil.copyfile, ) -> ExternalProofHostChannel: env_key = TARGET_ENV_KEYS[target] try: command = shlex.split(env.get(env_key, "")) except ValueError as exc: raise ProofHostError( [ _failure( "proof-host channel configuration is invalid", expected="shell-style command tokens", actual=str(exc), repair="bash scripts/release.sh --candidate", ) ] ) from None if not command: raise ProofHostError( [ _failure( "proof-host channel is not configured", expected=f"{env_key} command", actual="<missing>", repair="bash scripts/release.sh --candidate", ) ] ) return cls( target, command, runner=runner, cohort_id_factory=cohort_id_factory, file_copier=file_copier, )
def _cleanup_remote(self, *, cohort_id: str, ledger_sha256: str) -> None: try: result = _run( self._runner, [*self._command, "cleanup", cohort_id, ledger_sha256], ) except BaseException as exc: raise ProofHostError( [ _failure( "proof-host remote cleanup failed", expected="external proof-host cleanup completed", actual=type(exc).__name__, repair="bash scripts/release.sh --candidate", ) ] ) from None if result.returncode != 0: raise ProofHostError( [ _failure( "proof-host remote cleanup failed", expected="external proof-host cleanup exit 0", actual=result.stderr.strip() or result.stdout.strip() or f"exit {result.returncode}", repair="bash scripts/release.sh --candidate", ) ] )
def _cleanup_local( self, targets: Sequence[tuple[Path, DirectoryIdentity | None]], ) -> ProofHostError | None: failures: list[Failure] = [] preserved: list[Path] = [] for path, identity in targets: try: if any(path in preserved_path.parents for preserved_path in preserved): failures.append( _failure( "proof-host local cleanup failed", expected="owned proof-host transients removed", actual="preserved descendant residue", repair="bash scripts/release.sh --candidate", ) ) continue if identity is not None: identity_failure = _directory_identity_failure(identity) if identity_failure is not None: preserved.append(path) failures.append( _failure( "proof-host local cleanup failed", expected="owned proof-host transients removed", actual=f"{identity.label} residue", repair="bash scripts/release.sh --candidate", ) ) continue if path.is_dir() and not path.is_symlink(): shutil.rmtree(path) elif path.exists() or path.is_symlink(): path.unlink() except BaseException as exc: preserved.append(path) failures.append( _failure( "proof-host local cleanup failed", expected="owned proof-host transients removed", actual=type(exc).__name__, repair="bash scripts/release.sh --candidate", ) ) 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, *, cohort_id: str, target: str, version: str, source_commit: str, core_lock_sha256: str, 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 = [ { **entry, "path": f"candidate/{entry['basename']}", } 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, "target": target, "version": version, "source_commit": source_commit, "candidate_digest": candidate_digest, "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", "install_proof": "output/install-proof.json", "nvattest_proof": "output/nvattest-proof.json", }, "expected_host": {"os": os_name, "arch": arch}, } if set(payload) != REQUEST_KEYS: raise AssertionError("proof-host request key set drifted") 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] raise AssertionError("proof-host request host key set drifted") return payload
def _validate_response( self, payload: Mapping[str, object], *, cohort_id: str, target: str, candidate_digest: str, ledger_sha256: str, ) -> Mapping[str, object]: failures = _key_set_failures("proof-host response", payload, RESPONSE_KEYS) if failures: raise ProofHostError(failures) if payload.get("schema_version") != 1: raise ProofHostError( [ _failure( "proof-host response schema version is wrong", expected="1", actual=repr(payload.get("schema_version")), repair="bash scripts/release.sh --candidate", ) ] ) if payload.get("cohort_id") != cohort_id: raise ProofHostError( [ _failure( "proof-host response cohort id is wrong", expected=cohort_id, actual=repr(payload.get("cohort_id")), repair="bash scripts/release.sh --candidate", ) ] ) attestation = payload.get("attestation") if not isinstance(attestation, Mapping): raise ProofHostError( [ _failure( "proof-host response attestation is invalid", expected="attestation object", actual=type(attestation).__name__, repair="bash scripts/release.sh --candidate", ) ] ) failures = _key_set_failures( "proof-host response attestation", attestation, ATTESTATION_KEYS, ) if failures: raise ProofHostError(failures) expected_os, expected_arch = TARGET_POLICY[target] expected = { "os": expected_os, "arch": expected_arch, "candidate_digest": candidate_digest, "ledger_sha256": ledger_sha256, } failures = [ _failure( f"proof-host response attestation {key} is wrong", expected=repr(expected_value), actual=repr(attestation.get(key)), repair="bash scripts/release.sh --candidate", ) for key, expected_value in expected.items() if attestation.get(key) != expected_value ] if failures: raise ProofHostError(failures) 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, ) if failures: raise ProofHostError(failures) descriptors[key] = descriptor return descriptors
def run_target_proofs( self, *, target: str, version: str, source_commit: str, core_lock_sha256: str, candidate_digest: str, ledger_sha256: str, 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, nvattest_output_path: Path, ) -> TargetProofPaths: if target != self._target: raise ProofHostError( [ _failure( "proof-host channel target mismatch", expected=self._target, actual=target, repair="bash scripts/release.sh --candidate", ) ] ) _ = candidate_paths _validate_scalars( target=target, source_commit=source_commit, 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, schema_version=CURRENT_PROOF_SCHEMA_VERSION, ) 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" 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") 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" ) 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", ) ] ) (request_dir / "ledger.json").write_bytes( canonical_json_bytes(ledger_payload) ) request_payload = self._request_payload( cohort_id=cohort_id, target=target, version=version, source_commit=source_commit, core_lock_sha256=core_lock_sha256, 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( "proof_host.request", request_payload, ) if public_failures: raise ProofHostError(public_failures) remote_started = True result = _run( self._runner, [*self._command, "prove", "request.json"], cwd=request_dir, ) if result.returncode != 0: raise ProofHostError( [ _failure( "proof-host proof command failed", expected="external proof-host command exit 0", actual=result.stderr.strip() or result.stdout.strip() or f"exit {result.returncode}", repair="bash scripts/release.sh --candidate", ) ] ) _validate_directory_identities(identities) _validate_regular_file(response_path, label="response") response = _json_object(response_path) proof_descriptors = self._validate_response( response, cohort_id=cohort_id, target=target, candidate_digest=candidate_digest, ledger_sha256=ledger_sha256, ) 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 install proof path is invalid", expected="output/install-proof.json", actual=repr(install_proof_name), repair="bash scripts/release.sh --candidate", ) ] ) nvattest_proof_name = nvattest_descriptor.get("path") if nvattest_proof_name != "output/nvattest-proof.json": raise ProofHostError( [ _failure( "proof-host nvattest proof path is invalid", expected="output/nvattest-proof.json", actual=repr(nvattest_proof_name), repair="bash scripts/release.sh --candidate", ) ] ) _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 install proof SHA-256 is wrong", expected=install_sha256, actual=repr(install_descriptor.get("sha256")), repair="bash scripts/release.sh --candidate", ) ] ) 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) try: install_proof_payload = json.loads(install_proof_bytes) except json.JSONDecodeError as exc: raise ProofHostError( [ _failure( "proof-host install proof schema could not be read", expected=( f"schema_version {CURRENT_PROOF_SCHEMA_VERSION} " "install proof JSON" ), actual=str(exc), repair="bash scripts/release.sh --candidate", ) ] ) from None actual_schema = ( install_proof_payload.get("schema_version") if isinstance(install_proof_payload, Mapping) else install_proof_payload ) if actual_schema != CURRENT_PROOF_SCHEMA_VERSION: raise ProofHostError( [ _failure( "proof-host install proof schema is not current", expected=str(CURRENT_PROOF_SCHEMA_VERSION), actual=repr(actual_schema), repair="bash scripts/release.sh --candidate", ) ] ) proof_failures = validate_install_proof_bytes( install_proof_bytes, target=target, version=version, source_commit=source_commit, core_lock_sha256=core_lock_sha256, candidate_digest=candidate_digest, ledger_sha256=ledger_sha256, candidate_dir=candidate_dir, ledger_payload=ledger_payload, ) if proof_failures: raise ProofHostError(proof_failures) try: expected_candidate_wheels = candidate_wheel_entries( target_install_paths_from_ledger( ledger_payload, target=target, candidate_dir=request_candidate_dir, schema_version=CURRENT_PROOF_SCHEMA_VERSION, ) ) expected_support_distributions = ( support_distribution_entries_with_metadata( tuple( sorted( request_support_dir.iterdir(), key=lambda path: path.name, ) ) ) ) except NvattestProofError as exc: raise ProofHostError(exc.failures) from exc 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_candidate_wheels=expected_candidate_wheels, expected_support_distributions=expected_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 TargetProofPaths(install=output_path, nvattest=nvattest_output_path) except BaseException as exc: if isinstance(exc, ProofHostError): primary = exc else: primary = ProofHostError( [ _failure( "proof-host operation failed", expected="proof-host request completed", actual=type(exc).__name__, repair="bash scripts/release.sh --candidate", ) ] ) raise primary finally: cleanup_error: ProofHostError | None = None if remote_started: try: self._cleanup_remote( cohort_id=cohort_id, ledger_sha256=ledger_sha256, ) except ProofHostError as exc: cleanup_error = exc 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), ) ) cleanup_error = _combine_failures(cleanup_error, local_cleanup) if cleanup_error is not None: combined = _combine_failures(primary, cleanup_error) assert combined is not None if completed: raise combined raise combined
def proof_channels_from_env( env: Mapping[str, str], *, runner: Runner = subprocess.run,) -> dict[str, ProofHostChannel]: return { target: ExternalProofHostChannel.from_env(target, env, runner=runner) for target in PROOF_TARGETS }
def run_target_proofs_with_channels( channels: Mapping[str, ProofHostChannel], **kwargs: Any,) -> TargetProofPaths: target = str(kwargs.get("target", "")) channel = channels.get(target) if channel is None: raise ProofHostError( [ _failure( "proof-host channel is missing", expected=f"{target} configured proof-host channel", actual="<missing>", repair="bash scripts/release.sh --candidate", ) ] ) return channel.run_target_proofs(**kwargs)