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