"""Section 8 — Security Hardening (Adversarial) tests.

Covers:
  SSRF            : outbound URLs validated; RFC-1918 / loopback / link-local
                    blocked; non-HTTPS schemes rejected; DNS resolution guarded.
  Mass assignment : all request bodies use Pydantic models; no **kwargs from
                    raw request dicts passed to ORM constructors.
  Timing attacks  : hmac.compare_digest used for all secret comparisons.
  Object enumeration : repo / issue primary keys are content-addressed IDs, not sequential ints.
  Commit forgery  : REQUIRE_SIGNED_COMMITS enforcement gate in wire_push.
  Regex DoS       : no user-supplied pattern compiled with re.compile().
  Tar bomb        : no archive extraction; N/A for current feature set.
"""
from __future__ import annotations

import ast
import hashlib
import re
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

_ROOT = Path(__file__).resolve().parents[1]
_MUSEHUB_PKG   = _ROOT / "musehub"
_SSRF_MODULE   = _MUSEHUB_PKG / "security" / "ssrf.py"
_WEBHOOK_MODEL = _MUSEHUB_PKG / "models" / "musehub.py"
_DISPATCHER    = _MUSEHUB_PKG / "services" / "musehub_webhook_dispatcher.py"
_WIRE_SVC      = _MUSEHUB_PKG / "services" / "musehub_wire.py"
_CONFIG        = _MUSEHUB_PKG / "config.py"


# ═══════════════════════════════════════════════════════════════════════════════
# SSRF — check_url_safe (sync, no DNS)
# ═══════════════════════════════════════════════════════════════════════════════

class TestSsrfCheckUrlSafe:
    def _check(self, url: str) -> str:
        from musehub.security.ssrf import check_url_safe
        return check_url_safe(url)

    def test_https_public_url_allowed(self) -> None:
        result = self._check("https://hooks.example.com/deliver")
        assert result == "https://hooks.example.com/deliver"

    def test_http_scheme_blocked(self) -> None:
        with pytest.raises(ValueError, match="https://"):
            self._check("http://hooks.example.com/deliver")

    def test_ftp_scheme_blocked(self) -> None:
        with pytest.raises(ValueError, match="https://"):
            self._check("ftp://hooks.example.com/deliver")

    def test_file_scheme_blocked(self) -> None:
        with pytest.raises(ValueError, match="https://"):
            self._check("file:///etc/passwd")

    def test_loopback_127_blocked(self) -> None:
        with pytest.raises(ValueError, match="private/reserved"):
            self._check("https://127.0.0.1/internal")

    def test_loopback_127_other_blocked(self) -> None:
        with pytest.raises(ValueError, match="private/reserved"):
            self._check("https://127.255.255.255/internal")

    def test_rfc1918_10_blocked(self) -> None:
        with pytest.raises(ValueError, match="private/reserved"):
            self._check("https://10.0.0.1/")

    def test_rfc1918_172_16_blocked(self) -> None:
        with pytest.raises(ValueError, match="private/reserved"):
            self._check("https://172.16.0.1/")

    def test_rfc1918_192_168_blocked(self) -> None:
        with pytest.raises(ValueError, match="private/reserved"):
            self._check("https://192.168.1.1/")

    def test_link_local_169_254_blocked(self) -> None:
        """169.254.169.254 is the AWS instance metadata endpoint."""
        with pytest.raises(ValueError, match="private/reserved"):
            self._check("https://169.254.169.254/latest/meta-data/")

    def test_hostname_with_path_allowed(self) -> None:
        result = self._check("https://api.stripe.com/v1/webhooks")
        assert "stripe.com" in result

    def test_no_hostname_blocked(self) -> None:
        with pytest.raises(ValueError):
            self._check("https:///path")

    def test_empty_string_blocked(self) -> None:
        with pytest.raises(ValueError):
            self._check("")


# ═══════════════════════════════════════════════════════════════════════════════
# SSRF — validate_outbound_url (async, with DNS)
# ═══════════════════════════════════════════════════════════════════════════════

