gabriel / muse public
test_verify_extended.py python
999 lines 42.6 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 121 days ago
1 """Extended integrity tests for ``muse verify`` / ``run_verify``.
2
3 Covers gaps left by test_cmd_verify.py, test_cmd_verify_hardening.py, and
4 test_cmd_verify_shallow.py:
5
6 Signature verification (run_verify BFS path, not verify-commit):
7 S1 Valid Ed25519 signature — run_verify must NOT report a failure.
8 S2 Tampered commit payload — signature present but payload changed → kind="signature".
9 S3 Wrong signature bytes (bit-flip) — Ed25519 rejects → kind="signature".
10 S4 Unknown signature algorithm prefix (e.g. "ml-dsa-65:…") → kind="signature".
11 S5 Unknown public-key algorithm prefix (e.g. "ml-dsa-65:…") → kind="key_missing".
12 S6 Malformed public-key base64 ("ed25519:!!!") → decode_pubkey ValueError
13 → pub_bytes=b"" → kind="signature".
14 S7 Empty signer_public_key ("") → sig_algo("") == "" → kind="key_missing".
15 S8 signatures_checked counts only signed commits (not unsigned ones).
16 S9 Mixed chain: some commits signed, some unsigned — only signed ones verified.
17 S10 Error message for sig failure names agent_id and key_id.
18
19 Merge commit (parent2_commit_id):
20 M1 Merge commit: both parent chains walked, all objects verified.
21 M2 Merge commit: corrupt object in second-parent chain detected.
22 M3 Merge commit: missing second-parent commit → kind="commit".
23
24 Ref path traversal security:
25 P1 branch="../../evil" — _branch_refs cannot escape heads dir.
26 P2 branch="/absolute/path" — does not read outside the repo.
27 P3 Ref file with binary (non-UTF-8) content — decode errors handled gracefully.
28
29 IOError / TOCTOU:
30 T1 Object file deleted between object_state returning PRESENT and _rehash_object
31 reading it — OSError propagates; CLI exits with code 3.
32
33 JSON schema completeness:
34 J1 --json output includes "strict" key.
35 J2 --json "strict" is False by default, True when --strict is passed.
36 J3 --json "check_objects" key present in all branches.
37
38 Counter accuracy:
39 C1 Same object ID referenced by two different snapshots counted once.
40 C2 signatures_checked equals the number of commits with a non-empty signature.
41 C3 hash-mismatch error message contains both expected and actual short IDs.
42 """
43
44 from __future__ import annotations
45
46 import datetime
47 import json
48 import os
49 import pathlib
50 import threading
51 from collections.abc import Mapping
52 from typing import Any
53
54 import pytest
55 from tests.cli_test_helper import CliRunner, InvokeResult
56
57 from muse.core.object_store import object_path, write_object
58 from muse.core.provenance import (
59 encode_public_key,
60 provenance_payload,
61 sign_commit_ed25519,
62 sign_commit_record,
63 verify_commit_ed25519,
64 )
65 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
66 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
67 from muse.core.types import blob_id, encode_pubkey, long_id, short_id
68 from muse.core.verify import run_verify
69 from muse.core.paths import heads_dir, muse_dir, ref_path
70
71 runner = CliRunner()
72 _REPO_ID = "verify-extended-test"
73
74
75 # ---------------------------------------------------------------------------
76 # Shared helpers
77 # ---------------------------------------------------------------------------
78
79
80 def _init_repo(path: pathlib.Path) -> pathlib.Path:
81 muse = muse_dir(path)
82 for d in ("commits", "snapshots", "objects", "refs/heads"):
83 (muse / d).mkdir(parents=True, exist_ok=True)
84 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
85 (muse / "repo.json").write_text(
86 json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8"
87 )
88 return path
89
90
91
92
93 def _make_key() -> "Any":
94 """Generate a fresh Ed25519 private key."""
95 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
96 return Ed25519PrivateKey.generate()
97
98
99 def _commit(
100 root: pathlib.Path,
101 *,
102 branch: str = "main",
103 parent_id: str | None = None,
104 parent2_id: str | None = None,
105 content: bytes = b"data",
106 idx: int = 0,
107 private_key: "Any | None" = None,
108 agent_id: str = "test-agent",
109 ) -> str:
110 """Write a complete commit (object + snapshot + commit record) and update branch ref.
111
112 When *private_key* is given the commit is Ed25519-signed.
113 Returns the commit_id.
114 """
115 raw = content + idx.to_bytes(4, "big")
116 obj_id = blob_id(raw)
117 write_object(root, obj_id, raw)
118 manifest = {f"file_{idx}.txt": obj_id}
119 snap_id = compute_snapshot_id(manifest)
120 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
121
122 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) + datetime.timedelta(hours=idx)
123 parent_ids = [pid for pid in [parent_id, parent2_id] if pid]
124
125 # signer_public_key is included in the commit_id hash — must derive it BEFORE
126 # calling compute_commit_id so the stored record passes _verify_commit_id.
127 pub_b64 = ""
128 if private_key is not None:
129 _, pub_b64 = encode_public_key(private_key)
130
131 commit_id = compute_commit_id(
132 parent_ids=parent_ids,
133 snapshot_id=snap_id,
134 message=f"commit {idx}",
135 committed_at_iso=committed_at.isoformat(),
136 signer_public_key=pub_b64,
137 )
138
139 sig = key_id = ""
140 if private_key is not None:
141 sig, _, key_id = sign_commit_record(
142 commit_id,
143 agent_id=agent_id,
144 private_key=private_key,
145 committed_at=committed_at.isoformat(),
146 )
147
148 write_commit(root, CommitRecord(
149 commit_id=commit_id,
150 repo_id="test-repo",
151 branch=branch,
152 snapshot_id=snap_id,
153 message=f"commit {idx}",
154 committed_at=committed_at,
155 parent_commit_id=parent_id,
156 parent2_commit_id=parent2_id,
157 agent_id=agent_id if private_key else "",
158 signature=sig,
159 signer_public_key=pub_b64,
160 signer_key_id=key_id,
161 ))
162 (ref_path(root, branch)).write_text(commit_id, encoding="utf-8")
163 return commit_id
164
165
166 def _env(root: pathlib.Path) -> Mapping[str, str]:
167 return {"MUSE_REPO_ROOT": str(root)}
168
169
170 def _force_write_commit(root: pathlib.Path, record: "CommitRecord") -> None:
171 """Overwrite a commit file unconditionally, bypassing write_commit idempotency.
172
173 Use only in tests that need to inject tampered records after a valid commit
174 has already been written.
175 """
176 import msgpack
177 from muse.core.store import commit_path
178 commit_file = commit_path(root, record.commit_id)
179 commit_file.write_bytes(msgpack.packb(record.to_dict(), use_bin_type=True))
180
181
182 def _invoke(root: pathlib.Path, *args: str) -> InvokeResult:
183 from muse.cli.app import main as cli_main
184 return runner.invoke(cli_main, ["verify", *args], env=_env(root))
185
186
187 # ---------------------------------------------------------------------------
188 # S — Signature verification in run_verify BFS
189 # ---------------------------------------------------------------------------
190
191
192 class TestSignatureVerification:
193 """Ed25519 signature verification exercised through run_verify's BFS walk.
194
195 These tests cover the signature branch inside run_verify, which is
196 distinct from the muse verify-commit command (a separate plumbing tool).
197 """
198
199 def test_s1_valid_signed_commit_passes(self, tmp_path: pathlib.Path) -> None:
200 """S1: A properly signed commit must not produce any failure."""
201 repo = _init_repo(tmp_path)
202 key = _make_key()
203 _commit(repo, private_key=key, idx=0)
204
205 result = run_verify(repo)
206
207 assert result["all_ok"] is True, f"Unexpected failures: {result['failures']}"
208 assert result["signatures_checked"] == 1
209 assert result["failures"] == []
210
211 def test_s2_tampered_payload_detected(self, tmp_path: pathlib.Path) -> None:
212 """S2: A commit whose agent_id differs from what was signed → signature invalid."""
213 repo = _init_repo(tmp_path)
214 key = _make_key()
215 cid = _commit(repo, private_key=key, agent_id="real-agent", idx=0)
216
217 # Re-read and tamper the commit record: change agent_id to something
218 # different from what was signed. The signature still references the
219 # original agent_id in the provenance_payload.
220 from muse.core.store import read_commit
221 original = read_commit(repo, cid)
222 assert original is not None
223 tampered = CommitRecord(
224 commit_id=original.commit_id,
225 repo_id=original.repo_id,
226 branch=original.branch,
227 snapshot_id=original.snapshot_id,
228 message=original.message,
229 committed_at=original.committed_at,
230 agent_id="evil-agent", # tampered — differs from what was signed
231 signature=original.signature,
232 signer_public_key=original.signer_public_key,
233 signer_key_id=original.signer_key_id,
234 )
235 _force_write_commit(repo, tampered)
236
237 result = run_verify(repo)
238
239 assert result["all_ok"] is False
240 sig_failures = [f for f in result["failures"] if f["kind"] == "signature"]
241 assert len(sig_failures) >= 1, f"Expected signature failure, got: {result['failures']}"
242
243 def test_s3_bit_flip_in_signature_bytes_detected(self, tmp_path: pathlib.Path) -> None:
244 """S3: One bit flipped in the stored signature bytes → Ed25519 rejects → kind='signature'."""
245 repo = _init_repo(tmp_path)
246 key = _make_key()
247 cid = _commit(repo, private_key=key, idx=0)
248
249 from muse.core.store import read_commit
250 from muse.core.types import decode_sig, encode_sig
251 original = read_commit(repo, cid)
252 assert original is not None
253 _, sig_bytes = decode_sig(original.signature)
254 # Flip one bit in the middle of the signature
255 sig_list = bytearray(sig_bytes)
256 sig_list[32] ^= 0x01
257 bad_sig = encode_sig("ed25519", bytes(sig_list))
258
259 tampered = CommitRecord(
260 commit_id=original.commit_id,
261 repo_id=original.repo_id,
262 branch=original.branch,
263 snapshot_id=original.snapshot_id,
264 message=original.message,
265 committed_at=original.committed_at,
266 agent_id=original.agent_id,
267 signature=bad_sig,
268 signer_public_key=original.signer_public_key,
269 signer_key_id=original.signer_key_id,
270 )
271 _force_write_commit(repo, tampered)
272
273 result = run_verify(repo)
274
275 assert result["all_ok"] is False
276 kinds = [f["kind"] for f in result["failures"]]
277 assert "signature" in kinds, f"Expected 'signature' failure, got: {kinds}"
278
279 def test_s4_unknown_signature_algorithm_reported(self, tmp_path: pathlib.Path) -> None:
280 """S4: sig='ml-dsa-65:…' (unknown algorithm) → kind='signature', not 'key_missing'."""
281 repo = _init_repo(tmp_path)
282 key = _make_key()
283 _, pub_b64 = encode_public_key(key)
284 content = b"unknown-sig-alg"
285 obj_id = blob_id(content)
286 write_object(repo, obj_id, content)
287 manifest = {"f.txt": obj_id}
288 snap_id = compute_snapshot_id(manifest)
289 write_snapshot(repo, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
290 committed_at = datetime.datetime(2026, 3, 1, tzinfo=datetime.timezone.utc)
291 cid = compute_commit_id(
292 parent_ids=[], snapshot_id=snap_id,
293 message="unknown alg", committed_at_iso=committed_at.isoformat(),
294 signer_public_key=pub_b64,
295 )
296 write_commit(repo, CommitRecord(
297 commit_id=cid, repo_id=_REPO_ID, branch="main",
298 snapshot_id=snap_id, message="unknown alg", committed_at=committed_at,
299 signature=f"ml-dsa-65:{'A' * 80}", # unknown prefix
300 signer_public_key=pub_b64, # valid ed25519 key
301 agent_id="future-agent",
302 ))
303 (heads_dir(repo) / "main").write_text(cid)
304
305 result = run_verify(repo)
306
307 assert result["all_ok"] is False
308 kinds = [f["kind"] for f in result["failures"]]
309 assert "signature" in kinds, f"Expected 'signature', got: {kinds}"
310 assert "key_missing" not in kinds
311
312 def test_s5_unknown_pubkey_algorithm_reported_as_key_missing(self, tmp_path: pathlib.Path) -> None:
313 """S5: sig='ed25519:…' but pub_raw='ml-dsa-65:…' → kind='key_missing', not 'signature'."""
314 repo = _init_repo(tmp_path)
315 key = _make_key()
316 content = b"unknown-pk-alg"
317 obj_id = blob_id(content)
318 write_object(repo, obj_id, content)
319 manifest = {"f.txt": obj_id}
320 snap_id = compute_snapshot_id(manifest)
321 write_snapshot(repo, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
322 committed_at = datetime.datetime(2026, 3, 2, tzinfo=datetime.timezone.utc)
323 unknown_pk = f"ml-dsa-65:{'A' * 80}"
324 cid = compute_commit_id(
325 parent_ids=[], snapshot_id=snap_id,
326 message="unknown pk alg", committed_at_iso=committed_at.isoformat(),
327 signer_public_key=unknown_pk,
328 )
329 payload = provenance_payload(cid, agent_id="future-agent",
330 committed_at=committed_at.isoformat())
331 valid_sig = sign_commit_ed25519(payload, key)
332 write_commit(repo, CommitRecord(
333 commit_id=cid, repo_id=_REPO_ID, branch="main",
334 snapshot_id=snap_id, message="unknown pk alg", committed_at=committed_at,
335 signature=valid_sig,
336 signer_public_key=unknown_pk, # unknown prefix on key
337 agent_id="future-agent",
338 ))
339 (heads_dir(repo) / "main").write_text(cid)
340
341 result = run_verify(repo)
342
343 assert result["all_ok"] is False
344 kinds = [f["kind"] for f in result["failures"]]
345 assert "key_missing" in kinds, f"Expected 'key_missing', got: {kinds}"
346 assert "signature" not in kinds
347
348 def test_s6_malformed_pubkey_base64_causes_signature_failure(self, tmp_path: pathlib.Path) -> None:
349 """S6: pub_raw='ed25519:!!!' (valid prefix, invalid base64) → decode_pubkey raises
350 ValueError → pub_bytes=b'' → kind='signature'."""
351 repo = _init_repo(tmp_path)
352 key = _make_key()
353 content = b"bad-b64-key"
354 obj_id = blob_id(content)
355 write_object(repo, obj_id, content)
356 manifest = {"f.txt": obj_id}
357 snap_id = compute_snapshot_id(manifest)
358 write_snapshot(repo, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
359 committed_at = datetime.datetime(2026, 3, 3, tzinfo=datetime.timezone.utc)
360 bad_pk = "ed25519:!!!notvalidbase64!!!"
361 cid = compute_commit_id(
362 parent_ids=[], snapshot_id=snap_id,
363 message="bad b64 key", committed_at_iso=committed_at.isoformat(),
364 signer_public_key=bad_pk,
365 )
366 payload = provenance_payload(cid, agent_id="agent",
367 committed_at=committed_at.isoformat())
368 valid_sig = sign_commit_ed25519(payload, key)
369 write_commit(repo, CommitRecord(
370 commit_id=cid, repo_id=_REPO_ID, branch="main",
371 snapshot_id=snap_id, message="bad b64 key", committed_at=committed_at,
372 signature=valid_sig,
373 signer_public_key=bad_pk, # prefix ok, content not valid base64
374 agent_id="agent",
375 ))
376 (heads_dir(repo) / "main").write_text(cid)
377
378 result = run_verify(repo)
379
380 assert result["all_ok"] is False
381 kinds = [f["kind"] for f in result["failures"]]
382 assert "signature" in kinds, f"Expected 'signature' failure, got: {kinds}"
383
384 def test_s7_empty_signer_public_key_reported_as_key_missing(self, tmp_path: pathlib.Path) -> None:
385 """S7: signer_public_key='' → sig_algo('') == '' != 'ed25519' → kind='key_missing'."""
386 repo = _init_repo(tmp_path)
387 key = _make_key()
388 content = b"no-pk"
389 obj_id = blob_id(content)
390 write_object(repo, obj_id, content)
391 manifest = {"f.txt": obj_id}
392 snap_id = compute_snapshot_id(manifest)
393 write_snapshot(repo, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
394 committed_at = datetime.datetime(2026, 3, 4, tzinfo=datetime.timezone.utc)
395 cid = compute_commit_id(
396 parent_ids=[], snapshot_id=snap_id,
397 message="no pk", committed_at_iso=committed_at.isoformat(),
398 )
399 payload = provenance_payload(cid, committed_at=committed_at.isoformat())
400 valid_sig = sign_commit_ed25519(payload, key)
401 write_commit(repo, CommitRecord(
402 commit_id=cid, repo_id=_REPO_ID, branch="main",
403 snapshot_id=snap_id, message="no pk", committed_at=committed_at,
404 signature=valid_sig,
405 signer_public_key="", # key rotation / missing key
406 ))
407 (heads_dir(repo) / "main").write_text(cid)
408
409 result = run_verify(repo)
410
411 assert result["all_ok"] is False
412 kinds = [f["kind"] for f in result["failures"]]
413 assert "key_missing" in kinds, f"Expected 'key_missing', got: {kinds}"
414 assert "signature" not in kinds
415
416 def test_s8_unsigned_commits_not_counted(self, tmp_path: pathlib.Path) -> None:
417 """S8: Commits with empty signature field do not increment signatures_checked."""
418 repo = _init_repo(tmp_path)
419 prev = _commit(repo, idx=0) # unsigned
420 _commit(repo, parent_id=prev, idx=1) # unsigned
421
422 result = run_verify(repo)
423
424 assert result["all_ok"] is True
425 assert result["signatures_checked"] == 0
426
427 def test_s9_mixed_chain_counts_only_signed(self, tmp_path: pathlib.Path) -> None:
428 """S9: 3-commit chain: commit 0 unsigned, commit 1 signed, commit 2 unsigned.
429 signatures_checked must be exactly 1 and all_ok must be True."""
430 repo = _init_repo(tmp_path)
431 key = _make_key()
432 c0 = _commit(repo, idx=0) # unsigned
433 c1 = _commit(repo, parent_id=c0, idx=1, private_key=key) # signed
434 _commit(repo, parent_id=c1, idx=2) # unsigned
435
436 result = run_verify(repo)
437
438 assert result["all_ok"] is True, f"Failures: {result['failures']}"
439 assert result["signatures_checked"] == 1
440 assert result["commits_checked"] == 3
441
442 def test_s10_signature_failure_error_names_agent(self, tmp_path: pathlib.Path) -> None:
443 """S10: Signature failure error message includes agent_id and key reference."""
444 repo = _init_repo(tmp_path)
445 key = _make_key()
446 cid = _commit(repo, private_key=key, agent_id="my-special-agent", idx=0)
447
448 # Tamper the signature bytes so verification fails
449 from muse.core.store import read_commit
450 from muse.core.types import decode_sig, encode_sig
451 original = read_commit(repo, cid)
452 assert original is not None
453 _, sig_bytes = decode_sig(original.signature)
454 bad_sig = encode_sig("ed25519", bytes([sig_bytes[0] ^ 0xFF]) + sig_bytes[1:])
455 _force_write_commit(repo, CommitRecord(
456 commit_id=original.commit_id, repo_id=original.repo_id,
457 branch=original.branch, snapshot_id=original.snapshot_id,
458 message=original.message, committed_at=original.committed_at,
459 agent_id="my-special-agent",
460 signature=bad_sig,
461 signer_public_key=original.signer_public_key,
462 signer_key_id=original.signer_key_id,
463 ))
464
465 result = run_verify(repo)
466
467 assert result["all_ok"] is False
468 sig_failures = [f for f in result["failures"] if f["kind"] == "signature"]
469 assert sig_failures
470 error_msg = sig_failures[0]["error"]
471 assert "my-special-agent" in error_msg or short_id(cid) in error_msg, (
472 f"Error message should name agent or commit: {error_msg!r}"
473 )
474
475
476 # ---------------------------------------------------------------------------
477 # M — Merge commits (parent2_commit_id)
478 # ---------------------------------------------------------------------------
479
480
481 class TestMergeCommits:
482 """parent2_commit_id in the BFS walk — both parent chains verified."""
483
484 def _make_branch_commit(
485 self,
486 root: pathlib.Path,
487 branch: str,
488 idx: int,
489 parent_id: str | None = None,
490 ) -> tuple[str, str]:
491 """Create a commit on *branch* and return (commit_id, obj_id)."""
492 content = f"branch-{branch}-{idx}".encode()
493 obj_id = blob_id(content)
494 write_object(root, obj_id, content)
495 manifest = {f"{branch}_{idx}.py": obj_id}
496 snap_id = compute_snapshot_id(manifest)
497 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
498 committed_at = (
499 datetime.datetime(2026, 2, 1, tzinfo=datetime.timezone.utc)
500 + datetime.timedelta(hours=idx)
501 )
502 parent_ids = [parent_id] if parent_id else []
503 cid = compute_commit_id(
504 parent_ids=parent_ids, snapshot_id=snap_id,
505 message=f"{branch} commit {idx}", committed_at_iso=committed_at.isoformat(),
506 )
507 write_commit(root, CommitRecord(
508 commit_id=cid, repo_id=_REPO_ID, branch=branch,
509 snapshot_id=snap_id, message=f"{branch} commit {idx}",
510 committed_at=committed_at, parent_commit_id=parent_id,
511 ))
512 (ref_path(root, branch)).write_text(cid)
513 return cid, obj_id
514
515 def test_m1_merge_commit_both_parents_walked(self, tmp_path: pathlib.Path) -> None:
516 """M1: A merge commit with two parents; objects from both parent chains verified."""
517 repo = _init_repo(tmp_path)
518
519 # main branch: one commit
520 main_cid, main_obj = self._make_branch_commit(repo, "main", idx=0)
521 # feat branch: one commit
522 feat_cid, feat_obj = self._make_branch_commit(repo, "feat", idx=1)
523
524 # Merge commit: parent1=main, parent2=feat
525 merge_content = b"merge-content"
526 merge_obj = blob_id(merge_content)
527 write_object(repo, merge_obj, merge_content)
528 manifest = {"merge.py": merge_obj}
529 snap_id = compute_snapshot_id(manifest)
530 write_snapshot(repo, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
531 committed_at = datetime.datetime(2026, 2, 1, 12, tzinfo=datetime.timezone.utc)
532 merge_cid = compute_commit_id(
533 parent_ids=[main_cid, feat_cid],
534 snapshot_id=snap_id,
535 message="merge feat into main",
536 committed_at_iso=committed_at.isoformat(),
537 )
538 write_commit(repo, CommitRecord(
539 commit_id=merge_cid, repo_id=_REPO_ID, branch="main",
540 snapshot_id=snap_id, message="merge feat into main",
541 committed_at=committed_at, parent_commit_id=main_cid,
542 parent2_commit_id=feat_cid,
543 ))
544 (heads_dir(repo) / "main").write_text(merge_cid)
545
546 result = run_verify(repo)
547
548 assert result["all_ok"] is True, f"Failures: {result['failures']}"
549 # 3 distinct commits: main + feat + merge (feat also has its own branch ref)
550 assert result["commits_checked"] >= 3
551 # All 3 objects must have been checked
552 assert result["objects_checked"] >= 3
553
554 def test_m2_corrupt_object_in_second_parent_chain_detected(
555 self, tmp_path: pathlib.Path
556 ) -> None:
557 """M2: Corruption in an object reachable only via parent2 is caught."""
558 repo = _init_repo(tmp_path)
559
560 main_cid, _ = self._make_branch_commit(repo, "main", idx=0)
561 feat_cid, feat_obj = self._make_branch_commit(repo, "feat", idx=1)
562
563 # Corrupt the feat object
564 feat_file = object_path(repo, feat_obj)
565 os.chmod(feat_file, 0o644)
566 feat_file.write_bytes(b"corrupted by test")
567
568 # Merge with feat as parent2
569 merge_content = b"merge"
570 merge_obj = blob_id(merge_content)
571 write_object(repo, merge_obj, merge_content)
572 manifest = {"m.py": merge_obj}
573 snap_id = compute_snapshot_id(manifest)
574 write_snapshot(repo, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
575 committed_at = datetime.datetime(2026, 2, 2, tzinfo=datetime.timezone.utc)
576 merge_cid = compute_commit_id(
577 parent_ids=[main_cid, feat_cid], snapshot_id=snap_id,
578 message="merge", committed_at_iso=committed_at.isoformat(),
579 )
580 write_commit(repo, CommitRecord(
581 commit_id=merge_cid, repo_id=_REPO_ID, branch="main",
582 snapshot_id=snap_id, message="merge", committed_at=committed_at,
583 parent_commit_id=main_cid, parent2_commit_id=feat_cid,
584 ))
585 (heads_dir(repo) / "main").write_text(merge_cid)
586
587 result = run_verify(repo, check_objects=True)
588
589 assert result["all_ok"] is False
590 object_failures = [f for f in result["failures"] if f["kind"] == "object"]
591 assert any(f["id"] == feat_obj for f in object_failures), (
592 f"Expected feat_obj failure, got: {object_failures}"
593 )
594
595 def test_m3_missing_second_parent_commit_reported(self, tmp_path: pathlib.Path) -> None:
596 """M3: parent2_commit_id points to a nonexistent commit → kind='commit'."""
597 repo = _init_repo(tmp_path)
598
599 main_cid, _ = self._make_branch_commit(repo, "main", idx=0)
600 phantom_parent = long_id("d" * 64) # will be stubbed — verify must report it missing
601 from muse.core.store import commit_path as _cp
602 _stub = _cp(repo, phantom_parent)
603 _stub.parent.mkdir(parents=True, exist_ok=True)
604 _stub.write_bytes(b"") # unreadable stub; verify walks it and reports missing
605
606 merge_content = b"merge-phantom"
607 merge_obj = blob_id(merge_content)
608 write_object(repo, merge_obj, merge_content)
609 manifest = {"mp.py": merge_obj}
610 snap_id = compute_snapshot_id(manifest)
611 write_snapshot(repo, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
612 committed_at = datetime.datetime(2026, 2, 3, tzinfo=datetime.timezone.utc)
613 merge_cid = compute_commit_id(
614 parent_ids=[main_cid, phantom_parent], snapshot_id=snap_id,
615 message="merge phantom", committed_at_iso=committed_at.isoformat(),
616 )
617 write_commit(repo, CommitRecord(
618 commit_id=merge_cid, repo_id=_REPO_ID, branch="main",
619 snapshot_id=snap_id, message="merge phantom", committed_at=committed_at,
620 parent_commit_id=main_cid, parent2_commit_id=phantom_parent,
621 ))
622 (heads_dir(repo) / "main").write_text(merge_cid)
623
624 result = run_verify(repo)
625
626 assert result["all_ok"] is False
627 commit_failures = [f for f in result["failures"] if f["kind"] == "commit"]
628 assert any(f["id"] == phantom_parent for f in commit_failures), (
629 f"Expected commit failure for phantom parent: {commit_failures}"
630 )
631
632
633 # ---------------------------------------------------------------------------
634 # P — Path traversal and ref security
635 # ---------------------------------------------------------------------------
636
637
638 class TestRefSecurity:
639 """Ref file security: path traversal, binary content, oversized files."""
640
641 def test_p1_path_traversal_via_branch_param_does_not_escape(
642 self, tmp_path: pathlib.Path
643 ) -> None:
644 """P1: branch='../../evil' cannot traverse outside the heads directory.
645
646 _branch_refs constructs heads_dir / branch. Python's Path resolves
647 '..' lazily — 'heads/../../evil' normalises to '.muse/evil' which
648 should not exist. The result must be an empty ref list (not a
649 failure, just nothing found).
650 """
651 repo = _init_repo(tmp_path)
652 # Write a file the traversal might try to read
653 evil_file = muse_dir(repo) / "evil"
654 evil_file.write_text(long_id("a" * 64))
655
656 from muse.core.verify import _branch_refs # type: ignore[attr-defined]
657 refs = _branch_refs(repo, branch="../../evil")
658 # Must return empty — either the file didn't resolve into heads/ or
659 # was not found. The critical requirement: no crash and no refs returned
660 # that would cause BFS to walk attacker-controlled data as a commit ID.
661 assert refs == [] or all(commit_id.startswith("sha256:") for _, commit_id in refs)
662
663 def test_p2_absolute_path_branch_does_not_read_outside_repo(
664 self, tmp_path: pathlib.Path
665 ) -> None:
666 """P2: branch='/etc/passwd' is joined to heads_dir — Path joins strip leading /
667 on some platforms or produce a heads_dir-relative path. Either way no
668 sensitive file is read and no crash occurs."""
669 repo = _init_repo(tmp_path)
670
671 from muse.core.verify import _branch_refs # type: ignore[attr-defined]
672 # Must not raise; may return [] or a ref if heads_dir//etc/passwd exists (it won't)
673 try:
674 refs = _branch_refs(repo, branch="/etc/passwd")
675 except Exception as exc:
676 pytest.fail(f"_branch_refs raised on absolute branch path: {exc}")
677 # No valid commit ID should come from /etc/passwd content
678 for _, cid in refs:
679 assert cid.startswith("sha256:") and len(cid) == 71, (
680 f"Suspicious commit ID from absolute path branch: {cid!r}"
681 )
682
683 def test_p3_binary_ref_file_handled_gracefully(self, tmp_path: pathlib.Path) -> None:
684 """P3: Binary (non-UTF-8) content in a ref file is decoded with errors='replace'
685 and produces an invalid ref ID → kind='ref' failure, no crash."""
686 repo = _init_repo(tmp_path)
687 # Write binary garbage to the ref file
688 (heads_dir(repo) / "main").write_bytes(b"\xff\xfe\x00binary\x01garbage")
689
690 result = run_verify(repo)
691
692 # Must not raise; the invalid ref ID should be reported
693 assert result["all_ok"] is False
694 kinds = [f["kind"] for f in result["failures"]]
695 assert "ref" in kinds, f"Expected 'ref' failure for binary content, got: {kinds}"
696
697
698 # ---------------------------------------------------------------------------
699 # T — IOError / TOCTOU
700 # ---------------------------------------------------------------------------
701
702
703 class TestIOErrorHandling:
704 """IOError propagation from _rehash_object and related paths."""
705
706 def test_t1_object_deleted_between_state_check_and_read(
707 self, tmp_path: pathlib.Path
708 ) -> None:
709 """T1: Object file exists when object_state runs but is deleted before
710 _rehash_object opens it → OSError propagates through run_verify.
711 The CLI must exit with code 3 (INTERNAL_ERROR)."""
712 repo = _init_repo(tmp_path)
713 content = b"will be deleted"
714 obj_id = blob_id(content)
715 write_object(repo, obj_id, content)
716 manifest = {"toctou.py": obj_id}
717 snap_id = compute_snapshot_id(manifest)
718 write_snapshot(repo, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
719 committed_at = datetime.datetime(2026, 4, 10, tzinfo=datetime.timezone.utc)
720 cid = compute_commit_id(
721 parent_ids=[], snapshot_id=snap_id,
722 message="toctou test", committed_at_iso=committed_at.isoformat(),
723 )
724 write_commit(repo, CommitRecord(
725 commit_id=cid, repo_id=_REPO_ID, branch="main",
726 snapshot_id=snap_id, message="toctou test", committed_at=committed_at,
727 ))
728 (heads_dir(repo) / "main").write_text(cid)
729
730 # Delete the object after writing it (simulate TOCTOU)
731 obj_file = object_path(repo, obj_id)
732 os.chmod(obj_file, 0o644)
733 os.unlink(obj_file)
734
735 # run_verify itself should raise OSError (not silently swallow it)
736 # OR handle it and produce a failure. Both are acceptable; what's NOT
737 # acceptable is silently reporting all_ok=True.
738 try:
739 result = run_verify(repo, check_objects=True)
740 # If run_verify catches the OSError internally, it must report a failure
741 assert result["all_ok"] is False, (
742 "run_verify must not report all_ok=True when an object is unreadable"
743 )
744 except OSError:
745 # Also acceptable: OSError propagates to CLI level
746 pass
747
748
749 # ---------------------------------------------------------------------------
750 # J — JSON schema completeness
751 # ---------------------------------------------------------------------------
752
753
754 class TestJsonSchema:
755 """JSON output must include all documented fields."""
756
757 def test_j1_strict_field_present_in_json(self, tmp_path: pathlib.Path) -> None:
758 """J1: The 'strict' key must appear in --json output."""
759 repo = _init_repo(tmp_path)
760 _commit(repo, idx=0)
761 result = _invoke(repo, "--json")
762 assert result.exit_code == 0
763 data = json.loads(result.output)
764 assert "strict" in data, f"'strict' missing from JSON: {list(data.keys())}"
765
766 def test_j2_strict_false_by_default(self, tmp_path: pathlib.Path) -> None:
767 """J2: Default invocation must have strict=False in JSON output."""
768 repo = _init_repo(tmp_path)
769 _commit(repo, idx=0)
770 data = json.loads(_invoke(repo, "--json").output)
771 assert data["strict"] is False
772
773 def test_j2b_strict_true_when_flag_passed(self, tmp_path: pathlib.Path) -> None:
774 """J2b: --strict must set strict=True in JSON output."""
775 repo = _init_repo(tmp_path)
776 _commit(repo, idx=0)
777 data = json.loads(_invoke(repo, "--strict", "--json").output)
778 assert data["strict"] is True
779
780 def test_j3_check_objects_present_in_all_branches(self, tmp_path: pathlib.Path) -> None:
781 """J3: 'check_objects' must appear whether or not --no-objects is passed."""
782 repo = _init_repo(tmp_path)
783 _commit(repo, idx=0)
784 d1 = json.loads(_invoke(repo, "--json").output)
785 d2 = json.loads(_invoke(repo, "--no-objects", "--json").output)
786 assert "check_objects" in d1
787 assert "check_objects" in d2
788 assert d1["check_objects"] is True
789 assert d2["check_objects"] is False
790
791 def test_j4_all_documented_fields_present(self, tmp_path: pathlib.Path) -> None:
792 """J4: Every field documented in the command docstring appears in JSON."""
793 repo = _init_repo(tmp_path)
794 _commit(repo, idx=0)
795 data = json.loads(_invoke(repo, "--json").output)
796 required_fields = {
797 "repo_id", "refs_checked", "commits_checked", "snapshots_checked",
798 "objects_checked", "signatures_checked", "all_ok", "nothing_checked",
799 "check_objects", "strict", "branch", "fail_fast", "failures",
800 "shallow_commits", "promised_objects", "is_shallow", "promisor_remotes",
801 "muse_version", "schema", "exit_code", "duration_ms", "timestamp",
802 "warnings",
803 }
804 missing = required_fields - set(data.keys())
805 assert not missing, f"JSON output missing fields: {missing}"
806
807 def test_j5_failures_list_empty_when_all_ok(self, tmp_path: pathlib.Path) -> None:
808 """J5: When all_ok=True the failures list must be [] (not absent)."""
809 repo = _init_repo(tmp_path)
810 _commit(repo, idx=0)
811 data = json.loads(_invoke(repo, "--json").output)
812 assert data["all_ok"] is True
813 assert data["failures"] == []
814
815
816 # ---------------------------------------------------------------------------
817 # C — Counter accuracy
818 # ---------------------------------------------------------------------------
819
820
821 class TestCounterAccuracy:
822 """Verify that all counters are accurate, deduplicated, and never inflated."""
823
824 def test_c1_same_object_across_two_snapshots_counted_once(
825 self, tmp_path: pathlib.Path
826 ) -> None:
827 """C1: One object ID referenced by two different snapshots must appear
828 in objects_checked exactly once (deduplication via verified_objects set)."""
829 repo = _init_repo(tmp_path)
830 shared_content = b"shared object"
831 shared_obj = blob_id(shared_content)
832 write_object(repo, shared_obj, shared_content)
833
834 # Commit 0: snapshot references shared_obj
835 manifest0 = {"shared.py": shared_obj}
836 snap0 = compute_snapshot_id(manifest0)
837 write_snapshot(repo, SnapshotRecord(snapshot_id=snap0, manifest=manifest0))
838 committed_at0 = datetime.datetime(2026, 5, 1, tzinfo=datetime.timezone.utc)
839 cid0 = compute_commit_id(
840 parent_ids=[], snapshot_id=snap0,
841 message="c0", committed_at_iso=committed_at0.isoformat(),
842 )
843 write_commit(repo, CommitRecord(
844 commit_id=cid0, repo_id=_REPO_ID, branch="main",
845 snapshot_id=snap0, message="c0", committed_at=committed_at0,
846 ))
847
848 # Commit 1: different snapshot, same shared_obj
849 extra_content = b"extra"
850 extra_obj = blob_id(extra_content)
851 write_object(repo, extra_obj, extra_content)
852 manifest1 = {"shared.py": shared_obj, "extra.py": extra_obj}
853 snap1 = compute_snapshot_id(manifest1)
854 write_snapshot(repo, SnapshotRecord(snapshot_id=snap1, manifest=manifest1))
855 committed_at1 = datetime.datetime(2026, 5, 2, tzinfo=datetime.timezone.utc)
856 cid1 = compute_commit_id(
857 parent_ids=[cid0], snapshot_id=snap1,
858 message="c1", committed_at_iso=committed_at1.isoformat(),
859 )
860 write_commit(repo, CommitRecord(
861 commit_id=cid1, repo_id=_REPO_ID, branch="main",
862 snapshot_id=snap1, message="c1", committed_at=committed_at1,
863 parent_commit_id=cid0,
864 ))
865 (heads_dir(repo) / "main").write_text(cid1)
866
867 result = run_verify(repo, check_objects=True)
868
869 assert result["all_ok"] is True
870 # 2 distinct objects: shared_obj + extra_obj (shared_obj counted once)
871 assert result["objects_checked"] == 2, (
872 f"Expected 2 unique objects, got {result['objects_checked']}"
873 )
874
875 def test_c2_signatures_checked_exact_count(self, tmp_path: pathlib.Path) -> None:
876 """C2: signatures_checked equals exactly the number of commits with
877 a non-empty 'signature' field."""
878 repo = _init_repo(tmp_path)
879 key = _make_key()
880 prev = None
881 for i in range(5):
882 # Alternate: even-indexed commits are signed
883 pk = key if i % 2 == 0 else None
884 prev = _commit(repo, parent_id=prev, idx=i, private_key=pk)
885
886 result = run_verify(repo)
887
888 # Commits 0, 2, 4 are signed → 3 signatures_checked
889 assert result["all_ok"] is True, f"Failures: {result['failures']}"
890 assert result["signatures_checked"] == 3
891
892 def test_c3_hash_mismatch_error_shows_both_ids(self, tmp_path: pathlib.Path) -> None:
893 """C3: A hash mismatch failure's error string contains both the expected
894 short ID and the actual short ID computed from the corrupted content."""
895 repo = _init_repo(tmp_path)
896 content = b"original content for c3"
897 obj_id = blob_id(content)
898 write_object(repo, obj_id, content)
899 manifest = {"c3.py": obj_id}
900 snap_id = compute_snapshot_id(manifest)
901 write_snapshot(repo, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
902 committed_at = datetime.datetime(2026, 5, 3, tzinfo=datetime.timezone.utc)
903 cid = compute_commit_id(
904 parent_ids=[], snapshot_id=snap_id,
905 message="c3", committed_at_iso=committed_at.isoformat(),
906 )
907 write_commit(repo, CommitRecord(
908 commit_id=cid, repo_id=_REPO_ID, branch="main",
909 snapshot_id=snap_id, message="c3", committed_at=committed_at,
910 ))
911 (heads_dir(repo) / "main").write_text(cid)
912
913 corrupt_content = b"corrupted replacement bytes for c3"
914 obj_file = object_path(repo, obj_id)
915 os.chmod(obj_file, 0o644)
916 obj_file.write_bytes(corrupt_content)
917
918 result = run_verify(repo, check_objects=True)
919
920 assert result["all_ok"] is False
921 obj_failures = [f for f in result["failures"] if f["kind"] == "object"]
922 assert obj_failures
923 error_msg = obj_failures[0]["error"]
924 # Error must mention the expected short ID or the actual short ID
925 actual_id = blob_id(corrupt_content)
926 assert short_id(obj_id) in error_msg or short_id(actual_id) in error_msg, (
927 f"Error message should contain short ID reference: {error_msg!r}"
928 )
929 # Keyword "mismatch" or "corruption" must appear
930 assert "mismatch" in error_msg or "corruption" in error_msg, (
931 f"Error must describe the problem: {error_msg!r}"
932 )
933
934 def test_c4_commit_count_accurate_on_diamond_dag(self, tmp_path: pathlib.Path) -> None:
935 """C4: Diamond-shaped DAG (main←A, main←B, merge←A+B) — each commit
936 counted exactly once despite two paths to common ancestors."""
937 repo = _init_repo(tmp_path)
938
939 # Common ancestor
940 base_cid, _ = self._make_raw_commit(repo, "main", idx=0, parent=None)
941 # Two diverging branches
942 a_cid, _ = self._make_raw_commit(repo, "feat-a", idx=1, parent=base_cid)
943 b_cid, _ = self._make_raw_commit(repo, "feat-b", idx=2, parent=base_cid)
944 # Merge
945 merge_content = b"diamond-merge"
946 merge_obj = blob_id(merge_content)
947 write_object(repo, merge_obj, merge_content)
948 manifest = {"m.py": merge_obj}
949 snap_id = compute_snapshot_id(manifest)
950 write_snapshot(repo, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
951 committed_at = datetime.datetime(2026, 5, 10, tzinfo=datetime.timezone.utc)
952 merge_cid = compute_commit_id(
953 parent_ids=[a_cid, b_cid], snapshot_id=snap_id,
954 message="merge", committed_at_iso=committed_at.isoformat(),
955 )
956 write_commit(repo, CommitRecord(
957 commit_id=merge_cid, repo_id=_REPO_ID, branch="main",
958 snapshot_id=snap_id, message="merge", committed_at=committed_at,
959 parent_commit_id=a_cid, parent2_commit_id=b_cid,
960 ))
961 (heads_dir(repo) / "main").write_text(merge_cid)
962
963 result = run_verify(repo)
964
965 assert result["all_ok"] is True
966 # 4 commits: base + A + B + merge — base must NOT be counted twice
967 assert result["commits_checked"] == 4, (
968 f"Expected 4 commits in diamond DAG, got {result['commits_checked']}"
969 )
970
971 def _make_raw_commit(
972 self,
973 root: pathlib.Path,
974 branch: str,
975 idx: int,
976 parent: str | None,
977 ) -> tuple[str, str]:
978 content = f"raw-{branch}-{idx}".encode()
979 obj_id = blob_id(content)
980 write_object(root, obj_id, content)
981 manifest = {f"{branch}_{idx}.py": obj_id}
982 snap_id = compute_snapshot_id(manifest)
983 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
984 committed_at = (
985 datetime.datetime(2026, 5, 1, tzinfo=datetime.timezone.utc)
986 + datetime.timedelta(hours=idx)
987 )
988 parent_ids = [parent] if parent else []
989 cid = compute_commit_id(
990 parent_ids=parent_ids, snapshot_id=snap_id,
991 message=f"{branch} {idx}", committed_at_iso=committed_at.isoformat(),
992 )
993 write_commit(root, CommitRecord(
994 commit_id=cid, repo_id=_REPO_ID, branch=branch,
995 snapshot_id=snap_id, message=f"{branch} {idx}",
996 committed_at=committed_at, parent_commit_id=parent,
997 ))
998 (ref_path(root, branch)).write_text(cid)
999 return cid, obj_id
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago