gabriel / musehub public
test_security_hardening_section8.py python
523 lines 26.0 KB
Raw
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 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 re
18 from pathlib import Path
19 from unittest.mock import AsyncMock, MagicMock, patch
20
21 import pytest
22
23 _ROOT = Path(__file__).resolve().parents[1]
24 _MUSEHUB_PKG = _ROOT / "musehub"
25 _SSRF_MODULE = _MUSEHUB_PKG / "security" / "ssrf.py"
26 _WEBHOOK_MODEL = _MUSEHUB_PKG / "models" / "musehub.py"
27 _DISPATCHER = _MUSEHUB_PKG / "services" / "musehub_webhook_dispatcher.py"
28 _WIRE_SVC = _MUSEHUB_PKG / "services" / "musehub_wire.py"
29 _CONFIG = _MUSEHUB_PKG / "config.py"
30
31
32 # ═══════════════════════════════════════════════════════════════════════════════
33 # SSRF — check_url_safe (sync, no DNS)
34 # ═══════════════════════════════════════════════════════════════════════════════
35
36 class TestSsrfCheckUrlSafe:
37 def _check(self, url: str) -> str:
38 from musehub.security.ssrf import check_url_safe
39 return check_url_safe(url)
40
41 def test_https_public_url_allowed(self):
42 result = self._check("https://hooks.example.com/deliver")
43 assert result == "https://hooks.example.com/deliver"
44
45 def test_http_scheme_blocked(self):
46 with pytest.raises(ValueError, match="https://"):
47 self._check("http://hooks.example.com/deliver")
48
49 def test_ftp_scheme_blocked(self):
50 with pytest.raises(ValueError, match="https://"):
51 self._check("ftp://hooks.example.com/deliver")
52
53 def test_file_scheme_blocked(self):
54 with pytest.raises(ValueError, match="https://"):
55 self._check("file:///etc/passwd")
56
57 def test_loopback_127_blocked(self):
58 with pytest.raises(ValueError, match="private/reserved"):
59 self._check("https://127.0.0.1/internal")
60
61 def test_loopback_127_other_blocked(self):
62 with pytest.raises(ValueError, match="private/reserved"):
63 self._check("https://127.255.255.255/internal")
64
65 def test_rfc1918_10_blocked(self):
66 with pytest.raises(ValueError, match="private/reserved"):
67 self._check("https://10.0.0.1/")
68
69 def test_rfc1918_172_16_blocked(self):
70 with pytest.raises(ValueError, match="private/reserved"):
71 self._check("https://172.16.0.1/")
72
73 def test_rfc1918_192_168_blocked(self):
74 with pytest.raises(ValueError, match="private/reserved"):
75 self._check("https://192.168.1.1/")
76
77 def test_link_local_169_254_blocked(self):
78 """169.254.169.254 is the AWS instance metadata endpoint."""
79 with pytest.raises(ValueError, match="private/reserved"):
80 self._check("https://169.254.169.254/latest/meta-data/")
81
82 def test_hostname_with_path_allowed(self):
83 result = self._check("https://api.stripe.com/v1/webhooks")
84 assert "stripe.com" in result
85
86 def test_no_hostname_blocked(self):
87 with pytest.raises(ValueError):
88 self._check("https:///path")
89
90 def test_empty_string_blocked(self):
91 with pytest.raises(ValueError):
92 self._check("")
93
94
95 # ═══════════════════════════════════════════════════════════════════════════════
96 # SSRF — validate_outbound_url (async, with DNS)
97 # ═══════════════════════════════════════════════════════════════════════════════
98
99 class TestSsrfValidateOutboundUrl:
100 @pytest.mark.anyio
101 async def test_public_hostname_allowed(self):
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 @pytest.mark.anyio
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 @pytest.mark.anyio
121 async def test_dns_resolves_to_loopback_blocked(self):
122 from musehub.security.ssrf import validate_outbound_url
123 import socket
124 with patch("musehub.security.ssrf.socket.getaddrinfo",
125 return_value=[(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("127.0.0.1", 443))]):
126 with pytest.raises(ValueError, match="private/reserved"):
127 await validate_outbound_url("https://localhost.example.com/")
128
129 @pytest.mark.anyio
130 async def test_unresolvable_hostname_blocked(self):
131 from musehub.security.ssrf import validate_outbound_url
132 import socket
133 with patch("musehub.security.ssrf.socket.getaddrinfo",
134 side_effect=socket.gaierror("Name or service not known")):
135 with pytest.raises(ValueError, match="cannot be resolved"):
136 await validate_outbound_url("https://nonexistent.invalid/")
137
138 @pytest.mark.anyio
139 async def test_bare_ip_does_not_do_dns(self):
140 """Bare IP literals must not trigger DNS resolution."""
141 from musehub.security.ssrf import validate_outbound_url
142 # If DNS is called for a bare IP, the test would hang or fail.
143 with patch("musehub.security.ssrf.socket.getaddrinfo") as mock_dns:
144 with pytest.raises(ValueError, match="private/reserved"):
145 await validate_outbound_url("https://10.0.0.1/")
146 mock_dns.assert_not_called()
147
148
149 # ═══════════════════════════════════════════════════════════════════════════════
150 # SSRF — WebhookCreate validator
151 # ═══════════════════════════════════════════════════════════════════════════════
152
153 class TestWebhookCreateSsrf:
154 def _create(self, url: str):
155 from musehub.models.musehub import WebhookCreate
156 return WebhookCreate(url=url, events=["push"])
157
158 def test_https_public_accepted(self):
159 wh = self._create("https://hooks.example.com/musehub")
160 assert wh.url == "https://hooks.example.com/musehub"
161
162 def test_http_rejected_at_model_parse(self):
163 from pydantic import ValidationError
164 with pytest.raises(ValidationError, match="https://"):
165 self._create("http://hooks.example.com/musehub")
166
167 def test_rfc1918_rejected_at_model_parse(self):
168 from pydantic import ValidationError
169 with pytest.raises(ValidationError, match="private/reserved"):
170 self._create("https://192.168.0.1/hook")
171
172 def test_loopback_rejected_at_model_parse(self):
173 from pydantic import ValidationError
174 with pytest.raises(ValidationError, match="private/reserved"):
175 self._create("https://127.0.0.1/hook")
176
177 def test_aws_metadata_endpoint_rejected(self):
178 from pydantic import ValidationError
179 with pytest.raises(ValidationError, match="private/reserved"):
180 self._create("https://169.254.169.254/latest/meta-data/iam/")
181
182
183 # ═══════════════════════════════════════════════════════════════════════════════
184 # SSRF — defense-in-depth in _attempt_delivery
185 # ═══════════════════════════════════════════════════════════════════════════════
186
187 class TestWebhookDeliveryDefenseInDepth:
188 @pytest.mark.anyio
189 async def test_delivery_blocked_when_ssrf_check_fails(self):
190 """_attempt_delivery returns failure when the URL fails SSRF validation."""
191 from musehub.services.musehub_webhook_dispatcher import _attempt_delivery
192 from musehub.db.musehub_models import MusehubWebhook
193
194 webhook = MagicMock(spec=MusehubWebhook)
195 webhook.url = "https://192.168.1.1/hook" # private IP
196 webhook.secret = ""
197
198 client = AsyncMock()
199 success, status, msg = await _attempt_delivery(
200 client,
201 webhook=webhook,
202 event_type="push",
203 payload_bytes=b"{}",
204 delivery_id="test-id",
205 attempt=1,
206 )
207
208 assert success is False
209 assert "SSRF" in msg or "blocked" in msg.lower()
210 client.post.assert_not_called()
211
212 def test_dispatcher_imports_ssrf_module(self):
213 src = _DISPATCHER.read_text()
214 assert "validate_outbound_url" in src or "ssrf" in src.lower(), (
215 "musehub_webhook_dispatcher.py must use validate_outbound_url for defence-in-depth"
216 )
217
218
219 # ═══════════════════════════════════════════════════════════════════════════════
220 # Mass assignment
221 # ═══════════════════════════════════════════════════════════════════════════════
222
223 class TestMassAssignment:
224 def _route_files(self) -> list[Path]:
225 return list((_MUSEHUB_PKG / "api" / "routes").rglob("*.py"))
226
227 def test_no_kwargs_splat_from_request_body(self):
228 """No route handler should unpack request.body or raw dict into ORM."""
229 dangerous_patterns = [
230 re.compile(r"\*\*request\.body"),
231 re.compile(r"\*\*await request\.json"),
232 re.compile(r"dict\(request\)"),
233 ]
234 violations: list[str] = []
235 for path in self._route_files():
236 src = path.read_text()
237 for pat in dangerous_patterns:
238 if pat.search(src):
239 violations.append(f"{path.name}: {pat.pattern}")
240 assert not violations, f"Potential mass assignment found: {violations}"
241
242 def test_route_handler_body_params_are_pydantic_or_primitive(self):
243 """Spot-check that handler params with complex types are Pydantic models."""
244 # The most dangerous pattern is `body: dict = Body(...)` — raw dict
245 # passed to ORM. We check that no handler uses `dict` as a body type.
246 body_dict_re = re.compile(r":\s*dict\s*=\s*(?:Body|Depends)\(")
247 violations: list[str] = []
248 for path in self._route_files():
249 src = path.read_text()
250 if body_dict_re.search(src):
251 violations.append(path.name)
252 assert not violations, (
253 f"Raw dict Body() parameter found (mass assignment risk): {violations}"
254 )
255
256 def test_webhook_create_uses_pydantic_not_raw_dict(self):
257 src = (_MUSEHUB_PKG / "api" / "routes" / "musehub" / "webhooks.py").read_text()
258 assert "WebhookCreate" in src, (
259 "Webhook creation route does not use WebhookCreate Pydantic model"
260 )
261
262
263 # ═══════════════════════════════════════════════════════════════════════════════
264 # Timing attacks
265 # ═══════════════════════════════════════════════════════════════════════════════
266
267 class TestTimingAttacks:
268 def test_request_signing_uses_cryptographic_verify(self):
269 """MSign uses Ed25519 verify_signature — cryptographically constant-time."""
270 src = (_MUSEHUB_PKG / "auth" / "request_signing.py").read_text()
271 assert "verify_signature" in src, (
272 "request_signing.py must call verify_signature for MSign auth"
273 )
274 # Must NOT use == to compare signature bytes (timing leak)
275 assert "sig_bytes ==" not in src and "== sig_bytes" not in src, (
276 "request_signing.py uses == for signature bytes — timing attack"
277 )
278
279 def test_crypto_keys_uses_compare_digest(self):
280 src = (_MUSEHUB_PKG / "crypto" / "keys.py").read_text()
281 assert "compare_digest" in src, (
282 "crypto/keys.py must use hmac.compare_digest"
283 )
284
285 def test_compare_digest_used_not_equal_for_secrets(self):
286 """Grep all auth/crypto files for == comparisons against secret strings."""
287 secret_eq_re = re.compile(
288 r"(secret|token|password|sig|signature|key)\s*==\s*[a-zA-Z_]",
289 re.IGNORECASE,
290 )
291 checked_files = [
292 _MUSEHUB_PKG / "auth" / "request_signing.py",
293 _MUSEHUB_PKG / "crypto" / "keys.py",
294 _MUSEHUB_PKG / "services" / "musehub_auth.py",
295 ]
296 violations: list[str] = []
297 for path in checked_files:
298 for lineno, line in enumerate(path.read_text().splitlines(), 1):
299 stripped = line.strip()
300 if stripped.startswith("#"):
301 continue
302 if secret_eq_re.search(stripped):
303 violations.append(f"{path.name}:{lineno}: {stripped[:80]}")
304 assert not violations, (
305 f"Potential timing-unsafe comparison found:\n" + "\n".join(violations)
306 )
307
308
309 # ═══════════════════════════════════════════════════════════════════════════════
310 # Object enumeration
311 # ═══════════════════════════════════════════════════════════════════════════════
312
313 class TestObjectEnumeration:
314 def _db_models_src(self) -> str:
315 return (_MUSEHUB_PKG / "db" / "musehub_models.py").read_text()
316
317 def test_repo_primary_key_is_uuid_string(self):
318 src = self._db_models_src()
319 # repo_id must be String(36) default=_new_uuid — not Integer
320 m = re.search(r"repo_id.*primary_key=True", src)
321 assert m, "repo_id primary key not found in musehub_models.py"
322 # Check the line includes String, not Integer
323 line = [l for l in src.splitlines() if "repo_id" in l and "primary_key=True" in l]
324 assert line, "repo_id primary key line not found"
325 assert "String" in line[0], (
326 f"repo_id primary key is not a String (UUID): {line[0]}"
327 )
328 assert "Integer" not in line[0], (
329 "repo_id primary key is Integer — sequential IDs enable enumeration"
330 )
331
332 def test_issue_primary_key_is_uuid_string(self):
333 src = self._db_models_src()
334 lines = [l for l in src.splitlines() if "issue_id" in l and "primary_key=True" in l]
335 assert lines, "issue_id primary key line not found"
336 assert "String" in lines[0], (
337 f"issue_id primary key is not a UUID String: {lines[0]}"
338 )
339
340 def test_issue_number_is_per_repo_not_global(self):
341 """Per-repo sequential numbers (like GitHub) are acceptable;
342 a global monotone counter would enable cross-repo enumeration."""
343 src = self._db_models_src()
344 # Verify issue number is defined in the MusehubIssue table context
345 # (per-repo scoped by unique_constraint on (repo_id, number)).
346 assert "number" in src and "repo_id" in src, (
347 "Could not find issue number + repo_id in musehub_models.py"
348 )
349
350 def test_no_autoincrement_primary_keys_on_main_entities(self):
351 """Key entity tables must not use INTEGER AUTOINCREMENT as PK."""
352 src = self._db_models_src()
353 # Collect lines that set primary_key=True
354 pk_lines = [
355 (i + 1, l.strip())
356 for i, l in enumerate(src.splitlines())
357 if "primary_key=True" in l
358 ]
359 for lineno, line in pk_lines:
360 # If it says Integer AND primary_key=True AND not in a composite key,
361 # it is a sequential auto-increment PK — flag it.
362 if "Integer" in line and "primary_key=True" in line:
363 # Allow composite keys (they never appear alone as the entity ID)
364 # by checking the field name; reject generic 'id' or '<entity>_id'.
365 if re.search(r"\b(id|_id)\b", line):
366 pytest.fail(
367 f"musehub_models.py:{lineno}: Integer primary key found — "
368 f"use UUID strings to prevent enumeration:\n {line}"
369 )
370
371
372 # ═══════════════════════════════════════════════════════════════════════════════
373 # Commit ID forgery / server-side signature enforcement
374 # ═══════════════════════════════════════════════════════════════════════════════
375
376 class TestCommitIdForgery:
377 def test_require_signed_commits_setting_exists(self):
378 src = _CONFIG.read_text()
379 assert "require_signed_commits" in src, (
380 "Settings does not have a require_signed_commits field"
381 )
382
383 def test_wire_push_enforces_when_setting_true(self):
384 """wire_push must reject unsigned commits when require_signed_commits=True."""
385 src = _WIRE_SVC.read_text()
386 assert "require_signed_commits" in src, (
387 "musehub_wire.py does not enforce require_signed_commits"
388 )
389 assert "unsigned" in src.lower() or "signature" in src.lower(), (
390 "musehub_wire.py does not reference signature enforcement"
391 )
392
393 @pytest.mark.anyio
394 async def test_wire_push_rejects_unsigned_when_enforcement_on(self):
395 from musehub.services.musehub_wire import wire_push
396 from musehub.models.wire import WirePushRequest, WireBundle, WireCommit
397
398 unsigned_commit = WireCommit(
399 commit_id="abc123",
400 message="test commit",
401 signature="", # no signature
402 signer_key_id="", # no signer
403 )
404 req = WirePushRequest(
405 bundle=WireBundle(commits=[unsigned_commit]),
406 branch="main",
407 )
408
409 mock_session = AsyncMock()
410 mock_repo = MagicMock()
411 mock_repo.deleted_at = None
412 mock_repo.owner = "gabriel"
413 mock_session.get.return_value = mock_repo
414
415 with patch("musehub.services.musehub_wire.settings") as mock_settings:
416 mock_settings.require_signed_commits = True
417 mock_settings.per_repo_quota_bytes = 0
418 result = await wire_push(mock_session, "repo-123", req, pusher_id="gabriel")
419
420 assert result.ok is False
421 assert "unsigned" in result.message.lower() or "sign" in result.message.lower()
422
423 def test_wire_push_source_has_unsigned_warning_not_rejection(self):
424 """When enforcement is off, wire_push logs a debug warning but does not reject."""
425 src = _WIRE_SVC.read_text()
426 # With enforcement off: a debug/warning log, not a WirePushResponse(ok=False).
427 # The rejection path is inside 'if settings.require_signed_commits:'.
428 # Outside that block there must be no unconditional rejection for unsigned.
429 # Verify the soft-path uses logger.debug, not an early return.
430 enforcement_block = src[src.find("require_signed_commits"):]
431 else_block_start = enforcement_block.find("else:")
432 assert else_block_start != -1, "No else branch for require_signed_commits"
433 else_block = enforcement_block[else_block_start:else_block_start + 400]
434 assert "logger.debug" in else_block or "logger.warning" in else_block, (
435 "Unsigned commit soft-path must log at debug/warning, not silently pass"
436 )
437 # The else branch must not contain a WirePushResponse(ok=False)
438 assert 'ok=False' not in else_block, (
439 "Unsigned commits are rejected even when require_signed_commits=False"
440 )
441
442
443 # ═══════════════════════════════════════════════════════════════════════════════
444 # Regex DoS
445 # ═══════════════════════════════════════════════════════════════════════════════
446
447 class TestRegexDos:
448 def _search_service_src(self) -> str:
449 return (_MUSEHUB_PKG / "services" / "musehub_search.py").read_text()
450
451 def test_search_by_pattern_uses_python_in_not_regex(self):
452 """search_by_pattern must use Python `in` operator, not re.compile(user_input)."""
453 src = self._search_service_src()
454 # find the search_by_pattern function body
455 fn_start = src.find("def search_by_pattern")
456 fn_end = src.find("\nasync def ", fn_start + 1)
457 if fn_end == -1:
458 fn_end = len(src)
459 fn_body = src[fn_start:fn_end]
460 # Must not compile user pattern as regex
461 assert "re.compile" not in fn_body, (
462 "search_by_pattern uses re.compile on user input — ReDoS risk"
463 )
464 # Should use substring `in` operator
465 assert " in " in fn_body, (
466 "search_by_pattern does not use substring 'in' operator"
467 )
468
469 def test_search_by_ask_uses_fixed_tokenizer_not_user_regex(self):
470 """search_by_ask must tokenize with a fixed pattern, not user-supplied."""
471 src = self._search_service_src()
472 fn_start = src.find("def search_by_ask")
473 fn_end = src.find("\nasync def ", fn_start + 1)
474 if fn_end == -1:
475 fn_end = len(src)
476 fn_body = src[fn_start:fn_end]
477 # Any re.compile inside must be a literal fixed pattern (not a variable)
478 compile_calls = re.findall(r"re\.compile\((.+?)\)", fn_body)
479 for call in compile_calls:
480 assert not re.match(r"^[a-zA-Z_]\w*$", call.strip()), (
481 f"search_by_ask compiles a variable as regex: re.compile({call})"
482 )
483
484 def test_no_user_supplied_regex_in_search_routes(self):
485 """Search API routes must not pass user query string to re.compile()."""
486 src = (_MUSEHUB_PKG / "api" / "routes" / "musehub" / "search.py").read_text()
487 # re.compile should not appear in the route handler file
488 assert "re.compile" not in src, (
489 "search.py compiles user input as regex — ReDoS risk"
490 )
491
492 def test_global_search_uses_sql_or_python_in(self):
493 """Global search must use parameterized SQL or Python `in`, not Python regex."""
494 src = self._search_service_src()
495 # Verify _TOKEN_RE is a fixed, pre-compiled pattern
496 token_re_line = [l for l in src.splitlines() if "_TOKEN_RE" in l and "compile" in l]
497 assert token_re_line, "_TOKEN_RE pre-compiled pattern not found in musehub_search.py"
498 # It should be a literal string, not a variable
499 assert "[a-z" in token_re_line[0] or "[A-Z" in token_re_line[0], (
500 "_TOKEN_RE should be a literal character-class pattern"
501 )
502
503
504 # ═══════════════════════════════════════════════════════════════════════════════
505 # Tar bomb / zip bomb (N/A)
506 # ═══════════════════════════════════════════════════════════════════════════════
507
508 class TestTarBomb:
509 def test_no_archive_extraction_in_codebase(self):
510 """MuseHub does not extract archives — this check ensures it stays that way."""
511 dangerous = ["tarfile.open", "tarfile.extractall", "zipfile.ZipFile", ".extractall("]
512 violations: list[str] = []
513 for py_file in _MUSEHUB_PKG.rglob("*.py"):
514 if "test_" in py_file.name:
515 continue
516 src = py_file.read_text()
517 for pattern in dangerous:
518 if pattern in src:
519 violations.append(f"{py_file.relative_to(_ROOT)}: {pattern}")
520 assert not violations, (
521 "Archive extraction found — add size/count limits before extractall():\n"
522 + "\n".join(violations)
523 )
File History 1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago