learningcircuit--local-deep-research/tests/security/test_ssrf_validator.py

"""
SSRF Validator Tests

Tests for the SSRF (Server-Side Request Forgery) protection that validates URLs
before making outgoing HTTP requests.

Security model:
- By default, block all private/internal IPs (RFC1918, localhost, link-local, CGNAT)
- allow_localhost=True: Allow only loopback addresses (127.x.x.x, ::1)
- allow_private_ips=True: Allow all private/internal IPs + localhost:
  - RFC1918: 10.x.x.x, 172.16-31.x.x, 192.168.x.x
  - CGNAT: 100.64.x.x (used by Podman/rootless containers)
  - Link-local: 169.254.x.x (except cloud metadata endpoints)
  - IPv6 ULA: fc00::/7
  - IPv6 Link-local: fe80::/10
- Cloud metadata endpoints (AWS IMDS / ECS, Azure, OCI, DigitalOcean,
  AlibabaCloud, Tencent — see ALWAYS_BLOCKED_METADATA_IPS) are ALWAYS blocked

The allow_private_ips parameter is designed for trusted self-hosted services like
SearXNG or Ollama that may be running in containerized environments (Docker, Podman)
or on a different machine on the local network.
"""

import socket

import pytest
from unittest.mock import patch

from tests.test_utils import add_src_to_path

add_src_to_path()


class TestIsIpBlocked:
    """Test the is_ip_blocked function."""

    def test_localhost_blocked_by_default(self):
        """Localhost should be blocked by default."""
        from local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("127.0.0.1") is True
        assert is_ip_blocked("127.0.0.2") is True

    def test_localhost_allowed_with_allow_localhost(self):
        """Localhost should be allowed with allow_localhost=True."""
        from local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("127.0.0.1", allow_localhost=True) is False
        assert is_ip_blocked("127.0.0.2", allow_localhost=True) is False

    def test_private_ip_blocked_with_allow_localhost(self):
        """Private IPs should still be blocked with allow_localhost=True."""
        from local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("192.168.1.100", allow_localhost=True) is True
        assert is_ip_blocked("10.0.0.5", allow_localhost=True) is True
        assert is_ip_blocked("172.16.0.1", allow_localhost=True) is True

    def test_private_ip_allowed_with_allow_private_ips(self):
        """Private IPs should be allowed with allow_private_ips=True."""
        from local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        # 192.168.x.x range
        assert is_ip_blocked("192.168.1.100", allow_private_ips=True) is False
        assert is_ip_blocked("192.168.0.1", allow_private_ips=True) is False
        assert is_ip_blocked("192.168.255.255", allow_private_ips=True) is False

        # 10.x.x.x range
        assert is_ip_blocked("10.0.0.1", allow_private_ips=True) is False
        assert is_ip_blocked("10.255.255.255", allow_private_ips=True) is False

        # 172.16-31.x.x range
        assert is_ip_blocked("172.16.0.1", allow_private_ips=True) is False
        assert is_ip_blocked("172.31.255.255", allow_private_ips=True) is False

    def test_localhost_also_allowed_with_allow_private_ips(self):
        """Localhost should also be allowed with allow_private_ips=True."""
        from local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("127.0.0.1", allow_private_ips=True) is False
        assert is_ip_blocked("127.0.0.2", allow_private_ips=True) is False

    def test_aws_metadata_always_blocked(self):
        """AWS metadata endpoint should ALWAYS be blocked, even with allow_private_ips."""
        from local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        # Without any allowlist
        assert is_ip_blocked("169.254.169.254") is True

        # With allow_localhost
        assert is_ip_blocked("169.254.169.254", allow_localhost=True) is True

        # With allow_private_ips - CRITICAL: Must still be blocked!
        assert is_ip_blocked("169.254.169.254", allow_private_ips=True) is True

    def test_public_ip_not_blocked(self):
        """Public IPs should not be blocked."""
        from local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("8.8.8.8") is False
        assert is_ip_blocked("1.1.1.1") is False
        assert is_ip_blocked("142.250.185.206") is False  # google.com

    def test_link_local_blocked(self):
        """Link-local addresses should be blocked."""
        from local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("169.254.1.1") is True
        assert is_ip_blocked("169.254.100.100") is True

    def test_cgnat_blocked_by_default(self):
        """CGNAT addresses (100.64.x.x) should be blocked by default."""
        from src.local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("100.64.0.1") is True
        assert is_ip_blocked("100.100.100.100") is True
        assert is_ip_blocked("100.127.255.255") is True

    def test_cgnat_allowed_with_allow_private_ips(self):
        """CGNAT addresses (100.64.x.x) should be allowed with allow_private_ips=True."""
        from src.local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        # CGNAT range used by Podman rootless containers
        assert is_ip_blocked("100.64.0.1", allow_private_ips=True) is False
        assert is_ip_blocked("100.100.100.100", allow_private_ips=True) is False
        assert is_ip_blocked("100.127.255.255", allow_private_ips=True) is False

    def test_link_local_allowed_with_allow_private_ips(self):
        """Link-local addresses (169.254.x.x) should be allowed with allow_private_ips=True."""
        from src.local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        # Non-AWS-metadata link-local addresses should be allowed
        assert is_ip_blocked("169.254.1.1", allow_private_ips=True) is False
        assert is_ip_blocked("169.254.100.100", allow_private_ips=True) is False
        # AWS metadata endpoint MUST still be blocked
        assert is_ip_blocked("169.254.169.254", allow_private_ips=True) is True

    def test_ipv6_ula_blocked_by_default(self):
        """IPv6 Unique Local Addresses (fc00::/7) should be blocked by default."""
        from src.local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("fc00::1") is True
        assert is_ip_blocked("fd00::1") is True

    def test_ipv6_ula_allowed_with_allow_private_ips(self):
        """IPv6 ULA (fc00::/7) should be allowed with allow_private_ips=True."""
        from src.local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("fc00::1", allow_private_ips=True) is False
        assert is_ip_blocked("fd00::1", allow_private_ips=True) is False

    def test_ipv6_link_local_blocked_by_default(self):
        """IPv6 link-local addresses (fe80::/10) should be blocked by default."""
        from src.local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("fe80::1") is True
        assert is_ip_blocked("fe80::1234:5678") is True

    def test_ipv6_link_local_allowed_with_allow_private_ips(self):
        """IPv6 link-local (fe80::/10) should be allowed with allow_private_ips=True."""
        from src.local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("fe80::1", allow_private_ips=True) is False
        assert is_ip_blocked("fe80::1234:5678", allow_private_ips=True) is False


