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

"""
Extended tests for the SSRF validator module.

Provides comprehensive coverage of is_ip_blocked, get_safe_url, validate_url,
and module constants (ALWAYS_BLOCKED_METADATA_IPS, ALLOWED_SCHEMES).
"""

from unittest.mock import patch

import pytest

from local_deep_research.security.ssrf_validator import (
    ALWAYS_BLOCKED_METADATA_IPS,
    ALLOWED_SCHEMES,
    get_safe_url,
    is_ip_blocked,
    validate_url,
)


# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------


class TestAlwaysBlockedMetadataIPsConstant:
    """Verify the ALWAYS_BLOCKED_METADATA_IPS frozenset."""

    def test_contains_aws_imds(self):
        """AWS IMDS / Azure / OCI / DigitalOcean shared endpoint."""
        assert "169.254.169.254" in ALWAYS_BLOCKED_METADATA_IPS

    def test_contains_aws_ecs_v3_and_v4(self):
        assert "169.254.170.2" in ALWAYS_BLOCKED_METADATA_IPS
        assert "169.254.170.23" in ALWAYS_BLOCKED_METADATA_IPS

    def test_contains_alibaba_and_tencent(self):
        assert "100.100.100.200" in ALWAYS_BLOCKED_METADATA_IPS
        assert "169.254.0.23" in ALWAYS_BLOCKED_METADATA_IPS


class TestAllowedSchemesConstant:
    """Verify the ALLOWED_SCHEMES constant."""

    def test_http_in_allowed_schemes(self):
        """http must be an allowed scheme."""
        assert "http" in ALLOWED_SCHEMES

    def test_https_in_allowed_schemes(self):
        """https must be an allowed scheme."""
        assert "https" in ALLOWED_SCHEMES

    def test_ftp_not_in_allowed_schemes(self):
        """ftp must NOT be an allowed scheme."""
        assert "ftp" not in ALLOWED_SCHEMES

    def test_file_not_in_allowed_schemes(self):
        """file must NOT be an allowed scheme."""
        assert "file" not in ALLOWED_SCHEMES

    def test_allowed_schemes_is_a_set(self):
        """ALLOWED_SCHEMES should be a set for O(1) lookups."""
        assert isinstance(ALLOWED_SCHEMES, set)

    def test_allowed_schemes_contains_only_expected_values(self):
        """ALLOWED_SCHEMES should contain exactly http and https."""
        assert ALLOWED_SCHEMES == {"http", "https"}


# ---------------------------------------------------------------------------
# is_ip_blocked -- public / safe IPs
# ---------------------------------------------------------------------------


class TestIsIpBlockedPublicIPs:
    """Public IPs must never be blocked regardless of flags."""

    @pytest.mark.parametrize(
        "ip",
        [
            "8.8.8.8",
            "8.8.4.4",
            "1.1.1.1",
            "1.0.0.1",
            "208.67.222.222",
            "142.250.185.206",
            "93.184.216.34",
            "198.51.100.1",
            "203.0.113.5",
        ],
    )
    def test_public_ipv4_not_blocked(self, ip):
        """Well-known public IPv4 addresses are not blocked."""
        assert is_ip_blocked(ip) is False

    @pytest.mark.parametrize(
        "ip",
        [
            "8.8.8.8",
            "1.1.1.1",
            "93.184.216.34",
        ],
    )
    def test_public_ipv4_not_blocked_with_allow_localhost(self, ip):
        """Public IPs remain unblocked when allow_localhost=True."""
        assert is_ip_blocked(ip, allow_localhost=True) is False

    @pytest.mark.parametrize(
        "ip",
        [
            "8.8.8.8",
            "1.1.1.1",
        ],
    )
    def test_public_ipv4_not_blocked_with_allow_private_ips(self, ip):
        """Public IPs remain unblocked when allow_private_ips=True."""
        assert is_ip_blocked(ip, allow_private_ips=True) is False

    def test_public_ipv6_not_blocked(self):
        """Public IPv6 addresses are not blocked."""
        # Google public DNS IPv6
        assert is_ip_blocked("2001:4860:4860::8888") is False

    def test_public_ipv6_not_blocked_with_flags(self):
        """Public IPv6 remains unblocked regardless of flags."""
        assert (
            is_ip_blocked("2001:4860:4860::8888", allow_localhost=True) is False
        )
        assert (
            is_ip_blocked("2001:4860:4860::8888", allow_private_ips=True)
            is False
        )


