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

"""Tests for safe_requests module - SSRF-protected HTTP requests."""

import pytest
from unittest.mock import patch, MagicMock
import requests

from local_deep_research.security import ssrf_validator
from local_deep_research.security.safe_requests import (
    safe_get,
    safe_post,
    SafeSession,
    DEFAULT_TIMEOUT,
    MAX_RESPONSE_SIZE,
)


class TestSafeGetFunction:
    """Tests for safe_get function."""

    def test_valid_url_makes_request(self):
        """Should make request to valid external URL."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.get"
            ) as mock_get:
                mock_response = MagicMock()
                mock_response.headers = {}
                mock_get.return_value = mock_response

                response = safe_get("https://example.com")

                mock_get.assert_called_once()
                assert response == mock_response

    def test_rejects_invalid_url(self):
        """Should raise ValueError for URLs failing SSRF validation."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=False,
        ):
            with pytest.raises(ValueError, match="SSRF"):
                safe_get("http://127.0.0.1/admin")

    def test_uses_default_timeout(self):
        """Should use default timeout when not specified."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.get"
            ) as mock_get:
                mock_response = MagicMock()
                mock_response.headers = {}
                mock_get.return_value = mock_response

                safe_get("https://example.com")

                _, kwargs = mock_get.call_args
                assert kwargs["timeout"] == DEFAULT_TIMEOUT

    def test_custom_timeout(self):
        """Should use custom timeout when provided."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.get"
            ) as mock_get:
                mock_response = MagicMock()
                mock_response.headers = {}
                mock_get.return_value = mock_response

                safe_get("https://example.com", timeout=60)

                _, kwargs = mock_get.call_args
                assert kwargs["timeout"] == 60

    def test_underlying_requests_always_gets_redirects_disabled(self):
        """The underlying requests.get() always receives allow_redirects=False;
        safe_get handles redirects manually with SSRF validation."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.get"
            ) as mock_get:
                mock_response = MagicMock()
                mock_response.headers = {}
                mock_get.return_value = mock_response

                safe_get("https://example.com")

                _, kwargs = mock_get.call_args
                assert kwargs["allow_redirects"] is False

    def test_can_enable_redirects(self):
        """With allow_redirects=True, redirects are still disabled at
        requests level (handled manually for SSRF validation)."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.get"
            ) as mock_get:
                mock_response = MagicMock()
                mock_response.headers = {}
                mock_get.return_value = mock_response

                safe_get("https://example.com", allow_redirects=True)

                _, kwargs = mock_get.call_args
                # Redirects are always disabled at the requests level;
                # safe_get handles them manually with SSRF validation
                assert kwargs["allow_redirects"] is False

    def test_passes_params(self):
        """Should pass URL parameters."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.get"
            ) as mock_get:
                mock_response = MagicMock()
                mock_response.headers = {}
                mock_get.return_value = mock_response

                params = {"q": "test", "page": "1"}
                safe_get("https://example.com", params=params)

                args, _ = mock_get.call_args
                assert args == ("https://example.com",)
                _, kwargs = mock_get.call_args
                assert "params" not in kwargs or kwargs.get("params") == params

    def test_oversized_response_raises_error(self):
        """Oversized responses are rejected with ValueError."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.get"
            ) as mock_get:
                mock_response = MagicMock()
                mock_response.headers = {
                    "Content-Length": str(MAX_RESPONSE_SIZE + 1)
                }
                mock_get.return_value = mock_response

                with pytest.raises(ValueError, match="Response too large"):
                    safe_get("https://example.com")

    def test_accepts_response_within_limit(self):
        """Should accept response within size limit."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.get"
            ) as mock_get:
                mock_response = MagicMock()
                mock_response.headers = {"Content-Length": str(1024)}
                mock_get.return_value = mock_response

                response = safe_get("https://example.com")
                assert response == mock_response

    def test_handles_invalid_content_length(self):
        """Should ignore invalid Content-Length values."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.get"
            ) as mock_get:
                mock_response = MagicMock()
                mock_response.headers = {"Content-Length": "not-a-number"}
                mock_get.return_value = mock_response

                # Should not raise
                response = safe_get("https://example.com")
                assert response == mock_response

    def test_reraises_timeout(self):
        """Should re-raise timeout exceptions."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.get",
                side_effect=requests.Timeout("timeout"),
            ):
                with pytest.raises(requests.Timeout):
                    safe_get("https://example.com")

    def test_reraises_request_exception(self):
        """Should re-raise request exceptions."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.get",
                side_effect=requests.RequestException("connection error"),
            ):
                with pytest.raises(requests.RequestException):
                    safe_get("https://example.com")

    def test_allow_localhost_parameter(self):
        """Should pass allow_localhost to validate_url."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ) as mock_validate:
            with patch(
                "local_deep_research.security.safe_requests.requests.get"
            ) as mock_get:
                mock_response = MagicMock()
                mock_response.headers = {}
                mock_get.return_value = mock_response

                safe_get("http://localhost:8080", allow_localhost=True)

                mock_validate.assert_called_once_with(
                    "http://localhost:8080",
                    allow_localhost=True,
                    allow_private_ips=False,
                )

    def test_allow_private_ips_parameter(self):
        """Should pass allow_private_ips to validate_url."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ) as mock_validate:
            with patch(
                "local_deep_research.security.safe_requests.requests.get"
            ) as mock_get:
                mock_response = MagicMock()
                mock_response.headers = {}
                mock_get.return_value = mock_response

                safe_get("http://192.168.1.1", allow_private_ips=True)

                mock_validate.assert_called_once_with(
                    "http://192.168.1.1",
                    allow_localhost=False,
                    allow_private_ips=True,
                )