class TestValidateUrl:
    """Test the validate_url function."""

    def test_public_url_allowed(self):
        """Public URLs should be allowed."""
        from local_deep_research.security.ssrf_validator import validate_url

        with patch("socket.getaddrinfo") as mock_getaddrinfo:
            # Mock DNS resolution to return a public IP
            mock_getaddrinfo.return_value = [
                (2, 1, 6, "", ("142.250.185.206", 0))
            ]
            assert validate_url("https://google.com") is True

    def test_localhost_url_blocked_by_default(self):
        """Localhost URLs should be blocked by default."""
        from local_deep_research.security.ssrf_validator import validate_url

        assert validate_url("http://127.0.0.1:8080") is False
        assert validate_url("http://localhost:8080") is False

    def test_localhost_url_allowed_with_allow_localhost(self):
        """Localhost URLs should be allowed with allow_localhost=True."""
        from local_deep_research.security.ssrf_validator import validate_url

        with patch("socket.getaddrinfo") as mock_getaddrinfo:
            mock_getaddrinfo.return_value = [(2, 1, 6, "", ("127.0.0.1", 0))]
            assert (
                validate_url("http://localhost:8080", allow_localhost=True)
                is True
            )

        assert (
            validate_url("http://127.0.0.1:8080", allow_localhost=True) is True
        )

    def test_private_ip_url_blocked_with_allow_localhost(self):
        """Private IP URLs should still be blocked with allow_localhost=True."""
        from local_deep_research.security.ssrf_validator import validate_url

        assert (
            validate_url("http://192.168.1.100:8080", allow_localhost=True)
            is False
        )
        assert (
            validate_url("http://10.0.0.5:8080", allow_localhost=True) is False
        )

    def test_private_ip_url_allowed_with_allow_private_ips(self):
        """Private IP URLs should be allowed with allow_private_ips=True."""
        from local_deep_research.security.ssrf_validator import validate_url

        # 192.168.x.x - typical home network
        assert (
            validate_url("http://192.168.1.100:8080", allow_private_ips=True)
            is True
        )
        assert (
            validate_url("http://192.168.0.1:80", allow_private_ips=True)
            is True
        )

        # 10.x.x.x - typical corporate network
        assert (
            validate_url("http://10.0.0.5:8080", allow_private_ips=True) is True
        )
        assert (
            validate_url("http://10.10.10.10:3000", allow_private_ips=True)
            is True
        )

        # 172.16-31.x.x - Docker default network etc.
        assert (
            validate_url("http://172.16.0.1:8080", allow_private_ips=True)
            is True
        )
        assert (
            validate_url("http://172.20.0.2:5000", allow_private_ips=True)
            is True
        )

    def test_aws_metadata_url_always_blocked(self):
        """AWS metadata URL should ALWAYS be blocked."""
        from local_deep_research.security.ssrf_validator import validate_url

        aws_metadata_url = "http://169.254.169.254/latest/meta-data"

        # Without any allowlist
        assert validate_url(aws_metadata_url) is False

        # With allow_localhost
        assert validate_url(aws_metadata_url, allow_localhost=True) is False

        # With allow_private_ips - CRITICAL: Must still be blocked!
        assert validate_url(aws_metadata_url, allow_private_ips=True) is False

    def test_invalid_scheme_blocked(self):
        """Invalid schemes should be blocked."""
        from local_deep_research.security.ssrf_validator import validate_url

        assert validate_url("ftp://example.com") is False
        assert validate_url("file:///etc/passwd") is False
        assert validate_url("javascript:alert(1)") is False

    def test_hostname_resolving_to_private_ip_blocked(self):
        """Hostnames that resolve to private IPs should be blocked."""
        from local_deep_research.security.ssrf_validator import validate_url

        with patch("socket.getaddrinfo") as mock_getaddrinfo:
            # Simulate a hostname resolving to a private IP (DNS rebinding attack)
            mock_getaddrinfo.return_value = [(2, 1, 6, "", ("192.168.1.1", 0))]
            assert validate_url("http://evil.com") is False

    def test_hostname_resolving_to_private_ip_allowed_with_allow_private_ips(
        self,
    ):
        """Hostnames resolving to private IPs allowed with allow_private_ips=True."""
        from local_deep_research.security.ssrf_validator import validate_url

        with patch("socket.getaddrinfo") as mock_getaddrinfo:
            mock_getaddrinfo.return_value = [
                (2, 1, 6, "", ("192.168.1.100", 0))
            ]
            assert (
                validate_url("http://my-searxng.local", allow_private_ips=True)
                is True
            )


class TestSearXNGUseCase:
    """Test the specific SearXNG use case that motivated allow_private_ips."""

    def test_searxng_on_localhost(self):
        """SearXNG on localhost should work with allow_private_ips."""
        from local_deep_research.security.ssrf_validator import validate_url

        with patch("socket.getaddrinfo") as mock_getaddrinfo:
            mock_getaddrinfo.return_value = [(2, 1, 6, "", ("127.0.0.1", 0))]
            assert (
                validate_url("http://localhost:8080", allow_private_ips=True)
                is True
            )

        assert (
            validate_url("http://127.0.0.1:8080", allow_private_ips=True)
            is True
        )

    def test_searxng_on_lan(self):
        """SearXNG on LAN should work with allow_private_ips."""
        from local_deep_research.security.ssrf_validator import validate_url

        # Home network
        assert (
            validate_url("http://192.168.1.100:8080", allow_private_ips=True)
            is True
        )

        # NAS or server on network
        assert (
            validate_url("http://10.0.0.50:8888", allow_private_ips=True)
            is True
        )

        # Docker network
        assert (
            validate_url("http://172.17.0.2:8080", allow_private_ips=True)
            is True
        )

    def test_searxng_hostname_on_lan(self):
        """SearXNG with hostname on LAN should work with allow_private_ips."""
        from local_deep_research.security.ssrf_validator import validate_url

        with patch("socket.getaddrinfo") as mock_getaddrinfo:
            # Simulate local DNS or /etc/hosts entry
            mock_getaddrinfo.return_value = [(2, 1, 6, "", ("192.168.1.50", 0))]
            assert (
                validate_url(
                    "http://searxng.local:8080", allow_private_ips=True
                )
                is True
            )


class TestContainerNetworking:
    """Test container networking scenarios (Podman, Docker, etc.)."""

    def test_podman_host_containers_internal(self):
        """Podman's host.containers.internal (resolves to CGNAT) should work with allow_private_ips."""
        from src.local_deep_research.security.ssrf_validator import validate_url

        with patch("socket.getaddrinfo") as mock_getaddrinfo:
            # Podman rootless containers typically resolve host.containers.internal to 100.64.x.x
            mock_getaddrinfo.return_value = [(2, 1, 6, "", ("100.64.1.1", 0))]
            assert (
                validate_url(
                    "http://host.containers.internal:11434",
                    allow_private_ips=True,
                )
                is True
            )

    def test_ollama_in_podman(self):
        """Ollama running on host accessible via Podman's CGNAT should work."""
        from src.local_deep_research.security.ssrf_validator import validate_url

        with patch("socket.getaddrinfo") as mock_getaddrinfo:
            # Ollama on host via Podman CGNAT
            mock_getaddrinfo.return_value = [
                (2, 1, 6, "", ("100.100.100.100", 0))
            ]
            assert (
                validate_url(
                    "http://host.containers.internal:11434/api/generate",
                    allow_private_ips=True,
                )
                is True
            )

    def test_searxng_in_podman(self):
        """SearXNG running on host accessible via Podman's CGNAT should work."""
        from src.local_deep_research.security.ssrf_validator import validate_url

        with patch("socket.getaddrinfo") as mock_getaddrinfo:
            # SearXNG on host via Podman CGNAT
            mock_getaddrinfo.return_value = [(2, 1, 6, "", ("100.64.0.1", 0))]
            assert (
                validate_url(
                    "http://host.containers.internal:8080/search",
                    allow_private_ips=True,
                )
                is True
            )

    def test_cgnat_url_blocked_by_default(self):
        """CGNAT URLs should be blocked by default (without allow_private_ips)."""
        from src.local_deep_research.security.ssrf_validator import validate_url

        # Direct CGNAT IP
        assert validate_url("http://100.64.0.1:8080") is False

        with patch("socket.getaddrinfo") as mock_getaddrinfo:
            mock_getaddrinfo.return_value = [(2, 1, 6, "", ("100.64.1.1", 0))]
            assert (
                validate_url("http://host.containers.internal:11434") is False
            )

    def test_docker_bridge_network(self):
        """Docker bridge network IPs should work with allow_private_ips."""
        from src.local_deep_research.security.ssrf_validator import validate_url

        # Docker typically uses 172.17.x.x for bridge network
        assert (
            validate_url("http://172.17.0.2:8080", allow_private_ips=True)
            is True
        )


