gabriel / muse public
verify_commit.py python
297 lines 10.7 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """``muse verify-commit`` — single-commit Ed25519 + attestation verification.
2
3 Companion to ``muse verify`` (whole-repo walk). Verifies the four-tier
4 provenance chain for one commit:
5
6 * Tier 0: unsigned (no Ed25519 signature)
7 * Tier 1: signed only (Ed25519 OK, no attestation record)
8 * Tier 2: signed + HMAC-attested (AIR record present and HMAC valid)
9 * Tier 3: signed + HMAC-attested + ICP-anchored (canister returned matching record)
10
11 Partial-verification handling is mandatory per the Phase 7.5 contract:
12 when the verifier cannot complete the chain (e.g. ICP unreachable), the CLI
13 surfaces ``ICP_UNREACHABLE — cannot confirm anchor`` rather than silently
14 degrading to "signed only".
15
16 Usage::
17
18 muse verify-commit <commit_id> # text output
19 muse verify-commit <commit_id> --attest # include HMAC + ICP tiers
20 muse verify-commit <commit_id> --attest --json # machine-readable
21
22 JSON output schema::
23
24 {
25 "commit_id": "<sha256>",
26 "tier": 0|1|2|3,
27 "tier_label": "unsigned|signed|hmac-attested|icp-anchored",
28 "valid": <bool>,
29 "partial": <bool>,
30 "reason": "<str>",
31 "hmac_valid": <bool>,
32 "icp_anchored": <bool>,
33 "anchor_tx_id": "<str>",
34 "identity_depth": <int>,
35 "provider_id": "<str>"
36 }
37
38 Exit codes::
39
40 0 — verified at the requested tier
41 1 — verification failed (provably bad — invalid sig, tampered HMAC)
42 2 — repo not found / commit not found
43 4 — partial verification (ICP unreachable etc.) — verifier could not complete
44 """
45
46 from __future__ import annotations
47
48 import argparse
49 import json
50 import sys
51 from typing import Any
52
53 from muse.core.errors import ExitCode
54 from muse.core.repo import require_repo
55 from muse.core.store import read_commit
56 from muse.core.validation import sanitize_display
57 from muse.core.verify import (
58 AttestationVerifyResult,
59 TIER_HMAC_ATTESTED,
60 TIER_ICP_ANCHORED,
61 TIER_SIGNED_ONLY,
62 TIER_UNSIGNED,
63 verify_attestation,
64 )
65
66
67 # Tier label map — exposed as a module constant so the JSON-mode label is
68 # stable across releases.
69 TIER_LABELS: dict[int, str] = {
70 TIER_UNSIGNED: "unsigned",
71 TIER_SIGNED_ONLY: "signed",
72 TIER_HMAC_ATTESTED: "hmac-attested",
73 TIER_ICP_ANCHORED: "icp-anchored",
74 }
75
76
77 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
78 """Register the ``muse verify-commit`` subcommand."""
79 parser = subparsers.add_parser(
80 "verify-commit",
81 help="Verify a single commit's Ed25519 + attestation chain.",
82 description=__doc__,
83 formatter_class=argparse.RawDescriptionHelpFormatter,
84 )
85 parser.add_argument(
86 "commit_id", metavar="COMMIT_ID",
87 help="Full or short commit ID to verify.",
88 )
89 parser.add_argument(
90 "--attest", action="store_true",
91 help=(
92 "Include HMAC + ICP attestation tiers in the verification "
93 "(Phase 7.5). Without --attest, only Ed25519 signature is "
94 "checked (tier 0/1)."
95 ),
96 )
97 parser.add_argument(
98 "--no-icp", action="store_true", dest="no_icp",
99 help=(
100 "Skip the ICP canister query. Useful in offline environments "
101 "or when the canister is known to be unreachable. Result is "
102 "capped at tier 2 (HMAC-attested)."
103 ),
104 )
105 parser.add_argument(
106 "--icp-canister", default=None, dest="icp_canister",
107 help="Override the default ICP canister ID for the anchor check.",
108 )
109 parser.add_argument(
110 "--icp-timeout", type=float, default=5.0, dest="icp_timeout",
111 help="ICP HTTP query timeout in seconds (default: 5.0).",
112 )
113 parser.add_argument(
114 "--verify-identity", action="store_true", dest="verify_identity",
115 help=(
116 "Also verify the org-quorum signatures stored in "
117 "commit.metadata['quorum'] against the live identity records "
118 "(Phase 7.7). When set, the result includes ``quorum_valid`` "
119 "and ``quorum_reason`` and the exit code is non-zero if the "
120 "quorum is below threshold."
121 ),
122 )
123 parser.add_argument(
124 "--identity-domain", default="knowtation", dest="identity_domain",
125 help=(
126 "Domain for the identity-provider lookup used by "
127 "--verify-identity (default: knowtation)."
128 ),
129 )
130 parser.add_argument(
131 "--json", action="store_true", dest="json_out",
132 help="Emit a machine-readable JSON report on stdout.",
133 )
134 parser.set_defaults(func=run)
135
136
137 def run(args: argparse.Namespace) -> None:
138 """Verify the commit and emit a four-tier report."""
139 commit_id: str = args.commit_id
140 attest: bool = args.attest
141 no_icp: bool = args.no_icp
142 icp_canister: str | None = args.icp_canister
143 icp_timeout: float = args.icp_timeout
144 json_out: bool = args.json_out
145 verify_identity: bool = bool(getattr(args, "verify_identity", False))
146 identity_domain: str = getattr(args, "identity_domain", "knowtation") or "knowtation"
147
148 root = require_repo()
149 commit = read_commit(root, commit_id)
150 if commit is None:
151 print(
152 f"❌ Commit {sanitize_display(commit_id)!r} not found.",
153 file=sys.stderr,
154 )
155 raise SystemExit(ExitCode.NOT_FOUND)
156
157 cfg: dict[str, Any] = {"check_icp": not no_icp}
158 if icp_canister:
159 cfg["icp_canister_id"] = icp_canister
160 cfg["icp_timeout"] = icp_timeout
161
162 if attest:
163 result = verify_attestation(commit, root, cfg)
164 else:
165 # Lightweight path — only check the Ed25519 signature.
166 from muse.core.verify import _check_ed25519_signature
167 sig_ok, sig_reason = _check_ed25519_signature(commit)
168 result = AttestationVerifyResult(
169 valid=sig_ok,
170 reason=sig_reason if not sig_ok else "",
171 tier=TIER_SIGNED_ONLY if sig_ok else TIER_UNSIGNED,
172 hmac_valid=False,
173 icp_anchored=False,
174 anchor_tx_id="",
175 identity_depth=0,
176 provider_id="",
177 partial=False,
178 )
179
180 quorum_payload: dict[str, Any] | None = None
181 if verify_identity:
182 from muse.core.verify import verify_quorum_signatures
183 qresult = verify_quorum_signatures(
184 commit, root, {"identity_domain": identity_domain}
185 )
186 quorum_payload = {
187 "valid": bool(qresult.get("valid", False)),
188 "partial": bool(qresult.get("partial", False)),
189 "reason": qresult.get("reason", ""),
190 "threshold": int(qresult.get("threshold", 0)),
191 "valid_signers": int(qresult.get("valid_signers", 0)),
192 "total_signers": int(qresult.get("total_signers", 0)),
193 "depth": int(qresult.get("depth", 0)),
194 "org_handle": qresult.get("org_handle", ""),
195 }
196
197 if json_out:
198 payload: dict[str, Any] = {
199 "commit_id": commit.commit_id,
200 "tier": result["tier"],
201 "tier_label": TIER_LABELS.get(result["tier"], "unknown"),
202 "valid": bool(result.get("valid", False)),
203 "partial": bool(result.get("partial", False)),
204 "reason": result.get("reason", ""),
205 "hmac_valid": bool(result.get("hmac_valid", False)),
206 "icp_anchored": bool(result.get("icp_anchored", False)),
207 "anchor_tx_id": result.get("anchor_tx_id", ""),
208 "identity_depth": int(result.get("identity_depth", 0)),
209 "provider_id": result.get("provider_id", ""),
210 }
211 if quorum_payload is not None:
212 payload["quorum"] = quorum_payload
213 print(json.dumps(payload, indent=2, sort_keys=True))
214 else:
215 _print_text(commit.commit_id, result)
216 if quorum_payload is not None:
217 _print_quorum(quorum_payload)
218
219 overall_partial = bool(result.get("partial", False))
220 overall_valid = bool(result.get("valid", False))
221 if quorum_payload is not None:
222 if quorum_payload["partial"]:
223 overall_partial = True
224 if not quorum_payload["valid"]:
225 overall_valid = False
226
227 if overall_partial:
228 raise SystemExit(ExitCode.NOT_FOUND)
229 if result["tier"] == TIER_UNSIGNED or not overall_valid:
230 raise SystemExit(ExitCode.USER_ERROR)
231 raise SystemExit(ExitCode.SUCCESS)
232
233
234 def _print_quorum(qp: dict[str, Any]) -> None:
235 """Render a one-block quorum verification summary."""
236 org = sanitize_display(str(qp.get("org_handle", "")))
237 threshold = int(qp.get("threshold", 0))
238 valid_signers = int(qp.get("valid_signers", 0))
239 total = int(qp.get("total_signers", 0))
240 depth = int(qp.get("depth", 0))
241 print(f"Quorum org: {org}")
242 print(f"Quorum signers: {valid_signers}/{total} valid (threshold {threshold})")
243 if depth:
244 print(f"Identity depth: {depth}")
245 if qp.get("partial"):
246 print(f"Quorum status: ⚠️ PARTIAL — {qp.get('reason')}")
247 elif not qp.get("valid"):
248 print(f"Quorum status: ❌ INVALID — {qp.get('reason')}")
249 else:
250 print("Quorum status: ✅ verified")
251
252
253 def _print_text(commit_id: str, result: AttestationVerifyResult) -> None:
254 """Render a human-readable four-tier report."""
255 tier = result["tier"]
256 label = TIER_LABELS.get(tier, "unknown")
257 reason = result.get("reason", "")
258 partial = result.get("partial", False)
259
260 # Glyphs match the user's spec verbatim:
261 # ✗ unsigned
262 # ✓ signed only
263 # ✓✓ signed + HMAC-attested
264 # ✓✓✓ signed + HMAC-attested + ICP-anchored
265 if tier == TIER_UNSIGNED:
266 glyph = "✗"
267 elif tier == TIER_SIGNED_ONLY:
268 glyph = "✓"
269 elif tier == TIER_HMAC_ATTESTED:
270 glyph = "✓✓"
271 elif tier == TIER_ICP_ANCHORED:
272 glyph = "✓✓✓"
273 else:
274 glyph = "?"
275
276 print(f"Commit: {commit_id}")
277 print(f"Tier {tier}: {glyph} {label}")
278 if partial:
279 print(f"Status: ⚠️ PARTIAL — {reason}")
280 elif tier == TIER_UNSIGNED:
281 print(f"Status: ❌ {reason or 'unsigned'}")
282 elif not result.get("valid", False):
283 print(f"Status: ❌ INVALID — {reason}")
284 else:
285 print("Status: ✅ verified")
286 if result.get("provider_id"):
287 print(f"Provider: {sanitize_display(result['provider_id'])}")
288 if result.get("hmac_valid"):
289 print("HMAC: ✅ valid")
290 if result.get("icp_anchored"):
291 anchor_id = result.get("anchor_tx_id", "")
292 print(f"ICP anchor: ✅ {anchor_id}")
293 if int(result.get("identity_depth", 0)) > 0:
294 print(f"Identity depth: {result['identity_depth']}")
295
296
297 __all__ = ["TIER_LABELS", "register", "run"]
File History 4 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e Merge branch 'dev' into main Human 29 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce Merge branch 'dev' into main Human 48 days ago