"""``muse verify-commit`` — single-commit Ed25519 + attestation verification. Companion to ``muse verify`` (whole-repo walk). Verifies the four-tier provenance chain for one commit: * Tier 0: unsigned (no Ed25519 signature) * Tier 1: signed only (Ed25519 OK, no attestation record) * Tier 2: signed + HMAC-attested (AIR record present and HMAC valid) * Tier 3: signed + HMAC-attested + ICP-anchored (canister returned matching record) Partial-verification handling is mandatory per the Phase 7.5 contract: when the verifier cannot complete the chain (e.g. ICP unreachable), the CLI surfaces ``ICP_UNREACHABLE — cannot confirm anchor`` rather than silently degrading to "signed only". Usage:: muse verify-commit # text output muse verify-commit --attest # include HMAC + ICP tiers muse verify-commit --attest --json # machine-readable JSON output schema:: { "commit_id": "", "tier": 0|1|2|3, "tier_label": "unsigned|signed|hmac-attested|icp-anchored", "valid": , "partial": , "reason": "", "hmac_valid": , "icp_anchored": , "anchor_tx_id": "", "identity_depth": , "provider_id": "" } Exit codes:: 0 — verified at the requested tier 1 — verification failed (provably bad — invalid sig, tampered HMAC) 2 — repo not found / commit not found 4 — partial verification (ICP unreachable etc.) — verifier could not complete """ from __future__ import annotations import argparse import json import sys from typing import Any from muse.core.errors import ExitCode from muse.core.repo import require_repo from muse.core.store import read_commit from muse.core.validation import sanitize_display from muse.core.verify import ( AttestationVerifyResult, TIER_HMAC_ATTESTED, TIER_ICP_ANCHORED, TIER_SIGNED_ONLY, TIER_UNSIGNED, verify_attestation, ) # Tier label map — exposed as a module constant so the JSON-mode label is # stable across releases. TIER_LABELS: dict[int, str] = { TIER_UNSIGNED: "unsigned", TIER_SIGNED_ONLY: "signed", TIER_HMAC_ATTESTED: "hmac-attested", TIER_ICP_ANCHORED: "icp-anchored", } def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse verify-commit`` subcommand.""" parser = subparsers.add_parser( "verify-commit", help="Verify a single commit's Ed25519 + attestation chain.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "commit_id", metavar="COMMIT_ID", help="Full or short commit ID to verify.", ) parser.add_argument( "--attest", action="store_true", help=( "Include HMAC + ICP attestation tiers in the verification " "(Phase 7.5). Without --attest, only Ed25519 signature is " "checked (tier 0/1)." ), ) parser.add_argument( "--no-icp", action="store_true", dest="no_icp", help=( "Skip the ICP canister query. Useful in offline environments " "or when the canister is known to be unreachable. Result is " "capped at tier 2 (HMAC-attested)." ), ) parser.add_argument( "--icp-canister", default=None, dest="icp_canister", help="Override the default ICP canister ID for the anchor check.", ) parser.add_argument( "--icp-timeout", type=float, default=5.0, dest="icp_timeout", help="ICP HTTP query timeout in seconds (default: 5.0).", ) parser.add_argument( "--verify-identity", action="store_true", dest="verify_identity", help=( "Also verify the org-quorum signatures stored in " "commit.metadata['quorum'] against the live identity records " "(Phase 7.7). When set, the result includes ``quorum_valid`` " "and ``quorum_reason`` and the exit code is non-zero if the " "quorum is below threshold." ), ) parser.add_argument( "--identity-domain", default="knowtation", dest="identity_domain", help=( "Domain for the identity-provider lookup used by " "--verify-identity (default: knowtation)." ), ) parser.add_argument( "--json", action="store_true", dest="json_out", help="Emit a machine-readable JSON report on stdout.", ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Verify the commit and emit a four-tier report.""" commit_id: str = args.commit_id attest: bool = args.attest no_icp: bool = args.no_icp icp_canister: str | None = args.icp_canister icp_timeout: float = args.icp_timeout json_out: bool = args.json_out verify_identity: bool = bool(getattr(args, "verify_identity", False)) identity_domain: str = getattr(args, "identity_domain", "knowtation") or "knowtation" root = require_repo() commit = read_commit(root, commit_id) if commit is None: print( f"❌ Commit {sanitize_display(commit_id)!r} not found.", file=sys.stderr, ) raise SystemExit(ExitCode.NOT_FOUND) cfg: dict[str, Any] = {"check_icp": not no_icp} if icp_canister: cfg["icp_canister_id"] = icp_canister cfg["icp_timeout"] = icp_timeout if attest: result = verify_attestation(commit, root, cfg) else: # Lightweight path — only check the Ed25519 signature. from muse.core.verify import _check_ed25519_signature sig_ok, sig_reason = _check_ed25519_signature(commit) result = AttestationVerifyResult( valid=sig_ok, reason=sig_reason if not sig_ok else "", tier=TIER_SIGNED_ONLY if sig_ok else TIER_UNSIGNED, hmac_valid=False, icp_anchored=False, anchor_tx_id="", identity_depth=0, provider_id="", partial=False, ) quorum_payload: dict[str, Any] | None = None if verify_identity: from muse.core.verify import verify_quorum_signatures qresult = verify_quorum_signatures( commit, root, {"identity_domain": identity_domain} ) quorum_payload = { "valid": bool(qresult.get("valid", False)), "partial": bool(qresult.get("partial", False)), "reason": qresult.get("reason", ""), "threshold": int(qresult.get("threshold", 0)), "valid_signers": int(qresult.get("valid_signers", 0)), "total_signers": int(qresult.get("total_signers", 0)), "depth": int(qresult.get("depth", 0)), "org_handle": qresult.get("org_handle", ""), } if json_out: payload: dict[str, Any] = { "commit_id": commit.commit_id, "tier": result["tier"], "tier_label": TIER_LABELS.get(result["tier"], "unknown"), "valid": bool(result.get("valid", False)), "partial": bool(result.get("partial", False)), "reason": result.get("reason", ""), "hmac_valid": bool(result.get("hmac_valid", False)), "icp_anchored": bool(result.get("icp_anchored", False)), "anchor_tx_id": result.get("anchor_tx_id", ""), "identity_depth": int(result.get("identity_depth", 0)), "provider_id": result.get("provider_id", ""), } if quorum_payload is not None: payload["quorum"] = quorum_payload print(json.dumps(payload, indent=2, sort_keys=True)) else: _print_text(commit.commit_id, result) if quorum_payload is not None: _print_quorum(quorum_payload) overall_partial = bool(result.get("partial", False)) overall_valid = bool(result.get("valid", False)) if quorum_payload is not None: if quorum_payload["partial"]: overall_partial = True if not quorum_payload["valid"]: overall_valid = False if overall_partial: raise SystemExit(ExitCode.NOT_FOUND) if result["tier"] == TIER_UNSIGNED or not overall_valid: raise SystemExit(ExitCode.USER_ERROR) raise SystemExit(ExitCode.SUCCESS) def _print_quorum(qp: dict[str, Any]) -> None: """Render a one-block quorum verification summary.""" org = sanitize_display(str(qp.get("org_handle", ""))) threshold = int(qp.get("threshold", 0)) valid_signers = int(qp.get("valid_signers", 0)) total = int(qp.get("total_signers", 0)) depth = int(qp.get("depth", 0)) print(f"Quorum org: {org}") print(f"Quorum signers: {valid_signers}/{total} valid (threshold {threshold})") if depth: print(f"Identity depth: {depth}") if qp.get("partial"): print(f"Quorum status: ⚠️ PARTIAL — {qp.get('reason')}") elif not qp.get("valid"): print(f"Quorum status: ❌ INVALID — {qp.get('reason')}") else: print("Quorum status: ✅ verified") def _print_text(commit_id: str, result: AttestationVerifyResult) -> None: """Render a human-readable four-tier report.""" tier = result["tier"] label = TIER_LABELS.get(tier, "unknown") reason = result.get("reason", "") partial = result.get("partial", False) # Glyphs match the user's spec verbatim: # ✗ unsigned # ✓ signed only # ✓✓ signed + HMAC-attested # ✓✓✓ signed + HMAC-attested + ICP-anchored if tier == TIER_UNSIGNED: glyph = "✗" elif tier == TIER_SIGNED_ONLY: glyph = "✓" elif tier == TIER_HMAC_ATTESTED: glyph = "✓✓" elif tier == TIER_ICP_ANCHORED: glyph = "✓✓✓" else: glyph = "?" print(f"Commit: {commit_id}") print(f"Tier {tier}: {glyph} {label}") if partial: print(f"Status: ⚠️ PARTIAL — {reason}") elif tier == TIER_UNSIGNED: print(f"Status: ❌ {reason or 'unsigned'}") elif not result.get("valid", False): print(f"Status: ❌ INVALID — {reason}") else: print("Status: ✅ verified") if result.get("provider_id"): print(f"Provider: {sanitize_display(result['provider_id'])}") if result.get("hmac_valid"): print("HMAC: ✅ valid") if result.get("icp_anchored"): anchor_id = result.get("anchor_tx_id", "") print(f"ICP anchor: ✅ {anchor_id}") if int(result.get("identity_depth", 0)) > 0: print(f"Identity depth: {result['identity_depth']}") __all__ = ["TIER_LABELS", "register", "run"]