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

"""
Behavioral tests for ssrf_validator module.

Tests the SSRF (Server-Side Request Forgery) validation functions.
"""

import pytest


class TestIsIPBlockedLoopback:
    """Tests for loopback address blocking."""

    def test_127_0_0_1_blocked_by_default(self):
        """127.0.0.1 is blocked by default."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("127.0.0.1") is True

    def test_127_0_0_1_allowed_with_localhost_flag(self):
        """127.0.0.1 is 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

    def test_loopback_range_blocked(self):
        """Entire 127.x.x.x range is blocked."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("127.255.255.255") is True

    def test_ipv6_loopback_blocked(self):
        """IPv6 loopback ::1 is blocked."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("::1") is True

    def test_ipv6_loopback_allowed_with_flag(self):
        """IPv6 loopback allowed with allow_localhost=True."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("::1", allow_localhost=True) is False


class TestIsIPBlockedPrivate:
    """Tests for private IP address blocking."""

    def test_10_x_blocked(self):
        """10.x.x.x is blocked by default."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("10.0.0.1") is True

    def test_172_16_x_blocked(self):
        """172.16.x.x is blocked by default."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("172.16.0.1") is True

    def test_192_168_x_blocked(self):
        """192.168.x.x is blocked by default."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("192.168.1.1") is True

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

        assert is_ip_blocked("10.0.0.1", allow_private_ips=True) is False
        assert is_ip_blocked("172.16.0.1", allow_private_ips=True) is False
        assert is_ip_blocked("192.168.1.1", allow_private_ips=True) is False


class TestIsIPBlockedCGNAT:
    """Tests for CGNAT (100.64.0.0/10) address handling."""

    def test_cgnat_blocked_by_default(self):
        """CGNAT range is blocked by default."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("100.64.0.1") is True

    def test_cgnat_allowed_with_private_flag(self):
        """CGNAT allowed with allow_private_ips=True."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("100.64.0.1", allow_private_ips=True) is False

    def test_cgnat_end_of_range(self):
        """End of CGNAT range is blocked."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("100.127.255.255") is True


class TestIsIPBlockedLinkLocal:
    """Tests for link-local address handling."""

    def test_link_local_blocked(self):
        """Link-local 169.254.x.x is blocked."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("169.254.0.1") is True

    def test_link_local_allowed_with_private_flag(self):
        """Link-local allowed with allow_private_ips=True (except AWS)."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        # Regular link-local should be allowed
        assert is_ip_blocked("169.254.1.1", allow_private_ips=True) is False


class TestIsIPBlockedAWSMetadata:
    """Tests for AWS metadata endpoint blocking."""

    def test_aws_metadata_always_blocked(self):
        """AWS metadata endpoint 169.254.169.254 is always blocked."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        # Should be blocked even with all flags
        assert is_ip_blocked("169.254.169.254") is True
        assert is_ip_blocked("169.254.169.254", allow_localhost=True) is True
        assert is_ip_blocked("169.254.169.254", allow_private_ips=True) is True


class TestIsIPBlockedPublic:
    """Tests for public IP address handling."""

    def test_public_ip_not_blocked(self):
        """Public IPs are not 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("208.67.222.222") is False


class TestIsIPBlockedIPv6Private:
    """Tests for IPv6 private address handling."""

    def test_ipv6_fc00_blocked(self):
        """IPv6 fc00::/7 unique local is blocked."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

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

    def test_ipv6_fd00_blocked(self):
        """IPv6 fd00:: unique local is blocked."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

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

    def test_ipv6_fe80_blocked(self):
        """IPv6 fe80:: link-local is blocked."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

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

    def test_ipv6_private_allowed_with_flag(self):
        """IPv6 private allowed with allow_private_ips=True."""
        from 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("fe80::1", allow_private_ips=True) is False