class TestSsrfValidateOutboundUrl:
    async def test_public_hostname_allowed(self) -> None:
        from musehub.security.ssrf import validate_outbound_url
        import socket
        # Patch DNS to return a public IP.
        with patch("musehub.security.ssrf.socket.getaddrinfo",
                   return_value=[(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("93.184.216.34", 443))]):
            result = await validate_outbound_url("https://example.com/webhook")
        assert result == "https://example.com/webhook"

    async def test_dns_resolves_to_private_blocked(self) -> None:
        from musehub.security.ssrf import validate_outbound_url
        import socket
        # DNS rebinding: hostname looks public but resolves to internal IP.
        with patch("musehub.security.ssrf.socket.getaddrinfo",
                   return_value=[(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("192.168.1.100", 443))]):
            with pytest.raises(ValueError, match="private/reserved"):
                await validate_outbound_url("https://sneaky.example.com/webhook")

    async def test_dns_resolves_to_loopback_blocked(self) -> None:
        from musehub.security.ssrf import validate_outbound_url
        import socket
        with patch("musehub.security.ssrf.socket.getaddrinfo",
                   return_value=[(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("127.0.0.1", 443))]):
            with pytest.raises(ValueError, match="private/reserved"):
                await validate_outbound_url("https://localhost.example.com/")

    async def test_unresolvable_hostname_blocked(self) -> None:
        from musehub.security.ssrf import validate_outbound_url
        import socket
        with patch("musehub.security.ssrf.socket.getaddrinfo",
                   side_effect=socket.gaierror("Name or service not known")):
            with pytest.raises(ValueError, match="cannot be resolved"):
                await validate_outbound_url("https://nonexistent.invalid/")

    async def test_bare_ip_does_not_do_dns(self) -> None:
        """Bare IP literals must not trigger DNS resolution."""
        from musehub.security.ssrf import validate_outbound_url
        # If DNS is called for a bare IP, the test would hang or fail.
        with patch("musehub.security.ssrf.socket.getaddrinfo") as mock_dns:
            with pytest.raises(ValueError, match="private/reserved"):
                await validate_outbound_url("https://10.0.0.1/")
            mock_dns.assert_not_called()


# ═══════════════════════════════════════════════════════════════════════════════
# SSRF — WebhookCreate validator
# ═══════════════════════════════════════════════════════════════════════════════

class TestWebhookCreateSsrf:
    def _create(self, url: str) -> None:
        from musehub.models.musehub import WebhookCreate
        return WebhookCreate(url=url, events=["push"])

    def test_https_public_accepted(self) -> None:
        wh = self._create("https://hooks.example.com/musehub")
        assert wh.url == "https://hooks.example.com/musehub"

    def test_http_rejected_at_model_parse(self) -> None:
        from pydantic import ValidationError
        with pytest.raises(ValidationError, match="https://"):
            self._create("http://hooks.example.com/musehub")

    def test_rfc1918_rejected_at_model_parse(self) -> None:
        from pydantic import ValidationError
        with pytest.raises(ValidationError, match="private/reserved"):
            self._create("https://192.168.0.1/hook")

    def test_loopback_rejected_at_model_parse(self) -> None:
        from pydantic import ValidationError
        with pytest.raises(ValidationError, match="private/reserved"):
            self._create("https://127.0.0.1/hook")

    def test_aws_metadata_endpoint_rejected(self) -> None:
        from pydantic import ValidationError
        with pytest.raises(ValidationError, match="private/reserved"):
            self._create("https://169.254.169.254/latest/meta-data/iam/")


# ═══════════════════════════════════════════════════════════════════════════════
# SSRF — defense-in-depth in _attempt_delivery
# ═══════════════════════════════════════════════════════════════════════════════

class TestWebhookDeliveryDefenseInDepth:
    async def test_delivery_blocked_when_ssrf_check_fails(self) -> None:
        """_attempt_delivery returns failure when the URL fails SSRF validation."""
        from musehub.services.musehub_webhook_dispatcher import _attempt_delivery
        from musehub.db.musehub_models import MusehubWebhook

        webhook = MagicMock(spec=MusehubWebhook)
        webhook.url = "https://192.168.1.1/hook"  # private IP
        webhook.secret = ""

        client = AsyncMock()
        success, status, msg = await _attempt_delivery(
            client,
            webhook=webhook,
            event_type="push",
            payload_bytes=b"{}",
            delivery_id="test-id",
            attempt=1,
        )

        assert success is False
        assert "SSRF" in msg or "blocked" in msg.lower()
        client.post.assert_not_called()

    def test_dispatcher_imports_ssrf_module(self) -> None:
        src = _DISPATCHER.read_text()
        assert "validate_outbound_url" in src or "ssrf" in src.lower(), (
            "musehub_webhook_dispatcher.py must use validate_outbound_url for defence-in-depth"
        )


