learningcircuit--local-deep-research/tests/llm/test_provider_base_url_ssrf.py

"""SSRF guard for operator-configured LLM provider base_url.

Even though the SSRF validator (``ssrf_validator.validate_url``) gates
outbound HTTP through ``safe_requests``, LangChain provider SDKs
(``ChatOpenAI``, ``ChatOllama``) use their own internal ``httpx`` client
that bypasses ``safe_requests`` entirely. An authenticated user with
permission to edit ``llm.<provider>.url`` could otherwise route every
inference call at internal services / cloud-credential endpoints.

The fix gates ``base_url`` through ``assert_base_url_safe`` immediately
after ``normalize_url`` and before the SDK constructor. These tests lock
in:

1. Every IP in ``ALWAYS_BLOCKED_METADATA_IPS`` raises ValueError.
2. Legitimate localhost / RFC1918 destinations (Ollama / LM Studio /
   llama.cpp on private networks) keep working.
3. Validation runs BEFORE the SDK constructor (a future refactor that
   moved it after would silently regress).
4. Edge cases like empty / None base_url are rejected.
"""

from unittest.mock import MagicMock, Mock

import pytest


@pytest.mark.parametrize(
    "ip",
    [
        "169.254.169.254",  # AWS IMDS / Azure / OCI / DigitalOcean
        "169.254.170.2",  # AWS ECS task metadata v3
        "169.254.170.23",  # AWS ECS task metadata v4
        "169.254.0.23",  # Tencent Cloud
        "100.100.100.200",  # AlibabaCloud
    ],
)
def test_assert_base_url_safe_rejects_all_metadata_ips(ip):
    """Every cloud-metadata IP in ALWAYS_BLOCKED_METADATA_IPS must
    raise — this is the load-bearing security property."""
    from local_deep_research.security.ssrf_validator import (
        assert_base_url_safe,
    )

    with pytest.raises(ValueError, match="failed SSRF validation"):
        assert_base_url_safe(f"http://{ip}/", setting_key="llm.test.url")


def test_assert_base_url_safe_accepts_localhost():
    """Localhost is the typical Ollama / LM Studio / llama.cpp host."""
    from local_deep_research.security.ssrf_validator import (
        assert_base_url_safe,
    )

    assert (
        assert_base_url_safe(
            "http://localhost:11434/", setting_key="llm.ollama.url"
        )
        == "http://localhost:11434/"
    )


def test_assert_base_url_safe_accepts_rfc1918():
    """Docker / private-network deployments need RFC1918."""
    from local_deep_research.security.ssrf_validator import (
        assert_base_url_safe,
    )

    assert (
        assert_base_url_safe(
            "http://192.168.1.10:11434/", setting_key="llm.ollama.url"
        )
        == "http://192.168.1.10:11434/"
    )


def test_assert_base_url_safe_rejects_empty_string():
    from local_deep_research.security.ssrf_validator import (
        assert_base_url_safe,
    )

    with pytest.raises(ValueError, match="failed SSRF validation"):
        assert_base_url_safe("", setting_key="llm.x.url")


def test_assert_base_url_safe_rejects_none():
    """isinstance check in validate_url means None returns False, which
    means assert_base_url_safe raises rather than crashing on .strip()."""
    from local_deep_research.security.ssrf_validator import (
        assert_base_url_safe,
    )

    with pytest.raises(ValueError, match="failed SSRF validation"):
        assert_base_url_safe(None, setting_key="llm.x.url")


def test_assert_base_url_safe_error_message_includes_setting_key():
    """Operators see the error and need to know which setting to fix."""
    from local_deep_research.security.ssrf_validator import (
        assert_base_url_safe,
    )

    with pytest.raises(ValueError, match=r"llm\.lmstudio\.url"):
        assert_base_url_safe(
            "http://169.254.169.254/", setting_key="llm.lmstudio.url"
        )


# ---------------------------------------------------------------------
# Per-provider integration tests
# ---------------------------------------------------------------------