class TestParserDifferentialBypass:
    """
    Tests for the parser-differential SSRF bypass (GHSA-g23j-2vwm-5c25).

    Python's ``urllib.parse.urlparse`` and the ``requests``/``urllib3``
    parser disagree on URLs that contain a backslash before the userinfo
    ``@``.  ``urlparse`` treats ``\\`` as a literal char and ``@`` as the
    userinfo separator (so it extracts the post-``@`` host); ``requests``
    treats ``\\`` as a path delimiter and connects to the pre-``\\`` host.

    A pre-fix ``validate_url`` based on ``urlparse(url).hostname`` would
    pass URLs like ``http://127.0.0.1\\@1.1.1.1`` (it sees ``1.1.1.1``)
    while ``requests.get(url)`` would actually connect to ``127.0.0.1``.
    The fix combines a Layer-1 reject of RFC-illegal characters with a
    Layer-2 swap to ``urllib3.util.parse_url`` for hostname extraction.
    """

    def test_advisory_canonical_payload(self):
        """The exact PoC from the advisory must be rejected."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1:6666\\@1.1.1.1") is False

    def test_backslash_no_port(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1\\@1.1.1.1") is False

    def test_double_backslash(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1\\\\@1.1.1.1") is False

    def test_slash_then_backslash(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1/\\@1.1.1.1") is False

    def test_tab_at_seam(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1\t@1.1.1.1") is False

    def test_carriage_return_at_seam(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1\r@1.1.1.1") is False

    def test_newline_at_seam(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1\n@1.1.1.1") is False

    def test_space_at_seam(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1 @1.1.1.1") is False

    def test_ipv6_loopback_with_backslash(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[::1]\\@1.1.1.1") is False

    def test_ipv4_mapped_ipv6_with_backslash(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[::ffff:127.0.0.1]\\@1.1.1.1") is False

    def test_backslash_with_trailing_port(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1\\@1.1.1.1:80") is False

    def test_trailing_dot_loopback_with_backslash(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1.\\@1.1.1.1") is False

    def test_null_byte_in_userinfo(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1\x00@1.1.1.1") is False

    def test_idn_unicode_host_rejected(self):
        """IDN/Unicode hosts are rejected by urllib3 / ASCII guard."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        # Circled-digit homoglyphs of '127' resolve via NFKC to '127' on
        # some libcs.  urllib3 currently rejects these via
        # LocationParseError; the ASCII-printable guard backs that up.
        assert validate_url("http://①②⑦.0.0.1/") is False

    def test_octal_ip_resolves_to_loopback(self):
        """Octal IP form '0177.0.0.1' resolves to 127.0.0.1 via getaddrinfo."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("127.0.0.1", 0))],
        ):
            assert validate_url("http://0177.0.0.1/") is False

    def test_decimal_int_ip_resolves_to_loopback(self):
        """Decimal-int IP form '2130706433' resolves to 127.0.0.1."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("127.0.0.1", 0))],
        ):
            assert validate_url("http://2130706433/") is False

    def test_post_prepare_canonicalised_form(self):
        """
        Layer-2 verification: when ``requests.PreparedRequest.url``
        canonicalises ``\\`` to ``%5C``, the urllib3-based hostname
        extraction still returns ``127.0.0.1`` so the IP check fires.
        Layer 1 doesn't match ``%5C`` (it's three printable ASCII chars);
        Layer 2 is the load-bearing defence on the SafeSession.send path.
        """
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1:6666/%5C@1.1.1.1") is False

    def test_backslash_deep_in_path(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://example.com/path\\@1.1.1.1") is False

    def test_backslash_in_userinfo_password(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://user:pass\\@127.0.0.1/") is False

    def test_backslash_with_port_on_trailing_host(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1\\@evil.com:8080") is False

    def test_interior_whitespace_at_seam(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1 \\@1.1.1.1") is False

    def test_ipv6_unspecified_blocked(self):
        """``::`` is the IPv6 unspecified address — Linux routes
        connections to ``[::]:port`` to a service bound on ``[::1]:port``,
        so it must be blocked alongside ``0.0.0.0`` (the IPv4 equivalent,
        already covered via 0.0.0.0/8 in BLOCKED_IP_RANGES)."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[::]/") is False

    def test_ipv6_unspecified_zero_form_blocked(self):
        """Equivalent representation ``0::`` — must normalise to ``::``
        before the IP-range check or this bypasses the literal-string
        allow-list in notification_validator."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[0::]/") is False

    def test_ipv6_unspecified_full_form_blocked(self):
        """Equivalent representation ``0:0:0:0:0:0:0:0``."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[0:0:0:0:0:0:0:0]/") is False


class TestDnsResolvedBypass:
    """
    The validator's load-bearing path for hostnames (not IP literals) is:
    1. ``ipaddress.ip_address(hostname)`` raises ``ValueError`` (not an IP)
    2. ``socket.getaddrinfo(hostname, ...)`` resolves to one or more IPs
    3. Each resolved IP is checked against ``BLOCKED_IP_RANGES``

    These tests exercise step 3 directly by mocking ``getaddrinfo``.
    """

    def test_hostname_resolving_to_loopback_blocked(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("127.0.0.1", 0))],
        ):
            assert validate_url("http://attacker.example.com/") is False

    def test_hostname_resolving_to_rfc1918_blocked(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("10.0.0.5", 0))],
        ):
            assert validate_url("http://attacker.example.com/") is False

    def test_hostname_resolving_to_link_local_blocked(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("169.254.1.1", 0))],
        ):
            assert validate_url("http://attacker.example.com/") is False

    def test_hostname_resolving_to_aws_metadata_blocked(self):
        """Hardcoded AWS metadata block fires even with allow_private_ips."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("169.254.169.254", 0))],
        ):
            # Even with the most permissive flag, AWS metadata stays blocked.
            assert (
                validate_url(
                    "http://attacker.example.com/", allow_private_ips=True
                )
                is False
            )

    def test_multiple_resolved_ips_one_private_blocks(self):
        """
        DNS returning a public IP first then a private IP must still block.
        Round-robin / multi-A-record DNS could otherwise be a bypass.
        """
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch(
            "socket.getaddrinfo",
            return_value=[
                (2, 1, 6, "", ("93.184.216.34", 0)),  # public
                (2, 1, 6, "", ("127.0.0.1", 0)),  # private — must block
            ],
        ):
            assert validate_url("http://attacker.example.com/") is False

    def test_dns_resolution_failure_fails_closed(self):
        """``getaddrinfo`` raising ``gaierror`` must return False (not allow)."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch("socket.getaddrinfo", side_effect=socket.gaierror()):
            assert validate_url("http://nonexistent.invalid/") is False

    def test_ipv6_dns_resolution_to_loopback_blocked(self):
        """Hostname resolving to IPv6 ``::1`` must be blocked."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch(
            "socket.getaddrinfo",
            return_value=[(10, 1, 6, "", ("::1", 0, 0, 0))],
        ):
            assert validate_url("http://attacker.example.com/") is False

    def test_dns_resolves_to_ipv4_mapped_ipv6_loopback_blocked(self):
        """
        Hostname resolving to ``::ffff:127.0.0.1`` (IPv4-mapped IPv6) must
        be blocked — exercises the IPv4-mapped unwrap in is_ip_blocked.
        """
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch(
            "socket.getaddrinfo",
            return_value=[(10, 1, 6, "", ("::ffff:127.0.0.1", 0, 0, 0))],
        ):
            assert validate_url("http://attacker.example.com/") is False


class TestAlternateIpFormsBlocked:
    """
    Alternate textual representations of private IPv4/IPv6 addresses.
    On Linux ``getaddrinfo`` accepts most of these and resolves them to
    the canonical form, which the IP check then catches.
    """

    def test_octal_loopback_blocked(self):
        """``0177.0.0.1`` → ``127.0.0.1`` via getaddrinfo on Linux."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("127.0.0.1", 0))],
        ):
            assert validate_url("http://0177.0.0.1/") is False

    def test_decimal_int_loopback_blocked(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("127.0.0.1", 0))],
        ):
            assert validate_url("http://2130706433/") is False

    def test_short_ipv4_form_loopback_blocked(self):
        """``127.1`` (short form) → ``127.0.0.1`` via getaddrinfo on Linux."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("127.0.0.1", 0))],
        ):
            assert validate_url("http://127.1/") is False

    def test_ipv4_mapped_ipv6_loopback_literal_blocked(self):
        """``[::ffff:127.0.0.1]`` is an IPv4-mapped IPv6 of loopback."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[::ffff:127.0.0.1]/") is False

    def test_ipv4_mapped_ipv6_rfc1918_literal_blocked(self):
        """``[::ffff:10.0.0.1]`` — IPv4-mapped IPv6 of RFC1918."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[::ffff:10.0.0.1]/") is False

    def test_ipv4_mapped_ipv6_aws_metadata_literal_blocked(self):
        """``[::ffff:169.254.169.254]`` — AWS metadata via mapped IPv6."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[::ffff:169.254.169.254]/") is False


class TestAllowFlagMatrix:
    """
    Verify ``allow_localhost`` / ``allow_private_ips`` flag combinations
    against the new ``::/128`` blocklist entry, and confirm the AWS
    metadata hardcoded block holds under all flag combinations.
    """

    def test_loopback_default_blocked(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1/") is False

    def test_loopback_with_allow_localhost(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1/", allow_localhost=True) is True

    def test_loopback_with_allow_private_ips(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://127.0.0.1/", allow_private_ips=True) is True

    def test_ipv6_loopback_with_allow_localhost(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[::1]/", allow_localhost=True) is True

    def test_ipv6_unspecified_blocked_even_with_allow_localhost(self):
        """
        ``::`` is the unspecified address, NOT the loopback address.
        Linux happens to route it to local services, but conceptually
        ``::`` is "any address" — distinct from ``::1``.
        ``allow_localhost`` is therefore conservatively scoped to
        ``::1`` and ``127.0.0.0/8`` and does NOT permit ``::``.
        """
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[::]/", allow_localhost=True) is False

    def test_ipv6_unspecified_blocked_even_with_allow_private_ips(self):
        """Same reasoning: ``::`` is not in any allowed-range carve-out."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[::]/", allow_private_ips=True) is False

    def test_aws_metadata_blocked_under_allow_localhost(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert (
            validate_url("http://169.254.169.254/", allow_localhost=True)
            is False
        )

    def test_aws_metadata_blocked_under_allow_private_ips(self):
        """
        Codebase comments call this out as ALWAYS blocked. Locks in that
        the most permissive flag still doesn't reach AWS metadata.
        """
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert (
            validate_url("http://169.254.169.254/", allow_private_ips=True)
            is False
        )


class TestAlwaysBlockedMetadataIPs:
    """Cloud-metadata IPs blocked under every flag combination."""

    def test_metadata_ip_blocked_under_all_flags(self):
        """Every IP in the always-blocked set must be blocked under all
        allow-flag combinations."""
        from local_deep_research.security.ssrf_validator import (
            ALWAYS_BLOCKED_METADATA_IPS,
            is_ip_blocked,
        )

        for ip in sorted(ALWAYS_BLOCKED_METADATA_IPS):
            assert is_ip_blocked(ip) is True
            assert is_ip_blocked(ip, allow_localhost=True) is True
            assert is_ip_blocked(ip, allow_private_ips=True) is True

    def test_validate_url_blocks_all_metadata_ips_under_allow_private_ips(self):
        """Same coverage end-to-end through validate_url."""
        from local_deep_research.security.ssrf_validator import (
            ALWAYS_BLOCKED_METADATA_IPS,
            validate_url,
        )

        for ip in sorted(ALWAYS_BLOCKED_METADATA_IPS):
            assert (
                validate_url(f"http://{ip}/", allow_private_ips=True) is False
            )

    def test_dns_resolution_to_metadata_ip_blocked(self):
        """A hostname that resolves to a metadata IP must also be blocked
        even when allow_private_ips=True."""
        from local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("169.254.170.2", 0))],
        ):
            assert (
                validate_url("http://attacker.example/", allow_private_ips=True)
                is False
            )