class TestIsIPBlockedInvalid:
    """Tests for invalid IP address handling."""

    def test_invalid_ip_not_blocked(self):
        """Invalid IP returns False (not blocked)."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("not-an-ip") is False

    def test_empty_string_not_blocked(self):
        """Empty string returns False."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("") is False

    def test_hostname_not_blocked(self):
        """Hostname is not blocked (not an IP)."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("example.com") is False


class TestIsIPBlockedZeroNetwork:
    """Tests for 0.0.0.0/8 network handling."""

    def test_0_0_0_0_blocked(self):
        """0.0.0.0 is blocked."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("0.0.0.0") is True

    def test_0_x_x_x_blocked(self):
        """0.x.x.x network is blocked."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("0.1.2.3") is True


class TestPrivateIpRangesBehavior:
    """Verify is_ip_blocked rejects an interior address from every entry of
    PRIVATE_IP_RANGES.

    Replaces the prior TestBlockedIPRanges class, which asserted
    ``ip_network("127.0.0.0/8") in BLOCKED_IP_RANGES`` — a tautology that
    re-stated the constant rather than exercising blocking behavior. The
    parametrized assertions here would fail if any range entry were
    accidentally removed from PRIVATE_IP_RANGES in ip_ranges.py.
    """

    @pytest.mark.parametrize(
        "ip,range_label",
        [
            ("127.0.0.5", "127.0.0.0/8 IPv4 loopback"),
            ("::1", "::1/128 IPv6 loopback"),
            ("10.0.0.5", "10.0.0.0/8 RFC1918 class A"),
            ("172.16.0.5", "172.16.0.0/12 RFC1918 class B"),
            ("172.31.255.254", "172.16.0.0/12 RFC1918 class B upper bound"),
            ("192.168.0.5", "192.168.0.0/16 RFC1918 class C"),
            ("100.64.0.5", "100.64.0.0/10 CGNAT"),
            ("169.254.0.5", "169.254.0.0/16 link-local"),
            ("fe80::1", "fe80::/10 IPv6 link-local"),
            ("fc00::1", "fc00::/7 IPv6 ULA"),
            ("0.0.0.5", "0.0.0.0/8 unspecified IPv4"),
            ("::", "::/128 unspecified IPv6"),
            ("2002:7f00:1::", "2002::/16 6to4 wrapping loopback"),
            ("64:ff9b::a9fe:a9fe", "64:ff9b::/96 NAT64 well-known"),
            ("64:ff9b:1::a9fe:a9fe", "64:ff9b:1::/48 NAT64 local-use"),
            ("2001::1", "2001::/32 Teredo"),
            ("100::1", "100::/64 discard"),
            ("::7f00:1", "::/96 IPv4-compatible IPv6 wrapping loopback"),
        ],
    )
    def test_interior_address_blocked(self, ip, range_label):
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked(ip) is True, (
            f"Expected {ip} to be blocked because it falls in {range_label}"
        )


class TestAllowedSchemes:
    """Tests for ALLOWED_SCHEMES constant."""

    def test_allowed_schemes_is_set(self):
        """ALLOWED_SCHEMES is a set."""
        from local_deep_research.security.ssrf_validator import ALLOWED_SCHEMES

        assert isinstance(ALLOWED_SCHEMES, set)

    def test_http_in_allowed(self):
        """http is in ALLOWED_SCHEMES."""
        from local_deep_research.security.ssrf_validator import ALLOWED_SCHEMES

        assert "http" in ALLOWED_SCHEMES

    def test_https_in_allowed(self):
        """https is in ALLOWED_SCHEMES."""
        from local_deep_research.security.ssrf_validator import ALLOWED_SCHEMES

        assert "https" in ALLOWED_SCHEMES


class TestAlwaysBlockedMetadataIPs:
    """Tests for ALWAYS_BLOCKED_METADATA_IPS constant."""

    def test_aws_imds_in_always_blocked(self):
        """AWS / Azure / OCI / DigitalOcean shared IMDS IP is in the set."""
        from local_deep_research.security.ssrf_validator import (
            ALWAYS_BLOCKED_METADATA_IPS,
        )

        assert "169.254.169.254" in ALWAYS_BLOCKED_METADATA_IPS


class TestGetSafeUrl:
    """Tests for get_safe_url function."""

    def test_returns_url_if_safe(self):
        """Returns URL if it's safe."""
        from unittest.mock import patch

        from local_deep_research.security.ssrf_validator import get_safe_url

        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_returns_none_for_empty(self):
        """Returns default for empty URL."""
        from local_deep_research.security.ssrf_validator import get_safe_url

        result = get_safe_url("")
        assert result is None

    def test_returns_default_for_empty(self):
        """Returns specified default for empty URL."""
        from local_deep_research.security.ssrf_validator import get_safe_url

        result = get_safe_url("", default="https://fallback.com")
        assert result == "https://fallback.com"

    def test_returns_none_for_none_input(self):
        """Returns default for None input."""
        from local_deep_research.security.ssrf_validator import get_safe_url

        result = get_safe_url(None)
        assert result is None


class TestValidateUrlNoBypass:
    """Tests that validate_url performs real validation (no test bypass)."""

    def test_localhost_blocked_by_default(self):
        """validate_url blocks localhost by default."""
        from local_deep_research.security.ssrf_validator import validate_url

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

    def test_testing_env_does_not_bypass(self, monkeypatch):
        """TESTING=true environment variable does NOT bypass validation."""
        from local_deep_research.security.ssrf_validator import validate_url

        monkeypatch.setenv("TESTING", "true")
        result = validate_url("http://127.0.0.1")
        assert result is False


class TestAllowLocalhostFlag:
    """Tests for allow_localhost flag interaction."""

    def test_localhost_not_blocked_with_flag(self):
        """Localhost is not blocked when 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("::1", allow_localhost=True) is False

    def test_private_still_blocked_with_localhost_flag(self):
        """Private IPs still blocked with allow_localhost (not allow_private)."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        # allow_localhost only allows loopback, not all private
        assert is_ip_blocked("10.0.0.1", allow_localhost=True) is True
        assert is_ip_blocked("192.168.1.1", allow_localhost=True) is True


class TestAllowPrivateIPsFlag:
    """Tests for allow_private_ips flag interaction."""

    def test_private_ips_includes_loopback(self):
        """allow_private_ips also allows loopback."""
        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

    def test_aws_metadata_still_blocked_with_private(self):
        """AWS metadata is still blocked even with allow_private_ips."""
        from local_deep_research.security.ssrf_validator import is_ip_blocked

        assert is_ip_blocked("169.254.169.254", allow_private_ips=True) is True