class TestSafePostFunction:
    """Tests for safe_post function."""

    def test_valid_url_makes_request(self):
        """Should make POST request to valid external URL."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.post"
            ) as mock_post:
                mock_response = MagicMock()
                mock_response.headers = {}
                mock_post.return_value = mock_response

                response = safe_post(
                    "https://example.com/api", json={"key": "value"}
                )

                mock_post.assert_called_once()
                assert response == mock_response

    def test_rejects_invalid_url(self):
        """Should raise ValueError for URLs failing SSRF validation."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=False,
        ):
            with pytest.raises(ValueError, match="SSRF"):
                safe_post("http://169.254.169.254/metadata")

    def test_passes_data_parameter(self):
        """Should pass data parameter for form data."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.post"
            ) as mock_post:
                mock_response = MagicMock()
                mock_response.headers = {}
                mock_post.return_value = mock_response

                safe_post("https://example.com", data="raw data")

                _, kwargs = mock_post.call_args
                assert kwargs.get("data") == "raw data"

    def test_passes_json_parameter(self):
        """Should pass json parameter for JSON data."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.post"
            ) as mock_post:
                mock_response = MagicMock()
                mock_response.headers = {}
                mock_post.return_value = mock_response

                json_data = {"key": "value"}
                safe_post("https://example.com", json=json_data)

                _, kwargs = mock_post.call_args
                assert kwargs.get("json") == json_data

    def test_uses_default_timeout(self):
        """Should use default timeout when not specified."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.post"
            ) as mock_post:
                mock_response = MagicMock()
                mock_response.headers = {}
                mock_post.return_value = mock_response

                safe_post("https://example.com")

                _, kwargs = mock_post.call_args
                assert kwargs["timeout"] == DEFAULT_TIMEOUT

    def test_underlying_requests_always_gets_redirects_disabled(self):
        """The underlying requests.post() always receives allow_redirects=False;
        safe_post handles redirects manually with SSRF validation."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.post"
            ) as mock_post:
                mock_response = MagicMock()
                mock_response.headers = {}
                mock_post.return_value = mock_response

                safe_post("https://example.com")

                _, kwargs = mock_post.call_args
                assert kwargs["allow_redirects"] is False

    def test_can_enable_redirects(self):
        """With allow_redirects=True, redirects are still disabled at
        requests level (handled manually for SSRF validation)."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.post"
            ) as mock_post:
                mock_response = MagicMock()
                mock_response.headers = {}
                mock_post.return_value = mock_response

                safe_post("https://example.com", allow_redirects=True)

                _, kwargs = mock_post.call_args
                # Redirects are always disabled at the requests level;
                # safe_post handles them manually with SSRF validation
                assert kwargs["allow_redirects"] is False

    def test_oversized_response_raises_error(self):
        """Oversized responses are rejected with ValueError."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.post"
            ) as mock_post:
                mock_response = MagicMock()
                mock_response.headers = {
                    "Content-Length": str(MAX_RESPONSE_SIZE + 1)
                }
                mock_post.return_value = mock_response

                with pytest.raises(ValueError, match="Response too large"):
                    safe_post("https://example.com")

    def test_reraises_timeout(self):
        """Should re-raise timeout exceptions."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch(
                "local_deep_research.security.safe_requests.requests.post",
                side_effect=requests.Timeout("timeout"),
            ):
                with pytest.raises(requests.Timeout):
                    safe_post("https://example.com")


class TestSafeSession:
    """Tests for SafeSession class."""

    def test_init_default_values(self):
        """Should initialize with default security settings."""
        session = SafeSession()
        assert session.allow_localhost is False
        assert session.allow_private_ips is False

    def test_init_allow_localhost(self):
        """Should accept allow_localhost parameter."""
        session = SafeSession(allow_localhost=True)
        assert session.allow_localhost is True
        assert session.allow_private_ips is False

    def test_init_allow_private_ips(self):
        """Should accept allow_private_ips parameter."""
        session = SafeSession(allow_private_ips=True)
        assert session.allow_localhost is False
        assert session.allow_private_ips is True

    def test_request_validates_url(self):
        """Should validate URL before making request."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=False,
        ):
            session = SafeSession()
            with pytest.raises(ValueError, match="SSRF"):
                session.request("GET", "http://127.0.0.1/admin")

    def test_request_makes_call_on_valid_url(self):
        """Should make request when URL is valid."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch.object(requests.Session, "request") as mock_request:
                mock_response = MagicMock()
                mock_request.return_value = mock_response

                session = SafeSession()
                response = session.request("GET", "https://example.com")

                mock_request.assert_called_once()
                assert response == mock_response

    def test_request_uses_default_timeout(self):
        """Should set default timeout if not provided."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch.object(requests.Session, "request") as mock_request:
                mock_response = MagicMock()
                mock_request.return_value = mock_response

                session = SafeSession()
                session.request("GET", "https://example.com")

                _, kwargs = mock_request.call_args
                assert kwargs["timeout"] == DEFAULT_TIMEOUT

    def test_request_respects_custom_timeout(self):
        """Should respect custom timeout when provided."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with patch.object(requests.Session, "request") as mock_request:
                mock_response = MagicMock()
                mock_request.return_value = mock_response

                session = SafeSession()
                session.request("GET", "https://example.com", timeout=120)

                _, kwargs = mock_request.call_args
                assert kwargs["timeout"] == 120

    def test_context_manager(self):
        """Should work as context manager."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ):
            with SafeSession() as session:
                assert isinstance(session, SafeSession)

    def test_passes_allow_localhost_to_validate(self):
        """Should pass allow_localhost to validate_url."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ) as mock_validate:
            with patch.object(requests.Session, "request"):
                session = SafeSession(allow_localhost=True)
                session.request("GET", "http://localhost:8080")

                mock_validate.assert_called_once_with(
                    "http://localhost:8080",
                    allow_localhost=True,
                    allow_private_ips=False,
                )

    def test_passes_allow_private_ips_to_validate(self):
        """Should pass allow_private_ips to validate_url."""
        with patch.object(
            ssrf_validator,
            "validate_url",
            return_value=True,
        ) as mock_validate:
            with patch.object(requests.Session, "request"):
                session = SafeSession(allow_private_ips=True)
                session.request("GET", "http://192.168.1.1")

                mock_validate.assert_called_once_with(
                    "http://192.168.1.1",
                    allow_localhost=False,
                    allow_private_ips=True,
                )


class TestConstants:
    """Tests for module constants."""

    def test_default_timeout_reasonable(self):
        """DEFAULT_TIMEOUT should be a reasonable value."""
        assert DEFAULT_TIMEOUT == 30
        assert isinstance(DEFAULT_TIMEOUT, int)

    def test_max_response_size_reasonable(self):
        """MAX_RESPONSE_SIZE should be a reasonable value (1GB)."""
        assert MAX_RESPONSE_SIZE == 1024 * 1024 * 1024  # 1GB
        assert isinstance(MAX_RESPONSE_SIZE, int)


class TestUserAgentInjection:
    """The project User-Agent should be injected automatically when the
    caller doesn't supply one, and preserved when they do.

    This locks in the behavior added by PR #3081 — academic API endpoints
    (arXiv, OpenAlex, PubMed, …) need to identify the caller, and we
    don't want every call site to remember to set the header manually.
    """

    def _make_response(self):
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.headers = {}
        mock_response.url = "https://example.com"
        return mock_response

    def test_safe_get_injects_user_agent_when_missing(self):
        """safe_get with no headers should auto-inject the project UA."""
        from local_deep_research.constants import USER_AGENT

        with (
            patch.object(ssrf_validator, "validate_url", return_value=True),
            patch(
                "local_deep_research.security.safe_requests.requests.get",
                return_value=self._make_response(),
            ) as mock_get,
        ):
            safe_get("https://example.com")

            _, kwargs = mock_get.call_args
            assert "headers" in kwargs
            assert kwargs["headers"]["User-Agent"] == USER_AGENT

    def test_safe_get_preserves_explicit_user_agent(self):
        """A caller-supplied User-Agent must NOT be overwritten."""
        custom_ua = "Mozilla/5.0 Custom/1.0"

        with (
            patch.object(ssrf_validator, "validate_url", return_value=True),
            patch(
                "local_deep_research.security.safe_requests.requests.get",
                return_value=self._make_response(),
            ) as mock_get,
        ):
            safe_get("https://example.com", headers={"User-Agent": custom_ua})

            _, kwargs = mock_get.call_args
            assert kwargs["headers"]["User-Agent"] == custom_ua

    def test_safe_get_user_agent_check_is_case_insensitive(self):
        """A `user-agent` (lowercase) header should be respected."""
        custom_ua = "Mozilla/5.0 Custom/1.0"

        with (
            patch.object(ssrf_validator, "validate_url", return_value=True),
            patch(
                "local_deep_research.security.safe_requests.requests.get",
                return_value=self._make_response(),
            ) as mock_get,
        ):
            safe_get("https://example.com", headers={"user-agent": custom_ua})

            _, kwargs = mock_get.call_args
            # Original lowercase key preserved; no second User-Agent added
            assert kwargs["headers"]["user-agent"] == custom_ua
            assert "User-Agent" not in kwargs["headers"]

    def test_safe_get_does_not_mutate_caller_headers_dict(self):
        """safe_get must not add User-Agent to the caller's dict."""
        caller_headers = {"X-Custom": "value"}

        with (
            patch.object(ssrf_validator, "validate_url", return_value=True),
            patch(
                "local_deep_research.security.safe_requests.requests.get",
                return_value=self._make_response(),
            ),
        ):
            safe_get("https://example.com", headers=caller_headers)

            # Caller's dict is untouched — User-Agent injection is on a
            # copy, not the original.
            assert "User-Agent" not in caller_headers
            assert caller_headers == {"X-Custom": "value"}

    def test_safe_post_injects_user_agent_when_missing(self):
        """safe_post should auto-inject the project UA."""
        from local_deep_research.constants import USER_AGENT

        with (
            patch.object(ssrf_validator, "validate_url", return_value=True),
            patch(
                "local_deep_research.security.safe_requests.requests.post",
                return_value=self._make_response(),
            ) as mock_post,
        ):
            safe_post("https://example.com", data={"k": "v"})

            _, kwargs = mock_post.call_args
            assert kwargs["headers"]["User-Agent"] == USER_AGENT

    def test_safe_post_preserves_explicit_user_agent(self):
        custom_ua = "Mozilla/5.0 Custom/1.0"

        with (
            patch.object(ssrf_validator, "validate_url", return_value=True),
            patch(
                "local_deep_research.security.safe_requests.requests.post",
                return_value=self._make_response(),
            ) as mock_post,
        ):
            safe_post(
                "https://example.com",
                data={"k": "v"},
                headers={"User-Agent": custom_ua},
            )

            _, kwargs = mock_post.call_args
            assert kwargs["headers"]["User-Agent"] == custom_ua


