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

"""High-value tests for security/ssrf_validator.py - SSRF Prevention."""

import socket
from unittest.mock import patch

import pytest


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


# ---------------------------------------------------------------------------
# is_ip_blocked - loopback / RFC1918 / CGNAT / link-local / IPv6
# ---------------------------------------------------------------------------


class TestIsIpBlockedLoopback:
    """Loopback address blocking."""

    def test_ipv4_loopback_blocked_by_default(self):
        assert is_ip_blocked("127.0.0.1") is True

    def test_ipv4_loopback_127_0_0_2_blocked(self):
        assert is_ip_blocked("127.0.0.2") is True

    def test_ipv6_loopback_blocked_by_default(self):
        assert is_ip_blocked("::1") is True

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

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


class TestIsIpBlockedRFC1918:
    """RFC 1918 private ranges."""

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

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

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

    def test_10_network_allowed_with_private_flag(self):
        assert is_ip_blocked("10.0.0.1", allow_private_ips=True) is False

    def test_172_16_allowed_with_private_flag(self):
        assert is_ip_blocked("172.16.0.1", allow_private_ips=True) is False

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


class TestIsIpBlockedSpecialRanges:
    """CGNAT, link-local, IPv6 private ranges."""

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

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

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

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

    def test_cgnat_allowed_with_private_flag(self):
        assert is_ip_blocked("100.64.0.1", allow_private_ips=True) is False

    def test_link_local_allowed_with_private_flag(self):
        assert is_ip_blocked("169.254.1.1", allow_private_ips=True) is False

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


class TestIsIpBlockedCloudMetadata:
    """Cloud-provider metadata endpoints always blocked under all flags."""

    @pytest.mark.parametrize("ip", sorted(ALWAYS_BLOCKED_METADATA_IPS))
    def test_metadata_ip_blocked_default(self, ip):
        assert is_ip_blocked(ip) is True

    @pytest.mark.parametrize("ip", sorted(ALWAYS_BLOCKED_METADATA_IPS))
    def test_metadata_ip_blocked_with_allow_localhost(self, ip):
        assert is_ip_blocked(ip, allow_localhost=True) is True

    @pytest.mark.parametrize("ip", sorted(ALWAYS_BLOCKED_METADATA_IPS))
    def test_metadata_ip_blocked_with_allow_private_ips(self, ip):
        assert is_ip_blocked(ip, allow_private_ips=True) is True


class TestIsIpBlockedIPv4Mapped:
    """IPv4-mapped IPv6 address unwrapping."""

    def test_ipv4_mapped_loopback_blocked(self):
        assert is_ip_blocked("::ffff:127.0.0.1") is True

    def test_ipv4_mapped_private_blocked(self):
        assert is_ip_blocked("::ffff:10.0.0.1") is True

    def test_ipv4_mapped_aws_metadata_blocked(self):
        assert is_ip_blocked("::ffff:169.254.169.254") is True

    def test_ipv4_mapped_public_not_blocked(self):
        assert is_ip_blocked("::ffff:8.8.8.8") is False


class TestIsIpBlockedPublicAndInvalid:
    """Public IPs and invalid input."""

    def test_public_ip_not_blocked(self):
        assert is_ip_blocked("8.8.8.8") is False

    def test_another_public_ip(self):
        assert is_ip_blocked("1.1.1.1") is False

    def test_invalid_ip_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


# ---------------------------------------------------------------------------
# validate_url
# ---------------------------------------------------------------------------


class TestValidateUrlScheme:
    """Scheme validation."""

    def test_ftp_scheme_rejected(self):
        assert validate_url("ftp://example.com/file") is False

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

    def test_javascript_scheme_rejected(self):
        assert validate_url("javascript:alert(1)") is False


class TestValidateUrlHostname:
    """Hostname and IP-based URL validation."""

    def test_missing_hostname_rejected(self):
        assert validate_url("http://") is False

    def test_ip_loopback_url_blocked(self):
        assert validate_url("http://127.0.0.1/admin") is False

    def test_ip_private_url_blocked(self):
        assert validate_url("http://192.168.1.1/") is False

    def test_aws_metadata_url_blocked(self):
        assert validate_url("http://169.254.169.254/latest/meta-data/") is False

    @patch("local_deep_research.security.ssrf_validator.socket.getaddrinfo")
    def test_hostname_resolving_to_public_ip_allowed(self, mock_dns):
        mock_dns.return_value = [
            (socket.AF_INET, socket.SOCK_STREAM, 0, "", ("93.184.216.34", 0))
        ]
        assert validate_url("http://example.com") is True

    @patch("local_deep_research.security.ssrf_validator.socket.getaddrinfo")
    def test_hostname_resolving_to_private_ip_blocked(self, mock_dns):
        mock_dns.return_value = [
            (socket.AF_INET, socket.SOCK_STREAM, 0, "", ("10.0.0.1", 0))
        ]
        assert validate_url("http://internal.corp") is False

    @patch("local_deep_research.security.ssrf_validator.socket.getaddrinfo")
    def test_dns_failure_blocks_url(self, mock_dns):
        mock_dns.side_effect = socket.gaierror("Name resolution failed")
        assert validate_url("http://nonexistent.test") is False