# ═══════════════════════════════════════════════════════════════════════════════
# Mass assignment
# ═══════════════════════════════════════════════════════════════════════════════

class TestMassAssignment:
    def _route_files(self) -> list[Path]:
        return list((_MUSEHUB_PKG / "api" / "routes").rglob("*.py"))

    def test_no_kwargs_splat_from_request_body(self) -> None:
        """No route handler should unpack request.body or raw dict into ORM."""
        dangerous_patterns = [
            re.compile(r"\*\*request\.body"),
            re.compile(r"\*\*await request\.json"),
            re.compile(r"dict\(request\)"),
        ]
        violations: list[str] = []
        for path in self._route_files():
            src = path.read_text()
            for pat in dangerous_patterns:
                if pat.search(src):
                    violations.append(f"{path.name}: {pat.pattern}")
        assert not violations, f"Potential mass assignment found: {violations}"

    def test_route_handler_body_params_are_pydantic_or_primitive(self) -> None:
        """Spot-check that handler params with complex types are Pydantic models."""
        # The most dangerous pattern is `body: dict = Body(...)` — raw dict
        # passed to ORM. We check that no handler uses `dict` as a body type.
        body_dict_re = re.compile(r":\s*dict\s*=\s*(?:Body|Depends)\(")
        violations: list[str] = []
        for path in self._route_files():
            src = path.read_text()
            if body_dict_re.search(src):
                violations.append(path.name)
        assert not violations, (
            f"Raw dict Body() parameter found (mass assignment risk): {violations}"
        )

    def test_webhook_create_uses_pydantic_not_raw_dict(self) -> None:
        src = (_MUSEHUB_PKG / "api" / "routes" / "musehub" / "webhooks.py").read_text()
        assert "WebhookCreate" in src, (
            "Webhook creation route does not use WebhookCreate Pydantic model"
        )


# ═══════════════════════════════════════════════════════════════════════════════
# Timing attacks
# ═══════════════════════════════════════════════════════════════════════════════

class TestTimingAttacks:
    def test_request_signing_uses_cryptographic_verify(self) -> None:
        """MSign uses Ed25519 verify_signature — cryptographically constant-time."""
        src = (_MUSEHUB_PKG / "auth" / "request_signing.py").read_text()
        assert "verify_signature" in src, (
            "request_signing.py must call verify_signature for MSign auth"
        )
        # Must NOT use == to compare signature bytes (timing leak)
        assert "sig_bytes ==" not in src and "== sig_bytes" not in src, (
            "request_signing.py uses == for signature bytes — timing attack"
        )

    def test_crypto_keys_uses_compare_digest(self) -> None:
        src = (_MUSEHUB_PKG / "crypto" / "keys.py").read_text()
        assert "compare_digest" in src, (
            "crypto/keys.py must use hmac.compare_digest"
        )

    def test_compare_digest_used_not_equal_for_secrets(self) -> None:
        """Grep all auth/crypto files for == comparisons against secret strings."""
        secret_eq_re = re.compile(
            r"(secret|token|password|sig|signature|key)\s*==\s*[a-zA-Z_]",
            re.IGNORECASE,
        )
        checked_files = [
            _MUSEHUB_PKG / "auth" / "request_signing.py",
            _MUSEHUB_PKG / "crypto" / "keys.py",
            _MUSEHUB_PKG / "services" / "musehub_auth.py",
        ]
        violations: list[str] = []
        for path in checked_files:
            for lineno, line in enumerate(path.read_text().splitlines(), 1):
                stripped = line.strip()
                if stripped.startswith("#"):
                    continue
                if secret_eq_re.search(stripped):
                    violations.append(f"{path.name}:{lineno}: {stripped[:80]}")
        assert not violations, (
            f"Potential timing-unsafe comparison found:\n{'\n'.join(violations)}"
        )


