#!/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 _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", "profile_wheel_sha256", "command_identity", "source_root_role", "test_count", "status", "log_relative_path", "log_sha256", "recorded_result_text", "receipt_sha256", } _PROFILE_COMMAND = { "executable_role": "selected_qualification_python", "argv": ["-I", "-m", "pytest", "-q"], "working_directory_role": "profile_source_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_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 = { "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 == "postfreeze-seal.json": 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 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") 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") 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 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" log_path = bundle / "qualification/profile-qualification.log" private(receipt_path) private(log_path) persisted = object_value( json.loads(receipt_path.read_text(encoding="utf-8")), "persisted profile qualification", ) unsigned = { key: value for key, value in receipt.items() if key != "receipt_sha256" } if ( set(receipt) != _PROFILE_QUALIFICATION_KEYS or receipt.get("schema") != "task26-profile-qualification-receipt-v1" or receipt.get("profile_source_tree_digest") != source.get("source_tree_digest") or receipt.get("profile_wheel_sha256") != sha(profile) or receipt.get("command_identity") != _PROFILE_COMMAND or receipt.get("source_root_role") != "supplied_successor_profile_source_tree" or receipt.get("test_count") != 749 or type(receipt.get("test_count")) is not int or receipt.get("status") != "PASS" or receipt.get("log_relative_path") != "qualification/profile-qualification.log" or receipt.get("log_sha256") != sha(log_path) or receipt.get("recorded_result_text") != "749 passed" or log_path.read_bytes() != b"749 passed\n" 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_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) 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.json"), "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())