class TestValidateUrlFlagPassthrough:
    """Flag pass-through to is_ip_blocked."""

    @patch("local_deep_research.security.ssrf_validator.socket.getaddrinfo")
    def test_allow_localhost_passes_loopback(self, mock_dns):
        mock_dns.return_value = [
            (socket.AF_INET, socket.SOCK_STREAM, 0, "", ("127.0.0.1", 0))
        ]
        assert validate_url("http://127.0.0.1/", allow_localhost=True) is True

    @patch("local_deep_research.security.ssrf_validator.socket.getaddrinfo")
    def test_allow_private_ips_passes_private(self, mock_dns):
        mock_dns.return_value = [
            (socket.AF_INET, socket.SOCK_STREAM, 0, "", ("10.0.0.1", 0))
        ]
        assert validate_url("http://10.0.0.1/", allow_private_ips=True) is True

    @patch("local_deep_research.security.ssrf_validator.socket.getaddrinfo")
    def test_multi_address_dns_one_private_blocked(self, mock_dns):
        mock_dns.return_value = [
            (socket.AF_INET, socket.SOCK_STREAM, 0, "", ("93.184.216.34", 0)),
            (socket.AF_INET, socket.SOCK_STREAM, 0, "", ("10.0.0.1", 0)),
        ]
        assert validate_url("http://dual.example.com") is False


class TestIsIpBlockedZeroNetwork:
    """0.0.0.0/8 range blocking."""

    def test_0001_blocked(self):
        assert is_ip_blocked("0.0.0.1") is True

    def test_0001_blocked_even_with_allow_private(self):
        """0.0.0.0/8 is NOT in PRIVATE_RANGES, so it stays blocked."""
        assert is_ip_blocked("0.0.0.1", allow_private_ips=True) is True


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

    def test_ftp_scheme_blocked(self):
        assert validate_url("ftp://evil.com") is False

    def test_https_public_url_allowed(self):
        with patch("socket.getaddrinfo") as mock_dns:
            mock_dns.return_value = [(2, 1, 6, "", ("93.184.216.34", 0))]
            assert validate_url("https://example.com") is True


# ---------------------------------------------------------------------------
# get_safe_url
# ---------------------------------------------------------------------------


class TestGetSafeUrl:
    """get_safe_url() wrapper."""

    def test_none_returns_default(self):
        assert get_safe_url(None, "fallback") == "fallback"

    def test_empty_returns_default(self):
        assert get_safe_url("", "fallback") == "fallback"

    def test_safe_url_passes_through(self):
        """Exercises real validate_url end-to-end. Mocking only DNS means
        a regression in scheme/IP/host-extraction logic would surface
        here, not just whichever code path the mock previously covered.
        """
        with patch(
            "local_deep_research.security.ssrf_validator.socket.getaddrinfo",
            return_value=[(2, 1, 6, "", ("93.184.216.34", 0))],
        ):
            assert get_safe_url("http://example.com") == "http://example.com"

    def test_unsafe_url_returns_default(self):
        """Real-validator pass-through: the IP literal 10.0.0.5 is
        rejected without DNS, so removing the RFC1918 entry from
        PRIVATE_IP_RANGES would cause this test to fail.
        """
        assert get_safe_url("http://10.0.0.5/", "safe") == "safe"

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


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


class TestConstants:
    def test_allowed_schemes_http_https(self):
        assert ALLOWED_SCHEMES == {"http", "https"}

    def test_always_blocked_metadata_ips_membership(self):
        """Lock in the exact membership of the always-blocked set so a
        future contributor accidentally removing an IP fails loudly."""
        assert ALWAYS_BLOCKED_METADATA_IPS == frozenset(
            {
                "169.254.169.254",
                "169.254.170.2",
                "169.254.170.23",
                "169.254.0.23",
                "100.100.100.200",
            }
        )