# ═══════════════════════════════════════════════════════════════════════════════
# Object enumeration
# ═══════════════════════════════════════════════════════════════════════════════

class TestObjectEnumeration:
    def _db_models_src(self) -> str:
        return (_MUSEHUB_PKG / "db" / "musehub_models.py").read_text()

    def test_repo_primary_key_is_string(self) -> None:
        src = self._db_models_src()
        # repo_id must be String default=content-addressed ID — not Integer
        m = re.search(r"repo_id.*primary_key=True", src)
        assert m, "repo_id primary key not found in musehub_models.py"
        # Check the line includes String, not Integer
        line = [l for l in src.splitlines() if "repo_id" in l and "primary_key=True" in l]
        assert line, "repo_id primary key line not found"
        assert "String" in line[0], (
            f"repo_id primary key is not a String: {line[0]}"
        )
        assert "Integer" not in line[0], (
            "repo_id primary key is Integer — sequential IDs enable enumeration"
        )

    def test_issue_primary_key_is_string(self) -> None:
        src = self._db_models_src()
        lines = [l for l in src.splitlines() if "issue_id" in l and "primary_key=True" in l]
        assert lines, "issue_id primary key line not found"
        assert "String" in lines[0], (
            f"issue_id primary key is not a String: {lines[0]}"
        )

    def test_issue_number_is_per_repo_not_global(self) -> None:
        """Per-repo sequential numbers (like GitHub) are acceptable;
        a global monotone counter would enable cross-repo enumeration."""
        src = self._db_models_src()
        # Verify issue number is defined in the MusehubIssue table context
        # (per-repo scoped by unique_constraint on (repo_id, number)).
        assert "number" in src and "repo_id" in src, (
            "Could not find issue number + repo_id in musehub_models.py"
        )

    def test_no_autoincrement_primary_keys_on_main_entities(self) -> None:
        """Key entity tables must not use INTEGER AUTOINCREMENT as PK."""
        src = self._db_models_src()
        # Collect lines that set primary_key=True
        pk_lines = [
            (i + 1, l.strip())
            for i, l in enumerate(src.splitlines())
            if "primary_key=True" in l
        ]
        for lineno, line in pk_lines:
            # If it says Integer AND primary_key=True AND not in a composite key,
            # it is a sequential auto-increment PK — flag it.
            if "Integer" in line and "primary_key=True" in line:
                # Allow composite keys (they never appear alone as the entity ID)
                # by checking the field name; reject generic 'id' or '<entity>_id'.
                if re.search(r"\b(id|_id)\b", line):
                    pytest.fail(
                        f"musehub_models.py:{lineno}: Integer primary key found — "
                        f"use String IDs to prevent enumeration:\n  {line}"
                    )


# ═══════════════════════════════════════════════════════════════════════════════
# Commit ID forgery / server-side signature enforcement
# ═══════════════════════════════════════════════════════════════════════════════

class TestCommitIdForgery:
    def test_require_signed_commits_setting_exists(self) -> None:
        src = _CONFIG.read_text()
        assert "require_signed_commits" in src, (
            "Settings does not have a require_signed_commits field"
        )

    def test_wire_push_enforces_when_setting_true(self) -> None:
        """wire_push must reject unsigned commits when require_signed_commits=True."""
        src = _WIRE_SVC.read_text()
        assert "require_signed_commits" in src, (
            "musehub_wire.py does not enforce require_signed_commits"
        )
        assert "unsigned" in src.lower() or "signature" in src.lower(), (
            "musehub_wire.py does not reference signature enforcement"
        )

    def test_wire_push_stream_enforces_signed_commits(self) -> None:
        """wire_push_stream must reject unsigned commits when require_signed_commits=True."""
        src = _WIRE_SVC.read_text()
        wire_stream_src = src[src.find("async def wire_push_stream"):]
        assert "require_signed_commits" in wire_stream_src, (
            "wire_push_stream does not enforce require_signed_commits"
        )
        assert "unsigned" in wire_stream_src.lower() or "signature" in wire_stream_src.lower(), (
            "wire_push_stream does not reference signature enforcement"
        )

    def test_wire_push_source_has_unsigned_warning_not_rejection(self) -> None:
        """When enforcement is off, wire_push logs a debug warning but does not reject."""
        src = _WIRE_SVC.read_text()
        # With enforcement off: a debug/warning log, not a WirePushResponse(ok=False).
        # The rejection path is inside 'if settings.require_signed_commits:'.
        # Outside that block there must be no unconditional rejection for unsigned.
        # Verify the soft-path uses logger.debug, not an early return.
        enforcement_block = src[src.find("require_signed_commits"):]
        else_block_start = enforcement_block.find("else:")
        assert else_block_start != -1, "No else branch for require_signed_commits"
        else_block = enforcement_block[else_block_start:else_block_start + 400]
        assert "logger.debug" in else_block or "logger.warning" in else_block, (
            "Unsigned commit soft-path must log at debug/warning, not silently pass"
        )
        # The else branch must not contain a WirePushResponse(ok=False)
        assert 'ok=False' not in else_block, (
            "Unsigned commits are rejected even when require_signed_commits=False"
        )


