#!/usr/bin/env python3 """Private, fail-closed, read-only verifier for retained Task21-26 archives.""" from __future__ import annotations import hashlib import json import os import re import stat import sys from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import Callable, Never, TypeGuard HEX64 = re.compile(r"[0-9a-f]{64}\Z") ARCHIVE_ID = re.compile(r"[0-9a-f]{32}\Z") FLAGS = os.O_RDONLY | os.O_CLOEXEC | getattr(os, "O_NOFOLLOW", 0) DIR_FLAGS = FLAGS | os.O_DIRECTORY UID = os.getuid() class VerificationError(RuntimeError): pass @dataclass(frozen=True) class Scope: name: str relative_path: str kind: str LEGACY = ( Scope("customer_registry", "customers/registry.json", "file"), Scope("customer_state", "data/customers", "directory"), Scope("room_bootstrap", "data/onboarding/telegram-room-bootstrap-v1", "directory"), Scope("owner_actions", "data/owner-actions", "directory"), Scope("onboarding_publication", "data/onboarding/telegram-publication-outbox-v1", "directory"), Scope("activation_notices", "data/activation-completion-notices.jsonl", "file"), Scope("activation_audit", "data/customer-activation-audit.jsonl", "file"), Scope("activation_receipt", "data/customer-activation-journal.json", "file"), Scope("onboarding_projection", "data/nutrition-onboarding-projection-journal.jsonl", "file"), Scope("scheduled_delivery_ledger", "data/scheduled-deliveries.jsonl", "file"), Scope("scheduled_delivery_fence", "data/scheduled-deliveries-fence.json", "file"), Scope("scheduled_delivery_claims", "data/customer-schedule-claims", "directory"), ) V2 = LEGACY + ( Scope("adaptive_authority_lock", "data/.adaptive-authority.lock", "file"), Scope("nutrition_onboarding_projection_lock", "data/.nutrition-onboarding-projection.lock", "file"), Scope("scheduled_deliveries_lock", "data/.scheduled-deliveries.lock", "file"), Scope("activation_notice_lock", "data/activation-completion-notices.lock", "file"), ) INGRESS = "d0aacf0f4bdbb7c0" V3 = V2 + ( Scope(f"telegram_ingress_receipt_{INGRESS}", f"data/telegram-ingress-receipts-v1-{INGRESS}.json", "file"), Scope(f"telegram_ingress_receipt_{INGRESS}_lock", f"data/telegram-ingress-receipts-v1-{INGRESS}.json.lock", "file"), ) V4 = V2 + ( Scope("recovery_audits", "data/recovery-audits", "directory"), Scope("activation_readiness", "data/onboarding/activation-readiness-v2", "directory"), Scope("task23_expired_bootstrap_supersession", "data/onboarding/task23-expired-bootstrap-supersession-v1", "directory"), Scope("task23_expired_bootstrap_supersession_lock", "data/onboarding/.task23-expired-bootstrap-supersession-v1.migration.lock", "file"), Scope("scheduled_delivery_attempt_lock_042165491060cc17a886131e425042075a5e639e0ce212ec8239f023bf759428", "data/.scheduled-delivery-attempt-042165491060cc17a886131e425042075a5e639e0ce212ec8239f023bf759428.lock", "file"), *V3[-2:], ) SCHEMAS: dict[str, tuple[Scope, ...]] = { "dualcoach-rehearsal-archive-v1": LEGACY, "dualcoach-rehearsal-archive-v3": V3, "dualcoach-rehearsal-archive-v4": V4, "dualcoach-rehearsal-supplemental-quarantine-v1": ( Scope("historical_duplicate_data_root", "data/data", "directory"), ), } def canonical(value: object) -> bytes: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8") def sha256(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest() def reject_constant(value: str) -> Never: raise VerificationError(f"invalid JSON constant: {value}") def reject_duplicates(pairs: list[tuple[str, object]]) -> dict[str, object]: result: dict[str, object] = {} for key, value in pairs: if key in result: raise VerificationError(f"duplicate JSON key: {key}") result[key] = value return result def is_object_dict(value: object) -> TypeGuard[dict[str, object]]: return isinstance(value, dict) def is_object_list(value: object) -> TypeGuard[list[object]]: return isinstance(value, list) def object_value(value: object, label: str) -> dict[str, object]: if not is_object_dict(value): raise VerificationError(f"{label} is not an object with string keys") return dict(value) def object_rows(value: object, label: str) -> list[dict[str, object]]: if not is_object_list(value): raise VerificationError(f"{label} is not a list") return [object_value(item, label) for item in value] def is_int_list(value: object) -> TypeGuard[list[int]]: return is_object_list(value) and all(type(item) is int for item in value) def require_string(value: object, label: str) -> str: if not isinstance(value, str): raise VerificationError(f"{label} is not a string") return value def parse_object(raw: bytes, label: str) -> dict[str, object]: try: decoder: Callable[..., object] = json.loads decoded = decoder( raw.decode("utf-8"), object_pairs_hook=reject_duplicates, parse_constant=reject_constant, ) except VerificationError: raise except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as exc: raise VerificationError(f"{label} is invalid JSON") from exc return object_value(decoded, label) def fingerprint(info: os.stat_result) -> tuple[int, ...]: return (info.st_dev, info.st_ino, info.st_mode, info.st_nlink, info.st_uid, info.st_size, info.st_mtime_ns, info.st_ctime_ns) def assert_entry(info: os.stat_result, kind: str, label: str, *, sealed: bool = False, evidence: bool = False) -> None: mode = stat.S_IMODE(info.st_mode) if info.st_uid != UID: raise VerificationError(f"{label} owner drift") if kind == "directory": allowed_directories = ({0o500, 0o700} if evidence else ({0o500} if sealed else {0o700})) if not stat.S_ISDIR(info.st_mode) or mode not in allowed_directories: raise VerificationError(f"{label} directory mode/type drift") else: allowed = ({0o400, 0o500, 0o600, 0o664, 0o700} if evidence else ({0o400} if sealed else {0o600})) if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or mode not in allowed: raise VerificationError(f"{label} file mode/type/nlink drift") def read_at(parent_fd: int, name: str, label: str, *, sealed: bool = False, evidence: bool = False) -> tuple[bytes, os.stat_result]: try: before = os.stat(name, dir_fd=parent_fd, follow_symlinks=False) assert_entry(before, "file", label, sealed=sealed, evidence=evidence) fd = os.open(name, FLAGS, dir_fd=parent_fd) except OSError as exc: raise VerificationError(f"{label} cannot be opened with O_NOFOLLOW") from exc try: opened = os.fstat(fd) if fingerprint(opened) != fingerprint(before): raise VerificationError(f"{label} changed while opening") chunks: list[bytes] = [] while block := os.read(fd, 1 << 20): chunks.append(block) after = os.fstat(fd) if fingerprint(after) != fingerprint(opened): raise VerificationError(f"{label} drifted during read") return b"".join(chunks), after finally: os.close(fd) def open_path(path: Path, *, sealed: bool = False, evidence: bool = False) -> tuple[int, os.stat_result]: absolute = Path(os.path.abspath(os.fspath(path))) fd = os.open("/", DIR_FLAGS) try: parts = absolute.parts[1:] for index, part in enumerate(parts): before = os.stat(part, dir_fd=fd, follow_symlinks=False) child = os.open(part, DIR_FLAGS, dir_fd=fd) opened = os.fstat(child) if (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino): os.close(child) raise VerificationError("path ancestry changed while opening") os.close(fd) fd = child if index == len(parts) - 1: assert_entry(opened, "directory", "root", sealed=sealed, evidence=evidence) return fd, os.fstat(fd) except (OSError, VerificationError): os.close(fd) raise def read_path(path: Path, label: str, *, sealed: bool = True) -> bytes: parent_fd, parent_before = open_path(path.parent, sealed=sealed, evidence=True) try: raw, _ = read_at(parent_fd, path.name, label, sealed=sealed, evidence=True) if fingerprint(os.fstat(parent_fd)) != fingerprint(parent_before): raise VerificationError(f"{label} parent drifted") return raw finally: os.close(parent_fd) @dataclass(frozen=True) class Entry: kind: str mode: str digest: str | None def scan_archive(root_fd: int) -> dict[str, Entry]: observed: dict[str, Entry] = {} def walk(fd: int, parent: str) -> None: directory_before = os.fstat(fd) for name in sorted(os.listdir(fd)): if not name or name in {".", ".."} or "/" in name: raise VerificationError("archive entry name escapes parent") relative = name if not parent else f"{parent}/{name}" before = os.stat(name, dir_fd=fd, follow_symlinks=False) if stat.S_ISDIR(before.st_mode): assert_entry(before, "directory", relative) child = os.open(name, DIR_FLAGS, dir_fd=fd) try: opened = os.fstat(child) if fingerprint(opened) != fingerprint(before): raise VerificationError(f"{relative} changed while opening") observed[relative] = Entry("directory", "0700", None) walk(child, relative) if fingerprint(os.fstat(child)) != fingerprint(opened): raise VerificationError(f"{relative} drifted during scan") finally: os.close(child) elif stat.S_ISREG(before.st_mode): raw, info = read_at(fd, name, relative) observed[relative] = Entry("file", f"{stat.S_IMODE(info.st_mode):04o}", sha256(raw)) else: raise VerificationError(f"{relative} is a symlink or special entry") if fingerprint(os.fstat(fd)) != fingerprint(directory_before): raise VerificationError(f"{parent or 'archive'} drifted during scan") walk(root_fd, "") return observed def scope_digest(observed: dict[str, Entry], root: str, kind: str) -> tuple[str, int, int]: selected = [(path, entry) for path, entry in observed.items() if path == root or path.startswith(root + "/")] if not selected: raise VerificationError(f"scope payload missing: {root}") root_entry = observed[root] if root_entry.kind != kind: raise VerificationError(f"scope kind drift: {root}") rows: list[bytes] = [] files = directories = 0 for path, entry in sorted(selected): relative = "." if path == root else path[len(root) + 1:] if entry.kind == "directory": directories += 1 rows.append(b"d\0" + relative.encode("utf-8")) else: files += 1 rows.append(b"f\0" + relative.encode("utf-8") + b"\0" + str(entry.digest).encode("ascii")) return sha256(b"\n".join(rows)), files, directories def require_exact_keys(value: dict[str, object], keys: set[str], label: str) -> None: if set(value) != keys: raise VerificationError(f"{label} fields drift") def verify_archive_fd(root_fd: int, archive_id: str) -> dict[str, object]: root_before = os.fstat(root_fd) assert_entry(root_before, "directory", "archive") observed = scan_archive(root_fd) if "manifest.json" not in observed or "receipt.json" not in observed or observed.get("payload", Entry("", "", None)).kind != "directory": raise VerificationError("archive control layout is incomplete") manifest_raw, _ = read_at(root_fd, "manifest.json", "manifest") receipt_raw, _ = read_at(root_fd, "receipt.json", "receipt") manifest = parse_object(manifest_raw, "manifest") receipt = parse_object(receipt_raw, "receipt") schema = manifest.get("schema") if not isinstance(schema, str) or schema not in SCHEMAS: raise VerificationError("unknown schema") contract = SCHEMAS[schema] require_exact_keys(manifest, {"schema", "archive_id", "scope_count", "scopes"}, "manifest") require_exact_keys(receipt, {"schema", "archive_id", "archive_digest", "archived_scope_count"}, "receipt") records = object_rows(manifest.get("scopes"), "manifest scopes") if len(records) != len(contract): raise VerificationError("scope member count drift") if manifest.get("archive_id") != archive_id or receipt.get("archive_id") != archive_id or not ARCHIVE_ID.fullmatch(archive_id): raise VerificationError("archive identity drift") if manifest.get("scope_count") != len(contract) or receipt.get("archived_scope_count") != len(contract) or receipt.get("schema") != schema: raise VerificationError("scope count/schema drift") archive_digest = sha256(canonical(manifest)) if receipt.get("archive_digest") != archive_digest: raise VerificationError("canonical manifest hash drift") expected_fixed = {"manifest.json", "receipt.json", "payload"} directory_roots: list[str] = [] scope_results: list[dict[str, object]] = [] for scope, record in zip(contract, records, strict=True): require_exact_keys(record, {"name", "present", "digest", "files", "directories"}, "scope") if record.get("name") != scope.name: raise VerificationError("schema scope dispatch/order drift") target = f"payload/{scope.relative_path}" present = record.get("present") files, directories = record.get("files"), record.get("directories") if type(files) is not int or type(directories) is not int or files < 0 or directories < 0: raise VerificationError("scope counts invalid") if present is False: if record.get("digest") is not None or files != 0 or directories != 0 or target in observed: raise VerificationError("absent scope drift") elif present is True: expected_fixed.add(target) for parent in PurePosixPath(target).parents: if str(parent) != ".": expected_fixed.add(str(parent)) actual_digest, actual_files, actual_dirs = scope_digest(observed, target, scope.kind) if record.get("digest") != actual_digest or files != actual_files or directories != actual_dirs: raise VerificationError("scope digest/member count drift") if scope.kind == "directory": directory_roots.append(target) scope_results.append({"name_sha256": sha256(scope.name.encode()), "files": files, "directories": directories, "digest": actual_digest}) else: raise VerificationError("scope presence invalid") for path in observed: if path not in expected_fixed and not any(path.startswith(root + "/") for root in directory_roots): raise VerificationError("archive contains unexpected member") if not expected_fixed <= set(observed): raise VerificationError("archive is missing expected member") if fingerprint(os.fstat(root_fd)) != fingerprint(root_before): raise VerificationError("archive root drifted during verification") tree_rows = [{"kind": entry.kind, "mode": entry.mode, "path_sha256": sha256(path.encode()), "sha256": entry.digest} for path, entry in sorted(observed.items())] return { "status": "PASS", "archive_id_sha256": sha256(archive_id.encode()), "schema": schema, "scope_count": len(contract), "file_count": sum(row.kind == "file" for row in observed.values()), "directory_count": sum(row.kind == "directory" for row in observed.values()), "archive_digest": archive_digest, "manifest_sha256": sha256(manifest_raw), "receipt_sha256": sha256(receipt_raw), "tree_aggregate_sha256": sha256(canonical(tree_rows)), "scopes": scope_results, } def verify_archive_path(path: Path) -> dict[str, object]: fd, _ = open_path(path) try: return verify_archive_fd(fd, path.name) finally: os.close(fd) def verify_candidate_chain(candidate: dict[str, object]) -> dict[str, str]: require_exact_keys(candidate, {"root", "full_candidate_digest", "manifest_sha256", "checkpoint_sha256", "binding_sha256", "binding_artifact_sha256", "freeze_path", "freeze_sha256"}, "candidate inventory") root = Path(require_string(candidate["root"], "candidate root")) manifest_raw = read_path(root / "candidate-manifest.json", "candidate manifest") checkpoint_raw = read_path(root / "candidate-checkpoint.json", "candidate checkpoint") binding_raw = read_path(root / "bindings/historical-archive-binding.json", "archive binding") freeze_path = Path(require_string(candidate["freeze_path"], "candidate freeze path")) freeze_raw = read_path(freeze_path, "candidate freeze") actual = [sha256(manifest_raw), sha256(checkpoint_raw), sha256(binding_raw), sha256(freeze_raw)] expected = [ require_string(candidate["manifest_sha256"], "candidate manifest hash"), require_string(candidate["checkpoint_sha256"], "candidate checkpoint hash"), require_string(candidate["binding_artifact_sha256"], "candidate binding artifact hash"), require_string(candidate["freeze_sha256"], "candidate freeze hash"), ] if actual != expected: raise VerificationError("candidate chain raw hash drift") manifest = parse_object(manifest_raw, "candidate manifest") checkpoint = parse_object(checkpoint_raw, "candidate checkpoint") binding = parse_object(binding_raw, "archive binding") freeze = parse_object(freeze_raw, "candidate freeze") core = dict(manifest) full = require_string(core.pop("full_candidate_digest", None), "full candidate digest") claimed_core = require_string(core.pop("core_candidate_digest", None), "core candidate digest") calculated_core = sha256(canonical(core)) calculated_full = sha256(canonical({"core_candidate_digest": calculated_core, "manifest_core": core})) binding_core = dict(binding) binding_digest = require_string(binding_core.pop("binding_sha256", None), "archive binding digest") candidate_digest = require_string(candidate["full_candidate_digest"], "inventory candidate digest") inventory_binding_digest = require_string(candidate["binding_sha256"], "inventory binding digest") historical_binding = object_value(manifest.get("historical_archive_binding"), "historical archive binding") if ( full != candidate_digest or claimed_core != calculated_core or full != calculated_full or checkpoint.get("full_candidate_digest") != full or checkpoint.get("manifest_sha256") != actual[0] or binding_digest != inventory_binding_digest or binding_digest != sha256(canonical(binding_core)) or historical_binding.get("artifact_sha256") != actual[2] or historical_binding.get("binding_sha256") != binding_digest or freeze.get("full_candidate_digest") != full or freeze.get("manifest_sha256") != actual[0] or Path(str(freeze.get("candidate_root"))) != root ): raise VerificationError("candidate parent/binding chain drift") return {"full_candidate_digest": full, "manifest_sha256": actual[0], "binding_sha256": binding_digest, "freeze_sha256": actual[3]} def verify_inventory(path: Path, *, verify_candidate: bool = True) -> dict[str, object]: raw = read_path(path, "archive inventory", sealed=False) inventory = parse_object(raw, "archive inventory") require_exact_keys(inventory, {"schema", "archive_parent", "candidate", "archives"}, "inventory") if inventory.get("schema") != "task26-retained-archive-inventory-v1": raise VerificationError("inventory schema drift") rows = object_rows(inventory.get("archives"), "inventory archives") if not rows: raise VerificationError("inventory archive set is empty") expected_ids: list[str] = [] by_id: dict[str, dict[str, object]] = {} for row in rows: require_exact_keys(row, {"archive_id", "schema", "manifest_sha256", "receipt_sha256", "tasks", "binding_files"}, "archive inventory row") archive_id = row.get("archive_id") if not isinstance(archive_id, str) or not ARCHIVE_ID.fullmatch(archive_id) or archive_id in by_id: raise VerificationError("inventory archive identity invalid") expected_ids.append(archive_id) by_id[archive_id] = row parent = Path(str(inventory["archive_parent"])) parent_fd, parent_before = open_path(parent) results: list[dict[str, object]] = [] try: observed_ids = sorted(os.listdir(parent_fd)) if observed_ids != sorted(expected_ids): raise VerificationError("retained archive set drift") for archive_id in sorted(expected_ids): before = os.stat(archive_id, dir_fd=parent_fd, follow_symlinks=False) assert_entry(before, "directory", "archive") archive_fd = os.open(archive_id, DIR_FLAGS, dir_fd=parent_fd) try: if fingerprint(os.fstat(archive_fd)) != fingerprint(before): raise VerificationError("archive changed while opening") result = verify_archive_fd(archive_fd, archive_id) finally: os.close(archive_fd) row = by_id[archive_id] if result["schema"] != row["schema"] or result["manifest_sha256"] != row["manifest_sha256"]: raise VerificationError("raw manifest hash/schema drift") if result["receipt_sha256"] != row["receipt_sha256"]: raise VerificationError("raw receipt hash drift") tasks = row["tasks"] if not is_int_list(tasks) or not tasks or any(task < 21 or task > 26 for task in tasks): raise VerificationError("archive task binding invalid") bindings = object_rows(row["binding_files"], "archive provenance bindings") if not bindings: raise VerificationError("archive has no parent binding") for binding in bindings: require_exact_keys(binding, {"path", "sha256"}, "archive binding row") binding_path = require_string(binding["path"], "archive binding path") binding_hash = require_string(binding["sha256"], "archive binding hash") binding_raw = read_path(Path(binding_path), "archive provenance binding", sealed=False) if sha256(binding_raw) != binding_hash or archive_id.encode() not in binding_raw: raise VerificationError("archive provenance binding drift") if Path(binding_path).suffix == ".json": _ = parse_object(binding_raw, "archive provenance binding") result["tasks"] = tasks results.append(result) if fingerprint(os.fstat(parent_fd)) != fingerprint(parent_before): raise VerificationError("archive parent drifted during verification") finally: os.close(parent_fd) candidate_result = verify_candidate_chain(object_value(inventory["candidate"], "candidate inventory")) if verify_candidate else None return { "status": "PASS", "archive_count": len(results), "schema_count": len({row["schema"] for row in results}), "schemas": sorted({str(row["schema"]) for row in results}), "archives": results, "candidate": candidate_result, "inventory_sha256": sha256(raw), } def redacted_receipt( result: dict[str, object], permission_seal_sha256: str, supersedes_receipt_sha256: str ) -> dict[str, object]: archives = object_rows(result["archives"], "verified archives") rows = [{key: value for key, value in row.items() if key not in {"scopes"}} for row in archives] return { "schema": "task26-retained-archive-verification-receipt-v2", "status": "PASS", "supersedes_receipt_sha256": supersedes_receipt_sha256, "supersession_reason": "quality findings repaired and artifact set resealed", "ready_for_preflight": True, "archive_count": result["archive_count"], "schema_count": result["schema_count"], "schemas": result["schemas"], "archives": rows, "candidate": result["candidate"], "inventory_sha256": result["inventory_sha256"], "permission_seal_sha256": permission_seal_sha256, "privacy": "redacted deterministic; archive IDs and filesystem paths omitted", "non_actions": ["no restoration", "no extraction", "no runtime write", "no archive write", "no service action", "no network", "no git"], } def parse_cli(argv: list[str]) -> tuple[Path, Path, Path | None]: if len(argv) not in {3, 5} or argv[1] != "--permission-seal": raise VerificationError("usage: verifier INVENTORY --permission-seal SEAL [--receipt RECEIPT]") if len(argv) == 5 and argv[3] != "--receipt": raise VerificationError("usage: verifier INVENTORY --permission-seal SEAL [--receipt RECEIPT]") return Path(argv[0]), Path(argv[2]), Path(argv[4]) if len(argv) == 5 else None def main() -> int: try: inventory_path, permission_seal_path, receipt_path = parse_cli(sys.argv[1:]) seal_raw = read_path(permission_seal_path, "permission seal", sealed=False) seal = parse_object(seal_raw, "permission seal") require_exact_keys( seal, {"schema", "authority", "allowed_operation", "forbidden_operations", "verifier_sha256", "tests_sha256", "inventory_sha256", "candidate_digest", "supersedes_receipt_sha256"}, "permission seal", ) if seal.get("schema") != "task26-retained-archive-verifier-permission-seal-v1" or seal.get("allowed_operation") != "read_only_verification_and_redacted_receipt_write": raise VerificationError("permission seal schema/operation drift") result = verify_inventory(inventory_path) verifier_raw = read_path(Path(__file__), "sealed verifier", sealed=False) tests_raw = read_path(Path(__file__).with_name("test_archive_verifier.py"), "sealed verifier tests", sealed=False) candidate = object_value(result.get("candidate"), "verified candidate") if ( seal.get("verifier_sha256") != sha256(verifier_raw) or seal.get("tests_sha256") != sha256(tests_raw) or seal.get("inventory_sha256") != result.get("inventory_sha256") or seal.get("candidate_digest") != candidate.get("full_candidate_digest") ): raise VerificationError("permission seal binding drift") supersedes = require_string(seal.get("supersedes_receipt_sha256"), "superseded receipt hash") receipt = redacted_receipt(result, sha256(seal_raw), supersedes) encoded = canonical(receipt) + b"\n" if receipt_path is not None: flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC | getattr(os, "O_NOFOLLOW", 0) fd = os.open(receipt_path, flags, 0o600) try: os.fchmod(fd, 0o600) if os.write(fd, encoded) != len(encoded): raise VerificationError("short receipt write") os.fsync(fd) finally: os.close(fd) _ = sys.stdout.buffer.write(encoded) return 0 except (OSError, VerificationError, KeyError, TypeError) as exc: _ = sys.stderr.write(f"FAIL: {exc}\n") return 1 if __name__ == "__main__": raise SystemExit(main())