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

"""Tests for SSRF bypass simplification and debug hardening.

Verifies:
- Only PYTEST_CURRENT_TEST bypasses SSRF validation (behavioral)
- Removed ssrf.disable_validation setting has no effect
- Debug mode setting is non-editable in default_settings.json
- Full search engine validates URLs via SSRF before fetching
"""

import json
import os
import socket
from unittest.mock import patch, MagicMock


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


class TestSSRFValidationAlwaysActive:
    """Verify SSRF validation is always active (no test bypass)."""

    def test_aws_metadata_blocked(self):
        """AWS metadata endpoint should be blocked."""
        assert validate_url("http://169.254.169.254/latest/meta-data") is False

    def test_public_url_allowed(self):
        """Normal public URLs should pass."""
        with patch(
            "local_deep_research.security.ssrf_validator.socket.getaddrinfo"
        ) as mock_dns:
            mock_dns.return_value = [(2, 1, 6, "", ("93.184.216.34", 0))]
            assert validate_url("http://example.com") is True

    def test_testing_env_var_does_not_bypass(self):
        """Setting TESTING=true should NOT bypass SSRF validation."""
        with patch.dict(os.environ, {"TESTING": "true"}, clear=False):
            assert (
                validate_url("http://169.254.169.254/latest/meta-data") is False
            )


class TestRemovedSSRFDisableValidationSetting:
    """Verify the removed ssrf.disable_validation setting has no effect."""

    def test_ldr_ssrf_disable_validation_env_var_has_no_effect(self):
        """LDR_SECURITY_SSRF_DISABLE_VALIDATION env var must not bypass validation."""
        with patch.dict(
            os.environ,
            {"LDR_SECURITY_SSRF_DISABLE_VALIDATION": "true"},
            clear=False,
        ):
            assert (
                validate_url("http://169.254.169.254/latest/meta-data") is False
            )
            assert validate_url("http://192.168.1.1/admin") is False

    def test_ssrf_disable_validation_key_not_in_registry(self):
        """ssrf.disable_validation must not be registered in the env registry."""
        from local_deep_research.settings.env_registry import registry

        assert not registry.is_env_only("security.ssrf.disable_validation"), (
            "security.ssrf.disable_validation must NOT be in the env registry"
        )

    def test_dns_failure_fails_closed(self):
        """DNS resolution failure must block the URL (fail-closed)."""
        with patch(
            "local_deep_research.security.ssrf_validator.socket.getaddrinfo",
            side_effect=socket.gaierror("DNS resolution failed"),
        ):
            assert validate_url("http://nonexistent-host.example.com") is False


class TestDebugModeNonEditable:
    """Verify debug mode setting cannot be changed by users."""

    def test_debug_setting_not_editable(self):
        """app.debug setting should have editable=false."""
        settings_path = "src/local_deep_research/defaults/default_settings.json"
        with open(settings_path) as f:
            settings = json.load(f)

        debug_setting = settings.get("app.debug", {})
        assert debug_setting.get("editable") is False, (
            "app.debug must not be editable from the UI"
        )