# ═══════════════════════════════════════════════════════════════════════════════
# Regex DoS
# ═══════════════════════════════════════════════════════════════════════════════

class TestRegexDos:
    def _search_service_src(self) -> str:
        return (_MUSEHUB_PKG / "services" / "musehub_search.py").read_text()

    def test_search_by_pattern_uses_python_in_not_regex(self) -> None:
        """search_by_pattern must use Python `in` operator, not re.compile(user_input)."""
        src = self._search_service_src()
        # find the search_by_pattern function body
        fn_start = src.find("def search_by_pattern")
        fn_end = src.find("\nasync def ", fn_start + 1)
        if fn_end == -1:
            fn_end = len(src)
        fn_body = src[fn_start:fn_end]
        # Must not compile user pattern as regex
        assert "re.compile" not in fn_body, (
            "search_by_pattern uses re.compile on user input — ReDoS risk"
        )
        # Should use substring `in` operator
        assert " in " in fn_body, (
            "search_by_pattern does not use substring 'in' operator"
        )

    def test_search_by_ask_uses_fixed_tokenizer_not_user_regex(self) -> None:
        """search_by_ask must tokenize with a fixed pattern, not user-supplied."""
        src = self._search_service_src()
        fn_start = src.find("def search_by_ask")
        fn_end = src.find("\nasync def ", fn_start + 1)
        if fn_end == -1:
            fn_end = len(src)
        fn_body = src[fn_start:fn_end]
        # Any re.compile inside must be a literal fixed pattern (not a variable)
        compile_calls = re.findall(r"re\.compile\((.+?)\)", fn_body)
        for call in compile_calls:
            assert not re.match(r"^[a-zA-Z_]\w*$", call.strip()), (
                f"search_by_ask compiles a variable as regex: re.compile({call})"
            )

    def test_no_user_supplied_regex_in_search_routes(self) -> None:
        """Search API routes must not pass user query string to re.compile()."""
        src = (_MUSEHUB_PKG / "api" / "routes" / "musehub" / "search.py").read_text()
        # re.compile should not appear in the route handler file
        assert "re.compile" not in src, (
            "search.py compiles user input as regex — ReDoS risk"
        )

    def test_global_search_uses_sql_or_python_in(self) -> None:
        """Global search must use parameterized SQL or Python `in`, not Python regex."""
        src = self._search_service_src()
        # Verify _TOKEN_RE is a fixed, pre-compiled pattern
        token_re_line = [l for l in src.splitlines() if "_TOKEN_RE" in l and "compile" in l]
        assert token_re_line, "_TOKEN_RE pre-compiled pattern not found in musehub_search.py"
        # It should be a literal string, not a variable
        assert "[a-z" in token_re_line[0] or "[A-Z" in token_re_line[0], (
            "_TOKEN_RE should be a literal character-class pattern"
        )


# ═══════════════════════════════════════════════════════════════════════════════
# Tar bomb / zip bomb (N/A)
# ═══════════════════════════════════════════════════════════════════════════════

