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