"""muse 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. 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, "error": null}, {"object_id": "", "ok": false, "size_bytes": 512, "error": "hash mismatch: stored abc123… recomputed def456…"}, {"object_id": "", "ok": false, "size_bytes": null, "error": "object not found in store"} ], "all_ok": false, "checked": 3, "failed": 2, "duration_ms": 1.234, "exit_code": 1 } Text output (``--format text``):: OK (4096 bytes) FAIL hash mismatch: stored abc123… recomputed def456… FAIL object not found in store --- Checked: 3 Failed: 2 With ``--quiet``: no output; exits 0 if all pass, exits 1 otherwise. Output 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). - ``duration_ms`` and ``exit_code`` are always present in JSON output so agents can measure latency and check results without inspecting the process exit status separately. - Results are returned in the same order as the input object IDs. - ``error`` is always ``null`` when ``ok`` is ``true``. 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. Use ``--fail-fast`` in CI to bail immediately after the first failure instead of scanning the entire store. Agent use --------- Verify specific objects after a replication or backup:: muse verify-object --json Full store integrity check (fsck equivalent):: muse verify-object --all --json Fail fast — bail after the first corrupt object:: muse verify-object --all --fail-fast --json Pipe object IDs from a manifest:: cat manifest.txt | muse verify-object --stdin --json Quiet mode for CI:: muse verify-object --all --quiet && echo "store intact" """ import argparse import hashlib import json import logging import pathlib import sys from typing import TypedDict from muse.core.types import long_id from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.errors import ExitCode from muse.core.object_store import iter_stored_objects, object_path, objects_dir from muse.core.repo import require_repo from muse.core.timing import start_timer from muse.core.validation import sanitize_display, validate_object_id logger = logging.getLogger(__name__) _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 class _VerifyObjectJson(EnvelopeJson): results: list[_ObjectResult] all_ok: bool checked: int failed: int 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/sha256//``. 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 ``sha256:<64hex>`` object IDs present on disk. """ return sorted(oid for oid, _ in iter_stored_objects(root)) 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(): # Fall through to pack store before reporting missing. from muse.core.pack_store import read_object_from_packs try: content = read_object_from_packs(root, object_id) except OSError as exc: return { "object_id": object_id, "ok": False, "size_bytes": None, "error": f"pack integrity error: {exc}", } if content is None: return { "object_id": object_id, "ok": False, "size_bytes": None, "error": "object not found in store", } # read_object_from_packs verifies the hash internally — reaching here # means the object is present and correct. return {"object_id": object_id, "ok": True, "size_bytes": len(content), "error": None} try: h = hashlib.sha256() total_size = 0 payload_size: int | None = None leftover = b"" with dest.open("rb") as fh: for chunk in iter(lambda: fh.read(_CHUNK), b""): h.update(chunk) total_size += len(chunk) if payload_size is None: # Find the null byte that terminates the " \0" header. buf = leftover + chunk null_pos = buf.find(b"\0") if null_pos != -1: try: _, size_str = buf[:null_pos].decode("ascii").split(" ", 1) payload_size = int(size_str) except (ValueError, UnicodeDecodeError): payload_size = total_size # corrupt header: report raw size else: leftover = buf # If header was never found (truncated/corrupt), fall back to total size. if payload_size is None: payload_size = total_size actual = long_id(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": payload_size, "error": ( f"hash mismatch: stored {object_id} " f"recomputed {actual}" ), } return {"object_id": object_id, "ok": True, "size_bytes": payload_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( "--fail-fast", action="store_true", dest="fail_fast", help=( "Stop after the first failed object and exit 1 immediately. " "Useful in CI to avoid scanning an entire store when a single " "corruption has already been detected." ), ) parser.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit machine-readable JSON on stdout.", ) parser.set_defaults(func=run, json_out=False) def run(args: argparse.Namespace) -> None: """Re-hash stored objects to detect silent data corruption. Streams each object in 64 KiB chunks, computes SHA-256, and compares it to the content-addressed filename. Any mismatch indicates corruption. Results are returned in the same order as the input IDs; size is counted during hashing so no extra stat() call is needed. Agent quickstart:: muse verify-object sha256: --json muse verify-object --all --json # full store fsck muse verify-object --all --fail-fast --json cat manifest.txt | muse verify-object --stdin --json JSON fields:: results List of {object_id, ok, size_bytes, error} per object. all_ok true when every object passed. checked Total number of objects checked. failed Number of objects that failed verification. 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 All objects verified successfully. 1 One or more objects failed verification or were not found. 3 Unexpected I/O error during hashing. """ elapsed = start_timer() json_out: bool = args.json_out cli_ids: list[str] = args.object_ids or [] verify_all: bool = args.verify_all from_stdin: bool = args.from_stdin quiet: bool = args.quiet fail_fast: bool = args.fail_fast 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) if verify_all and from_stdin: print( json.dumps( {"error": "--all cannot be combined with --stdin."} ), 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) # Verify — optionally bail after the first failure. results: list[_ObjectResult] = [] for oid in object_ids: r = _verify_one(root, oid) results.append(r) if fail_fast and not r["ok"]: break 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 not json_out: 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}") print(f"---\nChecked: {len(results)} Failed: {failed_count}") if not all_ok: raise SystemExit(ExitCode.USER_ERROR) return exit_code = 0 if all_ok else int(ExitCode.USER_ERROR) print(json.dumps(_VerifyObjectJson( **make_envelope(elapsed, exit_code=exit_code), 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)