class TestTarBomb:
    def test_no_archive_extraction_in_codebase(self) -> None:
        """MuseHub does not extract archives — this check ensures it stays that way."""
        dangerous = ["tarfile.open", "tarfile.extractall", "zipfile.ZipFile", ".extractall("]
        violations: list[str] = []
        for py_file in _MUSEHUB_PKG.rglob("*.py"):
            if "test_" in py_file.name:
                continue
            src = py_file.read_text()
            for pattern in dangerous:
                if pattern in src:
                    violations.append(f"{py_file.relative_to(_ROOT)}: {pattern}")
        assert not violations, (
            f"Archive extraction found — add size/count limits before extractall():\n{'\n'.join(violations)}"
        )




from httpx import AsyncClient

_ROOT         = Path(__file__).resolve().parents[1]
_MUSEHUB_PKG  = _ROOT / "musehub"
_MAIN_PY      = _MUSEHUB_PKG / "main.py"
_AUTH_SVC     = _MUSEHUB_PKG / "services" / "musehub_auth.py"
_MCP_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) -> None:
        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) -> None:
        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) -> None:
        mp3_sync = bytes([0xFF, 0xFB]) + b"\x90\x00" * 10
        assert self._check("song.mp3", mp3_sync) == "MP3"

    def test_valid_webp(self) -> None:
        webp = b"RIFF\x00\x00\x00\x00WEBP" + b"\x00" * 10
        assert self._check("cover.webp", webp) == "WebP"

    def test_valid_png(self) -> None:
        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) -> None:
        jpeg = bytes([0xFF, 0xD8, 0xFF, 0xE0]) + b"\x00" * 10
        assert self._check("cover.jpg", jpeg) == "JPEG"

    def test_unknown_extension_passes_through(self) -> None:
        # .py files are not in the known-type map — no check performed.
        assert self._check("main.py", b"import os\n") == "unknown"

    def test_script_shebang_in_py_file_allowed(self) -> None:
        # Python scripts legitimately start with #! — this is NOT a polyglot attack.
        shebang = b"#!/usr/bin/env python3\ndef main(): pass\n"
        assert self._check("tools/audit.py", shebang) == "unknown"

    def test_script_shebang_in_sh_file_allowed(self) -> None:
        shebang = b"#!/bin/bash\necho hello\n"
        assert self._check("scripts/deploy.sh", shebang) == "unknown"

    def test_shebang_still_blocked_in_binary_extensions(self) -> None:
        # A .mp3 with a shebang IS a polyglot attack.
        self._expect_error("song.mp3", b"#!/bin/bash\nrm -rf /")

    def test_shebang_still_blocked_in_midi_extension(self) -> None:
        self._expect_error("track.mid", b"#!/usr/bin/php\n<?php system($_GET['cmd']); ?>")

    def test_empty_content_returns_empty(self) -> None:
        assert self._check("track.mid", b"") == "empty"

    # ── Polyglot attacks ─────────────────────────────────────────────────────

    def test_midi_extension_but_html_content_blocked(self) -> None:
        self._expect_error("track.mid", b"<!DOCTYPE html><html><script>alert(1)</script>")

    def test_midi_extension_but_php_shebang_blocked(self) -> None:
        self._expect_error("track.mid", b"#!/usr/bin/php\n<?php system($_GET['cmd']); ?>")

    def test_jpg_extension_but_html_content_blocked(self) -> None:
        self._expect_error("cover.jpg", b"<html><body>not an image</body></html>")

    def test_webp_extension_but_wrong_magic_blocked(self) -> None:
        self._expect_error("cover.webp", b"\x00\x00\x00\x00\x00\x00\x00\x00")

    def test_png_extension_but_wrong_magic_blocked(self) -> None:
        self._expect_error("cover.png", b"JFIF\x00\x00")

    def test_jpeg_extension_but_zip_content_blocked(self) -> None:
        self._expect_error("cover.jpg", bytes([0x50, 0x4B, 0x03, 0x04]) + b"\x00" * 10)

    def test_mp3_extension_but_html_shebang_blocked(self) -> None:
        self._expect_error("song.mp3", b"#!/bin/bash\nrm -rf /")

    def test_html_extension_allows_html_content(self) -> None:
        # .html extension is exempt from the forbidden-HTML check.
        result = self._check("readme.html", b"<!DOCTYPE html><html></html>")
        assert result == "HTML"  # returns "HTML" since the exemption triggers

    # ── Wire push integration ─────────────────────────────────────────────────

    def test_wire_push_imports_magic_bytes(self) -> None:
        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) -> None:
        src = _WIRE_SVC.read_text()
        assert "PolyglotFileError" in src, (
            "wire_push does not catch PolyglotFileError"
        )


