gabriel / musehub public
test_mpack_content_scanning_phase3.py python
485 lines 17.1 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 3: content scanning and DMCA takedown (issue #49).
2
3 Phase 3 invariants:
4 3a. Known-hash blocklist — mpack.index quarantines any mpack whose objects
5 appear in musehub_blocked_hashes, before any MinIO writes.
6 3b. content.scan job infrastructure — after indexing, binary objects above
7 the scan threshold get a content.scan job enqueued.
8 3c. DMCA takedown — POST /api/admin/takedown adds hashes to the blocklist,
9 moves existing MinIO objects to quarantine, marks repos with dmca_hold.
10 Requires is_admin=True; non-admins receive 403.
11 """
12 from __future__ import annotations
13
14 import datetime
15 import pathlib
16
17 import msgpack
18 import pytest
19 import pytest_asyncio
20 from httpx import AsyncClient, ASGITransport
21 from sqlalchemy import select
22 from sqlalchemy.ext.asyncio import AsyncSession
23 from unittest.mock import patch
24
25 from musehub.auth.request_signing import MSignContext, require_signed_request, optional_signed_request
26 from musehub.db.musehub_abuse_models import MusehubBlockedHash
27 from musehub.db.musehub_jobs_models import MusehubBackgroundJob
28 from musehub.db.musehub_repo_models import MusehubRepo
29 from musehub.db.database import get_db
30 from musehub.main import app
31
32 from muse.core.object_store import write_object
33 from muse.core.mpack import build_mpack, build_wire_mpack
34 from muse.core.paths import muse_dir
35 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
36 from muse.core.commits import CommitRecord, write_commit
37 from muse.core.refs import write_branch_ref
38 from muse.core.snapshots import SnapshotRecord, write_snapshot
39 from muse.core.types import blob_id
40 from musehub.types.json_types import JSONObject
41
42 pytestmark = pytest.mark.wire
43
44
45 _AUTH_CTX = MSignContext(
46 handle="gabriel",
47 identity_id="sha256:" + "0" * 64,
48 is_agent=False,
49 is_admin=True,
50 )
51
52 _NON_ADMIN_CTX = MSignContext(
53 handle="carol",
54 identity_id="sha256:" + "1" * 64,
55 is_agent=False,
56 is_admin=False,
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 non_admin_client(db_session: AsyncSession) -> None:
87 async def _override_get_db() -> None:
88 yield db_session
89
90 app.dependency_overrides[get_db] = _override_get_db
91 app.dependency_overrides[require_signed_request] = lambda: _NON_ADMIN_CTX
92 app.dependency_overrides[optional_signed_request] = lambda: _NON_ADMIN_CTX
93
94 async with AsyncClient(
95 transport=ASGITransport(app=app),
96 base_url="https://localhost:1337",
97 ) as c:
98 yield c
99
100 app.dependency_overrides.clear()
101
102
103 @pytest_asyncio.fixture()
104 async def repo(client: AsyncClient) -> None:
105 resp = await client.post(
106 "/api/repos",
107 json={"owner": "gabriel", "name": "phase3-scan-test", "visibility": "public", "initialize": False},
108 )
109 assert resp.status_code in (200, 201), resp.text
110 data = resp.json()
111 yield data
112 await client.delete(f"/api/repos/{data['repoId']}")
113
114
115 def _make_repo(tmp: pathlib.Path) -> tuple[pathlib.Path, str, dict]:
116 tmp.mkdir(parents=True, exist_ok=True)
117 dot = muse_dir(tmp)
118 dot.mkdir()
119 (dot / "repo.json").write_text('{"repo_id":"phase3-test","owner":"gabriel"}')
120 for d in ("commits", "snapshots", "objects"):
121 (dot / d).mkdir()
122 (dot / "refs" / "heads").mkdir(parents=True)
123 (dot / "HEAD").write_text("ref: refs/heads/main\n")
124 (dot / "config.toml").write_text("")
125
126 blob_ids: list[str] = []
127 for i in range(_N_FILES):
128 data = f"phase3-base-{i:04d}".encode() + b"x" * _BLOB_SIZE
129 oid = blob_id(data)
130 write_object(tmp, oid, data)
131 blob_ids.append(oid)
132
133 base_manifest = {f"src/file_{i:04d}.py": blob_ids[i] for i in range(_N_FILES)}
134 parent = None
135 tip = ""
136 ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
137
138 for i in range(_N_COMMITS):
139 manifest = dict(base_manifest)
140 for j in range(_FILES_CHANGED):
141 idx = (i * _FILES_CHANGED + j) % _N_FILES
142 raw = f"phase3-c{i:04d}-f{j}".encode() + b"y" * _BLOB_SIZE
143 oid = blob_id(raw)
144 write_object(tmp, oid, raw)
145 manifest[f"src/file_{idx:04d}.py"] = oid
146
147 sid = compute_snapshot_id(manifest)
148 write_snapshot(tmp, SnapshotRecord(snapshot_id=sid, manifest=manifest))
149 msg = f"commit-{i:05d}"
150 cid = compute_commit_id(
151 parent_ids=[parent] if parent else [],
152 snapshot_id=sid,
153 message=msg,
154 committed_at_iso=ts.isoformat(),
155 author="gabriel",
156 )
157 write_commit(tmp, CommitRecord(
158 commit_id=cid, branch="main",
159 snapshot_id=sid, message=msg, committed_at=ts,
160 parent_commit_id=parent, parent2_commit_id=None,
161 author="gabriel", metadata={}, structured_delta=None,
162 sem_ver_bump="none", breaking_changes=[],
163 agent_id="", model_id="", toolchain_id="",
164 prompt_hash="", signature="", signer_key_id="",
165 ))
166 parent = cid
167 tip = cid
168 ts += datetime.timedelta(seconds=60)
169
170 write_branch_ref(tmp, "main", tip)
171 mpack = build_mpack(tmp, [tip], have=[])
172 return tmp, tip, mpack
173
174
175 async def _push_mpack(client: AsyncClient, repo_slug: str, mpack: dict, head: str) -> None:
176 """Store mpack in MemoryBackend and call unpack-mpack. Raises AssertionError on non-200."""
177 import musehub.storage.backends as _backends_mod
178 from muse.core.types import blob_id as _blob_id
179
180 wire_bytes = build_wire_mpack(mpack)
181 mpack_key = _blob_id(wire_bytes)
182
183 backend = _backends_mod.get_backend()
184 await backend.put_mpack(mpack_key, wire_bytes)
185
186 n_commits = len(mpack.get("commits") or [])
187 n_blobs = len(mpack.get("blobs") or [])
188
189 resp = await client.post(
190 f"/gabriel/{repo_slug}/push/unpack-mpack",
191 content=msgpack.packb(
192 {
193 "mpack_key": mpack_key,
194 "branch": "main",
195 "head": head,
196 "commits_count": n_commits,
197 "blobs_count": n_blobs,
198 },
199 use_bin_type=True,
200 ),
201 headers={"Content-Type": "application/x-msgpack"},
202 )
203 assert resp.status_code == 200, f"push failed: {resp.status_code} {resp.text}"
204
205
206 async def _push_mpack_expect_blocked(client: AsyncClient, repo_slug: str, mpack: dict, head: str) -> dict:
207 """Store mpack in MemoryBackend and call unpack-mpack. Returns response for 422."""
208 import musehub.storage.backends as _backends_mod
209 from muse.core.types import blob_id as _blob_id
210
211 wire_bytes = build_wire_mpack(mpack)
212 mpack_key = _blob_id(wire_bytes)
213
214 backend = _backends_mod.get_backend()
215 await backend.put_mpack(mpack_key, wire_bytes)
216
217 n_commits = len(mpack.get("commits") or [])
218 n_blobs = len(mpack.get("blobs") or [])
219
220 resp = await client.post(
221 f"/gabriel/{repo_slug}/push/unpack-mpack",
222 content=msgpack.packb(
223 {
224 "mpack_key": mpack_key,
225 "branch": "main",
226 "head": head,
227 "commits_count": n_commits,
228 "blobs_count": n_blobs,
229 },
230 use_bin_type=True,
231 ),
232 headers={"Content-Type": "application/x-msgpack"},
233 )
234 return resp
235
236
237 # ── 3a: known-hash blocklist ─────────────────────────────────────────────────
238
239 @pytest.mark.asyncio
240 async def test_blocked_object_quarantines_mpack(
241 client: AsyncClient, repo: JSONObject, tmp_path: pathlib.Path, db_session: AsyncSession,
242 ) -> None:
243 """A push containing a blocked object is rejected with 422 before any writes.
244
245 The blocklist check runs inside wire_push_unpack_mpack before object writes.
246 """
247 _, head, mpack = _make_repo(tmp_path / "repo")
248
249 blobs = mpack.get("blobs") or []
250 assert blobs, "mpack must have blobs"
251 blocked_oid = blobs[0]["object_id"]
252
253 db_session.add(MusehubBlockedHash(object_id=blocked_oid, reason="test NCMEC block"))
254 await db_session.flush()
255
256 resp = await _push_mpack_expect_blocked(client, repo["slug"], mpack, head)
257 assert resp.status_code == 422, (
258 f"blocked object must cause 422, got {resp.status_code}: {resp.text}"
259 )
260 assert "blocked" in resp.text.lower(), (
261 f"response body must mention 'blocked', got: {resp.text}"
262 )
263
264
265 @pytest.mark.asyncio
266 async def test_blocked_check_fires_before_minio_puts(
267 client: AsyncClient, repo: JSONObject, tmp_path: pathlib.Path, db_session: AsyncSession,
268 ) -> None:
269 """Blocklist check fires before backend.put — zero object writes on blocked mpack.
270
271 The order enforced by wire_push_unpack_mpack: blocklist check → object writes.
272 When a blocked hash is detected, backend.put for individual objects must never run.
273 """
274 import musehub.storage.backends as _backends_mod
275
276 _, head, mpack = _make_repo(tmp_path / "repo")
277
278 blobs = mpack.get("blobs") or []
279 blocked_oid = blobs[0]["object_id"]
280 db_session.add(MusehubBlockedHash(object_id=blocked_oid, reason="test"))
281 await db_session.flush()
282
283 backend = _backends_mod.get_backend()
284 put_calls: list[str] = []
285 _real_put = backend.put
286
287 async def _spy(oid: str, data: bytes) -> None:
288 put_calls.append(oid)
289 return await _real_put(oid, data)
290
291 with patch.object(backend, "put", side_effect=_spy):
292 resp = await _push_mpack_expect_blocked(client, repo["slug"], mpack, head)
293
294 assert resp.status_code == 422
295 assert not put_calls, (
296 f"backend.put was called {len(put_calls)} time(s) despite blocked hash — "
297 f"blocklist check must run before object writes"
298 )
299
300
301 @pytest.mark.asyncio
302 async def test_multiple_blocked_objects_all_reported(
303 client: AsyncClient, repo: JSONObject, tmp_path: pathlib.Path, db_session: AsyncSession,
304 ) -> None:
305 """A mpack with multiple blocked objects is rejected with a 422 mentioning 'blocked'.
306
307 Any blocked object in the mpack is sufficient to reject the entire push.
308 """
309 _, head, mpack = _make_repo(tmp_path / "repo")
310
311 blobs = mpack.get("blobs") or []
312 assert len(blobs) >= 2, "need at least 2 blobs to test multi-blocked"
313 blocked_oids = [blobs[0]["object_id"], blobs[1]["object_id"]]
314 for oid in blocked_oids:
315 db_session.add(MusehubBlockedHash(object_id=oid, reason="test"))
316 await db_session.flush()
317
318 resp = await _push_mpack_expect_blocked(client, repo["slug"], mpack, head)
319 assert resp.status_code == 422
320 assert "blocked" in resp.text.lower(), (
321 f"response must mention 'blocked', got: {resp.text}"
322 )
323
324
325 @pytest.mark.asyncio
326 async def test_clean_mpack_bypasses_blocklist_unimpeded(
327 client: AsyncClient, repo: JSONObject, tmp_path: pathlib.Path, db_session: AsyncSession,
328 ) -> None:
329 """A mpack with no blocked objects pushes successfully despite a non-matching blocklist.
330
331 Blocklist entries for other objects must not affect clean pushes.
332 """
333 _, head, mpack = _make_repo(tmp_path / "repo")
334
335 # Block a hash that is NOT in this mpack
336 db_session.add(MusehubBlockedHash(object_id="sha256:" + "f" * 64, reason="unrelated block"))
337 await db_session.flush()
338
339 await _push_mpack(client, repo["slug"], mpack, head)
340
341
342 # ── 3c: DMCA takedown endpoint ───────────────────────────────────────────────
343
344 @pytest.mark.asyncio
345 async def test_dmca_takedown_adds_hashes_to_blocklist(
346 client: AsyncClient, tmp_path: pathlib.Path, db_session: AsyncSession,
347 ) -> None:
348 """POST /api/admin/takedown adds object_ids to musehub_blocked_hashes.
349
350 After a successful takedown request, each supplied object_id must appear
351 in musehub_blocked_hashes with the supplied reason.
352 """
353 oids = [
354 blob_id(f"dmca-object-{i}".encode())
355 for i in range(3)
356 ]
357 resp = await client.post(
358 "/api/admin/takedown",
359 json={"object_ids": oids, "reason": "DMCA request #12345"},
360 )
361 assert resp.status_code == 200, resp.text
362
363 db_session.expire_all()
364 rows = (await db_session.execute(
365 select(MusehubBlockedHash).where(MusehubBlockedHash.object_id.in_(oids))
366 )).scalars().all()
367 assert len(rows) == len(oids), (
368 f"expected {len(oids)} blocked_hash rows, got {len(rows)}"
369 )
370 for row in rows:
371 assert "DMCA" in (row.reason or ""), (
372 f"expected reason to contain 'DMCA', got {row.reason!r}"
373 )
374
375
376 @pytest.mark.asyncio
377 async def test_dmca_takedown_marks_repos_dmca_hold(
378 client: AsyncClient, repo: JSONObject, tmp_path: pathlib.Path, db_session: AsyncSession,
379 ) -> None:
380 """POST /api/admin/takedown with repo_ids sets dmca_hold=True on each repo.
381
382 A repo under dmca_hold must have its flag persisted so push gates and
383 serve paths can enforce the hold.
384 """
385 repo_id = repo["repoId"]
386 oids = [blob_id(b"dmca-hold-test")]
387
388 resp = await client.post(
389 "/api/admin/takedown",
390 json={"object_ids": oids, "reason": "DMCA #hold-test", "repo_ids": [repo_id]},
391 )
392 assert resp.status_code == 200, resp.text
393 data = resp.json()
394 assert data.get("repos_held", 0) == 1, (
395 f"expected repos_held=1 in response, got: {data}"
396 )
397
398 db_session.expire_all()
399 repo_row = (await db_session.execute(
400 select(MusehubRepo).where(MusehubRepo.repo_id == repo_id)
401 )).scalar_one()
402 assert repo_row.dmca_hold is True, (
403 f"expected repo.dmca_hold=True after takedown with repo_ids, got {repo_row.dmca_hold}"
404 )
405
406
407 @pytest.mark.asyncio
408 async def test_dmca_takedown_quarantines_existing_minio_objects(
409 client: AsyncClient, repo: JSONObject, tmp_path: pathlib.Path, db_session: AsyncSession,
410 ) -> None:
411 """POST /api/admin/takedown moves already-stored MinIO objects to quarantine.
412
413 Objects that were written to MinIO before the takedown must be moved to the
414 quarantine prefix so they are no longer publicly fetchable.
415 """
416 # Write a real object to MinIO first (simulates a previously indexed object)
417 from musehub.storage.backends import get_backend as _get_backend
418 backend = _get_backend()
419
420 obj_data = b"sensitive-content-to-quarantine-" + b"x" * 100
421 oid = blob_id(obj_data)
422 await backend.put(oid, obj_data)
423
424 # Verify it exists in MinIO before takedown
425 assert await backend.get(oid) is not None, "object must be in MinIO before takedown"
426
427 resp = await client.post(
428 "/api/admin/takedown",
429 json={"object_ids": [oid], "reason": "DMCA quarantine test"},
430 )
431 assert resp.status_code == 200, resp.text
432 data = resp.json()
433 assert data.get("quarantined_count", 0) >= 1, (
434 f"expected quarantined_count >= 1 in response, got: {data}"
435 )
436
437 # Object must no longer be accessible from the main bucket
438 assert await backend.get(oid) is None, (
439 "object must be removed from main MinIO bucket after DMCA takedown"
440 )
441
442
443 @pytest.mark.asyncio
444 async def test_dmca_takedown_requires_admin(
445 non_admin_client: AsyncClient, db_session: AsyncSession,
446 ) -> None:
447 """POST /api/admin/takedown returns 403 for non-admin callers.
448
449 The takedown endpoint is admin-only — a non-admin identity must receive
450 403 Forbidden, not 401 (auth header is valid, permissions are insufficient).
451 """
452 resp = await non_admin_client.post(
453 "/api/admin/takedown",
454 json={"object_ids": ["sha256:" + "a" * 64], "reason": "test"},
455 )
456 assert resp.status_code == 403, (
457 f"expected 403 for non-admin takedown request, got {resp.status_code}: {resp.text}"
458 )
459
460
461 @pytest.mark.asyncio
462 async def test_dmca_takedown_idempotent(
463 client: AsyncClient, db_session: AsyncSession,
464 ) -> None:
465 """POST /api/admin/takedown is idempotent — re-blocking an already-blocked hash is safe.
466
467 Calling takedown twice with the same object_ids must not raise an error or
468 create duplicate rows.
469 """
470 oid = blob_id(b"idempotent-takedown")
471 payload = {"object_ids": [oid], "reason": "idempotent test"}
472
473 resp1 = await client.post("/api/admin/takedown", json=payload)
474 assert resp1.status_code == 200, resp1.text
475
476 resp2 = await client.post("/api/admin/takedown", json=payload)
477 assert resp2.status_code == 200, resp2.text
478
479 db_session.expire_all()
480 rows = (await db_session.execute(
481 select(MusehubBlockedHash).where(MusehubBlockedHash.object_id == oid)
482 )).scalars().all()
483 assert len(rows) == 1, f"expected exactly 1 blocked_hash row after idempotent calls, got {len(rows)}"
484
485
File History 1 commit
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ 49 days ago