gabriel / muse public
test_security_msgpack_hardening.py python
793 lines 30.8 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """Phase 2.4 — Malicious msgpack payload hardening tests.
2
3 Attack surface covered
4 ----------------------
5 * ``safe_unpackb`` — the new canonical deserialization primitive:
6 - Size-bomb (payload larger than max_bytes cap)
7 - Billion-laughs via enormous maps / arrays
8 - String length bomb
9 - Binary blob injection (allow_binary=False default)
10 - ``strict_map_key=False`` passthrough for legacy staging index
11 - Clean inputs still deserialise correctly
12 - 10 000-entry random fuzz with no unhandled exceptions
13 - Concurrent deserialization stress (50 threads)
14
15 * ``read_msgpack_file`` — file-based wrapper:
16 - Stat check fires before read_bytes (no OOM on 4 GiB file placeholder)
17 - Per-value limits enforced after stat
18
19 * ``MAX_PACK_MSGPACK_BYTES`` — new pack/bundle constant:
20 - Exported and larger than ``MAX_MSGPACK_BYTES``
21 - Verified against the 512 MiB specification
22
23 * Callsite hardening (end-to-end via CliRunner / direct call):
24 - ``bundle._load_bundle`` rejects oversized bundle files
25 - ``unpack_objects`` stdin rejects size-bomb payloads
26 - ``verify_pack`` stdin rejects size-bomb payloads
27 - ``symbol_cache.SymbolCache.load`` rejects oversized cache files
28 - ``test_history.load_history`` rejects oversized history files
29 - ``transport._decode`` enforces per-value limits on server responses
30 - ``_invariants._FileCache.load`` rejects oversized cache files
31 - ``stage.read_stage`` rejects oversized staging index files
32
33 * TypeGuard narrowing (``_is_commit_dict``, ``_is_snapshot_dict``) —
34 non-dict entries in a wire bundle are silently dropped, not propagated.
35 """
36 from __future__ import annotations
37
38 import os
39 import pathlib
40 import random
41 import struct
42 import tempfile
43 import threading
44 import time
45 from unittest.mock import MagicMock, patch
46
47 import msgpack
48 import pytest
49
50 from muse.core.store import (
51 MAX_MSGPACK_BYTES,
52 MAX_PACK_MSGPACK_BYTES,
53 MsgpackValue,
54 read_msgpack_file,
55 safe_unpackb,
56 )
57
58
59 # ---------------------------------------------------------------------------
60 # Helpers
61 # ---------------------------------------------------------------------------
62
63 def _pack(obj: MsgpackValue) -> bytes:
64 raw = msgpack.packb(obj, use_bin_type=True)
65 assert isinstance(raw, bytes)
66 return raw
67
68
69 def _nested_map(depth: int) -> MsgpackDict:
70 """Build a dict nested *depth* levels deep."""
71 result: MsgpackDict = {"x": None}
72 for _ in range(depth - 1):
73 result = {"x": result}
74 return result
75
76
77 def _nested_list(depth: int) -> list[MsgpackValue]:
78 """Build a list nested *depth* levels deep."""
79 result: list[MsgpackValue] = [None]
80 for _ in range(depth - 1):
81 result = [result]
82 return result
83
84
85 # ---------------------------------------------------------------------------
86 # 1. safe_unpackb — unit tests
87 # ---------------------------------------------------------------------------
88
89 class TestSafeUnpackbSizeBomb:
90 """safe_unpackb raises ValueError before parsing when len(raw) > max_bytes."""
91
92 def test_exact_limit_is_accepted(self) -> None:
93 payload = _pack("x")
94 assert safe_unpackb(payload, max_bytes=len(payload)) == "x"
95
96 def test_one_byte_over_raises(self) -> None:
97 payload = _pack("x")
98 with pytest.raises(ValueError, match="safety cap"):
99 safe_unpackb(payload, max_bytes=len(payload) - 1)
100
101 def test_default_limit_is_max_msgpack_bytes(self) -> None:
102 """Payloads at exactly MAX_MSGPACK_BYTES are accepted."""
103 tiny = _pack({"k": "v"})
104 result = safe_unpackb(tiny)
105 assert result == {"k": "v"}
106
107 def test_size_error_includes_context_label(self) -> None:
108 payload = _pack("big")
109 with pytest.raises(ValueError, match="stdin"):
110 safe_unpackb(payload, context="stdin", max_bytes=0)
111
112 def test_empty_bytes_accepted(self) -> None:
113 """Empty msgpack (nil) is valid and accepted."""
114 nil_bytes = _pack(None)
115 result = safe_unpackb(nil_bytes)
116 assert result is None
117
118
119 class TestSafeUnpackbPerValueLimits:
120 """Per-value limits block billion-laughs payloads."""
121
122 def test_string_over_1_mib_raises(self) -> None:
123 big_str = "A" * (1_048_577) # 1 MiB + 1 byte
124 payload = _pack(big_str)
125 with pytest.raises(Exception):
126 safe_unpackb(payload, max_bytes=len(payload) + 100)
127
128 def test_string_exactly_at_1_mib_accepted(self) -> None:
129 ok_str = "A" * 1_048_576
130 payload = _pack(ok_str)
131 result = safe_unpackb(payload, max_bytes=len(payload) + 100)
132 assert result == ok_str
133
134 def test_huge_map_rejected(self) -> None:
135 """A map with 1_000_001 keys exceeds _MSGPACK_MAX_MAP_LEN."""
136 # Build a msgpack map header with count > limit directly to avoid
137 # allocating 1M Python strings in the test process.
138 count = 1_000_001
139 # msgpack map32: 0xdf + uint32 BE count
140 raw = struct.pack(">BI", 0xDF, count) + b"\xa1x\xa1y" * count
141 with pytest.raises(Exception):
142 safe_unpackb(raw, max_bytes=len(raw) + 1000)
143
144 def test_huge_array_rejected(self) -> None:
145 """An array with 1_000_001 entries exceeds _MSGPACK_MAX_ARRAY_LEN."""
146 count = 1_000_001
147 # msgpack array32: 0xdd + uint32 BE count
148 raw = struct.pack(">BI", 0xDD, count) + b"\xc0" * count
149 with pytest.raises(Exception):
150 safe_unpackb(raw, max_bytes=len(raw) + 1000)
151
152 def test_binary_blob_rejected_by_default(self) -> None:
153 """allow_binary=False (default) rejects msgpack binary blobs."""
154 payload = _pack(b"\x00\xff" * 10)
155 with pytest.raises(Exception):
156 safe_unpackb(payload, max_bytes=len(payload) + 100)
157
158 def test_binary_blob_accepted_with_allow_binary(self) -> None:
159 """allow_binary=True permits binary blobs (pack/bundle payloads)."""
160 blob = b"\xde\xad\xbe\xef" * 4
161 payload = _pack(blob)
162 result = safe_unpackb(payload, max_bytes=len(payload) + 100, allow_binary=True)
163 assert result == blob
164
165 def test_strict_map_key_true_rejects_integer_keys(self) -> None:
166 """By default, integer map keys are rejected."""
167 # Hand-craft msgpack with integer key
168 payload = struct.pack(">BB", 0x81, 0x01) + b"\xa1v" # {1: "v"}
169 with pytest.raises(Exception):
170 safe_unpackb(payload, max_bytes=100)
171
172 def test_strict_map_key_false_allows_integer_keys(self) -> None:
173 """strict_map_key=False permits legacy integer keys (e.g. {1: 'v'})."""
174 payload = struct.pack(">BB", 0x81, 0x01) + b"\xa1v" # {1: "v"}
175 result = safe_unpackb(payload, max_bytes=100, strict_map_key=False)
176 # Returned as a dict with one entry whose value is "v"
177 assert isinstance(result, dict) and list(result.values()) == ["v"]
178
179 def test_invalid_msgpack_raises(self) -> None:
180 with pytest.raises(Exception):
181 safe_unpackb(b"\xff\xfe\xfd\xfc", max_bytes=100)
182
183 def test_clean_dict_roundtrip(self) -> None:
184 original: MsgpackDict = {"key": "value", "n": 42, "flag": True}
185 result = safe_unpackb(_pack(original))
186 assert result == original
187
188 def test_clean_list_roundtrip(self) -> None:
189 original: list[MsgpackValue] = ["a", 1, None, True]
190 result = safe_unpackb(_pack(original))
191 assert result == original
192
193
194 class TestSafeUnpackbNestingBomb:
195 """Deeply nested structures — should raise or return, never hang."""
196
197 def test_500_nested_dicts_terminates_quickly(self) -> None:
198 """500 nested dicts must terminate (raise or succeed) within 1 second."""
199 import sys
200 old_limit = sys.getrecursionlimit()
201 sys.setrecursionlimit(max(old_limit, 5000))
202 try:
203 payload = _pack(_nested_map(500))
204 start = time.monotonic()
205 try:
206 safe_unpackb(payload, max_bytes=len(payload) + 10_000)
207 except Exception:
208 pass
209 elapsed = time.monotonic() - start
210 assert elapsed < 1.0, f"Nested dict deserialization hung ({elapsed:.2f}s)"
211 finally:
212 sys.setrecursionlimit(old_limit)
213
214 def test_500_nested_lists_terminates_quickly(self) -> None:
215 import sys
216 old_limit = sys.getrecursionlimit()
217 sys.setrecursionlimit(max(old_limit, 5000))
218 try:
219 payload = _pack(_nested_list(500))
220 start = time.monotonic()
221 try:
222 safe_unpackb(payload, max_bytes=len(payload) + 10_000)
223 except Exception:
224 pass
225 elapsed = time.monotonic() - start
226 assert elapsed < 1.0, f"Nested list deserialization hung ({elapsed:.2f}s)"
227 finally:
228 sys.setrecursionlimit(old_limit)
229
230
231 # ---------------------------------------------------------------------------
232 # 2. read_msgpack_file — unit tests
233 # ---------------------------------------------------------------------------
234
235 class TestReadMsgpackFile:
236 """read_msgpack_file enforces the size cap via stat before read_bytes."""
237
238 def test_normal_file_roundtrips(self, tmp_path: pathlib.Path) -> None:
239 f = tmp_path / "ok.msgpack"
240 f.write_bytes(_pack({"a": 1}))
241 result = read_msgpack_file(f)
242 assert result == {"a": 1}
243
244 def test_oversized_file_raises_os_error_before_read(self, tmp_path: pathlib.Path) -> None:
245 """The stat check must fire *before* read_bytes so no OOM occurs."""
246 f = tmp_path / "big.msgpack"
247 # Write minimal valid msgpack but lie to stat via mock
248 f.write_bytes(_pack("tiny"))
249 with patch.object(pathlib.Path, "stat") as mock_stat:
250 mock_stat.return_value = MagicMock(st_size=MAX_MSGPACK_BYTES + 1)
251 with pytest.raises(OSError, match="safety cap"):
252 read_msgpack_file(f)
253
254 def test_error_message_includes_filename(self, tmp_path: pathlib.Path) -> None:
255 f = tmp_path / "corrupt.msgpack"
256 f.write_bytes(_pack("x"))
257 with patch.object(pathlib.Path, "stat") as mock_stat:
258 mock_stat.return_value = MagicMock(st_size=MAX_MSGPACK_BYTES + 1)
259 with pytest.raises(OSError, match="corrupt.msgpack"):
260 read_msgpack_file(f)
261
262 def test_custom_max_bytes_respected(self, tmp_path: pathlib.Path) -> None:
263 f = tmp_path / "small.msgpack"
264 f.write_bytes(_pack("x"))
265 size = f.stat().st_size
266 # One byte under custom limit — OK
267 result = read_msgpack_file(f, max_bytes=size + 1)
268 assert result == "x"
269 # One byte over custom limit — raises
270 with pytest.raises(OSError):
271 read_msgpack_file(f, max_bytes=size - 1)
272
273 def test_strict_map_key_false_passed_through(self, tmp_path: pathlib.Path) -> None:
274 """strict_map_key=False is forwarded to safe_unpackb."""
275 f = tmp_path / "int_keys.msgpack"
276 # hand-craft msgpack with int key 1 -> "v"
277 f.write_bytes(struct.pack(">BB", 0x81, 0x01) + b"\xa1v")
278 result = read_msgpack_file(f, strict_map_key=False)
279 assert isinstance(result, dict) and list(result.values()) == ["v"]
280
281 def test_per_value_limits_apply_after_stat(self, tmp_path: pathlib.Path) -> None:
282 """Even within the size cap, a huge string is rejected."""
283 big_str = "Z" * (1_048_577)
284 f = tmp_path / "big_str.msgpack"
285 f.write_bytes(_pack(big_str))
286 size = f.stat().st_size
287 with pytest.raises(Exception):
288 read_msgpack_file(f, max_bytes=size + 1000)
289
290
291 # ---------------------------------------------------------------------------
292 # 3. MAX_PACK_MSGPACK_BYTES — constant tests
293 # ---------------------------------------------------------------------------
294
295 class TestMaxPackMsgpackBytes:
296 def test_exported(self) -> None:
297 assert MAX_PACK_MSGPACK_BYTES is not None
298
299 def test_is_int(self) -> None:
300 assert isinstance(MAX_PACK_MSGPACK_BYTES, int)
301
302 def test_larger_than_max_msgpack_bytes(self) -> None:
303 assert MAX_PACK_MSGPACK_BYTES > MAX_MSGPACK_BYTES
304
305 def test_is_512_mib(self) -> None:
306 assert MAX_PACK_MSGPACK_BYTES == 512 * 1024 * 1024
307
308
309 # ---------------------------------------------------------------------------
310 # 4. safe_unpackb fuzzing — 10 000 random inputs, no unhandled exceptions
311 # ---------------------------------------------------------------------------
312
313 class TestSafeUnpackbFuzz10k:
314 """Feed 10 000 random byte strings to safe_unpackb.
315
316 All calls must raise a known exception or return a valid MsgpackValue.
317 No unhandled exceptions (AttributeError, KeyError, etc.) are permitted.
318 """
319
320 def test_fuzz_10k_random_bytes(self) -> None:
321 rng = random.Random(0xDEADBEEF)
322 allowed_exc = (
323 ValueError, # size cap
324 msgpack.UnpackException,
325 msgpack.ExtraData,
326 msgpack.FormatError,
327 RecursionError,
328 UnicodeDecodeError,
329 MemoryError,
330 )
331 for i in range(10_000):
332 size = rng.randint(0, 64)
333 payload = bytes(rng.randint(0, 255) for _ in range(size))
334 try:
335 safe_unpackb(payload, max_bytes=256)
336 except allowed_exc:
337 pass
338 except Exception as exc:
339 pytest.fail(
340 f"Unexpected exception on iteration {i} "
341 f"(payload={payload.hex()!r}): {type(exc).__name__}: {exc}"
342 )
343
344 def test_fuzz_10k_valid_msgpack(self) -> None:
345 """All valid msgpack inputs (within limits) must deserialise cleanly."""
346 rng = random.Random(0xCAFEBABE)
347 allowed_exc = (
348 ValueError, # size cap or value limit
349 msgpack.UnpackException,
350 )
351 for i in range(10_000):
352 # Generate a small valid msgpack object
353 kind = rng.randint(0, 4)
354 if kind == 0:
355 obj: MsgpackValue = rng.randint(-(2**31), 2**31 - 1)
356 elif kind == 1:
357 obj = rng.random()
358 elif kind == 2:
359 obj = "".join(chr(rng.randint(32, 126)) for _ in range(rng.randint(0, 32)))
360 elif kind == 3:
361 obj = None
362 else:
363 obj = bool(rng.randint(0, 1))
364 payload = _pack(obj)
365 try:
366 safe_unpackb(payload, max_bytes=len(payload) + 10)
367 except allowed_exc:
368 pass
369 except Exception as exc:
370 pytest.fail(
371 f"Unexpected exception on iteration {i}: "
372 f"{type(exc).__name__}: {exc}"
373 )
374
375
376 # ---------------------------------------------------------------------------
377 # 5. Concurrent deserialization stress
378 # ---------------------------------------------------------------------------
379
380 class TestSafeUnpackbConcurrent:
381 """50 threads calling safe_unpackb simultaneously — no data races."""
382
383 def test_50_threads_concurrent_safe_unpackb(self) -> None:
384 payload = _pack({"key": "value", "n": 42})
385 errors: list[Exception] = []
386
387 def _worker() -> None:
388 try:
389 result = safe_unpackb(payload)
390 assert result == {"key": "value", "n": 42}
391 except Exception as exc:
392 errors.append(exc)
393
394 threads = [threading.Thread(target=_worker) for _ in range(50)]
395 for t in threads:
396 t.start()
397 for t in threads:
398 t.join(timeout=5)
399
400 assert not errors, f"Concurrent errors: {errors}"
401
402 def test_50_threads_size_bomb_rejected_concurrently(self) -> None:
403 payload = _pack("tiny")
404 errors: list[str] = []
405
406 def _worker() -> None:
407 try:
408 safe_unpackb(payload, max_bytes=0)
409 errors.append("Expected ValueError, got success")
410 except ValueError:
411 pass # correct
412 except Exception as exc:
413 errors.append(f"Wrong exception: {type(exc).__name__}: {exc}")
414
415 threads = [threading.Thread(target=_worker) for _ in range(50)]
416 for t in threads:
417 t.start()
418 for t in threads:
419 t.join(timeout=5)
420
421 assert not errors, f"Errors: {errors}"
422
423
424 # ---------------------------------------------------------------------------
425 # 6. Callsite hardening — bundle._load_bundle
426 # ---------------------------------------------------------------------------
427
428 class TestBundleLoadHardening:
429 """_load_bundle rejects oversized files before reading into memory."""
430
431 def test_oversized_bundle_file_exits_cleanly(self, tmp_path: pathlib.Path) -> None:
432 from muse.cli.commands.bundle import _load_bundle
433
434 bundle_file = tmp_path / "huge.bundle"
435 bundle_file.write_bytes(_pack({"commits": [], "snapshots": [], "objects": []}))
436 with patch.object(pathlib.Path, "stat") as mock_stat:
437 mock_stat.return_value = MagicMock(st_size=MAX_PACK_MSGPACK_BYTES + 1)
438 with pytest.raises(SystemExit):
439 _load_bundle(bundle_file)
440
441 def test_bundle_size_check_fires_before_read(self, tmp_path: pathlib.Path) -> None:
442 """Stat check must happen *before* read_bytes — never allocate the big buffer."""
443 from muse.cli.commands.bundle import _load_bundle
444
445 bundle_file = tmp_path / "fake_huge.bundle"
446 bundle_file.write_bytes(_pack({}))
447
448 read_bytes_calls: list[int] = []
449 original_read_bytes = pathlib.Path.read_bytes
450
451 def tracked_read_bytes(self: pathlib.Path) -> bytes:
452 read_bytes_calls.append(1)
453 return original_read_bytes(self)
454
455 with patch.object(pathlib.Path, "stat") as mock_stat, \
456 patch.object(pathlib.Path, "read_bytes", tracked_read_bytes):
457 mock_stat.return_value = MagicMock(st_size=MAX_PACK_MSGPACK_BYTES + 1)
458 with pytest.raises(SystemExit):
459 _load_bundle(bundle_file)
460
461 # read_bytes must NOT have been called — stat check fired first
462 assert not read_bytes_calls, "read_bytes was called despite oversized stat"
463
464 def test_valid_bundle_loads_correctly(self, tmp_path: pathlib.Path) -> None:
465 from muse.cli.commands.bundle import _load_bundle
466
467 bundle_file = tmp_path / "valid.bundle"
468 bundle_file.write_bytes(_pack({"commits": [], "snapshots": [], "objects": []}))
469 result = _load_bundle(bundle_file)
470 assert isinstance(result, dict)
471
472 def test_non_dict_bundle_payload_rejected(self, tmp_path: pathlib.Path) -> None:
473 from muse.cli.commands.bundle import _load_bundle
474
475 bundle_file = tmp_path / "list.bundle"
476 bundle_file.write_bytes(_pack([1, 2, 3]))
477 with pytest.raises(SystemExit):
478 _load_bundle(bundle_file)
479
480 def test_non_dict_commits_entries_silently_dropped(self, tmp_path: pathlib.Path) -> None:
481 """Non-dict entries in commits list are filtered by _is_commit_dict."""
482 from muse.cli.commands.bundle import _load_bundle
483
484 bundle_file = tmp_path / "mixed.bundle"
485 bundle_file.write_bytes(_pack({
486 "commits": ["string_entry", 42, None, {"real_key": "real_value"}],
487 "snapshots": [],
488 "objects": [],
489 }))
490 result = _load_bundle(bundle_file)
491 commits = result.get("commits", [])
492 # Only the dict entry should survive
493 assert len(commits) == 1
494 assert commits[0] == {"real_key": "real_value"}
495
496
497 # ---------------------------------------------------------------------------
498 # 7. Callsite hardening — unpack_objects (stdin)
499 # ---------------------------------------------------------------------------
500
501 class TestUnpackObjectsStdinHardening:
502 """unpack_objects rejects size-bomb payloads from stdin."""
503
504 def test_size_bomb_stdin_rejected(self, tmp_path: pathlib.Path) -> None:
505 """A stdin payload exceeding MAX_PACK_MSGPACK_BYTES is rejected."""
506 import sys
507 from io import BytesIO
508
509 import muse.cli.commands.plumbing.unpack_objects as _mod
510
511 tiny_payload = _pack({"commits": [], "snapshots": [], "objects": []})
512 # Mock stdin.buffer.read to return a payload that exceeds the limit
513 with patch.object(_mod, "sys") as mock_sys, \
514 patch("muse.cli.commands.plumbing.unpack_objects.require_repo",
515 return_value=tmp_path):
516 # Build a .muse dir so require_repo doesn't fail
517 (tmp_path / ".muse").mkdir(exist_ok=True)
518 mock_sys.stdin = MagicMock()
519 mock_sys.stdin.buffer = MagicMock()
520
521 # Simulate payload larger than MAX_PACK_MSGPACK_BYTES
522 oversized = b"X" * (MAX_PACK_MSGPACK_BYTES + 1)
523 mock_sys.stdin.buffer.read.return_value = oversized
524 mock_sys.stderr = sys.stderr
525 mock_sys.stdout = sys.stdout
526
527 with pytest.raises(SystemExit):
528 args = MagicMock()
529 args.fmt = "json"
530 _mod.run(args)
531
532 def test_invalid_msgpack_stdin_rejected(self, tmp_path: pathlib.Path) -> None:
533 """Garbage stdin bytes produce a clean error exit, not a traceback."""
534 import sys
535 import muse.cli.commands.plumbing.unpack_objects as _mod
536
537 with patch.object(_mod, "sys") as mock_sys, \
538 patch("muse.cli.commands.plumbing.unpack_objects.require_repo",
539 return_value=tmp_path):
540 (tmp_path / ".muse").mkdir(exist_ok=True)
541 mock_sys.stdin = MagicMock()
542 mock_sys.stdin.buffer = MagicMock()
543 mock_sys.stdin.buffer.read.return_value = b"\xff\xfe\xfd garbage"
544 mock_sys.stderr = sys.stderr
545 mock_sys.stdout = sys.stdout
546
547 with pytest.raises(SystemExit) as exc_info:
548 args = MagicMock()
549 args.fmt = "json"
550 _mod.run(args)
551
552 assert exc_info.value.code != 0
553
554
555 # ---------------------------------------------------------------------------
556 # 8. Callsite hardening — verify_pack (stdin / file)
557 # ---------------------------------------------------------------------------
558
559 class TestVerifyPackHardening:
560 """verify_pack rejects oversized and malformed msgpack."""
561
562 def test_size_bomb_raises_system_exit(self) -> None:
563 """A stdin payload exceeding MAX_PACK_MSGPACK_BYTES is rejected."""
564 import sys
565 import muse.cli.commands.plumbing.verify_pack as _mod
566
567 oversized = b"X" * (MAX_PACK_MSGPACK_BYTES + 1)
568
569 with patch.object(_mod, "sys") as mock_sys:
570 mock_sys.stdin = MagicMock()
571 mock_sys.stdin.buffer = MagicMock()
572 mock_sys.stdin.buffer.read.return_value = oversized
573 mock_sys.stderr = sys.stderr
574 mock_sys.stdout = sys.stdout
575
576 with pytest.raises(SystemExit):
577 args = MagicMock()
578 args.format = "json"
579 args.file = None
580 _mod.run(args)
581
582
583 # ---------------------------------------------------------------------------
584 # 9. Callsite hardening — symbol_cache.SymbolCache.load
585 # ---------------------------------------------------------------------------
586
587 class TestSymbolCacheHardening:
588 """SymbolCache.load returns empty cache for oversized or corrupt files."""
589
590 def test_oversized_cache_returns_empty(self, tmp_path: pathlib.Path) -> None:
591 from muse.core.symbol_cache import SymbolCache
592
593 muse_dir = tmp_path / ".muse"
594 muse_dir.mkdir()
595 cache_file = muse_dir / "symbol_cache.msgpack"
596 cache_file.write_bytes(_pack({"version": 1, "entries": {}}))
597
598 with patch.object(pathlib.Path, "stat") as mock_stat:
599 mock_stat.return_value = MagicMock(st_size=MAX_MSGPACK_BYTES + 1)
600 result = SymbolCache.load(muse_dir)
601
602 # size is a @property, not a method
603 assert result.size == 0
604
605 def test_corrupt_cache_returns_empty(self, tmp_path: pathlib.Path) -> None:
606 from muse.core.symbol_cache import SymbolCache
607
608 muse_dir = tmp_path / ".muse"
609 muse_dir.mkdir()
610 cache_file = muse_dir / "symbol_cache.msgpack"
611 cache_file.write_bytes(b"\xff\xfe garbage")
612
613 result = SymbolCache.load(muse_dir)
614 assert result.size == 0
615
616 def test_valid_cache_still_loads(self, tmp_path: pathlib.Path) -> None:
617 from muse.core.symbol_cache import SymbolCache
618
619 muse_dir = tmp_path / ".muse"
620 muse_dir.mkdir()
621 cache_file = muse_dir / "symbol_cache.msgpack"
622 # Write a well-formed but empty cache
623 cache_file.write_bytes(_pack({"version": 1, "entries": {}}))
624
625 # Should not raise even though version != _CACHE_VERSION — returns empty
626 result = SymbolCache.load(muse_dir)
627 assert result.size == 0
628
629
630 # ---------------------------------------------------------------------------
631 # 10. Callsite hardening — test_history.load_history
632 # ---------------------------------------------------------------------------
633
634 class TestTestHistoryHardening:
635 """load_history returns [] for oversized or corrupt history files."""
636
637 def test_oversized_history_returns_empty_list(self, tmp_path: pathlib.Path) -> None:
638 from muse.core.test_history import load_history
639
640 muse_dir = tmp_path / ".muse"
641 muse_dir.mkdir()
642 history_file = muse_dir / "test_history.msgpack"
643 history_file.write_bytes(_pack({"version": 1, "runs": []}))
644
645 with patch.object(pathlib.Path, "stat") as mock_stat:
646 mock_stat.return_value = MagicMock(st_size=MAX_MSGPACK_BYTES + 1)
647 result = load_history(tmp_path)
648
649 assert result == []
650
651 def test_corrupt_history_returns_empty_list(self, tmp_path: pathlib.Path) -> None:
652 from muse.core.test_history import load_history
653
654 muse_dir = tmp_path / ".muse"
655 muse_dir.mkdir()
656 history_file = muse_dir / "test_history.msgpack"
657 history_file.write_bytes(b"\xde\xad\xbe\xef garbage")
658
659 result = load_history(tmp_path)
660 assert result == []
661
662
663 # ---------------------------------------------------------------------------
664 # 11. Callsite hardening — transport._decode
665 # ---------------------------------------------------------------------------
666
667 class TestTransportDecodeHardening:
668 """_decode (static method on HttpTransport) enforces per-value limits."""
669
670 def test_empty_response_returns_empty_dict(self) -> None:
671 from muse.core.transport import HttpTransport
672 result = HttpTransport._decode(b"")
673 assert result == {}
674
675 def test_valid_msgpack_decodes(self) -> None:
676 from muse.core.transport import HttpTransport
677 payload = _pack({"status": "ok", "count": 42})
678 result = HttpTransport._decode(payload)
679 assert result == {"status": "ok", "count": 42}
680
681 def test_string_over_limit_raises_transport_error(self) -> None:
682 from muse.core.transport import HttpTransport, TransportError
683 big_str = "B" * (1_048_577)
684 payload = _pack({"msg": big_str})
685 with pytest.raises(TransportError, match="invalid msgpack"):
686 HttpTransport._decode(payload)
687
688 def test_non_dict_top_level_returns_empty_dict(self) -> None:
689 from muse.core.transport import HttpTransport
690 payload = _pack([1, 2, 3])
691 result = HttpTransport._decode(payload)
692 assert result == {}
693
694 def test_invalid_msgpack_raises_transport_error(self) -> None:
695 from muse.core.transport import HttpTransport, TransportError
696 with pytest.raises(TransportError, match="invalid msgpack"):
697 HttpTransport._decode(b"\xff\xfe garbage")
698
699
700 # ---------------------------------------------------------------------------
701 # 12. Callsite hardening — stage.read_stage
702 # ---------------------------------------------------------------------------
703
704 class TestReadStageHardening:
705 """read_stage returns {} for oversized or corrupt staging index files."""
706
707 def test_oversized_stage_returns_empty(self, tmp_path: pathlib.Path) -> None:
708 from muse.plugins.code.stage import read_stage, stage_path
709
710 root = tmp_path
711 sp = stage_path(root)
712 sp.parent.mkdir(parents=True, exist_ok=True)
713 sp.write_bytes(_pack({"version": 2, "entries": {}}))
714
715 with patch.object(pathlib.Path, "stat") as mock_stat:
716 mock_stat.return_value = MagicMock(st_size=MAX_MSGPACK_BYTES + 1)
717 result = read_stage(root)
718
719 assert result == {}
720
721 def test_corrupt_stage_returns_empty(self, tmp_path: pathlib.Path) -> None:
722 from muse.plugins.code.stage import read_stage, stage_path
723
724 root = tmp_path
725 sp = stage_path(root)
726 sp.parent.mkdir(parents=True, exist_ok=True)
727 sp.write_bytes(b"\xfe\xed garbage")
728
729 result = read_stage(root)
730 assert result == {}
731
732 def test_non_dict_stage_returns_empty(self, tmp_path: pathlib.Path) -> None:
733 from muse.plugins.code.stage import read_stage, stage_path
734
735 root = tmp_path
736 sp = stage_path(root)
737 sp.parent.mkdir(parents=True, exist_ok=True)
738 sp.write_bytes(_pack([1, 2, 3])) # list, not dict
739
740 result = read_stage(root)
741 assert result == {}
742
743
744 # ---------------------------------------------------------------------------
745 # 13. TypeGuard narrowing — bundle and unpack_objects
746 # ---------------------------------------------------------------------------
747
748 class TestTypeGuardNarrowing:
749 """Non-dict entries in wire bundles are filtered, not propagated."""
750
751 def test_is_commit_dict_rejects_non_dicts(self) -> None:
752 from muse.cli.commands.bundle import _is_commit_dict
753
754 assert _is_commit_dict({}) is True
755 assert _is_commit_dict({"commit_id": "abc"}) is True
756 assert _is_commit_dict("string") is False
757 assert _is_commit_dict(42) is False
758 assert _is_commit_dict(None) is False
759 assert _is_commit_dict([]) is False
760
761 def test_is_snapshot_dict_rejects_non_dicts(self) -> None:
762 from muse.cli.commands.bundle import _is_snapshot_dict
763
764 assert _is_snapshot_dict({}) is True
765 assert _is_snapshot_dict({"snapshot_id": "abc"}) is True
766 assert _is_snapshot_dict("string") is False
767 assert _is_snapshot_dict(42) is False
768 assert _is_snapshot_dict(None) is False
769
770 def test_unpack_objects_is_commit_dict(self) -> None:
771 from muse.cli.commands.plumbing.unpack_objects import _is_commit_dict
772
773 assert _is_commit_dict({"commit_id": "abc"}) is True
774 assert _is_commit_dict("not a dict") is False
775
776 def test_unpack_objects_is_snapshot_dict(self) -> None:
777 from muse.cli.commands.plumbing.unpack_objects import _is_snapshot_dict
778
779 assert _is_snapshot_dict({"snapshot_id": "abc"}) is True
780 assert _is_snapshot_dict(99) is False
781
782 def test_as_branch_heads_filters_non_str_values(self) -> None:
783 from muse.cli.commands.plumbing.unpack_objects import _as_branch_heads
784
785 result = _as_branch_heads({"main": "abc123", "bad": 42, "ok": "def456"})
786 assert result == {"main": "abc123", "ok": "def456"}
787
788 def test_as_branch_heads_non_dict_input(self) -> None:
789 from muse.cli.commands.plumbing.unpack_objects import _as_branch_heads
790
791 assert _as_branch_heads(None) == {}
792 assert _as_branch_heads("string") == {}
793 assert _as_branch_heads([]) == {}
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 101 days ago