"""Section 8 (part 2) — Advanced security hardening tests. Covers: Polyglot files : magic-bytes validation; extension/content mismatch rejected. Clickjacking : X-Frame-Options: DENY + CSP frame-ancestors 'none' in SecurityHeadersMiddleware. Open redirect : ?next= param stores path-only, never a full absolute URL; validate_redirect_target utility rejects external origins. Handle squatting : handle normalised to lowercase before DB insert; Gabriel and gabriel cannot coexist. MCP prompt inj. : tool results wrapped in tags; orientation prompt contains untrusted-content instruction. Agent impersonation: TRUSTED_AGENT_IDS flag; unknown agents flagged in metadata, never rejected. """ from __future__ import annotations import re from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch import pytest from httpx import AsyncClient _ROOT = Path(__file__).resolve().parents[1] _MUSEHUB_PKG = _ROOT / "musehub" _MAIN_PY = _MUSEHUB_PKG / "main.py" _ELICITATION = _MUSEHUB_PKG / "api" / "routes" / "musehub" / "ui_mcp_elicitation.py" _AUTH_SVC = _MUSEHUB_PKG / "services" / "musehub_auth.py" _DISPATCHER = _MUSEHUB_PKG / "mcp" / "dispatcher.py" _PROMPTS = _MUSEHUB_PKG / "mcp" / "prompts.py" _WIRE_SVC = _MUSEHUB_PKG / "services" / "musehub_wire.py" _CONFIG = _MUSEHUB_PKG / "config.py" _MAGIC_BYTES = _MUSEHUB_PKG / "security" / "magic_bytes.py" # ═══════════════════════════════════════════════════════════════════════════════ # Polyglot files — magic bytes # ═══════════════════════════════════════════════════════════════════════════════ class TestMagicBytes: def _check(self, path: str, content: bytes) -> str: from musehub.security.magic_bytes import check_magic_bytes return check_magic_bytes(path, content) def _expect_error(self, path: str, content: bytes) -> None: from musehub.security.magic_bytes import check_magic_bytes, PolyglotFileError with pytest.raises(PolyglotFileError): check_magic_bytes(path, content) # ── Valid files ────────────────────────────────────────────────────────── def test_valid_midi(self): midi_header = b"MThd\x00\x00\x00\x06\x00\x01\x00\x04\x01\xe0" assert self._check("track.mid", midi_header) == "MIDI" def test_valid_mp3_id3(self): mp3_id3 = b"ID3\x03\x00\x00\x00\x00\x00\x00" assert self._check("song.mp3", mp3_id3) == "MP3" def test_valid_mp3_sync(self): mp3_sync = bytes([0xFF, 0xFB]) + b"\x90\x00" * 10 assert self._check("song.mp3", mp3_sync) == "MP3" def test_valid_webp(self): webp = b"RIFF\x00\x00\x00\x00WEBP" + b"\x00" * 10 assert self._check("cover.webp", webp) == "WebP" def test_valid_png(self): png = bytes([0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A]) + b"\x00" * 10 assert self._check("cover.png", png) == "PNG" def test_valid_jpeg(self): jpeg = bytes([0xFF, 0xD8, 0xFF, 0xE0]) + b"\x00" * 10 assert self._check("cover.jpg", jpeg) == "JPEG" def test_unknown_extension_passes_through(self): # .py files are not in the known-type map — no check performed. assert self._check("main.py", b"import os\n") == "unknown" def test_empty_content_returns_empty(self): assert self._check("track.mid", b"") == "empty" # ── Polyglot attacks ───────────────────────────────────────────────────── def test_midi_extension_but_html_content_blocked(self): self._expect_error("track.mid", b"") def test_midi_extension_but_php_shebang_blocked(self): self._expect_error("track.mid", b"#!/usr/bin/php\n") def test_jpg_extension_but_html_content_blocked(self): self._expect_error("cover.jpg", b"not an image") def test_webp_extension_but_wrong_magic_blocked(self): self._expect_error("cover.webp", b"\x00\x00\x00\x00\x00\x00\x00\x00") def test_png_extension_but_wrong_magic_blocked(self): self._expect_error("cover.png", b"JFIF\x00\x00") def test_jpeg_extension_but_zip_content_blocked(self): self._expect_error("cover.jpg", bytes([0x50, 0x4B, 0x03, 0x04]) + b"\x00" * 10) def test_mp3_extension_but_html_shebang_blocked(self): self._expect_error("song.mp3", b"#!/bin/bash\nrm -rf /") def test_html_extension_allows_html_content(self): # .html extension is exempt from the forbidden-HTML check. result = self._check("readme.html", b"") assert result == "HTML" # returns "HTML" since the exemption triggers # ── Wire push integration ───────────────────────────────────────────────── def test_wire_push_imports_magic_bytes(self): src = _WIRE_SVC.read_text() assert "magic_bytes" in src or "PolyglotFileError" in src, ( "musehub_wire.py does not check magic bytes on push" ) def test_wire_push_rejects_polyglot_path(self): src = _WIRE_SVC.read_text() assert "PolyglotFileError" in src, ( "wire_push does not catch PolyglotFileError" ) # ═══════════════════════════════════════════════════════════════════════════════ # Clickjacking # ═══════════════════════════════════════════════════════════════════════════════ class TestClickjacking: @pytest.mark.anyio async def test_x_frame_options_deny(self, client: AsyncClient): resp = await client.get("/healthz") assert resp.headers.get("x-frame-options", "").upper() == "DENY", ( "X-Frame-Options: DENY not set — clickjacking risk" ) @pytest.mark.anyio async def test_csp_frame_ancestors_none(self, client: AsyncClient): resp = await client.get("/healthz") csp = resp.headers.get("content-security-policy", "") assert "frame-ancestors" in csp and "'none'" in csp, ( "CSP frame-ancestors 'none' not set — clickjacking risk" ) def test_security_headers_middleware_sets_x_frame_options(self): src = _MAIN_PY.read_text() assert "X-Frame-Options" in src, ( "SecurityHeadersMiddleware does not set X-Frame-Options" ) assert "DENY" in src def test_security_headers_middleware_sets_frame_ancestors(self): src = _MAIN_PY.read_text() assert "frame-ancestors" in src and "'none'" in src, ( "SecurityHeadersMiddleware CSP does not include frame-ancestors 'none'" ) # ═══════════════════════════════════════════════════════════════════════════════ # Open redirect # ═══════════════════════════════════════════════════════════════════════════════ class TestOpenRedirect: def test_elicitation_stores_path_only_in_next_param(self): src = _ELICITATION.read_text() # Must use request.url.path, NOT request.url (which is the full absolute URL) assert "request.url.path" in src, ( "ui_mcp_elicitation.py stores full absolute URL in ?next= — " "open redirect: attacker can inject external URL via crafted host header" ) # Must not store the full URL object directly assert "f\"/login?next={callback}\"" in src or "/login?next=" in src, ( "?next= redirect not found in elicitation" ) def test_elicitation_does_not_store_full_absolute_url(self): src = _ELICITATION.read_text() # Verify the callback variable is not assigned from bare request.url # (without .path or .components) bad_pattern = re.compile(r"callback\s*=\s*request\.url\b(?!\.path|\.query|\.components)") assert not bad_pattern.search(src), ( "callback assigned from request.url (full URL) — should be request.url.path" ) def test_elicitation_path_only_for_both_routes(self): """Both elicitation routes must store path-only.""" src = _ELICITATION.read_text() path_assignments = src.count("request.url.path") assert path_assignments >= 2, ( f"Only {path_assignments} place(s) use request.url.path — " "both elicitation routes must use path-only redirect" ) # ═══════════════════════════════════════════════════════════════════════════════ # Handle squatting # ═══════════════════════════════════════════════════════════════════════════════ class TestHandleSquatting: def test_auth_service_normalizes_handle_to_lowercase(self): src = _AUTH_SVC.read_text() # Must call .lower() on the handle before creating the identity assert ".lower()" in src, ( "musehub_auth.py does not normalize handle to lowercase — " "Gabriel and gabriel could register as separate accounts" ) def test_lowercase_normalization_precedes_identity_creation(self): src = _AUTH_SVC.read_text() lower_pos = src.find(".lower()") identity_pos = src.find("MusehubIdentity(") assert lower_pos < identity_pos, ( "Handle lowercasing must happen before MusehubIdentity() construction" ) def test_handle_strip_applied(self): """Leading/trailing whitespace in handles must be stripped.""" src = _AUTH_SVC.read_text() assert ".strip()" in src, ( "musehub_auth.py does not strip whitespace from handle" ) def test_gabriel_and_gabriel_uppercase_normalize_to_same(self): """Functional: normalization must make Gabriel == gabriel.""" handle_upper = "Gabriel" handle_lower = handle_upper.strip().lower() assert handle_lower == "gabriel" assert handle_lower == handle_upper.strip().lower() # idempotent def test_handle_with_spaces_stripped(self): handle = " gabriel " normalized = handle.strip().lower() assert normalized == "gabriel" # ═══════════════════════════════════════════════════════════════════════════════ # MCP prompt injection # ═══════════════════════════════════════════════════════════════════════════════ class TestMcpPromptInjection: def test_dispatcher_wraps_result_in_delimiter_tags(self): src = _DISPATCHER.read_text() assert "" in src, ( "MCP dispatcher does not wrap tool results in delimiter" ) assert "" in src def test_delimiter_appears_in_success_path_not_error(self): """Delimiter should only wrap successful results, not error envelopes.""" src = _DISPATCHER.read_text() # Find the success block (isError: False) and confirm delimiter is there ok_section = src[src.find("isError"):] first_ok = ok_section.find("False") # The delimiter must appear before the first isError: False delimiter_pos = src.find("") assert delimiter_pos < src.find('"isError": False'), ( " delimiter must appear in the success response path" ) def test_prompts_instructs_model_to_treat_tool_results_as_data(self): src = _PROMPTS.read_text() assert "musehub_tool_result" in src, ( "prompts.py does not mention — " "model has no instruction to treat tool results as untrusted data" ) def test_prompts_mentions_prompt_injection_risk(self): src = _PROMPTS.read_text() assert "prompt" in src.lower() and "inject" in src.lower(), ( "prompts.py does not warn about prompt injection — agents are unprotected" ) def test_prompts_instructs_treat_as_data_not_instructions(self): src = _PROMPTS.read_text() # Must say something like "treat as data" or "not as instructions" assert "data" in src.lower() and ( "instruction" in src.lower() or "directive" in src.lower() ), ( "prompts.py does not instruct the model to treat tool results as data, " "not instructions" ) def test_prompts_names_user_controlled_fields(self): """Prompt must call out which fields are user-controlled (not vague).""" src = _PROMPTS.read_text() user_fields = ["commit message", "issue", "file path", "repository name", "branch name"] found = [f for f in user_fields if f in src.lower()] assert len(found) >= 3, ( f"prompts.py names only {found} as user-controlled — should name commit messages, " "issue bodies, file paths, repo names, and branch names" ) @pytest.mark.anyio async def test_healthz_tool_result_not_wrapped(self, client: AsyncClient): """/healthz returns plain JSON — not an MCP tool call, no wrapping needed.""" resp = await client.get("/healthz") text = resp.text assert "" not in text, ( "/healthz response should be plain JSON, not wrapped in MCP delimiters" ) # ═══════════════════════════════════════════════════════════════════════════════ # Agent impersonation # ═══════════════════════════════════════════════════════════════════════════════ class TestAgentImpersonation: def test_trusted_agent_ids_setting_exists(self): src = _CONFIG.read_text() assert "trusted_agent_ids" in src, ( "Settings does not have a trusted_agent_ids field" ) def test_wire_push_checks_agent_ids(self): src = _WIRE_SVC.read_text() assert "trusted_agent_ids" in src, ( "wire_push does not check agent_id against trusted_agent_ids" ) def test_wire_push_flags_not_rejects(self): """Unknown agents must be flagged, never rejected.""" src = _WIRE_SVC.read_text() agent_section = src[src.find("trusted_agent_ids"):] # Rejection would look like return WirePushResponse(ok=False, ...) # after the trusted check — there must be no such rejection assert "ok=False" not in agent_section[:500], ( "wire_push rejects unknown agents — must only flag them" ) assert "untrusted_agent" in agent_section[:600], ( "wire_push does not flag unknown agents with untrusted_agent metadata" ) def test_unknown_agent_flagged_in_metadata(self): """untrusted_agent flag must be injected into commit metadata.""" src = _WIRE_SVC.read_text() assert "\"untrusted_agent\"" in src or "'untrusted_agent'" in src, ( "wire_push does not inject untrusted_agent flag into commit metadata" ) @pytest.mark.anyio async def test_unknown_agent_flagged_when_registry_set(self): from musehub.services.musehub_wire import wire_push from musehub.models.wire import WirePushRequest, WireBundle, WireCommit commit = WireCommit( commit_id="abc999", message="agent push", agent_id="evil-agent/1.0", ) req = WirePushRequest(bundle=WireBundle(commits=[commit]), branch="main") mock_session = AsyncMock() mock_repo = MagicMock() mock_repo.deleted_at = None mock_repo.owner = "gabriel" mock_session.get.return_value = mock_repo logged: list[str] = [] with patch("musehub.services.musehub_wire.settings") as mock_settings, \ patch("musehub.services.musehub_wire.logger") as mock_logger: mock_settings.require_signed_commits = False mock_settings.trusted_agent_ids = ["agentception-worker", "claude-opus-4-6"] mock_settings.per_repo_quota_bytes = 0 mock_logger.warning.side_effect = lambda msg, *args, **kw: logged.append( msg % args if args else msg ) # We only need the flagging to run — the rest of push will fail on DB mocks try: await wire_push(mock_session, "repo-123", req, pusher_id="gabriel") except Exception: pass # DB mock failures are expected; we only care about the warning assert any("untrusted" in m or "unknown agent" in m for m in logged), ( "Unknown agent_id was not flagged with a warning" ) @pytest.mark.anyio async def test_known_agent_not_flagged(self): from musehub.services.musehub_wire import wire_push from musehub.models.wire import WirePushRequest, WireBundle, WireCommit commit = WireCommit( commit_id="def000", message="known agent push", agent_id="agentception-worker-42", ) req = WirePushRequest(bundle=WireBundle(commits=[commit]), branch="main") mock_session = AsyncMock() mock_repo = MagicMock() mock_repo.deleted_at = None mock_repo.owner = "gabriel" mock_session.get.return_value = mock_repo logged: list[str] = [] with patch("musehub.services.musehub_wire.settings") as mock_settings, \ patch("musehub.services.musehub_wire.logger") as mock_logger: mock_settings.require_signed_commits = False mock_settings.trusted_agent_ids = ["agentception-worker"] mock_settings.per_repo_quota_bytes = 0 mock_logger.warning.side_effect = lambda msg, *args, **kw: logged.append( msg % args if args else msg ) try: await wire_push(mock_session, "repo-123", req, pusher_id="gabriel") except Exception: pass assert not any("untrusted" in m for m in logged), ( "Known agent was incorrectly flagged as untrusted" )