class TestRedactUrlForLog:
    """The redact_url_for_log helper used at all log sites."""

    def test_strips_userinfo(self):
        from local_deep_research.security.ssrf_validator import (
            redact_url_for_log,
        )

        assert (
            redact_url_for_log("http://user:secret@example.com/path?token=x")
            == "http://example.com"
        )

    def test_strips_percent_encoded_password(self):
        from local_deep_research.security.ssrf_validator import (
            redact_url_for_log,
        )

        assert (
            redact_url_for_log("http://u:p%40ss@example.com/")
            == "http://example.com"
        )

    def test_keeps_port(self):
        from local_deep_research.security.ssrf_validator import (
            redact_url_for_log,
        )

        assert (
            redact_url_for_log("http://example.com:8080/path")
            == "http://example.com:8080"
        )

    def test_ipv6_host_keeps_brackets(self):
        from local_deep_research.security.ssrf_validator import (
            redact_url_for_log,
        )

        assert redact_url_for_log("http://[::1]:8080/") == "http://[::1]:8080"

    def test_no_scheme_uses_question_mark(self):
        """Scheme-relative URLs use '?' as the scheme sentinel."""
        from local_deep_research.security.ssrf_validator import (
            redact_url_for_log,
        )

        # urllib3 may parse '//example.com/path' with scheme=None.
        result = redact_url_for_log("//example.com/path")
        assert result.startswith("?://") or result == "<unparseable>"

    def test_unparseable_returns_sentinel(self):
        """urllib3 rejects malformed IPv6 brackets and out-of-range
        ports; helper falls back to <unparseable>."""
        from local_deep_research.security.ssrf_validator import (
            redact_url_for_log,
        )

        assert redact_url_for_log("http://[::") == "<unparseable>"
        assert redact_url_for_log("http://1.2.3.4:99999") == "<unparseable>"

    def test_validate_url_log_does_not_leak_userinfo(self, loguru_caplog):
        """End-to-end: validate_url's rejection log must not contain the
        password from the URL's userinfo. Also assert at least one log
        record was emitted, otherwise the not-in assertion is vacuously
        true and we'd have false confidence."""
        from local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        # Mock DNS to a public IP so the URL passes Layer 1+2 and reaches
        # the IP-block log site (which does log the URL).
        with (
            loguru_caplog.at_level("WARNING"),
            patch(
                "socket.getaddrinfo",
                return_value=[(2, 1, 6, "", ("127.0.0.1", 0))],
            ),
        ):
            validate_url("http://user:supersecret123@evilhost.example/")
        assert "supersecret123" not in loguru_caplog.text, (
            "Password leaked into log output"
        )
        # Anti-silent-pass: verify we actually did log (otherwise the
        # not-in assertion above is trivially true on empty text).
        assert len(loguru_caplog.records) > 0, "No log records emitted"