# ---------------------------------------------------------------------------
# is_ip_blocked -- localhost / loopback
# ---------------------------------------------------------------------------


class TestIsIpBlockedLoopback:
    """Loopback addresses blocked by default, allowed via flags."""

    def test_127_0_0_1_blocked_by_default(self):
        """Standard IPv4 loopback is blocked."""
        assert is_ip_blocked("127.0.0.1") is True

    def test_127_0_0_2_blocked_by_default(self):
        """Alternate IPv4 loopback address in 127/8 is blocked."""
        assert is_ip_blocked("127.0.0.2") is True

    def test_127_255_255_255_blocked_by_default(self):
        """End of IPv4 loopback range is blocked."""
        assert is_ip_blocked("127.255.255.255") is True

    def test_ipv6_loopback_blocked_by_default(self):
        """IPv6 loopback ::1 is blocked."""
        assert is_ip_blocked("::1") is True

    def test_127_0_0_1_allowed_with_allow_localhost(self):
        """allow_localhost=True unblocks 127.0.0.1."""
        assert is_ip_blocked("127.0.0.1", allow_localhost=True) is False

    def test_127_range_allowed_with_allow_localhost(self):
        """allow_localhost=True unblocks the whole 127/8 range."""
        assert is_ip_blocked("127.0.0.2", allow_localhost=True) is False
        assert is_ip_blocked("127.255.255.255", allow_localhost=True) is False

    def test_ipv6_loopback_allowed_with_allow_localhost(self):
        """allow_localhost=True unblocks ::1."""
        assert is_ip_blocked("::1", allow_localhost=True) is False

    def test_127_0_0_1_allowed_with_allow_private_ips(self):
        """allow_private_ips=True also unblocks loopback."""
        assert is_ip_blocked("127.0.0.1", allow_private_ips=True) is False

    def test_ipv6_loopback_allowed_with_allow_private_ips(self):
        """allow_private_ips=True also unblocks IPv6 loopback."""
        assert is_ip_blocked("::1", allow_private_ips=True) is False


# ---------------------------------------------------------------------------
# is_ip_blocked -- RFC1918 private ranges
# ---------------------------------------------------------------------------


class TestIsIpBlockedRFC1918:
    """RFC1918 private addresses blocked by default, allowed with allow_private_ips."""

    # -- 10.0.0.0/8 --

    def test_10_0_0_1_blocked(self):
        assert is_ip_blocked("10.0.0.1") is True

    def test_10_255_255_255_blocked(self):
        assert is_ip_blocked("10.255.255.255") is True

    def test_10_x_allowed_with_allow_private_ips(self):
        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

    def test_10_x_still_blocked_with_allow_localhost(self):
        """allow_localhost does NOT unblock RFC1918."""
        assert is_ip_blocked("10.0.0.1", allow_localhost=True) is True

    # -- 172.16.0.0/12 --

    def test_172_16_0_1_blocked(self):
        assert is_ip_blocked("172.16.0.1") is True

    def test_172_31_255_255_blocked(self):
        assert is_ip_blocked("172.31.255.255") is True

    def test_172_15_255_255_not_blocked(self):
        """172.15.x.x is outside 172.16.0.0/12 and therefore public."""
        assert is_ip_blocked("172.15.255.255") is False

    def test_172_32_0_0_not_blocked(self):
        """172.32.x.x is outside 172.16.0.0/12 and therefore public."""
        assert is_ip_blocked("172.32.0.0") is False

    def test_172_16_x_allowed_with_allow_private_ips(self):
        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_172_16_x_still_blocked_with_allow_localhost(self):
        assert is_ip_blocked("172.16.0.1", allow_localhost=True) is True

    # -- 192.168.0.0/16 --

    def test_192_168_1_1_blocked(self):
        assert is_ip_blocked("192.168.1.1") is True

    def test_192_168_0_1_blocked(self):
        assert is_ip_blocked("192.168.0.1") is True

    def test_192_168_255_255_blocked(self):
        assert is_ip_blocked("192.168.255.255") is True

    def test_192_168_x_allowed_with_allow_private_ips(self):
        assert is_ip_blocked("192.168.1.1", allow_private_ips=True) is False
        assert is_ip_blocked("192.168.255.255", allow_private_ips=True) is False

    def test_192_168_x_still_blocked_with_allow_localhost(self):
        assert is_ip_blocked("192.168.1.1", allow_localhost=True) is True


