"""muse plumbing verify-object — verify the integrity of stored objects. Reads one or more objects from the content-addressed store and re-hashes each one to confirm that its on-disk content still matches its claimed SHA-256 identity. Reports the result per object and exits non-zero if any object fails verification. This is the integrity primitive used by backup systems, replication agents, and CI pipelines to detect silent data corruption without a full fsck. Output (JSON, default):: { "results": [ {"object_id": "", "ok": true, "size_bytes": 4096}, {"object_id": "", "ok": false, "size_bytes": 512, "error": "hash mismatch: stored recomputed "}, {"object_id": "", "ok": false, "size_bytes": null, "error": "object not found in store"} ], "all_ok": false, "checked": 3, "failed": 2 } Text output (``--format text``):: OK (4096 bytes) FAIL hash mismatch: stored abc123… recomputed def456… FAIL object not found in store With ``--quiet``: no output; exits 0 if all pass, exits 1 otherwise. Plumbing contract ----------------- - Exit 0: all objects verified successfully. - Exit 1: one or more objects failed verification; object not found; bad args. - Exit 3: unexpected I/O error (e.g. disk read failure). Performance ----------- Object size is measured by counting bytes as they are read during hashing — no separate ``stat()`` call is made. For ``--all`` over a large store this saves one syscall per object. Agent use --------- Verify specific objects after a replication or backup:: muse plumbing verify-object --json Full store integrity check (fsck equivalent):: muse plumbing verify-object --all --json Pipe object IDs from a manifest:: cat .muse/snapshot_manifest.txt | muse plumbing verify-object --stdin Quiet mode for CI:: muse plumbing verify-object --all --quiet && echo "store intact" """ from __future__ import annotations import argparse import hashlib import json import logging import pathlib import sys from typing import TypedDict from muse.core.errors import ExitCode from muse.core.object_store import object_path, objects_dir from muse.core.repo import require_repo from muse.core.validation import sanitize_display, validate_object_id logger = logging.getLogger(__name__) _FORMAT_CHOICES = ("json", "text") _CHUNK = 65536 # 64 KiB read chunks — keeps the heap clean for large blobs class _ObjectResult(TypedDict): object_id: str ok: bool size_bytes: int | None error: str | None def _iter_all_object_ids(root: pathlib.Path) -> list[str]: """Walk the object store shard tree and return every stored object ID. The store layout is ``/.muse/objects//``. Symlinks in the shard directories are skipped — they are not a valid Muse store layout and could point outside the repository. Returns: Sorted list of 64-hex-char object IDs present on disk. """ store = objects_dir(root) if not store.exists(): return [] ids: list[str] = [] for shard_dir in sorted(store.iterdir()): if shard_dir.is_symlink() or not shard_dir.is_dir(): continue if len(shard_dir.name) != 2: continue prefix = shard_dir.name for obj_file in sorted(shard_dir.iterdir()): if obj_file.is_symlink() or not obj_file.is_file(): continue oid = prefix + obj_file.name if len(oid) == 64: ids.append(oid) return ids def _verify_one(root: pathlib.Path, object_id: str) -> _ObjectResult: """Integrity-check a single object and return its result record. Streams the object in 64 KiB chunks to avoid loading large blobs into memory. Measures size by counting bytes during hashing — no separate ``stat()`` call is needed. Returns an :class:`_ObjectResult` — never raises. """ try: validate_object_id(object_id) except ValueError as exc: return { "object_id": object_id, "ok": False, "size_bytes": None, "error": str(exc), } dest = object_path(root, object_id) if not dest.exists(): return { "object_id": object_id, "ok": False, "size_bytes": None, "error": "object not found in store", } try: size = 0 h = hashlib.sha256() with dest.open("rb") as fh: for chunk in iter(lambda: fh.read(_CHUNK), b""): h.update(chunk) size += len(chunk) actual = h.hexdigest() except OSError as exc: return { "object_id": object_id, "ok": False, "size_bytes": None, "error": f"I/O error: {exc}", } if actual != object_id: return { "object_id": object_id, "ok": False, "size_bytes": size, "error": ( f"hash mismatch: stored {object_id[:12]}… " f"recomputed {actual[:12]}…" ), } return {"object_id": object_id, "ok": True, "size_bytes": size, "error": None} def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the verify-object subcommand.""" parser = subparsers.add_parser( "verify-object", help="Re-hash stored objects to detect data corruption.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "object_ids", nargs="*", help=( "One or more SHA-256 object IDs to verify. " "Required unless --all or --stdin is used." ), ) parser.add_argument( "--all", "-a", action="store_true", dest="verify_all", help=( "Verify every object in the store — the fsck equivalent. " "No object ID arguments needed. " "Cannot be combined with positional object IDs." ), ) parser.add_argument( "--stdin", action="store_true", dest="from_stdin", help=( "Read additional object IDs from stdin, one per line. " "Blank lines and '#'-comments are skipped. " "Combines with positional object ID arguments." ), ) parser.add_argument( "--quiet", "-q", action="store_true", help="No output. Exit 0 if all objects are intact, exit 1 otherwise.", ) parser.add_argument( "--format", "-f", dest="fmt", default="json", metavar="FORMAT", help="Output format: json or text. (default: json)", ) parser.add_argument( "--json", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.", ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Verify the integrity of one or more objects in the store. Re-hashes each object's on-disk content and confirms it matches the SHA-256 identity used as its filename. Any mismatch indicates silent data corruption and is reported as a failure. Size is counted during hashing — no separate stat() call is made. """ fmt: str = args.fmt cli_ids: list[str] = args.object_ids or [] verify_all: bool = args.verify_all from_stdin: bool = args.from_stdin quiet: bool = args.quiet if fmt not in _FORMAT_CHOICES: print( json.dumps( {"error": f"Unknown format {fmt!r}. Valid: {', '.join(_FORMAT_CHOICES)}"} ), file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if verify_all and cli_ids: print( json.dumps( {"error": "--all cannot be combined with explicit object ID arguments."} ), file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() # Collect object IDs from all sources. if verify_all: object_ids: list[str] = _iter_all_object_ids(root) else: object_ids = list(cli_ids) if from_stdin: for line in sys.stdin: # Strip \r\n — CRLF from Windows or injection would embed \r # in the ID, causing validate_object_id to reject it with a # confusing error rather than a clear "malformed input" message. stripped = line.rstrip("\r\n") if stripped and not stripped.startswith("#"): object_ids.append(stripped) if not object_ids and not verify_all: print( json.dumps( {"error": "At least one object ID is required (or use --all / --stdin)."} ), file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) results: list[_ObjectResult] = [_verify_one(root, oid) for oid in object_ids] all_ok = all(r["ok"] for r in results) failed_count = sum(1 for r in results if not r["ok"]) if quiet: raise SystemExit(0 if all_ok else ExitCode.USER_ERROR) if fmt == "text": for r in results: status = "OK " if r["ok"] else "FAIL" oid_safe = sanitize_display(r["object_id"]) size_str = f" ({r['size_bytes']} bytes)" if r["size_bytes"] is not None else "" err_str = ( f" {sanitize_display(r['error'])}" if not r["ok"] and r["error"] else "" ) print(f"{status} {oid_safe}{size_str}{err_str}") if not all_ok: raise SystemExit(ExitCode.USER_ERROR) return print( json.dumps({ "results": [dict(r) for r in results], "all_ok": all_ok, "checked": len(results), "failed": failed_count, }) ) if not all_ok: raise SystemExit(ExitCode.USER_ERROR)