gabriel / musehub public
test_mpack_validation_phase2.py python
415 lines 15.4 KB
Raw
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ breaking 49 days ago
1 """TDD — Phase 2: mpack content validation.
2
3 Validation invariants (issue #49) live in wire_push_unpack_mpack (the
4 synchronous push path), not in the background indexing job. The background
5 job (process_mpack_index_job) trusts the mpack_key integrity proof and does
6 no per-object or decompression validation.
7
8 Phase 2 invariants tested here:
9 2a. Decompression size guard (zip bomb) — wire_push_unpack_mpack returns 422
10 when cumulative decompressed bytes exceeds settings.mpack_max_decompressed_bytes.
11 2b. Hash integrity gate — unpack-mpack returns 422 when the stored bytes do
12 not hash to the claimed mpack_key.
13 2c. Quarantine state — on zip bomb: backend.quarantine_mpack removes the mpack
14 from normal storage.
15 2d. No commits written on validation failure — zip bomb fires before any DB writes.
16 2e. quarantine_job — standalone utility that marks a background job as quarantined.
17 2f. Valid mpack — process_mpack_index_job completes normally for a well-formed mpack.
18 """
19 from __future__ import annotations
20
21 import copy
22 import datetime
23 import hashlib
24 import pathlib
25
26 import msgpack
27 import pytest
28 import pytest_asyncio
29
30 from httpx import AsyncClient, ASGITransport
31 from sqlalchemy import select
32 from sqlalchemy.ext.asyncio import AsyncSession
33 from unittest.mock import patch
34
35 from musehub.auth.request_signing import MSignContext, require_signed_request, optional_signed_request
36 from musehub.db.musehub_jobs_models import MusehubBackgroundJob
37 from musehub.db.musehub_repo_models import MusehubCommit, MusehubRepo
38 from musehub.db.database import get_db
39 from musehub.main import app
40
41 from muse.core.object_store import write_object
42 from muse.core.mpack import build_mpack, build_wire_mpack
43 from muse.core.paths import muse_dir
44 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
45 from muse.core.commits import CommitRecord, write_commit
46 from muse.core.refs import write_branch_ref
47 from muse.core.snapshots import SnapshotRecord, write_snapshot
48 from muse.core.types import blob_id
49 from musehub.types.json_types import JSONObject
50
51
52 _AUTH_CTX = MSignContext(
53 handle="gabriel",
54 identity_id="sha256:" + "0" * 64,
55 is_agent=False,
56 is_admin=True,
57 )
58
59 _N_FILES = 8
60 _N_COMMITS = 4
61 _FILES_CHANGED = 2
62 _BLOB_SIZE = 128
63
64
65 # ── fixtures ────────────────────────────────────────────────────────────────
66
67 @pytest_asyncio.fixture()
68 async def client(db_session: AsyncSession) -> None:
69 async def _override_get_db() -> None:
70 yield db_session
71
72 app.dependency_overrides[get_db] = _override_get_db
73 app.dependency_overrides[require_signed_request] = lambda: _AUTH_CTX
74 app.dependency_overrides[optional_signed_request] = lambda: _AUTH_CTX
75
76 async with AsyncClient(
77 transport=ASGITransport(app=app),
78 base_url="https://localhost:1337",
79 ) as c:
80 yield c
81
82 app.dependency_overrides.clear()
83
84
85 @pytest_asyncio.fixture()
86 async def repo(client: AsyncClient) -> None:
87 resp = await client.post(
88 "/api/repos",
89 json={"owner": "gabriel", "name": "phase2-validation-test", "visibility": "public", "initialize": False},
90 )
91 assert resp.status_code in (200, 201), resp.text
92 data = resp.json()
93 yield data["slug"]
94 await client.delete(f"/api/repos/{data['repoId']}")
95
96
97 def _make_repo(tmp: pathlib.Path) -> tuple[pathlib.Path, str, dict]:
98 tmp.mkdir(parents=True, exist_ok=True)
99 dot = muse_dir(tmp)
100 dot.mkdir()
101 (dot / "repo.json").write_text('{"repo_id":"phase2-test","owner":"gabriel"}')
102 for d in ("commits", "snapshots", "objects"):
103 (dot / d).mkdir()
104 (dot / "refs" / "heads").mkdir(parents=True)
105 (dot / "HEAD").write_text("ref: refs/heads/main\n")
106 (dot / "config.toml").write_text("")
107
108 blob_ids: list[str] = []
109 for i in range(_N_FILES):
110 data = f"base-{i:04d}".encode() + b"x" * _BLOB_SIZE
111 oid = blob_id(data)
112 write_object(tmp, oid, data)
113 blob_ids.append(oid)
114
115 base_manifest = {f"src/file_{i:04d}.py": blob_ids[i] for i in range(_N_FILES)}
116 parent = None
117 tip = ""
118 ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
119
120 for i in range(_N_COMMITS):
121 manifest = dict(base_manifest)
122 for j in range(_FILES_CHANGED):
123 idx = (i * _FILES_CHANGED + j) % _N_FILES
124 raw = f"c{i:04d}-f{j}".encode() + b"y" * _BLOB_SIZE
125 oid = blob_id(raw)
126 write_object(tmp, oid, raw)
127 manifest[f"src/file_{idx:04d}.py"] = oid
128
129 sid = compute_snapshot_id(manifest)
130 write_snapshot(tmp, SnapshotRecord(snapshot_id=sid, manifest=manifest))
131 msg = f"commit-{i:05d}"
132 cid = compute_commit_id(
133 parent_ids=[parent] if parent else [],
134 snapshot_id=sid,
135 message=msg,
136 committed_at_iso=ts.isoformat(),
137 author="gabriel",
138 )
139 write_commit(tmp, CommitRecord(
140 commit_id=cid, branch="main",
141 snapshot_id=sid, message=msg, committed_at=ts,
142 parent_commit_id=parent, parent2_commit_id=None,
143 author="gabriel", metadata={}, structured_delta=None,
144 sem_ver_bump="none", breaking_changes=[],
145 agent_id="", model_id="", toolchain_id="",
146 prompt_hash="", signature="", signer_key_id="",
147 ))
148 parent = cid
149 tip = cid
150 ts += datetime.timedelta(seconds=60)
151
152 write_branch_ref(tmp, "main", tip)
153 mpack = build_mpack(tmp, [tip], have=[])
154 return tmp, tip, mpack
155
156
157 async def _store_mpack_for_index_job(
158 repo_slug: str,
159 mpack: dict,
160 head: str,
161 db_session: AsyncSession,
162 ) -> str:
163 """Store raw msgpack in MemoryBackend and create a mpack.index job row. Returns job_id."""
164 import musehub.storage.backends as _backends_mod
165 from datetime import datetime, timezone
166 from musehub.core.genesis import compute_job_id as _compute_job_id
167
168 repo_row = (await db_session.execute(
169 select(MusehubRepo).where(MusehubRepo.slug == repo_slug)
170 )).scalar_one()
171 repo_id = repo_row.repo_id
172
173 wire_bytes = msgpack.packb(mpack, use_bin_type=True)
174 mpack_key = "sha256:" + hashlib.sha256(wire_bytes).hexdigest()
175 n_objects = len(mpack.get("blobs") or [])
176
177 backend = _backends_mod.get_backend()
178 await backend.put_mpack(mpack_key, wire_bytes)
179
180 now = datetime.now(tz=timezone.utc)
181 job_id = _compute_job_id(repo_id, "mpack.index", now.isoformat())
182 db_session.add(MusehubBackgroundJob(
183 job_id=job_id,
184 repo_id=repo_id,
185 job_type="mpack.index",
186 payload={
187 "mpack_key": mpack_key,
188 "pusher_id": "sha256:" + "0" * 64,
189 "branch": "main",
190 "head": head,
191 "force": False,
192 "declared_objects_count": n_objects,
193 },
194 status="pending",
195 created_at=now,
196 attempt=0,
197 ))
198 await db_session.flush()
199 return job_id
200
201
202 async def _store_wire_mpack_for_http(
203 wire_bytes: bytes,
204 ) -> str:
205 """Store MUSE binary mpack in MemoryBackend. Returns mpack_key (blob_id).
206
207 For tests that call the unpack-mpack HTTP endpoint — the route's integrity
208 check uses blob_id (sha256("blob <size>\\0<data>")), so the key must match.
209 """
210 import musehub.storage.backends as _backends_mod
211 mpack_key = blob_id(wire_bytes)
212 backend = _backends_mod.get_backend()
213 await backend.put_mpack(mpack_key, wire_bytes)
214 return mpack_key
215
216
217 async def _call_unpack_mpack(
218 client: AsyncClient,
219 repo_slug: str,
220 mpack_key: str,
221 head: str,
222 ) -> dict:
223 """Call the unpack-mpack HTTP endpoint and return the parsed response."""
224 resp = await client.post(
225 f"/gabriel/{repo_slug}/push/unpack-mpack",
226 content=msgpack.packb(
227 {"mpack_key": mpack_key, "branch": "main", "head": head},
228 use_bin_type=True,
229 ),
230 headers={"Content-Type": "application/x-msgpack"},
231 )
232 return resp
233
234
235 # ── Phase 2 tests ───────────────────────────────────────────────────────────
236
237 @pytest.mark.asyncio
238 async def test_hash_mismatch_caught_at_unpack_mpack(
239 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
240 ) -> None:
241 """Mpack-level integrity gate: bytes that don't hash to mpack_key → 422.
242
243 The per-mpack blob_id check lives in wire_push_unpack_mpack. If the stored
244 bytes hash to Y but the client claims key X, unpack-mpack rejects with 422.
245 Per-object hash checking was intentionally removed: blob_id(wire_bytes) ==
246 mpack_key already authenticates every byte in the mpack.
247 """
248 _, head, raw_mpack = _make_repo(tmp_path / "repo")
249 wire_bytes = build_wire_mpack(raw_mpack)
250
251 # PUT real bytes under a deliberately wrong key (all aaaa...).
252 wrong_key = "sha256:" + "a" * 64
253 import musehub.storage.backends as _backends_mod
254 backend = _backends_mod.get_backend()
255 await backend.put_mpack(wrong_key, wire_bytes)
256
257 resp = await _call_unpack_mpack(client, repo, wrong_key, head)
258 assert resp.status_code == 422, resp.text
259 assert "integrity" in resp.text.lower() or "sha256" in resp.text.lower() or "mismatch" in resp.text.lower()
260
261
262 @pytest.mark.asyncio
263 async def test_zip_bomb_rejected_at_unpack_mpack(
264 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
265 ) -> None:
266 """Zip bomb guard triggers in wire_push_unpack_mpack → 422.
267
268 Patching mpack_max_decompressed_bytes to 1 forces the guard on the first
269 blob. The check fires before any DB or storage writes.
270 """
271 _, head, raw_mpack = _make_repo(tmp_path / "repo")
272 wire_bytes = build_wire_mpack(raw_mpack)
273 mpack_key = await _store_wire_mpack_for_http(wire_bytes)
274
275 from musehub.config import settings as _real_settings
276
277 class _TinyLimitSettings:
278 mpack_max_decompressed_bytes = 1
279 def __getattr__(self, name: str):
280 return getattr(_real_settings, name)
281
282 with patch("musehub.services.musehub_wire_push.settings", _TinyLimitSettings()):
283 resp = await _call_unpack_mpack(client, repo, mpack_key, head)
284
285 assert resp.status_code == 422, (
286 f"expected 422 on zip bomb, got {resp.status_code}: {resp.text[:200]}"
287 )
288 assert "decompressed" in resp.text.lower() or "zip" in resp.text.lower() or "limit" in resp.text.lower(), (
289 f"422 body does not mention decompression limit: {resp.text[:200]}"
290 )
291
292
293 @pytest.mark.asyncio
294 async def test_zip_bomb_quarantines_mpack_in_storage(
295 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
296 ) -> None:
297 """After a zip bomb rejection, the mpack is quarantined (removed from normal storage).
298
299 wire_push_unpack_mpack calls backend.quarantine_mpack before raising
300 MPackValidationError, so the mpack bytes are no longer accessible at the
301 original key.
302 """
303 _, head, raw_mpack = _make_repo(tmp_path / "repo")
304 wire_bytes = build_wire_mpack(raw_mpack)
305 mpack_key = await _store_wire_mpack_for_http(wire_bytes)
306
307 import musehub.storage.backends as _backends_mod
308 backend = _backends_mod.get_backend()
309 assert await backend.exists_mpack(mpack_key), "mpack must exist before the call"
310
311 from musehub.config import settings as _real_settings
312
313 class _TinyLimitSettings:
314 mpack_max_decompressed_bytes = 1
315 def __getattr__(self, name: str):
316 return getattr(_real_settings, name)
317
318 with patch("musehub.services.musehub_wire_push.settings", _TinyLimitSettings()):
319 resp = await _call_unpack_mpack(client, repo, mpack_key, head)
320
321 assert resp.status_code == 422
322 assert not await backend.exists_mpack(mpack_key), (
323 "mpack must be quarantined (removed from normal storage) after zip bomb rejection"
324 )
325
326
327 @pytest.mark.asyncio
328 async def test_zip_bomb_leaves_no_commit_rows(
329 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
330 ) -> None:
331 """Zip bomb fires before any DB writes — no commits are inserted.
332
333 The decompression guard runs at step 5 of wire_push_unpack_mpack, before
334 the commit/snapshot/object write phase. A rejected mpack must leave the DB
335 in the same state as before the call.
336 """
337 _, head, raw_mpack = _make_repo(tmp_path / "repo")
338 wire_bytes = build_wire_mpack(raw_mpack)
339 mpack_key = await _store_wire_mpack_for_http(wire_bytes)
340
341 raw_commits = raw_mpack.get("commits") or []
342 all_cids = [c["commit_id"] for c in raw_commits if "commit_id" in c]
343 assert all_cids, "mpack must contain commits for this test to be meaningful"
344
345 from musehub.config import settings as _real_settings
346
347 class _TinyLimitSettings:
348 mpack_max_decompressed_bytes = 1
349 def __getattr__(self, name: str):
350 return getattr(_real_settings, name)
351
352 with patch("musehub.services.musehub_wire_push.settings", _TinyLimitSettings()):
353 resp = await _call_unpack_mpack(client, repo, mpack_key, head)
354
355 assert resp.status_code == 422
356
357 rows = (await db_session.execute(
358 select(MusehubCommit).where(MusehubCommit.commit_id.in_(all_cids))
359 )).scalars().all()
360 assert not rows, (
361 f"{len(rows)} commit rows were inserted despite zip bomb rejection: "
362 f"{[r.commit_id[:16] for r in rows[:3]]}"
363 )
364
365
366 @pytest.mark.asyncio
367 async def test_quarantine_job_sets_status_and_reason(
368 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
369 ) -> None:
370 """quarantine_job sets status='quarantined' and stores the reason on a job row.
371
372 quarantine_job is the worker-level utility that marks a background job as
373 quarantined after any validation or integrity failure.
374 """
375 _, head, mpack = _make_repo(tmp_path / "repo")
376 job_id = await _store_mpack_for_index_job(repo, mpack, head, db_session)
377
378 from musehub.services.musehub_jobs import quarantine_job
379
380 reason = "object sha256:deadbeef content does not match declared id"
381 await quarantine_job(db_session, job_id, reason)
382 await db_session.commit()
383
384 db_session.expire_all()
385 job_row = (await db_session.execute(
386 select(MusehubBackgroundJob).where(MusehubBackgroundJob.job_id == job_id)
387 )).scalar_one()
388
389 assert job_row.status == "quarantined", (
390 f"expected status 'quarantined' after quarantine_job, got '{job_row.status}'"
391 )
392 assert job_row.quarantine_reason is not None, "quarantine_reason must be set"
393 assert reason[:50] in job_row.quarantine_reason, (
394 f"quarantine_reason {job_row.quarantine_reason!r} does not contain the error message"
395 )
396
397
398 @pytest.mark.asyncio
399 async def test_valid_mpack_passes_all_validation_checks(
400 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession,
401 ) -> None:
402 """Regression: a well-formed mpack passes all Phase 2 checks and is indexed.
403
404 All blobs have correct IDs and decompressed size is within limit —
405 process_mpack_index_job must complete normally and return positive counts.
406 """
407 _, head, mpack = _make_repo(tmp_path / "repo")
408 job_id = await _store_mpack_for_index_job(repo, mpack, head, db_session)
409
410 from musehub.services.musehub_wire import process_mpack_index_job
411 result = await process_mpack_index_job(db_session, job_id)
412 await db_session.commit()
413
414 assert result["commit_graph_written"] > 0, f"expected commits indexed for a valid mpack, got {result}"
415 assert result["mpack_index_written"] > 0, f"expected objects indexed for a valid mpack, got {result}"
File History 1 commit
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ 49 days ago