class TestFullSearchSSRFValidation:
    """Verify full search engine validates URLs before fetching content."""

    def test_ssrf_blocked_urls_logged_and_skipped(self):
        """URLs that fail SSRF validation should be logged and not fetched.

        Exercises the real ssrf_validator (no validate_url mock) so a
        regression in metadata-IP blocking would surface here too. The
        bad URL is an IP literal that is rejected without DNS, and the
        good URL has socket.getaddrinfo mocked to a public IP.
        """
        from local_deep_research.web_search_engines.engines.full_search import (
            FullSearchResults,
        )

        mock_llm = MagicMock()
        mock_web_search = MagicMock()
        mock_web_search.invoke.return_value = [
            {"link": "http://169.254.169.254/secret", "title": "bad"},
            {"link": "http://example.com/good", "title": "good"},
        ]

        fs = FullSearchResults(llm=mock_llm, web_search=mock_web_search)

        with (
            patch(
                "local_deep_research.web_search_engines.engines.full_search.QUALITY_CHECK_DDG_URLS",
                False,
            ),
            patch(
                "local_deep_research.security.ssrf_validator.socket.getaddrinfo",
                return_value=[(2, 1, 6, "", ("93.184.216.34", 0))],
            ),
            patch(
                "local_deep_research.web_search_engines.engines.full_search.batch_fetch_and_extract"
            ) as mock_batch,
        ):
            mock_batch.return_value = {
                "http://example.com/good": "Safe content",
            }

            results = fs.run("test query")

            mock_batch.assert_called_once()
            fetched_urls = mock_batch.call_args[0][0]
            assert "http://169.254.169.254/secret" not in fetched_urls
            assert "http://example.com/good" in fetched_urls
            assert len(results) == 2

    def test_blocked_url_gets_none_safe_url_gets_content(self):
        """Blocked results must have full_content=None; safe results get real content.

        Exercises the real ssrf_validator so removing the
        ALWAYS_BLOCKED_METADATA_IPS check in ssrf_validator.py would
        cause this test to fail (bad URL would no longer be blocked).
        """
        from local_deep_research.web_search_engines.engines.full_search import (
            FullSearchResults,
        )

        mock_llm = MagicMock()
        mock_web_search = MagicMock()
        mock_web_search.invoke.return_value = [
            {"link": "http://169.254.169.254/secret", "title": "bad"},
            {"link": "http://example.com/good", "title": "good"},
        ]

        fs = FullSearchResults(llm=mock_llm, web_search=mock_web_search)

        with (
            patch(
                "local_deep_research.web_search_engines.engines.full_search.QUALITY_CHECK_DDG_URLS",
                False,
            ),
            patch(
                "local_deep_research.security.ssrf_validator.socket.getaddrinfo",
                return_value=[(2, 1, 6, "", ("93.184.216.34", 0))],
            ),
            patch(
                "local_deep_research.web_search_engines.engines.full_search.batch_fetch_and_extract"
            ) as mock_batch,
        ):
            mock_batch.return_value = {
                "http://example.com/good": "Safe content",
            }

            results = fs.run("test query")

            bad = next(
                r
                for r in results
                if r["link"] == "http://169.254.169.254/secret"
            )
            good = next(
                r for r in results if r["link"] == "http://example.com/good"
            )

            assert bad["full_content"] is None
            assert good["full_content"] is not None

    def test_validate_url_called_with_raw_url(self):
        """validate_url must receive the exact URL from search results."""
        from local_deep_research.web_search_engines.engines.full_search import (
            FullSearchResults,
        )

        raw_url = "http://169.254.169.254:80/secret"
        mock_web_search = MagicMock()
        mock_web_search.invoke.return_value = [
            {"link": raw_url, "title": "bad"}
        ]

        fs = FullSearchResults(llm=MagicMock(), web_search=[MagicMock()])
        fs.web_search = mock_web_search

        with (
            patch(
                "local_deep_research.web_search_engines.engines.full_search.QUALITY_CHECK_DDG_URLS",
                False,
            ),
            patch(
                "local_deep_research.web_search_engines.engines.full_search.validate_url"
            ) as mock_validate,
        ):
            mock_validate.return_value = False
            fs.run("test")
            mock_validate.assert_called_once_with(raw_url)

    def test_all_urls_blocked_returns_results_without_content(self):
        """When all URLs are blocked, results are returned with full_content=None.

        Uses a real private IP literal (10.0.0.1) so the real SSRF
        validator does the blocking — no validate_url mock needed.
        """
        from local_deep_research.web_search_engines.engines.full_search import (
            FullSearchResults,
        )

        mock_llm = MagicMock()
        mock_web_search = MagicMock()
        mock_web_search.invoke.return_value = [
            {"link": "http://10.0.0.1/internal", "title": "internal"},
        ]

        fs = FullSearchResults(llm=mock_llm, web_search=[mock_web_search])
        fs.web_search = mock_web_search

        with patch(
            "local_deep_research.web_search_engines.engines.full_search.QUALITY_CHECK_DDG_URLS",
            False,
        ):
            results = fs.run("test query")

            assert len(results) == 1
            assert results[0]["full_content"] is None


# ---------------------------------------------------------------------------
# Group 1: CRITICAL — IPv4-Mapped IPv6 Bypass
# ---------------------------------------------------------------------------
class TestIPv4MappedIPv6Bypass:
    """IPv4-mapped IPv6 addresses (::ffff:x.x.x.x) must be unwrapped and blocked."""

    def test_ipv4_mapped_ipv6_aws_metadata_blocked(self):
        """::ffff:169.254.169.254 must be blocked (AWS credential theft via IPv6)."""
        assert is_ip_blocked("::ffff:169.254.169.254") is True

    def test_ipv4_mapped_ipv6_loopback_blocked(self):
        """::ffff:127.0.0.1 must be blocked (localhost SSRF via IPv6)."""
        assert is_ip_blocked("::ffff:127.0.0.1") is True

    def test_ipv4_mapped_ipv6_private_ip_blocked(self):
        """::ffff:192.168.1.1 must be blocked (internal network via IPv6)."""
        assert is_ip_blocked("::ffff:192.168.1.1") is True

    def test_ipv4_mapped_ipv6_dns_resolution_blocked(self):
        """DNS returning ::ffff:127.0.0.1 must be caught."""
        with patch(
            "local_deep_research.security.ssrf_validator.socket.getaddrinfo"
        ) as mock_dns:
            mock_dns.return_value = [
                (socket.AF_INET6, 1, 6, "", ("::ffff:127.0.0.1", 0, 0, 0))
            ]
            assert validate_url("http://evil.example.com/") is False