class TestSchemeRejection:
    """Non-http(s) schemes must be rejected outright (not just the host check)."""

    def test_file_scheme_rejected(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("file:///etc/passwd") is False

    def test_ftp_scheme_rejected(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("ftp://example.com/") is False

    def test_gopher_scheme_rejected(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("gopher://example.com/") is False

    def test_dict_scheme_rejected(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("dict://example.com:11211/stat") is False

    def test_no_scheme_rejected(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("127.0.0.1") is False

    def test_scheme_relative_url_rejected(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("//127.0.0.1/") is False

    def test_uppercase_https_scheme_accepted(self):
        """Schemes are case-insensitive per RFC 3986."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("93.184.216.34", 0))],
        ):
            assert validate_url("HTTPS://example.com/") is True


class TestNeverRaises:
    """
    Property-based-style robustness: ``validate_url`` is a security
    boundary that takes untrusted input. It must NEVER raise — only
    return ``True``/``False``. A crash here is a DoS vector.
    """

    @pytest.mark.parametrize(
        "weird_input",
        [
            "",
            " ",
            "\x00",
            "\x00" * 100,
            ":",
            "::",
            "://",
            "http",
            "http:",
            "http:/",
            "http://",
            "http:// ",
            "http://[",
            "http://[::",
            "http://]",
            "http://@",
            "http://@@@",
            "http://:@",
            "http://:80",
            "http://:0",
            "http://example.com:99999999",  # overflow port
            "http://example.com:-1",  # negative port
            "http://%00",
            "http://%2F%2F",
            "h" * 10_000,
            "http://" + "a" * 100_000,  # huge URL
            "http://" + "[" * 100,
            "http://." + ("a." * 1000) + "com",
            "http://example.com/" + "?" * 1000,
            "\udcff",  # lone surrogate (Python str-only, raises on encode)
            "http://\udcff/",
        ],
    )
    def test_pathological_input_returns_bool(self, weird_input):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        result = validate_url(weird_input)
        assert isinstance(result, bool)


class TestIPv6TransitionPrefixesBlocked:
    """IPv6 transition prefixes (6to4, NAT64, Teredo, discard) are now
    blocked. On Linux hosts with kernel sit0/NAT64 routes configured,
    these prefixes wrap private IPv4 destinations. Default Linux has no
    such routes (so this isn't exploitable in the typical deployment),
    but blocking them closes the gap for operators who do enable
    transition tunnels."""

    def test_6to4_wrapped_loopback_blocked(self):
        """``[2002:7f00:1::]`` — 6to4 wrap of 127.0.0.1."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[2002:7f00:1::]/") is False

    def test_6to4_wrapped_rfc1918_blocked(self):
        """``[2002:c0a8:101::]`` — 6to4 wrap of 192.168.1.1."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[2002:c0a8:101::]/") is False

    def test_nat64_wrapped_loopback_blocked(self):
        """``[64:ff9b::7f00:1]`` — NAT64 wrap of 127.0.0.1."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[64:ff9b::7f00:1]/") is False

    def test_teredo_prefix_blocked(self):
        """Teredo (2001::/32) tunnels IPv6-over-UDP/IPv4."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[2001::1]/") is False

    def test_ipv6_discard_prefix_blocked(self):
        """RFC 6666 discard prefix (100::/64) is reserved for sinkholes."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[100::1]/") is False

    def test_6to4_wraps_aws_metadata_blocked(self):
        """[2002:a9fe:a9fe::] — 6to4 wrap of 169.254.169.254 (AWS IMDS).
        Cloud metadata is the highest-value SSRF target; the 2002::/16
        block is what catches this case (the IMDS hardcoded literal
        check is on the IPv4 form only)."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[2002:a9fe:a9fe::]/") is False

    def test_nat64_wraps_aws_metadata_blocked(self):
        """[64:ff9b::a9fe:a9fe] — NAT64 wrap of 169.254.169.254 (AWS IMDS)."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[64:ff9b::a9fe:a9fe]/") is False

    def test_6to4_wraps_rfc1918_class_a_blocked(self):
        """[2002:0a00:1::] — 6to4 wrap of 10.0.0.1."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[2002:0a00:1::]/") is False

    def test_6to4_wraps_rfc1918_class_b_blocked(self):
        """[2002:ac10:1::] — 6to4 wrap of 172.16.0.1."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[2002:ac10:1::]/") is False

    def test_nat64_wraps_rfc1918_class_a_blocked(self):
        """[64:ff9b::a00:1] — NAT64 wrap of 10.0.0.1."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[64:ff9b::a00:1]/") is False

    def test_nat64_local_use_prefix_blocked(self):
        """RFC 8215's 64:ff9b:1::/48 (NAT64 local-use) is the same SSRF
        threat class as the well-known /96. On hosts configured to route
        the local-use prefix, [64:ff9b:1::a9fe:a9fe] reaches AWS IMDS
        identically to the WKP form. Missing this prefix earned a
        HackerOne bounty against the Ruby ssrf_filter library."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[64:ff9b:1::1]/") is False

    def test_nat64_local_use_wraps_aws_metadata_blocked(self):
        """[64:ff9b:1::a9fe:a9fe] — local-use NAT64 wrap of 169.254.169.254."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[64:ff9b:1::a9fe:a9fe]/") is False

    def test_ipv4_compatible_imds_blocked(self):
        """[::169.254.169.254] — RFC 4291 IPv4-Compatible IPv6 form
        (DEPRECATED 2006). On hosts with ::/96 routes this reaches IMDS
        identically to the IPv4-mapped and NAT64-wrapped forms. Same
        defense-in-depth class as the transition prefixes."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[::169.254.169.254]/") is False

    def test_ipv4_compatible_imds_hex_form_blocked(self):
        """Same address, hex form: [::a9fe:a9fe]."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[::a9fe:a9fe]/") is False

    def test_ipv4_compatible_rfc1918_blocked(self):
        """[::192.168.1.1] — IPv4-Compatible wrap of RFC1918."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[::192.168.1.1]/") is False


