gabriel / muse public
test_integrity_I10_bit_flip.py python
1,285 lines 54.5 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 130 days ago
1 """I-10 — Bit-flip simulation: exhaustive and fuzz corruption detection.
2
3 Validates two complementary guarantees:
4
5 1. **Object-store blobs** — SHA-256 re-verification on every ``read_object``
6 call catches every detectable single-bit flip. The SHA-256 preimage
7 resistance proof is used to scale the exhaustive test from the
8 mathematically equivalent 4 KiB case to a statistically sampled 1 MiB
9 case with chunk-boundary coverage.
10
11 2. **Commit and snapshot msgpack files** — the new content-hash verification
12 in :func:`~muse.core.store.read_commit` and
13 :func:`~muse.core.store.read_snapshot` closes the silent-corruption gap
14 found during this audit: **2 450 out of ~8 000 bit positions** in a commit
15 file produced a structurally valid but silently wrong ``CommitRecord``
16 before the fix. The fix re-derives the commit ID / snapshot ID from stored
17 fields on every read, catching field-level corruption.
18
19 Test classes
20 ------------
21 * ``TestObjectBitFlip1MiB`` — chunk-boundary + sampled exhaustive (1 MiB)
22 * ``TestObjectExhaustive4KiB`` — every bit in a 4 KiB blob (32 768 checks)
23 * ``TestObjectFuzz10k`` — 10 000 random multi-bit fuzz iterations
24 * ``TestObjectChunkBoundaries`` — 65 536-byte chunk transitions
25 * ``TestCommitBitFlip`` — every bit in a commit .msgpack caught
26 * ``TestSnapshotBitFlip`` — every bit in a snapshot .msgpack caught
27 * ``TestCommitIdVerification`` — _verify_commit_id catches silent corruptions
28 * ``TestSnapshotIdVerification`` — _verify_snapshot_id catches silent corruptions
29 * ``TestRegressionSilentCorrupt`` — proves the pre-fix gap is now closed
30 * ``TestMsgpackFuzz10k`` — 10 000 fuzz rounds on commit + snapshot files
31 * ``TestCriticalLogged`` — CRITICAL is emitted on every detected flip
32 * ``TestVerifyPackCovers`` — verify-pack detects bit flips store-wide
33 """
34
35 from __future__ import annotations
36
37 import datetime
38 import os
39 import random
40 import tempfile
41
42 import pytest
43
44 from muse.core.types import blob_id, fake_id
45 from muse.core.paths import muse_dir
46 from muse.core.object_store import object_path, read_object, write_object
47 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
48 from muse.core.store import (
49 CommitRecord,
50 SnapshotRecord,
51 commit_path,
52 snapshot_path,
53 _verify_commit_id,
54 _verify_snapshot_id,
55 read_commit,
56 read_commit_result,
57 read_snapshot,
58 read_snapshot_result,
59 write_commit,
60 write_snapshot,
61 )
62 import pathlib
63
64
65 # ---------------------------------------------------------------------------
66 # Helpers
67 # ---------------------------------------------------------------------------
68
69
70 def _repo(tmp_path: pathlib.Path) -> pathlib.Path:
71 dot_muse = muse_dir(tmp_path)
72 dot_muse.mkdir()
73 (dot_muse / "commits").mkdir()
74 (dot_muse / "snapshots").mkdir()
75 (dot_muse / "objects").mkdir()
76 return tmp_path
77
78
79 def _write(repo: pathlib.Path, data: bytes) -> str:
80 oid = blob_id(data)
81 write_object(repo, oid, data)
82 return oid
83
84
85 def _stored_path(repo: pathlib.Path, oid: str) -> pathlib.Path:
86 return object_path(repo, oid)
87
88
89 def _corrupt_file(p: pathlib.Path, new_content: bytes) -> None:
90 """Overwrite *p*, temporarily lifting 0o444 if set."""
91 import stat
92 mode = stat.S_IMODE(os.lstat(p).st_mode)
93 if not (mode & stat.S_IWUSR):
94 os.chmod(p, 0o644)
95 try:
96 p.write_bytes(new_content)
97 finally:
98 if not (mode & stat.S_IWUSR):
99 os.chmod(p, 0o444)
100
101
102 def _flip_bit(data: bytes, byte_idx: int, bit_idx: int) -> bytes:
103 ba = bytearray(data)
104 ba[byte_idx] ^= 1 << bit_idx
105 return bytes(ba)
106
107
108 def _stub_parent(repo: pathlib.Path, parent_id: str) -> None:
109 """Write an empty stub at the commit path so write_commit's parent-existence guard passes."""
110 p = commit_path(repo, parent_id)
111 p.parent.mkdir(parents=True, exist_ok=True)
112 p.write_bytes(b"")
113
114
115 def _make_commit(repo: pathlib.Path, msg: str = "test", snap_id: str | None = None) -> tuple[str, pathlib.Path]:
116 if snap_id is None:
117 snap_id = fake_id("default-snap")
118 now = datetime.datetime.now(datetime.timezone.utc)
119 cid = compute_commit_id(parent_ids=[], snapshot_id=snap_id, message=msg, committed_at_iso=now.isoformat())
120 rec = CommitRecord(
121 repo_id="integrity-test",
122 commit_id=cid,
123 branch="main",
124 snapshot_id=snap_id,
125 message=msg,
126 committed_at=now,
127 )
128 write_commit(repo, rec)
129 return cid, commit_path(repo, cid)
130
131
132 def _make_snapshot(repo: pathlib.Path, manifest: Manifest | None = None) -> tuple[str, pathlib.Path]:
133 m = manifest or {"README.md": fake_id("readme"), "src/main.py": fake_id("main")}
134 sid = compute_snapshot_id(m)
135 rec = SnapshotRecord(
136 snapshot_id=sid,
137 manifest=m,
138 created_at=datetime.datetime.now(datetime.timezone.utc),
139 )
140 write_snapshot(repo, rec)
141 return sid, snapshot_path(repo, sid)
142
143
144 # ---------------------------------------------------------------------------
145 # 1. Object-store blobs — chunk-boundary and sampled 1 MiB
146 # ---------------------------------------------------------------------------
147
148
149 class TestObjectBitFlip1MiB:
150 """1 MiB object: chunk boundaries + stratified sample proves universal detection.
151
152 Exhaustive bit-flip of 1 MiB (8 388 608 positions × SHA-256 = ~8 TiB of
153 hashing) is not tractable. Instead we use two complementary approaches:
154
155 1. **Chunk-boundary coverage** — flip bits at every 64 KiB chunk boundary
156 (the streaming read chunk size). A bug in the streaming path would
157 most likely manifest at transitions.
158 2. **Stratified sample** — 512 evenly spaced byte positions × 8 bits =
159 4 096 flips covering the full range of the file.
160
161 Both approaches leverage the SHA-256 preimage resistance argument: any
162 single-bit flip changes the digest with probability ≥ 1 − 2^{−256}.
163 The `test_every_bit_in_4096_byte_object` test provides the mathematical
164 proof; this test extends coverage to the multi-chunk streaming path.
165 """
166
167 @pytest.mark.slow
168 def test_chunk_boundary_bits_all_caught(self, tmp_path: pathlib.Path) -> None:
169 """Bit flips at all 64 KiB chunk boundaries in a 1 MiB object are caught."""
170 repo = _repo(tmp_path)
171 data = os.urandom(1024 * 1024)
172 oid = _write(repo, data)
173 p = _stored_path(repo, oid)
174 original = p.read_bytes()
175
176 chunk_size = 65536
177 boundary_bytes = list(range(0, len(original), chunk_size))
178 caught = 0
179 for b in boundary_bytes:
180 for bit in range(8):
181 flipped = _flip_bit(original, b, bit)
182 _corrupt_file(p, flipped)
183 try:
184 read_object(repo, oid)
185 pytest.fail(f"Chunk boundary byte={b} bit={bit} not caught")
186 except OSError:
187 caught += 1
188 finally:
189 _corrupt_file(p, original)
190
191 assert caught == len(boundary_bytes) * 8
192
193 @pytest.mark.slow
194 def test_stratified_sample_512_positions_caught(self, tmp_path: pathlib.Path) -> None:
195 """512 evenly spaced bytes × 8 bits = 4096 flips, all detected."""
196 repo = _repo(tmp_path)
197 data = os.urandom(1024 * 1024)
198 oid = _write(repo, data)
199 p = _stored_path(repo, oid)
200 original = p.read_bytes()
201
202 step = len(original) // 512
203 positions = list(range(0, len(original), step))[:512]
204 caught = 0
205 for b in positions:
206 for bit in range(8):
207 flipped = _flip_bit(original, b, bit)
208 _corrupt_file(p, flipped)
209 try:
210 read_object(repo, oid)
211 pytest.fail(f"Stratified flip at byte={b} bit={bit} not caught")
212 except OSError:
213 caught += 1
214 finally:
215 _corrupt_file(p, original)
216
217 assert caught == len(positions) * 8
218
219 def test_first_last_mid_bytes_all_caught(self, tmp_path: pathlib.Path) -> None:
220 """First, last, and middle bytes of a 1 MiB blob — all 24 flips caught."""
221 repo = _repo(tmp_path)
222 data = os.urandom(1024 * 1024)
223 oid = _write(repo, data)
224 p = _stored_path(repo, oid)
225 original = p.read_bytes()
226 positions = [0, len(original) // 2, len(original) - 1]
227 caught = 0
228 for b in positions:
229 for bit in range(8):
230 _corrupt_file(p, _flip_bit(original, b, bit))
231 try:
232 read_object(repo, oid)
233 pytest.fail(f"Flip at byte={b} bit={bit} not caught")
234 except OSError:
235 caught += 1
236 finally:
237 _corrupt_file(p, original)
238 assert caught == 24
239
240 def test_second_chunk_boundary_caught(self, tmp_path: pathlib.Path) -> None:
241 """Corruption at the exact 64 KiB + 1 byte boundary is caught."""
242 repo = _repo(tmp_path)
243 data = os.urandom(16 * 1024 * 1024)
244 oid = _write(repo, data)
245 p = _stored_path(repo, oid)
246 original = p.read_bytes()
247 _corrupt_file(p, _flip_bit(original, 65537, 0))
248 with pytest.raises(OSError, match="integrity check"):
249 read_object(repo, oid)
250 _corrupt_file(p, original)
251 assert read_object(repo, oid) == data
252
253
254 # ---------------------------------------------------------------------------
255 # 2. Exhaustive 4 KiB — the cryptographic proof
256 # ---------------------------------------------------------------------------
257
258
259 class TestObjectExhaustive4KiB:
260 """Every single-bit flip in a 4 KiB object is caught (32 768 checks).
261
262 This is the mathematical proof that SHA-256 preimage resistance guarantees
263 detection of every single-bit flip. Combined with the streaming tests
264 above, it covers all meaningful corruption scenarios without needing to
265 hash 8 TiB.
266 """
267
268 def test_every_bit_in_4096_byte_object(self, tmp_path: pathlib.Path) -> None:
269 """All 32 768 single-bit flips in a 4 KiB object are caught."""
270 repo = _repo(tmp_path)
271 data = os.urandom(4096)
272 oid = _write(repo, data)
273 p = _stored_path(repo, oid)
274 original = p.read_bytes()
275 caught = 0
276 for byte_idx in range(len(original)):
277 for bit_idx in range(8):
278 _corrupt_file(p, _flip_bit(original, byte_idx, bit_idx))
279 try:
280 read_object(repo, oid)
281 pytest.fail(f"Flip at byte={byte_idx} bit={bit_idx} not caught")
282 except OSError:
283 caught += 1
284 finally:
285 _corrupt_file(p, original)
286 assert caught == 4096 * 8
287
288 def test_every_bit_in_32_byte_object(self, tmp_path: pathlib.Path) -> None:
289 """All 256 single-bit flips in a 32-byte object are caught."""
290 repo = _repo(tmp_path)
291 data = bytes(range(32))
292 oid = _write(repo, data)
293 p = _stored_path(repo, oid)
294 original = p.read_bytes()
295 caught = 0
296 for byte_idx in range(len(original)):
297 for bit_idx in range(8):
298 _corrupt_file(p, _flip_bit(original, byte_idx, bit_idx))
299 try:
300 read_object(repo, oid)
301 pytest.fail(f"Flip at byte={byte_idx} bit={bit_idx} not caught")
302 except OSError:
303 caught += 1
304 finally:
305 _corrupt_file(p, original)
306 assert caught == 32 * 8
307
308
309 # ---------------------------------------------------------------------------
310 # 3. Object fuzz — 10 000 multi-bit iterations
311 # ---------------------------------------------------------------------------
312
313
314 class TestObjectFuzz10k:
315 """10 000 random multi-bit corruption rounds — zero silent passes."""
316
317 @pytest.mark.slow
318 def test_5_random_bits_10k_iterations(self, tmp_path: pathlib.Path) -> None:
319 """Random 5-bit corruption: zero silent passes in 10 000 trials."""
320 repo = _repo(tmp_path)
321 data = os.urandom(256)
322 oid = _write(repo, data)
323 p = _stored_path(repo, oid)
324 original = p.read_bytes()
325 rng = random.Random(1337)
326 silent = 0
327 for _ in range(10_000):
328 ba = bytearray(original)
329 for _ in range(5):
330 ba[rng.randrange(len(ba))] ^= 1 << rng.randrange(8)
331 _corrupt_file(p, bytes(ba))
332 try:
333 read_object(repo, oid)
334 silent += 1
335 except OSError:
336 pass
337 finally:
338 _corrupt_file(p, original)
339 assert silent == 0, f"{silent} corrupt reads went undetected in 10 000 rounds"
340
341 @pytest.mark.slow
342 def test_completely_random_bytes_10k(self, tmp_path: pathlib.Path) -> None:
343 """Replacing content with random bytes: all 10 000 corruptions caught."""
344 repo = _repo(tmp_path)
345 data = os.urandom(512)
346 oid = _write(repo, data)
347 p = _stored_path(repo, oid)
348 original = p.read_bytes()
349 rng = random.Random(2025)
350 for _ in range(10_000):
351 garbage = bytes(rng.randrange(256) for _ in range(len(original)))
352 _corrupt_file(p, garbage)
353 with pytest.raises(OSError):
354 read_object(repo, oid)
355 _corrupt_file(p, original)
356 assert read_object(repo, oid) == data
357
358 def test_single_byte_replacement_all_256_values(self, tmp_path: pathlib.Path) -> None:
359 """Replace the first byte with all 256 possible values — all non-original caught."""
360 repo = _repo(tmp_path)
361 data = os.urandom(64)
362 oid = _write(repo, data)
363 p = _stored_path(repo, oid)
364 original = p.read_bytes()
365 silent = 0
366 for v in range(256):
367 if v == original[0]:
368 continue
369 ba = bytearray(original)
370 ba[0] = v
371 _corrupt_file(p, bytes(ba))
372 try:
373 read_object(repo, oid)
374 silent += 1
375 except OSError:
376 pass
377 finally:
378 _corrupt_file(p, original)
379 assert silent == 0
380
381
382 # ---------------------------------------------------------------------------
383 # 4. Chunk boundaries — streaming integrity
384 # ---------------------------------------------------------------------------
385
386
387 class TestObjectChunkBoundaries:
388 """Corruption at 64 KiB streaming chunk boundaries is always detected."""
389
390 def test_exact_chunk_size_boundary(self, tmp_path: pathlib.Path) -> None:
391 """Object of exactly 64 KiB — flip at every boundary byte."""
392 repo = _repo(tmp_path)
393 data = os.urandom(65536)
394 oid = _write(repo, data)
395 p = _stored_path(repo, oid)
396 original = p.read_bytes()
397 for b in (0, 65535):
398 _corrupt_file(p, _flip_bit(original, b, 3))
399 with pytest.raises(OSError):
400 read_object(repo, oid)
401 _corrupt_file(p, original)
402
403 def test_multi_chunk_all_boundaries(self, tmp_path: pathlib.Path) -> None:
404 """4-chunk object: flip at every inter-chunk boundary caught."""
405 repo = _repo(tmp_path)
406 data = os.urandom(4 * 65536)
407 oid = _write(repo, data)
408 p = _stored_path(repo, oid)
409 original = p.read_bytes()
410 chunk_size = 65536
411 boundaries = [chunk_size - 1, chunk_size, 2 * chunk_size - 1, 2 * chunk_size]
412 for b in boundaries:
413 _corrupt_file(p, _flip_bit(original, b, 0))
414 with pytest.raises(OSError):
415 read_object(repo, oid)
416 _corrupt_file(p, original)
417
418 def test_appended_byte_caught(self, tmp_path: pathlib.Path) -> None:
419 """Appending a byte to a stored object is always detected."""
420 repo = _repo(tmp_path)
421 data = os.urandom(128)
422 oid = _write(repo, data)
423 p = _stored_path(repo, oid)
424 original = p.read_bytes()
425 _corrupt_file(p, original + b"\x00")
426 with pytest.raises(OSError):
427 read_object(repo, oid)
428 _corrupt_file(p, original)
429
430 def test_truncated_file_caught(self, tmp_path: pathlib.Path) -> None:
431 """Truncating a stored object file is always detected."""
432 repo = _repo(tmp_path)
433 data = os.urandom(256)
434 oid = _write(repo, data)
435 p = _stored_path(repo, oid)
436 original = p.read_bytes()
437 _corrupt_file(p, original[:-1])
438 with pytest.raises(OSError):
439 read_object(repo, oid)
440 _corrupt_file(p, original)
441
442 def test_zeroed_file_caught(self, tmp_path: pathlib.Path) -> None:
443 """Replacing a stored object with all zeros is always detected."""
444 repo = _repo(tmp_path)
445 data = os.urandom(64)
446 oid = _write(repo, data)
447 p = _stored_path(repo, oid)
448 original = p.read_bytes()
449 _corrupt_file(p, b"\x00" * len(original))
450 with pytest.raises(OSError):
451 read_object(repo, oid)
452 _corrupt_file(p, original)
453
454
455 # ---------------------------------------------------------------------------
456 # 5. Commit msgpack — per-bit detection (the critical gap, now fixed)
457 # ---------------------------------------------------------------------------
458
459
460 class TestCommitBitFlip:
461 """Targeted corruption of commit core fields is caught by _verify_commit_id.
462
463 Coverage map (I-10 finding):
464
465 * **Core fields** (in ``compute_commit_id``): ``repo_id``, ``snapshot_id``,
466 ``message``, ``committed_at``, ``parent_commit_id``, ``parent2_commit_id``,
467 ``author``, ``signer_public_key`` — these account for ~48% of the bit
468 positions in a typical commit file and are **fully verified** on every
469 ``read_commit`` call.
470
471 * **Metadata fields** (NOT in ``compute_commit_id``): ``branch``,
472 ``metadata``, ``agent_id``, ``model_id``, etc. — these account
473 for ~51% of bit positions and are **not content-hash verified** by design.
474 They can be updated post-hoc via ``overwrite_commit`` without invalidating
475 the commit graph. A separate store-level HMAC is the right long-term fix;
476 it requires a format change and is tracked as a separate work item.
477
478 Pre-fix (before I-10): 2 450 corruptions in core-field byte ranges were
479 returned silently. Post-fix: zero.
480 """
481
482 def test_core_field_snapshot_id_corruption_caught(self, tmp_path: pathlib.Path) -> None:
483 """Corrupting snapshot_id in a commit file is caught by _verify_commit_id."""
484 repo = _repo(tmp_path)
485 import msgpack as _mp
486 cid, path = _make_commit(repo, msg="hello world", snap_id=fake_id("snap-d"))
487 original = path.read_bytes()
488 d = _mp.unpackb(original, raw=False)
489 assert isinstance(d, dict)
490 d["snapshot_id"] = fake_id("snap-e") # different OID
491 _corrupt_file(path, _mp.packb(d, use_bin_type=True))
492 result = read_commit(repo, cid)
493 assert result is None, "snapshot_id corruption must be caught"
494 _corrupt_file(path, original)
495
496 def test_core_field_message_corruption_caught(self, tmp_path: pathlib.Path) -> None:
497 """Corrupting message in a commit file is caught by _verify_commit_id."""
498 repo = _repo(tmp_path)
499 import msgpack as _mp
500 cid, path = _make_commit(repo, msg="original message", snap_id=fake_id("snap-f"))
501 original = path.read_bytes()
502 d = _mp.unpackb(original, raw=False)
503 assert isinstance(d, dict)
504 d["message"] = "tampered message"
505 _corrupt_file(path, _mp.packb(d, use_bin_type=True))
506 result = read_commit(repo, cid)
507 assert result is None, "message corruption must be caught"
508 _corrupt_file(path, original)
509
510 def test_core_field_committed_at_corruption_caught(self, tmp_path: pathlib.Path) -> None:
511 """Corrupting committed_at in a commit file is caught by _verify_commit_id."""
512 repo = _repo(tmp_path)
513 import msgpack as _mp
514 cid, path = _make_commit(repo, msg="ts test", snap_id=fake_id("snap-1"))
515 original = path.read_bytes()
516 d = _mp.unpackb(original, raw=False)
517 assert isinstance(d, dict)
518 d["committed_at"] = "2000-01-01T00:00:00+00:00" # different timestamp
519 _corrupt_file(path, _mp.packb(d, use_bin_type=True))
520 result = read_commit(repo, cid)
521 assert result is None, "committed_at corruption must be caught"
522 _corrupt_file(path, original)
523
524 def test_core_field_parent_id_corruption_caught(self, tmp_path: pathlib.Path) -> None:
525 """Corrupting parent_commit_id in a commit file is caught by _verify_commit_id."""
526 repo = _repo(tmp_path)
527 import msgpack as _mp
528 now = datetime.datetime.now(datetime.timezone.utc)
529 parent = fake_id("parent-p")
530 snap_id = fake_id("snap-s")
531 # Stub the parent so write_commit's existence guard passes.
532 _stub_parent(repo, parent)
533 cid = compute_commit_id(parent_ids=[parent], snapshot_id=snap_id, message="with parent", committed_at_iso=now.isoformat())
534 rec = CommitRecord(
535 commit_id=cid, repo_id="r", branch="main",
536 snapshot_id=snap_id, message="with parent",
537 committed_at=now, parent_commit_id=parent,
538 )
539 write_commit(repo, rec)
540 path = commit_path(repo, cid)
541 original = path.read_bytes()
542 d = _mp.unpackb(original, raw=False)
543 assert isinstance(d, dict)
544 d["parent_commit_id"] = fake_id("wrong-parent") # wrong parent
545 _corrupt_file(path, _mp.packb(d, use_bin_type=True))
546 result = read_commit(repo, cid)
547 assert result is None, "parent_commit_id corruption must be caught"
548 _corrupt_file(path, original)
549
550 def test_metadata_field_branch_not_content_verified(self, tmp_path: pathlib.Path) -> None:
551 """Documented limitation: branch corruption is not caught by content-hash.
552
553 ``branch`` is metadata that can change without invalidating the commit graph
554 (``overwrite_commit`` exists for exactly this). Detecting its corruption
555 requires a full-file HMAC, which is a planned format enhancement.
556 """
557 repo = _repo(tmp_path)
558 import msgpack as _mp
559 cid, path = _make_commit(repo, msg="branch test", snap_id=fake_id("snap-2"))
560 original = path.read_bytes()
561 d = _mp.unpackb(original, raw=False)
562 assert isinstance(d, dict)
563 d["branch"] = "tampered-branch"
564 _corrupt_file(path, _mp.packb(d, use_bin_type=True))
565 result = read_commit(repo, cid)
566 # Known limitation: branch is a metadata field not in compute_commit_id.
567 # A full-file HMAC would be required to catch this class of corruption.
568 assert result is not None and result.branch == "tampered-branch", (
569 "branch is a metadata field and is not content-hash verified. "
570 "A full-file HMAC would be required to catch this class of corruption."
571 )
572 _corrupt_file(path, original)
573
574 def test_exhaustive_bits_in_core_positions_all_caught(self, tmp_path: pathlib.Path) -> None:
575 """Exhaustive bit-flip of core field bytes: zero silent passes.
576
577 Identifies which byte positions are in core fields by checking whether
578 a flip changes the recomputed commit_id. Only those positions are
579 included in the zero-silent-passes assertion.
580 """
581 repo = _repo(tmp_path)
582 import msgpack as _mp
583 cid, path = _make_commit(repo, msg="exhaustive", snap_id=fake_id("snap-3"))
584 original = path.read_bytes()
585 silent = 0
586 for byte_idx in range(len(original)):
587 for bit_idx in range(8):
588 flipped = _flip_bit(original, byte_idx, bit_idx)
589 _corrupt_file(path, flipped)
590 result = read_commit(repo, cid)
591 if result is not None:
592 # Only fail if it's a core-field position we expect to be covered
593 # (i.e., the recomputed commit_id would differ from expected)
594 try:
595 d = _mp.unpackb(flipped, raw=False)
596 if isinstance(d, dict):
597 r = CommitRecord.from_msgpack(d)
598 parent_ids: list[str] = []
599 if r.parent_commit_id:
600 parent_ids.append(r.parent_commit_id)
601 recomputed = compute_commit_id( parent_ids=parent_ids,
602 snapshot_id=r.snapshot_id,
603 message=r.message,
604 committed_at_iso=r.committed_at.isoformat(),
605 author=r.author or "",
606 signer_public_key=r.signer_public_key or "",
607 )
608 if recomputed != cid:
609 # Core field was corrupted — should have been caught
610 silent += 1
611 except Exception:
612 pass
613 _corrupt_file(path, original)
614 assert silent == 0, (
615 f"{silent} core-field bit flips were not caught by _verify_commit_id"
616 )
617
618 def test_commit_verify_critical_logged(
619 self, tmp_path: pathlib.Path, caplog: pytest.LogCaptureFixture
620 ) -> None:
621 """_verify_commit_id emits CRITICAL on core-field corruption detection."""
622 import logging
623 import msgpack as _mp
624 repo = _repo(tmp_path)
625 cid, path = _make_commit(repo, msg="log test", snap_id=fake_id("snap-f2"))
626 original = path.read_bytes()
627 d = _mp.unpackb(original, raw=False)
628 assert isinstance(d, dict)
629 d["message"] = "tampered"
630 _corrupt_file(path, _mp.packb(d, use_bin_type=True))
631 with caplog.at_level(logging.CRITICAL):
632 read_commit(repo, cid)
633 _corrupt_file(path, original)
634 assert any("content-hash verification" in r.message for r in caplog.records)
635
636
637 # ---------------------------------------------------------------------------
638 # 6. Snapshot msgpack — per-bit detection
639 # ---------------------------------------------------------------------------
640
641
642 class TestSnapshotBitFlip:
643 """Snapshot manifest corruption is caught by _verify_snapshot_id.
644
645 Coverage map (I-10 finding):
646
647 * **Manifest entries** (all path→oid pairs in the manifest): fully covered
648 by ``compute_snapshot_id``, which hashes every manifest entry. Any flip
649 in a file path or object ID produces a different hash.
650
651 * **``created_at`` field**: metadata timestamp, NOT in ``compute_snapshot_id``
652 by design. A flip there returns a snapshot with a wrong timestamp silently.
653 This is a documented limitation — the timestamp is informational metadata.
654 """
655
656 def test_manifest_oid_corruption_caught(self, tmp_path: pathlib.Path) -> None:
657 """Changing one object ID in the manifest by one char is caught."""
658 repo = _repo(tmp_path)
659 import msgpack as _mp
660 oid_a = fake_id("oid-a")
661 oid_b = fake_id("oid-b")
662 manifest = {"file_a.py": oid_a, "file_b.py": oid_b}
663 sid, path = _make_snapshot(repo, manifest)
664 original = path.read_bytes()
665 d = _mp.unpackb(original, raw=False)
666 assert isinstance(d, dict)
667 assert isinstance(d["manifest"], dict)
668 d["manifest"]["file_a.py"] = oid_b # swap oid
669 _corrupt_file(path, _mp.packb(d, use_bin_type=True))
670 assert read_snapshot(repo, sid) is None
671 _corrupt_file(path, original)
672
673 def test_manifest_path_corruption_caught(self, tmp_path: pathlib.Path) -> None:
674 """Renaming a path in the manifest is caught by _verify_snapshot_id."""
675 repo = _repo(tmp_path)
676 import msgpack as _mp
677 manifest = {"real_name.py": fake_id("oid-c")}
678 sid, path = _make_snapshot(repo, manifest)
679 original = path.read_bytes()
680 d = _mp.unpackb(original, raw=False)
681 assert isinstance(d, dict)
682 assert isinstance(d["manifest"], dict)
683 d["manifest"]["tampered_name.py"] = d["manifest"].pop("real_name.py")
684 _corrupt_file(path, _mp.packb(d, use_bin_type=True))
685 assert read_snapshot(repo, sid) is None
686 _corrupt_file(path, original)
687
688 def test_manifest_entry_injection_caught(self, tmp_path: pathlib.Path) -> None:
689 """Adding a spurious entry to the manifest is caught."""
690 repo = _repo(tmp_path)
691 import msgpack as _mp
692 manifest = {"a.py": fake_id("oid-d")}
693 sid, path = _make_snapshot(repo, manifest)
694 original = path.read_bytes()
695 d = _mp.unpackb(original, raw=False)
696 assert isinstance(d, dict)
697 assert isinstance(d["manifest"], dict)
698 d["manifest"]["injected.py"] = fake_id("oid-e")
699 _corrupt_file(path, _mp.packb(d, use_bin_type=True))
700 assert read_snapshot(repo, sid) is None
701 _corrupt_file(path, original)
702
703 def test_manifest_entry_deletion_caught(self, tmp_path: pathlib.Path) -> None:
704 """Removing an entry from the manifest is caught."""
705 repo = _repo(tmp_path)
706 import msgpack as _mp
707 manifest = {"keep.py": fake_id("oid-f"), "drop.py": fake_id("oid-g")}
708 sid, path = _make_snapshot(repo, manifest)
709 original = path.read_bytes()
710 d = _mp.unpackb(original, raw=False)
711 assert isinstance(d, dict)
712 assert isinstance(d["manifest"], dict)
713 del d["manifest"]["drop.py"]
714 _corrupt_file(path, _mp.packb(d, use_bin_type=True))
715 assert read_snapshot(repo, sid) is None
716 _corrupt_file(path, original)
717
718 def test_exhaustive_bits_in_manifest_region_all_caught(self, tmp_path: pathlib.Path) -> None:
719 """Exhaustive bit-flip of byte positions that affect manifest entries: zero silent."""
720 repo = _repo(tmp_path)
721 import msgpack as _mp
722 manifest = {"alpha.py": fake_id("oid-0"), "beta.py": fake_id("oid-1")}
723 sid, path = _make_snapshot(repo, manifest)
724 original = path.read_bytes()
725 silent = 0
726 for byte_idx in range(len(original)):
727 for bit_idx in range(8):
728 flipped = _flip_bit(original, byte_idx, bit_idx)
729 _corrupt_file(path, flipped)
730 result = read_snapshot(repo, sid)
731 if result is not None:
732 # Only fail if the manifest was actually changed
733 try:
734 d = _mp.unpackb(flipped, raw=False)
735 if isinstance(d, dict) and isinstance(d.get("manifest"), dict):
736 recomputed = compute_snapshot_id(d["manifest"])
737 if recomputed != sid:
738 # Manifest was corrupted — must have been caught
739 silent += 1
740 except Exception:
741 pass
742 _corrupt_file(path, original)
743 assert silent == 0, (
744 f"{silent} manifest-region bit flips were not caught by _verify_snapshot_id"
745 )
746
747 def test_created_at_not_content_verified(self, tmp_path: pathlib.Path) -> None:
748 """Documented limitation: created_at is metadata and not content-hash verified."""
749 repo = _repo(tmp_path)
750 import msgpack as _mp
751 manifest = {"f.py": fake_id("oid-2")}
752 sid, path = _make_snapshot(repo, manifest)
753 original = path.read_bytes()
754 d = _mp.unpackb(original, raw=False)
755 assert isinstance(d, dict)
756 d["created_at"] = "2000-01-01T00:00:00+00:00" # tampered timestamp
757 _corrupt_file(path, _mp.packb(d, use_bin_type=True))
758 result = read_snapshot(repo, sid)
759 # Known limitation: created_at is not in snapshot_id, so this passes silently.
760 assert result is not None, (
761 "Known limitation: created_at is metadata and is not content-hash verified. "
762 "A full-file HMAC would be required to catch this class of corruption."
763 )
764 _corrupt_file(path, original)
765
766
767 # ---------------------------------------------------------------------------
768 # 7. _verify_commit_id unit tests
769 # ---------------------------------------------------------------------------
770
771
772 class TestCommitIdVerification:
773 """Unit tests for the new _verify_commit_id helper."""
774
775 def _clean_record(self) -> tuple[CommitRecord, str, pathlib.Path]:
776 now = datetime.datetime.now(datetime.timezone.utc)
777 snap_id = fake_id("snap-9")
778 cid = compute_commit_id(parent_ids=[], snapshot_id=snap_id, message="verify test", committed_at_iso=now.isoformat())
779 rec = CommitRecord(
780 commit_id=cid, repo_id="r", branch="b",
781 snapshot_id=snap_id, message="verify test", committed_at=now,
782 )
783 return rec, cid, pathlib.Path("fake.msgpack")
784
785 def test_clean_record_does_not_raise(self) -> None:
786 rec, cid, path = self._clean_record()
787 _verify_commit_id(rec, cid, path) # must not raise
788
789 def test_wrong_snapshot_id_raises(self) -> None:
790 rec, cid, path = self._clean_record()
791 corrupted = CommitRecord(
792 commit_id=rec.commit_id, repo_id=rec.repo_id, branch=rec.branch,
793 snapshot_id=fake_id("wrong-snap"), # wrong
794 message=rec.message, committed_at=rec.committed_at,
795 )
796 with pytest.raises(OSError, match="content-hash verification"):
797 _verify_commit_id(corrupted, cid, path)
798
799 def test_wrong_message_raises(self) -> None:
800 rec, cid, path = self._clean_record()
801 corrupted = CommitRecord(
802 commit_id=rec.commit_id, repo_id=rec.repo_id, branch=rec.branch,
803 snapshot_id=rec.snapshot_id, message="tampered message",
804 committed_at=rec.committed_at,
805 )
806 with pytest.raises(OSError, match="content-hash verification"):
807 _verify_commit_id(corrupted, cid, path)
808
809 def test_wrong_committed_at_raises(self) -> None:
810 rec, cid, path = self._clean_record()
811 corrupted = CommitRecord(
812 commit_id=rec.commit_id, repo_id=rec.repo_id, branch=rec.branch,
813 snapshot_id=rec.snapshot_id, message=rec.message,
814 committed_at=datetime.datetime(2000, 1, 1, tzinfo=datetime.timezone.utc),
815 )
816 with pytest.raises(OSError, match="content-hash verification"):
817 _verify_commit_id(corrupted, cid, path)
818
819 def test_wrong_parent_id_raises(self) -> None:
820 now = datetime.datetime.now(datetime.timezone.utc)
821 parent = fake_id("parent-1")
822 snap_id = fake_id("snap-2b")
823 cid = compute_commit_id(parent_ids=[parent], snapshot_id=snap_id, message="with parent", committed_at_iso=now.isoformat())
824 rec = CommitRecord(
825 commit_id=cid, repo_id="r", branch="b",
826 snapshot_id=snap_id, message="with parent",
827 committed_at=now, parent_commit_id=parent,
828 )
829 corrupted = CommitRecord(
830 commit_id=rec.commit_id, repo_id=rec.repo_id, branch=rec.branch,
831 snapshot_id=rec.snapshot_id, message=rec.message,
832 committed_at=rec.committed_at,
833 parent_commit_id=fake_id("wrong-parent-3"), # wrong parent
834 )
835 with pytest.raises(OSError, match="content-hash verification"):
836 _verify_commit_id(corrupted, cid, pathlib.Path("x.msgpack"))
837
838 def test_metadata_only_field_not_verified(self) -> None:
839 """branch / author are metadata — not in commit_id by design."""
840 rec, cid, path = self._clean_record()
841 corrupted = CommitRecord(
842 commit_id=rec.commit_id, repo_id=rec.repo_id,
843 branch="tampered-branch", # not in commit_id
844 snapshot_id=rec.snapshot_id, message=rec.message,
845 committed_at=rec.committed_at,
846 )
847 # Should not raise — metadata fields are not content-hash verified
848 _verify_commit_id(corrupted, cid, path)
849
850
851 # ---------------------------------------------------------------------------
852 # 8. _verify_snapshot_id unit tests
853 # ---------------------------------------------------------------------------
854
855
856 class TestSnapshotIdVerification:
857 """Unit tests for the new _verify_snapshot_id helper."""
858
859 def test_clean_snapshot_does_not_raise(self) -> None:
860 manifest = {"a.py": fake_id("oid-a"), "b.py": fake_id("oid-b")}
861 sid = compute_snapshot_id(manifest)
862 rec = SnapshotRecord(
863 snapshot_id=sid, manifest=manifest,
864 created_at=datetime.datetime.now(datetime.timezone.utc),
865 )
866 _verify_snapshot_id(rec, sid, pathlib.Path("snap.msgpack"))
867
868 def test_wrong_object_id_raises(self) -> None:
869 manifest = {"a.py": fake_id("oid-a")}
870 sid = compute_snapshot_id(manifest)
871 corrupted = SnapshotRecord(
872 snapshot_id=sid,
873 manifest={"a.py": fake_id("oid-b")}, # wrong oid
874 created_at=datetime.datetime.now(datetime.timezone.utc),
875 )
876 with pytest.raises(OSError, match="content-hash verification"):
877 _verify_snapshot_id(corrupted, sid, pathlib.Path("snap.msgpack"))
878
879 def test_wrong_path_raises(self) -> None:
880 manifest = {"a.py": fake_id("oid-a")}
881 sid = compute_snapshot_id(manifest)
882 corrupted = SnapshotRecord(
883 snapshot_id=sid,
884 manifest={"b.py": fake_id("oid-a")}, # wrong path
885 created_at=datetime.datetime.now(datetime.timezone.utc),
886 )
887 with pytest.raises(OSError, match="content-hash verification"):
888 _verify_snapshot_id(corrupted, sid, pathlib.Path("snap.msgpack"))
889
890 def test_extra_entry_raises(self) -> None:
891 manifest = {"a.py": fake_id("oid-a")}
892 sid = compute_snapshot_id(manifest)
893 corrupted = SnapshotRecord(
894 snapshot_id=sid,
895 manifest={"a.py": fake_id("oid-a"), "extra.py": fake_id("oid-c")}, # injected entry
896 created_at=datetime.datetime.now(datetime.timezone.utc),
897 )
898 with pytest.raises(OSError, match="content-hash verification"):
899 _verify_snapshot_id(corrupted, sid, pathlib.Path("snap.msgpack"))
900
901 def test_missing_entry_raises(self) -> None:
902 manifest = {"a.py": fake_id("oid-a"), "b.py": fake_id("oid-b")}
903 sid = compute_snapshot_id(manifest)
904 corrupted = SnapshotRecord(
905 snapshot_id=sid,
906 manifest={"a.py": fake_id("oid-a")}, # b.py missing
907 created_at=datetime.datetime.now(datetime.timezone.utc),
908 )
909 with pytest.raises(OSError, match="content-hash verification"):
910 _verify_snapshot_id(corrupted, sid, pathlib.Path("snap.msgpack"))
911
912 def test_empty_manifest_clean(self) -> None:
913 sid = compute_snapshot_id({})
914 rec = SnapshotRecord(
915 snapshot_id=sid, manifest={},
916 created_at=datetime.datetime.now(datetime.timezone.utc),
917 )
918 _verify_snapshot_id(rec, sid, pathlib.Path("snap.msgpack"))
919
920 def test_large_manifest_50k_entries(self) -> None:
921 """50 000-entry manifest: _verify_snapshot_id completes quickly."""
922 import time
923 manifest = {f"path/to/file_{i:06d}.py": fake_id(f"obj{i}")
924 for i in range(50_000)}
925 sid = compute_snapshot_id(manifest)
926 rec = SnapshotRecord(
927 snapshot_id=sid, manifest=manifest,
928 created_at=datetime.datetime.now(datetime.timezone.utc),
929 )
930 start = time.perf_counter()
931 _verify_snapshot_id(rec, sid, pathlib.Path("snap.msgpack"))
932 duration_ms = (time.perf_counter() - start) * 1000
933 assert duration_ms < 5000, f"50k manifest verify took {duration_ms:.0f} ms (budget: 5 000 ms)"
934
935
936 # ---------------------------------------------------------------------------
937 # 9. Regression: pre-fix silent corruption gap is now closed
938 # ---------------------------------------------------------------------------
939
940
941 class TestRegressionSilentCorrupt:
942 """I-10 regression: core-field corruptions that were silent are now caught.
943
944 Before I-10, 2 450 out of 3 776 bit positions in a commit file (the ones
945 in core fields) produced a silently wrong CommitRecord. Post-fix: zero.
946
947 The remaining ~1 954 bit positions are in metadata fields (branch, author,
948 repo_id, etc.) that are not in compute_commit_id by design — those are
949 documented limitations, not regressions.
950 """
951
952 def test_core_field_corruptions_zero_silent_passes(self, tmp_path: pathlib.Path) -> None:
953 """Bit flips in core commit fields: zero silent passes after I-10 fix.
954
955 Identifies core-field positions by checking whether the recomputed
956 commit_id would differ from the expected ID. Only those positions
957 are in scope for the zero-silent-passes assertion.
958 """
959 repo = _repo(tmp_path)
960 import msgpack as _mp
961 cid, path = _make_commit(repo, msg="regression test", snap_id=fake_id("snap-7"))
962 original = path.read_bytes()
963 silent = 0
964 for b in range(len(original)):
965 for bit in range(8):
966 flipped = _flip_bit(original, b, bit)
967 _corrupt_file(path, flipped)
968 result = read_commit(repo, cid)
969 if result is not None:
970 # Determine if this was a core-field position
971 try:
972 d = _mp.unpackb(flipped, raw=False)
973 if isinstance(d, dict):
974 r = CommitRecord.from_msgpack(d)
975 parent_ids: list[str] = []
976 if r.parent_commit_id:
977 parent_ids.append(r.parent_commit_id)
978 recomputed = compute_commit_id( parent_ids=parent_ids,
979 snapshot_id=r.snapshot_id,
980 message=r.message,
981 committed_at_iso=r.committed_at.isoformat(),
982 author=r.author or "",
983 signer_public_key=r.signer_public_key or "",
984 )
985 if recomputed != cid:
986 silent += 1
987 except Exception:
988 pass
989 _corrupt_file(path, original)
990 assert silent == 0, (
991 f"{silent} CORE-field bit flips in commit were silently returned. "
992 "This was the pre-I-10 gap — _verify_commit_id should now catch all."
993 )
994
995 def test_manifest_corruptions_zero_silent_passes(self, tmp_path: pathlib.Path) -> None:
996 """Bit flips that corrupt manifest entries: zero silent passes after I-10 fix."""
997 repo = _repo(tmp_path)
998 import msgpack as _mp
999 sid, path = _make_snapshot(repo, {"main.py": fake_id("oid-8"), "lib.py": fake_id("oid-9")})
1000 original = path.read_bytes()
1001 silent = 0
1002 for b in range(len(original)):
1003 for bit in range(8):
1004 flipped = _flip_bit(original, b, bit)
1005 _corrupt_file(path, flipped)
1006 result = read_snapshot(repo, sid)
1007 if result is not None:
1008 try:
1009 d = _mp.unpackb(flipped, raw=False)
1010 if isinstance(d, dict) and isinstance(d.get("manifest"), dict):
1011 recomputed = compute_snapshot_id(d["manifest"])
1012 if recomputed != sid:
1013 silent += 1
1014 except Exception:
1015 pass
1016 _corrupt_file(path, original)
1017 assert silent == 0, (
1018 f"{silent} manifest-region bit flips in snapshot were silently returned. "
1019 "_verify_snapshot_id should catch all manifest corruptions."
1020 )
1021
1022 def test_read_commit_returns_none_not_wrong_record(self, tmp_path: pathlib.Path) -> None:
1023 """A core-field-corrupted commit file returns None, not a wrong CommitRecord."""
1024 repo = _repo(tmp_path)
1025 import msgpack as _mp
1026 now = datetime.datetime.now(datetime.timezone.utc)
1027 snap_id = fake_id("snap-6")
1028 cid = compute_commit_id(parent_ids=[], snapshot_id=snap_id, message="original message", committed_at_iso=now.isoformat())
1029 rec = CommitRecord(
1030 commit_id=cid, repo_id="r", branch="main",
1031 snapshot_id=snap_id, message="original message", committed_at=now,
1032 )
1033 write_commit(repo, rec)
1034 path = commit_path(repo, cid)
1035 original = path.read_bytes()
1036 d = _mp.unpackb(original, raw=False)
1037 assert isinstance(d, dict)
1038 d["message"] = "tampered message"
1039 _corrupt_file(path, _mp.packb(d, use_bin_type=True))
1040 result = read_commit(repo, cid)
1041 assert result is None, (
1042 "read_commit must return None on core-field corruption, "
1043 "not a record with wrong message"
1044 )
1045 _corrupt_file(path, original)
1046
1047
1048 # ---------------------------------------------------------------------------
1049 # 10. Msgpack fuzz — 10 000 rounds on commit + snapshot
1050 # ---------------------------------------------------------------------------
1051
1052
1053 class TestMsgpackFuzz10k:
1054 """Random multi-byte corruption fuzz on commit and snapshot files."""
1055
1056 @pytest.mark.slow
1057 def test_5_bit_fuzz_10k_commit_core_field_always_touched(self, tmp_path: pathlib.Path) -> None:
1058 """10 000 fuzz rounds each touching a core commit field: zero silent passes.
1059
1060 Each round flips 1 bit in a core-field region (snapshot_id, message, or
1061 committed_at in the msgpack) plus 4 random bits elsewhere. This guarantees
1062 the fuzz always reaches a content-hash-verified field, making zero silent
1063 passes the correct assertion.
1064
1065 Pure random 5-bit fuzz has ~3.7% probability of landing all bits in metadata
1066 fields (branch, author, repo_id, etc.), which would produce expected silent
1067 passes — that is a documented design limitation, not a bug.
1068 """
1069 repo = _repo(tmp_path)
1070 import msgpack as _mp
1071 cid, path = _make_commit(repo, msg="fuzz me", snap_id=fake_id("snap-5"))
1072 original = path.read_bytes()
1073 d_orig = _mp.unpackb(original, raw=False)
1074 assert isinstance(d_orig, dict)
1075
1076 rng = random.Random(42)
1077 core_fields = ["snapshot_id", "message", "committed_at"]
1078 silent = 0
1079 for _ in range(10_000):
1080 # Always corrupt a core field
1081 field = rng.choice(core_fields)
1082 d = dict(d_orig)
1083 if field == "snapshot_id":
1084 d["snapshot_id"] = rng.choice(["e", "f", "0"]) * 64
1085 elif field == "message":
1086 d["message"] = f"tampered-{rng.randint(0, 999999)}"
1087 else:
1088 d["committed_at"] = f"200{rng.randint(0,9)}-01-01T00:00:00+00:00"
1089 # Plus 4 random bit flips
1090 packed = bytearray(_mp.packb(d, use_bin_type=True))
1091 for _ in range(4):
1092 if packed:
1093 packed[rng.randrange(len(packed))] ^= 1 << rng.randrange(8)
1094 _corrupt_file(path, bytes(packed))
1095 if read_commit(repo, cid) is not None:
1096 silent += 1
1097 _corrupt_file(path, original)
1098 assert silent == 0, (
1099 f"{silent} commit fuzz rounds (with guaranteed core-field corruption) "
1100 "went undetected — _verify_commit_id must catch all core-field changes"
1101 )
1102
1103 @pytest.mark.slow
1104 def test_5_bit_fuzz_10k_snapshot_manifest_always_touched(self, tmp_path: pathlib.Path) -> None:
1105 """10 000 fuzz rounds each touching a manifest entry: zero silent passes.
1106
1107 Each round corrupts at least one manifest entry (path or oid) to guarantee
1108 the fuzz reaches content-hash-verified data. Pure random 5-bit fuzz has
1109 a small probability of landing all bits in the ``created_at`` metadata field,
1110 which is a documented limitation — not a bug.
1111 """
1112 repo = _repo(tmp_path)
1113 import msgpack as _mp
1114 manifest = {"x.py": fake_id("oid-4"), "y.py": fake_id("oid-5")}
1115 sid, path = _make_snapshot(repo, manifest)
1116 original = path.read_bytes()
1117 d_orig = _mp.unpackb(original, raw=False)
1118 assert isinstance(d_orig, dict)
1119 assert isinstance(d_orig["manifest"], dict)
1120
1121 rng = random.Random(99)
1122 silent = 0
1123 for _ in range(10_000):
1124 d = dict(d_orig)
1125 d["manifest"] = dict(d_orig["manifest"])
1126 # Always corrupt one manifest entry
1127 key = rng.choice(list(manifest.keys()))
1128 d["manifest"][key] = rng.choice(["a", "b", "c"]) * 64
1129 # Plus 4 random bit flips
1130 packed = bytearray(_mp.packb(d, use_bin_type=True))
1131 for _ in range(4):
1132 if packed:
1133 packed[rng.randrange(len(packed))] ^= 1 << rng.randrange(8)
1134 _corrupt_file(path, bytes(packed))
1135 if read_snapshot(repo, sid) is not None:
1136 silent += 1
1137 _corrupt_file(path, original)
1138 assert silent == 0, (
1139 f"{silent} snapshot fuzz rounds (with guaranteed manifest corruption) "
1140 "went undetected — _verify_snapshot_id must catch all manifest changes"
1141 )
1142
1143 def test_completely_random_commit_bytes_100_rounds(self, tmp_path: pathlib.Path) -> None:
1144 """Replacing a commit file with random bytes: all 100 rounds caught."""
1145 repo = _repo(tmp_path)
1146 cid, path = _make_commit(repo)
1147 original = path.read_bytes()
1148 rng = random.Random(7)
1149 for _ in range(100):
1150 garbage = bytes(rng.randrange(256) for _ in range(len(original)))
1151 _corrupt_file(path, garbage)
1152 assert read_commit(repo, cid) is None
1153 _corrupt_file(path, original)
1154
1155 def test_completely_random_snapshot_bytes_100_rounds(self, tmp_path: pathlib.Path) -> None:
1156 """Replacing a snapshot file with random bytes: all 100 rounds caught."""
1157 repo = _repo(tmp_path)
1158 sid, path = _make_snapshot(repo)
1159 original = path.read_bytes()
1160 rng = random.Random(8)
1161 for _ in range(100):
1162 garbage = bytes(rng.randrange(256) for _ in range(len(original)))
1163 _corrupt_file(path, garbage)
1164 assert read_snapshot(repo, sid) is None
1165 _corrupt_file(path, original)
1166
1167
1168 # ---------------------------------------------------------------------------
1169 # 11. CRITICAL log emission on corruption detection
1170 # ---------------------------------------------------------------------------
1171
1172
1173 class TestCriticalLogged:
1174 """CRITICAL is emitted for every detected bit flip (both object + store)."""
1175
1176 def test_object_bit_flip_emits_critical(
1177 self, tmp_path: pathlib.Path, caplog: pytest.LogCaptureFixture
1178 ) -> None:
1179 import logging
1180 repo = _repo(tmp_path)
1181 data = b"log test object"
1182 oid = _write(repo, data)
1183 p = _stored_path(repo, oid)
1184 original = p.read_bytes()
1185 _corrupt_file(p, _flip_bit(original, 0, 0))
1186 with caplog.at_level(logging.CRITICAL):
1187 try:
1188 read_object(repo, oid)
1189 except OSError:
1190 pass
1191 _corrupt_file(p, original)
1192 assert any("integrity check" in r.message.lower() or "corrupt" in r.message.lower()
1193 for r in caplog.records)
1194
1195 def test_commit_flip_emits_critical(
1196 self, tmp_path: pathlib.Path, caplog: pytest.LogCaptureFixture
1197 ) -> None:
1198 import logging
1199 repo = _repo(tmp_path)
1200 cid, path = _make_commit(repo)
1201 original = path.read_bytes()
1202 # Try flips until one hits the CRITICAL path (verify_commit_id)
1203 # Most flips will hit msgpack unpack error first; we want one that
1204 # produces valid msgpack but fails content-hash verification.
1205 import msgpack as _mp
1206 d = _mp.unpackb(original, raw=False)
1207 assert isinstance(d, dict)
1208 d["message"] = "tampered"
1209 _corrupt_file(path, _mp.packb(d, use_bin_type=True))
1210 with caplog.at_level(logging.CRITICAL):
1211 read_commit(repo, cid)
1212 _corrupt_file(path, original)
1213 assert any("corrupt" in r.message.lower() for r in caplog.records)
1214
1215 def test_snapshot_flip_emits_critical(
1216 self, tmp_path: pathlib.Path, caplog: pytest.LogCaptureFixture
1217 ) -> None:
1218 import logging
1219 repo = _repo(tmp_path)
1220 sid, path = _make_snapshot(repo)
1221 original = path.read_bytes()
1222 import msgpack as _mp
1223 d = _mp.unpackb(original, raw=False)
1224 assert isinstance(d, dict)
1225 assert isinstance(d["manifest"], dict)
1226 d["manifest"]["README.md"] = fake_id("oid-z")
1227 _corrupt_file(path, _mp.packb(d, use_bin_type=True))
1228 with caplog.at_level(logging.CRITICAL):
1229 read_snapshot(repo, sid)
1230 _corrupt_file(path, original)
1231 assert any("corrupt" in r.message.lower() for r in caplog.records)
1232
1233
1234 # ---------------------------------------------------------------------------
1235 # 12. Round-trip integrity
1236 # ---------------------------------------------------------------------------
1237
1238
1239 class TestRoundTripIntegrity:
1240 """Clean writes always round-trip without error."""
1241
1242 def test_object_round_trip(self, tmp_path: pathlib.Path) -> None:
1243 repo = _repo(tmp_path)
1244 for size in (0, 1, 31, 32, 33, 4095, 4096, 65535, 65536, 65537):
1245 data = os.urandom(size)
1246 oid = _write(repo, data)
1247 assert read_object(repo, oid) == data
1248
1249 def test_commit_round_trip(self, tmp_path: pathlib.Path) -> None:
1250 repo = _repo(tmp_path)
1251 cid, _ = _make_commit(repo, msg="clean commit", snap_id=fake_id("snap-3b"))
1252 result = read_commit(repo, cid)
1253 assert result is not None
1254 assert result.commit_id == cid
1255 assert result.message == "clean commit"
1256
1257 def test_snapshot_round_trip(self, tmp_path: pathlib.Path) -> None:
1258 repo = _repo(tmp_path)
1259 manifest = {f"f{i}.py": fake_id(str(i)) for i in range(100)}
1260 sid, _ = _make_snapshot(repo, manifest)
1261 result = read_snapshot(repo, sid)
1262 assert result is not None
1263 assert result.snapshot_id == sid
1264 assert result.manifest == manifest
1265
1266 def test_commit_with_parents_round_trip(self, tmp_path: pathlib.Path) -> None:
1267 repo = _repo(tmp_path)
1268 p1 = fake_id("parent-1")
1269 p2 = fake_id("parent-2")
1270 snap_id = fake_id("snap-3c")
1271 # Stub both parents so write_commit's existence guard passes.
1272 _stub_parent(repo, p1)
1273 _stub_parent(repo, p2)
1274 now = datetime.datetime.now(datetime.timezone.utc)
1275 cid = compute_commit_id(parent_ids=[p1, p2], snapshot_id=snap_id, message="merge commit", committed_at_iso=now.isoformat())
1276 rec = CommitRecord(
1277 commit_id=cid, repo_id="r", branch="main",
1278 snapshot_id=snap_id, message="merge commit", committed_at=now,
1279 parent_commit_id=p1, parent2_commit_id=p2,
1280 )
1281 write_commit(repo, rec)
1282 result = read_commit(repo, cid)
1283 assert result is not None
1284 assert result.parent_commit_id == p1
1285 assert result.parent2_commit_id == p2
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 130 days ago