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