# ═══════════════════════════════════════════════════════════════════════════════
# Clickjacking
# ═══════════════════════════════════════════════════════════════════════════════

class TestClickjacking:
    async def test_x_frame_options_deny(self, client: AsyncClient) -> None:
        resp = await client.get("/healthz")
        assert resp.headers.get("x-frame-options", "").upper() == "DENY", (
            "X-Frame-Options: DENY not set — clickjacking risk"
        )

    async def test_csp_frame_ancestors_none(self, client: AsyncClient) -> None:
        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) -> None:
        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) -> None:
        src = _MAIN_PY.read_text()
        assert "frame-ancestors" in src and "'none'" in src, (
            "SecurityHeadersMiddleware CSP does not include frame-ancestors 'none'"
        )


# ═══════════════════════════════════════════════════════════════════════════════
# Handle squatting
# ═══════════════════════════════════════════════════════════════════════════════

class TestHandleSquatting:
    def test_auth_service_normalizes_handle_to_lowercase(self) -> None:
        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) -> None:
        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) -> None:
        """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) -> None:
        """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) -> None:
        handle = "  gabriel  "
        normalized = handle.strip().lower()
        assert normalized == "gabriel"


# ═══════════════════════════════════════════════════════════════════════════════
# MCP prompt injection
# ═══════════════════════════════════════════════════════════════════════════════

class TestMcpPromptInjection:
    def test_dispatcher_wraps_result_in_delimiter_tags(self) -> None:
        src = _MCP_DISPATCHER.read_text()
        assert "<musehub_tool_result>" in src, (
            "MCP dispatcher does not wrap tool results in <musehub_tool_result> delimiter"
        )
        assert "</musehub_tool_result>" in src

    def test_delimiter_appears_in_success_path_not_error(self) -> None:
        """Delimiter should only wrap successful results, not error envelopes."""
        src = _MCP_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("<musehub_tool_result>")
        assert delimiter_pos < src.find('"isError": False'), (
            "<musehub_tool_result> delimiter must appear in the success response path"
        )

    def test_prompts_instructs_model_to_treat_tool_results_as_data(self) -> None:
        src = _PROMPTS.read_text()
        assert "musehub_tool_result" in src, (
            "prompts.py does not mention <musehub_tool_result> — "
            "model has no instruction to treat tool results as untrusted data"
        )

    def test_prompts_mentions_prompt_injection_risk(self) -> None:
        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) -> None:
        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) -> None:
        """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"
        )

    async def test_healthz_tool_result_not_wrapped(self, client: AsyncClient) -> None:
        """/healthz returns plain JSON — not an MCP tool call, no wrapping needed."""
        resp = await client.get("/healthz")
        text = resp.text
        assert "<musehub_tool_result>" 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) -> None:
        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) -> None:
        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) -> None:
        """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) -> None:
        """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"
        )

    def test_wire_push_stream_flags_unknown_agents(self) -> None:
        """wire_push_stream must flag unknown agent_ids with a warning."""
        src = _WIRE_SVC.read_text()
        wire_stream_src = src[src.find("async def wire_push_stream"):]
        assert "trusted_agent_ids" in wire_stream_src, (
            "wire_push_stream does not check trusted_agent_ids"
        )
        assert "untrusted_agent" in wire_stream_src, (
            "wire_push_stream does not inject untrusted_agent flag into commit metadata"
        )

    def test_wire_push_stream_checks_trusted_prefix(self) -> None:
        """wire_push_stream uses prefix matching for trusted agent IDs."""
        src = _WIRE_SVC.read_text()
        wire_stream_src = src[src.find("async def wire_push_stream"):]
        assert "startswith" in wire_stream_src, (
            "wire_push_stream must use prefix matching (startswith) for trusted_agent_ids"
        )