gabriel / musehub public
test_security_hardening.py python
815 lines 41.3 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago
1 """Section 8 — Security Hardening (Adversarial) tests.
2
3 Covers:
4 SSRF : outbound URLs validated; RFC-1918 / loopback / link-local
5 blocked; non-HTTPS schemes rejected; DNS resolution guarded.
6 Mass assignment : all request bodies use Pydantic models; no **kwargs from
7 raw request dicts passed to ORM constructors.
8 Timing attacks : hmac.compare_digest used for all secret comparisons.
9 Object enumeration : repo / issue primary keys are content-addressed IDs, not sequential ints.
10 Commit forgery : REQUIRE_SIGNED_COMMITS enforcement gate in wire_push.
11 Regex DoS : no user-supplied pattern compiled with re.compile().
12 Tar bomb : no archive extraction; N/A for current feature set.
13 """
14 from __future__ import annotations
15
16 import ast
17 import hashlib
18 import re
19 from pathlib import Path
20 from unittest.mock import AsyncMock, MagicMock, patch
21
22 import pytest
23
24 _ROOT = Path(__file__).resolve().parents[1]
25 _MUSEHUB_PKG = _ROOT / "musehub"
26 _SSRF_MODULE = _MUSEHUB_PKG / "security" / "ssrf.py"
27 _WEBHOOK_MODEL = _MUSEHUB_PKG / "models" / "musehub.py"
28 _DISPATCHER = _MUSEHUB_PKG / "services" / "musehub_webhook_dispatcher.py"
29 _WIRE_SVC = _MUSEHUB_PKG / "services" / "musehub_wire.py"
30 _CONFIG = _MUSEHUB_PKG / "config.py"
31
32
33 # ═══════════════════════════════════════════════════════════════════════════════
34 # SSRF — check_url_safe (sync, no DNS)
35 # ═══════════════════════════════════════════════════════════════════════════════
36
37 class TestSsrfCheckUrlSafe:
38 def _check(self, url: str) -> str:
39 from musehub.security.ssrf import check_url_safe
40 return check_url_safe(url)
41
42 def test_https_public_url_allowed(self) -> None:
43 result = self._check("https://hooks.example.com/deliver")
44 assert result == "https://hooks.example.com/deliver"
45
46 def test_http_scheme_blocked(self) -> None:
47 with pytest.raises(ValueError, match="https://"):
48 self._check("http://hooks.example.com/deliver")
49
50 def test_ftp_scheme_blocked(self) -> None:
51 with pytest.raises(ValueError, match="https://"):
52 self._check("ftp://hooks.example.com/deliver")
53
54 def test_file_scheme_blocked(self) -> None:
55 with pytest.raises(ValueError, match="https://"):
56 self._check("file:///etc/passwd")
57
58 def test_loopback_127_blocked(self) -> None:
59 with pytest.raises(ValueError, match="private/reserved"):
60 self._check("https://127.0.0.1/internal")
61
62 def test_loopback_127_other_blocked(self) -> None:
63 with pytest.raises(ValueError, match="private/reserved"):
64 self._check("https://127.255.255.255/internal")
65
66 def test_rfc1918_10_blocked(self) -> None:
67 with pytest.raises(ValueError, match="private/reserved"):
68 self._check("https://10.0.0.1/")
69
70 def test_rfc1918_172_16_blocked(self) -> None:
71 with pytest.raises(ValueError, match="private/reserved"):
72 self._check("https://172.16.0.1/")
73
74 def test_rfc1918_192_168_blocked(self) -> None:
75 with pytest.raises(ValueError, match="private/reserved"):
76 self._check("https://192.168.1.1/")
77
78 def test_link_local_169_254_blocked(self) -> None:
79 """169.254.169.254 is the AWS instance metadata endpoint."""
80 with pytest.raises(ValueError, match="private/reserved"):
81 self._check("https://169.254.169.254/latest/meta-data/")
82
83 def test_hostname_with_path_allowed(self) -> None:
84 result = self._check("https://api.stripe.com/v1/webhooks")
85 assert "stripe.com" in result
86
87 def test_no_hostname_blocked(self) -> None:
88 with pytest.raises(ValueError):
89 self._check("https:///path")
90
91 def test_empty_string_blocked(self) -> None:
92 with pytest.raises(ValueError):
93 self._check("")
94
95
96 # ═══════════════════════════════════════════════════════════════════════════════
97 # SSRF — validate_outbound_url (async, with DNS)
98 # ═══════════════════════════════════════════════════════════════════════════════
99
100 class TestSsrfValidateOutboundUrl:
101 async def test_public_hostname_allowed(self) -> None:
102 from musehub.security.ssrf import validate_outbound_url
103 import socket
104 # Patch DNS to return a public IP.
105 with patch("musehub.security.ssrf.socket.getaddrinfo",
106 return_value=[(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("93.184.216.34", 443))]):
107 result = await validate_outbound_url("https://example.com/webhook")
108 assert result == "https://example.com/webhook"
109
110 async def test_dns_resolves_to_private_blocked(self) -> None:
111 from musehub.security.ssrf import validate_outbound_url
112 import socket
113 # DNS rebinding: hostname looks public but resolves to internal IP.
114 with patch("musehub.security.ssrf.socket.getaddrinfo",
115 return_value=[(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("192.168.1.100", 443))]):
116 with pytest.raises(ValueError, match="private/reserved"):
117 await validate_outbound_url("https://sneaky.example.com/webhook")
118
119 async def test_dns_resolves_to_loopback_blocked(self) -> None:
120 from musehub.security.ssrf import validate_outbound_url
121 import socket
122 with patch("musehub.security.ssrf.socket.getaddrinfo",
123 return_value=[(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("127.0.0.1", 443))]):
124 with pytest.raises(ValueError, match="private/reserved"):
125 await validate_outbound_url("https://localhost.example.com/")
126
127 async def test_unresolvable_hostname_blocked(self) -> None:
128 from musehub.security.ssrf import validate_outbound_url
129 import socket
130 with patch("musehub.security.ssrf.socket.getaddrinfo",
131 side_effect=socket.gaierror("Name or service not known")):
132 with pytest.raises(ValueError, match="cannot be resolved"):
133 await validate_outbound_url("https://nonexistent.invalid/")
134
135 async def test_bare_ip_does_not_do_dns(self) -> None:
136 """Bare IP literals must not trigger DNS resolution."""
137 from musehub.security.ssrf import validate_outbound_url
138 # If DNS is called for a bare IP, the test would hang or fail.
139 with patch("musehub.security.ssrf.socket.getaddrinfo") as mock_dns:
140 with pytest.raises(ValueError, match="private/reserved"):
141 await validate_outbound_url("https://10.0.0.1/")
142 mock_dns.assert_not_called()
143
144
145 # ═══════════════════════════════════════════════════════════════════════════════
146 # SSRF — WebhookCreate validator
147 # ═══════════════════════════════════════════════════════════════════════════════
148
149 class TestWebhookCreateSsrf:
150 def _create(self, url: str) -> None:
151 from musehub.models.musehub import WebhookCreate
152 return WebhookCreate(url=url, events=["push"])
153
154 def test_https_public_accepted(self) -> None:
155 wh = self._create("https://hooks.example.com/musehub")
156 assert wh.url == "https://hooks.example.com/musehub"
157
158 def test_http_rejected_at_model_parse(self) -> None:
159 from pydantic import ValidationError
160 with pytest.raises(ValidationError, match="https://"):
161 self._create("http://hooks.example.com/musehub")
162
163 def test_rfc1918_rejected_at_model_parse(self) -> None:
164 from pydantic import ValidationError
165 with pytest.raises(ValidationError, match="private/reserved"):
166 self._create("https://192.168.0.1/hook")
167
168 def test_loopback_rejected_at_model_parse(self) -> None:
169 from pydantic import ValidationError
170 with pytest.raises(ValidationError, match="private/reserved"):
171 self._create("https://127.0.0.1/hook")
172
173 def test_aws_metadata_endpoint_rejected(self) -> None:
174 from pydantic import ValidationError
175 with pytest.raises(ValidationError, match="private/reserved"):
176 self._create("https://169.254.169.254/latest/meta-data/iam/")
177
178
179 # ═══════════════════════════════════════════════════════════════════════════════
180 # SSRF — defense-in-depth in _attempt_delivery
181 # ═══════════════════════════════════════════════════════════════════════════════
182
183 class TestWebhookDeliveryDefenseInDepth:
184 async def test_delivery_blocked_when_ssrf_check_fails(self) -> None:
185 """_attempt_delivery returns failure when the URL fails SSRF validation."""
186 from musehub.services.musehub_webhook_dispatcher import _attempt_delivery
187 from musehub.db.musehub_models import MusehubWebhook
188
189 webhook = MagicMock(spec=MusehubWebhook)
190 webhook.url = "https://192.168.1.1/hook" # private IP
191 webhook.secret = ""
192
193 client = AsyncMock()
194 success, status, msg = await _attempt_delivery(
195 client,
196 webhook=webhook,
197 event_type="push",
198 payload_bytes=b"{}",
199 delivery_id="test-id",
200 attempt=1,
201 )
202
203 assert success is False
204 assert "SSRF" in msg or "blocked" in msg.lower()
205 client.post.assert_not_called()
206
207 def test_dispatcher_imports_ssrf_module(self) -> None:
208 src = _DISPATCHER.read_text()
209 assert "validate_outbound_url" in src or "ssrf" in src.lower(), (
210 "musehub_webhook_dispatcher.py must use validate_outbound_url for defence-in-depth"
211 )
212
213
214 # ═══════════════════════════════════════════════════════════════════════════════
215 # Mass assignment
216 # ═══════════════════════════════════════════════════════════════════════════════
217
218 class TestMassAssignment:
219 def _route_files(self) -> list[Path]:
220 return list((_MUSEHUB_PKG / "api" / "routes").rglob("*.py"))
221
222 def test_no_kwargs_splat_from_request_body(self) -> None:
223 """No route handler should unpack request.body or raw dict into ORM."""
224 dangerous_patterns = [
225 re.compile(r"\*\*request\.body"),
226 re.compile(r"\*\*await request\.json"),
227 re.compile(r"dict\(request\)"),
228 ]
229 violations: list[str] = []
230 for path in self._route_files():
231 src = path.read_text()
232 for pat in dangerous_patterns:
233 if pat.search(src):
234 violations.append(f"{path.name}: {pat.pattern}")
235 assert not violations, f"Potential mass assignment found: {violations}"
236
237 def test_route_handler_body_params_are_pydantic_or_primitive(self) -> None:
238 """Spot-check that handler params with complex types are Pydantic models."""
239 # The most dangerous pattern is `body: dict = Body(...)` — raw dict
240 # passed to ORM. We check that no handler uses `dict` as a body type.
241 body_dict_re = re.compile(r":\s*dict\s*=\s*(?:Body|Depends)\(")
242 violations: list[str] = []
243 for path in self._route_files():
244 src = path.read_text()
245 if body_dict_re.search(src):
246 violations.append(path.name)
247 assert not violations, (
248 f"Raw dict Body() parameter found (mass assignment risk): {violations}"
249 )
250
251 def test_webhook_create_uses_pydantic_not_raw_dict(self) -> None:
252 src = (_MUSEHUB_PKG / "api" / "routes" / "musehub" / "webhooks.py").read_text()
253 assert "WebhookCreate" in src, (
254 "Webhook creation route does not use WebhookCreate Pydantic model"
255 )
256
257
258 # ═══════════════════════════════════════════════════════════════════════════════
259 # Timing attacks
260 # ═══════════════════════════════════════════════════════════════════════════════
261
262 class TestTimingAttacks:
263 def test_request_signing_uses_cryptographic_verify(self) -> None:
264 """MSign uses Ed25519 verify_signature — cryptographically constant-time."""
265 src = (_MUSEHUB_PKG / "auth" / "request_signing.py").read_text()
266 assert "verify_signature" in src, (
267 "request_signing.py must call verify_signature for MSign auth"
268 )
269 # Must NOT use == to compare signature bytes (timing leak)
270 assert "sig_bytes ==" not in src and "== sig_bytes" not in src, (
271 "request_signing.py uses == for signature bytes — timing attack"
272 )
273
274 def test_crypto_keys_uses_compare_digest(self) -> None:
275 src = (_MUSEHUB_PKG / "crypto" / "keys.py").read_text()
276 assert "compare_digest" in src, (
277 "crypto/keys.py must use hmac.compare_digest"
278 )
279
280 def test_compare_digest_used_not_equal_for_secrets(self) -> None:
281 """Grep all auth/crypto files for == comparisons against secret strings."""
282 secret_eq_re = re.compile(
283 r"(secret|token|password|sig|signature|key)\s*==\s*[a-zA-Z_]",
284 re.IGNORECASE,
285 )
286 checked_files = [
287 _MUSEHUB_PKG / "auth" / "request_signing.py",
288 _MUSEHUB_PKG / "crypto" / "keys.py",
289 _MUSEHUB_PKG / "services" / "musehub_auth.py",
290 ]
291 violations: list[str] = []
292 for path in checked_files:
293 for lineno, line in enumerate(path.read_text().splitlines(), 1):
294 stripped = line.strip()
295 if stripped.startswith("#"):
296 continue
297 if secret_eq_re.search(stripped):
298 violations.append(f"{path.name}:{lineno}: {stripped[:80]}")
299 assert not violations, (
300 f"Potential timing-unsafe comparison found:\n{'\n'.join(violations)}"
301 )
302
303
304 # ═══════════════════════════════════════════════════════════════════════════════
305 # Object enumeration
306 # ═══════════════════════════════════════════════════════════════════════════════
307
308 class TestObjectEnumeration:
309 def _db_models_src(self) -> str:
310 return (_MUSEHUB_PKG / "db" / "musehub_models.py").read_text()
311
312 def test_repo_primary_key_is_string(self) -> None:
313 src = self._db_models_src()
314 # repo_id must be String default=content-addressed ID — not Integer
315 m = re.search(r"repo_id.*primary_key=True", src)
316 assert m, "repo_id primary key not found in musehub_models.py"
317 # Check the line includes String, not Integer
318 line = [l for l in src.splitlines() if "repo_id" in l and "primary_key=True" in l]
319 assert line, "repo_id primary key line not found"
320 assert "String" in line[0], (
321 f"repo_id primary key is not a String: {line[0]}"
322 )
323 assert "Integer" not in line[0], (
324 "repo_id primary key is Integer — sequential IDs enable enumeration"
325 )
326
327 def test_issue_primary_key_is_string(self) -> None:
328 src = self._db_models_src()
329 lines = [l for l in src.splitlines() if "issue_id" in l and "primary_key=True" in l]
330 assert lines, "issue_id primary key line not found"
331 assert "String" in lines[0], (
332 f"issue_id primary key is not a String: {lines[0]}"
333 )
334
335 def test_issue_number_is_per_repo_not_global(self) -> None:
336 """Per-repo sequential numbers (like GitHub) are acceptable;
337 a global monotone counter would enable cross-repo enumeration."""
338 src = self._db_models_src()
339 # Verify issue number is defined in the MusehubIssue table context
340 # (per-repo scoped by unique_constraint on (repo_id, number)).
341 assert "number" in src and "repo_id" in src, (
342 "Could not find issue number + repo_id in musehub_models.py"
343 )
344
345 def test_no_autoincrement_primary_keys_on_main_entities(self) -> None:
346 """Key entity tables must not use INTEGER AUTOINCREMENT as PK."""
347 src = self._db_models_src()
348 # Collect lines that set primary_key=True
349 pk_lines = [
350 (i + 1, l.strip())
351 for i, l in enumerate(src.splitlines())
352 if "primary_key=True" in l
353 ]
354 for lineno, line in pk_lines:
355 # If it says Integer AND primary_key=True AND not in a composite key,
356 # it is a sequential auto-increment PK — flag it.
357 if "Integer" in line and "primary_key=True" in line:
358 # Allow composite keys (they never appear alone as the entity ID)
359 # by checking the field name; reject generic 'id' or '<entity>_id'.
360 if re.search(r"\b(id|_id)\b", line):
361 pytest.fail(
362 f"musehub_models.py:{lineno}: Integer primary key found — "
363 f"use String IDs to prevent enumeration:\n {line}"
364 )
365
366
367 # ═══════════════════════════════════════════════════════════════════════════════
368 # Commit ID forgery / server-side signature enforcement
369 # ═══════════════════════════════════════════════════════════════════════════════
370
371 class TestCommitIdForgery:
372 def test_require_signed_commits_setting_exists(self) -> None:
373 src = _CONFIG.read_text()
374 assert "require_signed_commits" in src, (
375 "Settings does not have a require_signed_commits field"
376 )
377
378 def test_wire_push_enforces_when_setting_true(self) -> None:
379 """wire_push must reject unsigned commits when require_signed_commits=True."""
380 src = _WIRE_SVC.read_text()
381 assert "require_signed_commits" in src, (
382 "musehub_wire.py does not enforce require_signed_commits"
383 )
384 assert "unsigned" in src.lower() or "signature" in src.lower(), (
385 "musehub_wire.py does not reference signature enforcement"
386 )
387
388 def test_wire_push_stream_enforces_signed_commits(self) -> None:
389 """wire_push_stream must reject unsigned commits when require_signed_commits=True."""
390 src = _WIRE_SVC.read_text()
391 wire_stream_src = src[src.find("async def wire_push_stream"):]
392 assert "require_signed_commits" in wire_stream_src, (
393 "wire_push_stream does not enforce require_signed_commits"
394 )
395 assert "unsigned" in wire_stream_src.lower() or "signature" in wire_stream_src.lower(), (
396 "wire_push_stream does not reference signature enforcement"
397 )
398
399 def test_wire_push_source_has_unsigned_warning_not_rejection(self) -> None:
400 """When enforcement is off, wire_push logs a debug warning but does not reject."""
401 src = _WIRE_SVC.read_text()
402 # With enforcement off: a debug/warning log, not a WirePushResponse(ok=False).
403 # The rejection path is inside 'if settings.require_signed_commits:'.
404 # Outside that block there must be no unconditional rejection for unsigned.
405 # Verify the soft-path uses logger.debug, not an early return.
406 enforcement_block = src[src.find("require_signed_commits"):]
407 else_block_start = enforcement_block.find("else:")
408 assert else_block_start != -1, "No else branch for require_signed_commits"
409 else_block = enforcement_block[else_block_start:else_block_start + 400]
410 assert "logger.debug" in else_block or "logger.warning" in else_block, (
411 "Unsigned commit soft-path must log at debug/warning, not silently pass"
412 )
413 # The else branch must not contain a WirePushResponse(ok=False)
414 assert 'ok=False' not in else_block, (
415 "Unsigned commits are rejected even when require_signed_commits=False"
416 )
417
418
419 # ═══════════════════════════════════════════════════════════════════════════════
420 # Regex DoS
421 # ═══════════════════════════════════════════════════════════════════════════════
422
423 class TestRegexDos:
424 def _search_service_src(self) -> str:
425 return (_MUSEHUB_PKG / "services" / "musehub_search.py").read_text()
426
427 def test_search_by_pattern_uses_python_in_not_regex(self) -> None:
428 """search_by_pattern must use Python `in` operator, not re.compile(user_input)."""
429 src = self._search_service_src()
430 # find the search_by_pattern function body
431 fn_start = src.find("def search_by_pattern")
432 fn_end = src.find("\nasync def ", fn_start + 1)
433 if fn_end == -1:
434 fn_end = len(src)
435 fn_body = src[fn_start:fn_end]
436 # Must not compile user pattern as regex
437 assert "re.compile" not in fn_body, (
438 "search_by_pattern uses re.compile on user input — ReDoS risk"
439 )
440 # Should use substring `in` operator
441 assert " in " in fn_body, (
442 "search_by_pattern does not use substring 'in' operator"
443 )
444
445 def test_search_by_ask_uses_fixed_tokenizer_not_user_regex(self) -> None:
446 """search_by_ask must tokenize with a fixed pattern, not user-supplied."""
447 src = self._search_service_src()
448 fn_start = src.find("def search_by_ask")
449 fn_end = src.find("\nasync def ", fn_start + 1)
450 if fn_end == -1:
451 fn_end = len(src)
452 fn_body = src[fn_start:fn_end]
453 # Any re.compile inside must be a literal fixed pattern (not a variable)
454 compile_calls = re.findall(r"re\.compile\((.+?)\)", fn_body)
455 for call in compile_calls:
456 assert not re.match(r"^[a-zA-Z_]\w*$", call.strip()), (
457 f"search_by_ask compiles a variable as regex: re.compile({call})"
458 )
459
460 def test_no_user_supplied_regex_in_search_routes(self) -> None:
461 """Search API routes must not pass user query string to re.compile()."""
462 src = (_MUSEHUB_PKG / "api" / "routes" / "musehub" / "search.py").read_text()
463 # re.compile should not appear in the route handler file
464 assert "re.compile" not in src, (
465 "search.py compiles user input as regex — ReDoS risk"
466 )
467
468 def test_global_search_uses_sql_or_python_in(self) -> None:
469 """Global search must use parameterized SQL or Python `in`, not Python regex."""
470 src = self._search_service_src()
471 # Verify _TOKEN_RE is a fixed, pre-compiled pattern
472 token_re_line = [l for l in src.splitlines() if "_TOKEN_RE" in l and "compile" in l]
473 assert token_re_line, "_TOKEN_RE pre-compiled pattern not found in musehub_search.py"
474 # It should be a literal string, not a variable
475 assert "[a-z" in token_re_line[0] or "[A-Z" in token_re_line[0], (
476 "_TOKEN_RE should be a literal character-class pattern"
477 )
478
479
480 # ═══════════════════════════════════════════════════════════════════════════════
481 # Tar bomb / zip bomb (N/A)
482 # ═══════════════════════════════════════════════════════════════════════════════
483
484 class TestTarBomb:
485 def test_no_archive_extraction_in_codebase(self) -> None:
486 """MuseHub does not extract archives — this check ensures it stays that way."""
487 dangerous = ["tarfile.open", "tarfile.extractall", "zipfile.ZipFile", ".extractall("]
488 violations: list[str] = []
489 for py_file in _MUSEHUB_PKG.rglob("*.py"):
490 if "test_" in py_file.name:
491 continue
492 src = py_file.read_text()
493 for pattern in dangerous:
494 if pattern in src:
495 violations.append(f"{py_file.relative_to(_ROOT)}: {pattern}")
496 assert not violations, (
497 f"Archive extraction found — add size/count limits before extractall():\n{'\n'.join(violations)}"
498 )
499
500
501
502
503 from httpx import AsyncClient
504
505 _ROOT = Path(__file__).resolve().parents[1]
506 _MUSEHUB_PKG = _ROOT / "musehub"
507 _MAIN_PY = _MUSEHUB_PKG / "main.py"
508 _AUTH_SVC = _MUSEHUB_PKG / "services" / "musehub_auth.py"
509 _MCP_DISPATCHER = _MUSEHUB_PKG / "mcp" / "dispatcher.py"
510 _PROMPTS = _MUSEHUB_PKG / "mcp" / "prompts.py"
511 _WIRE_SVC = _MUSEHUB_PKG / "services" / "musehub_wire.py"
512 _CONFIG = _MUSEHUB_PKG / "config.py"
513 _MAGIC_BYTES = _MUSEHUB_PKG / "security" / "magic_bytes.py"
514
515
516 # ═══════════════════════════════════════════════════════════════════════════════
517 # Polyglot files — magic bytes
518 # ═══════════════════════════════════════════════════════════════════════════════
519
520 class TestMagicBytes:
521 def _check(self, path: str, content: bytes) -> str:
522 from musehub.security.magic_bytes import check_magic_bytes
523 return check_magic_bytes(path, content)
524
525 def _expect_error(self, path: str, content: bytes) -> None:
526 from musehub.security.magic_bytes import check_magic_bytes, PolyglotFileError
527 with pytest.raises(PolyglotFileError):
528 check_magic_bytes(path, content)
529
530 # ── Valid files ──────────────────────────────────────────────────────────
531
532 def test_valid_midi(self) -> None:
533 midi_header = b"MThd\x00\x00\x00\x06\x00\x01\x00\x04\x01\xe0"
534 assert self._check("track.mid", midi_header) == "MIDI"
535
536 def test_valid_mp3_id3(self) -> None:
537 mp3_id3 = b"ID3\x03\x00\x00\x00\x00\x00\x00"
538 assert self._check("song.mp3", mp3_id3) == "MP3"
539
540 def test_valid_mp3_sync(self) -> None:
541 mp3_sync = bytes([0xFF, 0xFB]) + b"\x90\x00" * 10
542 assert self._check("song.mp3", mp3_sync) == "MP3"
543
544 def test_valid_webp(self) -> None:
545 webp = b"RIFF\x00\x00\x00\x00WEBP" + b"\x00" * 10
546 assert self._check("cover.webp", webp) == "WebP"
547
548 def test_valid_png(self) -> None:
549 png = bytes([0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A]) + b"\x00" * 10
550 assert self._check("cover.png", png) == "PNG"
551
552 def test_valid_jpeg(self) -> None:
553 jpeg = bytes([0xFF, 0xD8, 0xFF, 0xE0]) + b"\x00" * 10
554 assert self._check("cover.jpg", jpeg) == "JPEG"
555
556 def test_unknown_extension_passes_through(self) -> None:
557 # .py files are not in the known-type map — no check performed.
558 assert self._check("main.py", b"import os\n") == "unknown"
559
560 def test_script_shebang_in_py_file_allowed(self) -> None:
561 # Python scripts legitimately start with #! — this is NOT a polyglot attack.
562 shebang = b"#!/usr/bin/env python3\ndef main(): pass\n"
563 assert self._check("tools/audit.py", shebang) == "unknown"
564
565 def test_script_shebang_in_sh_file_allowed(self) -> None:
566 shebang = b"#!/bin/bash\necho hello\n"
567 assert self._check("scripts/deploy.sh", shebang) == "unknown"
568
569 def test_shebang_still_blocked_in_binary_extensions(self) -> None:
570 # A .mp3 with a shebang IS a polyglot attack.
571 self._expect_error("song.mp3", b"#!/bin/bash\nrm -rf /")
572
573 def test_shebang_still_blocked_in_midi_extension(self) -> None:
574 self._expect_error("track.mid", b"#!/usr/bin/php\n<?php system($_GET['cmd']); ?>")
575
576 def test_empty_content_returns_empty(self) -> None:
577 assert self._check("track.mid", b"") == "empty"
578
579 # ── Polyglot attacks ─────────────────────────────────────────────────────
580
581 def test_midi_extension_but_html_content_blocked(self) -> None:
582 self._expect_error("track.mid", b"<!DOCTYPE html><html><script>alert(1)</script>")
583
584 def test_midi_extension_but_php_shebang_blocked(self) -> None:
585 self._expect_error("track.mid", b"#!/usr/bin/php\n<?php system($_GET['cmd']); ?>")
586
587 def test_jpg_extension_but_html_content_blocked(self) -> None:
588 self._expect_error("cover.jpg", b"<html><body>not an image</body></html>")
589
590 def test_webp_extension_but_wrong_magic_blocked(self) -> None:
591 self._expect_error("cover.webp", b"\x00\x00\x00\x00\x00\x00\x00\x00")
592
593 def test_png_extension_but_wrong_magic_blocked(self) -> None:
594 self._expect_error("cover.png", b"JFIF\x00\x00")
595
596 def test_jpeg_extension_but_zip_content_blocked(self) -> None:
597 self._expect_error("cover.jpg", bytes([0x50, 0x4B, 0x03, 0x04]) + b"\x00" * 10)
598
599 def test_mp3_extension_but_html_shebang_blocked(self) -> None:
600 self._expect_error("song.mp3", b"#!/bin/bash\nrm -rf /")
601
602 def test_html_extension_allows_html_content(self) -> None:
603 # .html extension is exempt from the forbidden-HTML check.
604 result = self._check("readme.html", b"<!DOCTYPE html><html></html>")
605 assert result == "HTML" # returns "HTML" since the exemption triggers
606
607 # ── Wire push integration ─────────────────────────────────────────────────
608
609 def test_wire_push_imports_magic_bytes(self) -> None:
610 src = _WIRE_SVC.read_text()
611 assert "magic_bytes" in src or "PolyglotFileError" in src, (
612 "musehub_wire.py does not check magic bytes on push"
613 )
614
615 def test_wire_push_rejects_polyglot_path(self) -> None:
616 src = _WIRE_SVC.read_text()
617 assert "PolyglotFileError" in src, (
618 "wire_push does not catch PolyglotFileError"
619 )
620
621
622 # ═══════════════════════════════════════════════════════════════════════════════
623 # Clickjacking
624 # ═══════════════════════════════════════════════════════════════════════════════
625
626 class TestClickjacking:
627 async def test_x_frame_options_deny(self, client: AsyncClient) -> None:
628 resp = await client.get("/healthz")
629 assert resp.headers.get("x-frame-options", "").upper() == "DENY", (
630 "X-Frame-Options: DENY not set — clickjacking risk"
631 )
632
633 async def test_csp_frame_ancestors_none(self, client: AsyncClient) -> None:
634 resp = await client.get("/healthz")
635 csp = resp.headers.get("content-security-policy", "")
636 assert "frame-ancestors" in csp and "'none'" in csp, (
637 "CSP frame-ancestors 'none' not set — clickjacking risk"
638 )
639
640 def test_security_headers_middleware_sets_x_frame_options(self) -> None:
641 src = _MAIN_PY.read_text()
642 assert "X-Frame-Options" in src, (
643 "SecurityHeadersMiddleware does not set X-Frame-Options"
644 )
645 assert "DENY" in src
646
647 def test_security_headers_middleware_sets_frame_ancestors(self) -> None:
648 src = _MAIN_PY.read_text()
649 assert "frame-ancestors" in src and "'none'" in src, (
650 "SecurityHeadersMiddleware CSP does not include frame-ancestors 'none'"
651 )
652
653
654 # ═══════════════════════════════════════════════════════════════════════════════
655 # Handle squatting
656 # ═══════════════════════════════════════════════════════════════════════════════
657
658 class TestHandleSquatting:
659 def test_auth_service_normalizes_handle_to_lowercase(self) -> None:
660 src = _AUTH_SVC.read_text()
661 # Must call .lower() on the handle before creating the identity
662 assert ".lower()" in src, (
663 "musehub_auth.py does not normalize handle to lowercase — "
664 "Gabriel and gabriel could register as separate accounts"
665 )
666
667 def test_lowercase_normalization_precedes_identity_creation(self) -> None:
668 src = _AUTH_SVC.read_text()
669 lower_pos = src.find(".lower()")
670 identity_pos = src.find("MusehubIdentity(")
671 assert lower_pos < identity_pos, (
672 "Handle lowercasing must happen before MusehubIdentity() construction"
673 )
674
675 def test_handle_strip_applied(self) -> None:
676 """Leading/trailing whitespace in handles must be stripped."""
677 src = _AUTH_SVC.read_text()
678 assert ".strip()" in src, (
679 "musehub_auth.py does not strip whitespace from handle"
680 )
681
682 def test_gabriel_and_gabriel_uppercase_normalize_to_same(self) -> None:
683 """Functional: normalization must make Gabriel == gabriel."""
684 handle_upper = "Gabriel"
685 handle_lower = handle_upper.strip().lower()
686 assert handle_lower == "gabriel"
687 assert handle_lower == handle_upper.strip().lower() # idempotent
688
689 def test_handle_with_spaces_stripped(self) -> None:
690 handle = " gabriel "
691 normalized = handle.strip().lower()
692 assert normalized == "gabriel"
693
694
695 # ═══════════════════════════════════════════════════════════════════════════════
696 # MCP prompt injection
697 # ═══════════════════════════════════════════════════════════════════════════════
698
699 class TestMcpPromptInjection:
700 def test_dispatcher_wraps_result_in_delimiter_tags(self) -> None:
701 src = _MCP_DISPATCHER.read_text()
702 assert "<musehub_tool_result>" in src, (
703 "MCP dispatcher does not wrap tool results in <musehub_tool_result> delimiter"
704 )
705 assert "</musehub_tool_result>" in src
706
707 def test_delimiter_appears_in_success_path_not_error(self) -> None:
708 """Delimiter should only wrap successful results, not error envelopes."""
709 src = _MCP_DISPATCHER.read_text()
710 # Find the success block (isError: False) and confirm delimiter is there
711 ok_section = src[src.find("isError"):]
712 first_ok = ok_section.find("False")
713 # The delimiter must appear before the first isError: False
714 delimiter_pos = src.find("<musehub_tool_result>")
715 assert delimiter_pos < src.find('"isError": False'), (
716 "<musehub_tool_result> delimiter must appear in the success response path"
717 )
718
719 def test_prompts_instructs_model_to_treat_tool_results_as_data(self) -> None:
720 src = _PROMPTS.read_text()
721 assert "musehub_tool_result" in src, (
722 "prompts.py does not mention <musehub_tool_result> — "
723 "model has no instruction to treat tool results as untrusted data"
724 )
725
726 def test_prompts_mentions_prompt_injection_risk(self) -> None:
727 src = _PROMPTS.read_text()
728 assert "prompt" in src.lower() and "inject" in src.lower(), (
729 "prompts.py does not warn about prompt injection — agents are unprotected"
730 )
731
732 def test_prompts_instructs_treat_as_data_not_instructions(self) -> None:
733 src = _PROMPTS.read_text()
734 # Must say something like "treat as data" or "not as instructions"
735 assert "data" in src.lower() and (
736 "instruction" in src.lower() or "directive" in src.lower()
737 ), (
738 "prompts.py does not instruct the model to treat tool results as data, "
739 "not instructions"
740 )
741
742 def test_prompts_names_user_controlled_fields(self) -> None:
743 """Prompt must call out which fields are user-controlled (not vague)."""
744 src = _PROMPTS.read_text()
745 user_fields = ["commit message", "issue", "file path", "repository name", "branch name"]
746 found = [f for f in user_fields if f in src.lower()]
747 assert len(found) >= 3, (
748 f"prompts.py names only {found} as user-controlled — should name commit messages, "
749 "issue bodies, file paths, repo names, and branch names"
750 )
751
752 async def test_healthz_tool_result_not_wrapped(self, client: AsyncClient) -> None:
753 """/healthz returns plain JSON — not an MCP tool call, no wrapping needed."""
754 resp = await client.get("/healthz")
755 text = resp.text
756 assert "<musehub_tool_result>" not in text, (
757 "/healthz response should be plain JSON, not wrapped in MCP delimiters"
758 )
759
760
761 # ═══════════════════════════════════════════════════════════════════════════════
762 # Agent impersonation
763 # ═══════════════════════════════════════════════════════════════════════════════
764
765 class TestAgentImpersonation:
766 def test_trusted_agent_ids_setting_exists(self) -> None:
767 src = _CONFIG.read_text()
768 assert "trusted_agent_ids" in src, (
769 "Settings does not have a trusted_agent_ids field"
770 )
771
772 def test_wire_push_checks_agent_ids(self) -> None:
773 src = _WIRE_SVC.read_text()
774 assert "trusted_agent_ids" in src, (
775 "wire_push does not check agent_id against trusted_agent_ids"
776 )
777
778 def test_wire_push_flags_not_rejects(self) -> None:
779 """Unknown agents must be flagged, never rejected."""
780 src = _WIRE_SVC.read_text()
781 agent_section = src[src.find("trusted_agent_ids"):]
782 # Rejection would look like return WirePushResponse(ok=False, ...)
783 # after the trusted check — there must be no such rejection
784 assert "ok=False" not in agent_section[:500], (
785 "wire_push rejects unknown agents — must only flag them"
786 )
787 assert "untrusted_agent" in agent_section[:600], (
788 "wire_push does not flag unknown agents with untrusted_agent metadata"
789 )
790
791 def test_unknown_agent_flagged_in_metadata(self) -> None:
792 """untrusted_agent flag must be injected into commit metadata."""
793 src = _WIRE_SVC.read_text()
794 assert "\"untrusted_agent\"" in src or "'untrusted_agent'" in src, (
795 "wire_push does not inject untrusted_agent flag into commit metadata"
796 )
797
798 def test_wire_push_stream_flags_unknown_agents(self) -> None:
799 """wire_push_stream must flag unknown agent_ids with a warning."""
800 src = _WIRE_SVC.read_text()
801 wire_stream_src = src[src.find("async def wire_push_stream"):]
802 assert "trusted_agent_ids" in wire_stream_src, (
803 "wire_push_stream does not check trusted_agent_ids"
804 )
805 assert "untrusted_agent" in wire_stream_src, (
806 "wire_push_stream does not inject untrusted_agent flag into commit metadata"
807 )
808
809 def test_wire_push_stream_checks_trusted_prefix(self) -> None:
810 """wire_push_stream uses prefix matching for trusted agent IDs."""
811 src = _WIRE_SVC.read_text()
812 wire_stream_src = src[src.find("async def wire_push_stream"):]
813 assert "startswith" in wire_stream_src, (
814 "wire_push_stream must use prefix matching (startswith) for trusted_agent_ids"
815 )
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago