gabriel / musehub public
test_security_hardening.py python
921 lines 45.3 KB
Raw
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ breaking 156 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):
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):
48 with pytest.raises(ValueError, match="https://"):
49 self._check("http://hooks.example.com/deliver")
50
51 def test_ftp_scheme_blocked(self):
52 with pytest.raises(ValueError, match="https://"):
53 self._check("ftp://hooks.example.com/deliver")
54
55 def test_file_scheme_blocked(self):
56 with pytest.raises(ValueError, match="https://"):
57 self._check("file:///etc/passwd")
58
59 def test_loopback_127_blocked(self):
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):
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):
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):
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):
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):
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):
85 result = self._check("https://api.stripe.com/v1/webhooks")
86 assert "stripe.com" in result
87
88 def test_no_hostname_blocked(self):
89 with pytest.raises(ValueError):
90 self._check("https:///path")
91
92 def test_empty_string_blocked(self):
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):
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):
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):
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):
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):
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):
152 from musehub.models.musehub import WebhookCreate
153 return WebhookCreate(url=url, events=["push"])
154
155 def test_https_public_accepted(self):
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):
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):
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):
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):
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):
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):
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):
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):
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):
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):
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):
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):
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):
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):
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):
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):
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):
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):
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 async def test_wire_push_rejects_unsigned_when_enforcement_on(self):
390 from musehub.services.musehub_wire import wire_push
391 from musehub.models.wire import WirePushRequest, WireBundle, WireCommit
392
393 unsigned_commit = WireCommit(
394 commit_id="sha256:" + hashlib.sha256(b"unsigned-commit").hexdigest(),
395 message="test commit",
396 signature="", # no signature
397 signer_key_id="", # no signer
398 )
399 req = WirePushRequest(
400 bundle=WireBundle(commits=[unsigned_commit]),
401 branch="main",
402 )
403
404 mock_session = AsyncMock()
405 mock_repo = MagicMock()
406 mock_repo.owner = "gabriel"
407 mock_session.get.return_value = mock_repo
408
409 with patch("musehub.services.musehub_wire.settings") as mock_settings:
410 mock_settings.require_signed_commits = True
411 mock_settings.per_repo_quota_bytes = 0
412 result = await wire_push(mock_session, "repo-123", req, pusher_id="gabriel")
413
414 assert result.ok is False
415 assert "unsigned" in result.message.lower() or "sign" in result.message.lower()
416
417 def test_wire_push_source_has_unsigned_warning_not_rejection(self):
418 """When enforcement is off, wire_push logs a debug warning but does not reject."""
419 src = _WIRE_SVC.read_text()
420 # With enforcement off: a debug/warning log, not a WirePushResponse(ok=False).
421 # The rejection path is inside 'if settings.require_signed_commits:'.
422 # Outside that block there must be no unconditional rejection for unsigned.
423 # Verify the soft-path uses logger.debug, not an early return.
424 enforcement_block = src[src.find("require_signed_commits"):]
425 else_block_start = enforcement_block.find("else:")
426 assert else_block_start != -1, "No else branch for require_signed_commits"
427 else_block = enforcement_block[else_block_start:else_block_start + 400]
428 assert "logger.debug" in else_block or "logger.warning" in else_block, (
429 "Unsigned commit soft-path must log at debug/warning, not silently pass"
430 )
431 # The else branch must not contain a WirePushResponse(ok=False)
432 assert 'ok=False' not in else_block, (
433 "Unsigned commits are rejected even when require_signed_commits=False"
434 )
435
436
437 # ═══════════════════════════════════════════════════════════════════════════════
438 # Regex DoS
439 # ═══════════════════════════════════════════════════════════════════════════════
440
441 class TestRegexDos:
442 def _search_service_src(self) -> str:
443 return (_MUSEHUB_PKG / "services" / "musehub_search.py").read_text()
444
445 def test_search_by_pattern_uses_python_in_not_regex(self):
446 """search_by_pattern must use Python `in` operator, not re.compile(user_input)."""
447 src = self._search_service_src()
448 # find the search_by_pattern function body
449 fn_start = src.find("def search_by_pattern")
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 # Must not compile user pattern as regex
455 assert "re.compile" not in fn_body, (
456 "search_by_pattern uses re.compile on user input — ReDoS risk"
457 )
458 # Should use substring `in` operator
459 assert " in " in fn_body, (
460 "search_by_pattern does not use substring 'in' operator"
461 )
462
463 def test_search_by_ask_uses_fixed_tokenizer_not_user_regex(self):
464 """search_by_ask must tokenize with a fixed pattern, not user-supplied."""
465 src = self._search_service_src()
466 fn_start = src.find("def search_by_ask")
467 fn_end = src.find("\nasync def ", fn_start + 1)
468 if fn_end == -1:
469 fn_end = len(src)
470 fn_body = src[fn_start:fn_end]
471 # Any re.compile inside must be a literal fixed pattern (not a variable)
472 compile_calls = re.findall(r"re\.compile\((.+?)\)", fn_body)
473 for call in compile_calls:
474 assert not re.match(r"^[a-zA-Z_]\w*$", call.strip()), (
475 f"search_by_ask compiles a variable as regex: re.compile({call})"
476 )
477
478 def test_no_user_supplied_regex_in_search_routes(self):
479 """Search API routes must not pass user query string to re.compile()."""
480 src = (_MUSEHUB_PKG / "api" / "routes" / "musehub" / "search.py").read_text()
481 # re.compile should not appear in the route handler file
482 assert "re.compile" not in src, (
483 "search.py compiles user input as regex — ReDoS risk"
484 )
485
486 def test_global_search_uses_sql_or_python_in(self):
487 """Global search must use parameterized SQL or Python `in`, not Python regex."""
488 src = self._search_service_src()
489 # Verify _TOKEN_RE is a fixed, pre-compiled pattern
490 token_re_line = [l for l in src.splitlines() if "_TOKEN_RE" in l and "compile" in l]
491 assert token_re_line, "_TOKEN_RE pre-compiled pattern not found in musehub_search.py"
492 # It should be a literal string, not a variable
493 assert "[a-z" in token_re_line[0] or "[A-Z" in token_re_line[0], (
494 "_TOKEN_RE should be a literal character-class pattern"
495 )
496
497
498 # ═══════════════════════════════════════════════════════════════════════════════
499 # Tar bomb / zip bomb (N/A)
500 # ═══════════════════════════════════════════════════════════════════════════════
501
502 class TestTarBomb:
503 def test_no_archive_extraction_in_codebase(self):
504 """MuseHub does not extract archives — this check ensures it stays that way."""
505 dangerous = ["tarfile.open", "tarfile.extractall", "zipfile.ZipFile", ".extractall("]
506 violations: list[str] = []
507 for py_file in _MUSEHUB_PKG.rglob("*.py"):
508 if "test_" in py_file.name:
509 continue
510 src = py_file.read_text()
511 for pattern in dangerous:
512 if pattern in src:
513 violations.append(f"{py_file.relative_to(_ROOT)}: {pattern}")
514 assert not violations, (
515 "Archive extraction found — add size/count limits before extractall():\n"
516 + "\n".join(violations)
517 )
518
519
520
521
522 from httpx import AsyncClient
523
524 _ROOT = Path(__file__).resolve().parents[1]
525 _MUSEHUB_PKG = _ROOT / "musehub"
526 _MAIN_PY = _MUSEHUB_PKG / "main.py"
527 _ELICITATION = _MUSEHUB_PKG / "api" / "routes" / "musehub" / "ui_mcp_elicitation.py"
528 _AUTH_SVC = _MUSEHUB_PKG / "services" / "musehub_auth.py"
529 _MCP_DISPATCHER = _MUSEHUB_PKG / "mcp" / "dispatcher.py"
530 _PROMPTS = _MUSEHUB_PKG / "mcp" / "prompts.py"
531 _WIRE_SVC = _MUSEHUB_PKG / "services" / "musehub_wire.py"
532 _CONFIG = _MUSEHUB_PKG / "config.py"
533 _MAGIC_BYTES = _MUSEHUB_PKG / "security" / "magic_bytes.py"
534
535
536 # ═══════════════════════════════════════════════════════════════════════════════
537 # Polyglot files — magic bytes
538 # ═══════════════════════════════════════════════════════════════════════════════
539
540 class TestMagicBytes:
541 def _check(self, path: str, content: bytes) -> str:
542 from musehub.security.magic_bytes import check_magic_bytes
543 return check_magic_bytes(path, content)
544
545 def _expect_error(self, path: str, content: bytes) -> None:
546 from musehub.security.magic_bytes import check_magic_bytes, PolyglotFileError
547 with pytest.raises(PolyglotFileError):
548 check_magic_bytes(path, content)
549
550 # ── Valid files ──────────────────────────────────────────────────────────
551
552 def test_valid_midi(self):
553 midi_header = b"MThd\x00\x00\x00\x06\x00\x01\x00\x04\x01\xe0"
554 assert self._check("track.mid", midi_header) == "MIDI"
555
556 def test_valid_mp3_id3(self):
557 mp3_id3 = b"ID3\x03\x00\x00\x00\x00\x00\x00"
558 assert self._check("song.mp3", mp3_id3) == "MP3"
559
560 def test_valid_mp3_sync(self):
561 mp3_sync = bytes([0xFF, 0xFB]) + b"\x90\x00" * 10
562 assert self._check("song.mp3", mp3_sync) == "MP3"
563
564 def test_valid_webp(self):
565 webp = b"RIFF\x00\x00\x00\x00WEBP" + b"\x00" * 10
566 assert self._check("cover.webp", webp) == "WebP"
567
568 def test_valid_png(self):
569 png = bytes([0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A]) + b"\x00" * 10
570 assert self._check("cover.png", png) == "PNG"
571
572 def test_valid_jpeg(self):
573 jpeg = bytes([0xFF, 0xD8, 0xFF, 0xE0]) + b"\x00" * 10
574 assert self._check("cover.jpg", jpeg) == "JPEG"
575
576 def test_unknown_extension_passes_through(self):
577 # .py files are not in the known-type map — no check performed.
578 assert self._check("main.py", b"import os\n") == "unknown"
579
580 def test_script_shebang_in_py_file_allowed(self):
581 # Python scripts legitimately start with #! — this is NOT a polyglot attack.
582 shebang = b"#!/usr/bin/env python3\ndef main(): pass\n"
583 assert self._check("tools/audit.py", shebang) == "unknown"
584
585 def test_script_shebang_in_sh_file_allowed(self):
586 shebang = b"#!/bin/bash\necho hello\n"
587 assert self._check("scripts/deploy.sh", shebang) == "unknown"
588
589 def test_shebang_still_blocked_in_binary_extensions(self):
590 # A .mp3 with a shebang IS a polyglot attack.
591 self._expect_error("song.mp3", b"#!/bin/bash\nrm -rf /")
592
593 def test_shebang_still_blocked_in_midi_extension(self):
594 self._expect_error("track.mid", b"#!/usr/bin/php\n<?php system($_GET['cmd']); ?>")
595
596 def test_empty_content_returns_empty(self):
597 assert self._check("track.mid", b"") == "empty"
598
599 # ── Polyglot attacks ─────────────────────────────────────────────────────
600
601 def test_midi_extension_but_html_content_blocked(self):
602 self._expect_error("track.mid", b"<!DOCTYPE html><html><script>alert(1)</script>")
603
604 def test_midi_extension_but_php_shebang_blocked(self):
605 self._expect_error("track.mid", b"#!/usr/bin/php\n<?php system($_GET['cmd']); ?>")
606
607 def test_jpg_extension_but_html_content_blocked(self):
608 self._expect_error("cover.jpg", b"<html><body>not an image</body></html>")
609
610 def test_webp_extension_but_wrong_magic_blocked(self):
611 self._expect_error("cover.webp", b"\x00\x00\x00\x00\x00\x00\x00\x00")
612
613 def test_png_extension_but_wrong_magic_blocked(self):
614 self._expect_error("cover.png", b"JFIF\x00\x00")
615
616 def test_jpeg_extension_but_zip_content_blocked(self):
617 self._expect_error("cover.jpg", bytes([0x50, 0x4B, 0x03, 0x04]) + b"\x00" * 10)
618
619 def test_mp3_extension_but_html_shebang_blocked(self):
620 self._expect_error("song.mp3", b"#!/bin/bash\nrm -rf /")
621
622 def test_html_extension_allows_html_content(self):
623 # .html extension is exempt from the forbidden-HTML check.
624 result = self._check("readme.html", b"<!DOCTYPE html><html></html>")
625 assert result == "HTML" # returns "HTML" since the exemption triggers
626
627 # ── Wire push integration ─────────────────────────────────────────────────
628
629 def test_wire_push_imports_magic_bytes(self):
630 src = _WIRE_SVC.read_text()
631 assert "magic_bytes" in src or "PolyglotFileError" in src, (
632 "musehub_wire.py does not check magic bytes on push"
633 )
634
635 def test_wire_push_rejects_polyglot_path(self):
636 src = _WIRE_SVC.read_text()
637 assert "PolyglotFileError" in src, (
638 "wire_push does not catch PolyglotFileError"
639 )
640
641
642 # ═══════════════════════════════════════════════════════════════════════════════
643 # Clickjacking
644 # ═══════════════════════════════════════════════════════════════════════════════
645
646 class TestClickjacking:
647 async def test_x_frame_options_deny(self, client: AsyncClient):
648 resp = await client.get("/healthz")
649 assert resp.headers.get("x-frame-options", "").upper() == "DENY", (
650 "X-Frame-Options: DENY not set — clickjacking risk"
651 )
652
653 async def test_csp_frame_ancestors_none(self, client: AsyncClient):
654 resp = await client.get("/healthz")
655 csp = resp.headers.get("content-security-policy", "")
656 assert "frame-ancestors" in csp and "'none'" in csp, (
657 "CSP frame-ancestors 'none' not set — clickjacking risk"
658 )
659
660 def test_security_headers_middleware_sets_x_frame_options(self):
661 src = _MAIN_PY.read_text()
662 assert "X-Frame-Options" in src, (
663 "SecurityHeadersMiddleware does not set X-Frame-Options"
664 )
665 assert "DENY" in src
666
667 def test_security_headers_middleware_sets_frame_ancestors(self):
668 src = _MAIN_PY.read_text()
669 assert "frame-ancestors" in src and "'none'" in src, (
670 "SecurityHeadersMiddleware CSP does not include frame-ancestors 'none'"
671 )
672
673
674 # ═══════════════════════════════════════════════════════════════════════════════
675 # Open redirect
676 # ═══════════════════════════════════════════════════════════════════════════════
677
678 class TestOpenRedirect:
679 def test_elicitation_stores_path_only_in_next_param(self):
680 src = _ELICITATION.read_text()
681 # Must use request.url.path, NOT request.url (which is the full absolute URL)
682 assert "request.url.path" in src, (
683 "ui_mcp_elicitation.py stores full absolute URL in ?next= — "
684 "open redirect: attacker can inject external URL via crafted host header"
685 )
686 # Must not store the full URL object directly
687 assert "f\"/login?next={callback}\"" in src or "/login?next=" in src, (
688 "?next= redirect not found in elicitation"
689 )
690
691 def test_elicitation_does_not_store_full_absolute_url(self):
692 src = _ELICITATION.read_text()
693 # Verify the callback variable is not assigned from bare request.url
694 # (without .path or .components)
695 bad_pattern = re.compile(r"callback\s*=\s*request\.url\b(?!\.path|\.query|\.components)")
696 assert not bad_pattern.search(src), (
697 "callback assigned from request.url (full URL) — should be request.url.path"
698 )
699
700 def test_elicitation_path_only_for_both_routes(self):
701 """Both elicitation routes must store path-only."""
702 src = _ELICITATION.read_text()
703 path_assignments = src.count("request.url.path")
704 assert path_assignments >= 2, (
705 f"Only {path_assignments} place(s) use request.url.path — "
706 "both elicitation routes must use path-only redirect"
707 )
708
709
710 # ═══════════════════════════════════════════════════════════════════════════════
711 # Handle squatting
712 # ═══════════════════════════════════════════════════════════════════════════════
713
714 class TestHandleSquatting:
715 def test_auth_service_normalizes_handle_to_lowercase(self):
716 src = _AUTH_SVC.read_text()
717 # Must call .lower() on the handle before creating the identity
718 assert ".lower()" in src, (
719 "musehub_auth.py does not normalize handle to lowercase — "
720 "Gabriel and gabriel could register as separate accounts"
721 )
722
723 def test_lowercase_normalization_precedes_identity_creation(self):
724 src = _AUTH_SVC.read_text()
725 lower_pos = src.find(".lower()")
726 identity_pos = src.find("MusehubIdentity(")
727 assert lower_pos < identity_pos, (
728 "Handle lowercasing must happen before MusehubIdentity() construction"
729 )
730
731 def test_handle_strip_applied(self):
732 """Leading/trailing whitespace in handles must be stripped."""
733 src = _AUTH_SVC.read_text()
734 assert ".strip()" in src, (
735 "musehub_auth.py does not strip whitespace from handle"
736 )
737
738 def test_gabriel_and_gabriel_uppercase_normalize_to_same(self):
739 """Functional: normalization must make Gabriel == gabriel."""
740 handle_upper = "Gabriel"
741 handle_lower = handle_upper.strip().lower()
742 assert handle_lower == "gabriel"
743 assert handle_lower == handle_upper.strip().lower() # idempotent
744
745 def test_handle_with_spaces_stripped(self):
746 handle = " gabriel "
747 normalized = handle.strip().lower()
748 assert normalized == "gabriel"
749
750
751 # ═══════════════════════════════════════════════════════════════════════════════
752 # MCP prompt injection
753 # ═══════════════════════════════════════════════════════════════════════════════
754
755 class TestMcpPromptInjection:
756 def test_dispatcher_wraps_result_in_delimiter_tags(self):
757 src = _MCP_DISPATCHER.read_text()
758 assert "<musehub_tool_result>" in src, (
759 "MCP dispatcher does not wrap tool results in <musehub_tool_result> delimiter"
760 )
761 assert "</musehub_tool_result>" in src
762
763 def test_delimiter_appears_in_success_path_not_error(self):
764 """Delimiter should only wrap successful results, not error envelopes."""
765 src = _MCP_DISPATCHER.read_text()
766 # Find the success block (isError: False) and confirm delimiter is there
767 ok_section = src[src.find("isError"):]
768 first_ok = ok_section.find("False")
769 # The delimiter must appear before the first isError: False
770 delimiter_pos = src.find("<musehub_tool_result>")
771 assert delimiter_pos < src.find('"isError": False'), (
772 "<musehub_tool_result> delimiter must appear in the success response path"
773 )
774
775 def test_prompts_instructs_model_to_treat_tool_results_as_data(self):
776 src = _PROMPTS.read_text()
777 assert "musehub_tool_result" in src, (
778 "prompts.py does not mention <musehub_tool_result> — "
779 "model has no instruction to treat tool results as untrusted data"
780 )
781
782 def test_prompts_mentions_prompt_injection_risk(self):
783 src = _PROMPTS.read_text()
784 assert "prompt" in src.lower() and "inject" in src.lower(), (
785 "prompts.py does not warn about prompt injection — agents are unprotected"
786 )
787
788 def test_prompts_instructs_treat_as_data_not_instructions(self):
789 src = _PROMPTS.read_text()
790 # Must say something like "treat as data" or "not as instructions"
791 assert "data" in src.lower() and (
792 "instruction" in src.lower() or "directive" in src.lower()
793 ), (
794 "prompts.py does not instruct the model to treat tool results as data, "
795 "not instructions"
796 )
797
798 def test_prompts_names_user_controlled_fields(self):
799 """Prompt must call out which fields are user-controlled (not vague)."""
800 src = _PROMPTS.read_text()
801 user_fields = ["commit message", "issue", "file path", "repository name", "branch name"]
802 found = [f for f in user_fields if f in src.lower()]
803 assert len(found) >= 3, (
804 f"prompts.py names only {found} as user-controlled — should name commit messages, "
805 "issue bodies, file paths, repo names, and branch names"
806 )
807
808 async def test_healthz_tool_result_not_wrapped(self, client: AsyncClient):
809 """/healthz returns plain JSON — not an MCP tool call, no wrapping needed."""
810 resp = await client.get("/healthz")
811 text = resp.text
812 assert "<musehub_tool_result>" not in text, (
813 "/healthz response should be plain JSON, not wrapped in MCP delimiters"
814 )
815
816
817 # ═══════════════════════════════════════════════════════════════════════════════
818 # Agent impersonation
819 # ═══════════════════════════════════════════════════════════════════════════════
820
821 class TestAgentImpersonation:
822 def test_trusted_agent_ids_setting_exists(self):
823 src = _CONFIG.read_text()
824 assert "trusted_agent_ids" in src, (
825 "Settings does not have a trusted_agent_ids field"
826 )
827
828 def test_wire_push_checks_agent_ids(self):
829 src = _WIRE_SVC.read_text()
830 assert "trusted_agent_ids" in src, (
831 "wire_push does not check agent_id against trusted_agent_ids"
832 )
833
834 def test_wire_push_flags_not_rejects(self):
835 """Unknown agents must be flagged, never rejected."""
836 src = _WIRE_SVC.read_text()
837 agent_section = src[src.find("trusted_agent_ids"):]
838 # Rejection would look like return WirePushResponse(ok=False, ...)
839 # after the trusted check — there must be no such rejection
840 assert "ok=False" not in agent_section[:500], (
841 "wire_push rejects unknown agents — must only flag them"
842 )
843 assert "untrusted_agent" in agent_section[:600], (
844 "wire_push does not flag unknown agents with untrusted_agent metadata"
845 )
846
847 def test_unknown_agent_flagged_in_metadata(self):
848 """untrusted_agent flag must be injected into commit metadata."""
849 src = _WIRE_SVC.read_text()
850 assert "\"untrusted_agent\"" in src or "'untrusted_agent'" in src, (
851 "wire_push does not inject untrusted_agent flag into commit metadata"
852 )
853
854 async def test_unknown_agent_flagged_when_registry_set(self):
855 from musehub.services.musehub_wire import wire_push
856 from musehub.models.wire import WirePushRequest, WireBundle, WireCommit
857
858 commit = WireCommit(
859 commit_id="sha256:" + hashlib.sha256(b"evil-agent-commit").hexdigest(),
860 message="agent push",
861 agent_id="evil-agent/1.0",
862 )
863 req = WirePushRequest(bundle=WireBundle(commits=[commit]), branch="main")
864
865 mock_session = AsyncMock()
866 mock_repo = MagicMock()
867 mock_repo.owner = "gabriel"
868 mock_session.get.return_value = mock_repo
869
870 logged: list[str] = []
871 with patch("musehub.services.musehub_wire.settings") as mock_settings, \
872 patch("musehub.services.musehub_wire.logger") as mock_logger:
873 mock_settings.require_signed_commits = False
874 mock_settings.trusted_agent_ids = ["agentception-worker", "claude-opus-4-6"]
875 mock_settings.per_repo_quota_bytes = 0
876 mock_logger.warning.side_effect = lambda msg, *args, **kw: logged.append(
877 msg % args if args else msg
878 )
879 # We only need the flagging to run — the rest of push will fail on DB mocks
880 try:
881 await wire_push(mock_session, "repo-123", req, pusher_id="gabriel")
882 except Exception:
883 pass # DB mock failures are expected; we only care about the warning
884
885 assert any("untrusted" in m or "unknown agent" in m for m in logged), (
886 "Unknown agent_id was not flagged with a warning"
887 )
888
889 async def test_known_agent_not_flagged(self):
890 from musehub.services.musehub_wire import wire_push
891 from musehub.models.wire import WirePushRequest, WireBundle, WireCommit
892
893 commit = WireCommit(
894 commit_id="sha256:" + hashlib.sha256(b"known-agent-commit").hexdigest(),
895 message="known agent push",
896 agent_id="agentception-worker-42",
897 )
898 req = WirePushRequest(bundle=WireBundle(commits=[commit]), branch="main")
899
900 mock_session = AsyncMock()
901 mock_repo = MagicMock()
902 mock_repo.owner = "gabriel"
903 mock_session.get.return_value = mock_repo
904
905 logged: list[str] = []
906 with patch("musehub.services.musehub_wire.settings") as mock_settings, \
907 patch("musehub.services.musehub_wire.logger") as mock_logger:
908 mock_settings.require_signed_commits = False
909 mock_settings.trusted_agent_ids = ["agentception-worker"]
910 mock_settings.per_repo_quota_bytes = 0
911 mock_logger.warning.side_effect = lambda msg, *args, **kw: logged.append(
912 msg % args if args else msg
913 )
914 try:
915 await wire_push(mock_session, "repo-123", req, pusher_id="gabriel")
916 except Exception:
917 pass
918
919 assert not any("untrusted" in m for m in logged), (
920 "Known agent was incorrectly flagged as untrusted"
921 )
File History 1 commit
sha256:9590cee1e0ccd6c76528f005b95d634d80f5019f0dcb7c371e149adc31d1fb65 refactor: enforce gRPC framing on all MWP wire traffic Sonnet 4.6 minor ⚠ 156 days ago