class TestOllamaBaseUrlSsrf:
    """OllamaProvider gates base_url at three sites: list_models_for_api,
    create_llm, is_available. Each behaves differently on rejection."""

    def test_create_llm_rejects_aws_metadata(self):
        """create_llm propagates ValueError so the research-query caller
        sees a clear config error instead of silent SSRF."""
        from local_deep_research.llm.providers.implementations.ollama import (
            OllamaProvider,
        )

        snapshot = {"llm.ollama.url": "http://169.254.169.254/"}
        with pytest.raises(ValueError, match="failed SSRF validation"):
            OllamaProvider.create_llm(
                model_name="llama3", settings_snapshot=snapshot
            )

    def test_create_llm_validate_runs_before_chatollama_constructor(
        self, monkeypatch
    ):
        """A future refactor that moved the validate call below the SDK
        constructor would silently regress (constructor is mocked in
        tests). Make assert_base_url_safe raise; assert ChatOllama mock
        has call_count == 0.

        Patch the SYMBOL inside ollama.py — ``from langchain_ollama
        import ChatOllama`` binds the name locally at import time;
        patching ``langchain_ollama.ChatOllama`` would not affect that
        local binding.
        """
        from local_deep_research.llm.providers.implementations.ollama import (
            OllamaProvider,
        )

        mock_ctor = MagicMock(return_value=Mock())
        monkeypatch.setattr(
            "local_deep_research.llm.providers.implementations.ollama.ChatOllama",
            mock_ctor,
        )
        monkeypatch.setattr(
            "local_deep_research.llm.providers.implementations.ollama.assert_base_url_safe",
            Mock(side_effect=ValueError("simulated SSRF rejection")),
        )

        with pytest.raises(ValueError):
            OllamaProvider.create_llm(
                model_name="llama3",
                settings_snapshot={"llm.ollama.url": "http://example.com/"},
            )
        assert mock_ctor.call_count == 0, (
            "ChatOllama was instantiated despite SSRF guard raising — "
            "ordering regression"
        )

    def test_is_available_returns_false_on_bad_base_url(self):
        """is_available swallows ValueError and returns False so the
        model-list UI degrades gracefully."""
        from local_deep_research.llm.providers.implementations.ollama import (
            OllamaProvider,
        )

        snapshot = {"llm.ollama.url": "http://169.254.169.254/"}
        assert OllamaProvider.is_available(settings_snapshot=snapshot) is False

    def test_list_models_for_api_returns_empty_on_bad_base_url(self):
        """list_models_for_api swallows ValueError and returns []."""
        from local_deep_research.llm.providers.implementations.ollama import (
            OllamaProvider,
        )

        result = OllamaProvider.list_models_for_api(
            api_key="test",
            base_url="http://169.254.169.254/",
        )
        assert result == []


class TestOpenAICompatibleBaseUrlSsrf:
    """OpenAICompatibleProvider gates base_url in create_llm and
    _create_llm_instance, but only when ``cls.url_setting`` is set
    (subclasses with hardcoded default_base_url skip validation)."""

    def test_subclass_with_url_setting_rejects_metadata(self):
        """A subclass with cls.url_setting set (e.g. CustomOpenAIEndpoint
        which uses llm.openai_endpoint.url) must reject metadata IPs."""
        from local_deep_research.llm.providers.openai_base import (
            OpenAICompatibleProvider,
        )

        # Create a temporary subclass to exercise the cls.url_setting gate
        class _TestProvider(OpenAICompatibleProvider):
            provider_name = "TestProvider"
            url_setting = "llm.test.url"
            api_key_setting = None
            default_base_url = "http://169.254.169.254/"

        with pytest.raises(ValueError, match="failed SSRF validation"):
            _TestProvider.create_llm(model_name="test")

    def test_pure_provider_skips_validation(self, monkeypatch):
        """Providers with ``url_setting = None`` (Anthropic, OpenAI,
        OpenRouter, xAI, IONOS) must NOT invoke assert_base_url_safe.
        Their default_base_url is hardcoded; there's no operator URL to
        attack, so validation would be wasted work."""
        from local_deep_research.llm.providers.openai_base import (
            OpenAICompatibleProvider,
        )

        mock_guard = Mock(return_value="passthrough")
        monkeypatch.setattr(
            "local_deep_research.llm.providers.openai_base.assert_base_url_safe",
            mock_guard,
        )
        monkeypatch.setattr(
            "local_deep_research.llm.providers.openai_base.ChatOpenAI",
            Mock(),
        )

        class _PureProvider(OpenAICompatibleProvider):
            provider_name = "PureProvider"
            url_setting = None
            api_key_setting = None
            default_base_url = "https://api.example.com/v1"

        _PureProvider.create_llm(model_name="test-model")
        assert mock_guard.call_count == 0, (
            "Pure-default provider invoked SSRF guard unexpectedly"
        )

    def test_validate_runs_before_chatopenai_constructor(self, monkeypatch):
        """Same ordering invariant as the Ollama test, for the OpenAI
        compat path. Patch the local binding of ChatOpenAI inside
        openai_base.py."""
        from local_deep_research.llm.providers.openai_base import (
            OpenAICompatibleProvider,
        )

        mock_ctor = MagicMock(return_value=Mock())
        monkeypatch.setattr(
            "local_deep_research.llm.providers.openai_base.ChatOpenAI",
            mock_ctor,
        )
        monkeypatch.setattr(
            "local_deep_research.llm.providers.openai_base.assert_base_url_safe",
            Mock(side_effect=ValueError("simulated SSRF rejection")),
        )

        class _GuardedProvider(OpenAICompatibleProvider):
            provider_name = "GuardedProvider"
            url_setting = "llm.guarded.url"
            api_key_setting = None
            default_base_url = "https://api.example.com/v1"

        with pytest.raises(ValueError):
            _GuardedProvider.create_llm(model_name="test-model")
        assert mock_ctor.call_count == 0, (
            "ChatOpenAI was instantiated despite SSRF guard raising — "
            "ordering regression"
        )