class TestIPv6TransitionPrefixesAllowFlagMatrix:
    """Lock in the design decision: ``allow_private_ips=True`` does NOT
    bypass the IPv6 transition prefixes (2002::/16, 64:ff9b::/96,
    2001::/32, 100::/64). The override carve-out only covers the local
    LOOPBACK_RANGES + PRIVATE_RANGES lists in ssrf_validator.py; the
    transition prefixes are intentionally excluded so that an attacker
    cannot reach a private IPv4 destination by tunneling through 6to4
    or NAT64 even when the operator has set ``allow_private_ips=True``
    for a self-hosted service like Ollama.

    If you ever need a self-hosted service reachable via 6to4 or
    NAT64, that's a deliberate config decision and the design here
    forces it to be made explicitly.
    """

    def test_6to4_blocked_under_allow_localhost(self):
        from src.local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("2002:7f00:1::", allow_localhost=True) is True

    def test_6to4_blocked_under_allow_private_ips(self):
        from src.local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("2002:c0a8:101::", allow_private_ips=True) is True

    def test_nat64_blocked_under_allow_localhost(self):
        from src.local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("64:ff9b::7f00:1", allow_localhost=True) is True

    def test_nat64_blocked_under_allow_private_ips(self):
        from src.local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("64:ff9b::a00:1", allow_private_ips=True) is True

    def test_teredo_blocked_under_allow_private_ips(self):
        from src.local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("2001::1", allow_private_ips=True) is True

    def test_discard_blocked_under_allow_private_ips(self):
        from src.local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        assert is_ip_blocked("100::1", allow_private_ips=True) is True

    def test_6to4_aws_metadata_blocked_under_allow_private_ips(self):
        """High-value: even with the most permissive flag, the 6to4 wrap
        of AWS IMDS must remain blocked."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert (
            validate_url("http://[2002:a9fe:a9fe::]/", allow_private_ips=True)
            is False
        )

    def test_nat64_aws_metadata_blocked_under_allow_private_ips(self):
        """Same locking-in for NAT64 wrap of IMDS."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert (
            validate_url("http://[64:ff9b::a9fe:a9fe]/", allow_private_ips=True)
            is False
        )


class TestIPv6TransitionPrefixesAntiCollision:
    """Anti-regression: legitimate IPv6 destinations adjacent to the new
    transition prefixes must still pass validation. These tests guard
    against accidental over-blocking if anyone widens a prefix later."""

    def test_google_dns_v6_passes(self):
        """2001:4860:4860::8888 — Google Public DNS. Second hextet 0x4860
        is outside the 2001::/32 Teredo block."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[2001:4860:4860::8888]/") is True

    def test_cloudflare_dns_v6_passes(self):
        """2606:4700:4700::1111 — Cloudflare Public DNS, far from any
        transition prefix."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[2606:4700:4700::1111]/") is True

    def test_root_server_v6_passes(self):
        """2001:500::/30 root-server allocation — second hextet 0x0500
        is outside Teredo."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[2001:500:88::1]/") is True

    def test_he_tunnelbroker_v6_passes(self):
        """2001:470::/32 Hurricane Electric — second hextet 0x0470."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[2001:470:1f04::1]/") is True

    def test_neighbor_above_6to4_passes(self):
        """2003::/16 sits adjacent to 2002::/16 but is not in it."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[2003::1]/") is True

    def test_discard_prefix_neighbor_passes(self):
        """100:1::/16 sits outside the 100::/64 discard prefix
        (second hextet 0x0001)."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[100:1::1]/") is True


class TestNat64EnvOptOut:
    """Operator escape hatch: ``LDR_SECURITY_ALLOW_NAT64=true`` opens the
    two NAT64 prefixes for IPv6-only deployments using DNS64+NAT64.

    Critical invariants:
    - The carve-out is ONLY for the two NAT64 prefixes (well-known and
      RFC 8215 local-use). 6to4, Teredo, discard remain blocked.
    - The carve-out does NOT reopen the IPv4-form cloud-metadata block
      (169.254.169.254 stays blocked).
    - Reading the env var lazily (per-call, not at import) means
      monkeypatching works in tests.
    """

    def test_nat64_wkp_blocked_when_env_unset(self, monkeypatch):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        monkeypatch.delenv("LDR_SECURITY_ALLOW_NAT64", raising=False)
        assert validate_url("http://[64:ff9b::a00:1]/") is False

    def test_nat64_wkp_allowed_when_env_true(self, monkeypatch):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        # 64:ff9b::8.8.8.8 — NAT64 wrap of Google DNS, the canonical
        # IPv6-only-deployment use case.
        assert validate_url("http://[64:ff9b::808:808]/") is True

    def test_nat64_local_use_allowed_when_env_true(self, monkeypatch):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert validate_url("http://[64:ff9b:1::808:808]/") is True

    def test_env_does_not_unblock_6to4(self, monkeypatch):
        """6to4 has no live legitimate use; the operator switch must
        not extend to it."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert validate_url("http://[2002:c0a8:101::]/") is False

    def test_env_does_not_unblock_teredo(self, monkeypatch):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert validate_url("http://[2001::1]/") is False

    def test_env_does_not_unblock_discard(self, monkeypatch):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert validate_url("http://[100::1]/") is False

    def test_env_does_not_unblock_imds_v4_literal(self, monkeypatch):
        """The IPv4-form metadata literal is in ALWAYS_BLOCKED_METADATA_IPS
        and is checked BEFORE the prefix loop. The NAT64 carve-out
        cannot reach it."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert validate_url("http://169.254.169.254/") is False

    def test_env_does_not_unblock_imds_via_nat64_wkp_wrap(self, monkeypatch):
        """The IMDS embedded-IPv4 check fires before the NAT64 carve-out:
        even with operator opt-in, [64:ff9b::a9fe:a9fe] (NAT64 WKP wrap
        of 169.254.169.254) stays blocked. ALWAYS_BLOCKED_METADATA_IPS
        is absolute by design."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert validate_url("http://[64:ff9b::a9fe:a9fe]/") is False

    def test_env_does_not_unblock_imds_via_nat64_local_use_wrap(
        self, monkeypatch
    ):
        """Same lock-in for the RFC 8215 local-use prefix wrap."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert validate_url("http://[64:ff9b:1::a9fe:a9fe]/") is False

    def test_env_does_not_unblock_ecs_metadata_via_nat64_wrap(
        self, monkeypatch
    ):
        """169.254.170.2 (AWS ECS task metadata v3) is also in the
        always-blocked set; NAT64 wrap stays blocked under opt-in."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        # 169.254.170.2 = 0xa9feaa02
        assert validate_url("http://[64:ff9b::a9fe:aa02]/") is False

    def test_env_falsy_values_keep_blocked(self, monkeypatch):
        """'false', '0', and unset must all keep the block in place."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        for value in ("false", "0", "no", ""):
            monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", value)
            assert validate_url("http://[64:ff9b::a00:1]/") is False, (
                f"NAT64 must remain blocked for env value {value!r}"
            )

    def test_env_true_does_not_bypass_loopback_in_block_list(self, monkeypatch):
        """Sanity: opting into NAT64 must not accidentally unblock
        non-NAT64 entries that share a prefix family."""
        from src.local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert is_ip_blocked("127.0.0.1") is True
        assert is_ip_blocked("::1") is True

    def test_env_true_does_not_unblock_ipv6_ula(self, monkeypatch):
        """The carve-out's ``continue`` lives in the same loop that walks
        ULA (fc00::/7) and link-local (fe80::/10). Pin that opting into
        NAT64 does not accidentally unblock these adjacent IPv6 ranges."""
        from src.local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert is_ip_blocked("fc00::1") is True
        assert is_ip_blocked("fd12:3456:789a::1") is True

    def test_env_true_does_not_unblock_ipv6_link_local(self, monkeypatch):
        from src.local_deep_research.security.ssrf_validator import (
            is_ip_blocked,
        )

        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert is_ip_blocked("fe80::1") is True


