gabriel / muse public
test_cmd_verify_commit.py python
454 lines 16.8 KB
Raw
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ breaking 150 days ago
1 """Tests for ``muse verify-commit`` — verify Ed25519 signatures on commits.
2
3 Coverage tiers:
4 - Unit: _verify_one (valid sig, tampered commit_id, missing sig, missing public key,
5 bit-flip in signature, unsigned commit with --strict)
6 - Integration: commit --sign → verify-commit succeeds; tamper stored record → verify fails;
7 batch verify (multiple commit IDs); --json schema; text output format;
8 unsigned commit exits nonzero with --strict; unsigned commit exits 0 without;
9 nonexistent commit_id; --check-key-status returns unknown when no hub
10 - Security: bit-flip in signature; canonical message tamper; ANSI in commit ref rejected
11 - Stress: 100 signed commits verified correctly; key_status_cache deduplicates calls
12 """
13
14 from __future__ import annotations
15
16 import base64
17 import datetime
18 import hashlib
19 import json
20 import pathlib
21 from unittest.mock import patch
22
23 import pytest
24
25 from tests.cli_test_helper import CliRunner
26 from muse.core.object_store import write_object
27 from muse.core.provenance import (
28 encode_public_key,
29 provenance_payload,
30 sign_commit_record,
31 verify_commit_ed25519,
32 )
33 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
34 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
35 from muse.core._types import Manifest
36
37 runner = CliRunner()
38
39 _REPO_ID = "verify-commit-test"
40 _counter = 0
41
42
43 # ---------------------------------------------------------------------------
44 # Helpers
45 # ---------------------------------------------------------------------------
46
47
48 def _sha(data: bytes) -> str:
49 return hashlib.sha256(data).hexdigest()
50
51
52 def _init_repo(path: pathlib.Path) -> pathlib.Path:
53 muse = path / ".muse"
54 for d in ("commits", "snapshots", "objects", "refs/heads", "code"):
55 (muse / d).mkdir(parents=True, exist_ok=True)
56 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
57 (muse / "repo.json").write_text(
58 json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8"
59 )
60 return path
61
62
63 def _env(repo: pathlib.Path) -> dict[str, str]:
64 return {"MUSE_REPO_ROOT": str(repo)}
65
66
67 def _make_key():
68 """Generate a fresh Ed25519 private key."""
69 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
70 return Ed25519PrivateKey.generate()
71
72
73 def _commit_files(
74 root: pathlib.Path,
75 files: dict[str, bytes],
76 branch: str = "main",
77 message: str | None = None,
78 sign: bool = False,
79 private_key=None,
80 agent_id: str = "test-agent",
81 ) -> tuple[str, CommitRecord]:
82 """Create a commit; optionally sign it. Returns (commit_id, CommitRecord)."""
83 global _counter
84 _counter += 1
85 manifest: Manifest = {}
86 for rel_path, content in files.items():
87 obj_id = _sha(content)
88 write_object(root, obj_id, content)
89 manifest[rel_path] = obj_id
90 abs_path = root / rel_path
91 abs_path.parent.mkdir(parents=True, exist_ok=True)
92 abs_path.write_bytes(content)
93 snap_id = compute_snapshot_id(manifest)
94 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
95 committed_at = datetime.datetime.now(datetime.timezone.utc)
96 ref_path = root / ".muse" / "refs" / "heads" / branch
97 parent_id = ref_path.read_text(encoding="utf-8").strip() if ref_path.exists() else None
98 parents = [parent_id] if parent_id else []
99 msg = message or f"commit {_counter}"
100 commit_id = compute_commit_id(parents, snap_id, msg, committed_at.isoformat())
101
102 sig = ""
103 pub_b64 = ""
104 key_id = ""
105 if sign and private_key is not None:
106 result = sign_commit_record(
107 commit_id,
108 agent_id,
109 private_key,
110 committed_at=committed_at.isoformat(),
111 )
112 if result:
113 sig, pub_b64, key_id = result
114
115 record = CommitRecord(
116 commit_id=commit_id,
117 repo_id=_REPO_ID,
118 branch=branch,
119 snapshot_id=snap_id,
120 message=msg,
121 committed_at=committed_at,
122 parent_commit_id=parent_id,
123 agent_id=agent_id if sign else "",
124 signature=sig,
125 signer_public_key=pub_b64,
126 signer_key_id=key_id,
127 )
128 write_commit(root, record)
129 ref_path.write_text(commit_id, encoding="utf-8")
130 return commit_id, record
131
132
133 def _invoke(repo: pathlib.Path, *args: str):
134 from muse.cli.app import main as cli
135 return runner.invoke(cli, ["verify-commit", *args], env=_env(repo))
136
137
138 def _force_write_commit(root: pathlib.Path, record: CommitRecord) -> None:
139 """Overwrite a commit file unconditionally (bypasses write_commit idempotency)."""
140 import msgpack
141 commit_file = root / ".muse" / "commits" / f"{record.commit_id}.msgpack"
142 commit_file.write_bytes(msgpack.packb(record.to_dict(), use_bin_type=True))
143
144
145 # ---------------------------------------------------------------------------
146 # Unit — _verify_one
147 # ---------------------------------------------------------------------------
148
149
150 def test_verify_one_valid_signature(tmp_path: pathlib.Path) -> None:
151 from muse.cli.commands.verify_commit import _verify_one
152 root = _init_repo(tmp_path)
153 key = _make_key()
154 commit_id, _ = _commit_files(root, {"a.py": b"x = 1\n"}, sign=True, private_key=key)
155 result = _verify_one(root, commit_id)
156 assert result["valid"] is True
157 assert result["commit_id"] == commit_id
158 assert len(result["key_id"]) > 0
159
160
161 def test_verify_one_unsigned_commit(tmp_path: pathlib.Path) -> None:
162 from muse.cli.commands.verify_commit import _verify_one
163 root = _init_repo(tmp_path)
164 commit_id, _ = _commit_files(root, {"a.py": b"x = 1\n"}, sign=False)
165 result = _verify_one(root, commit_id)
166 assert result["valid"] is False
167 assert result["signer"] == ""
168
169
170 def test_verify_one_tampered_commit_id(tmp_path: pathlib.Path) -> None:
171 """Querying a non-existent commit_id returns valid=False."""
172 from muse.cli.commands.verify_commit import _verify_one
173 root = _init_repo(tmp_path)
174 fake_id = "a" * 64
175 result = _verify_one(root, fake_id)
176 assert result["valid"] is False
177
178
179 def test_verify_one_missing_public_key(tmp_path: pathlib.Path) -> None:
180 """A commit with a signature but no public key returns valid=False."""
181 from muse.cli.commands.verify_commit import _verify_one
182 root = _init_repo(tmp_path)
183 key = _make_key()
184 commit_id, record = _commit_files(root, {"a.py": b"x = 1\n"}, sign=True, private_key=key)
185
186 # Force-overwrite the commit with signer_public_key stripped
187 tampered = CommitRecord(
188 commit_id=record.commit_id,
189 repo_id=record.repo_id,
190 branch=record.branch,
191 snapshot_id=record.snapshot_id,
192 message=record.message,
193 committed_at=record.committed_at,
194 parent_commit_id=record.parent_commit_id,
195 agent_id=record.agent_id,
196 signature=record.signature,
197 signer_public_key="", # stripped
198 signer_key_id=record.signer_key_id,
199 )
200 _force_write_commit(root, tampered)
201 result = _verify_one(root, commit_id)
202 assert result["valid"] is False
203
204
205 def test_verify_one_bit_flip_in_signature(tmp_path: pathlib.Path) -> None:
206 """A single bit flip in the stored signature must invalidate it."""
207 from muse.cli.commands.verify_commit import _verify_one
208 root = _init_repo(tmp_path)
209 key = _make_key()
210 commit_id, record = _commit_files(root, {"a.py": b"x = 1\n"}, sign=True, private_key=key)
211
212 # Flip one byte in the base64 signature
213 sig_bytes = base64.urlsafe_b64decode(record.signature + "==")
214 flipped = bytes([sig_bytes[0] ^ 0xFF]) + sig_bytes[1:]
215 bad_sig = base64.urlsafe_b64encode(flipped).rstrip(b"=").decode()
216
217 tampered = CommitRecord(
218 commit_id=record.commit_id,
219 repo_id=record.repo_id,
220 branch=record.branch,
221 snapshot_id=record.snapshot_id,
222 message=record.message,
223 committed_at=record.committed_at,
224 parent_commit_id=record.parent_commit_id,
225 agent_id=record.agent_id,
226 signature=bad_sig,
227 signer_public_key=record.signer_public_key,
228 signer_key_id=record.signer_key_id,
229 )
230 _force_write_commit(root, tampered)
231 result = _verify_one(root, commit_id)
232 assert result["valid"] is False
233
234
235 def test_verify_one_key_status_unknown_without_hub(tmp_path: pathlib.Path) -> None:
236 """Without a hub configured, key_status must be 'unknown'."""
237 from muse.cli.commands.verify_commit import _verify_one
238 root = _init_repo(tmp_path)
239 key = _make_key()
240 commit_id, _ = _commit_files(root, {"a.py": b"x = 1\n"}, sign=True, private_key=key)
241 result = _verify_one(root, commit_id, check_key_status=True, hub_url=None)
242 assert result["key_status"] == "unknown"
243
244
245 def test_verify_one_json_schema_keys(tmp_path: pathlib.Path) -> None:
246 from muse.cli.commands.verify_commit import _verify_one
247 root = _init_repo(tmp_path)
248 key = _make_key()
249 commit_id, _ = _commit_files(root, {"a.py": b"x = 1\n"}, sign=True, private_key=key)
250 result = _verify_one(root, commit_id)
251 assert "commit_id" in result
252 assert "valid" in result
253 assert "signer" in result
254 assert "key_id" in result
255 assert "signed_at" in result
256 assert "key_status" in result
257
258
259 # ---------------------------------------------------------------------------
260 # Integration — CLI
261 # ---------------------------------------------------------------------------
262
263
264 def test_verify_commit_valid_exits_zero(tmp_path: pathlib.Path) -> None:
265 root = _init_repo(tmp_path)
266 key = _make_key()
267 commit_id, _ = _commit_files(root, {"a.py": b"x = 1\n"}, sign=True, private_key=key)
268 result = _invoke(root, commit_id)
269 assert result.exit_code == 0
270
271
272 def test_verify_commit_json_output(tmp_path: pathlib.Path) -> None:
273 root = _init_repo(tmp_path)
274 key = _make_key()
275 commit_id, _ = _commit_files(root, {"a.py": b"x = 1\n"}, sign=True, private_key=key)
276 result = _invoke(root, commit_id, "--json")
277 assert result.exit_code == 0
278 data = json.loads(result.stdout)
279 assert data["commit_id"] == commit_id
280 assert data["valid"] is True
281 assert "signer" in data
282 assert "key_id" in data
283 assert "key_status" in data
284
285
286 def test_verify_commit_invalid_exits_nonzero(tmp_path: pathlib.Path) -> None:
287 root = _init_repo(tmp_path)
288 key = _make_key()
289 commit_id, record = _commit_files(root, {"a.py": b"x = 1\n"}, sign=True, private_key=key)
290
291 sig_bytes = base64.urlsafe_b64decode(record.signature + "==")
292 flipped = bytes([sig_bytes[0] ^ 0xFF]) + sig_bytes[1:]
293 bad_sig = base64.urlsafe_b64encode(flipped).rstrip(b"=").decode()
294 tampered = CommitRecord(
295 commit_id=record.commit_id, repo_id=record.repo_id, branch=record.branch,
296 snapshot_id=record.snapshot_id, message=record.message,
297 committed_at=record.committed_at, parent_commit_id=record.parent_commit_id,
298 agent_id=record.agent_id, signature=bad_sig,
299 signer_public_key=record.signer_public_key, signer_key_id=record.signer_key_id,
300 )
301 _force_write_commit(root, tampered)
302 result = _invoke(root, commit_id, "--json")
303 assert result.exit_code != 0
304 data = json.loads(result.stdout)
305 assert data["valid"] is False
306
307
308 def test_verify_commit_unsigned_no_strict_exits_zero(tmp_path: pathlib.Path) -> None:
309 """Unsigned commit without --strict: exits 0 but valid=False in output."""
310 root = _init_repo(tmp_path)
311 commit_id, _ = _commit_files(root, {"a.py": b"x = 1\n"}, sign=False)
312 result = _invoke(root, commit_id, "--json")
313 assert result.exit_code == 0
314 data = json.loads(result.stdout)
315 assert data["valid"] is False
316
317
318 def test_verify_commit_unsigned_strict_exits_nonzero(tmp_path: pathlib.Path) -> None:
319 root = _init_repo(tmp_path)
320 commit_id, _ = _commit_files(root, {"a.py": b"x = 1\n"}, sign=False)
321 result = _invoke(root, commit_id, "--strict")
322 assert result.exit_code != 0
323
324
325 def test_verify_commit_head_shorthand(tmp_path: pathlib.Path) -> None:
326 root = _init_repo(tmp_path)
327 key = _make_key()
328 _commit_files(root, {"a.py": b"x = 1\n"}, sign=True, private_key=key)
329 result = _invoke(root, "HEAD", "--json")
330 assert result.exit_code == 0
331 data = json.loads(result.stdout)
332 assert data["valid"] is True
333
334
335 def test_verify_commit_nonexistent_ref_exits_nonzero(tmp_path: pathlib.Path) -> None:
336 root = _init_repo(tmp_path)
337 result = _invoke(root, "a" * 64)
338 assert result.exit_code != 0
339
340
341 def test_verify_commit_text_output_format(tmp_path: pathlib.Path) -> None:
342 """Text output: '<status> <short_commit_id> <signer>'"""
343 root = _init_repo(tmp_path)
344 key = _make_key()
345 commit_id, _ = _commit_files(root, {"a.py": b"x = 1\n"}, sign=True, private_key=key)
346 result = _invoke(root, commit_id)
347 assert result.exit_code == 0
348 assert commit_id[:8] in result.stdout
349
350
351 def test_verify_commit_batch_all_valid(tmp_path: pathlib.Path) -> None:
352 root = _init_repo(tmp_path)
353 key = _make_key()
354 ids = []
355 for i in range(3):
356 cid, _ = _commit_files(root, {"a.py": f"x = {i}\n".encode()}, sign=True, private_key=key)
357 ids.append(cid)
358 result = _invoke(root, *ids, "--json")
359 # Batch: stdout is newline-separated JSON objects or a JSON array
360 assert result.exit_code == 0
361
362
363 def test_verify_commit_batch_one_invalid_exits_nonzero(tmp_path: pathlib.Path) -> None:
364 root = _init_repo(tmp_path)
365 key = _make_key()
366 cid1, _ = _commit_files(root, {"a.py": b"x = 1\n"}, sign=True, private_key=key)
367 cid2, _ = _commit_files(root, {"a.py": b"x = 2\n"}, sign=False)
368 result = _invoke(root, cid1, cid2, "--strict")
369 assert result.exit_code != 0
370
371
372 def test_verify_commit_check_key_status_unknown_no_hub(tmp_path: pathlib.Path) -> None:
373 root = _init_repo(tmp_path)
374 key = _make_key()
375 commit_id, _ = _commit_files(root, {"a.py": b"x = 1\n"}, sign=True, private_key=key)
376 result = _invoke(root, commit_id, "--check-key-status", "--json")
377 assert result.exit_code == 0
378 data = json.loads(result.stdout)
379 assert data["key_status"] == "unknown"
380
381
382 # ---------------------------------------------------------------------------
383 # Security
384 # ---------------------------------------------------------------------------
385
386
387 def test_verify_commit_ansi_in_ref_rejected(tmp_path: pathlib.Path) -> None:
388 root = _init_repo(tmp_path)
389 _commit_files(root, {"a.py": b"x = 1\n"})
390 result = _invoke(root, "\x1b[31mbad\x1b[0m")
391 assert result.exit_code != 0
392
393
394 def test_verify_commit_canonical_message_tamper_detected(tmp_path: pathlib.Path) -> None:
395 """Changing agent_id in the stored record must invalidate the signature."""
396 from muse.cli.commands.verify_commit import _verify_one
397 root = _init_repo(tmp_path)
398 key = _make_key()
399 commit_id, record = _commit_files(
400 root, {"a.py": b"x = 1\n"}, sign=True, private_key=key, agent_id="agent-A"
401 )
402 # Overwrite with a different agent_id — canonical message will differ
403 tampered = CommitRecord(
404 commit_id=record.commit_id, repo_id=record.repo_id, branch=record.branch,
405 snapshot_id=record.snapshot_id, message=record.message,
406 committed_at=record.committed_at, parent_commit_id=record.parent_commit_id,
407 agent_id="agent-B", # tampered
408 signature=record.signature,
409 signer_public_key=record.signer_public_key,
410 signer_key_id=record.signer_key_id,
411 )
412 _force_write_commit(root, tampered)
413 result = _verify_one(root, commit_id)
414 assert result["valid"] is False
415
416
417 # ---------------------------------------------------------------------------
418 # Stress — 100 signed commits
419 # ---------------------------------------------------------------------------
420
421
422 def test_verify_commit_100_signed_commits(tmp_path: pathlib.Path) -> None:
423 """100 signed commits all verify correctly."""
424 from muse.cli.commands.verify_commit import _verify_one
425 root = _init_repo(tmp_path)
426 key = _make_key()
427 for i in range(100):
428 commit_id, _ = _commit_files(
429 root, {"f.py": f"v = {i}\n".encode()}, sign=True, private_key=key
430 )
431 result = _verify_one(root, commit_id)
432 assert result["valid"] is True, f"commit {i} failed"
433
434
435 def test_verify_commit_key_status_cache_deduplicates(tmp_path: pathlib.Path) -> None:
436 """key_status_cache is populated on first lookup and reused on subsequent ones."""
437 from muse.cli.commands.verify_commit import _verify_one
438 root = _init_repo(tmp_path)
439 key = _make_key()
440
441 call_count = 0
442
443 def mock_check(hub_url, key_id):
444 nonlocal call_count
445 call_count += 1
446 return "active"
447
448 cache: dict[str, str] = {}
449 with patch("muse.cli.commands.verify_commit._fetch_key_status", side_effect=mock_check):
450 commit_id, _ = _commit_files(root, {"a.py": b"x = 1\n"}, sign=True, private_key=key)
451 _verify_one(root, commit_id, check_key_status=True, hub_url="http://fake", key_status_cache=cache)
452 _verify_one(root, commit_id, check_key_status=True, hub_url="http://fake", key_status_cache=cache)
453
454 assert call_count == 1, "cache must prevent duplicate key status lookups"
File History 1 commit
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 150 days ago