class TestOpenAICompatibleListModelsSsrf:
    """OpenAICompatibleProvider.list_models_for_api (used by
    CustomOpenAIEndpoint / LMStudio / LlamaCpp) gates base_url through the
    SSRF guard when ``cls.url_setting`` is set. On rejection it must
    degrade gracefully and return [] (model-listing should NOT 500)."""

    def test_list_models_for_api_returns_empty_on_metadata_url(
        self, monkeypatch, loguru_caplog
    ):
        """A subclass with cls.url_setting set must return [] (and log a
        warning) when base_url is a blocked metadata endpoint, without ever
        instantiating the OpenAI SDK client."""
        from local_deep_research.llm.providers.openai_base import (
            OpenAICompatibleProvider,
        )

        # If the guard failed to short-circuit, the SDK client would be
        # constructed; make that an explicit failure so the test can't pass
        # for the wrong reason.
        import openai

        def _boom(*args, **kwargs):
            raise AssertionError(
                "OpenAI client constructed despite blocked base_url"
            )

        monkeypatch.setattr(openai, "OpenAI", _boom)

        class _TestProvider(OpenAICompatibleProvider):
            provider_name = "TestListProvider"
            url_setting = "llm.test.url"
            api_key_setting = None
            default_base_url = "https://api.example.com/v1"

            @classmethod
            def requires_auth_for_models(cls):
                return False

        with loguru_caplog.at_level("WARNING"):
            result = _TestProvider.list_models_for_api(
                api_key="test-key",
                base_url="http://169.254.169.254/",
            )

        assert result == []
        assert "SSRF validation" in loguru_caplog.text, (
            "expected a warning naming SSRF validation"
        )

    def test_list_models_for_api_allows_localhost(self, monkeypatch):
        """Legitimate local endpoints (LM Studio / llama.cpp on localhost)
        must pass the guard and reach the SDK client."""
        from local_deep_research.llm.providers.openai_base import (
            OpenAICompatibleProvider,
        )

        captured = {}

        class _FakeModels:
            def list(self):
                return Mock(data=[])

        class _FakeClient:
            def __init__(self, api_key=None, base_url=None):
                captured["base_url"] = base_url
                self.models = _FakeModels()

        import openai

        monkeypatch.setattr(openai, "OpenAI", _FakeClient)

        class _TestProvider(OpenAICompatibleProvider):
            provider_name = "TestListProvider"
            url_setting = "llm.test.url"
            api_key_setting = None
            default_base_url = "https://api.example.com/v1"

            @classmethod
            def requires_auth_for_models(cls):
                return False

        result = _TestProvider.list_models_for_api(
            api_key="test-key",
            base_url="http://localhost:1234/v1",
        )
        assert result == []
        assert captured["base_url"] == "http://localhost:1234/v1", (
            "localhost base_url should reach the SDK client unchanged"
        )