# ---------------------------------------------------------------------------
# is_ip_blocked -- AWS metadata endpoint (ALWAYS blocked)
# ---------------------------------------------------------------------------


class TestIsIpBlockedAWSMetadata:
    """169.254.169.254 must ALWAYS be blocked -- the #1 SSRF target."""

    def test_aws_metadata_blocked_default(self):
        assert is_ip_blocked("169.254.169.254") is True

    def test_aws_metadata_blocked_with_allow_localhost(self):
        assert is_ip_blocked("169.254.169.254", allow_localhost=True) is True

    def test_aws_metadata_blocked_with_allow_private_ips(self):
        """Critical: even when all private IPs are allowed, AWS metadata stays blocked."""
        assert is_ip_blocked("169.254.169.254", allow_private_ips=True) is True

    def test_aws_metadata_blocked_with_both_flags(self):
        """Both flags together must not unblock AWS metadata."""
        assert (
            is_ip_blocked(
                "169.254.169.254", allow_localhost=True, allow_private_ips=True
            )
            is True
        )


# ---------------------------------------------------------------------------
# is_ip_blocked -- CGNAT (100.64.0.0/10)
# ---------------------------------------------------------------------------


class TestIsIpBlockedCGNAT:
    """CGNAT range used by Podman/rootless containers."""

    def test_cgnat_start_blocked(self):
        assert is_ip_blocked("100.64.0.1") is True

    def test_cgnat_middle_blocked(self):
        assert is_ip_blocked("100.100.100.100") is True

    def test_cgnat_end_blocked(self):
        assert is_ip_blocked("100.127.255.255") is True

    def test_cgnat_allowed_with_allow_private_ips(self):
        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_cgnat_blocked_with_allow_localhost_only(self):
        """allow_localhost does NOT unblock CGNAT."""
        assert is_ip_blocked("100.64.0.1", allow_localhost=True) is True

    def test_just_outside_cgnat_not_blocked(self):
        """100.128.0.0 is outside the 100.64.0.0/10 range and public."""
        assert is_ip_blocked("100.128.0.0") is False


# ---------------------------------------------------------------------------
# is_ip_blocked -- link-local (169.254.0.0/16)
# ---------------------------------------------------------------------------


class TestIsIpBlockedLinkLocal:
    """Link-local addresses (169.254.x.x) blocked by default."""

    def test_link_local_blocked(self):
        assert is_ip_blocked("169.254.1.1") is True
        assert is_ip_blocked("169.254.100.100") is True

    def test_link_local_beginning_blocked(self):
        assert is_ip_blocked("169.254.0.1") is True

    def test_link_local_end_blocked(self):
        assert is_ip_blocked("169.254.255.255") is True

    def test_non_aws_link_local_allowed_with_allow_private_ips(self):
        """Non-AWS link-local addresses allowed with allow_private_ips."""
        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
        assert is_ip_blocked("169.254.0.1", allow_private_ips=True) is False

    def test_aws_metadata_still_blocked_within_link_local(self):
        """AWS metadata is a special case within link-local; always blocked."""
        assert is_ip_blocked("169.254.169.254", allow_private_ips=True) is True

    def test_link_local_blocked_with_allow_localhost_only(self):
        """allow_localhost does NOT unblock link-local."""
        assert is_ip_blocked("169.254.1.1", allow_localhost=True) is True


# ---------------------------------------------------------------------------
# is_ip_blocked -- IPv6 private ranges
# ---------------------------------------------------------------------------