class TestParserDifferentialEndToEnd:
    """
    End-to-end integration tests for the parser-differential SSRF bypass
    fix (GHSA-g23j-2vwm-5c25).

    Approach: bind a TCP socket on 127.0.0.1:<random> WITHOUT calling
    listen() — the kernel responds RST to any incoming connect.  If the
    fix regresses and ``safe_get`` actually attempts to connect to the
    bound port, ``requests`` raises ``ConnectionError`` (kernel RST), so
    a strict ``pytest.raises(ValueError, match=...)`` distinguishes
    "validator caught it" from "validator failed and the kernel saved us".
    """

    @staticmethod
    def _bind_unused_port():
        import socket as _socket

        sock = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM)
        sock.bind(("127.0.0.1", 0))
        return sock

    def test_safe_get_blocks_parser_differential_no_socket_connect(self):
        from local_deep_research.security.safe_requests import safe_get

        sock = self._bind_unused_port()
        try:
            port = sock.getsockname()[1]
            bypass_url = f"http://127.0.0.1:{port}\\@1.1.1.1"
            with pytest.raises(ValueError, match="SSRF|security validation"):
                safe_get(bypass_url, timeout=2)
        finally:
            sock.close()

    def test_safe_post_blocks_parser_differential(self):
        from local_deep_research.security.safe_requests import safe_post

        sock = self._bind_unused_port()
        try:
            port = sock.getsockname()[1]
            bypass_url = f"http://127.0.0.1:{port}\\@1.1.1.1"
            with pytest.raises(ValueError, match="SSRF|security validation"):
                safe_post(bypass_url, data={"k": "v"}, timeout=2)
        finally:
            sock.close()

    def test_safesession_blocks_parser_differential(self):
        """SafeSession validates at both request() and send() — exercises
        the double-validation path. This URL contains ``\\`` so Layer 1
        catches it at request() before .prepare() canonicalises it."""
        from local_deep_research.security.safe_requests import SafeSession

        sock = self._bind_unused_port()
        try:
            port = sock.getsockname()[1]
            bypass_url = f"http://127.0.0.1:{port}\\@1.1.1.1"
            with SafeSession() as sess:
                with pytest.raises(
                    ValueError, match="SSRF|security validation"
                ):
                    sess.get(bypass_url, timeout=2)
        finally:
            sock.close()

    def test_safesession_send_blocks_canonicalised_form(self):
        """
        Layer-2 verification: ``SafeSession.send()`` is called with a
        ``PreparedRequest`` whose URL contains ``%5C`` (the canonicalised
        form of ``\\``).  Layer 1 doesn't match ``%5C``, so Layer 2's
        urllib3-based hostname extraction is what blocks this — proving
        Layer 2 carries the load on this path.
        """
        from local_deep_research.security.safe_requests import SafeSession

        with SafeSession() as sess:
            req = requests.Request("GET", "http://127.0.0.1:6666/%5C@1.1.1.1")
            prepared = sess.prepare_request(req)
            with pytest.raises(ValueError, match="SSRF|security validation"):
                sess.send(prepared, timeout=2)