class TestOpenAIProviderApiBaseSsrf:
    """OpenAIProvider overrides create_llm without calling super and has
    url_setting = None, so the base-class guard never runs. It gates
    llm.openai.api_base directly and lets ValueError propagate (fail-fast,
    matching ollama.create_llm)."""

    def test_create_llm_rejects_metadata_api_base(self, monkeypatch):
        """A blocked llm.openai.api_base must raise ValueError before the
        ChatOpenAI constructor runs."""
        from local_deep_research.llm.providers.implementations.openai import (
            OpenAIProvider,
        )

        mock_ctor = MagicMock(return_value=Mock())
        monkeypatch.setattr(
            "local_deep_research.llm.providers.implementations.openai.ChatOpenAI",
            mock_ctor,
        )

        snapshot = {
            "llm.openai.api_key": "sk-test-key",
            "llm.openai.api_base": "http://169.254.169.254/v1",
        }
        with pytest.raises(ValueError, match="failed SSRF validation"):
            OpenAIProvider.create_llm(
                model_name="gpt-4o-mini", settings_snapshot=snapshot
            )
        assert mock_ctor.call_count == 0, (
            "ChatOpenAI was instantiated despite SSRF guard raising"
        )

    def test_create_llm_allows_localhost_api_base(self, monkeypatch):
        """A legitimate local api_base (proxy / gateway on localhost) must
        pass the guard and reach the ChatOpenAI constructor."""
        from local_deep_research.llm.providers.implementations.openai import (
            OpenAIProvider,
        )

        captured = {}

        def _fake_ctor(**kwargs):
            captured.update(kwargs)
            return Mock()

        monkeypatch.setattr(
            "local_deep_research.llm.providers.implementations.openai.ChatOpenAI",
            _fake_ctor,
        )

        snapshot = {
            "llm.openai.api_key": "sk-test-key",
            "llm.openai.api_base": "http://localhost:8080/v1",
        }
        OpenAIProvider.create_llm(
            model_name="gpt-4o-mini", settings_snapshot=snapshot
        )
        assert captured["openai_api_base"] == "http://localhost:8080/v1", (
            "localhost api_base should pass the guard and reach ChatOpenAI"
        )


class TestPublicApi:
    """assert_base_url_safe and redact_url_for_log are exported from
    security/__init__.py for downstream importers."""

    def test_assert_base_url_safe_re_exported(self):
        from local_deep_research.security import assert_base_url_safe

        assert callable(assert_base_url_safe)

    def test_redact_url_for_log_re_exported(self):
        from local_deep_research.security import redact_url_for_log

        assert callable(redact_url_for_log)


class TestCustomAnthropicEndpointSsrf:
    """CustomAnthropicEndpointProvider gates its operator-configured
    base_url (``llm.anthropic_endpoint.url``) through the SSRF guard in the
    inherited AnthropicProvider.create_llm, and in list_models_for_api. The
    SDK is ChatAnthropic / the anthropic SDK, both of which use their own
    httpx client that bypasses safe_requests — so the guard is load-bearing."""

    def test_create_llm_rejects_aws_metadata(self):
        from local_deep_research.llm.providers.implementations.custom_anthropic_endpoint import (
            CustomAnthropicEndpointProvider,
        )

        snapshot = {"llm.anthropic_endpoint.url": "http://169.254.169.254/"}
        with pytest.raises(ValueError, match="failed SSRF validation"):
            CustomAnthropicEndpointProvider.create_llm(
                model_name="claude-3-5-sonnet", settings_snapshot=snapshot
            )

    def test_create_llm_validate_runs_before_chatanthropic_constructor(
        self, monkeypatch
    ):
        """Make assert_base_url_safe raise; ChatAnthropic must never be
        instantiated. Both names are bound in anthropic.py (the parent's
        create_llm), so patch them there."""
        from local_deep_research.llm.providers.implementations.custom_anthropic_endpoint import (
            CustomAnthropicEndpointProvider,
        )

        mock_ctor = MagicMock(return_value=Mock())
        monkeypatch.setattr(
            "local_deep_research.llm.providers.implementations.anthropic.ChatAnthropic",
            mock_ctor,
        )
        monkeypatch.setattr(
            "local_deep_research.llm.providers.implementations.anthropic.assert_base_url_safe",
            Mock(side_effect=ValueError("simulated SSRF rejection")),
        )

        with pytest.raises(ValueError):
            CustomAnthropicEndpointProvider.create_llm(
                model_name="claude-3-5-sonnet",
                settings_snapshot={
                    "llm.anthropic_endpoint.url": "http://example.com/"
                },
            )
        assert mock_ctor.call_count == 0, (
            "ChatAnthropic was instantiated despite SSRF guard raising — "
            "ordering regression"
        )

    def test_list_models_for_api_returns_empty_on_metadata_url(self):
        from local_deep_research.llm.providers.implementations.custom_anthropic_endpoint import (
            CustomAnthropicEndpointProvider,
        )

        result = CustomAnthropicEndpointProvider.list_models_for_api(
            api_key="test", base_url="http://169.254.169.254/"
        )
        assert result == []