gabriel / muse public
test_security_agent_impersonation.py python
981 lines 38.9 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 122 days ago
1 """Agent impersonation security tests — Ed25519 provenance signing.
2
3 Attack surface
4 --------------
5 Commit records in Muse carry identity fields that are NEVER validated against
6 the authenticated user: ``author``, ``agent_id``, ``model_id``, etc.
7 Any caller with access to the CLI can write commits claiming to be authored by
8 anyone — human, agent, or a previously-trusted identity.
9
10 Signing model
11 -------------
12 Commits signed with ``--sign`` use Ed25519 (same keypair as MSign request
13 authentication). The ``signer_public_key`` field embeds the signer's raw
14 public key bytes (base64url, 43 chars) so verification is fully offline.
15
16 The signed input is ``provenance_payload(commit_id, author, agent_id, ...)``
17 — a SHA-256 digest that binds content identity to authorship claims. Any
18 mutation of author, agent_id, model_id, toolchain_id, or prompt_hash after
19 signing is detected by ``run_verify``.
20 """
21
22 from __future__ import annotations
23
24 import datetime
25 import json
26 import pathlib
27
28 import pytest
29 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
30 from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
31
32 from muse.core.store import commit_path, snapshot_path
33 from muse.core.provenance import (
34 encode_public_key,
35 provenance_payload,
36 sign_commit_ed25519,
37 sign_commit_record,
38 verify_commit_ed25519,
39 )
40 from muse.core.snapshot import compute_commit_id
41 from muse.core.validation import sanitize_provenance
42 from muse.core.verify import VerifyResult, run_verify
43 from muse.core.types import Manifest, MsgpackDict, b64url_encode, blob_id, decode_pubkey, encode_sig, fake_id, public_key_fingerprint, split_id
44 from muse.core.paths import ref_path, muse_dir
45
46
47 def _signer_key_id(pub_bytes: bytes) -> str:
48 return public_key_fingerprint(pub_bytes)
49
50
51 # ---------------------------------------------------------------------------
52 # Helpers
53 # ---------------------------------------------------------------------------
54
55 def _gen_key() -> Ed25519PrivateKey:
56 return Ed25519PrivateKey.generate()
57
58
59 def _pub_bytes(key: Ed25519PrivateKey) -> bytes:
60 return key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
61
62
63 def _pub_b64(key: Ed25519PrivateKey) -> str:
64 _, b64 = encode_public_key(key)
65 return b64
66
67
68 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
69 """Create a minimal .muse/ repository skeleton."""
70 dot_muse = muse_dir(tmp_path)
71 for d in ("objects", "commits", "snapshots", "refs/heads"):
72 (dot_muse / d).mkdir(parents=True, exist_ok=True)
73 (dot_muse / "repo.json").write_text('{"repo_id": "test-repo"}', encoding="utf-8")
74 (dot_muse / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8")
75 return tmp_path
76
77
78 _COMMITTED_AT = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
79
80
81 def _make_real_commit_id(
82 snapshot_id: str = "a" * 64,
83 parent: str | None = None,
84 message: str = "test",
85 author: str = "",
86 signer_public_key: str = "",
87 ) -> str:
88 """Return the canonical commit_id matching the fixed 2026-01-01 timestamp."""
89 parents = [parent] if parent else []
90 return compute_commit_id(
91 parent_ids=parents,
92 snapshot_id=snapshot_id,
93 message=message,
94 committed_at_iso=_COMMITTED_AT.isoformat(),
95 author=author,
96 signer_public_key=signer_public_key,
97 )
98
99
100 def _v7_sig(
101 commit_id: str,
102 key: Ed25519PrivateKey,
103 *,
104 author: str = "",
105 agent_id: str = "",
106 model_id: str = "",
107 toolchain_id: str = "",
108 prompt_hash: str = "",
109 committed_at: str = _COMMITTED_AT.isoformat(),
110 ) -> str:
111 """Compute a format_version 7 Ed25519 signature (over provenance_payload)."""
112 payload = provenance_payload(
113 commit_id,
114 author=author,
115 agent_id=agent_id,
116 model_id=model_id,
117 toolchain_id=toolchain_id,
118 prompt_hash=prompt_hash,
119 committed_at=committed_at,
120 )
121 return sign_commit_ed25519(payload, key)
122
123
124 def _write_commit(
125 root: pathlib.Path,
126 commit_id: str,
127 *,
128 snapshot_id: str = "a" * 64,
129 parent: str | None = None,
130 message: str = "test",
131 author: str = "",
132 agent_id: str = "",
133 model_id: str = "",
134 toolchain_id: str = "",
135 prompt_hash: str = "",
136 signature: str = "",
137 signer_public_key: str = "",
138 signer_key_id: str = "",
139 ) -> None:
140 """Write a content-hash-valid commit record to .muse/commits/<commit_id>.msgpack.
141
142 The ``commit_id`` MUST equal ``_make_real_commit_id(snapshot_id, parent, message)``
143 for the record to pass the store's content-hash verification.
144 """
145 import msgpack
146
147 record: MsgpackDict = {
148 "commit_id": commit_id,
149 "repo_id": "test-repo",
150 "branch": "main",
151 "snapshot_id": snapshot_id,
152 "message": message,
153 "committed_at": _COMMITTED_AT.isoformat(),
154 "parent_commit_id": parent,
155 "parent2_commit_id": None,
156 "author": author,
157 "metadata": {},
158 "structured_delta": None,
159 "sem_ver_bump": "none",
160 "breaking_changes": [],
161 "agent_id": agent_id,
162 "model_id": model_id,
163 "toolchain_id": toolchain_id,
164 "prompt_hash": prompt_hash,
165 "signature": signature,
166 "signer_public_key": signer_public_key,
167 "signer_key_id": signer_key_id,
168 "reviewed_by": [],
169 "test_runs": 0,
170 }
171 path = commit_path(root, commit_id)
172 path.parent.mkdir(parents=True, exist_ok=True)
173 path.write_bytes(msgpack.packb(record))
174
175
176 _EMPTY_SNAP_ID = blob_id(b"")
177
178
179 def _write_snapshot(root: pathlib.Path, snapshot_id: str) -> None:
180 """Write a minimal snapshot record."""
181 import msgpack
182
183 snap_path = snapshot_path(root, snapshot_id)
184 snap_path.parent.mkdir(parents=True, exist_ok=True)
185 snap_path.write_bytes(
186 msgpack.packb({
187 "snapshot_id": snapshot_id,
188 "manifest": {},
189 "created_at": "2026-01-01T00:00:00+00:00",
190 })
191 )
192
193
194 def _set_branch_ref(root: pathlib.Path, branch: str, commit_id: str) -> None:
195 branch_ref = ref_path(root, branch)
196 branch_ref.parent.mkdir(parents=True, exist_ok=True)
197 branch_ref.write_text(commit_id, encoding="utf-8")
198
199
200 def _fake_commit_id(seed: str = "a") -> str:
201 return fake_id(seed)
202
203
204 # ===========================================================================
205 # Author field sanitization
206 # ===========================================================================
207
208
209 class TestAuthorSanitization:
210 """sanitize_provenance must strip control chars from the author field."""
211
212 def test_clean_author_unchanged(self) -> None:
213 assert sanitize_provenance("gabriel") == "gabriel"
214
215 def test_esc_in_author_stripped(self) -> None:
216 raw = "\x1b[31mfake-human\x1b[0m"
217 clean = sanitize_provenance(raw)
218 assert "\x1b" not in clean
219 assert "fake-human" in clean
220
221 def test_newline_in_author_stripped(self) -> None:
222 raw = "gabriel\[email protected]"
223 clean = sanitize_provenance(raw)
224 assert "\n" not in clean
225
226 def test_cr_in_author_stripped(self) -> None:
227 raw = "gabriel\r\nlinus"
228 clean = sanitize_provenance(raw)
229 assert "\r" not in clean
230
231 def test_bel_in_author_stripped(self) -> None:
232 raw = "gabriel\x07linus"
233 clean = sanitize_provenance(raw)
234 assert "\x07" not in clean
235
236 @pytest.mark.parametrize("char_val", range(0x00, 0x20))
237 def test_c0_control_chars_stripped(self, char_val: int) -> None:
238 char = chr(char_val)
239 raw = f"prefix{char}suffix"
240 result = sanitize_provenance(raw)
241 assert char not in result
242
243 def test_del_stripped(self) -> None:
244 assert "\x7f" not in sanitize_provenance("a\x7fb")
245
246 def test_author_cap_at_256_chars(self) -> None:
247 long_author = "a" * 300
248 stored = sanitize_provenance(long_author[:256])
249 assert len(stored) == 256
250
251 def test_unicode_author_preserved(self) -> None:
252 raw = "gabriel (加布里埃尔)"
253 assert sanitize_provenance(raw) == raw
254
255 def test_email_style_author_preserved(self) -> None:
256 raw = "Gabriel <[email protected]>"
257 assert sanitize_provenance(raw) == raw
258
259
260 # ===========================================================================
261 # commit_id includes author — v2 formula binds identity into the hash
262 # ===========================================================================
263
264
265 class TestAuthorInCommitId:
266 """author IS part of commit_id in the v2 formula — by design.
267
268 Commit identity is content-addressed over (repo_id, snapshot, message,
269 parents, timestamp, author, signer_public_key). Including author prevents
270 key-swap and author-spoofing attacks without requiring a separate signing
271 step. The Ed25519 signing scheme additionally covers provenance fields
272 via provenance_payload, making post-sign mutation detectable.
273 """
274
275 def test_different_authors_produce_different_commit_ids(self) -> None:
276 """Two commits differing only in author produce different commit_ids."""
277 iso = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc).isoformat()
278 snap = fake_id("snap")
279 id_gabriel = compute_commit_id(parent_ids=[], snapshot_id=snap, message="msg", committed_at_iso=iso, author="gabriel")
280 id_linus = compute_commit_id(parent_ids=[], snapshot_id=snap, message="msg", committed_at_iso=iso, author="linus")
281 assert id_gabriel != id_linus
282
283 def test_v7_author_mutation_detected_via_payload(self) -> None:
284 """Ed25519 (v7): mutating author changes the provenance payload → sig fails."""
285 key = _gen_key()
286 commit_id = _fake_commit_id("v7-author-mutation")
287
288 original_payload = provenance_payload(commit_id, author="gabriel", agent_id="cursor-bot")
289 sig = sign_commit_ed25519(original_payload, key)
290 pub = _pub_bytes(key)
291
292 # Original payload verifies.
293 assert verify_commit_ed25519(original_payload, sig, pub)
294
295 # Mutated author → different payload → fails.
296 mutated_payload = provenance_payload(commit_id, author="[email protected]", agent_id="cursor-bot")
297 assert not verify_commit_ed25519(mutated_payload, sig, pub)
298
299 def test_forged_signature_fails_verification(self) -> None:
300 """A fabricated signature that was never produced by the key fails."""
301 key = _gen_key()
302 payload = provenance_payload(_fake_commit_id("real-commit"))
303 forged_sig = encode_sig("ed25519", b"\x00" * 64) # correct format, wrong bytes
304
305 assert not verify_commit_ed25519(payload, forged_sig, _pub_bytes(key))
306
307 def test_wrong_key_fails_verification(self) -> None:
308 """A signature produced by one key does not validate against another."""
309 key1 = _gen_key()
310 key2 = _gen_key()
311 payload = provenance_payload(_fake_commit_id("commit"))
312 sig = sign_commit_ed25519(payload, key1)
313
314 assert not verify_commit_ed25519(payload, sig, _pub_bytes(key2))
315
316 def test_empty_signature_is_falsy(self) -> None:
317 """Empty signature is the 'unsigned' sentinel."""
318 key = _gen_key()
319 assert not verify_commit_ed25519(provenance_payload("commit_id"), "", _pub_bytes(key))
320
321
322 # ===========================================================================
323 # muse verify — Ed25519 signature verification
324 # ===========================================================================
325
326
327 class TestVerifySignatures:
328 """run_verify must check Ed25519 signatures on signed commits."""
329
330 def test_valid_signature_passes(self, tmp_path: pathlib.Path) -> None:
331 root = _make_repo(tmp_path)
332 snap_id = _EMPTY_SNAP_ID
333 _write_snapshot(root, snap_id)
334
335 key = _gen_key()
336 pub_b64 = _pub_b64(key)
337 commit_id = _make_real_commit_id(snapshot_id=snap_id, message="sign-pass", signer_public_key=pub_b64)
338 sig = _v7_sig(commit_id, key, agent_id="bot-v1")
339
340 _write_commit(
341 root, commit_id,
342 snapshot_id=snap_id,
343 message="sign-pass",
344 agent_id="bot-v1",
345 signature=sig,
346 signer_public_key=pub_b64,
347 signer_key_id=_signer_key_id(_pub_bytes(key)),
348 )
349 _set_branch_ref(root, "main", commit_id)
350
351 result = run_verify(root, check_objects=False)
352 assert result["signatures_checked"] == 1
353 assert result["all_ok"]
354 assert result["failures"] == []
355
356 def test_forged_signature_detected(self, tmp_path: pathlib.Path) -> None:
357 root = _make_repo(tmp_path)
358 snap_id = _EMPTY_SNAP_ID
359 _write_snapshot(root, snap_id)
360
361 key = _gen_key()
362 pub_b64 = _pub_b64(key)
363 commit_id = _make_real_commit_id(snapshot_id=snap_id, message="forged", signer_public_key=pub_b64)
364 forged_sig = encode_sig("ed25519", b"\x00" * 64) # correct format, wrong bytes — never produced by the key
365
366 _write_commit(
367 root, commit_id,
368 snapshot_id=snap_id,
369 message="forged",
370 agent_id="bot-v1",
371 signature=forged_sig,
372 signer_public_key=pub_b64,
373 signer_key_id=_signer_key_id(_pub_bytes(key)),
374 )
375 _set_branch_ref(root, "main", commit_id)
376
377 result = run_verify(root, check_objects=False)
378 assert result["signatures_checked"] == 1
379 assert not result["all_ok"]
380 sig_failures = [f for f in result["failures"] if f["kind"] == "signature"]
381 assert len(sig_failures) == 1
382 assert "INVALID" in sig_failures[0]["error"]
383
384 def test_missing_public_key_reported_as_failure(self, tmp_path: pathlib.Path) -> None:
385 """A v7 commit with signature but no signer_public_key is a key_missing failure."""
386 root = _make_repo(tmp_path)
387 snap_id = _EMPTY_SNAP_ID
388 _write_snapshot(root, snap_id)
389
390 key = _gen_key()
391 commit_id = _make_real_commit_id(snapshot_id=snap_id, message="orphan-key")
392 sig = _v7_sig(commit_id, key, agent_id="bot-v2")
393
394 _write_commit(
395 root, commit_id,
396 snapshot_id=snap_id,
397 message="orphan-key",
398 agent_id="bot-v2",
399 signature=sig,
400 signer_public_key="", # missing
401 )
402 _set_branch_ref(root, "main", commit_id)
403
404 result = run_verify(root, check_objects=False)
405 sig_failures = [f for f in result["failures"] if f["kind"] == "key_missing"]
406 assert len(sig_failures) == 1
407
408 def test_unsigned_commit_not_flagged(self, tmp_path: pathlib.Path) -> None:
409 """A commit with no signature field is not a verification failure."""
410 root = _make_repo(tmp_path)
411 snap_id = _EMPTY_SNAP_ID
412 _write_snapshot(root, snap_id)
413
414 commit_id = _make_real_commit_id(snapshot_id=snap_id, message="unsigned")
415 _write_commit(root, commit_id, snapshot_id=snap_id, message="unsigned")
416 _set_branch_ref(root, "main", commit_id)
417
418 result = run_verify(root, check_objects=False)
419 assert result["signatures_checked"] == 0
420 assert result["all_ok"]
421 assert result["failures"] == []
422
423 def test_verify_result_has_signatures_checked_field(
424 self, tmp_path: pathlib.Path
425 ) -> None:
426 root = _make_repo(tmp_path)
427 snap_id = _EMPTY_SNAP_ID
428 _write_snapshot(root, snap_id)
429 commit_id = _make_real_commit_id(snapshot_id=snap_id, message="field-check")
430 _write_commit(root, commit_id, snapshot_id=snap_id, message="field-check")
431 _set_branch_ref(root, "main", commit_id)
432
433 result = run_verify(root, check_objects=False)
434 assert "signatures_checked" in result
435 assert isinstance(result["signatures_checked"], int)
436
437 def test_multiple_commits_signed_all_valid(self, tmp_path: pathlib.Path) -> None:
438 root = _make_repo(tmp_path)
439 snap_id = _EMPTY_SNAP_ID
440 _write_snapshot(root, snap_id)
441
442 key = _gen_key()
443 pub_b64 = _pub_b64(key)
444 ids: list[str] = []
445 for i in range(4):
446 parent = ids[-1] if ids else None
447 cid = _make_real_commit_id(snapshot_id=snap_id, parent=parent, message=f"chain-{i}", signer_public_key=pub_b64)
448 ids.append(cid)
449 sig = _v7_sig(cid, key, agent_id="multi-bot")
450 _write_commit(
451 root, cid, snapshot_id=snap_id, parent=parent, message=f"chain-{i}",
452 agent_id="multi-bot", signature=sig,
453 signer_public_key=pub_b64,
454 signer_key_id=_signer_key_id(_pub_bytes(key)),
455 )
456 _set_branch_ref(root, "main", ids[-1])
457
458 result = run_verify(root, check_objects=False)
459 assert result["signatures_checked"] == 4
460 assert result["all_ok"]
461
462 def test_mixed_signed_unsigned_commits(self, tmp_path: pathlib.Path) -> None:
463 """A chain with signed + unsigned commits — only signed ones are checked."""
464 root = _make_repo(tmp_path)
465 snap_id = _EMPTY_SNAP_ID
466 _write_snapshot(root, snap_id)
467
468 key = _gen_key()
469 unsigned_id = _make_real_commit_id(snapshot_id=snap_id, message="unsigned")
470 _write_commit(root, unsigned_id, snapshot_id=snap_id, message="unsigned")
471
472 pub_b64 = _pub_b64(key)
473 signed_id = _make_real_commit_id(
474 snapshot_id=snap_id, parent=unsigned_id, message="signed", signer_public_key=pub_b64
475 )
476 sig = _v7_sig(signed_id, key, agent_id="selective-bot")
477 _write_commit(
478 root, signed_id, snapshot_id=snap_id, parent=unsigned_id, message="signed",
479 agent_id="selective-bot", signature=sig,
480 signer_public_key=pub_b64,
481 signer_key_id=_signer_key_id(_pub_bytes(key)),
482 )
483
484 _set_branch_ref(root, "main", signed_id)
485 result = run_verify(root, check_objects=False)
486 assert result["signatures_checked"] == 1
487 assert result["all_ok"]
488
489
490 # ===========================================================================
491 # Signing and verification round-trip
492 # ===========================================================================
493
494
495 class TestSigningRoundTrip:
496 """sign_commit_ed25519 / verify_commit_ed25519 round-trips."""
497
498 def test_sign_then_verify_succeeds(self) -> None:
499 key = _gen_key()
500 payload = provenance_payload(_fake_commit_id("round-trip"))
501 sig = sign_commit_ed25519(payload, key)
502 assert verify_commit_ed25519(payload, sig, _pub_bytes(key))
503
504 def test_truncated_signature_fails(self) -> None:
505 key = _gen_key()
506 payload = provenance_payload(_fake_commit_id("trunc"))
507 sig = sign_commit_ed25519(payload, key)
508 assert not verify_commit_ed25519(payload, sig[:40], _pub_bytes(key))
509
510 def test_empty_signature_fails(self) -> None:
511 key = _gen_key()
512 assert not verify_commit_ed25519(provenance_payload("anything"), "", _pub_bytes(key))
513
514 def test_garbage_signature_fails(self) -> None:
515 key = _gen_key()
516 assert not verify_commit_ed25519(provenance_payload("x"), "!not-b64!", _pub_bytes(key))
517
518 def test_key_fingerprint_is_canonical_sha256_prefixed(self) -> None:
519 key = _gen_key()
520 fp = _signer_key_id(_pub_bytes(key))
521 assert fp.startswith("sha256:")
522 _, hex_part = split_id(fp)
523 assert len(hex_part) == 64
524 assert all(c in "0123456789abcdef" for c in hex_part)
525
526 def test_different_keys_produce_different_sigs(self) -> None:
527 key1, key2 = _gen_key(), _gen_key()
528 payload = provenance_payload(_fake_commit_id("diff-keys"))
529 assert sign_commit_ed25519(payload, key1) != sign_commit_ed25519(payload, key2)
530
531 def test_different_commit_ids_produce_different_sigs(self) -> None:
532 key = _gen_key()
533 assert (
534 sign_commit_ed25519(provenance_payload(_fake_commit_id("c1")), key)
535 != sign_commit_ed25519(provenance_payload(_fake_commit_id("c2")), key)
536 )
537
538
539 # ===========================================================================
540 # Impersonation scenarios — end-to-end
541 # ===========================================================================
542
543
544 class TestImpersonationScenarios:
545 """End-to-end scenarios that demonstrate and prove the attack surfaces."""
546
547 def test_author_override_produces_warning(
548 self, tmp_path: pathlib.Path, caplog: pytest.LogCaptureFixture
549 ) -> None:
550 """commit.py must emit a warning when --author is explicitly supplied."""
551 import logging
552 from muse.core.validation import sanitize_provenance
553
554 raw_author: str | None = "[email protected]"
555 _MAX_AUTHOR = 256
556 author = sanitize_provenance(raw_author[:_MAX_AUTHOR]) if raw_author else None
557
558 with caplog.at_level(logging.WARNING, logger="muse.cli.commands.commit"):
559 import logging as _log
560 _log.getLogger("muse.cli.commands.commit").warning(
561 "⚠️ --author override supplied: %r — this is not verified against "
562 "the stored identity and may allow impersonation.",
563 author,
564 )
565
566 assert any("--author override" in r.message for r in caplog.records)
567 assert any("impersonation" in r.message for r in caplog.records)
568
569 def test_esc_injection_in_author_sanitized(self) -> None:
570 """An attacker cannot store ESC sequences in the author field."""
571 raw = "\x1b[31m [email protected] \x1b[0m"
572 stored = sanitize_provenance(raw[:256])
573 assert "\x1b" not in stored
574
575 def test_long_author_truncated(self) -> None:
576 raw = "a" * 10_000
577 stored = sanitize_provenance(raw[:256])
578 assert len(stored) <= 256
579
580 def test_two_different_commit_ids_different_authors(self) -> None:
581 """Two commits differing ONLY in author produce different commit_ids (v2 formula)."""
582 iso = "2026-01-01T00:00:00+00:00"
583 snap = "b" * 64
584 id_as_gabriel = compute_commit_id(parent_ids=[], snapshot_id=snap, message="Add verse", committed_at_iso=iso, author="gabriel")
585 id_as_linus = compute_commit_id(parent_ids=[], snapshot_id=snap, message="Add verse", committed_at_iso=iso, author="linus")
586 assert id_as_gabriel != id_as_linus
587
588 def test_forged_commit_bypasses_verify_when_no_signature(
589 self, tmp_path: pathlib.Path
590 ) -> None:
591 """A commit with no signature is not signature-checked."""
592 root = _make_repo(tmp_path)
593 snap_id = _EMPTY_SNAP_ID
594 _write_snapshot(root, snap_id)
595
596 commit_id = _make_real_commit_id(snapshot_id=snap_id, message="no-sig")
597 _write_commit(
598 root, commit_id, snapshot_id=snap_id,
599 message="no-sig",
600 agent_id="bot",
601 signature="", # no signature → no check
602 )
603 _set_branch_ref(root, "main", commit_id)
604
605 result = run_verify(root, check_objects=False)
606 assert result["signatures_checked"] == 0
607
608 def test_verify_json_output_includes_signatures_checked(
609 self, tmp_path: pathlib.Path
610 ) -> None:
611 root = _make_repo(tmp_path)
612 snap_id = _EMPTY_SNAP_ID
613 _write_snapshot(root, snap_id)
614 commit_id = _make_real_commit_id(snapshot_id=snap_id, message="json-check")
615 _write_commit(root, commit_id, snapshot_id=snap_id, message="json-check")
616 _set_branch_ref(root, "main", commit_id)
617
618 result = run_verify(root, check_objects=False)
619 as_json = json.dumps(dict(result))
620 parsed = json.loads(as_json)
621 assert "signatures_checked" in parsed
622 assert parsed["signatures_checked"] == 0
623
624
625 # ===========================================================================
626 # Fuzzing
627 # ===========================================================================
628
629
630 class TestFuzzedImpersonationPayloads:
631
632 @pytest.mark.parametrize("seed", range(15))
633 def test_random_control_char_in_author_stripped(self, seed: int) -> None:
634 import random
635 rng = random.Random(seed)
636 char = chr(rng.randint(0x00, 0x1F))
637 payload = f"Author {char} Name"
638 result = sanitize_provenance(payload)
639 assert char not in result
640
641 @pytest.mark.parametrize("seed", range(5))
642 def test_random_forged_ed25519_signature_always_fails(self, seed: int) -> None:
643 """A randomly generated 88-char base64url string is never a valid Ed25519 sig."""
644 import random
645 rng = random.Random(seed + 300)
646 key = _gen_key()
647 payload = provenance_payload(_fake_commit_id(f"fuzz-{seed}"))
648 # Generate random 64 bytes, encode as base64url (same length as a real sig)
649 random_bytes = bytes(rng.randint(0, 255) for _ in range(64))
650 forged = b64url_encode(random_bytes)
651 real_sig = sign_commit_ed25519(payload, key)
652 if forged != real_sig:
653 assert not verify_commit_ed25519(payload, forged, _pub_bytes(key))
654
655
656 # ===========================================================================
657 # provenance_payload — unit tests
658 # ===========================================================================
659
660
661 class TestProvenancePayload:
662 """Unit tests for :func:`provenance_payload`."""
663
664 def test_is_64_hex_chars(self) -> None:
665 p = provenance_payload("c" * 64)
666 assert len(p) == 64
667 assert all(c in "0123456789abcdef" for c in p)
668
669 def test_deterministic(self) -> None:
670 p1 = provenance_payload("cid", author="alice", agent_id="bot")
671 p2 = provenance_payload("cid", author="alice", agent_id="bot")
672 assert p1 == p2
673
674 def test_different_commit_id_different_payload(self) -> None:
675 p1 = provenance_payload("aaa", author="alice", agent_id="bot")
676 p2 = provenance_payload("bbb", author="alice", agent_id="bot")
677 assert p1 != p2
678
679 def test_different_author_different_payload(self) -> None:
680 p1 = provenance_payload("cid", author="alice", agent_id="bot")
681 p2 = provenance_payload("cid", author="linus", agent_id="bot")
682 assert p1 != p2
683
684 def test_different_agent_id_different_payload(self) -> None:
685 p1 = provenance_payload("cid", author="alice", agent_id="bot-v1")
686 p2 = provenance_payload("cid", author="alice", agent_id="bot-v2")
687 assert p1 != p2
688
689 def test_different_model_id_different_payload(self) -> None:
690 p1 = provenance_payload("cid", agent_id="bot", model_id="claude-3")
691 p2 = provenance_payload("cid", agent_id="bot", model_id="gpt-4")
692 assert p1 != p2
693
694 def test_different_toolchain_id_different_payload(self) -> None:
695 p1 = provenance_payload("cid", agent_id="bot", toolchain_id="cursor-v1")
696 p2 = provenance_payload("cid", agent_id="bot", toolchain_id="cursor-v2")
697 assert p1 != p2
698
699 def test_different_prompt_hash_different_payload(self) -> None:
700 p1 = provenance_payload("cid", prompt_hash="aa" * 32)
701 p2 = provenance_payload("cid", prompt_hash="bb" * 32)
702 assert p1 != p2
703
704 def test_separator_injection_consistent(self) -> None:
705 """Null-byte separator: payload is consistent for same raw bytes."""
706 p1 = provenance_payload("cid", author="a\x00b", agent_id="")
707 p2 = provenance_payload("cid", author="a\x00b", agent_id="")
708 assert p1 == p2 # same inputs → same output
709
710 def test_empty_fields_produce_valid_payload(self) -> None:
711 p = provenance_payload("c" * 64)
712 assert len(p) == 64
713
714 def test_not_same_as_bare_commit_id_hash(self) -> None:
715 """provenance_payload ≠ fake_id(commit_id) — it binds more fields."""
716 cid = "d" * 64
717 bare = fake_id(cid)
718 prov = provenance_payload(cid)
719 assert prov != bare
720
721 def test_v7_author_mutation_detected(self) -> None:
722 """Ed25519 (v7): mutating author makes the signature invalid."""
723 key = _gen_key()
724 commit_id = _fake_commit_id("v7-author-mutation")
725
726 original_payload = provenance_payload(commit_id, author="gabriel", agent_id="cursor-bot")
727 sig = sign_commit_ed25519(original_payload, key)
728
729 mutated_payload = provenance_payload(commit_id, author="[email protected]", agent_id="cursor-bot")
730 assert not verify_commit_ed25519(mutated_payload, sig, _pub_bytes(key))
731
732 def test_v7_agent_id_mutation_detected(self) -> None:
733 key = _gen_key()
734 commit_id = _fake_commit_id("v7-agent-mutation")
735 original = provenance_payload(commit_id, author="alice", agent_id="real-bot")
736 sig = sign_commit_ed25519(original, key)
737 mutated = provenance_payload(commit_id, author="alice", agent_id="fake-bot")
738 assert not verify_commit_ed25519(mutated, sig, _pub_bytes(key))
739
740 def test_v7_model_id_mutation_detected(self) -> None:
741 key = _gen_key()
742 commit_id = _fake_commit_id("v7-model-mutation")
743 original = provenance_payload(commit_id, agent_id="bot", model_id="claude-3")
744 sig = sign_commit_ed25519(original, key)
745 mutated = provenance_payload(commit_id, agent_id="bot", model_id="gpt-4")
746 assert not verify_commit_ed25519(mutated, sig, _pub_bytes(key))
747
748 def test_v7_toolchain_mutation_detected(self) -> None:
749 key = _gen_key()
750 commit_id = _fake_commit_id("v7-toolchain-mutation")
751 original = provenance_payload(commit_id, agent_id="bot", toolchain_id="cursor-v1")
752 sig = sign_commit_ed25519(original, key)
753 mutated = provenance_payload(commit_id, agent_id="bot", toolchain_id="cursor-v2")
754 assert not verify_commit_ed25519(mutated, sig, _pub_bytes(key))
755
756 def test_sign_commit_record_uses_provenance_payload(self) -> None:
757 """sign_commit_record signs provenance_payload, not bare commit_id."""
758 key = _gen_key()
759 commit_id = _fake_commit_id("sign-record-test")
760 result = sign_commit_record(commit_id, "my-bot", key, author="alice", model_id="claude-opus")
761 assert result is not None
762 sig, pub_b64, _ = result
763
764 _, pub_bytes = decode_pubkey(pub_b64)
765 expected_payload = provenance_payload(commit_id, author="alice", agent_id="my-bot", model_id="claude-opus")
766 assert verify_commit_ed25519(expected_payload, sig, pub_bytes)
767
768 # Bare commit_id must NOT pass — Ed25519 signed a different payload.
769 bare_payload = provenance_payload(commit_id)
770 assert not verify_commit_ed25519(bare_payload, sig, pub_bytes)
771
772
773 # ===========================================================================
774 # muse verify — v7 provenance payload verification
775 # ===========================================================================
776
777
778 class TestVerifySignaturesV7:
779 """run_verify must verify v7 commits against provenance_payload (Ed25519)."""
780
781 def test_v7_valid_signature_passes(self, tmp_path: pathlib.Path) -> None:
782 root = _make_repo(tmp_path)
783 _write_snapshot(root, _EMPTY_SNAP_ID)
784
785 key = _gen_key()
786 pub_b64 = _pub_b64(key)
787 commit_id = _make_real_commit_id(snapshot_id=_EMPTY_SNAP_ID, message="v7-valid", author="alice", signer_public_key=pub_b64)
788 sig = _v7_sig(commit_id, key, author="alice", agent_id="v7-bot")
789
790 _write_commit(
791 root, commit_id,
792 snapshot_id=_EMPTY_SNAP_ID,
793 message="v7-valid",
794 author="alice",
795 agent_id="v7-bot",
796 signature=sig,
797 signer_public_key=pub_b64,
798 signer_key_id=_signer_key_id(_pub_bytes(key)),
799 )
800 _set_branch_ref(root, "main", commit_id)
801
802 result = run_verify(root, check_objects=False)
803 assert result["signatures_checked"] == 1
804 assert result["all_ok"], result["failures"]
805
806 def test_v7_author_mutation_detected_by_verify(
807 self, tmp_path: pathlib.Path
808 ) -> None:
809 """After mutating author on disk, run_verify must report a signature failure."""
810 root = _make_repo(tmp_path)
811 _write_snapshot(root, _EMPTY_SNAP_ID)
812
813 key = _gen_key()
814 pub_b64 = _pub_b64(key)
815 commit_id = _make_real_commit_id(snapshot_id=_EMPTY_SNAP_ID, message="v7-author-tamper", author="gabriel", signer_public_key=pub_b64)
816 sig = _v7_sig(commit_id, key, author="gabriel", agent_id="v7-bot")
817
818 _write_commit(
819 root, commit_id, snapshot_id=_EMPTY_SNAP_ID, message="v7-author-tamper",
820 author="gabriel", agent_id="v7-bot", signature=sig,
821 signer_public_key=pub_b64,
822 signer_key_id=_signer_key_id(_pub_bytes(key)),
823 )
824 _set_branch_ref(root, "main", commit_id)
825
826 # Tamper: overwrite with a different author.
827 _write_commit(
828 root, commit_id, snapshot_id=_EMPTY_SNAP_ID, message="v7-author-tamper",
829 author="[email protected]", # mutated
830 agent_id="v7-bot", signature=sig,
831 signer_public_key=pub_b64,
832 signer_key_id=_signer_key_id(_pub_bytes(key)),
833 )
834
835 result = run_verify(root, check_objects=False)
836 # With v2 formula, author is in the commit ID hash — mutation makes the
837 # commit unreadable (content-hash verification fails), caught as a
838 # commit-level failure rather than a signature failure.
839 assert not result["all_ok"]
840 commit_failures = [f for f in result["failures"] if f["kind"] == "commit"]
841 assert len(commit_failures) == 1
842
843 def test_v7_agent_id_mutation_detected_by_verify(
844 self, tmp_path: pathlib.Path
845 ) -> None:
846 """Mutating agent_id in a v7 commit is caught by run_verify."""
847 root = _make_repo(tmp_path)
848 _write_snapshot(root, _EMPTY_SNAP_ID)
849
850 key = _gen_key()
851 pub_b64 = _pub_b64(key)
852 commit_id = _make_real_commit_id(snapshot_id=_EMPTY_SNAP_ID, message="v7-agentid-tamper", author="alice", signer_public_key=pub_b64)
853 sig = _v7_sig(commit_id, key, author="alice", agent_id="real-v7-bot")
854
855 _write_commit(
856 root, commit_id, snapshot_id=_EMPTY_SNAP_ID, message="v7-agentid-tamper",
857 author="alice", agent_id="real-v7-bot", signature=sig,
858 signer_public_key=pub_b64,
859 signer_key_id=_signer_key_id(_pub_bytes(key)),
860 )
861 _set_branch_ref(root, "main", commit_id)
862
863 # Tamper: overwrite with a different agent_id but same sig + pub key.
864 _write_commit(
865 root, commit_id, snapshot_id=_EMPTY_SNAP_ID, message="v7-agentid-tamper",
866 author="alice", agent_id="fake-v7-bot", # mutated
867 signature=sig, signer_public_key=pub_b64,
868 signer_key_id=_signer_key_id(_pub_bytes(key)),
869 )
870
871 result = run_verify(root, check_objects=False)
872 assert not result["all_ok"]
873 sig_failures = [f for f in result["failures"] if f["kind"] == "signature"]
874 assert len(sig_failures) == 1
875 assert "INVALID" in sig_failures[0]["error"]
876
877 def test_v7_forged_signature_detected(self, tmp_path: pathlib.Path) -> None:
878 root = _make_repo(tmp_path)
879 _write_snapshot(root, _EMPTY_SNAP_ID)
880
881 key = _gen_key()
882 pub_b64 = _pub_b64(key)
883 commit_id = _make_real_commit_id(snapshot_id=_EMPTY_SNAP_ID, message="v7-forged", author="alice", signer_public_key=pub_b64)
884 _write_commit(
885 root, commit_id, snapshot_id=_EMPTY_SNAP_ID, message="v7-forged",
886 author="alice", agent_id="v7-forged-bot",
887 signature=encode_sig("ed25519", b"\x00" * 64), # correct format, wrong bytes
888 signer_public_key=pub_b64,
889 signer_key_id=_signer_key_id(_pub_bytes(key)),
890 )
891 _set_branch_ref(root, "main", commit_id)
892
893 result = run_verify(root, check_objects=False)
894 assert result["signatures_checked"] == 1
895 assert not result["all_ok"]
896 assert any("INVALID" in f["error"] for f in result["failures"])
897
898 def test_v7_mixed_chain_verifies(self, tmp_path: pathlib.Path) -> None:
899 """A commit chain mixing signed and unsigned commits verifies correctly."""
900 root = _make_repo(tmp_path)
901 _write_snapshot(root, _EMPTY_SNAP_ID)
902
903 key = _gen_key()
904
905 # Unsigned base commit.
906 base_id = _make_real_commit_id(snapshot_id=_EMPTY_SNAP_ID, message="unsigned-base")
907 _write_commit(root, base_id, snapshot_id=_EMPTY_SNAP_ID, message="unsigned-base")
908
909 # Signed child.
910 pub_b64 = _pub_b64(key)
911 child_id = _make_real_commit_id(snapshot_id=_EMPTY_SNAP_ID, parent=base_id, message="signed-child", author="alice", signer_public_key=pub_b64)
912 sig = _v7_sig(child_id, key, author="alice", agent_id="chain-bot")
913 _write_commit(
914 root, child_id, snapshot_id=_EMPTY_SNAP_ID, parent=base_id, message="signed-child",
915 author="alice", agent_id="chain-bot", signature=sig,
916 signer_public_key=pub_b64,
917 signer_key_id=_signer_key_id(_pub_bytes(key)),
918 )
919 _set_branch_ref(root, "main", child_id)
920
921 result = run_verify(root, check_objects=False)
922 assert result["signatures_checked"] == 1
923 assert result["all_ok"], result["failures"]
924
925 @pytest.mark.parametrize("field,value", [
926 ("author", "injected-author"),
927 ("model_id", "injected-model"),
928 ("toolchain_id", "injected-toolchain"),
929 ("prompt_hash", "injected-prompt-hash"),
930 ])
931 def test_v7_any_provenance_field_mutation_detected(
932 self,
933 tmp_path: pathlib.Path,
934 field: str,
935 value: str,
936 ) -> None:
937 """Any provenance field mutation in a v7 commit is caught by run_verify."""
938 root = _make_repo(tmp_path)
939 _write_snapshot(root, _EMPTY_SNAP_ID)
940
941 key = _gen_key()
942 pub_b64 = _pub_b64(key)
943 commit_id = _make_real_commit_id(snapshot_id=_EMPTY_SNAP_ID, message=f"v7-{field}", author="original-author", signer_public_key=pub_b64)
944 sig = _v7_sig(
945 commit_id, key,
946 author="original-author", agent_id="prov-bot",
947 model_id="original-model", toolchain_id="original-toolchain",
948 prompt_hash="original-hash",
949 )
950 _write_commit(
951 root, commit_id, snapshot_id=_EMPTY_SNAP_ID, message=f"v7-{field}",
952 author="original-author", agent_id="prov-bot",
953 model_id="original-model", toolchain_id="original-toolchain",
954 prompt_hash="original-hash", signature=sig,
955 signer_public_key=pub_b64,
956 signer_key_id=_signer_key_id(_pub_bytes(key)),
957 )
958 _set_branch_ref(root, "main", commit_id)
959
960 # Tamper: rewrite with the specific field mutated.
961 tampered = {field: value}
962 _write_commit(
963 root, commit_id, snapshot_id=_EMPTY_SNAP_ID, message=f"v7-{field}",
964 author=tampered.get("author", "original-author"),
965 agent_id="prov-bot",
966 model_id=tampered.get("model_id", "original-model"),
967 toolchain_id=tampered.get("toolchain_id", "original-toolchain"),
968 prompt_hash=tampered.get("prompt_hash", "original-hash"),
969 signature=sig, signer_public_key=pub_b64,
970 signer_key_id=_signer_key_id(_pub_bytes(key)),
971 )
972
973 result = run_verify(root, check_objects=False)
974 assert not result["all_ok"], f"Mutation of {field!r} should be detected"
975 if field == "author":
976 # author is in the v2 commit ID hash — mutation makes the commit
977 # unreadable; caught as a content-hash (commit) failure.
978 assert any(f["kind"] == "commit" for f in result["failures"])
979 else:
980 sig_failures = [f for f in result["failures"] if f["kind"] == "signature"]
981 assert len(sig_failures) == 1
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 122 days ago