class TestIsNat64WrappedMetadataIp:
    """Direct unit tests for the shared helper. Both validators rely on
    its IPv4 short-circuit and on the metadata-set membership check;
    surface those contracts explicitly so a refactor of either branch
    can't silently flip them."""

    def test_returns_false_for_ipv4(self):
        """The helper must short-circuit for IPv4 inputs because it's
        called after ``is_ip_blocked`` unwraps IPv4-mapped IPv6 — at
        that point the address is no longer IPv6 and the NAT64 check
        does not apply."""
        import ipaddress
        from src.local_deep_research.security.ssrf_validator import (
            is_nat64_wrapped_metadata_ip,
        )

        assert (
            is_nat64_wrapped_metadata_ip(
                ipaddress.IPv4Address("169.254.169.254")
            )
            is False
        )

    def test_returns_false_for_non_nat64_ipv6(self):
        """Public IPv6 (Google DNS) is not in any NAT64 prefix."""
        import ipaddress
        from src.local_deep_research.security.ssrf_validator import (
            is_nat64_wrapped_metadata_ip,
        )

        assert (
            is_nat64_wrapped_metadata_ip(
                ipaddress.IPv6Address("2001:4860:4860::8888")
            )
            is False
        )

    def test_returns_false_for_nat64_wrap_of_non_metadata(self):
        """[64:ff9b::a00:1] (NAT64 wrap of 10.0.0.1) is in a NAT64
        prefix but the embedded IPv4 is not metadata — helper returns
        False so the broader carve-out logic can apply."""
        import ipaddress
        from src.local_deep_research.security.ssrf_validator import (
            is_nat64_wrapped_metadata_ip,
        )

        assert (
            is_nat64_wrapped_metadata_ip(
                ipaddress.IPv6Address("64:ff9b::a00:1")
            )
            is False
        )

    def test_returns_true_for_imds_via_wkp(self):
        import ipaddress
        from src.local_deep_research.security.ssrf_validator import (
            is_nat64_wrapped_metadata_ip,
        )

        assert (
            is_nat64_wrapped_metadata_ip(
                ipaddress.IPv6Address("64:ff9b::a9fe:a9fe")
            )
            is True
        )

    def test_returns_true_for_imds_via_local_use(self):
        import ipaddress
        from src.local_deep_research.security.ssrf_validator import (
            is_nat64_wrapped_metadata_ip,
        )

        assert (
            is_nat64_wrapped_metadata_ip(
                ipaddress.IPv6Address("64:ff9b:1::a9fe:a9fe")
            )
            is True
        )


