#!/usr/bin/env python3 """Stdlib-only postfreeze verifier for a cleaned Task26 candidate bundle.""" from __future__ import annotations import base64 import csv import hashlib import json import os import shutil import stat import subprocess import sys import tempfile import zipfile from pathlib import Path from typing import cast _OUTPUT_LIMIT = 65536 _POSTFREEZE_SEAL = "postfreeze-seal.json" _DELIVERED_SEAL = "delivered-bundle-seal.json" _TY_TARGET_PYTHON_VERSION = "3.12" _TY_RAW_DIAGNOSTICS_SHA256 = "d01195a6c0de359b77948625d1e97c35cffbc7dcd145daaa4eed1a5c252ce063" _TY_DIAGNOSTICS_FINGERPRINT_SHA256 = "ba03fe2e484683a8bdba0121f014caf614891c84da06a674a25314da1fce658a" _TY_COMMAND_ARGV = [ "sealed-ty", "check", "gateway/platforms/nutrition_coaching.py", "--output-format", "gitlab", "--python-version", _TY_TARGET_PYTHON_VERSION, ] _EXPECTED_STATE_KEYS = { "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", "profile_qualification", "trust_boundary", "document_sha256", } _PROFILE_QUALIFICATION_KEYS = { "schema", "profile_source_tree_digest", "source_tree_sha256_before", "source_tree_sha256_after", "profile_wheel_sha256", "command_identity", "environment_contract", "source_root_role", "timeout_seconds", "test_count", "status", "exit_code", "interpreter", "pytest_version", "runner_tool", "stdout", "stderr", "collection_manifest", "receipt_sha256", } _PROFILE_MANIFEST_KEYS = { "schema", "pytest_version", "interpreter_version", "interpreter_executable_sha256", "collected_count", "executed_count", "passed_count", "test_ids", "passed_test_ids", "exit_code", "manifest_sha256", } _PROFILE_RUNNER_SHA256 = "ef7c3ce7adef58da280e53e3567ec8cdfc867f8d4a116050d7b45af3db5e65f9" _PROFILE_COMMAND = { "executable_role": "selected_qualification_python", "argv": ["-I", "qualification/profile-pytest-runner.py", "qualification/profile-collection-manifest.json"], "pytest_argv": ["-q", "-p", "no:cacheprovider", "."], "working_directory_role": "profile_source_root", } _PROFILE_ENV = { "schema": "task26-profile-pytest-environment-v1", "inherited_variables": [], "private_cache": True, "variables": { "HOME": "private_cache_root", "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8", "PATH": "", "PIP_CONFIG_FILE": "os.devnull", "PIP_NO_INDEX": "1", "PYTHONDONTWRITEBYTECODE": "1", "PYTHONNOUSERSITE": "1", "PYTEST_ADDOPTS": "", "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1", "TMPDIR": "private_cache_root", "XDG_CACHE_HOME": "private_cache_root", }, } class _BootstrapFailure(RuntimeError): def __init__(self, document: dict[str, object]) -> None: super().__init__("frozen bootstrap failed") self.document = document def stream_diagnostic(data: bytes) -> dict[str, object]: try: data.decode("utf-8", errors="strict") valid_utf8 = True except UnicodeDecodeError: valid_utf8 = False return { "byte_count": len(data), "sha256": hashlib.sha256(data).hexdigest(), "valid_utf8": valid_utf8, "truncated": len(data) > _OUTPUT_LIMIT, "text": data[:_OUTPUT_LIMIT].decode("utf-8", errors="replace"), } def parsed_json(data: bytes) -> object | None: try: return json.loads(data.decode("utf-8", errors="strict")) except (UnicodeDecodeError, json.JSONDecodeError): return None PACKAGE_TOOL_MEMBERS = { "candidate_authority": "gateway/platforms/task26_candidate_authority.py", "candidate_derivation": "gateway/platforms/task26_candidate_derivation.py", "commit_observer": "gateway/commit_observer.py", "dualcoach_controller": "gateway/platforms/dualcoach_tasks21_25_controller.py", "evidence_contract": "gateway/platforms/task26_evidence_contract.py", "final_state": "gateway/platforms/task26_final_state.py", "runtime_authority": "gateway/platforms/task26_runtime_authority.py", "telegram_adapter": "gateway/platforms/telegram.py", } SCRIPT_TOOL_PATHS = { "delivered_bundle_sealer": "verification-tools/task26_seal_delivered_bundle.py", "frozen_bootstrap": "verification-tools/task26_frozen_bootstrap.py", "independent_candidate_verifier": "verification-tools/independent_verify_candidate.py", "installed_provenance": "verification-tools/installed_wheel_provenance.py", "local_telegram_qa": "verification-tools/task26_local_telegram_qa.py", "source_golden_path": "verification-tools/source_golden_path.py", "ty_hermetic_runner": "verification-tools/task26_hermetic_ty_runner.py", "ty_surface_gate": "verification-tools/task26_ty_surface_gate.py", "ty_wheel_builder": "verification-tools/task26_build_ty_executable_wheel.py", "verify_source_golden_path": "verification-tools/verify_source_golden_path.py", } def canonical(value: object) -> bytes: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() def sha(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"unsafe sealed path: {path}") def object_value(value: object, label: str) -> dict[str, object]: 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, object], value) def scan_inventory(bundle: Path) -> list[dict[str, object]]: private(bundle, True) rows: list[dict[str, object]] = [] for path in sorted(bundle.rglob("*")): if path.is_symlink(): raise ValueError("sealed bundle contains a symlink") relative = path.relative_to(bundle).as_posix() if relative in {_POSTFREEZE_SEAL, _DELIVERED_SEAL}: continue private(path, path.is_dir()) row: dict[str, object] = { "relative_path": relative, "kind": "directory" if path.is_dir() else "file", "mode": stat.S_IMODE(path.stat().st_mode), } if path.is_file(): row.update({"sha256": sha(path), "size": path.stat().st_size}) rows.append(row) return rows def delivered_inventory(bundle: Path) -> list[dict[str, object]]: rows: list[dict[str, object]] = [] for path in sorted(bundle.rglob("*")): relative = path.relative_to(bundle).as_posix() if relative == _DELIVERED_SEAL: continue if path.is_symlink(): raise ValueError("sealed bundle contains a symlink") private(path, path.is_dir()) row: dict[str, object] = { "relative_path": relative, "kind": "directory" if path.is_dir() else "file", "mode": stat.S_IMODE(path.stat().st_mode), } if path.is_file(): row.update({"sha256": sha(path), "size": path.stat().st_size}) rows.append(row) return rows def receipt_inventory(receipt_root: Path) -> list[dict[str, object]]: private(receipt_root, True) rows: list[dict[str, object]] = [] for path in sorted(receipt_root.rglob("*")): relative = path.relative_to(receipt_root).as_posix() if relative == "SEAL.json": continue if path.is_symlink() or not path.is_file(): if path.is_dir(): private(path, True) continue raise ValueError("postfreeze receipt root contains an unsafe path") private(path) rows.append({"path": relative, "sha256": sha(path), "size": path.stat().st_size}) return rows def verify_delivered_bundle(bundle: Path) -> tuple[Path, str]: final_path = bundle / _DELIVERED_SEAL private(final_path) final_bytes = final_path.read_bytes() final = object_value(json.loads(final_bytes), "delivered bundle seal") unsigned = {key: value for key, value in final.items() if key != "seal_sha256"} relative_receipt = final.get("receipt_root_relative_path") if ( set(final) != { "schema", "candidate_digest", "prefreeze_seal_sha256", "postfreeze_seal_sha256", "permitted_postfreeze_additions", "receipt_root_relative_path", "inventory", "inventory_sha256", "seal_sha256", } or final.get("schema") != "task26-delivered-bundle-seal-v1" or final.get("permitted_postfreeze_additions") != [_POSTFREEZE_SEAL, _DELIVERED_SEAL] or not isinstance(relative_receipt, str) or Path(relative_receipt).is_absolute() or final.get("seal_sha256") != hashlib.sha256(canonical(unsigned)).hexdigest() ): raise ValueError("delivered bundle seal differs") receipt_binding = bundle / relative_receipt if receipt_binding.is_symlink(): raise ValueError("postfreeze receipt root binding is invalid") receipt_root = receipt_binding.resolve(strict=True) expected_receipt_relative = Path(os.path.relpath(receipt_root, bundle)).as_posix() if ( receipt_root.parent != bundle.parent or receipt_root == bundle or relative_receipt != expected_receipt_relative ): raise ValueError("postfreeze receipt root binding is invalid") final_inventory = delivered_inventory(bundle) if ( final.get("inventory") != final_inventory or final.get("inventory_sha256") != hashlib.sha256(canonical(final_inventory)).hexdigest() or final.get("postfreeze_seal_sha256") != sha(bundle / _POSTFREEZE_SEAL) or final.get("prefreeze_seal_sha256") != sha(bundle / "SEAL.json") ): raise ValueError("delivered bundle inventory differs") prefreeze = object_value( json.loads((bundle / "SEAL.json").read_text(encoding="utf-8")), "prefreeze seal", ) prefreeze_unsigned = { key: value for key, value in prefreeze.items() if key != "document_sha256" } if prefreeze.get("document_sha256") != hashlib.sha256(canonical(prefreeze_unsigned)).hexdigest(): raise ValueError("prefreeze seal differs") receipt_seal_path = receipt_root / "SEAL.json" pass_path = receipt_root / "PASS.json" receipt_postfreeze = receipt_root / _POSTFREEZE_SEAL for path in (receipt_seal_path, pass_path, receipt_postfreeze): private(path) if receipt_postfreeze.read_bytes() != (bundle / _POSTFREEZE_SEAL).read_bytes(): raise ValueError("receipt and delivered postfreeze seals differ") receipt_seal = object_value(json.loads(receipt_seal_path.read_bytes()), "receipt seal") receipt_unsigned = { key: value for key, value in receipt_seal.items() if key != "document_sha256" } files = receipt_inventory(receipt_root) if ( receipt_seal.get("schema") != "task26-postfreeze-receipt-seal-v2" or receipt_seal.get("document_sha256") != hashlib.sha256(canonical(receipt_unsigned)).hexdigest() or receipt_seal.get("files") != files or receipt_seal.get("file_count") != len(files) or receipt_seal.get("inventory_sha256") != hashlib.sha256(canonical(files)).hexdigest() ): raise ValueError("postfreeze receipt seal differs") passed = object_value(json.loads(pass_path.read_bytes()), "PASS receipt") passed_unsigned = {key: value for key, value in passed.items() if key != "receipt_sha256"} delivery = object_value(passed.get("delivered_bundle"), "PASS delivered bundle") postfreeze = object_value(passed.get("postfreeze_seal"), "PASS postfreeze seal") expected_delivery_relative = Path(os.path.relpath(bundle, receipt_root)).as_posix() if ( passed.get("status") != "PASS" or passed.get("candidate_digest") != final.get("candidate_digest") or passed.get("receipt_sha256") != hashlib.sha256(canonical(passed_unsigned)).hexdigest() or delivery != { "relative_path": expected_delivery_relative, "seal_relative_path": _DELIVERED_SEAL, "seal_sha256": hashlib.sha256(final_bytes).hexdigest(), } or postfreeze != { "receipt_relative_path": _POSTFREEZE_SEAL, "delivered_relative_path": _POSTFREEZE_SEAL, "sha256": sha(bundle / _POSTFREEZE_SEAL), } ): raise ValueError("PASS receipt delivery binding differs") return receipt_root, hashlib.sha256(final_bytes).hexdigest() def verify_record(wheel: Path) -> None: private(wheel) with zipfile.ZipFile(wheel) 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") records = [name for name in names if name.endswith(".dist-info/RECORD")] if len(records) != 1: raise ValueError("wheel RECORD cardinality differs") rows = list(csv.reader(archive.read(records[0]).decode("utf-8").splitlines())) recorded = {row[0] for row in rows if len(row) == 3} if recorded != set(names): raise ValueError("wheel RECORD inventory differs") for name, encoded, size in rows: data = archive.read(name) if size and int(size) != len(data): raise ValueError("wheel RECORD size differs") if encoded: algorithm, value = encoded.split("=", 1) if algorithm != "sha256": raise ValueError("wheel RECORD algorithm differs") actual = base64.urlsafe_b64encode(hashlib.sha256(data).digest()).rstrip(b"=").decode() if actual != value: raise ValueError("wheel RECORD hash differs") def locate_bound_wheel(bundle: Path, binding: dict[str, object], label: str) -> Path: filename = binding.get("filename") if not isinstance(filename, str) or Path(filename).name != filename: raise ValueError(f"{label} filename is invalid") matches = [path for path in bundle.rglob(filename) if path.is_file() and not path.is_symlink()] if len(matches) != 1: raise ValueError(f"{label} cardinality differs") wheel = matches[0] if sha(wheel) != binding.get("sha256"): raise ValueError(f"{label} hash differs") verify_record(wheel) return wheel def verify_candidate_and_tools(bundle: Path, expected: dict[str, object], hermes: Path, profile: Path) -> None: parity = object_value(expected.get("candidate_parity"), "candidate parity") outer = object_value(parity.get("outer_manifest"), "outer manifest binding") relative = outer.get("relative_path") if not isinstance(relative, str) or Path(relative).is_absolute() or ".." in Path(relative).parts: raise ValueError("candidate document path is invalid") outer_path = bundle / relative private(outer_path) if sha(outer_path) != outer.get("sha256"): raise ValueError("candidate document hash differs") outer_document = object_value(json.loads(outer_path.read_text(encoding="utf-8")), "candidate document") binding = object_value(outer_document.get("product_binding"), "product binding") binding_unsigned = {key: value for key, value in binding.items() if key != "binding_sha256"} if binding.get("binding_sha256") != hashlib.sha256(canonical(binding_unsigned)).hexdigest(): raise ValueError("product binding hash differs") inputs = object_value(binding.get("derivation_inputs"), "candidate derivation inputs") if binding.get("candidate_digest") != hashlib.sha256(canonical(inputs)).hexdigest() or binding.get("candidate_digest") != expected.get("candidate_digest"): raise ValueError("candidate digest differs") if inputs.get("hermes_wheel_sha256") != sha(hermes) or inputs.get("profile_wheel_sha256") != sha(profile): raise ValueError("candidate wheel binding differs") claimed_tools = object_value(inputs.get("qualification_tool_sha256"), "qualification tools") hermetic_ty = object_value(inputs.get("hermetic_ty"), "hermetic Ty binding") if ( hermetic_ty.get("schema") != "task26-preexecution-hermetic-ty-binding-v2" or hermetic_ty.get("target_python_version") != _TY_TARGET_PYTHON_VERSION or hermetic_ty.get("qualification_interpreter_role") != "selected_qualification_python" or hermetic_ty.get("ty_command_argv") != _TY_COMMAND_ARGV or _TY_COMMAND_ARGV.count("--python-version") != 1 ): raise ValueError("hermetic Ty semantic target binding differs") actual_tools: dict[str, str] = {} for name, relative_path in SCRIPT_TOOL_PATHS.items(): target = bundle / relative_path private(target) actual_tools[name] = sha(target) with zipfile.ZipFile(hermes) as archive: for name, member in PACKAGE_TOOL_MEMBERS.items(): try: actual_tools[name] = hashlib.sha256(archive.read(member)).hexdigest() except KeyError as exc: raise ValueError(f"qualification tool is absent: {name}") from exc if claimed_tools != actual_tools: raise ValueError("qualification tool hashes differ") attestation = object_value( expected.get("hermetic_ty_attestation"), "sealed hermetic Ty attestation" ) interpreter_version = attestation.get("qualification_interpreter_version") attestation_unsigned = { key: value for key, value in attestation.items() if key != "attestation_sha256" } if ( set(attestation) != { "schema", "candidate_digest", "corrected_ty_wheel_sha256", "ty_version", "target_python_version", "qualification_interpreter_role", "qualification_interpreter_version", "ty_command_argv", "runner_tool_sha256", "gate_tool_sha256", "binary_member_sha256", "raw_diagnostics_sha256", "raw_diagnostics_count", "diagnostics_semantic_fingerprint_sha256", "execution_provenance_sha256", "receipt_sha256", "attestation_sha256", } or attestation.get("schema") != "task26-sealed-hermetic-ty-attestation-v2" or attestation.get("target_python_version") != _TY_TARGET_PYTHON_VERSION or attestation.get("qualification_interpreter_role") != "selected_qualification_python" or not isinstance(interpreter_version, list) or interpreter_version[:2] != [3, 12] or attestation.get("ty_command_argv") != _TY_COMMAND_ARGV or attestation.get("candidate_digest") != expected.get("candidate_digest") or attestation.get("corrected_ty_wheel_sha256") != hermetic_ty.get("corrected_ty_wheel_sha256") or attestation.get("binary_member_sha256") != hermetic_ty.get("binary_member_sha256") or attestation.get("runner_tool_sha256") != actual_tools.get("ty_hermetic_runner") or attestation.get("gate_tool_sha256") != actual_tools.get("ty_surface_gate") or attestation.get("raw_diagnostics_sha256") != _TY_RAW_DIAGNOSTICS_SHA256 or attestation.get("diagnostics_semantic_fingerprint_sha256") != _TY_DIAGNOSTICS_FINGERPRINT_SHA256 or attestation.get("attestation_sha256") != hashlib.sha256(canonical(attestation_unsigned)).hexdigest() ): raise ValueError("sealed Ty semantic target attestation differs") def verify_transcript_binding( bundle: Path, parity: dict[str, object], binding: object, *, mode: str, ) -> None: row = object_value(binding, f"{mode} transcript binding") expected_relative = f"data/task26-local-http-telegram-api-qa-{mode}.json" if ( row.get("schema") != "task26-local-http-telegram-api-transcript-binding-v1" or row.get("runtime_mode") != mode or row.get("relative_path") != expected_relative ): raise ValueError("local socket transcript binding is invalid") bundle_relative = parity.get(f"{mode}_bundle_relative_path") if not isinstance(bundle_relative, str): raise ValueError("transcript bundle path is invalid") relative_root = Path(bundle_relative) if relative_root.is_absolute() or ".." in relative_root.parts: raise ValueError("transcript bundle path is invalid") path = bundle / relative_root / expected_relative private(path) transcript = object_value( json.loads(path.read_text(encoding="utf-8")), f"{mode} socket transcript", ) methods = transcript.get("methods") transcript_hash = hashlib.sha256(canonical(transcript)).hexdigest() if ( row.get("transcript_sha256") != transcript_hash or transcript.get("schema") != "task26-local-http-telegram-api-qa-v1" or transcript.get("runtime_mode") != mode or transcript.get("actual_socket") is not True or transcript.get("mock_network") is not False or transcript.get("revocation_committed") is not True or transcript.get("disconnect") != "PASS" or transcript.get("post_revoke_updates") != 0 or transcript.get("server_cleanup") is not True or transcript.get("socket_cleanup") is not True or transcript.get("server_socket_closed") is not True or transcript.get("active_connections") != 0 or transcript.get("watcher_cleanup") is not True or transcript.get("external_traffic") is not False or transcript.get("getMe_observed") is not True or transcript.get("getUpdates_observed") is not True or not isinstance(methods, list) or "getMe" not in methods or "getUpdates" not in methods ): raise ValueError("local socket transcript differs") def hex_digest(value: object) -> bool: return ( isinstance(value, str) and len(value) == 64 and all(character in "0123456789abcdef" for character in value) ) def verify_nonportable_direct_url_audit(expected: dict[str, object]) -> None: audit = object_value( expected.get("original_nonportable_record_audit"), "nonportable installed RECORD audit", ) unsigned = {key: value for key, value in audit.items() if key != "audit_sha256"} distributions = object_value(audit.get("distributions"), "raw audit distributions") portable = object_value(expected.get("runtime_portable"), "portable runtime") portable_distributions = object_value( portable.get("distributions"), "portable distributions" ) if ( set(audit) != { "schema", "install_role", "candidate_digest", "distributions", "raw_direct_url_hash_equality_required_or_claimed", "audit_sha256", } or audit.get("schema") != "task26-nonportable-installed-record-audit-v1" or audit.get("install_role") != "original" or audit.get("candidate_digest") != expected.get("candidate_digest") or audit.get("raw_direct_url_hash_equality_required_or_claimed") is not False or audit.get("audit_sha256") != hashlib.sha256(canonical(unsigned)).hexdigest() or set(distributions) != {"profile", "hermes"} ): raise ValueError("nonportable direct URL audit differs") for role in ("profile", "hermes"): row = object_value(distributions.get(role), f"{role} raw audit") portable_row = object_value( portable_distributions.get(role), f"portable {role} distribution" ) classifications = row.get("path_dependent_classifications") if ( set(row) != { "distribution", "wheel_sha256", "portable_projection_sha256", "raw_installed_record_sha256", "raw_installed_record_portable", "raw_direct_url_sha256", "raw_direct_url_record_row_sha256", "raw_direct_url_equality_required_or_claimed", "reason", "path_dependent_classifications", } or row.get("distribution") != portable_row.get("distribution_name") or row.get("wheel_sha256") != portable_row.get("wheel_sha256") or row.get("portable_projection_sha256") != portable_row.get("portable_record_projection_sha256") or not hex_digest(row.get("raw_installed_record_sha256")) or row.get("raw_installed_record_portable") is not False or not hex_digest(row.get("raw_direct_url_sha256")) or row.get("raw_direct_url_record_row_sha256") != row.get("raw_direct_url_sha256") or row.get("raw_direct_url_equality_required_or_claimed") is not False or not isinstance(classifications, list) or "direct_url_absolute_file_url" not in classifications or "direct_url_raw_record_hash" not in classifications ): raise ValueError("nonportable direct URL audit differs") def verify_portable_direct_url_provenance( expected: dict[str, object], profile: Path, hermes: Path ) -> None: runtime = object_value(expected.get("runtime_portable"), "portable runtime") distributions = object_value( runtime.get("distributions"), "portable distributions" ) if runtime.get("schema") != "installed-golden-runtime-portable-v2" or set( distributions ) != {"profile", "hermes"}: raise ValueError("portable runtime direct URL binding is invalid") for role, wheel in (("profile", profile), ("hermes", hermes)): distribution = object_value( distributions.get(role), f"portable {role} distribution" ) inventory = distribution.get("installed_inventory") if not isinstance(inventory, list): raise ValueError("portable direct URL inventory is invalid") rows: list[dict[str, object]] = [] for raw_row in inventory: if not isinstance(raw_row, dict): continue row = cast(dict[str, object], raw_row) if row.get("kind") == "canonical_direct_url_archive_provenance": rows.append(row) if len(rows) != 1: raise ValueError("portable direct URL provenance cardinality differs") row = rows[0] provenance = object_value( row.get("provenance"), "portable direct URL provenance" ) member = provenance.get("dist_info_member_path") wheel_hash = sha(wheel) if ( set(row) != {"path", "kind", "provenance"} or row.get("path") != member or not isinstance(member, str) or not member.endswith(".dist-info/direct_url.json") or set(provenance) != { "schema", "distribution", "dist_info_member_path", "sealed_wheel_basename", "sealed_wheel_sha256", "archive_sha256", } or provenance.get("schema") != "task26-portable-direct-url-provenance-v1" or provenance.get("distribution") != distribution.get("distribution_name") or provenance.get("sealed_wheel_basename") != wheel.name or provenance.get("sealed_wheel_sha256") != wheel_hash or provenance.get("archive_sha256") != wheel_hash ): raise ValueError("portable direct URL provenance differs") def _qualification_artifact(bundle: Path, value: object, relative: str) -> Path: binding = object_value(value, "profile qualification artifact") if set(binding) != {"relative_path", "sha256", "size"} or binding.get("relative_path") != relative: raise ValueError("profile qualification artifact binding is invalid") path = bundle / relative private(path) if binding.get("sha256") != sha(path) or binding.get("size") != path.stat().st_size: raise ValueError("profile qualification artifact differs") return path def verify_profile_qualification(bundle: Path, expected: dict[str, object], profile: Path) -> None: receipt = object_value(expected.get("profile_qualification"), "profile qualification") parity = object_value(expected.get("candidate_parity"), "candidate parity") source = object_value(parity.get("source"), "source candidate projection") receipt_path = bundle / "qualification/profile-qualification-receipt.json" private(receipt_path) persisted = object_value(json.loads(receipt_path.read_text(encoding="utf-8")), "persisted profile qualification") stdout = _qualification_artifact(bundle, receipt.get("stdout"), "qualification/profile-pytest.stdout") _qualification_artifact(bundle, receipt.get("stderr"), "qualification/profile-pytest.stderr") manifest_path = _qualification_artifact(bundle, receipt.get("collection_manifest"), "qualification/profile-collection-manifest.json") runner_path = bundle / "qualification/profile-pytest-runner.py" private(runner_path) manifest = object_value(json.loads(manifest_path.read_text(encoding="utf-8")), "profile pytest manifest") interpreter = object_value(receipt.get("interpreter"), "profile interpreter") runner = object_value(receipt.get("runner_tool"), "profile runner") ids = manifest.get("test_ids"); passed = manifest.get("passed_test_ids"); version = interpreter.get("version") timeout = receipt.get("timeout_seconds") unsigned = {key: value for key, value in receipt.items() if key != "receipt_sha256"} manifest_unsigned = {key: value for key, value in manifest.items() if key != "manifest_sha256"} if ( set(receipt) != _PROFILE_QUALIFICATION_KEYS or receipt.get("schema") != "task26-profile-qualification-receipt-v2" or receipt.get("profile_source_tree_digest") != source.get("source_tree_digest") or receipt.get("source_tree_sha256_before") != source.get("source_tree_digest") or receipt.get("source_tree_sha256_after") != source.get("source_tree_digest") or receipt.get("profile_wheel_sha256") != sha(profile) or receipt.get("command_identity") != _PROFILE_COMMAND or receipt.get("environment_contract") != _PROFILE_ENV or receipt.get("source_root_role") != "supplied_successor_profile_source_tree" or not isinstance(timeout, int) or not 1 <= timeout <= 900 or receipt.get("test_count") != 749 or type(receipt.get("test_count")) is not int or receipt.get("status") != "PASS" or receipt.get("exit_code") != 0 or runner != {"relative_path": "qualification/profile-pytest-runner.py", "sha256": _PROFILE_RUNNER_SHA256} or sha(runner_path) != _PROFILE_RUNNER_SHA256 or interpreter.get("executable_role") != "selected_qualification_python" or not hex_digest(interpreter.get("executable_sha256")) or not isinstance(version, list) or version[:2] != [3, 12] or receipt.get("pytest_version") != manifest.get("pytest_version") or manifest.get("interpreter_version") != version or manifest.get("interpreter_executable_sha256") != interpreter.get("executable_sha256") or set(manifest) != _PROFILE_MANIFEST_KEYS or manifest.get("schema") != "task26-profile-pytest-manifest-v1" or manifest.get("manifest_sha256") != hashlib.sha256(canonical(manifest_unsigned)).hexdigest() or not isinstance(ids, list) or len(ids) != 749 or len(set(ids)) != 749 or not isinstance(passed, list) or sorted(ids) != passed or any(manifest.get(key) != 749 for key in ("collected_count", "executed_count", "passed_count")) or manifest.get("exit_code") != 0 or b"749 passed in " not in stdout.read_bytes() or receipt.get("receipt_sha256") != hashlib.sha256(canonical(unsigned)).hexdigest() or persisted != receipt ): raise ValueError("profile qualification receipt differs") def verify_preexecution(bundle: Path) -> tuple[dict[str, object], Path, Path, Path]: expected_path = bundle / "sealed-expected-state.json" postfreeze_path = bundle / "postfreeze-seal.json" private(expected_path) private(postfreeze_path) expected = object_value(json.loads(expected_path.read_text(encoding="utf-8")), "expected state") unsigned = {key: value for key, value in expected.items() if key != "document_sha256"} if ( set(expected) != _EXPECTED_STATE_KEYS or expected.get("schema") != "task26-sealed-final-state-v7" or expected.get("document_sha256") != hashlib.sha256(canonical(unsigned)).hexdigest() ): raise ValueError("expected-state schema or hash differs") independent = object_value(expected.get("independent_candidate_verifier"), "independent verifier") verifier_path = bundle / str(independent.get("relative_path", "")) private(verifier_path) if independent.get("sha256") != sha(verifier_path) or sha(Path(__file__)) != sha(verifier_path): raise ValueError("independent verifier hash differs") postfreeze = object_value(json.loads(postfreeze_path.read_text(encoding="utf-8")), "postfreeze seal") postfreeze_unsigned = {key: value for key, value in postfreeze.items() if key != "seal_sha256"} if postfreeze.get("schema") != "task26-postfreeze-seal-v1" or postfreeze.get("seal_sha256") != hashlib.sha256(canonical(postfreeze_unsigned)).hexdigest(): raise ValueError("postfreeze seal hash differs") inventory = scan_inventory(bundle) if postfreeze.get("inventory") != inventory or postfreeze.get("inventory_sha256") != hashlib.sha256(canonical(inventory)).hexdigest() or postfreeze.get("expected_state_sha256") != sha(expected_path) or postfreeze.get("candidate_digest") != expected.get("candidate_digest"): raise ValueError("postfreeze inventory differs") hermes = locate_bound_wheel(bundle, object_value(expected.get("hermes_wheel"), "Hermes wheel"), "Hermes wheel") profile = locate_bound_wheel(bundle, object_value(expected.get("profile_wheel"), "profile wheel"), "profile wheel") verify_candidate_and_tools(bundle, expected, hermes, profile) verify_portable_direct_url_provenance(expected, profile, hermes) verify_nonportable_direct_url_audit(expected) verify_profile_qualification(bundle, expected, profile) parity = object_value(expected.get("candidate_parity"), "candidate parity") transcripts = object_value( expected.get("local_socket_transcripts"), "local socket transcripts" ) if set(transcripts) != {"source", "installed"}: raise ValueError("local socket transcript inventory differs") verify_transcript_binding( bundle, parity, transcripts["source"], mode="source" ) verify_transcript_binding( bundle, parity, transcripts["installed"], mode="installed" ) source_binding = object_value(transcripts["source"], "source transcript") installed_binding = object_value( transcripts["installed"], "installed transcript" ) if source_binding.get("transcript_sha256") == installed_binding.get( "transcript_sha256" ): raise ValueError("source and installed transcripts were substituted") authority = object_value(expected.get("authority"), "authority") for filename, field in (("registry.json", "registry_sha256"), ("qualification-ledger.json", "qualification-ledger_sha256")): target = bundle / "candidate-authority" / filename private(target) if sha(target) != authority.get(field): raise ValueError("sealed authority differs") return expected, expected_path, hermes, profile def clean_environment() -> dict[str, str]: environment = {key: value for key, value in os.environ.items() if not key.startswith(("PYTHON", "PIP")) and key != "PATH"} environment.update({"PATH": "", "PYTHONNOUSERSITE": "1", "PIP_NO_INDEX": "1", "PIP_CONFIG_FILE": os.devnull}) return environment def main() -> int: if len(sys.argv) != 2 or not sys.flags.isolated: print(json.dumps({"status": "TASK26_INDEPENDENT_VERIFY_FAIL", "reason": "isolated invocation required"}), file=sys.stderr) return 1 bundle = Path(sys.argv[1]).absolute() temporary: Path | None = None try: private(bundle, True) receipt_root, delivered_seal_sha256 = verify_delivered_bundle(bundle) expected, expected_path, hermes, profile = verify_preexecution(bundle) temporary = Path(tempfile.mkdtemp(prefix="task26-independent-verify-")) temporary.chmod(0o700) cwd = temporary / "cwd" cwd.mkdir(mode=0o700) bootstrap = bundle / "verification-tools/task26_frozen_bootstrap.py" command = [sys.executable, "-I", str(bootstrap), str(bundle)] completed = subprocess.run( command, cwd=cwd, env=clean_environment(), capture_output=True, check=False, ) if completed.returncode != 0: raise _BootstrapFailure( { "status": "TASK26_INDEPENDENT_VERIFY_FAIL", "stage": "frozen_bootstrap", "reason": "nested subprocess failed", "nested_process": { "command": command, "returncode": completed.returncode, "stdout": stream_diagnostic(completed.stdout), "stderr": stream_diagnostic(completed.stderr), "parsed_stdout": parsed_json(completed.stdout), "parsed_stderr": parsed_json(completed.stderr), }, } ) result = object_value( json.loads(completed.stdout.decode("utf-8", errors="strict")), "frozen verification result", ) if result.get("status") != "TASK26_FROZEN_BOOTSTRAP_PASS" or result.get("expected_state_sha256") != sha(expected_path): raise ValueError("installed frozen verification differs") receipt: dict[str, object] = { "schema": "task26-independent-candidate-verification-receipt-v1", "status": "TASK26_INDEPENDENT_CANDIDATE_PASS", "candidate_digest": expected.get("candidate_digest"), "expected_state_sha256": sha(expected_path), "postfreeze_seal_sha256": sha(bundle / _POSTFREEZE_SEAL), "delivered_bundle_seal_sha256": delivered_seal_sha256, "receipt_root": receipt_root.name, "hermes_wheel_sha256": sha(hermes), "profile_wheel_sha256": sha(profile), "runtime_source": "sealed_wheelhouse_only", "isolated": True, } receipt["receipt_sha256"] = hashlib.sha256(canonical(receipt)).hexdigest() print(json.dumps(receipt, sort_keys=True, separators=(",", ":"))) return 0 except _BootstrapFailure as exc: print( json.dumps(exc.document, sort_keys=True, separators=(",", ":")), file=sys.stderr, ) return 1 except BaseException as exc: if isinstance(exc, KeyboardInterrupt): raise print(json.dumps({"status": "TASK26_INDEPENDENT_VERIFY_FAIL", "stage": "preflight", "reason": str(exc)}, sort_keys=True), file=sys.stderr) return 1 finally: if temporary is not None and temporary.exists(): for path in temporary.rglob("*"): try: if not path.is_symlink(): path.chmod(0o700 if path.is_dir() else 0o600) except OSError: pass shutil.rmtree(temporary) if __name__ == "__main__": raise SystemExit(main())