#!/usr/bin/env python3 """Build or verify a private, hash-pinned V3 clean-genesis ledger bundle. This command is local and read-only with respect to PostgreSQL. It accepts only exact exporter-shaped, source-normalized V3 ``approve_claim`` materials. An admission manifest is never inferred from proposal approval: it is emitted only when the operator supplies the explicit review flag and every review field. The result is hash-pinned review metadata, not authenticated review authority, and does not make a reconstruction canonically eligible by itself. """ from __future__ import annotations import argparse import hashlib import json import os import stat import sys import tempfile import uuid from dataclasses import dataclass from pathlib import Path from typing import Any try: from . import run_local_genesis_ledger_rebuild as genesis_rebuild except ImportError: # pragma: no cover - direct script execution import run_local_genesis_ledger_rebuild as genesis_rebuild REPO_ROOT = Path(__file__).resolve().parents[1] SCRIPTS_DIR = REPO_ROOT / "scripts" if str(SCRIPTS_DIR) not in sys.path: sys.path.insert(0, str(SCRIPTS_DIR)) import export_kb_transition_replay_material as replay_exporter # noqa: E402 RECEIPT_ARTIFACT = "teleo_v3_genesis_ledger_bundle_builder_receipt" CANONICAL_REVIEWER_HANDLE = "m3taversal" REVIEW_METADATA_PROVENANCE = "operator_supplied_explicit_review_metadata" NORMALIZATION_SCHEMA = "livingip.kbSourceProposalNormalization.v3" EXTRACTION_MANIFEST_SCHEMA = "livingip.kbSourceExtractionManifest.v3" MANIFEST_IDENTITY_BINDING = "sha256(exact duplicate-free UTF-8 JSON file bytes)" NORMALIZATION_FIELDS = frozenset( { "schema", "semantic_extraction_statement", "semantic_extraction_is_retained_input", "model_training_claim", "manifest_schema", "manifest_file_sha256", "manifest_identity_binding", "manifest_canonical_sha256", "artifact_sha256", "extracted_text_sha256", "extractor", "source_identity", "source_locator", "normalized_candidate_content_sha256", "identity_material", "packet_identity_sha256", "parent_packet_sha256", "quote_bindings", } ) IDENTITY_FIELDS = frozenset( { "artifact_sha256", "extracted_text_sha256", "extractor", "manifest_file_sha256", "source_identity", "source_locator", "normalized_candidate_content_sha256", } ) EXTRACTOR_FIELDS = frozenset({"name", "version", "spec_sha256"}) QUOTE_BINDING_FIELDS = frozenset({"claim_key", "evidence_key", "kind", "quote", "quote_sha256", "unit", "start", "end"}) LOCATOR_FIELDS = frozenset( { "schema", "source_identity", "source_locator", "artifact_sha256", "extracted_text_sha256", "extractor", "unit", "start", "end", "quote_sha256", } ) class BundleError(RuntimeError): """A fail-closed error whose text is safe for command output.""" def __init__(self, code: str, safe_message: str) -> None: super().__init__(safe_message) self.code = code self.safe_message = safe_message @dataclass(frozen=True) class MaterialPin: path: Path sha256: str replay_material_sha256: str entry: genesis_rebuild.LedgerEntry @dataclass(frozen=True) class StagedFile: destination: Path temporary: Path payload: bytes @dataclass(frozen=True) class PublishedFile: destination: Path payload: bytes device: int inode: int def _canonical_json_bytes(value: Any) -> bytes: try: rendered = json.dumps( value, ensure_ascii=False, indent=2, sort_keys=True, allow_nan=False, ) except (TypeError, ValueError) as exc: raise BundleError("invalid_json_value", "bundle data is not deterministic JSON") from exc return (rendered + "\n").encode("utf-8") def _compiler_canonical_bytes(value: Any) -> bytes: return json.dumps(value, ensure_ascii=True, separators=(",", ":"), sort_keys=True).encode("utf-8") def _compiler_canonical_sha256(value: Any) -> str: return hashlib.sha256(_compiler_canonical_bytes(value)).hexdigest() def _sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def _stable_v3_uuid(row_kind: str, packet_identity_sha256: str, semantic_key: Any) -> str: material = _compiler_canonical_bytes( ["livingip", "kb-source-compiler", 3, packet_identity_sha256, row_kind, semantic_key] ).decode("ascii") return str(uuid.uuid5(uuid.NAMESPACE_URL, material)) def _absolute_path(path: Path) -> Path: expanded = path.expanduser() return expanded if expanded.is_absolute() else Path.cwd() / expanded def _regular_input(path: Path, *, label: str, private_json: bool = False) -> Path: absolute = _absolute_path(path) try: metadata = absolute.lstat() except OSError as exc: raise BundleError("missing_input", f"{label} must be an existing regular non-symlink file") from exc if absolute.is_symlink() or not stat.S_ISREG(metadata.st_mode): raise BundleError("unsafe_input", f"{label} must be an existing regular non-symlink file") if private_json and stat.S_IMODE(metadata.st_mode) != 0o600: raise BundleError("unsafe_private_permissions", f"{label} permissions must be exactly 0600") return absolute.resolve() def _prepare_new_output(path: Path, *, label: str) -> Path: absolute = _absolute_path(path) if absolute.name in {"", ".", ".."}: raise BundleError("unsafe_output", f"{label} must name a new file") try: absolute.lstat() except FileNotFoundError: pass except OSError as exc: raise BundleError("unsafe_output", f"{label} cannot be inspected safely") from exc else: raise BundleError("existing_output", f"{label} must not already exist") parent = absolute.parent if parent.exists() and (parent.is_symlink() or not parent.is_dir()): raise BundleError("unsafe_output_parent", f"{label} parent must be a real directory") return parent.resolve(strict=False) / absolute.name def _assert_distinct_paths(paths: list[tuple[Path, str]]) -> None: seen: dict[Path, str] = {} for path, label in paths: resolved = path.resolve(strict=False) if resolved in seen: raise BundleError("path_alias", f"{label} must not alias another bundle path") seen[resolved] = label def _read_json_object(path: Path, *, label: str) -> dict[str, Any]: try: raw = path.read_text(encoding="utf-8") return replay_exporter._parse_json_object(raw, label=label) except replay_exporter.ExportError as exc: raise BundleError("invalid_json_input", f"{label} is not exact duplicate-free UTF-8 JSON") from exc except (OSError, UnicodeError) as exc: raise BundleError("invalid_json_input", f"{label} is not exact duplicate-free UTF-8 JSON") from exc def _require_exact_fields(value: Any, expected: frozenset[str], *, label: str) -> dict[str, Any]: if not isinstance(value, dict) or frozenset(value) != expected: raise BundleError("invalid_v3_provenance", f"{label} does not match the current V3 provenance contract") return value def _require_sha256(value: Any, *, label: str) -> str: if not isinstance(value, str) or not genesis_rebuild.SHA256_RE.fullmatch(value): raise BundleError("invalid_v3_provenance", f"{label} must be a lowercase SHA-256 digest") return value def _canonical_uuid(value: Any, *, label: str) -> str: if not isinstance(value, str): raise BundleError("invalid_reviewer_identity", f"{label} must be a canonical UUID") try: normalized = str(uuid.UUID(value)) except (AttributeError, TypeError, ValueError) as exc: raise BundleError("invalid_reviewer_identity", f"{label} must be a canonical UUID") from exc if value != normalized: raise BundleError("invalid_reviewer_identity", f"{label} must be a canonical UUID") return normalized def _validate_source_normalization(entry: genesis_rebuild.LedgerEntry) -> None: proposal = entry.applied_proposal if entry.contract_version != 3 or proposal.get("proposal_type") != "approve_claim": raise BundleError("legacy_material_forbidden", "every material must be a V3 approve_claim transition") payload = proposal.get("payload") apply_payload = payload.get("apply_payload") if isinstance(payload, dict) else None if not isinstance(apply_payload, dict) or apply_payload.get("contract_version") != 3: raise BundleError("legacy_material_forbidden", "every material must be a V3 approve_claim transition") manifest = _require_exact_fields( apply_payload.get("normalization_manifest"), NORMALIZATION_FIELDS, label="normalization manifest", ) if ( manifest["schema"] != NORMALIZATION_SCHEMA or manifest["manifest_schema"] != EXTRACTION_MANIFEST_SCHEMA or manifest["manifest_identity_binding"] != MANIFEST_IDENTITY_BINDING or manifest["semantic_extraction_is_retained_input"] is not True or manifest["model_training_claim"] is not False or not isinstance(manifest["semantic_extraction_statement"], str) or "retained input" not in manifest["semantic_extraction_statement"].lower() ): raise BundleError( "invalid_v3_provenance", "normalization manifest does not retain the current source-backed V3 provenance statement", ) digest_fields = ( "manifest_file_sha256", "manifest_canonical_sha256", "artifact_sha256", "extracted_text_sha256", "normalized_candidate_content_sha256", "packet_identity_sha256", "parent_packet_sha256", ) digests = { field: _require_sha256(manifest[field], label=f"normalization manifest {field}") for field in digest_fields } extractor = _require_exact_fields(manifest["extractor"], EXTRACTOR_FIELDS, label="normalization extractor") if any(not isinstance(extractor[field], str) or not extractor[field].strip() for field in ("name", "version")): raise BundleError("invalid_v3_provenance", "normalization extractor identity must be non-empty") _require_sha256(extractor["spec_sha256"], label="normalization extractor spec_sha256") identity = _require_exact_fields( manifest["identity_material"], IDENTITY_FIELDS, label="normalization identity material", ) for field in IDENTITY_FIELDS: if identity[field] != manifest[field]: raise BundleError("invalid_v3_provenance", "normalization identity material does not match its pins") if _compiler_canonical_sha256(identity) != digests["packet_identity_sha256"]: raise BundleError("invalid_v3_provenance", "normalization packet identity is not recoverable") artifact_sha256 = digests["artifact_sha256"] source_identity = f"document:sha256:{artifact_sha256}" source_locator = f"artifact://sha256/{artifact_sha256}" if manifest["source_identity"] != source_identity or manifest["source_locator"] != source_locator: raise BundleError("invalid_v3_provenance", "normalization source identity is not content-addressed") source_proposal_id = apply_payload.get("source_proposal_id") try: canonical_source_proposal_id = str(uuid.UUID(str(source_proposal_id))) except (AttributeError, TypeError, ValueError) as exc: raise BundleError("invalid_v3_provenance", "V3 source proposal identity is invalid") from exc expected_source_proposal_id = _stable_v3_uuid( "source_proposal", digests["packet_identity_sha256"], digests["normalized_candidate_content_sha256"], ) if source_proposal_id != canonical_source_proposal_id or source_proposal_id != expected_source_proposal_id: raise BundleError("invalid_v3_provenance", "V3 source proposal identity is not recoverable") payload_sha256 = _compiler_canonical_sha256(proposal["payload"]) expected_proposal_id = _stable_v3_uuid( "pending_review_proposal", digests["packet_identity_sha256"], payload_sha256, ) expected_source_ref = f"source-compiler-v3:{digests['packet_identity_sha256'][:32]}:{payload_sha256[:16]}" if proposal.get("id") != expected_proposal_id or proposal.get("source_ref") != expected_source_ref: raise BundleError("invalid_v3_provenance", "V3 normalized proposal identity is not recoverable") sources = apply_payload.get("sources") evidence_rows = apply_payload.get("evidence") claims = apply_payload.get("claims") assessments = apply_payload.get("assessments") if ( not isinstance(sources, list) or len(sources) != 1 or not isinstance(sources[0], dict) or not isinstance(evidence_rows, list) or not evidence_rows or not isinstance(claims, list) or not claims or not isinstance(assessments, list) or len(assessments) != len(claims) ): raise BundleError("invalid_v3_provenance", "V3 material is not an exact source-backed claim bundle") source = sources[0] if ( source.get("content_hash") != artifact_sha256 or source.get("canonical_url") is not None or source.get("storage_uri") != source_locator ): raise BundleError("invalid_v3_provenance", "V3 source row does not retain the normalized artifact pin") raw_bindings = manifest["quote_bindings"] if not isinstance(raw_bindings, list) or not raw_bindings or len(raw_bindings) != len(evidence_rows): raise BundleError("invalid_v3_provenance", "normalization quote bindings must select every evidence row") bindings: list[dict[str, Any]] = [] binding_keys: list[tuple[str, str]] = [] for raw_binding in raw_bindings: binding = _require_exact_fields(raw_binding, QUOTE_BINDING_FIELDS, label="normalization quote binding") claim_key = binding["claim_key"] evidence_key = binding["evidence_key"] quote = binding["quote"] start = binding["start"] end = binding["end"] if ( not isinstance(claim_key, str) or not claim_key.strip() or not isinstance(evidence_key, str) or not evidence_key.strip() or binding["kind"] != "evidence" or not isinstance(quote, str) or not quote or binding["unit"] != "utf8_bytes" or type(start) is not int or type(end) is not int or start < 0 or end <= start or end - start != len(quote.encode("utf-8")) or _sha256_bytes(quote.encode("utf-8")) != _require_sha256(binding["quote_sha256"], label="normalization quote SHA-256") ): raise BundleError("invalid_v3_provenance", "normalization quote binding is invalid") binding_keys.append((claim_key, evidence_key)) bindings.append(binding) if len(binding_keys) != len(set(binding_keys)) or binding_keys != sorted(binding_keys): raise BundleError("invalid_v3_provenance", "normalization quote bindings are duplicate or ambiguously ordered") packet_identity = digests["packet_identity_sha256"] claim_keys = sorted({claim_key for claim_key, _ in binding_keys}) expected_claim_ids = {_stable_v3_uuid("claim", packet_identity, claim_key) for claim_key in claim_keys} observed_claim_ids = {row.get("id") for row in claims if isinstance(row, dict)} if len(observed_claim_ids) != len(claims) or observed_claim_ids != expected_claim_ids: raise BundleError("invalid_v3_provenance", "normalized V3 claim identities are not recoverable") evidence_by_id = { row.get("id"): row for row in evidence_rows if isinstance(row, dict) and isinstance(row.get("id"), str) } if len(evidence_by_id) != len(evidence_rows): raise BundleError("invalid_v3_provenance", "normalized V3 evidence identities are not unique") source_id = source.get("id") for binding in bindings: claim_key = binding["claim_key"] evidence_key = binding["evidence_key"] expected_evidence_id = _stable_v3_uuid("evidence", packet_identity, [claim_key, evidence_key]) evidence = evidence_by_id.get(expected_evidence_id) expected_locator = { "schema": "livingip.extractedTextQuoteLocator.v1", "source_identity": source_identity, "source_locator": source_locator, "artifact_sha256": artifact_sha256, "extracted_text_sha256": digests["extracted_text_sha256"], "extractor": extractor, "unit": "utf8_bytes", "start": binding["start"], "end": binding["end"], "quote_sha256": binding["quote_sha256"], } if ( evidence is None or evidence.get("claim_id") != _stable_v3_uuid("claim", packet_identity, claim_key) or evidence.get("source_id") != source_id or evidence.get("excerpt") != binding["quote"] or evidence.get("source_content_hash") != artifact_sha256 or not isinstance(evidence.get("locator_json"), dict) or frozenset(evidence["locator_json"]) != LOCATOR_FIELDS or evidence["locator_json"] != expected_locator ): raise BundleError("invalid_v3_provenance", "normalized V3 evidence provenance is not recoverable") expected_assessments = { _stable_v3_uuid("assessment", packet_identity, claim_key): _stable_v3_uuid("claim", packet_identity, claim_key) for claim_key in claim_keys } observed_assessments = { row.get("id"): row.get("claim_id") for row in assessments if isinstance(row, dict) and isinstance(row.get("id"), str) } if observed_assessments != expected_assessments: raise BundleError("invalid_v3_provenance", "normalized V3 assessment identities are not recoverable") def _validate_materials(paths: list[Path]) -> list[MaterialPin]: if not paths: raise BundleError("missing_material", "at least one ordered --material is required") pins: list[MaterialPin] = [] proposal_ids: set[str] = set() material_paths: set[Path] = set() for expected_sequence, raw_path in enumerate(paths, 1): path = _regular_input(raw_path, label=f"material {expected_sequence}") if path in material_paths: raise BundleError("duplicate_material", "ordered material paths must be unique") material_paths.add(path) raw_material = _read_json_object(path, label=f"material {expected_sequence}") try: material = replay_exporter.validate_material(raw_material, expected_sequence=expected_sequence) except replay_exporter.ExportError as exc: raise BundleError( "invalid_exported_material", f"material {expected_sequence} failed the exact exporter contract", ) from exc if _compiler_canonical_bytes(material) != _compiler_canonical_bytes(raw_material): raise BundleError( "noncanonical_exported_material", f"material {expected_sequence} is not in exact exporter-canonical form", ) material_sha256 = genesis_rebuild.canonical_rebuild.sha256_file(path) hashes = material.get("replay_receipt", {}).get("hashes") replay_material_sha256 = hashes.get("replay_material_sha256") if isinstance(hashes, dict) else None _require_sha256(replay_material_sha256, label=f"material {expected_sequence} replay hash") try: entry = genesis_rebuild._validate_entry_material( material, expected_sequence=expected_sequence, expected_replay_hash=replay_material_sha256, material_path=path, material_sha256=material_sha256, ) except genesis_rebuild.ReconstructionError as exc: raise BundleError( "invalid_genesis_material", f"material {expected_sequence} failed the genesis replay contract", ) from exc _validate_source_normalization(entry) proposal_id = entry.applied_proposal["id"] if proposal_id in proposal_ids: raise BundleError("duplicate_proposal", "ordered materials contain a duplicate proposal identity") proposal_ids.add(proposal_id) pins.append( MaterialPin( path=path, sha256=material_sha256, replay_material_sha256=replay_material_sha256, entry=entry, ) ) return pins def _relative_artifact(path: Path, *, ledger_parent: Path) -> str: rendered = Path(os.path.relpath(path, start=ledger_parent)).as_posix() if Path(rendered).is_absolute(): # pragma: no cover - os.path.relpath is relative on one filesystem raise BundleError("invalid_relative_path", "ledger artifact paths must be relative") return rendered def _validate_dump_and_manifests( genesis_dump: Path, genesis_manifest: Path, final_manifest: Path, ) -> None: try: genesis_rebuild.canonical_rebuild.validate_custom_dump(genesis_dump) except (OSError, ValueError) as exc: raise BundleError("invalid_genesis_dump", "genesis dump failed the custom-format preflight") from exc for path, label in ((genesis_manifest, "genesis parity manifest"), (final_manifest, "final parity manifest")): try: genesis_rebuild.load_manifest(path) except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError) as exc: raise BundleError("invalid_parity_manifest", f"{label} failed the existing manifest validator") from exc def _ledger_payload( *, ledger_path: Path, genesis_dump: Path, genesis_manifest: Path, final_manifest: Path, materials: list[MaterialPin], ) -> dict[str, Any]: return { "artifact": genesis_rebuild.LEDGER_ARTIFACT, "contract_version": genesis_rebuild.LEDGER_CONTRACT_VERSION, "engine": genesis_rebuild._engine_hashes(), "genesis": { "dump_sha256": genesis_rebuild.canonical_rebuild.sha256_file(genesis_dump), "parity_manifest_sha256": genesis_rebuild.canonical_rebuild.sha256_file(genesis_manifest), }, "entries": [ { "sequence": pin.entry.sequence, "material": _relative_artifact(pin.path, ledger_parent=ledger_path.parent), "sha256": pin.sha256, "replay_material_sha256": pin.replay_material_sha256, } for pin in materials ], "final_parity": { "manifest": _relative_artifact(final_manifest, ledger_parent=ledger_path.parent), "sha256": genesis_rebuild.canonical_rebuild.sha256_file(final_manifest), }, } def _admission_payload( args: argparse.Namespace, *, ledger_sha256: str, materials: list[MaterialPin], review_note_file: Path, ) -> dict[str, Any]: if args.reviewed_by != CANONICAL_REVIEWER_HANDLE: raise BundleError("invalid_reviewer_handle", "--reviewed-by must equal m3taversal exactly") reviewer_id = _canonical_uuid(args.reviewed_by_agent_id, label="--reviewed-by-agent-id") try: reviewed_at = genesis_rebuild.replay_receipt.canonical_utc_timestamp( args.reviewed_at, path="--reviewed-at", ) except (TypeError, ValueError) as exc: raise BundleError("invalid_reviewed_at", "--reviewed-at must be timezone-aware") from exc try: review_note = review_note_file.read_text(encoding="utf-8") except (OSError, UnicodeError) as exc: raise BundleError("invalid_review_note", "--review-note-file must contain readable UTF-8 text") from exc if not review_note.strip(): raise BundleError("invalid_review_note", "--review-note-file must be non-empty") return { "artifact": genesis_rebuild.ADMISSION_ARTIFACT, "contract_version": genesis_rebuild.ADMISSION_CONTRACT_VERSION, "ledger_sha256": ledger_sha256, "reviewed_by_handle": CANONICAL_REVIEWER_HANDLE, "reviewed_by_agent_id": reviewer_id, "reviewed_at": reviewed_at, "review_note": review_note, "entries": [ { "sequence": pin.entry.sequence, "proposal_id": pin.entry.applied_proposal["id"], "material_sha256": pin.sha256, "replay_material_sha256": pin.replay_material_sha256, "admission_scope": args.admission_scope, "basis": args.admission_basis, } for pin in materials ], } def _sync_directory(path: Path) -> None: descriptor = os.open(path, os.O_RDONLY) try: os.fsync(descriptor) finally: os.close(descriptor) def _stage_private_json(destination: Path, payload: dict[str, Any]) -> StagedFile: parent_preexisting = destination.parent.exists() destination.parent.mkdir(parents=True, exist_ok=True, mode=0o700) if destination.parent.is_symlink() or not destination.parent.is_dir(): raise BundleError("unsafe_output_parent", "output parent must be a real directory") if not parent_preexisting: os.chmod(destination.parent, 0o700) try: destination.lstat() except FileNotFoundError: pass else: raise BundleError("existing_output", "bundle output appeared before publication") payload_bytes = _canonical_json_bytes(payload) descriptor, temporary_name = tempfile.mkstemp(prefix=f".{destination.name}.", dir=str(destination.parent)) temporary = Path(temporary_name) try: os.fchmod(descriptor, 0o600) with os.fdopen(descriptor, "wb") as handle: handle.write(payload_bytes) handle.flush() os.fsync(handle.fileno()) except BaseException: try: os.close(descriptor) except OSError: pass try: temporary.unlink() except FileNotFoundError: pass raise return StagedFile(destination=destination, temporary=temporary, payload=payload_bytes) def _unlink_if_owned(published: PublishedFile) -> bool: try: metadata = published.destination.lstat() if ( stat.S_ISREG(metadata.st_mode) and metadata.st_dev == published.device and metadata.st_ino == published.inode and published.destination.read_bytes() == published.payload ): published.destination.unlink() return True except FileNotFoundError: return True except OSError: return False return False def _rollback_published(files: list[PublishedFile]) -> None: changed_directories: set[Path] = set() for published in reversed(files): if _unlink_if_owned(published): changed_directories.add(published.destination.parent) for directory in changed_directories: try: _sync_directory(directory) except OSError: pass def _publish_no_replace(staged_files: list[StagedFile]) -> list[PublishedFile]: published: list[PublishedFile] = [] directories = {staged.destination.parent for staged in staged_files} try: for staged in staged_files: temporary_metadata = staged.temporary.lstat() try: os.link(staged.temporary, staged.destination) except FileExistsError as exc: raise BundleError("existing_output", "bundle output appeared during no-replace publication") from exc published_item = PublishedFile( destination=staged.destination, payload=staged.payload, device=temporary_metadata.st_dev, inode=temporary_metadata.st_ino, ) published.append(published_item) metadata = staged.destination.lstat() if ( not stat.S_ISREG(metadata.st_mode) or metadata.st_dev != temporary_metadata.st_dev or metadata.st_ino != temporary_metadata.st_ino or stat.S_IMODE(metadata.st_mode) != 0o600 or staged.destination.read_bytes() != staged.payload ): raise BundleError("publication_readback_failed", "published bundle output failed exact readback") for directory in directories: _sync_directory(directory) for staged in staged_files: staged.temporary.unlink() for directory in directories: _sync_directory(directory) for item in published: metadata = item.destination.lstat() if ( item.destination.is_symlink() or not stat.S_ISREG(metadata.st_mode) or stat.S_IMODE(metadata.st_mode) != 0o600 or metadata.st_nlink != 1 or item.destination.read_bytes() != item.payload ): raise BundleError("publication_readback_failed", "published bundle output failed exact readback") return published except BaseException: _rollback_published(published) raise finally: for staged in staged_files: try: staged.temporary.unlink() except FileNotFoundError: pass def _strict_admission_readback(path: Path) -> None: manifest = _read_json_object(path, label="admission manifest") if manifest.get("reviewed_by_handle") != CANONICAL_REVIEWER_HANDLE: raise BundleError("invalid_reviewer_handle", "admission reviewer must equal m3taversal exactly") _canonical_uuid(manifest.get("reviewed_by_agent_id"), label="admission reviewer agent id") def _preflight_bundle( *, genesis_dump: Path, genesis_manifest: Path, ledger: Path, ledger_sha256: str, admission_manifest: Path | None, admission_manifest_sha256: str | None, expected_final_manifest: Path | None = None, expected_materials: list[Path] | None = None, ) -> genesis_rebuild.ReconstructionBundle: arguments = argparse.Namespace( genesis_dump=genesis_dump, genesis_manifest=genesis_manifest, ledger=ledger, ledger_sha256=ledger_sha256, admission_manifest=admission_manifest, admission_manifest_sha256=admission_manifest_sha256, output=ledger.parent / f".{ledger.name}.preflight-receipt-{uuid.uuid4().hex}.json", ) try: bundle = genesis_rebuild.load_bundle(arguments) except genesis_rebuild.ReconstructionError as exc: raise BundleError("genesis_preflight_refused", "existing genesis ledger preflight refused the bundle") from exc for entry in bundle.entries: _validate_source_normalization(entry) if expected_final_manifest is not None and bundle.final_manifest_path != expected_final_manifest.resolve(): raise BundleError("final_manifest_mismatch", "ledger does not select the supplied final parity manifest") if expected_materials is not None: observed = [entry.material_path for entry in bundle.entries] expected = [path.resolve() for path in expected_materials] if observed != expected: raise BundleError("material_order_mismatch", "ledger does not select the supplied material order exactly") if admission_manifest is not None: _strict_admission_readback(admission_manifest) return bundle def _verify_deterministic_json(path: Path, *, label: str) -> dict[str, Any]: value = _read_json_object(path, label=label) if path.read_bytes() != _canonical_json_bytes(value): raise BundleError("nondeterministic_json", f"{label} is not in deterministic builder encoding") return value def _admission_summary(*, emitted: bool, sha256: str | None = None) -> dict[str, Any]: if not emitted: return {"status": "not_requested", "emitted": False} return { "status": "explicit_review_metadata_supplied", "emitted": True, "sha256": sha256, "review_metadata_provenance": REVIEW_METADATA_PROVENANCE, "cryptographically_authenticated": False, } def _require_build_arguments(args: argparse.Namespace) -> None: if args.output_ledger is None: raise BundleError("missing_output", "build mode requires --output-ledger") if args.final_manifest is None: raise BundleError("missing_final_manifest", "build mode requires --final-manifest") if not args.materials: raise BundleError("missing_material", "build mode requires at least one ordered --material") if ( args.ledger_sha256 is not None or args.admission_manifest is not None or args.admission_manifest_sha256 is not None ): raise BundleError("invalid_build_option", "pre-existing ledger or admission pins are verify-only options") admission_values = ( args.admission_output, args.reviewed_by, args.reviewed_by_agent_id, args.reviewed_at, args.review_note_file, args.admission_scope, args.admission_basis, ) if not args.emit_admission_manifest and any(value is not None for value in admission_values): raise BundleError( "implicit_admission_forbidden", "admission metadata requires the explicit --emit-admission-manifest flag", ) if args.emit_admission_manifest and any(value is None for value in admission_values): raise BundleError( "incomplete_admission_review", "explicit admission requires output, reviewer, timestamp, note file, scope, and basis", ) def _require_verify_arguments(args: argparse.Namespace) -> Path: supplied = args.verify_ledger ledger = Path(supplied) if supplied else args.verify_ledger_alias or args.output_ledger if ledger is None: raise BundleError("missing_ledger", "--verify-ledger requires a ledger path") if args.ledger_sha256 is None or not genesis_rebuild.SHA256_RE.fullmatch(args.ledger_sha256): raise BundleError("missing_ledger_pin", "--verify-ledger requires --ledger-sha256") if args.emit_admission_manifest or any( value is not None for value in ( args.admission_output, args.reviewed_by, args.reviewed_by_agent_id, args.reviewed_at, args.review_note_file, args.admission_scope, args.admission_basis, ) ): raise BundleError("invalid_verify_option", "verify mode never emits or derives admission metadata") if (args.admission_manifest is None) != (args.admission_manifest_sha256 is None): raise BundleError( "incomplete_admission_pin", "verify mode requires both --admission-manifest and --admission-manifest-sha256", ) return ledger def build(args: argparse.Namespace) -> dict[str, Any]: _require_build_arguments(args) ledger_output = _prepare_new_output(args.output_ledger, label="--output-ledger") admission_output = ( _prepare_new_output(args.admission_output, label="--admission-output") if args.emit_admission_manifest else None ) genesis_dump = _regular_input(args.genesis_dump, label="--genesis-dump") genesis_manifest = _regular_input(args.genesis_manifest, label="--genesis-manifest") final_manifest = _regular_input(args.final_manifest, label="--final-manifest") review_note_file = ( _regular_input(args.review_note_file, label="--review-note-file") if args.emit_admission_manifest else None ) material_paths = [_regular_input(path, label=f"material {index}") for index, path in enumerate(args.materials, 1)] path_inventory = [ (genesis_dump, "--genesis-dump"), (genesis_manifest, "--genesis-manifest"), (final_manifest, "--final-manifest"), (ledger_output, "--output-ledger"), *[(path, f"material {index}") for index, path in enumerate(material_paths, 1)], ] if review_note_file is not None: path_inventory.append((review_note_file, "--review-note-file")) if admission_output is not None: path_inventory.append((admission_output, "--admission-output")) _assert_distinct_paths(path_inventory) _validate_dump_and_manifests(genesis_dump, genesis_manifest, final_manifest) materials = _validate_materials(material_paths) ledger_payload = _ledger_payload( ledger_path=ledger_output, genesis_dump=genesis_dump, genesis_manifest=genesis_manifest, final_manifest=final_manifest, materials=materials, ) staged_files: list[StagedFile] = [] published: list[PublishedFile] = [] try: staged_ledger = _stage_private_json(ledger_output, ledger_payload) staged_files.append(staged_ledger) ledger_sha256 = _sha256_bytes(staged_ledger.payload) staged_admission: StagedFile | None = None admission_sha256: str | None = None if admission_output is not None and review_note_file is not None: admission_payload = _admission_payload( args, ledger_sha256=ledger_sha256, materials=materials, review_note_file=review_note_file, ) staged_admission = _stage_private_json(admission_output, admission_payload) staged_files.append(staged_admission) admission_sha256 = _sha256_bytes(staged_admission.payload) _preflight_bundle( genesis_dump=genesis_dump, genesis_manifest=genesis_manifest, ledger=staged_ledger.temporary, ledger_sha256=ledger_sha256, admission_manifest=staged_admission.temporary if staged_admission is not None else None, admission_manifest_sha256=admission_sha256, expected_final_manifest=final_manifest, expected_materials=material_paths, ) published = _publish_no_replace(staged_files) _preflight_bundle( genesis_dump=genesis_dump, genesis_manifest=genesis_manifest, ledger=ledger_output, ledger_sha256=ledger_sha256, admission_manifest=admission_output, admission_manifest_sha256=admission_sha256, expected_final_manifest=final_manifest, expected_materials=material_paths, ) except BaseException: if published: _rollback_published(published) raise finally: for staged in staged_files: try: staged.temporary.unlink() except FileNotFoundError: pass return { "artifact": RECEIPT_ARTIFACT, "status": "pass", "action": "build", "entry_count": len(materials), "ledger_sha256": ledger_sha256, "admission": _admission_summary( emitted=admission_output is not None, sha256=admission_sha256, ), } def verify(args: argparse.Namespace) -> dict[str, Any]: raw_ledger = _require_verify_arguments(args) ledger = _regular_input(raw_ledger, label="ledger", private_json=True) genesis_dump = _regular_input(args.genesis_dump, label="--genesis-dump") genesis_manifest = _regular_input(args.genesis_manifest, label="--genesis-manifest") expected_final_manifest = ( _regular_input(args.final_manifest, label="--final-manifest") if args.final_manifest is not None else None ) expected_materials = [ _regular_input(path, label=f"material {index}") for index, path in enumerate(args.materials, 1) ] admission_manifest = ( _regular_input(args.admission_manifest, label="--admission-manifest", private_json=True) if args.admission_manifest is not None else None ) _verify_deterministic_json(ledger, label="ledger") actual_ledger_sha256 = genesis_rebuild.canonical_rebuild.sha256_file(ledger) if actual_ledger_sha256 != args.ledger_sha256: raise BundleError("ledger_hash_mismatch", "--ledger-sha256 does not match the exact ledger bytes") if admission_manifest is not None: _verify_deterministic_json(admission_manifest, label="admission manifest") actual_admission_sha256 = genesis_rebuild.canonical_rebuild.sha256_file(admission_manifest) if actual_admission_sha256 != args.admission_manifest_sha256: raise BundleError( "admission_hash_mismatch", "--admission-manifest-sha256 does not match the exact admission bytes", ) else: actual_admission_sha256 = None bundle = _preflight_bundle( genesis_dump=genesis_dump, genesis_manifest=genesis_manifest, ledger=ledger, ledger_sha256=args.ledger_sha256, admission_manifest=admission_manifest, admission_manifest_sha256=args.admission_manifest_sha256, expected_final_manifest=expected_final_manifest, expected_materials=expected_materials or None, ) return { "artifact": RECEIPT_ARTIFACT, "status": "pass", "action": "verify", "entry_count": len(bundle.entries), "ledger_sha256": actual_ledger_sha256, "admission": _admission_summary( emitted=admission_manifest is not None, sha256=actual_admission_sha256, ), } def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--genesis-dump", required=True, type=Path) parser.add_argument("--genesis-manifest", required=True, type=Path) parser.add_argument("--final-manifest", type=Path) parser.add_argument("--material", dest="materials", action="append", default=[], type=Path) parser.add_argument("--output-ledger", "--ledger-output", "--output", dest="output_ledger", type=Path) parser.add_argument( "--verify-ledger", nargs="?", const="", metavar="LEDGER", help="verify an existing ledger; path may instead be supplied with --ledger or --output-ledger", ) parser.add_argument("--ledger", dest="verify_ledger_alias", type=Path) parser.add_argument("--ledger-sha256") parser.add_argument( "--emit-admission-manifest", "--write-admission-manifest", dest="emit_admission_manifest", action="store_true", help="emit unauthenticated operator-supplied review metadata; this never grants canonical eligibility", ) parser.add_argument("--admission-output", "--output-admission-manifest", dest="admission_output", type=Path) parser.add_argument("--reviewed-by") parser.add_argument("--reviewed-by-agent-id", "--reviewer-agent-id", dest="reviewed_by_agent_id") parser.add_argument("--reviewed-at") parser.add_argument("--review-note-file", type=Path) parser.add_argument("--admission-scope", choices=sorted(genesis_rebuild.ADMISSION_SCOPES)) parser.add_argument("--admission-basis", choices=sorted(genesis_rebuild.ADMISSION_BASES)) parser.add_argument("--admission-manifest", type=Path) parser.add_argument("--admission-manifest-sha256") return parser.parse_args(argv) def run(args: argparse.Namespace) -> dict[str, Any]: verify_mode = args.verify_ledger is not None or args.verify_ledger_alias is not None return verify(args) if verify_mode else build(args) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) try: payload = run(args) except BundleError as exc: print( json.dumps( { "artifact": RECEIPT_ARTIFACT, "status": "refused", "code": exc.code, "message": exc.safe_message, }, indent=2, sort_keys=True, ), file=sys.stderr, ) return 1 except OSError: print( json.dumps( { "artifact": RECEIPT_ARTIFACT, "status": "refused", "code": "private_file_operation_failed", "message": "private bundle file operation failed; details withheld", }, indent=2, sort_keys=True, ), file=sys.stderr, ) return 1 print(json.dumps(payload, indent=2, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())