"""``muse verify-commit ...`` — verify Ed25519 signatures on commits. Reads the ``signature``, ``signer_public_key``, and ``signer_key_id`` fields from one or more commit records, reconstructs the canonical provenance payload, and verifies the Ed25519 signature. Output (per commit) ------------------- Text (default):: OK signer= key= BAD (no signature) BAD (invalid signature) JSON (``--json``):: {"commit_id": "...", "valid": true, "signer": "agent-abc", "key_id": "...", "signed_at": "2026-04-14T17:00:00Z", "key_status": "active|revoked|unknown"} For batch invocations each commit produces one JSON object per line. Flags ----- ``--strict`` Exit non-zero if any commit is unsigned (no signature present). Without ``--strict``, unsigned commits are reported as ``valid=false`` but do not affect the exit code unless the signature is *invalid*. ``--check-key-status`` Query MuseHub to check whether the signing key is still active. Fails closed: returns ``"unknown"`` on timeout, network error, or when no hub is configured. ``--json`` Emit one JSON object per commit, one per line. Exit codes:: 0 — all commits valid (or unsigned without --strict) 1 — at least one invalid signature, or unsigned with --strict 2 — usage error (bad ref, ANSI injection) Examples:: muse verify-commit HEAD muse verify-commit HEAD --json muse verify-commit --strict muse verify-commit --json muse verify-commit HEAD --check-key-status """ import argparse import json as _json import logging import pathlib import re import sys from concurrent.futures import ThreadPoolExecutor, as_completed from typing import TypedDict from muse.core.types import DEFAULT_SIGN_ALGO, decode_pubkey, long_id, short_id, sig_algo from muse.core.paths import ref_path as _ref_path from muse.core.envelope import make_envelope from muse.core.errors import ExitCode from muse.core.provenance import provenance_payload, verify_commit_ed25519 from muse.core.refs import read_ref from muse.core.repo import require_repo from muse.core.refs import ( get_head_commit_id, read_current_branch, ) from muse.core.commits import read_commit from muse.core.timing import start_timer from muse.core.validation import sanitize_display class _VerifyResult(TypedDict, total=False): commit_id: str valid: bool signer: str key_id: str signed_at: str key_status: str logger = logging.getLogger(__name__) # Timeout for hub key-status network calls. _KEY_STATUS_TIMEOUT = 5.0 # sha256:-prefixed commit IDs, HEAD, or branch names. # Colon is required for the sha256: prefix; all other path chars are alphanumeric, _, /, ., -. _SAFE_REF_RE = re.compile(r"^[a-zA-Z0-9_/:.\-]+$") # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _resolve_ref(root: pathlib.Path, treeish: str) -> str | None: """Resolve HEAD or a commit ID / branch name to a full commit ID. Returns None when the ref cannot be resolved. """ if treeish.upper() == "HEAD": try: branch = read_current_branch(root) return get_head_commit_id(root, branch) except Exception: return None # sha256:-prefixed or bare 64-char hex commit ID. if re.fullmatch(r"sha256:[0-9a-f]{64}", treeish): return treeish if re.fullmatch(r"[0-9a-f]{64}", treeish): return long_id(treeish) # Branch name → ref file. ref_file = _ref_path(root, treeish) return read_ref(ref_file) def _fetch_key_status(hub_url: str, key_id: str) -> str: """Query MuseHub for the status of *key_id*. Returns ``"active"``, ``"revoked"``, or ``"unknown"`` (on any error). """ try: import urllib.request url = f"{hub_url.rstrip('/')}/api/keys/{key_id}/status" req = urllib.request.Request(url, method="GET") with urllib.request.urlopen(req, timeout=_KEY_STATUS_TIMEOUT) as resp: body = _json.loads(resp.read().decode("utf-8")) status = body.get("status", "unknown") if status in ("active", "revoked"): return status return "unknown" except Exception: return "unknown" def _verify_one( root: pathlib.Path, commit_id: str, *, check_key_status: bool = False, hub_url: str | None = None, key_status_cache: dict[str, str] | None = None, ) -> _VerifyResult: """Verify the Ed25519 signature on a single commit. Args: root: Repository root path. commit_id: Full 64-char hex commit ID. check_key_status: When True, query MuseHub for key revocation status. hub_url: Hub URL for key-status queries. None → "unknown". key_status_cache: Shared cache dict to deduplicate key-status lookups within a batch invocation. Returns: Dict with keys: commit_id, valid, signer, key_id, signed_at, key_status. ``valid`` is False for unsigned commits, missing public keys, or failed signature verification. """ result: _VerifyResult = { "commit_id": commit_id, "valid": False, "signer": "", "key_id": "", "signed_at": "", "key_status": "unknown", "error": None, } commit = read_commit(root, commit_id) if commit is None: result["error"] = "commit not found" return result # musehub#223: agent_id is empty for human-authored commits -- falling back # to the local author handle avoids "signer": "" reading as broken/confusing # for the common human-signed case. Both fields are already on the local # commit record; no hub lookup needed. result["signer"] = commit.agent_id or commit.author result["key_id"] = commit.signer_key_id result["signed_at"] = commit.committed_at.isoformat() if commit.committed_at else "" # Unsigned commit — not an error unless --strict is applied by the caller. if not commit.signature: return result # Dispatch on algorithm prefix — the prefix is the sole discriminator. sig = commit.signature pub_raw = commit.signer_public_key if sig_algo(sig) != DEFAULT_SIGN_ALGO: result["error"] = f"unrecognised signature algorithm {sig_algo(sig)!r} — re-sign to fix" return result if sig_algo(pub_raw) != DEFAULT_SIGN_ALGO: result["error"] = f"unrecognised public key algorithm {sig_algo(pub_raw)!r} — re-sign to fix" return result try: _, pub_bytes = decode_pubkey(pub_raw) except ValueError: return result if not pub_bytes: return result payload = provenance_payload( commit_id, author=commit.author, agent_id=commit.agent_id, model_id=commit.model_id, toolchain_id=commit.toolchain_id, prompt_hash=commit.prompt_hash, committed_at=commit.committed_at.isoformat(), ) result["valid"] = verify_commit_ed25519(payload, sig, pub_bytes) # Key status enrichment. if check_key_status and commit.signer_key_id: if hub_url is None: result["key_status"] = "unknown" else: if key_status_cache is not None and commit.signer_key_id in key_status_cache: result["key_status"] = key_status_cache[commit.signer_key_id] else: status = _fetch_key_status(hub_url, commit.signer_key_id) result["key_status"] = status if key_status_cache is not None: key_status_cache[commit.signer_key_id] = status return result # --------------------------------------------------------------------------- # Registration # --------------------------------------------------------------------------- def register( subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]", ) -> None: """Register the ``muse verify-commit`` subcommand.""" parser = subparsers.add_parser( "verify-commit", help="Verify Ed25519 signatures on commits.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "commits", metavar="COMMIT", nargs="+", help="Commit ID(s) or HEAD to verify.", ) parser.add_argument( "--strict", action="store_true", help="Exit non-zero if any commit is unsigned.", ) parser.add_argument( "--check-key-status", action="store_true", dest="check_key_status", help="Query MuseHub to check key revocation status.", ) parser.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit one JSON object per commit, one per line.", ) parser.set_defaults(func=run, json_out=False) # --------------------------------------------------------------------------- # Run # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Verify Ed25519 signatures on one or more commits. Resolves each ref to a full commit ID, then verifies the embedded Ed25519 signature against the signer public key stored in the commit record. Per-commit JSON objects are emitted one per line; ``duration_ms`` and ``exit_code`` are included in each line so streaming consumers can act without buffering the full output. Agent quickstart:: muse verify-commit HEAD --json muse verify-commit sha256: --json muse verify-commit sha256: sha256: --strict --json muse verify-commit HEAD --check-key-status --json JSON fields (one object per commit, one per line):: commit_id Full sha256-prefixed commit ID. valid true if the signature verified successfully. signer agent_id string from the commit record. key_id Signer key fingerprint (first 16 hex chars of SHA-256(pubkey)). signed_at ISO-8601 timestamp from the commit record. key_status "active", "revoked", or "unknown" (requires --check-key-status). muse_version Muse release that produced this output. schema Envelope schema version (int). exit_code 0 if all commits valid so far, 1 if any failure. duration_ms Wall-clock milliseconds elapsed so far. timestamp ISO-8601 UTC timestamp of command completion. warnings List of non-fatal advisory messages. Exit codes:: 0 All commits valid (or unsigned without --strict). 1 At least one invalid signature, or unsigned with --strict. 2 Usage error (bad ref, ANSI injection attempt). """ elapsed = start_timer() raw_refs: list[str] = args.commits strict: bool = args.strict check_key_status: bool = args.check_key_status json_out: bool = args.json_out # Validate all refs before doing any work. _HEX_CHARS = frozenset("0123456789abcdef") for ref in raw_refs: if not _SAFE_REF_RE.match(ref): print( f"❌ Invalid ref: {sanitize_display(ref)}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # Bare hex is rejected at the CLI boundary — sha256: prefix is required. # HEAD and branch names contain non-hex characters and are never caught here. if all(c in _HEX_CHARS for c in ref): safe = sanitize_display(ref) print( f"❌ Bare hex IDs are not accepted — use 'sha256:{safe}' instead.\n" f" Even a short prefix works: 'sha256:{safe[:12]}'", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() # Resolve hub URL for key-status queries. hub_url: str | None = None if check_key_status: try: # musehub#223: this previously imported a nonexistent # muse.core.repo.read_hub_url, so the exception handler below # silently caught an ImportError on every single call -- hub_url # was always None regardless of whether a hub was configured, # which is the true root cause of --check-key-status always # reporting "unknown" (more fundamental than the missing # GET /api/keys/{key_id}/status endpoint, fixed separately). from muse.cli.config import get_hub_url hub_url = get_hub_url(root) except Exception: hub_url = None # Resolve refs to commit IDs. commit_ids: list[str] = [] for ref in raw_refs: cid = _resolve_ref(root, ref) if cid is None: print(f"❌ Cannot resolve ref: {sanitize_display(ref)}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) commit_ids.append(cid) key_status_cache: dict[str, str] = {} any_failure = False # Verify commits (parallel for large batches). results: list[dict] = [{}] * len(commit_ids) def _verify_indexed(idx_cid: tuple[int, str]) -> tuple[int, _VerifyResult]: idx, cid = idx_cid return idx, _verify_one( root, cid, check_key_status=check_key_status, hub_url=hub_url, key_status_cache=key_status_cache, ) with ThreadPoolExecutor(max_workers=min(8, len(commit_ids))) as pool: futures = {pool.submit(_verify_indexed, (i, cid)): i for i, cid in enumerate(commit_ids)} for future in as_completed(futures): idx, r = future.result() results[idx] = r for r in results: valid = r.get("valid", False) error = r.get("error") # "commit not found" is always a hard failure. if error: any_failure = True elif not valid: is_signed = bool(r.get("key_id") or r.get("signer")) # Invalid signature on a signed commit → always fail. # Unsigned commit → only fail with --strict. if is_signed or strict: any_failure = True if json_out: exit_code = int(ExitCode.USER_ERROR) if any_failure else 0 emit = {k: v for k, v in r.items() if k != "error"} print(_json.dumps({**make_envelope(elapsed, exit_code=exit_code), **emit})) else: if error: print(f"ERR {r['commit_id']} ({error})") else: status = "OK " if valid else "BAD" cid = short_id(r["commit_id"]) signer = r["signer"] or "(unsigned)" key = r["key_id"] or "" key_part = f" key={key}" if key else "" print(f"{status} {cid} signer={signer}{key_part}") if any_failure: raise SystemExit(ExitCode.USER_ERROR)