"""muse verify-pack — verify the integrity of a MPack. Reads a MPack msgpack binary from stdin (or ``--file``) and performs three levels of integrity checking: 1. **Blob integrity** — every ``blobs`` entry has its SHA-256 recomputed from the raw ``content`` bytes. The digest must match the declared ``object_id``. 2. **Snapshot consistency** — every snapshot in the mpack references only object IDs that are either in the mpack itself or already present in the local store. Orphaned manifest entries are reported as failures. 3. **Commit consistency** — every commit in the mpack references a ``snapshot_id`` that is either in the mpack or already in the local store. Pipe from ``pack-objects`` to validate before sending to a remote:: muse pack-objects | muse verify-pack Or verify a saved mpack file:: muse verify-pack --file mpack.muse Quick structural inspection without full hash verification:: muse verify-pack --stat --file mpack.muse Output (JSON, default):: { "blobs_checked": 42, "snapshots_checked": 5, "commits_checked": 5, "all_ok": true, "failures": [], "promised_objects": 0, "base_objects": 0, "bundle_mode": "full", "base_commits": [], "duration_ms": 1.234, "exit_code": 0 } With failures:: { "blobs_checked": 42, "snapshots_checked": 5, "commits_checked": 5, "all_ok": false, "failures": [ {"kind": "object", "id": "", "error": "hash mismatch"}, {"kind": "snapshot", "id": "", "error": "missing object: "} ], "promised_objects": 3, "base_objects": 0, "bundle_mode": "full", "base_commits": [], "duration_ms": 1.234, "exit_code": 1 } Stat-only output (``--stat``):: {"blobs": 42, "snapshots": 5, "commits": 5, "duration_ms": 0.456, "exit_code": 0} Object availability model ------------------------- Objects absent from the mpack are resolved against the local store using a three-state model identical to ``muse verify``: - **PRESENT** — object exists in the local ``.muse/objects/`` store (hash verified). Not a failure. - **PROMISED** — absent locally but at least one promisor remote is configured (``.muse/config.toml``). Counted in ``promised_objects``; does not affect ``all_ok``. Use ``--strict`` to treat promised objects as failures. - **MISSING** — absent locally with no promisor remote. Always a failure. Output contract --------------- - Exit 0: mpack is fully intact; or ``--stat`` completed. - Exit 1: one or more integrity failures; malformed msgpack input; bad format. - Exit 3: I/O error reading stdin or the mpack file. Agent use --------- Verify before pushing:: muse pack-objects "$TIP" \\ | muse verify-pack --json \\ | python3 -c "import sys,json; d=json.load(sys.stdin); sys.exit(0 if d['all_ok'] else 1)" Inspect mpack structure without hashing (fast):: muse verify-pack --stat --file mpack.muse --json Quiet CI gate — fails pipeline if mpack is corrupt:: muse pack-objects "$TIP" --file mpack.muse muse verify-pack --quiet --file mpack.muse """ import argparse import json import logging import pathlib import sys from typing import TypedDict from muse.core.types import blob_id from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.errors import ExitCode from muse.core.object_availability import ObjectState, load_promisor_remotes, object_state from muse.core.object_store import read_object from muse.core.repo import require_repo from muse.core.io import MAX_PACK_MSGPACK_BYTES, safe_unpackb from muse.core.snapshots import read_snapshot from muse.core.validation import sanitize_display, validate_object_id from muse.core.timing import start_timer logger = logging.getLogger(__name__) class _Failure(TypedDict): kind: str id: str error: str class _VerifyPackJson(EnvelopeJson): blobs_checked: int snapshots_checked: int commits_checked: int all_ok: bool failures: list[_Failure] promised_objects: int base_objects: int bundle_mode: str base_commits: list[str] class _StatResultJson(EnvelopeJson): blobs: int snapshots: int commits: int def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the verify-pack subcommand.""" parser = subparsers.add_parser( "verify-pack", help="Verify the integrity of a MPack.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--file", "-i", default=None, dest="bundle_file", metavar="PATH", help="Path to a MPack file. Reads from stdin when omitted.", ) parser.add_argument( "--stat", action="store_true", dest="stat_only", help=( "Fast structural inspection: count objects, snapshots, and commits " "without computing any hashes. Exits 0 on valid msgpack structure." ), ) parser.add_argument( "--quiet", "-q", action="store_true", help="No output. Exit 0 if all checks pass, exit 1 otherwise.", ) parser.add_argument( "--no-local", "-L", action="store_true", dest="skip_local_check", help="Skip checking the local store for missing snapshot/commit refs.", ) parser.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit machine-readable JSON on stdout.", ) parser.add_argument( "--strict", action="store_true", help=( "Treat promised objects (absent locally but covered by a promisor remote) " "as integrity failures. By default promised objects are counted in " "``promised_objects`` and do not affect ``all_ok``." ), ) parser.set_defaults(func=run, json_out=False) def run(args: argparse.Namespace) -> None: """Verify the integrity of a MPack. Reads a MPack from stdin or ``--file`` and checks: - Every object's payload re-hashes to its declared SHA-256 ID. - Every snapshot's manifest references objects present in the mpack, the local store, or a configured promisor remote. - Every commit's snapshot ID is present in the mpack or the local store. Objects absent from the mpack are resolved with the three-state model: PRESENT (local store, hash-verified), PROMISED (absent locally but a promisor remote is configured — counted in ``promised_objects``, not a failure unless ``--strict``), or MISSING (absent with no promisor — always a failure). Use ``--stat`` for a fast structural count without hash verification. Use ``--strict`` to require full self-containment (treats promised objects as failures). Agent quickstart:: muse pack-objects HEAD | muse verify-pack --json muse verify-pack --file mpack.muse --json muse verify-pack --stat --file mpack.muse --json muse verify-pack --strict --file mpack.muse --json JSON fields:: blobs_checked Number of blobs whose SHA-256 was recomputed. snapshots_checked Number of snapshot manifests checked. commits_checked Number of commit records checked. all_ok true when every check passed. failures List of {kind, id, error} failure records. promised_objects Count of objects deferred to promisor remotes. base_objects Count of objects expected at the incremental base. bundle_mode "full" or "incremental". base_commits List of base commit IDs for incremental bundles. muse_version Muse release that produced this output. schema Envelope schema version (int). exit_code 0 all ok, 1 any failure. duration_ms Wall-clock milliseconds for the command. timestamp ISO-8601 UTC timestamp of command completion. warnings List of non-fatal advisory messages. Exit codes:: 0 MPack fully intact, or --stat completed. 1 One or more integrity failures; malformed msgpack; bad format. 3 I/O error reading stdin or the mpack file. """ elapsed = start_timer() json_out: bool = args.json_out bundle_file: str | None = args.bundle_file quiet: bool = args.quiet skip_local_check: bool = args.skip_local_check stat_only: bool = args.stat_only strict: bool = args.strict # Read mpack bytes. if bundle_file is not None: try: raw_bytes = pathlib.Path(bundle_file).read_bytes() except OSError as exc: print( json.dumps({"error": f"Cannot read file: {sanitize_display(str(exc))}"}), file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) else: try: raw_bytes = sys.stdin.buffer.read() except OSError as exc: print( json.dumps({"error": f"Cannot read stdin: {exc}"}), file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) try: mpack = safe_unpackb( raw_bytes, context="pack input", max_bytes=MAX_PACK_MSGPACK_BYTES, allow_binary=True, ) except (ValueError, TypeError, Exception) as exc: print(json.dumps({"error": f"Invalid msgpack: {exc}"}), file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if not isinstance(mpack, dict): print( json.dumps({"error": "MPack must be a msgpack map."}), file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # --stat: fast structural count — no hash verification. if stat_only: blobs_raw = mpack.get("blobs", []) snapshots_raw = mpack.get("snapshots", []) commits_raw = mpack.get("commits", []) stat_result = { "blobs": len(blobs_raw) if isinstance(blobs_raw, list) else 0, "snapshots": len(snapshots_raw) if isinstance(snapshots_raw, list) else 0, "commits": len(commits_raw) if isinstance(commits_raw, list) else 0, } if not json_out: print( f"blobs={stat_result['blobs']} " f"snapshots={stat_result['snapshots']} " f"commits={stat_result['commits']} " f"duration_ms={elapsed()}" ) else: print(json.dumps(_StatResultJson( **make_envelope(elapsed), blobs=stat_result["blobs"], snapshots=stat_result["snapshots"], commits=stat_result["commits"], ))) return # Every mpack must carry a meta field — it is always written by build_mpack. bundle_meta = mpack.get("meta") if not isinstance(bundle_meta, dict): print(json.dumps({"error": "MPack is missing required 'meta' field."}), file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) bundle_mode: str = bundle_meta.get("mode", "") if bundle_mode not in ("full", "incremental"): print(json.dumps({"error": f"meta.mode must be 'full' or 'incremental', got {bundle_mode!r}."}), file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) raw_base = bundle_meta.get("base_commits") if not isinstance(raw_base, list): print(json.dumps({"error": "meta.base_commits must be a list."}), file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) base_commits: list[str] = [str(c) for c in raw_base] # Incremental bundles have unresolved refs expected at the base — track separately. base_objects_count = 0 # We need the repo root for local-store checks (optional). root: pathlib.Path | None = require_repo() if not skip_local_check else None # Load promisor remotes once — used in snapshot manifest checks to distinguish # PROMISED objects (absent locally but expected on a remote) from MISSING ones. promisor_remotes: list[str] = load_promisor_remotes(root) if root is not None else [] promised_count = 0 failures: list[_Failure] = [] # ----------------------------------------------------------------------- # 1. Blob integrity — re-hash each payload. # ----------------------------------------------------------------------- bundle_object_ids: set[str] = set() blobs_raw = mpack.get("blobs", []) if not isinstance(blobs_raw, list): print( json.dumps({"error": "'blobs' field must be a list."}), file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) for entry in blobs_raw: if not isinstance(entry, dict): failures.append( _Failure(kind="object", id="(unknown)", error="entry is not a dict") ) continue oid = entry.get("object_id", "") content = entry.get("content") if not isinstance(oid, str) or not isinstance(content, (bytes, bytearray)): failures.append( _Failure( kind="object", id="(unknown)", error="missing or invalid object_id / content fields", ) ) continue # Validate object ID format before embedding it anywhere. try: validate_object_id(oid) except ValueError: failures.append( _Failure( kind="object", id="(invalid)", error=f"object_id is not a valid sha256-prefixed hex ID: {oid[:24]!r}", ) ) continue # Hash without copying: msgpack returns bytes; hashlib accepts bytes/bytearray. actual = blob_id(content) if actual != oid: failures.append( _Failure( kind="object", id=oid, error=( f"hash mismatch: declared {oid} " f"recomputed {actual}" ), ) ) else: bundle_object_ids.add(oid) blobs_checked = len(blobs_raw) # ----------------------------------------------------------------------- # 2. Snapshot consistency — manifest entries must be present. # ----------------------------------------------------------------------- bundle_snapshot_ids: set[str] = set() snapshots_raw = mpack.get("snapshots", []) if not isinstance(snapshots_raw, list): print( json.dumps({"error": "'snapshots' field must be a list."}), file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) for snap_entry in snapshots_raw: if not isinstance(snap_entry, dict): failures.append( _Failure( kind="snapshot", id="(unknown)", error="snapshot entry is not a dict" ) ) continue snap_id = snap_entry.get("snapshot_id", "") if not isinstance(snap_id, str): failures.append( _Failure(kind="snapshot", id="(unknown)", error="missing snapshot_id") ) continue bundle_snapshot_ids.add(snap_id) manifest = snap_entry.get("manifest", {}) if not isinstance(manifest, dict): continue for path, obj_id in manifest.items(): if not isinstance(obj_id, str): continue if obj_id in bundle_object_ids: continue # Object not in mpack. if root is not None: # Local store available — use the PRESENT / PROMISED / MISSING tristate. try: local_content = read_object(root, obj_id) except OSError as exc: failures.append( _Failure( kind="object", id=obj_id, error=( f"local store object {obj_id} failed " f"SHA-256 integrity check: {exc}" ), ) ) continue if local_content is not None: continue # PRESENT and verified state = object_state(root, obj_id, promisor_remotes) if state == ObjectState.PROMISED and not strict: promised_count += 1 continue else: # No local store (--no-local). For incremental bundles, unresolved # refs are expected to exist at the declared base — not a failure # unless --strict is set. if bundle_mode == "incremental" and not strict: base_objects_count += 1 continue failures.append( _Failure( kind="snapshot", id=snap_id, error=( f"manifest path {path!r} references " f"missing object {obj_id}" ), ) ) snapshots_checked = len(snapshots_raw) # ----------------------------------------------------------------------- # 3. Commit consistency — snapshot_id must be resolvable. # ----------------------------------------------------------------------- commits_raw = mpack.get("commits", []) if not isinstance(commits_raw, list): print( json.dumps({"error": "'commits' field must be a list."}), file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) for commit_entry in commits_raw: if not isinstance(commit_entry, dict): failures.append( _Failure( kind="commit", id="(unknown)", error="commit entry is not a dict" ) ) continue commit_id = commit_entry.get("commit_id", "") snap_id = commit_entry.get("snapshot_id", "") if not isinstance(commit_id, str) or not isinstance(snap_id, str) or not snap_id: failures.append( _Failure( kind="commit", id=commit_id if isinstance(commit_id, str) else "(unknown)", error="missing commit_id or snapshot_id", ) ) continue if snap_id in bundle_snapshot_ids: continue if root is not None and read_snapshot(root, snap_id) is not None: continue if not skip_local_check: failures.append( _Failure( kind="commit", id=commit_id, error=f"references snapshot {snap_id} not in mpack or local store", ) ) commits_checked = len(commits_raw) all_ok = len(failures) == 0 if quiet: raise SystemExit(0 if all_ok else ExitCode.USER_ERROR) if not json_out: print( f"blobs={blobs_checked} snapshots={snapshots_checked} " f"commits={commits_checked} all_ok={all_ok}" ) for f in failures: print( f" FAIL [{sanitize_display(f['kind'])}] " f"{sanitize_display(f['id'])} " f"{sanitize_display(f['error'])}" ) if not all_ok: raise SystemExit(ExitCode.USER_ERROR) return print(json.dumps(_VerifyPackJson( **make_envelope(elapsed, exit_code=0 if all_ok else int(ExitCode.USER_ERROR)), blobs_checked=blobs_checked, snapshots_checked=snapshots_checked, commits_checked=commits_checked, all_ok=all_ok, failures=failures, promised_objects=promised_count, base_objects=base_objects_count, bundle_mode=bundle_mode, base_commits=base_commits, ))) if not all_ok: raise SystemExit(ExitCode.USER_ERROR)