# ---------------------------------------------------------------------------
# Group 2: HIGH — Alternative IP Notation Defense-in-Depth
# ---------------------------------------------------------------------------
class TestAlternativeIPNotations:
    """Alternative IP representations must not bypass SSRF checks."""

    def test_decimal_dword_loopback_blocked(self):
        """Decimal/dword notation (2130706433 = 127.0.0.1) must be blocked."""
        with patch(
            "local_deep_research.security.ssrf_validator.socket.getaddrinfo"
        ) as mock_dns:
            mock_dns.return_value = [(2, 1, 6, "", ("127.0.0.1", 0))]
            assert validate_url("http://2130706433/") is False

    def test_hex_ip_aws_metadata_blocked(self):
        """Hex notation (0xa9fea9fe = 169.254.169.254) must be blocked."""
        with patch(
            "local_deep_research.security.ssrf_validator.socket.getaddrinfo"
        ) as mock_dns:
            mock_dns.return_value = [(2, 1, 6, "", ("169.254.169.254", 0))]
            assert validate_url("http://0xa9fea9fe/") is False

    def test_octal_ip_loopback_blocked(self):
        """Octal notation (0177.0.0.1 = 127.0.0.1) must be blocked."""
        with patch(
            "local_deep_research.security.ssrf_validator.socket.getaddrinfo"
        ) as mock_dns:
            mock_dns.return_value = [(2, 1, 6, "", ("127.0.0.1", 0))]
            assert validate_url("http://0177.0.0.1/") is False

    def test_zero_address_blocked(self):
        """http://0/ resolving to 0.0.0.0 must be blocked."""
        with patch(
            "local_deep_research.security.ssrf_validator.socket.getaddrinfo"
        ) as mock_dns:
            mock_dns.return_value = [(2, 1, 6, "", ("0.0.0.0", 0))]
            assert validate_url("http://0/") is False


# ---------------------------------------------------------------------------
# Group 3: MEDIUM — URL Parsing Edge Cases
# ---------------------------------------------------------------------------
class TestURLParsingEdgeCases:
    """URL parsing quirks must not create SSRF bypass opportunities."""

    def test_leading_whitespace_blocked(self):
        """Leading whitespace must not bypass validation (CVE-2023-24329 regression)."""
        assert validate_url(" http://169.254.169.254/") is False

    def test_scheme_case_insensitive(self):
        """Mixed-case scheme (HTTP://) must still validate the IP."""
        assert validate_url("HTTP://169.254.169.254/") is False

    def test_url_without_scheme_rejected(self):
        """Scheme-less URL (//host/path) must be rejected."""
        assert validate_url("//169.254.169.254/latest") is False

    def test_crlf_injection_blocked(self):
        r"""URL with CRLF (\r\n) must be blocked."""
        assert (
            validate_url("http://169.254.169.254\r\nX-Injected: true") is False
        )


# ---------------------------------------------------------------------------
# Group 4: MEDIUM — Debug Mode Completeness
# ---------------------------------------------------------------------------
class TestDebugModeCompleteness:
    """Additional debug mode setting integrity checks."""

    def test_debug_setting_value_defaults_to_false(self):
        """app.debug default value must be False."""
        settings_path = "src/local_deep_research/defaults/default_settings.json"
        with open(settings_path) as f:
            settings = json.load(f)
        assert settings["app.debug"]["value"] is False

    def test_debug_description_mentions_env_var(self):
        """app.debug description must mention the LDR_APP_DEBUG env var."""
        settings_path = "src/local_deep_research/defaults/default_settings.json"
        with open(settings_path) as f:
            settings = json.load(f)
        assert "LDR_APP_DEBUG" in settings["app.debug"]["description"]


# ---------------------------------------------------------------------------
# Group 5: LOW — get_safe_url Coverage
# ---------------------------------------------------------------------------
class TestGetSafeUrl:
    """Coverage for the get_safe_url helper."""

    def test_returns_default_for_blocked_url(self):
        """Blocked URLs must return the default value."""
        result = get_safe_url(
            "http://169.254.169.254/", default="https://fallback.example.com"
        )
        assert result == "https://fallback.example.com"

    def test_returns_url_for_safe_url(self):
        """Safe URLs must be returned as-is."""
        with patch(
            "local_deep_research.security.ssrf_validator.socket.getaddrinfo"
        ) as mock_dns:
            mock_dns.return_value = [(2, 1, 6, "", ("93.184.216.34", 0))]
            result = get_safe_url("http://example.com/page")
        assert result == "http://example.com/page"

    def test_returns_default_for_none(self):
        """None input must return the default without raising."""
        result = get_safe_url(None, default="https://default.com")
        assert result == "https://default.com"