class TestIsIpBlockedIPv6Private:
    """IPv6 ULA (fc00::/7) and link-local (fe80::/10) blocked by default."""

    # -- Unique Local Addresses (fc00::/7) --

    def test_fc00_blocked(self):
        assert is_ip_blocked("fc00::1") is True

    def test_fd00_blocked(self):
        assert is_ip_blocked("fd00::1") is True

    def test_fd_range_blocked(self):
        """fd12:3456:789a::1 is in fc00::/7."""
        assert is_ip_blocked("fd12:3456:789a::1") is True

    def test_fc00_allowed_with_allow_private_ips(self):
        assert is_ip_blocked("fc00::1", allow_private_ips=True) is False

    def test_fd00_allowed_with_allow_private_ips(self):
        assert is_ip_blocked("fd00::1", allow_private_ips=True) is False

    def test_fc00_blocked_with_allow_localhost_only(self):
        assert is_ip_blocked("fc00::1", allow_localhost=True) is True

    # -- Link-Local (fe80::/10) --

    def test_fe80_blocked(self):
        assert is_ip_blocked("fe80::1") is True

    def test_fe80_with_interface_id_blocked(self):
        assert is_ip_blocked("fe80::1234:5678") is True

    def test_fe80_allowed_with_allow_private_ips(self):
        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

    def test_fe80_blocked_with_allow_localhost_only(self):
        assert is_ip_blocked("fe80::1", allow_localhost=True) is True


# ---------------------------------------------------------------------------
# is_ip_blocked -- "this" network (0.0.0.0/8)
# ---------------------------------------------------------------------------


class TestIsIpBlockedZeroNetwork:
    """The 0.0.0.0/8 'this' network is blocked."""

    def test_0_0_0_0_blocked(self):
        assert is_ip_blocked("0.0.0.0") is True

    def test_0_x_x_x_blocked(self):
        assert is_ip_blocked("0.1.2.3") is True

    def test_0_255_255_255_blocked(self):
        assert is_ip_blocked("0.255.255.255") is True


# ---------------------------------------------------------------------------
# is_ip_blocked -- invalid / edge-case inputs
# ---------------------------------------------------------------------------


class TestIsIpBlockedInvalidInputs:
    """Invalid or non-IP inputs must return False (not blocked)."""

    def test_invalid_ip_string_returns_false(self):
        assert is_ip_blocked("not-an-ip") is False

    def test_empty_string_returns_false(self):
        assert is_ip_blocked("") is False

    def test_hostname_returns_false(self):
        assert is_ip_blocked("example.com") is False

    def test_url_returns_false(self):
        """A full URL is not a valid IP string."""
        assert is_ip_blocked("http://127.0.0.1") is False

    def test_ip_with_port_returns_false(self):
        """IP:port is not a valid IP address."""
        assert is_ip_blocked("127.0.0.1:8080") is False

    def test_ip_with_trailing_space_returns_false(self):
        """Whitespace around an IP makes it invalid."""
        assert is_ip_blocked(" 127.0.0.1 ") is False

    def test_ip_with_cidr_returns_false(self):
        """CIDR notation is not a single IP address."""
        assert is_ip_blocked("10.0.0.0/8") is False

    def test_negative_numbers_returns_false(self):
        assert is_ip_blocked("-1.0.0.0") is False

    def test_oversized_octet_returns_false(self):
        assert is_ip_blocked("256.256.256.256") is False


# ---------------------------------------------------------------------------
# is_ip_blocked -- combined flags
# ---------------------------------------------------------------------------


class TestIsIpBlockedCombinedFlags:
    """Test behaviour when multiple flags are provided together."""

    def test_both_flags_allows_loopback(self):
        assert (
            is_ip_blocked(
                "127.0.0.1", allow_localhost=True, allow_private_ips=True
            )
            is False
        )

    def test_both_flags_allows_private(self):
        assert (
            is_ip_blocked(
                "192.168.1.1", allow_localhost=True, allow_private_ips=True
            )
            is False
        )

    def test_both_flags_still_blocks_aws_metadata(self):
        assert (
            is_ip_blocked(
                "169.254.169.254", allow_localhost=True, allow_private_ips=True
            )
            is True
        )

    def test_both_flags_does_not_block_public(self):
        assert (
            is_ip_blocked(
                "8.8.8.8", allow_localhost=True, allow_private_ips=True
            )
            is False
        )


# ---------------------------------------------------------------------------
# is_ip_blocked -- boundary IPs
# ---------------------------------------------------------------------------