class TestValidateUrlEdgeCases:
    """Robustness: validate_url must never raise, only return bool."""

    def test_empty_string_returns_false(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("") is False

    def test_whitespace_only_returns_false(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("   ") is False

    def test_tab_newline_only_returns_false(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("\t\n") is False

    def test_none_returns_false(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url(None) is False

    def test_int_returns_false(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url(123) is False

    def test_malformed_ipv6_no_crash(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url("http://[::") is False

    def test_extremely_long_url_no_crash(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        # Should reject (no DNS) but most importantly must not crash or
        # consume excessive memory.
        with patch(
            "socket.getaddrinfo", side_effect=__import__("socket").gaierror()
        ):
            assert validate_url("http://" + "a" * 100_000) is False


class TestLegitimateUrlsStillPass:
    """Anti-regression: ensure the fix doesn't reject RFC-legal URLs."""

    @staticmethod
    def _public_dns_mock():
        # 93.184.216.34 is the documented IP for example.com (RFC-2606).
        return patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("93.184.216.34", 0))],
        )

    def test_simple_http_url(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with self._public_dns_mock():
            assert validate_url("http://example.com/") is True

    def test_explicit_port(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with self._public_dns_mock():
            assert validate_url("http://example.com:8080/") is True

    def test_userinfo_is_rfc_legal(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with self._public_dns_mock():
            assert validate_url("http://user:pass@example.com/") is True

    def test_userinfo_with_port(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with self._public_dns_mock():
            assert validate_url("http://user:pass@example.com:8080/") is True

    def test_trailing_dot_hostname(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with self._public_dns_mock():
            assert validate_url("http://example.com./") is True

    def test_path_query_fragment(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with self._public_dns_mock():
            assert validate_url("http://example.com/path?q=1#frag") is True

    def test_plus_in_query_string(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with self._public_dns_mock():
            assert validate_url("http://example.com/?q=foo+bar") is True

    def test_encoded_backslash_in_path_is_rfc_legal(self):
        """%5C in a PATH (not a host bypass) is RFC-legal and must pass."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with self._public_dns_mock():
            assert validate_url("http://example.com/path%5Cfile") is True

    def test_encoded_space_in_path(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with self._public_dns_mock():
            assert (
                validate_url("http://example.com/path%20with%20encoded%20space")
                is True
            )

    def test_uppercase_hostname_case_folded(self):
        """Locks in case-folding parity between urlparse and urllib3."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with self._public_dns_mock():
            assert validate_url("http://EXAMPLE.COM/") is True

    def test_ipv6_public(self):
        """IPv6 hosts unwrap from brackets and check correctly."""
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        # 2001:db8::1 is the documentation prefix; not in any blocked
        # range, so this should pass.
        assert validate_url("http://[2001:db8::1]/") is True


# Security-model documentation lives in the module docstring at the top of
# this file. The previous TestDocumentation class held a single skipped
# placeholder test with no assertions; removed for clarity.


# ===========================================================================
# Edge-case coverage — gaps identified by a systematic review of the
# validator's branches against the existing 7-file test suite. Each test
# documents which production line(s) it pins and how it would fail under
# the documented mutation.
# ===========================================================================


class TestUnspecifiedIPv4Blocked:
    """validate_url end-to-end coverage for the 0.0.0.0/8 entry of
    PRIVATE_IP_RANGES (ip_ranges.py:24). Prior tests covered
    ``is_ip_blocked("0.0.0.0")`` but never the full URL path through the
    parser, IP-literal branch (ssrf_validator.py:258-268), and
    is_ip_blocked. Mutation: removing 0.0.0.0/8 from PRIVATE_IP_RANGES
    flips all three parametrize cases to True.
    """

    @pytest.mark.parametrize(
        "url",
        [
            "http://0.0.0.0/",
            "http://0.0.0.5/",
            "http://0.255.255.254/",
        ],
    )
    def test_unspecified_ipv4_literal_blocked(self, url):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url(url) is False


class TestDnsResolutionNonGaierror:
    """The DNS-resolution block in validate_url has two except handlers:
    a specific ``socket.gaierror`` (ssrf_validator.py:307-309) and a
    generic ``except Exception`` (lines 310-312). The latter is hit when
    getaddrinfo raises anything that is not a gaierror (PermissionError
    from a restricted environment, UnicodeError from idna encoding of
    oversized labels, etc.). Prior tests covered only gaierror.

    Mutation: deleting lines 310-312 lets the exception propagate out
    of validate_url. pytest reports the test as an ERROR (uncaught
    exception), which still counts as a failure for these assertions.
    """

    @pytest.mark.parametrize(
        "exc",
        [PermissionError("eperm"), OSError("eio"), RuntimeError("boom")],
    )
    def test_non_gaierror_during_dns_fails_closed(self, exc, loguru_caplog):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch("socket.getaddrinfo", side_effect=exc):
            with loguru_caplog.at_level("WARNING"):
                result = validate_url("http://example.com/")

        assert result is False
        assert "Error during hostname resolution" in loguru_caplog.text


class TestRfcForbiddenControlChars:
    """RFC_FORBIDDEN_URL_CHARS_RE (ssrf_validator.py:63) is defined as
    ``[\\\\s\\x00-\\x1f\\x7f]``. Prior tests heavily covered backslash
    and ``\\x00``; this class pins the rest of the control-byte range
    (low-range \\x01, mid-range \\x1f, and high-range \\x7f / DEL).

    Mutation: narrowing the character class (e.g. dropping \\x7f or the
    \\x01-\\x1f run) lets the corresponding URL pass through to urllib3
    parsing — which may then disagree with the eventual HTTP client on
    where the boundary between userinfo and host falls, the exact bug
    class GHSA-g23j-2vwm-5c25 closed.
    """

    @pytest.mark.parametrize(
        "ctrl_char",
        ["\x01", "\x1f", "\x7f"],
    )
    def test_control_byte_in_url_rejected(self, ctrl_char):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        url = f"http://127.0.0.1{ctrl_char}@1.1.1.1"
        assert validate_url(url) is False


class TestAlternateIpHexForm:
    """Single-DWORD hex notation (``0x7f000001``) is not parseable by
    ``ipaddress.ip_address`` so the IP-literal branch raises ValueError
    and the validator falls through to DNS resolution
    (ssrf_validator.py:269-271 → 287-289). On Linux glibc, getaddrinfo
    resolves the hex form to the canonical IPv4, which the post-DNS
    is_ip_blocked check then catches.

    This pins the load-bearing assumption that the IP-parse failure
    correctly falls through, rather than short-circuiting to a wrong
    outcome (e.g. returning True because no IP was matched).
    """

    def test_hex_dword_loopback_resolves_and_blocks(self):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("127.0.0.1", 0))],
        ):
            # 0x7f000001 == 127.0.0.1. glibc resolves this; non-glibc
            # libc implementations may reject — the test mocks DNS to
            # the canonical form so we exercise the post-DNS branch
            # regardless of platform getaddrinfo behavior.
            assert validate_url("http://0x7f000001/") is False


class TestPortEdgeCases:
    """Ports outside the legal range (>65535 and 0). 65536 trips
    urllib3's ``parse_url`` and falls into the LocationParseError
    handler (ssrf_validator.py:229-233). Port 0 parses successfully but
    the host is still 127.0.0.1, which the IP-literal branch rejects.
    """

    @pytest.mark.parametrize(
        "url",
        [
            "http://127.0.0.1:65536/",
            "http://127.0.0.1:0/",
        ],
    )
    def test_port_edge_case_blocked(self, url):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url(url) is False


class TestMultipleAtSignsContract:
    """RFC 3986 says when a URL contains multiple ``@`` characters the
    LAST one delimits userinfo from host. ``http://a@b@1.1.1.1/`` parses
    to host=``1.1.1.1``. urllib3 implements this; urllib.parse.urlparse
    has historically agreed.

    Validator behavior: when urllib3 reports a public host, validate_url
    allows the URL (and a request library subsequently connects to the
    same host). This test pins that contract. If urllib3 ever changes
    its multi-@ handling, the explicit parse-url assertion catches the
    drift.
    """

    def test_double_at_resolves_to_last_segment(self):
        from urllib3.util import parse_url

        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        url = "http://user:pass@127.0.0.1@1.1.1.1/"
        # Contract check — if this fails, urllib3 changed its multi-@
        # semantics and the validator's assumption is no longer safe.
        assert parse_url(url).host == "1.1.1.1"

        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("1.1.1.1", 0))],
        ):
            assert validate_url(url) is True


class TestUserinfoContainsIpShape:
    """An IP-shaped string inside the userinfo portion of a URL is not
    a bypass: urllib3 parses ``http://127.0.0.1@evil.com/`` with
    host=``evil.com``, and ``requests`` connects to the same host. The
    validator therefore (correctly) allows it.

    This test pins that the validator and urllib3 agree on this case,
    which is essential to the parser-differential defense documented
    at ssrf_validator.py:224-228.
    """

    def test_userinfo_ip_shape_is_not_a_bypass(self):
        from urllib3.util import parse_url

        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        url = "http://127.0.0.1@evil.com/"
        # The IP-shaped string is userinfo, not host.
        assert parse_url(url).host == "evil.com"

        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("93.184.216.34", 0))],
        ):
            assert validate_url(url) is True


class TestIpv6ZoneIdBlocked:
    """IPv6 link-local addresses with zone identifiers (RFC 6874 syntax)
    must still be blocked because the host is in fe80::/10 regardless
    of the trailing ``%zone`` part.

    Python's ``ipaddress.ip_address`` accepts ``fe80::1%eth0`` since
    3.9; the validator's IP-literal branch (ssrf_validator.py:258-268)
    therefore handles this directly via is_ip_blocked without falling
    through to DNS. The percent-encoded variant (%25eth0) is the
    URI-safe form that some clients emit; urllib3 may parse-fail it,
    in which case the LocationParseError branch also returns False.
    """

    @pytest.mark.parametrize(
        "url",
        [
            "http://[fe80::1%eth0]/",
            "http://[fe80::1%25eth0]/",
        ],
    )
    def test_ipv6_zone_id_link_local_blocked(self, url):
        from src.local_deep_research.security.ssrf_validator import (
            validate_url,
        )

        assert validate_url(url) is False