#!/usr/bin/env python3 """Independently verify the Task22 compatibility implementation successor. The verifier reads only source/package paths declared in the chained manifests. It never opens onboarding ledgers, outboxes, journals, callbacks, or services. """ from __future__ import annotations import hashlib import json import os import stat import subprocess import sys import zipfile from collections import Counter from pathlib import Path from typing import Protocol, cast EVIDENCE = Path(__file__).resolve().parent PREDECESSOR = EVIDENCE / "dualcoach-task22-supplemental-successor-manifest.json" SUCCESSOR = EVIDENCE / "dualcoach-task22-supplemental-compatibility-manifest.json" WHEEL = EVIDENCE / "dualcoach-task22-supplemental-compatibility.whl" WHEEL_RECEIPT = EVIDENCE / "dualcoach-task22-supplemental-compatibility-wheel.json" FREEZE = EVIDENCE / "dualcoach-task22-supplemental-compatibility-freeze-receipt.json" _PREDECESSOR_CLASSIFICATION = ( "pinned supplemental authority successor retained; exact terminal recovery is offline-sealed" ) def _run( argv: list[str], *, cwd: Path, allowed: set[int] ) -> subprocess.CompletedProcess[bytes]: result = subprocess.run( argv, cwd=cwd, env={**os.environ, "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8"}, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, ) if result.returncode not in allowed or result.stderr: raise RuntimeError(f"command failed: {argv!r}") return result def _category( root: str, path: str, roots: dict[str, Path] ) -> tuple[str, bytes, int]: if root == "gateway": tracked = subprocess.run( ["git", "ls-files", "--error-unmatch", "--", path], cwd=roots[root], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, ).returncode == 0 if tracked: result = _run( [ "git", "diff", "--binary", "--no-ext-diff", "--no-textconv", "HEAD", "--", path, ], cwd=roots[root], allowed={0}, ) return "gateway-tracked", result.stdout, result.returncode result = _run( [ "git", "diff", "--no-index", "--binary", "--no-ext-diff", "--", "/dev/null", f"./{path}", ], cwd=roots[root], allowed={0, 1}, ) return "gateway-untracked", result.stdout, result.returncode result = _run( [ "git", "diff", "--no-index", "--binary", "--no-ext-diff", "--", "/dev/null", f"./{path}", ], cwd=roots[root], allowed={0, 1}, ) return "profile-package-external", result.stdout, result.returncode class HashDigest(Protocol): def update(self, data: bytes, /) -> None: ... def hexdigest(self) -> str: ... def _canonical(value: object) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False ).encode() def _verify_wheel( *, candidate_digest: str, manifest_digest: str, predecessor_manifest_digest: str, gateway_root: Path, manifest_wheel_provenance: object, ) -> dict[str, object]: receipt_bytes = WHEEL_RECEIPT.read_bytes() decoded = cast(object, json.loads(receipt_bytes)) if not isinstance(decoded, dict): raise RuntimeError("deterministic wheel receipt is invalid") receipt = cast(dict[str, object], decoded) unsigned = dict(receipt) digest = unsigned.pop("receipt_digest", None) if not isinstance(digest, str) or digest != hashlib.sha256(_canonical(unsigned)).hexdigest(): raise RuntimeError("deterministic wheel receipt authentication failed") if ( stat.S_IMODE(SUCCESSOR.stat().st_mode) != 0o600 or stat.S_IMODE(WHEEL.stat().st_mode) != 0o444 or stat.S_IMODE(WHEEL_RECEIPT.stat().st_mode) != 0o600 or receipt.get("schema") != "task22-supplemental-deterministic-wheel-v1" or receipt.get("candidate_digest") != candidate_digest or receipt.get("candidate_manifest_sha256") != manifest_digest or receipt.get("predecessor_manifest_sha256") != predecessor_manifest_digest or receipt.get("wheel_filename") != WHEEL.name or receipt.get("wheel_sha256") != hashlib.sha256(WHEEL.read_bytes()).hexdigest() or receipt.get("wheel_bytes") != WHEEL.stat().st_size or receipt.get("zip_timestamp") != "2000-01-01T00:00:00Z" or receipt.get("source_epochs") != [1700000000, 1800000000] or receipt.get("builder_relative_path") != "scripts/reproducible-wheel-build" or receipt.get("builder_sha256") != hashlib.sha256( (gateway_root / "scripts" / "reproducible-wheel-build").read_bytes() ).hexdigest() or receipt.get("build_contract") != "two isolated offline source copies with distinct mtimes produced byte-identical wheels" or receipt.get("reproduced_exactly") is not True or receipt.get("live_profile_accessed") is not False ): raise RuntimeError("deterministic wheel receipt binding failed") with zipfile.ZipFile(WHEEL) as archive: members = [ { "filename": item.filename, "compress_size": item.compress_size, "file_size": item.file_size, "crc": item.CRC, "date_time": list(item.date_time), } for item in archive.infolist() ] member_index = hashlib.sha256(_canonical(members)).hexdigest() if ( receipt.get("wheel_members") != len(members) or receipt.get("wheel_member_index_sha256") != member_index or any(member["date_time"] != [2000, 1, 1, 0, 0, 0] for member in members) or manifest_wheel_provenance != { "receipt_schema": "task22-supplemental-deterministic-wheel-v1", "filename": WHEEL.name, "sha256": receipt["wheel_sha256"], "bytes": receipt["wheel_bytes"], "members": len(members), "member_index_sha256": member_index, "zip_timestamp": "2000-01-01T00:00:00Z", "source_epochs": [1700000000, 1800000000], "builder_sha256": receipt["builder_sha256"], "build_contract": "two isolated offline source copies with distinct mtimes produced byte-identical wheels", "reproduced_exactly": True, "live_profile_accessed": False, } ): raise RuntimeError("deterministic wheel member binding failed") return receipt def _verify_freeze( *, candidate_digest: str, manifest_digest: str, predecessor: dict[str, object], wheel: dict[str, object], ) -> dict[str, object]: raw = FREEZE.read_bytes() decoded = cast(object, json.loads(raw)) if not isinstance(decoded, dict): raise RuntimeError("compatibility freeze receipt is invalid") receipt = cast(dict[str, object], decoded) unsigned = dict(receipt) digest = unsigned.pop("receipt_digest", None) expected = { "schema": "task22-supplemental-compatibility-freeze-v1", "candidate_digest": candidate_digest, "manifest_sha256": manifest_digest, "predecessor_candidate_digest": predecessor["candidate_digest"], "predecessor_manifest_sha256": hashlib.sha256(PREDECESSOR.read_bytes()).hexdigest(), "wheel_sha256": wheel["wheel_sha256"], "wheel_members": wheel["wheel_members"], "wheel_receipt_sha256": hashlib.sha256(WHEEL_RECEIPT.read_bytes()).hexdigest(), "wheel_receipt_digest": wheel["receipt_digest"], "source_epochs": [1700000000, 1800000000], "verification": { "mode": "offline-code-only-compatibility-freeze", "live_profile_accessed": False, }, } if ( stat.S_IMODE(FREEZE.stat().st_mode) != 0o600 or not isinstance(digest, str) or digest != hashlib.sha256(_canonical(unsigned)).hexdigest() or unsigned != expected ): raise RuntimeError("compatibility freeze receipt binding failed") return receipt def _category_summary( counts: Counter[str], byte_counts: Counter[str], return_codes: Counter[tuple[str, int]], digests: dict[str, HashDigest], ) -> dict[str, dict[str, object]]: return { category: { "patch_digest": digests[category].hexdigest(), "patch_bytes": byte_counts[category], "leaf_count": counts[category], "return_code_histogram": { str(code): count for (name, code), count in sorted(return_codes.items()) if name == category }, } for category in sorted(counts) } def main() -> int: if len(sys.argv) != 1: raise SystemExit("usage: verify-task22-supplemental-candidate.py") predecessor_bytes = PREDECESSOR.read_bytes() successor_bytes = SUCCESSOR.read_bytes() predecessor = cast(dict[str, object], json.loads(predecessor_bytes)) successor = cast(dict[str, object], json.loads(successor_bytes)) roots_raw = cast(dict[str, object], successor["roots"]) roots = {key: Path(cast(str, value)) for key, value in roots_raw.items()} old_leaves = cast(list[object], predecessor["candidate_files"]) leaves = cast(list[object], successor["candidate_files"]) if len(old_leaves) != len(leaves) or successor.get("candidate_path_count") != len(leaves): raise RuntimeError("candidate path scope changed") if successor.get("predecessor") != { "candidate_digest": predecessor.get("candidate_digest"), "manifest_digest": hashlib.sha256(predecessor_bytes).hexdigest(), "classification": _PREDECESSOR_CLASSIFICATION, }: raise RuntimeError("successor predecessor binding is invalid") if successor.get("verification") != { "mode": "offline-code-only-compatibility-reseal", "predecessor_candidate_retained": True, "live_profile_accessed": False, }: raise RuntimeError("successor verification provenance is invalid") candidate = hashlib.sha256() package = hashlib.sha256() diff = hashlib.sha256() gateway_candidate = hashlib.sha256() gateway_diff = hashlib.sha256() counts: Counter[str] = Counter() byte_counts: Counter[str] = Counter() return_codes: Counter[tuple[str, int]] = Counter() category_digests: dict[str, HashDigest] = { "gateway-tracked": hashlib.sha256(), "gateway-untracked": hashlib.sha256(), "profile-package-external": hashlib.sha256(), } changed: list[str] = [] previous: tuple[bytes, bytes] | None = None paths: list[str] = [] for old_raw, raw in zip(old_leaves, leaves, strict=True): old = cast(dict[str, object], old_raw) leaf = cast(dict[str, object], raw) root = cast(str, leaf["root"]) path = cast(str, leaf["path"]) category = cast(str, leaf["diff_category"]) if (root, path, category) != ( old.get("root"), old.get("path"), old.get("diff_category") ): raise RuntimeError("candidate leaf identity changed") key = (root.encode(), path.encode()) if previous is not None and key <= previous: raise RuntimeError("candidate leaves are not in canonical order") previous = key content = (roots[root] / path).read_bytes() current_category, patch, return_code = _category(root, path, roots) if current_category != category: raise RuntimeError(f"candidate category drifted: {root}:{path}") if ( len(content) != leaf.get("bytes") or hashlib.sha256(content).hexdigest() != leaf.get("sha256") or len(patch) != leaf.get("patch_bytes") or return_code != leaf.get("patch_return_code") or hashlib.sha256(patch).hexdigest() != leaf.get("patch_sha256") ): raise RuntimeError(f"candidate leaf mismatch: {root}:{path}") if old.get("sha256") != leaf.get("sha256"): changed.append(f"{root}:{path}") paths.append(f"{root}:{path}") candidate.update(root.encode() + b"\0" + path.encode() + b"\0" + content + b"\0") if root == "profile_package": package.update(path.encode() + b"\0" + content + b"\0") for frame in (root.encode(), path.encode(), category.encode(), patch): diff.update(frame + b"\0") category_digests[category].update(frame + b"\0") if root == "gateway": gateway_diff.update(frame + b"\0") if root == "gateway": gateway_candidate.update( root.encode() + b"\0" + path.encode() + b"\0" + content + b"\0" ) counts[category] += 1 byte_counts[category] += len(patch) return_codes[(category, return_code)] += 1 status = _run( ["git", "status", "--porcelain=v1", "-z", "--untracked-files=all"], cwd=roots["gateway"], allowed={0}, ).stdout records = [record for record in status.split(b"\0") if record] scope = successor.get("successor_scope") if scope != { "changed_candidate_paths": changed, "candidate_path_scope_preserved": True, "gateway_status_scope_preserved": True, "live_profile_accessed": False, }: raise RuntimeError("successor scope binding is invalid") if ( not changed or paths != successor.get("candidate_paths") or candidate.hexdigest() != successor.get("candidate_digest") or package.hexdigest() != successor.get("profile_package_digest") or diff.hexdigest() != successor.get("candidate_diff_sha256") or gateway_candidate.hexdigest() != successor.get("gateway_candidate_digest") or gateway_diff.hexdigest() != successor.get("gateway_diff_sha256") or dict(counts) != successor.get("candidate_diff_category_counts") or _category_summary( counts, byte_counts, return_codes, category_digests ) != successor.get("candidate_diff_categories") or hashlib.sha256(status).hexdigest() != successor.get("gateway_status_snapshot_sha256") or len(status) != successor.get("gateway_status_byte_length") or len(records) != successor.get("gateway_status_entry_count") ): raise RuntimeError("successor candidate verification failed") manifest_digest = hashlib.sha256(successor_bytes).hexdigest() wheel = _verify_wheel( candidate_digest=candidate.hexdigest(), manifest_digest=manifest_digest, predecessor_manifest_digest=hashlib.sha256(predecessor_bytes).hexdigest(), gateway_root=roots["gateway"], manifest_wheel_provenance=successor.get("wheel_provenance"), ) freeze = _verify_freeze( candidate_digest=candidate.hexdigest(), manifest_digest=manifest_digest, predecessor=predecessor, wheel=wheel, ) print(json.dumps({ "candidate": candidate.hexdigest(), "changed_paths": changed, "manifest": hashlib.sha256(successor_bytes).hexdigest(), "mode": "narrow-code-only-compatibility-verification", "paths": len(leaves), "wheel": wheel["wheel_sha256"], "wheel_receipt": wheel["receipt_digest"], "freeze": freeze["receipt_digest"], }, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())