"""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 UUIDs, 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 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):
        result = self._check("https://hooks.example.com/deliver")
        assert result == "https://hooks.example.com/deliver"

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

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

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

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

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

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

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

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

    def test_link_local_169_254_blocked(self):
        """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):
        result = self._check("https://api.stripe.com/v1/webhooks")
        assert "stripe.com" in result

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

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


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

class TestSsrfValidateOutboundUrl:
    @pytest.mark.anyio
    async def test_public_hostname_allowed(self):
        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"

    @pytest.mark.anyio
    async def test_dns_resolves_to_private_blocked(self):
        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")

    @pytest.mark.anyio
    async def test_dns_resolves_to_loopback_blocked(self):
        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/")

    @pytest.mark.anyio
    async def test_unresolvable_hostname_blocked(self):
        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/")

    @pytest.mark.anyio
    async def test_bare_ip_does_not_do_dns(self):
        """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):
        from musehub.models.musehub import WebhookCreate
        return WebhookCreate(url=url, events=["push"])

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

    def test_http_rejected_at_model_parse(self):
        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):
        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):
        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):
        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:
    @pytest.mark.anyio
    async def test_delivery_blocked_when_ssrf_check_fails(self):
        """_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):
        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):
        """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):
        """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):
        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):
        """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):
        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):
        """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_uuid_string(self):
        src = self._db_models_src()
        # repo_id must be String(36) default=_new_uuid — 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 (UUID): {line[0]}"
        )
        assert "Integer" not in line[0], (
            "repo_id primary key is Integer — sequential IDs enable enumeration"
        )

    def test_issue_primary_key_is_uuid_string(self):
        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 UUID String: {lines[0]}"
        )

    def test_issue_number_is_per_repo_not_global(self):
        """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):
        """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 UUID strings to prevent enumeration:\n  {line}"
                    )


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

class TestCommitIdForgery:
    def test_require_signed_commits_setting_exists(self):
        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):
        """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"
        )

    @pytest.mark.anyio
    async def test_wire_push_rejects_unsigned_when_enforcement_on(self):
        from musehub.services.musehub_wire import wire_push
        from musehub.models.wire import WirePushRequest, WireBundle, WireCommit

        unsigned_commit = WireCommit(
            commit_id="abc123",
            message="test commit",
            signature="",        # no signature
            signer_key_id="",    # no signer
        )
        req = WirePushRequest(
            bundle=WireBundle(commits=[unsigned_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

        with patch("musehub.services.musehub_wire.settings") as mock_settings:
            mock_settings.require_signed_commits = True
            mock_settings.per_repo_quota_bytes = 0
            result = await wire_push(mock_session, "repo-123", req, pusher_id="gabriel")

        assert result.ok is False
        assert "unsigned" in result.message.lower() or "sign" in result.message.lower()

    def test_wire_push_source_has_unsigned_warning_not_rejection(self):
        """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):
        """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):
        """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):
        """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):
        """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):
        """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, (
            "Archive extraction found — add size/count limits before extractall():\n"
            + "\n".join(violations)
        )