class TestIsIpBlockedBoundaries:
    """Test IPs at the exact boundaries of blocked ranges."""

    def test_first_ip_of_10_range(self):
        assert is_ip_blocked("10.0.0.0") is True

    def test_last_ip_of_10_range(self):
        assert is_ip_blocked("10.255.255.255") is True

    def test_just_before_10_range_not_blocked(self):
        """9.255.255.255 is just before 10.0.0.0/8."""
        assert is_ip_blocked("9.255.255.255") is False

    def test_just_after_10_range_not_blocked(self):
        """11.0.0.0 is just after 10.0.0.0/8."""
        assert is_ip_blocked("11.0.0.0") is False

    def test_first_ip_of_172_16_range(self):
        assert is_ip_blocked("172.16.0.0") is True

    def test_last_ip_of_172_16_range(self):
        assert is_ip_blocked("172.31.255.255") is True

    def test_ip_172_15_255_255_not_blocked(self):
        """Just below the 172.16.0.0/12 range."""
        assert is_ip_blocked("172.15.255.255") is False

    def test_ip_172_32_0_0_not_blocked(self):
        """Just above the 172.16.0.0/12 range."""
        assert is_ip_blocked("172.32.0.0") is False

    def test_first_ip_of_192_168_range(self):
        assert is_ip_blocked("192.168.0.0") is True

    def test_last_ip_of_192_168_range(self):
        assert is_ip_blocked("192.168.255.255") is True

    def test_ip_192_167_255_255_not_blocked(self):
        assert is_ip_blocked("192.167.255.255") is False

    def test_ip_192_169_0_0_not_blocked(self):
        assert is_ip_blocked("192.169.0.0") is False

    def test_first_ip_of_cgnat(self):
        assert is_ip_blocked("100.64.0.0") is True

    def test_last_ip_of_cgnat(self):
        assert is_ip_blocked("100.127.255.255") is True

    def test_ip_100_63_255_255_not_blocked(self):
        """Just below CGNAT range."""
        assert is_ip_blocked("100.63.255.255") is False

    def test_ip_100_128_0_0_not_blocked(self):
        """Just above CGNAT range."""
        assert is_ip_blocked("100.128.0.0") is False


# ---------------------------------------------------------------------------
# get_safe_url -- None / empty handling
# ---------------------------------------------------------------------------


class TestGetSafeUrlNoneAndEmpty:
    """get_safe_url returns default for None or empty strings."""

    def test_none_returns_default_none(self):
        assert get_safe_url(None) is None

    def test_none_returns_custom_default(self):
        assert (
            get_safe_url(None, default="https://fallback.com")
            == "https://fallback.com"
        )

    def test_empty_string_returns_default_none(self):
        assert get_safe_url("") is None

    def test_empty_string_returns_custom_default(self):
        assert (
            get_safe_url("", default="https://fallback.com")
            == "https://fallback.com"
        )


class TestGetSafeUrlPassthrough:
    """get_safe_url passes through valid URLs and blocks unsafe ones."""

    def test_non_empty_url_returned(self):
        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("93.184.216.34", 0))],
        ):
            result = get_safe_url("https://example.com")
        assert result == "https://example.com"

    def test_http_url_returned(self):
        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("93.184.216.34", 0))],
        ):
            result = get_safe_url("http://example.com")
        assert result == "http://example.com"

    def test_url_with_path_returned(self):
        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("93.184.216.34", 0))],
        ):
            result = get_safe_url("https://example.com/path?q=1")
        assert result == "https://example.com/path?q=1"

    def test_localhost_url_returns_default(self):
        """validate_url blocks localhost, so get_safe_url returns default."""
        result = get_safe_url("http://127.0.0.1:8080", default="safe")
        assert result == "safe"

    def test_default_not_used_for_valid_url(self):
        with patch(
            "socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("93.184.216.34", 0))],
        ):
            result = get_safe_url(
                "https://example.com", default="https://fallback.com"
            )
        assert result == "https://example.com"


# ---------------------------------------------------------------------------
# validate_url -- real validation (no test bypass)
# ---------------------------------------------------------------------------


class TestValidateUrlRealValidation:
    """Confirm validate_url performs real SSRF validation."""

    def test_blocks_private_ip_url(self):
        """Private IP URL is blocked."""
        assert validate_url("http://192.168.1.1") is False

    def test_blocks_localhost_url(self):
        """Localhost URL is blocked."""
        assert validate_url("http://127.0.0.1") is False

    def test_blocks_aws_metadata_url(self):
        """AWS metadata URL is blocked."""
        assert validate_url("http://169.254.169.254/latest/meta-data") is False