gabriel / musehub public
test_mist_security.py python
678 lines 25.6 KB
Raw
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ breaking 156 days ago
1 """Section 17 — Mists Security Hardening: adversarial and access-control tests.
2
3 Covers every security property stated in the Phase 8 spec:
4
5 Filename attacks Path traversal, null bytes, control characters,
6 ANSI escape sequences, path separators — all must
7 return 422 Unprocessable Entity.
8
9 Content injection HTML/JS payloads stored verbatim; no server-side
10 sanitisation that would alter or strip content.
11 The embedding contract lives at the template layer
12 (Jinja2 auto-escapes {{ mist.content | e }}).
13
14 Access control Secret mists invisible to non-owners in list and
15 explore endpoints; non-owner update/delete blocked;
16 secret mist detail returns 403 to prevent leaking
17 existence.
18
19 Collision resistance Identical bytes → identical mist_id → POST returns
20 409 Conflict (idempotent, not an error the caller
21 should retry).
22
23 Large content Body ≥ 11 MiB triggers ContentSizeLimitMiddleware
24 → 413 Request Entity Too Large before any DB write.
25
26 Fork depth Fork chain capped at 5 levels; attempting to fork
27 a depth-5 mist returns 422.
28 """
29 from __future__ import annotations
30
31 import uuid
32
33 import pytest
34 from httpx import AsyncClient
35 from sqlalchemy.ext.asyncio import AsyncSession
36
37 from musehub.types.json_types import JSONObject
38
39 _OWNER = "testuser" # matches conftest._TEST_HANDLE
40 _OTHER = "otheruser"
41
42 _PY_CONTENT = "def hello():\n return 'hello world'\n"
43
44
45 def _payload(**overrides: object) -> JSONObject:
46 base: JSONObject = {
47 "filename": f"sec_{uuid.uuid4().hex[:8]}.py",
48 "content": _PY_CONTENT + uuid.uuid4().hex, # unique content per call
49 "visibility": "public",
50 }
51 base.update(overrides)
52 return base
53
54
55 async def _create(client: AsyncClient, headers: dict, **overrides: object) -> JSONObject:
56 r = await client.post("/api/mists", json=_payload(**overrides), headers=headers)
57 assert r.status_code == 201, r.text
58 return dict(r.json())
59
60
61 # ═══════════════════════════════════════════════════════════════════════════════
62 # Filename attacks
63 # ═══════════════════════════════════════════════════════════════════════════════
64
65 class TestFilenameAttacks:
66 """POST /api/mists with malicious filenames must be rejected (422)."""
67
68 @pytest.mark.anyio
69 async def test_path_traversal_dotdot(
70 self, client: AsyncClient, auth_headers: dict
71 ) -> None:
72 r = await client.post(
73 "/api/mists",
74 json=_payload(filename="../evil.py"),
75 headers=auth_headers,
76 )
77 assert r.status_code == 422
78
79 @pytest.mark.anyio
80 async def test_path_traversal_deep(
81 self, client: AsyncClient, auth_headers: dict
82 ) -> None:
83 r = await client.post(
84 "/api/mists",
85 json=_payload(filename="../../etc/passwd"),
86 headers=auth_headers,
87 )
88 assert r.status_code == 422
89
90 @pytest.mark.anyio
91 async def test_null_byte(
92 self, client: AsyncClient, auth_headers: dict
93 ) -> None:
94 r = await client.post(
95 "/api/mists",
96 json=_payload(filename="evil\x00.py"),
97 headers=auth_headers,
98 )
99 assert r.status_code == 422
100
101 @pytest.mark.anyio
102 async def test_forward_slash_separator(
103 self, client: AsyncClient, auth_headers: dict
104 ) -> None:
105 r = await client.post(
106 "/api/mists",
107 json=_payload(filename="subdir/evil.py"),
108 headers=auth_headers,
109 )
110 assert r.status_code == 422
111
112 @pytest.mark.anyio
113 async def test_backslash_separator(
114 self, client: AsyncClient, auth_headers: dict
115 ) -> None:
116 r = await client.post(
117 "/api/mists",
118 json=_payload(filename="subdir\\evil.py"),
119 headers=auth_headers,
120 )
121 assert r.status_code == 422
122
123 @pytest.mark.anyio
124 async def test_control_character_tab(
125 self, client: AsyncClient, auth_headers: dict
126 ) -> None:
127 r = await client.post(
128 "/api/mists",
129 json=_payload(filename="evil\t.py"),
130 headers=auth_headers,
131 )
132 assert r.status_code == 422
133
134 @pytest.mark.anyio
135 async def test_control_character_newline(
136 self, client: AsyncClient, auth_headers: dict
137 ) -> None:
138 r = await client.post(
139 "/api/mists",
140 json=_payload(filename="evil\n.py"),
141 headers=auth_headers,
142 )
143 assert r.status_code == 422
144
145 @pytest.mark.anyio
146 async def test_ansi_escape_sequence(
147 self, client: AsyncClient, auth_headers: dict
148 ) -> None:
149 r = await client.post(
150 "/api/mists",
151 json=_payload(filename="\x1b[31mevil\x1b[0m.py"),
152 headers=auth_headers,
153 )
154 assert r.status_code == 422
155
156 @pytest.mark.anyio
157 async def test_overlong_filename(
158 self, client: AsyncClient, auth_headers: dict
159 ) -> None:
160 r = await client.post(
161 "/api/mists",
162 json=_payload(filename="a" * 256 + ".py"),
163 headers=auth_headers,
164 )
165 assert r.status_code == 422
166
167 @pytest.mark.anyio
168 async def test_empty_filename_rejected(
169 self, client: AsyncClient, auth_headers: dict
170 ) -> None:
171 r = await client.post(
172 "/api/mists",
173 json=_payload(filename=""),
174 headers=auth_headers,
175 )
176 assert r.status_code == 422
177
178 @pytest.mark.anyio
179 async def test_valid_filename_accepted(
180 self, client: AsyncClient, auth_headers: dict
181 ) -> None:
182 """Confirm the gate accepts ordinary safe filenames."""
183 r = await client.post(
184 "/api/mists",
185 json=_payload(filename="valid_name.py"),
186 headers=auth_headers,
187 )
188 assert r.status_code == 201
189
190
191 # ═══════════════════════════════════════════════════════════════════════════════
192 # Content injection
193 # ═══════════════════════════════════════════════════════════════════════════════
194
195 class TestContentInjection:
196 """HTML/JS payloads must be stored verbatim; no server-side stripping."""
197
198 @pytest.mark.anyio
199 async def test_xss_script_tag_stored_verbatim(
200 self,
201 client: AsyncClient,
202 auth_headers: dict,
203 db_session: AsyncSession,
204 ) -> None:
205 xss = '<script>alert("xss")</script>'
206 body = _payload(content=xss + uuid.uuid4().hex)
207 r = await client.post("/api/mists", json=body, headers=auth_headers)
208 assert r.status_code == 201
209 mist_id = r.json()["mistId"]
210
211 r2 = await client.get(f"/api/mists/{mist_id}")
212 assert r2.status_code == 200
213 assert xss in r2.json()["content"], "XSS payload must be stored verbatim"
214
215 @pytest.mark.anyio
216 async def test_html_entity_stored_verbatim(
217 self,
218 client: AsyncClient,
219 auth_headers: dict,
220 db_session: AsyncSession,
221 ) -> None:
222 payload = '<img src=x onerror=alert(1)> &lt;not-encoded&gt;'
223 body = _payload(content=payload + uuid.uuid4().hex)
224 r = await client.post("/api/mists", json=body, headers=auth_headers)
225 assert r.status_code == 201
226 mist_id = r.json()["mistId"]
227
228 r2 = await client.get(f"/api/mists/{mist_id}")
229 assert r2.status_code == 200
230 # Content stored as-is — sanitisation is the template's responsibility.
231 assert payload in r2.json()["content"]
232
233 @pytest.mark.anyio
234 async def test_unicode_content_roundtrips(
235 self,
236 client: AsyncClient,
237 auth_headers: dict,
238 db_session: AsyncSession,
239 ) -> None:
240 unicode_content = "# 日本語テスト\nprint('こんにちは世界')\n" + uuid.uuid4().hex
241 body = _payload(content=unicode_content)
242 r = await client.post("/api/mists", json=body, headers=auth_headers)
243 assert r.status_code == 201
244 mist_id = r.json()["mistId"]
245
246 r2 = await client.get(f"/api/mists/{mist_id}")
247 assert r2.status_code == 200
248 assert unicode_content in r2.json()["content"]
249
250
251 # ═══════════════════════════════════════════════════════════════════════════════
252 # Access control
253 # ═══════════════════════════════════════════════════════════════════════════════
254
255 class TestAccessControl:
256 """Secret mists invisible to non-owners; non-owner mutations blocked."""
257
258 @pytest.mark.anyio
259 async def test_secret_mist_not_in_other_owners_list(
260 self,
261 client: AsyncClient,
262 auth_headers: dict,
263 db_session: AsyncSession,
264 ) -> None:
265 # Create a secret mist owned by "otheruser" directly via service layer.
266 # auth_headers authenticates as "testuser" — a legitimate non-owner.
267 from muse.plugins.mist.plugin import compute_mist_id
268 from musehub.db.musehub_models import MusehubRepo
269 from musehub.services.musehub_mists import create_mist as _svc_create
270
271 content = f"secret_list {uuid.uuid4().hex}"
272 mid = compute_mist_id(content.encode())
273 repo = MusehubRepo(
274 name=mid, owner="otheruser", slug=mid,
275 visibility="secret", owner_user_id="otheruser",
276 )
277 db_session.add(repo)
278 await db_session.flush()
279 await _svc_create(
280 db_session, mist_id=mid, filename="secret.py", content=content,
281 owner="otheruser", repo_id=str(repo.repo_id), visibility="secret",
282 )
283 await db_session.commit()
284
285 # testuser (non-owner) fetching otheruser's list must not see the secret mist.
286 r = await client.get("/api/otheruser/mists", headers=auth_headers)
287 assert r.status_code == 200
288 ids = [m["mistId"] for m in r.json()["mists"]]
289 assert mid not in ids, "Secret mist must not appear in non-owner's list view"
290
291 @pytest.mark.anyio
292 async def test_secret_mist_not_in_explore(
293 self,
294 client: AsyncClient,
295 auth_headers: dict,
296 db_session: AsyncSession,
297 ) -> None:
298 mist = await _create(client, auth_headers, visibility="secret")
299 mist_id = mist["mistId"]
300
301 r = await client.get("/api/mists/explore")
302 assert r.status_code == 200
303 ids = [m["mistId"] for m in r.json()["mists"]]
304 assert mist_id not in ids, "Secret mist must not appear in explore feed"
305
306 @pytest.mark.anyio
307 async def test_secret_mist_detail_returns_403_for_non_owner(
308 self,
309 client: AsyncClient,
310 auth_headers: dict,
311 db_session: AsyncSession,
312 ) -> None:
313 # Create a secret mist owned by "otheruser" directly via service layer.
314 # testuser (from auth_headers override) is the authenticated non-owner caller.
315 from muse.plugins.mist.plugin import compute_mist_id
316 from musehub.db.musehub_models import MusehubRepo
317 from musehub.services.musehub_mists import create_mist as _svc_create
318
319 content = f"secret_detail {uuid.uuid4().hex}"
320 mid = compute_mist_id(content.encode())
321 repo = MusehubRepo(
322 name=mid, owner="otheruser", slug=mid,
323 visibility="secret", owner_user_id="otheruser",
324 )
325 db_session.add(repo)
326 await db_session.flush()
327 await _svc_create(
328 db_session, mist_id=mid, filename="secret.py", content=content,
329 owner="otheruser", repo_id=str(repo.repo_id), visibility="secret",
330 )
331 await db_session.commit()
332
333 # testuser is authenticated but is not the owner of this secret mist.
334 r = await client.get(f"/api/mists/{mid}", headers=auth_headers)
335 assert r.status_code in (403, 404), (
336 "Secret mist must not be accessible to non-owner (even authenticated)"
337 )
338
339 @pytest.mark.anyio
340 async def test_non_owner_update_returns_404(
341 self,
342 client: AsyncClient,
343 auth_headers: dict,
344 db_session: AsyncSession,
345 ) -> None:
346 # Create a mist directly via service layer owned by "otheruser".
347 # auth_headers authenticates as "testuser" — a legitimate non-owner caller.
348 from muse.plugins.mist.plugin import compute_mist_id
349 from musehub.db.musehub_models import MusehubRepo
350 from musehub.services.musehub_mists import create_mist as _svc_create
351
352 content = f"non_owner_upd {uuid.uuid4().hex}"
353 mid = compute_mist_id(content.encode())
354 repo = MusehubRepo(
355 name=mid, owner="otheruser", slug=mid,
356 visibility="public", owner_user_id="otheruser",
357 )
358 db_session.add(repo)
359 await db_session.flush()
360 await _svc_create(
361 db_session, mist_id=mid, filename="f.py", content=content,
362 owner="otheruser", repo_id=str(repo.repo_id),
363 )
364 await db_session.commit()
365
366 r = await client.patch(
367 f"/api/mists/{mid}",
368 json={"title": "Hijacked"},
369 headers=auth_headers, # testuser ≠ otheruser
370 )
371 assert r.status_code in (403, 404)
372
373 @pytest.mark.anyio
374 async def test_non_owner_delete_returns_404(
375 self,
376 client: AsyncClient,
377 auth_headers: dict,
378 db_session: AsyncSession,
379 ) -> None:
380 from muse.plugins.mist.plugin import compute_mist_id
381 from musehub.db.musehub_models import MusehubRepo
382 from musehub.services.musehub_mists import create_mist as _svc_create
383
384 content = f"non_owner_del {uuid.uuid4().hex}"
385 mid = compute_mist_id(content.encode())
386 repo = MusehubRepo(
387 name=mid, owner="otheruser", slug=mid,
388 visibility="public", owner_user_id="otheruser",
389 )
390 db_session.add(repo)
391 await db_session.flush()
392 await _svc_create(
393 db_session, mist_id=mid, filename="f.py", content=content,
394 owner="otheruser", repo_id=str(repo.repo_id),
395 )
396 await db_session.commit()
397
398 r = await client.delete(
399 f"/api/mists/{mid}",
400 headers=auth_headers, # testuser ≠ otheruser
401 )
402 assert r.status_code in (403, 404)
403
404 @pytest.mark.anyio
405 async def test_unauthenticated_create_returns_401(
406 self, client: AsyncClient
407 ) -> None:
408 r = await client.post("/api/mists", json=_payload())
409 assert r.status_code == 401
410
411 @pytest.mark.anyio
412 async def test_unauthenticated_update_returns_401(
413 self, client: AsyncClient, db_session: AsyncSession
414 ) -> None:
415 # Create via service layer so auth_headers fixture is NOT active.
416 from muse.plugins.mist.plugin import compute_mist_id
417 from musehub.db.musehub_models import MusehubRepo
418 from musehub.services.musehub_mists import create_mist as _svc_create
419
420 content = f"unauth_upd {uuid.uuid4().hex}"
421 mid = compute_mist_id(content.encode())
422 repo = MusehubRepo(
423 name=mid, owner="testuser", slug=mid,
424 visibility="public", owner_user_id="testuser",
425 )
426 db_session.add(repo)
427 await db_session.flush()
428 await _svc_create(
429 db_session, mist_id=mid, filename="f.py", content=content,
430 owner="testuser", repo_id=str(repo.repo_id),
431 )
432 await db_session.commit()
433
434 r = await client.patch(f"/api/mists/{mid}", json={"title": "x"})
435 assert r.status_code == 401
436
437 @pytest.mark.anyio
438 async def test_unauthenticated_delete_returns_401(
439 self, client: AsyncClient, db_session: AsyncSession
440 ) -> None:
441 from muse.plugins.mist.plugin import compute_mist_id
442 from musehub.db.musehub_models import MusehubRepo
443 from musehub.services.musehub_mists import create_mist as _svc_create
444
445 content = f"unauth_del {uuid.uuid4().hex}"
446 mid = compute_mist_id(content.encode())
447 repo = MusehubRepo(
448 name=mid, owner="testuser", slug=mid,
449 visibility="public", owner_user_id="testuser",
450 )
451 db_session.add(repo)
452 await db_session.flush()
453 await _svc_create(
454 db_session, mist_id=mid, filename="f.py", content=content,
455 owner="testuser", repo_id=str(repo.repo_id),
456 )
457 await db_session.commit()
458
459 r = await client.delete(f"/api/mists/{mid}")
460 assert r.status_code == 401
461
462 @pytest.mark.anyio
463 async def test_owner_can_see_own_secret_mist(
464 self,
465 client: AsyncClient,
466 auth_headers: dict,
467 db_session: AsyncSession,
468 ) -> None:
469 mist = await _create(client, auth_headers, visibility="secret")
470 mist_id = mist["mistId"]
471
472 r = await client.get(f"/api/mists/{mist_id}", headers=auth_headers)
473 assert r.status_code == 200
474 assert r.json()["mistId"] == mist_id
475
476
477 # ═══════════════════════════════════════════════════════════════════════════════
478 # Collision resistance
479 # ═══════════════════════════════════════════════════════════════════════════════
480
481 class TestCollisionResistance:
482 """Identical content bytes → identical mist_id → POST returns 409."""
483
484 @pytest.mark.anyio
485 async def test_duplicate_content_returns_409(
486 self,
487 client: AsyncClient,
488 auth_headers: dict,
489 db_session: AsyncSession,
490 ) -> None:
491 fixed_content = "def idempotent(): return 42\n"
492 body = _payload(content=fixed_content, filename="idempotent.py")
493
494 r1 = await client.post("/api/mists", json=body, headers=auth_headers)
495 assert r1.status_code == 201
496 mist_id = r1.json()["mistId"]
497
498 r2 = await client.post("/api/mists", json=body, headers=auth_headers)
499 assert r2.status_code == 409, (
500 "Re-posting identical content must return 409 (content-addressed)"
501 )
502
503 @pytest.mark.anyio
504 async def test_different_content_different_id(
505 self,
506 client: AsyncClient,
507 auth_headers: dict,
508 db_session: AsyncSession,
509 ) -> None:
510 r1 = await _create(client, auth_headers, content="content_a " + uuid.uuid4().hex)
511 r2 = await _create(client, auth_headers, content="content_b " + uuid.uuid4().hex)
512 assert r1["mistId"] != r2["mistId"]
513
514 @pytest.mark.anyio
515 async def test_mist_id_deterministic_from_content(
516 self,
517 client: AsyncClient,
518 auth_headers: dict,
519 db_session: AsyncSession,
520 ) -> None:
521 """mist_id is deterministic — recomputing it offline verifies integrity."""
522 from muse.plugins.mist.plugin import compute_mist_id
523
524 content = "def check(): pass\n# unique: " + uuid.uuid4().hex
525 r = await _create(client, auth_headers, content=content)
526
527 expected_id = compute_mist_id(content.encode("utf-8"))
528 assert r["mistId"] == expected_id
529
530
531 # ═══════════════════════════════════════════════════════════════════════════════
532 # Large content rejection
533 # ═══════════════════════════════════════════════════════════════════════════════
534
535 class TestLargeContentRejection:
536 """Requests whose body exceeds 10 MiB must be rejected (413)."""
537
538 @pytest.mark.anyio
539 async def test_oversized_content_returns_413(
540 self,
541 client: AsyncClient,
542 auth_headers: dict,
543 db_session: AsyncSession,
544 ) -> None:
545 # 11 MiB of ASCII content — well above the 10 MiB middleware cap.
546 oversized = "x" * (11 * 1024 * 1024)
547 body: JSONObject = {
548 "filename": "large.py",
549 "content": oversized,
550 "visibility": "public",
551 }
552 r = await client.post("/api/mists", json=body, headers=auth_headers)
553 assert r.status_code == 413, (
554 "Content > 10 MiB must be rejected by ContentSizeLimitMiddleware"
555 )
556
557 @pytest.mark.anyio
558 async def test_near_limit_content_accepted(
559 self,
560 client: AsyncClient,
561 auth_headers: dict,
562 db_session: AsyncSession,
563 ) -> None:
564 """Content just under 1 MiB should succeed (sanity check)."""
565 content = "a" * (512 * 1024) # 512 KiB — well within limit
566 r = await client.post(
567 "/api/mists",
568 json=_payload(content=content),
569 headers=auth_headers,
570 )
571 assert r.status_code == 201
572
573
574 # ═══════════════════════════════════════════════════════════════════════════════
575 # Fork depth enforcement
576 # ═══════════════════════════════════════════════════════════════════════════════
577
578 class TestForkDepthEnforcement:
579 """Fork chain is capped at depth 5; further forks must return 422."""
580
581 @pytest.mark.anyio
582 async def test_fork_chain_to_max_depth(
583 self,
584 client: AsyncClient,
585 auth_headers: dict,
586 db_session: AsyncSession,
587 ) -> None:
588 # Create the root mist.
589 root = await _create(client, auth_headers)
590 current_id = root["mistId"]
591
592 # Fork 5 times — all must succeed.
593 for depth in range(1, 6):
594 r = await client.post(
595 f"/api/mists/{current_id}/fork", headers=auth_headers
596 )
597 assert r.status_code == 201, (
598 f"Fork at depth {depth} must succeed; got {r.status_code}: {r.text}"
599 )
600 current_id = r.json()["mistId"]
601
602 @pytest.mark.anyio
603 async def test_fork_past_max_depth_returns_422(
604 self,
605 client: AsyncClient,
606 auth_headers: dict,
607 db_session: AsyncSession,
608 ) -> None:
609 # Build a chain of depth 5.
610 root = await _create(client, auth_headers)
611 current_id = root["mistId"]
612 for _ in range(5):
613 r = await client.post(
614 f"/api/mists/{current_id}/fork", headers=auth_headers
615 )
616 assert r.status_code == 201
617 current_id = r.json()["mistId"]
618
619 # Forking the depth-5 mist must fail.
620 r = await client.post(
621 f"/api/mists/{current_id}/fork", headers=auth_headers
622 )
623 assert r.status_code == 422, (
624 f"Fork past depth 5 must be rejected; got {r.status_code}: {r.text}"
625 )
626
627 @pytest.mark.anyio
628 async def test_fork_nonexistent_mist_returns_404(
629 self,
630 client: AsyncClient,
631 auth_headers: dict,
632 db_session: AsyncSession,
633 ) -> None:
634 r = await client.post(
635 "/api/mists/doesnotexist/fork", headers=auth_headers
636 )
637 assert r.status_code == 404
638
639
640 # ═══════════════════════════════════════════════════════════════════════════════
641 # Tag injection
642 # ═══════════════════════════════════════════════════════════════════════════════
643
644 class TestTagSecurity:
645 """Tags have count and length limits; HTML-special and null-byte tags rejected."""
646
647 @pytest.mark.anyio
648 async def test_too_many_tags_rejected(
649 self, client: AsyncClient, auth_headers: dict
650 ) -> None:
651 r = await client.post(
652 "/api/mists",
653 json=_payload(tags=[f"tag{i}" for i in range(11)]), # max is 10
654 headers=auth_headers,
655 )
656 assert r.status_code == 422
657
658 @pytest.mark.anyio
659 async def test_overlong_tag_rejected(
660 self, client: AsyncClient, auth_headers: dict
661 ) -> None:
662 r = await client.post(
663 "/api/mists",
664 json=_payload(tags=["a" * 65]), # max is 64
665 headers=auth_headers,
666 )
667 assert r.status_code == 422
668
669 @pytest.mark.anyio
670 async def test_null_byte_in_tag_rejected(
671 self, client: AsyncClient, auth_headers: dict
672 ) -> None:
673 r = await client.post(
674 "/api/mists",
675 json=_payload(tags=["evil\x00tag"]),
676 headers=auth_headers,
677 )
678 assert r.status_code == 422
File History 1 commit
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ 156 days ago