"""Offline rehydration and sealed expected-state checks for Task26 bundles.""" from __future__ import annotations import argparse import hashlib import json import os import re import shutil import stat import subprocess import sys import tempfile import zipfile from email.parser import BytesParser from email.policy import compat32 from pathlib import Path from typing import Any, Mapping, Sequence, cast from packaging.requirements import InvalidRequirement, Requirement from packaging.version import InvalidVersion, Version from gateway.platforms.task26_candidate_derivation import ( qualification_tool_hashes, trust_boundary_digest, validate_product_binding, ) from gateway.platforms.task26_runtime_authority import validate_snapshot TRUST_BOUNDARY: dict[str, object] = { "schema": "task26-evidence-trust-boundary-v1", "trust_anchors": [ "trusted_operator_account", "host_kernel", "selected_python_pip_toolchain", "sealed_wheel_bytes", ], "guarantees_within_boundary": [ "content_identity", "deterministic_tamper_detection", "reproducibility", "revocation_and_current_authority_selection", "fail_closed_workflow", ], "scope_statement": ( "These guarantees apply only while the trusted operator account, host " "kernel, selected Python/pip toolchain, and sealed wheel bytes remain " "trustworthy." ), "outside_scope": ( "Coherent compromise of the trusted operator account or host administrator, " "including coherent modification of the repository, ledger, or verifier, is " "outside the Task26 threat model." ), "not_claimed": [ "external_authenticity", "non_repudiation", "owner_tamper_proof_or_worm", "resistance_to_compromised_operator_or_admin", ], "external_authenticity_claimed": False, "signature_claim": "none", } _EMPTY_WHEELHOUSE_PAYLOAD: dict[str, object] = { "root_mode": None, "directories": [], "entries": [], } _EMPTY_WHEELHOUSE: dict[str, object] = { "schema": "task26-sealed-wheelhouse-inventory-v1", **_EMPTY_WHEELHOUSE_PAYLOAD, "inventory_sha256": hashlib.sha256( json.dumps( _EMPTY_WHEELHOUSE_PAYLOAD, sort_keys=True, separators=(",", ":"), ).encode() ).hexdigest(), } _NORMALIZE_PROJECT = re.compile(r"[-_.]+") _HEX_DIGEST = re.compile(r"^[0-9a-f]{64}$") def canonical(value: object) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode() def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _private(path: Path, *, directory: bool = False) -> None: info = path.lstat() expected = stat.S_ISDIR if directory else stat.S_ISREG modes = {0o700, 0o500} if directory else {0o600, 0o400} if ( stat.S_ISLNK(info.st_mode) or not expected(info.st_mode) or info.st_uid != os.geteuid() or (not directory and info.st_nlink != 1) or stat.S_IMODE(info.st_mode) not in modes ): raise ValueError(f"sealed private path is invalid: {path}") def _object(value: object, label: str) -> dict[str, Any]: if not isinstance(value, dict) or any(not isinstance(key, str) for key in value): raise ValueError(f"{label} is invalid") return cast(dict[str, Any], value) def _normal_project(value: str) -> str: return _NORMALIZE_PROJECT.sub("-", value).lower() def _wheel_identity(path: Path) -> dict[str, object]: _private(path) try: with zipfile.ZipFile(path) as archive: names = archive.namelist() if ( len(names) != len(set(names)) or any(name.startswith("/") or ".." in Path(name).parts for name in names) ): raise ValueError("wheel members are invalid") metadata_names = [ name for name in names if name.endswith(".dist-info/METADATA") ] if len(metadata_names) != 1: raise ValueError("wheel METADATA cardinality is invalid") message = BytesParser(policy=compat32).parsebytes( archive.read(metadata_names[0]) ) except zipfile.BadZipFile as exc: raise ValueError("dependency wheel is invalid") from exc project = str(message.get("Name", "")).strip() version_text = str(message.get("Version", "")).strip() if not project or not version_text: raise ValueError("wheel project/version is unavailable") try: version = str(Version(version_text)) except InvalidVersion as exc: raise ValueError("wheel version is invalid") from exc requirements = sorted(str(value) for value in (message.get_all("Requires-Dist") or [])) return { "project": _normal_project(project), "version": version, "requires_dist": requirements, "sha256": sha256_file(path), "size": path.stat().st_size, "mode": stat.S_IMODE(path.stat().st_mode), } def _wheelhouse_files( root: Path, ) -> tuple[list[Path], list[dict[str, object]]]: _private(root, directory=True) wheels: list[Path] = [] directories: list[dict[str, object]] = [] pending = [root] while pending: directory = pending.pop() _private(directory, directory=True) for entry in sorted(os.scandir(directory), key=lambda item: item.name): path = Path(entry.path) info = path.lstat() if stat.S_ISLNK(info.st_mode): raise ValueError("wheelhouse symlink is forbidden") if stat.S_ISDIR(info.st_mode): _private(path, directory=True) directories.append( { "relative_path": path.relative_to(root).as_posix(), "mode": stat.S_IMODE(info.st_mode), } ) pending.append(path) elif stat.S_ISREG(info.st_mode) and path.suffix == ".whl": _private(path) wheels.append(path) else: raise ValueError("wheelhouse contains an unbound non-wheel entry") return ( sorted(wheels, key=lambda path: path.relative_to(root).as_posix()), sorted(directories, key=lambda row: str(row["relative_path"])), ) def _validate_dependency_closure( wheels: Sequence[tuple[Path, Mapping[str, object]]], ) -> None: projects: dict[str, tuple[Path, Mapping[str, object]]] = {} identities: set[tuple[str, str]] = set() for path, identity in wheels: project = str(identity.get("project", "")) version = str(identity.get("version", "")) key = (project, version) if key in identities or project in projects: raise ValueError("duplicate project/version in explicit wheel set") identities.add(key) projects[project] = (path, identity) for _path, identity in wheels: requirements = identity.get("requires_dist") if not isinstance(requirements, list): raise ValueError("wheel dependency declarations are invalid") for raw in requirements: try: requirement = Requirement(str(raw)) except InvalidRequirement as exc: raise ValueError("wheel dependency declaration is invalid") from exc if requirement.url is not None: raise ValueError("direct URL dependencies are forbidden") if requirement.marker is not None and not requirement.marker.evaluate(): continue dependency = projects.get(_normal_project(requirement.name)) if dependency is None: raise ValueError(f"missing dependency: {requirement.name}") selected_version = str(dependency[1].get("version", "")) if requirement.specifier and not requirement.specifier.contains( selected_version, prereleases=True ): raise ValueError(f"dependency version mismatch: {requirement.name}") def build_wheelhouse_inventory( wheelhouse: Path | None, *, required_wheels: Sequence[Path], ) -> dict[str, object]: """Build an exact recursive inventory and prove the explicit dependency closure.""" root_wheels = [(path, _wheel_identity(path)) for path in required_wheels] if wheelhouse is None: dependency_wheels: list[tuple[Path, Mapping[str, object]]] = [] entries: list[dict[str, object]] = [] else: paths, directories = _wheelhouse_files(wheelhouse) dependency_wheels = [(path, _wheel_identity(path)) for path in paths] entries = [ { "relative_path": path.relative_to(wheelhouse).as_posix(), **identity, } for path, identity in dependency_wheels ] if wheelhouse is None: root_mode: int | None = None directories = [] else: root_mode = stat.S_IMODE(wheelhouse.stat().st_mode) _validate_dependency_closure([*root_wheels, *dependency_wheels]) payload = { "root_mode": root_mode, "directories": directories, "entries": entries, } return { "schema": "task26-sealed-wheelhouse-inventory-v1", **payload, "inventory_sha256": hashlib.sha256(canonical(payload)).hexdigest(), } def verify_wheelhouse_inventory( wheelhouse: Path | None, expected: object, *, required_wheels: Sequence[Path], ) -> dict[str, object]: document = _object(expected, "sealed wheelhouse inventory") if set(document) != { "schema", "root_mode", "directories", "entries", "inventory_sha256", } or document.get( "schema" ) != "task26-sealed-wheelhouse-inventory-v1": raise ValueError("sealed wheelhouse inventory schema is invalid") entries = document.get("entries") directories = document.get("directories") payload = { "root_mode": document.get("root_mode"), "directories": directories, "entries": entries, } if ( not isinstance(entries, list) or not isinstance(directories, list) or document.get("inventory_sha256") != hashlib.sha256(canonical(payload)).hexdigest() ): raise ValueError("sealed wheelhouse inventory digest is invalid") actual = build_wheelhouse_inventory( wheelhouse, required_wheels=required_wheels ) if actual != document: raise ValueError("wheelhouse inventory differs from sealed expectation") return actual def _bound_wheel_paths( wheelhouse: Path | None, inventory: Mapping[str, object] ) -> list[Path]: entries = inventory.get("entries") if not isinstance(entries, list): raise ValueError("sealed wheelhouse entries are invalid") if wheelhouse is None: if entries: raise ValueError("sealed wheelhouse is unavailable") return [] paths: list[Path] = [] for raw in entries: row = _object(raw, "sealed wheelhouse row") relative = row.get("relative_path") if ( not isinstance(relative, str) or Path(relative).is_absolute() or ".." in Path(relative).parts ): raise ValueError("sealed wheelhouse path is invalid") paths.append(wheelhouse / relative) return paths def _contains_nonportable_record_material(value: object) -> bool: if isinstance(value, dict): return any( key in { "raw_installed_records_nonportable", "raw_installed_record_sha256", "raw_installed_record_portable", "record_path", } or _contains_nonportable_record_material(item) for key, item in value.items() ) if isinstance(value, list): return any(_contains_nonportable_record_material(item) for item in value) return False def _validate_portable_origin_contract(runtime_portable: object) -> None: runtime = _object(runtime_portable, "portable installed runtime") proof = _object(runtime.get("origin_proof"), "portable origin proof") execution = _object(proof.get("execution"), "portable origin execution") modules = proof.get("modules") if ( runtime.get("schema") != "installed-golden-runtime-portable-v2" or _contains_nonportable_record_material(runtime) or proof.get("schema") != "task26-installed-origin-proof-v1" or execution != { "cwd_authority": "private_empty_controlled", "python_isolated": True, "pythonpath_inherited": False, } or not isinstance(modules, list) or len(modules) != 2 ): raise ValueError("portable installed origin proof is invalid") expected_pairs = [("gateway", "hermes"), ("checkin_cli", "profile")] for raw, (module, distribution) in zip(modules, expected_pairs, strict=True): row = _object(raw, "portable origin module") relative = row.get("installed_relative_path") if ( set(row) != { "distribution", "module", "installed_relative_path", "sha256", "size", } or row.get("module") != module or row.get("distribution") != distribution or not isinstance(relative, str) or Path(relative).is_absolute() or ".." in Path(relative).parts or _HEX_DIGEST.fullmatch(str(row.get("sha256", ""))) is None or type(row.get("size")) is not int or row["size"] < 1 ): raise ValueError("portable installed origin module proof is invalid") def _validate_nonportable_record_audit( value: object, *, install_role: str, candidate_digest: str, runtime_portable: Mapping[str, object], ) -> dict[str, object]: audit = _object(value, "nonportable installed RECORD audit") if set(audit) != { "schema", "install_role", "candidate_digest", "distributions", "audit_sha256", }: raise ValueError("nonportable installed RECORD audit schema is invalid") unsigned = {key: item for key, item in audit.items() if key != "audit_sha256"} audit_distributions = _object( audit.get("distributions"), "nonportable audit distributions" ) portable_distributions = _object( runtime_portable.get("distributions"), "portable runtime distributions" ) if ( audit.get("schema") != "task26-nonportable-installed-record-audit-v1" or audit.get("install_role") != install_role or audit.get("candidate_digest") != candidate_digest or audit.get("audit_sha256") != hashlib.sha256(canonical(unsigned)).hexdigest() or set(audit_distributions) != {"profile", "hermes"} ): raise ValueError("nonportable installed RECORD audit binding is invalid") for role in ("profile", "hermes"): row = _object(audit_distributions.get(role), f"{role} raw RECORD audit") portable = _object( portable_distributions.get(role), f"{role} portable distribution" ) classifications = row.get("path_dependent_classifications") if ( set(row) != { "distribution", "wheel_sha256", "portable_projection_sha256", "raw_installed_record_sha256", "raw_installed_record_portable", "reason", "path_dependent_classifications", } or row.get("distribution") != portable.get("distribution_name") or row.get("wheel_sha256") != portable.get("wheel_sha256") or row.get("portable_projection_sha256") != portable.get("portable_record_projection_sha256") or _HEX_DIGEST.fullmatch( str(row.get("raw_installed_record_sha256", "")) ) is None or row.get("raw_installed_record_portable") is not False or row.get("reason") != "validated_path_dependent_installer_material_may_change_raw_record" or not isinstance(classifications, list) or classifications != sorted(set(classifications)) or any( item not in { "path_dependent_launcher", "console_entry_point_launcher", } for item in classifications ) ): raise ValueError("nonportable installed RECORD audit row is invalid") return audit def verify_nonportable_record_audit_pair( original: object, rehydrated: object, *, candidate_digest: str, original_runtime_portable: Mapping[str, object], rehydrated_runtime_portable: Mapping[str, object], ) -> dict[str, object]: """Validate both raw audits without comparing their nonportable hashes.""" original_audit = _validate_nonportable_record_audit( original, install_role="original", candidate_digest=candidate_digest, runtime_portable=original_runtime_portable, ) rehydrated_audit = _validate_nonportable_record_audit( rehydrated, install_role="rehydrated", candidate_digest=candidate_digest, runtime_portable=rehydrated_runtime_portable, ) original_distributions = _object( original_runtime_portable.get("distributions"), "original portable distributions", ) rehydrated_distributions = _object( rehydrated_runtime_portable.get("distributions"), "rehydrated portable distributions", ) projection_digests: dict[str, object] = {} for role in ("profile", "hermes"): original_distribution = _object( original_distributions.get(role), f"original {role} distribution" ) rehydrated_distribution = _object( rehydrated_distributions.get(role), f"rehydrated {role} distribution" ) projection = original_distribution.get( "portable_record_projection_sha256" ) if ( projection != rehydrated_distribution.get("portable_record_projection_sha256") or original_distribution.get("wheel_sha256") != rehydrated_distribution.get("wheel_sha256") ): raise ValueError("original and rehydrated portable projection differs") projection_digests[role] = projection return { "schema": "task26-frozen-installed-record-verification-v2", "original_nonportable_record_audit": original_audit, "rehydrated_nonportable_record_audit": rehydrated_audit, "raw_record_hash_equality_required_or_claimed": False, "portable_projection_sha256": projection_digests, } def build_outer_candidate_manifest( product_binding: Mapping[str, object], ) -> dict[str, object]: """Bind the complete canonical product derivation, never a digest tautology.""" validated = validate_product_binding(dict(product_binding)) document: dict[str, object] = { "schema": "task26-outer-candidate-manifest-v2", "candidate_digest": validated["candidate_digest"], "product_binding": validated, } document["manifest_sha256"] = hashlib.sha256(canonical(document)).hexdigest() return document def _verify_outer_candidate_manifest( bundle: Path, relative_path: str ) -> tuple[str, str, dict[str, object]]: relative = Path(relative_path) if relative.is_absolute() or ".." in relative.parts: raise ValueError("outer candidate manifest path is invalid") path = bundle / relative _private(path) document = _object( json.loads(path.read_text(encoding="utf-8")), "outer candidate manifest" ) if set(document) != { "schema", "candidate_digest", "product_binding", "manifest_sha256", }: raise ValueError("outer candidate manifest schema is invalid") unsigned = { key: value for key, value in document.items() if key != "manifest_sha256" } product_binding = validate_product_binding(document.get("product_binding")) candidate = str(document.get("candidate_digest", "")) if ( document.get("schema") != "task26-outer-candidate-manifest-v2" or candidate != product_binding["candidate_digest"] or document.get("manifest_sha256") != hashlib.sha256(canonical(unsigned)).hexdigest() ): raise ValueError("outer candidate manifest binding is invalid") return candidate, sha256_file(path), product_binding def _candidate_result_projection( result: Mapping[str, object], *, mode: str ) -> dict[str, object]: authority = _object(result.get("candidate_authority"), f"{mode} authority") contract = _object(result.get("task26_contract"), f"{mode} Task26 contract") transcript = _object( contract.get("local_socket_transcript"), f"{mode} local socket transcript", ) deployment = _object(result.get("deployment_wheels"), f"{mode} deployment wheels") candidate = result.get("candidate_digest") source_tree = result.get("source_tree_digest") capability_identity = _object( result.get("capability_identity_receipt"), f"{mode} capability identity receipt", ) runtime_authority_snapshot = validate_snapshot( result.get("runtime_authority_snapshot") ) capability_unsigned = { key: value for key, value in capability_identity.items() if key != "receipt_sha256" } if ( result.get("runtime_mode") != mode or _HEX_DIGEST.fullmatch(str(candidate)) is None or _HEX_DIGEST.fullmatch(str(source_tree)) is None or authority.get("current_qualified_candidate") != candidate or _HEX_DIGEST.fullmatch( str(authority.get("registry_head_sha256", "")) ) is None or _HEX_DIGEST.fullmatch( str(authority.get("ledger_head_sha256", "")) ) is None or _HEX_DIGEST.fullmatch(str(contract.get("chain_head", ""))) is None or transcript.get("schema") != "task26-local-http-telegram-api-transcript-binding-v1" or transcript.get("runtime_mode") != mode or transcript.get("relative_path") != f"data/task26-local-http-telegram-api-qa-{mode}.json" or _HEX_DIGEST.fullmatch( str(transcript.get("transcript_sha256", "")) ) is None or deployment.get("schema") != "task26-deployment-wheel-binding-v1" or deployment.get("candidate_digest") != candidate or _HEX_DIGEST.fullmatch( str(deployment.get("candidate_product_binding_sha256", "")) ) is None or _HEX_DIGEST.fullmatch( str(deployment.get("hermes_wheel_sha256", "")) ) is None or _HEX_DIGEST.fullmatch( str(deployment.get("profile_wheel_sha256", "")) ) is None or deployment.get("hermes_wheel_sha256") == deployment.get("profile_wheel_sha256") or candidate in { deployment.get("hermes_wheel_sha256"), deployment.get("profile_wheel_sha256"), } or capability_unsigned != { "schema": "task26-delivery-capability-identity-receipt-v1", "candidate_digest": candidate, "candidate_product_binding_sha256": deployment.get( "candidate_product_binding_sha256" ), "hermes_wheel_sha256": deployment.get("hermes_wheel_sha256"), "profile_wheel_sha256": deployment.get("profile_wheel_sha256"), "wheel_digest": deployment.get("hermes_wheel_sha256"), } or capability_identity.get("receipt_sha256") != hashlib.sha256(canonical(capability_unsigned)).hexdigest() or runtime_authority_snapshot.get("candidate_digest") != candidate ): raise ValueError(f"{mode} candidate or authority projection is invalid") return { "candidate_digest": candidate, "source_tree_digest": source_tree, "task26_chain_head": contract["chain_head"], "authority_current_candidate": authority["current_qualified_candidate"], "authority_registry_head": authority.get("registry_head_sha256"), "authority_ledger_head": authority.get("ledger_head_sha256"), "deployment_wheels": deployment, "capability_identity_receipt": capability_identity, "runtime_authority_snapshot": runtime_authority_snapshot, "local_socket_transcript": transcript, } def build_candidate_parity( bundle: Path, *, candidate_digest: str, source_result: Mapping[str, object], installed_result: Mapping[str, object], source_bundle_relative_path: str, installed_bundle_relative_path: str, outer_manifest_relative_path: str, ) -> dict[str, object]: source = _candidate_result_projection(source_result, mode="source") installed = _candidate_result_projection(installed_result, mode="installed") outer_candidate, outer_sha, product_binding = _verify_outer_candidate_manifest( bundle, outer_manifest_relative_path ) product_inputs = _object( product_binding.get("derivation_inputs"), "outer product derivation inputs" ) if source["source_tree_digest"] != installed["source_tree_digest"]: raise ValueError("source and installed tree provenance digests differ") source_transcript = _object( source["local_socket_transcript"], "source local socket transcript" ) installed_transcript = _object( installed["local_socket_transcript"], "installed local socket transcript" ) if ( source_transcript.get("runtime_mode") != "source" or installed_transcript.get("runtime_mode") != "installed" or source_transcript.get("relative_path") == installed_transcript.get("relative_path") or source_transcript.get("transcript_sha256") == installed_transcript.get("transcript_sha256") ): raise ValueError("source and installed socket transcripts were substituted") if ( source["deployment_wheels"] != installed["deployment_wheels"] or _object(source["deployment_wheels"], "source deployment wheels").get( "candidate_product_binding_sha256" ) != product_binding["binding_sha256"] or _object(source["deployment_wheels"], "source deployment wheels").get( "hermes_wheel_sha256" ) != product_inputs["hermes_wheel_sha256"] or _object(source["deployment_wheels"], "source deployment wheels").get( "profile_wheel_sha256" ) != product_inputs["profile_wheel_sha256"] ): raise ValueError("source and installed deployment wheel receipts differ") if any( value != candidate_digest for value in ( source["candidate_digest"], source["authority_current_candidate"], installed["candidate_digest"], installed["authority_current_candidate"], outer_candidate, ) ): raise ValueError("source, installed, authority, and outer candidates differ") for relative in (source_bundle_relative_path, installed_bundle_relative_path): path = Path(relative) if path.is_absolute() or ".." in path.parts: raise ValueError("candidate parity bundle path is invalid") document: dict[str, object] = { "schema": "task26-cross-mode-candidate-parity-v1", "candidate_digest": candidate_digest, "source_bundle_relative_path": source_bundle_relative_path, "installed_bundle_relative_path": installed_bundle_relative_path, "source": source, "installed": installed, "outer_manifest": { "relative_path": outer_manifest_relative_path, "sha256": outer_sha, "candidate_digest": outer_candidate, "product_binding_sha256": product_binding["binding_sha256"], }, } document["parity_sha256"] = hashlib.sha256(canonical(document)).hexdigest() return document def verify_candidate_parity( bundle: Path, parity: object, *, expected_candidate: str ) -> dict[str, object]: document = _object(parity, "candidate parity") if set(document) != { "schema", "candidate_digest", "source_bundle_relative_path", "installed_bundle_relative_path", "source", "installed", "outer_manifest", "parity_sha256", }: raise ValueError("candidate parity schema is invalid") unsigned = { key: value for key, value in document.items() if key != "parity_sha256" } outer = _object(document.get("outer_manifest"), "outer manifest parity") relative = outer.get("relative_path") if ( document.get("schema") != "task26-cross-mode-candidate-parity-v1" or document.get("candidate_digest") != expected_candidate or document.get("parity_sha256") != hashlib.sha256(canonical(unsigned)).hexdigest() or not isinstance(relative, str) ): raise ValueError("candidate parity digest is invalid") outer_candidate, outer_sha, product_binding = _verify_outer_candidate_manifest( bundle, relative ) source = _object(document.get("source"), "source candidate parity") installed = _object(document.get("installed"), "installed candidate parity") product_inputs = _object( product_binding.get("derivation_inputs"), "outer product derivation inputs" ) source_deployment = _object( source.get("deployment_wheels"), "source deployment wheels" ) if ( source.get("source_tree_digest") != installed.get("source_tree_digest") or source.get("deployment_wheels") != installed.get("deployment_wheels") or source_deployment.get("candidate_product_binding_sha256") != product_binding.get("binding_sha256") or source_deployment.get("hermes_wheel_sha256") != product_inputs.get("hermes_wheel_sha256") or source_deployment.get("profile_wheel_sha256") != product_inputs.get("profile_wheel_sha256") or outer.get("sha256") != outer_sha or outer.get("candidate_digest") != outer_candidate or outer.get("product_binding_sha256") != product_binding.get("binding_sha256") or any( value != expected_candidate for value in ( outer_candidate, source.get("candidate_digest"), source.get("authority_current_candidate"), installed.get("candidate_digest"), installed.get("authority_current_candidate"), ) ) ): raise ValueError("source, installed, authority, and outer candidates differ") return cast(dict[str, object], document) def verify_expected_state( bundle: Path, expected_path: Path, *, hermes_wheel: Path, profile_wheel: Path, ) -> dict[str, Any]: """Validate the hash-bound portable expectation before creating a runtime.""" _private(bundle, directory=True) _private(expected_path) _private(hermes_wheel) _private(profile_wheel) expected = _object( json.loads(expected_path.read_text(encoding="utf-8")), "sealed expected state", ) if set(expected) != { "schema", "candidate_digest", "runtime_portable", "original_nonportable_record_audit", "hermetic_ty_attestation", "hermes_wheel", "profile_wheel", "wheelhouse", "verifier", "provenance_helper", "frozen_bootstrap", "independent_candidate_verifier", "expected_status", "expected_result_sha256", "authority", "candidate_parity", "local_socket_transcripts", "trust_boundary", "document_sha256", } or expected.get("schema") != "task26-sealed-final-state-v6": raise ValueError("sealed expected-state schema is invalid") claimed = expected.get("document_sha256") unsigned = {key: value for key, value in expected.items() if key != "document_sha256"} if claimed != hashlib.sha256(canonical(unsigned)).hexdigest(): raise ValueError("sealed expected-state digest is invalid") if expected.get("trust_boundary") != TRUST_BOUNDARY: raise ValueError("sealed evidence trust boundary is invalid") _validate_portable_origin_contract(expected.get("runtime_portable")) _validate_nonportable_record_audit( expected.get("original_nonportable_record_audit"), install_role="original", candidate_digest=str(expected.get("candidate_digest", "")), runtime_portable=_object( expected.get("runtime_portable"), "portable installed runtime" ), ) verified_parity = verify_candidate_parity( bundle, expected.get("candidate_parity"), expected_candidate=str(expected.get("candidate_digest", "")), ) transcript_bindings = _object( expected.get("local_socket_transcripts"), "sealed local socket transcripts", ) source_projection = _object( verified_parity.get("source"), "source candidate projection" ) installed_projection = _object( verified_parity.get("installed"), "installed candidate projection" ) if transcript_bindings != { "source": source_projection.get("local_socket_transcript"), "installed": installed_projection.get("local_socket_transcript"), }: raise ValueError("sealed local socket transcript bindings differ") for label, path in ( ("hermes_wheel", hermes_wheel), ("profile_wheel", profile_wheel), ): binding = _object(expected.get(label), label) if binding != {"filename": path.name, "sha256": sha256_file(path)}: raise ValueError(f"sealed {label} digest differs") wheelhouse = _object(expected.get("wheelhouse"), "sealed wheelhouse inventory") if set(wheelhouse) != { "schema", "root_mode", "directories", "entries", "inventory_sha256", }: raise ValueError("sealed wheelhouse inventory schema is invalid") entries = wheelhouse.get("entries") if not isinstance(entries, list): raise ValueError("sealed wheelhouse entries are invalid") parity = _object(expected.get("candidate_parity"), "candidate parity") outer = _object(parity.get("outer_manifest"), "outer manifest parity") relative = outer.get("relative_path") if not isinstance(relative, str): raise ValueError("outer candidate manifest path is invalid") _, _, product_binding = _verify_outer_candidate_manifest(bundle, relative) validate_product_binding( product_binding, expected_tool_hashes=qualification_tool_hashes( scripts_dir=bundle / "verification-tools", package_root=Path(__file__).resolve().parents[2], ), expected_trust_boundary_sha256=trust_boundary_digest(TRUST_BOUNDARY), expected_wheelhouse_inventory_sha256=str(wheelhouse["inventory_sha256"]), expected_wheelhouse_entry_count=len(entries), actual_hermes_wheel=hermes_wheel, actual_profile_wheel=profile_wheel, ) _validate_hermetic_ty_attestation( expected.get("hermetic_ty_attestation"), candidate_digest=str(expected.get("candidate_digest", "")), product_binding=product_binding, ) for label in ( "verifier", "provenance_helper", "frozen_bootstrap", "independent_candidate_verifier", ): binding = _object(expected.get(label), label) relative = binding.get("relative_path") if ( not isinstance(relative, str) or Path(relative).is_absolute() or ".." in Path(relative).parts ): raise ValueError(f"sealed {label} path is invalid") target = bundle / relative _private(target) if binding.get("sha256") != sha256_file(target): raise ValueError(f"sealed {label} digest differs") authority = _object(expected.get("authority"), "sealed authority") for name in ("registry.json", "qualification-ledger.json"): target = bundle / "candidate-authority" / name _private(target) if authority.get(name.removesuffix(".json") + "_sha256") != sha256_file( target ): raise ValueError("sealed authority digest differs") return expected def _portable_verifier_result(result: Mapping[str, object]) -> dict[str, object]: return { key: value for key, value in result.items() if key not in {"package_authority", "rehydration_origin_proof"} } def _validate_hermetic_ty_attestation( value: object, *, candidate_digest: str, product_binding: Mapping[str, object], ) -> dict[str, object]: attestation = _object(value, "sealed hermetic Ty attestation") if set(attestation) != { "schema", "candidate_digest", "corrected_ty_wheel_sha256", "ty_version", "binary_member_sha256", "raw_diagnostics_sha256", "raw_diagnostics_count", "execution_provenance_sha256", "receipt_sha256", "attestation_sha256", }: raise ValueError("sealed hermetic Ty attestation schema is invalid") inputs = _object(product_binding.get("derivation_inputs"), "product inputs") hermetic_ty = _object(inputs.get("hermetic_ty"), "hermetic Ty binding") unsigned = { key: item for key, item in attestation.items() if key != "attestation_sha256" } if ( attestation.get("schema") != "task26-sealed-hermetic-ty-attestation-v1" or attestation.get("candidate_digest") != candidate_digest or attestation.get("corrected_ty_wheel_sha256") != hermetic_ty.get("corrected_ty_wheel_sha256") or attestation.get("ty_version") != hermetic_ty.get("ty_version") or attestation.get("binary_member_sha256") != hermetic_ty.get("binary_member_sha256") or any( _HEX_DIGEST.fullmatch(str(attestation.get(field, ""))) is None for field in ( "raw_diagnostics_sha256", "execution_provenance_sha256", "receipt_sha256", ) ) or type(attestation.get("raw_diagnostics_count")) is not int or attestation["raw_diagnostics_count"] < 1 or attestation.get("attestation_sha256") != hashlib.sha256(canonical(unsigned)).hexdigest() ): raise ValueError("sealed hermetic Ty attestation binding is invalid") return attestation def build_expected_state( bundle: Path, *, candidate_digest: str, runtime_portable: Mapping[str, object], original_nonportable_record_audit: Mapping[str, object], hermetic_ty_attestation: Mapping[str, object], hermes_wheel: Path, profile_wheel: Path, verifier_relative_path: str, provenance_relative_path: str, verifier_result: Mapping[str, object], source_verifier_result: Mapping[str, object], source_bundle_relative_path: str, installed_bundle_relative_path: str, outer_manifest_relative_path: str, wheelhouse: Path | None = None, ) -> dict[str, object]: """Build, but do not write or freeze, one successor expected-state document.""" authority_result = _object( verifier_result.get("candidate_authority"), "candidate authority result" ) _validate_portable_origin_contract(runtime_portable) wheelhouse_inventory = ( dict(_EMPTY_WHEELHOUSE) if wheelhouse is None else build_wheelhouse_inventory( wheelhouse, required_wheels=(profile_wheel, hermes_wheel) ) ) _, _, outer_product_binding = _verify_outer_candidate_manifest( bundle, outer_manifest_relative_path ) wheelhouse_entries = wheelhouse_inventory.get("entries") if not isinstance(wheelhouse_entries, list): raise ValueError("wheelhouse entries are invalid") validate_product_binding( outer_product_binding, expected_tool_hashes=qualification_tool_hashes( scripts_dir=bundle / "verification-tools", package_root=Path(__file__).resolve().parents[2], ), expected_trust_boundary_sha256=trust_boundary_digest(TRUST_BOUNDARY), expected_wheelhouse_inventory_sha256=str( wheelhouse_inventory["inventory_sha256"] ), expected_wheelhouse_entry_count=len(wheelhouse_entries), actual_hermes_wheel=hermes_wheel, actual_profile_wheel=profile_wheel, ) runtime_expectation = dict(runtime_portable) original_audit = _validate_nonportable_record_audit( dict(original_nonportable_record_audit), install_role="original", candidate_digest=candidate_digest, runtime_portable=runtime_expectation, ) sealed_ty_attestation = _validate_hermetic_ty_attestation( dict(hermetic_ty_attestation), candidate_digest=candidate_digest, product_binding=outer_product_binding, ) candidate_parity = build_candidate_parity( bundle, candidate_digest=candidate_digest, source_result=source_verifier_result, installed_result=verifier_result, source_bundle_relative_path=source_bundle_relative_path, installed_bundle_relative_path=installed_bundle_relative_path, outer_manifest_relative_path=outer_manifest_relative_path, ) source_projection = _object( candidate_parity.get("source"), "source candidate projection" ) installed_projection = _object( candidate_parity.get("installed"), "installed candidate projection" ) document: dict[str, object] = { "schema": "task26-sealed-final-state-v6", "candidate_digest": candidate_digest, "runtime_portable": runtime_expectation, "original_nonportable_record_audit": original_audit, "hermetic_ty_attestation": sealed_ty_attestation, "hermes_wheel": { "filename": hermes_wheel.name, "sha256": sha256_file(hermes_wheel), }, "profile_wheel": { "filename": profile_wheel.name, "sha256": sha256_file(profile_wheel), }, "wheelhouse": wheelhouse_inventory, "verifier": { "relative_path": verifier_relative_path, "sha256": sha256_file(bundle / verifier_relative_path), }, "provenance_helper": { "relative_path": provenance_relative_path, "sha256": sha256_file(bundle / provenance_relative_path), }, "frozen_bootstrap": { "relative_path": "verification-tools/task26_frozen_bootstrap.py", "sha256": sha256_file( bundle / "verification-tools/task26_frozen_bootstrap.py" ), }, "independent_candidate_verifier": { "relative_path": ( "verification-tools/independent_verify_candidate.py" ), "sha256": sha256_file( bundle / "verification-tools/independent_verify_candidate.py" ), }, "expected_status": verifier_result.get("status"), "expected_result_sha256": hashlib.sha256( canonical(_portable_verifier_result(verifier_result)) ).hexdigest(), "authority": { "registry_sha256": sha256_file( bundle / "candidate-authority/registry.json" ), "qualification-ledger_sha256": sha256_file( bundle / "candidate-authority/qualification-ledger.json" ), "registry_head_sha256": authority_result.get("registry_head_sha256"), "ledger_head_sha256": authority_result.get("ledger_head_sha256"), }, "candidate_parity": candidate_parity, "local_socket_transcripts": { "source": source_projection["local_socket_transcript"], "installed": installed_projection["local_socket_transcript"], }, "trust_boundary": TRUST_BOUNDARY, } document["document_sha256"] = hashlib.sha256(canonical(document)).hexdigest() return document def verify_runtime_origin_proof( runtime: Mapping[str, object], site_packages: Path, probe: object, *, controlled_cwd: Path | None = None, ) -> dict[str, object]: """Bind imported package origins to exact installed inventory rows.""" evidence = _object(probe, "rehydrated origin probe") modules = _object(evidence.get("modules"), "rehydrated module origins") if ( evidence.get("schema") != "task26-rehydrated-origin-probe-v1" or evidence.get("isolated") != 1 or evidence.get("pythonpath_present") is not False or set(modules) != {"gateway", "checkin_cli"} or not isinstance(evidence.get("cwd"), str) or ( controlled_cwd is not None and Path(str(evidence["cwd"])).resolve(strict=True) != controlled_cwd.resolve(strict=True) ) ): raise ValueError("rehydrated Python isolation proof is invalid") distributions = _object(runtime.get("distributions"), "runtime distributions") proof_rows: list[dict[str, object]] = [] for module, role in (("gateway", "hermes"), ("checkin_cli", "profile")): raw_path = modules.get(module) if not isinstance(raw_path, str): raise ValueError("rehydrated module origin is invalid") path = Path(raw_path) try: resolved = path.resolve(strict=True) relative = resolved.relative_to(site_packages.resolve(strict=True)).as_posix() except (OSError, ValueError) as exc: raise ValueError( "module origin is outside rehydrated site-packages" ) from exc metadata = resolved.stat() if ( resolved.is_symlink() or not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1 ): raise ValueError("rehydrated module origin is not a regular file") distribution = _object(distributions.get(role), f"{role} distribution") inventory = distribution.get("installed_inventory") if not isinstance(inventory, list): raise ValueError("installed inventory is invalid") matches = [ _object(row, "installed inventory row") for row in inventory if isinstance(row, dict) and row.get("path") == relative ] digest = sha256_file(resolved) if ( len(matches) != 1 or matches[0].get("sha256") != digest or matches[0].get("size") != metadata.st_size ): raise ValueError("module origin differs from installed RECORD inventory") proof_rows.append( { "distribution": role, "module": module, "installed_relative_path": relative, "sha256": digest, "size": metadata.st_size, } ) return { "schema": "task26-installed-origin-proof-v1", "execution": { "cwd_authority": "private_empty_controlled", "python_isolated": True, "pythonpath_inherited": False, }, "modules": proof_rows, } def _portable_runtime(runtime: Mapping[str, object]) -> dict[str, object]: distributions = _object(runtime.get("distributions"), "runtime distributions") portable_distributions: dict[str, object] = {} for role in ("profile", "hermes"): distribution = _object(distributions.get(role), f"{role} distribution") raw_inventory = distribution.get("installed_inventory") if not isinstance(raw_inventory, list) or any( not isinstance(row, dict) for row in raw_inventory ): raise ValueError(f"{role} installed inventory is invalid") inventory = [ row for row in raw_inventory if ".." not in Path(str(row.get("path", ""))).parts and not str(row.get("path", "")).endswith(".dist-info/RECORD") ] portable_distribution: dict[str, object] = { key: distribution.get(key) for key in ( "distribution_name", "distribution_version", "metadata_sha256", "wheel_filename", "wheel_sha256", "wheel_record_sha256", ) } projection = _object( distribution.get("portable_record_projection"), f"{role} portable RECORD projection", ) portable_distribution["portable_record_projection_sha256"] = projection.get( "projection_sha256" ) portable_distribution["installed_inventory"] = inventory portable_distribution["installed_inventory_sha256"] = hashlib.sha256( canonical(inventory) ).hexdigest() portable_distributions[role] = portable_distribution interpreter = _object(runtime.get("interpreter"), "runtime interpreter") return { "schema": "installed-golden-runtime-portable-v2", "interpreter": { "sha256": interpreter.get("sha256"), "version": interpreter.get("version"), }, "distributions": portable_distributions, "origin_proof": runtime.get("origin_proof"), } def _clean_environment() -> dict[str, str]: environment = { key: value for key, value in os.environ.items() if not key.startswith("PYTHON") and not key.startswith("PIP") } environment.update( { "PYTHONNOUSERSITE": "1", "PIP_CONFIG_FILE": os.devnull, "PIP_NO_INDEX": "1", "PIP_DISABLE_PIP_VERSION_CHECK": "1", } ) return environment def _run_python( command: Sequence[str], *, controlled_cwd: Path, environment: Mapping[str, str], ) -> subprocess.CompletedProcess[str]: _private(controlled_cwd, directory=True) if any(controlled_cwd.iterdir()): raise ValueError("controlled Python cwd is not empty") completed = subprocess.run( list(command), check=True, cwd=controlled_cwd, env=dict(environment), capture_output=True, text=True, ) if any(controlled_cwd.iterdir()): raise ValueError("Python subprocess wrote to controlled cwd") return completed def _verify_bound_wheel_hashes( expected: Mapping[str, object], hermes_wheel: Path, profile_wheel: Path ) -> None: for label, path in ( ("hermes_wheel", hermes_wheel), ("profile_wheel", profile_wheel), ): binding = _object(expected.get(label), label) if binding != {"filename": path.name, "sha256": sha256_file(path)}: raise ValueError(f"sealed {label} digest differs") def rehydrate_and_verify( bundle: Path, expected_path: Path, *, hermes_wheel: Path, profile_wheel: Path, wheelhouse: Path | None = None, temp_parent: Path | None = None, ) -> dict[str, object]: """Create an isolated offline runtime, rerun verification, and scrub it.""" expected = verify_expected_state( bundle, expected_path, hermes_wheel=hermes_wheel, profile_wheel=profile_wheel, ) inventory = verify_wheelhouse_inventory( wheelhouse, expected.get("wheelhouse"), required_wheels=(profile_wheel, hermes_wheel), ) temporary = Path(tempfile.mkdtemp(prefix="task26-rehydrate-", dir=temp_parent)) temporary.chmod(0o700) controlled_cwd = temporary / "controlled-cwd" controlled_cwd.mkdir(mode=0o700) environment = _clean_environment() try: venv = temporary / "venv" _run_python( [sys.executable, "-I", "-m", "venv", str(venv)], controlled_cwd=controlled_cwd, environment=environment, ) venv.chmod(0o700) python = venv / ("Scripts/python.exe" if os.name == "nt" else "bin/python") # Recompute immediately before install, then pass only these exact paths. inventory = verify_wheelhouse_inventory( wheelhouse, inventory, required_wheels=(profile_wheel, hermes_wheel), ) _verify_bound_wheel_hashes(expected, hermes_wheel, profile_wheel) dependency_paths = _bound_wheel_paths(wheelhouse, inventory) install = [ str(python), "-I", "-m", "pip", "install", "--no-index", "--no-deps", "--no-compile", *(str(path) for path in dependency_paths), str(profile_wheel), str(hermes_wheel), ] _run_python( install, controlled_cwd=controlled_cwd, environment=environment ) site_packages_text = _run_python( [ str(python), "-I", "-c", "import sysconfig; print(sysconfig.get_paths()['purelib'])", ], controlled_cwd=controlled_cwd, environment=environment, ).stdout.strip() site_packages = Path(site_packages_text) runtime_path = temporary / "runtime.json" helper = bundle / str( _object(expected["provenance_helper"], "provenance helper")[ "relative_path" ] ) probe = ( "import hashlib,importlib.util,json,os,pathlib,sys;" "p=pathlib.Path(sys.argv[1]);" "s=importlib.util.spec_from_file_location('sealed_provenance',p);" "m=importlib.util.module_from_spec(s);s.loader.exec_module(m);" "r=m.collect_installed_runtime(venv=pathlib.Path(sys.argv[2])," "site_packages=pathlib.Path(sys.argv[3])," "profile_wheel=pathlib.Path(sys.argv[4])," "hermes_wheel=pathlib.Path(sys.argv[5]));" "a=m.nonportable_installed_record_audit(r,install_role='rehydrated'," "candidate_digest=sys.argv[6]);" "import checkin_cli,gateway;" "o={'schema':'task26-rehydrated-origin-probe-v1'," "'isolated':sys.flags.isolated,'pythonpath_present':'PYTHONPATH' in os.environ," "'cwd':os.getcwd(),'modules':{'gateway':gateway.__file__," "'checkin_cli':checkin_cli.__file__}};" "print(json.dumps({'runtime':r,'record_audit':a,'origin_probe':o}," "sort_keys=True,separators=(',',':')))" ) probe_output = _run_python( [ str(python), "-I", "-c", probe, str(helper), str(venv), str(site_packages), str(profile_wheel), str(hermes_wheel), str(expected.get("candidate_digest", "")), ], controlled_cwd=controlled_cwd, environment=environment, ).stdout probe_document = _object( json.loads(probe_output), "rehydrated probe document" ) runtime_document = _object( probe_document.get("runtime"), "rehydrated runtime" ) origin_proof = verify_runtime_origin_proof( runtime_document, site_packages, probe_document.get("origin_probe"), controlled_cwd=controlled_cwd, ) runtime_document["origin_proof"] = origin_proof rehydrated_portable = _portable_runtime(runtime_document) rehydrated_audit = probe_document.get("record_audit") if rehydrated_portable != expected.get("runtime_portable"): raise ValueError("rehydrated runtime differs from sealed portable bytes") sealed_ty = _validate_hermetic_ty_attestation( expected.get("hermetic_ty_attestation"), candidate_digest=str(expected.get("candidate_digest", "")), product_binding=_object( _verify_outer_candidate_manifest( bundle, str( _object( _object( expected.get("candidate_parity"), "candidate parity", ).get("outer_manifest"), "outer manifest parity", ).get("relative_path", "") ), )[2], "outer product binding", ), ) if wheelhouse is None: raise ValueError("sealed corrected Ty wheelhouse is unavailable") ty_wheel_matches = [ wheelhouse / str(_object(row, "wheelhouse row").get("relative_path")) for row in cast(list[object], inventory.get("entries", [])) if _object(row, "wheelhouse row").get("sha256") == sealed_ty.get("corrected_ty_wheel_sha256") ] if len(ty_wheel_matches) != 1: raise ValueError("sealed corrected Ty wheel is unavailable") ty_output = temporary / "ty-attestation" ty_runner = bundle / "verification-tools/task26_hermetic_ty_runner.py" ty_gate = bundle / "verification-tools/task26_ty_surface_gate.py" ty_completed = _run_python( [ str(python), "-I", str(ty_runner), "--source", str(site_packages / "gateway/platforms/nutrition_coaching.py"), "--ty-wheel", str(ty_wheel_matches[0]), "--expected-wheel-sha256", str(sealed_ty["corrected_ty_wheel_sha256"]), "--expected-executable-sha256", str(sealed_ty["binary_member_sha256"]), "--gate-tool", str(ty_gate), "--output-directory", str(ty_output), ], controlled_cwd=controlled_cwd, environment=environment, ) ty_result = _object( json.loads(ty_completed.stdout), "rehydrated hermetic Ty result" ) ty_provenance = _object( ty_result.get("execution_provenance"), "rehydrated Ty execution provenance", ) ty_receipt = _object( ty_result.get("surface_receipt"), "rehydrated Ty surface receipt" ) if ( ty_result.get("ty_wheel_sha256") != sealed_ty.get("corrected_ty_wheel_sha256") or ty_result.get("installed_executable_sha256") != sealed_ty.get("binary_member_sha256") or ty_result.get("raw_diagnostics_sha256") != sealed_ty.get("raw_diagnostics_sha256") or ty_provenance.get("raw_diagnostics_count") != sealed_ty.get("raw_diagnostics_count") or ty_provenance.get("provenance_sha256") != sealed_ty.get("execution_provenance_sha256") or ty_receipt.get("receipt_sha256") != sealed_ty.get("receipt_sha256") ): raise ValueError("rehydrated hermetic Ty attestation differs") runtime_path.write_bytes(canonical(runtime_document) + b"\n") runtime_path.chmod(0o600) verifier = bundle / str( _object(expected["verifier"], "verifier")["relative_path"] ) parity = verify_candidate_parity( bundle, expected.get("candidate_parity"), expected_candidate=str(expected.get("candidate_digest", "")), ) def run_verifier(relative_key: str, label: str) -> dict[str, Any]: relative = Path(str(parity.get(relative_key, ""))) target_bundle = bundle / relative completed = _run_python( [ str(python), "-I", str(verifier), str(target_bundle), "--rehydrated-runtime", str(runtime_path), ], controlled_cwd=controlled_cwd, environment=environment, ) return _object(json.loads(completed.stdout), label) source_result = run_verifier( "source_bundle_relative_path", "rehydrated source verifier result" ) result = run_verifier( "installed_bundle_relative_path", "rehydrated installed verifier result" ) if ( _candidate_result_projection(source_result, mode="source") != parity.get("source") or _candidate_result_projection(result, mode="installed") != parity.get("installed") ): raise ValueError("rehydrated source/installed candidate parity differs") authority = _object( result.get("candidate_authority"), "rehydrated authority" ) expected_authority = _object( expected.get("authority"), "expected authority" ) if ( result.get("status") != expected.get("expected_status") or result.get("candidate_digest") != expected.get("candidate_digest") or authority.get("registry_head_sha256") != expected_authority.get("registry_head_sha256") or authority.get("ledger_head_sha256") != expected_authority.get("ledger_head_sha256") or hashlib.sha256( canonical(_portable_verifier_result(result)) ).hexdigest() != expected.get("expected_result_sha256") ): raise ValueError( "rehydrated verifier result differs from sealed expected state" ) result["installed_record_provenance"] = ( verify_nonportable_record_audit_pair( expected.get("original_nonportable_record_audit"), rehydrated_audit, candidate_digest=str(expected.get("candidate_digest", "")), original_runtime_portable=_object( expected.get("runtime_portable"), "original portable runtime", ), rehydrated_runtime_portable=rehydrated_portable, ) ) result["hermetic_ty_attestation"] = ty_result result["rehydration_origin_proof"] = origin_proof return result finally: # Verify immutable inputs again after every success or failure path. try: verify_wheelhouse_inventory( wheelhouse, inventory, required_wheels=(profile_wheel, hermes_wheel), ) _verify_bound_wheel_hashes(expected, hermes_wheel, profile_wheel) finally: if temporary.exists(): for path in temporary.rglob("*"): try: if path.is_symlink(): continue path.chmod(0o700 if path.is_dir() else 0o600) except OSError: pass shutil.rmtree(temporary, ignore_errors=False) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("bundle", type=Path) parser.add_argument("expected_state", type=Path) parser.add_argument("--hermes-wheel", type=Path, required=True) parser.add_argument("--profile-wheel", type=Path, required=True) parser.add_argument("--wheelhouse", type=Path) parser.add_argument("--temp-parent", type=Path) args = parser.parse_args() try: result = rehydrate_and_verify( args.bundle, args.expected_state, hermes_wheel=args.hermes_wheel, profile_wheel=args.profile_wheel, wheelhouse=args.wheelhouse, temp_parent=args.temp_parent, ) except ( OSError, ValueError, subprocess.SubprocessError, json.JSONDecodeError, ) as exc: print( json.dumps( {"status": "TASK26_REHYDRATION_FAIL", "reason": str(exc)}, sort_keys=True, ), file=sys.stderr, ) return 1 print(json.dumps(result, sort_keys=True, separators=(",", ":"))) return 0 if __name__ == "__main__": raise SystemExit(main())