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

"""Tests that LLM provider error paths never leak API key bytes into logs
or exception messages reaching callers.

Bundled with this file: a production fix to
``src/local_deep_research/llm/providers/implementations/google.py`` whose
``list_models_for_api`` method previously used ``logger.exception(...)``
to log a ``requests`` exception whose message embedded the full request
URL — and Google's API requires the API key as a ``?key=...`` query
parameter, so the key value was written to every loguru sink.

The tests below pin no-leak behavior across a few representative
providers, with the Google case being the one that previously failed.
"""

import base64
from unittest.mock import Mock, patch
from urllib.parse import quote, quote_plus

import pytest
import requests

# A recognizable sentinel that should never appear in any logged or
# returned text after these tests run. Kept URL-safe so it can be
# embedded literally in test URLs without confusing `urllib.parse`; the
# encoding-matrix helper covers non-URL transformations (base64, repr,
# truncation) that *do* change shape.
_LEAKED_KEY = "sk-leaked-sentinel-DO-NOT-APPEAR-12345"
# An opaque key with NO regex-detectable shape (no sk-/pk- prefix, not a URL
# param, not a Bearer token). Only the literal-redaction pass (self.api_key via
# _secret_attrs) can scrub it — used to prove the literal net for header keys.
_LEAKED_OPAQUE_KEY = "opaqueleakedsentinelDONOTAPPEAR0123456789"


def _all_encodings_of(secret: str) -> list:
    """Return every encoding of *secret* a leak might appear under.

    Used by the leak tests to assert no form of the sentinel reaches the
    log output. New providers / engines that introduce different encoding
    paths (e.g., a SDK that wraps the key in a JWT) should extend this
    helper so the contract scales.
    """
    return [
        secret,
        quote(secret, safe=""),  # %-encoded for ?key=
        quote_plus(secret),  # +-encoded for form-urlencoded
        repr(secret)[1:-1],  # f-string {x!r} leak shape
        base64.b64encode(secret.encode()).decode(),
        secret[:8],  # partial-leak (mask formatters)
    ]


def _stub_rate_tracker() -> Mock:
    """Return a minimal rate_tracker stub for tests that exercise the
    request-layer catch block but not rate limiting.

    Bypasses rate limiting without pulling in the real
    ``AdaptiveRateLimitTracker`` (which would trigger DB imports and a
    settings snapshot). The leak tests want the catch block to run, not
    the rate-limit code path.

    - ``apply_rate_limit`` returns ``0`` so the engine's
      ``time.sleep(self._last_wait_time)`` is a no-op (keeps the test
      fast). Forgetting this would have ``Mock`` return a truthy
      ``Mock`` instance and either sleep garbage or raise.
    - ``enabled = False`` so any code path that gates on
      ``self.rate_tracker.enabled`` sees the stub as disabled.
    - ``record_event`` is a no-op (the real implementation writes to
      a DB; the stub has no DB to write to).
    """
    tracker = Mock()
    tracker.enabled = False
    tracker.apply_rate_limit.return_value = 0
    return tracker


@pytest.fixture
def google_provider_module():
    """Late-import the Google provider so the test's loguru_caplog
    fixture has a chance to enable propagation first.
    """
    from local_deep_research.llm.providers.implementations import google

    return google


class TestGoogleProviderKeyLeakage:
    """The Google provider's ``list_models_for_api`` previously built a
    URL with the API key as a query parameter (Google's then-documented
    requirement). If the upstream request raised with the URL in its
    exception message, ``logger.exception`` would write the key to logs
    verbatim. The fix at
    :file:`src/local_deep_research/llm/providers/implementations/google.py`
    is two-layered:

    1. **Prevention by construction (primary defense, issue #4184):** the
       key is now passed via the ``x-goog-api-key`` header, so the URL
       handed to ``requests`` no longer carries the secret. HTTP
       exception messages embed the URL but never the headers.
    2. **Log-side redaction (defense-in-depth):** the except handler
       still wraps ``str(e)`` with ``redact_secrets(..., api_key)`` and
       uses ``logger.warning`` (no traceback) so any future code path
       that reintroduces the key into a logged string is still scrubbed.

    Both properties are pinned below — the construction-layer test
    fails if a maintainer ever reverts to ``?key=...``; the log-side
    tests fail if the redaction wrap is removed.
    """

    def test_api_key_is_passed_via_header_not_url(self, google_provider_module):
        """Prevention-by-construction: the URL handed to ``safe_get`` must
        not contain the api_key under any name, and the ``x-goog-api-key``
        header must carry it instead. This is the primary defense from
        issue #4184 — reverting to the old ``?key=`` form would make this
        test fail before any exception-handling code even runs.
        """
        import local_deep_research.security as sec_pkg

        captured = {}

        class _Resp:
            status_code = 200

            def json(self):
                return {"models": []}

        def _capture(url, *args, **kwargs):
            captured["url"] = url
            captured["headers"] = kwargs.get("headers") or {}
            return _Resp()

        with patch.object(sec_pkg, "safe_get", side_effect=_capture):
            google_provider_module.GoogleProvider.list_models_for_api(
                api_key=_LEAKED_KEY
            )

        assert "url" in captured, "safe_get was not called"
        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in captured["url"], (
                "API key found in the URL handed to safe_get (as encoding "
                f"{encoding!r}) - The Google provider must pass the key via the"
                " x-goog-api-key header (see issue #4184) so HTTP exception"
                " messages — which embed the URL but not headers — cannot "
                "carry the secret."
            )
        assert captured["headers"].get("x-goog-api-key") == _LEAKED_KEY, (
            "API key must be passed via the x-goog-api-key header. "
            f"Got headers: {captured['headers']!r}"
        )

    def test_no_leak_when_safe_get_raises_with_url_in_message(
        self, loguru_caplog, google_provider_module
    ):
        """Defense-in-depth: even if some upstream exception still embeds
        the key (e.g., a future code path reintroduces it, or an
        intermediate proxy echoes it back), the except handler's
        ``redact_secrets`` wrap must scrub it from the logged message.
        """
        import local_deep_research.security as sec_pkg

        exc = requests.exceptions.ConnectionError(
            "HTTPSConnectionPool(host='generativelanguage.googleapis.com', "
            "port=443): Max retries exceeded with url: "
            f"/v1beta/models?key={_LEAKED_KEY}"
        )

        with loguru_caplog.at_level("DEBUG"):
            with patch.object(sec_pkg, "safe_get", side_effect=exc):
                result = (
                    google_provider_module.GoogleProvider.list_models_for_api(
                        api_key=_LEAKED_KEY
                    )
                )

        assert result == []
        assert _LEAKED_KEY not in loguru_caplog.text, (
            "API key value leaked into logs via the upstream exception "
            "message. The except handler must redact the key before "
            "logging."
        )
        # Sanity: we did log something — proving the test exercised the
        # except branch rather than passing trivially.
        assert "Error fetching Google Gemini models" in loguru_caplog.text

    def test_no_leak_when_safe_get_raises_generic_runtime_error(
        self, loguru_caplog, google_provider_module
    ):
        """Some upstream failures raise a generic exception whose ``str()``
        contains the URL. Redaction must handle that path too.
        """
        import local_deep_research.security as sec_pkg

        exc = RuntimeError(
            f"upstream failure calling /v1beta/models?key={_LEAKED_KEY}"
        )

        with loguru_caplog.at_level("DEBUG"):
            with patch.object(sec_pkg, "safe_get", side_effect=exc):
                google_provider_module.GoogleProvider.list_models_for_api(
                    api_key=_LEAKED_KEY
                )

        assert _LEAKED_KEY not in loguru_caplog.text

    def test_non_200_response_does_not_leak_key(
        self, loguru_caplog, google_provider_module
    ):
        """The status-code branch must also not surface the URL. The
        existing warning at line 88-90 only includes ``response.status_code``
        — verify that contract holds.
        """
        import local_deep_research.security as sec_pkg

        class _Resp:
            status_code = 503
            text = "upstream busy"

            def json(self):
                return {}

        with loguru_caplog.at_level("DEBUG"):
            with patch.object(sec_pkg, "safe_get", return_value=_Resp()):
                result = (
                    google_provider_module.GoogleProvider.list_models_for_api(
                        api_key=_LEAKED_KEY
                    )
                )

        assert result == []
        assert _LEAKED_KEY not in loguru_caplog.text


class TestCredentialStoreKeyLeakage:
    """Pin no-leak behavior in the credential store base class. The class
    is a small wrapper around a dict; verify ``__repr__``,
    ``__str__``, and any exception paths do not expose stored secrets.
    """

    def test_repr_does_not_expose_stored_passwords(self):
        from local_deep_research.scheduler.background import (
            SchedulerCredentialStore,
        )

        store = SchedulerCredentialStore(ttl_hours=1)
        store.store("alice", _LEAKED_KEY)

        # repr / str must not expose the password
        assert _LEAKED_KEY not in repr(store)
        assert _LEAKED_KEY not in str(store)

    def test_clear_entry_does_not_log_store_state(self, loguru_caplog):
        """``clear_entry`` must be completely silent. The implementation
        in ``credential_store_base.py`` does not call ``logger`` at all
        — this test pins that contract so a future
        ``logger.debug(f"store contents: {self._store}")`` (which would
        expose every stored credential) is caught immediately. Exercises
        both the present-key and missing-key paths and asserts not just
        the leaked sentinel but that *no records were emitted at all*.
        """
        from local_deep_research.scheduler.background import (
            SchedulerCredentialStore,
        )

        store = SchedulerCredentialStore(ttl_hours=1)
        store.store("alice", _LEAKED_KEY)
        store.store("bob", "another-stored-secret-87654321")

        with loguru_caplog.at_level("DEBUG"):
            store.clear_entry("never-stored-user")
            store.clear_entry("alice")

        assert not loguru_caplog.records, (
            "clear_entry must be silent. Got log records: "
            f"{[r.getMessage() for r in loguru_caplog.records]}"
        )
        assert _LEAKED_KEY not in loguru_caplog.text


class TestOpenAICompatErrorRedaction:
    """The OpenAI-compat error helper at
    ``src/local_deep_research/error_handling/openai_compat_errors.py``
    runs ``_strip_credentials`` on ``base_url`` and appends ``{exc!s}`` to
    the returned friendly message. Verify that an embedded-credential
    base URL is stripped from the final string.
    """

    def test_friendly_error_strips_credentials_from_base_url(self):
        from local_deep_research.error_handling.openai_compat_errors import (
            friendly_openai_compatible_error,
        )

        # Some users embed API keys in the base URL itself
        embedded_url = f"https://user:{_LEAKED_KEY}@host.example.com/v1"
        exc = RuntimeError("upstream failed")  # exc!s does NOT contain the key

        result = friendly_openai_compatible_error(
            exc,
            provider="lmstudio",
            base_url=embedded_url,
            model="some-model",
        )

        assert _LEAKED_KEY not in result, (
            "_strip_credentials must remove userinfo from base_url before "
            "the URL is embedded in the friendly message"
        )


class TestOpenAIBaseProviderKeyLeakage:
    """``OpenAICompatibleProvider.list_models`` wraps the inner
    ``list_models_for_api`` call. If a subclass override raises (or the
    settings-fetch path raises while the api_key is in scope), the
    upstream exception's ``str()`` may embed the key. The except block at
    ``openai_base.py`` redacts the key from the exception string and uses
    ``logger.warning`` (no traceback) to keep the cause chain off the log.

    These tests use ``loguru_caplog_full`` — the stricter fixture that
    captures the rendered exception block — so a leak that lives only in
    the traceback would be caught. The vanilla ``loguru_caplog`` fixture
    uses ``format='{message}'`` and would false-pass on a traceback leak.
    """

    def test_no_leak_when_inner_call_raises_with_url_in_message(
        self, loguru_caplog_full
    ):
        """A subclass override of ``list_models_for_api`` that constructs
        the auth URL with the key in a query parameter (the Google
        pattern) and then raises a ``requests`` exception embedding that
        URL must not leak the key.
        """
        from local_deep_research.llm.providers.openai_base import (
            OpenAICompatibleProvider,
        )

        exc = requests.exceptions.ConnectionError(
            "HTTPSConnectionPool: Max retries exceeded with url: "
            f"/v1/models?key={_LEAKED_KEY}"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            with patch.object(
                OpenAICompatibleProvider,
                "list_models_for_api",
                side_effect=exc,
            ):
                with patch.object(
                    OpenAICompatibleProvider,
                    "requires_auth_for_models",
                    return_value=False,
                ):
                    with patch(
                        "local_deep_research.config.thread_settings."
                        "get_setting_from_snapshot",
                        return_value=_LEAKED_KEY,
                    ):
                        # Subclass override drives auth-required to True so
                        # api_key flows through the settings path.
                        OpenAICompatibleProvider.requires_auth_for_models = (
                            classmethod(lambda cls: True)
                        )
                        try:
                            result = OpenAICompatibleProvider.list_models()
                        finally:
                            del OpenAICompatibleProvider.requires_auth_for_models

        assert result == []
        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text, (
                f"API key leaked into logs as encoding {encoding!r}. "
                f"The except handler must pass the key to redact_secrets "
                f"and use logger.warning (not logger.exception)."
            )
        # Sanity: we did exercise the except branch.
        assert "Error listing models" in loguru_caplog_full.text

    def test_no_leak_when_inner_call_raises_generic_runtime_error(
        self, loguru_caplog_full
    ):
        """Some upstream failures raise generic exceptions whose ``str()``
        embeds the URL. Redaction must handle that path too — not just
        ``requests`` exception subclasses.
        """
        from local_deep_research.llm.providers.openai_base import (
            OpenAICompatibleProvider,
        )

        exc = RuntimeError(
            f"call to /v1/models?key={_LEAKED_KEY} failed: connection reset"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            with patch.object(
                OpenAICompatibleProvider,
                "list_models_for_api",
                side_effect=exc,
            ):
                with patch(
                    "local_deep_research.config.thread_settings."
                    "get_setting_from_snapshot",
                    return_value=_LEAKED_KEY,
                ):
                    OpenAICompatibleProvider.requires_auth_for_models = (
                        classmethod(lambda cls: True)
                    )
                    try:
                        OpenAICompatibleProvider.list_models()
                    finally:
                        del OpenAICompatibleProvider.requires_auth_for_models

        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text, (
                f"API key leaked as encoding {encoding!r}"
            )


class TestOpenAIEmbeddingProviderKeyLeakage:
    """``OpenAIEmbeddingsProvider.get_available_models`` calls the OpenAI
    SDK's ``client.models.list()`` with the api_key in scope. If the SDK
    raises (e.g., the base_url points at a misconfigured proxy that
    echoes the bearer token in its error body), the except handler at
    ``embeddings/providers/implementations/openai.py`` must redact the
    key before logging.
    """

    def test_no_leak_when_models_list_raises_with_key_in_message(
        self, loguru_caplog_full
    ):
        from local_deep_research.embeddings.providers.implementations.openai import (
            OpenAIEmbeddingsProvider,
        )

        def _settings_lookup(key, default=None, settings_snapshot=None):
            if key == "embeddings.openai.api_key":
                return _LEAKED_KEY
            if key == "embeddings.openai.base_url":
                return None
            return default

        exc = RuntimeError(
            f"upstream proxy echoed Authorization: Bearer {_LEAKED_KEY}"
        )

        class _Models:
            def list(self_inner):
                raise exc

        class _FakeOpenAI:
            def __init__(self, **kwargs):
                self.models = _Models()

        with loguru_caplog_full.at_level("DEBUG"):
            with patch(
                "local_deep_research.embeddings.providers.implementations."
                "openai.get_setting_from_snapshot",
                side_effect=_settings_lookup,
            ):
                with patch("openai.OpenAI", _FakeOpenAI):
                    result = OpenAIEmbeddingsProvider.get_available_models()

        assert result == []
        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text, (
                f"API key leaked as encoding {encoding!r}. The except "
                f"handler at openai.py:218 must redact via "
                f"redact_secrets(str(e), api_key) and use logger.warning."
            )
        # Sanity: the except branch ran.
        assert (
            "Error fetching OpenAI embedding models" in loguru_caplog_full.text
        )


class TestSearchEngineKeyLeakage:
    """Pin no-leak behavior across representative search engines whose
    HTTP-call error paths were wrapped in #4131. The base-class fix in
    ``search_engine_base.py`` covers the catch-all path, but engines
    that catch+swallow inside ``_get_previews`` need their own wrap —
    these tests cover those direct call sites.
    """

    def test_tavily_get_previews_does_not_leak_on_request_exception(
        self, loguru_caplog_full
    ):
        """Tavily's ``_get_previews`` catches
        ``requests.exceptions.RequestException`` and logs it. The except
        block must redact ``self.api_key`` from the rendered exception
        text. We drive this by patching ``safe_post`` to raise an
        exception whose ``str()`` embeds the sentinel in the URL.

        Tavily is the representative engine for this contract; the
        base-class fix in ``search_engine_base.py`` covers the
        catch-all path for all subclasses (see ``run()``). Other
        engines have engine-specific attribute requirements that make
        a parametrized test brittle without a fuller fixture suite —
        tracked as follow-up to #4131.
        """
        from local_deep_research.web_search_engines.engines import (
            search_engine_tavily as mod,
        )

        EngineCls = mod.TavilySearchEngine
        engine_class = "TavilySearchEngine"
        api_attr = "api_key"

        # Build an instance without going through normal __init__ — the
        # engines have settings-dependent __init__ paths that complicate
        # test setup. Provide every attribute referenced before safe_post
        # so the test actually reaches the leak-vector code path.
        engine = EngineCls.__new__(EngineCls)
        setattr(engine, api_attr, _LEAKED_KEY)
        engine.max_results = 10
        engine.search_depth = "basic"
        engine.include_full_content = False
        engine.include_domains = []
        engine.exclude_domains = []
        engine.base_url = "https://api.example.com"
        engine.engine_type = "test_engine"
        engine._search_results = None
        # Bypass the rate-limit-tracker dependency — see _stub_rate_tracker.
        # Tavily/Exa each call ``self.rate_tracker.apply_rate_limit(...)``
        # before the HTTP request.
        engine.rate_tracker = _stub_rate_tracker()
        # ``_raise_if_rate_limit`` is a base-class method that inspects
        # the response status code; with a non-HTTP exception passed in,
        # it must not itself raise. Stub it to a no-op for safety.
        engine._raise_if_rate_limit = lambda *a, **k: None

        exc = requests.exceptions.ConnectionError(
            f"HTTPSConnectionPool: Max retries exceeded with url: "
            f"/v1/search?api_key={_LEAKED_KEY}"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            with patch.object(mod, "safe_post", side_effect=exc, create=True):
                try:
                    engine._get_previews("test query")
                except Exception:
                    pass

        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text, (
                f"{engine_class} leaked api_key as encoding {encoding!r}. "
                f"The except block must call "
                f"redact_secrets(str(e), self.{api_attr}) before logging."
            )
        # Sanity: the except branch must have actually run — otherwise the
        # leak assertion passes trivially.
        assert "Error getting Tavily" in loguru_caplog_full.text, (
            f"{engine_class} test did not exercise the except branch. "
            f"Captured logs: {loguru_caplog_full.text!r}"
        )

    def test_nasa_ads_get_previews_does_not_leak_on_request_exception(
        self, loguru_caplog_full
    ):
        """NASA ADS sends its key as an ``Authorization: Bearer`` header
        (``self.headers``) and its ``_get_previews`` catch-all logged the
        exception unredacted. This engine was *not* in the original #4131
        engine list — it is covered by the #4131 follow-up. The except
        block must redact ``self.api_key`` from the rendered exception
        text. We drive this by patching ``safe_get`` to raise an exception
        whose ``str()`` embeds the sentinel in the URL.
        """
        from local_deep_research.web_search_engines.engines import (
            search_engine_nasa_ads as mod,
        )

        EngineCls = mod.NasaAdsSearchEngine

        engine = EngineCls.__new__(EngineCls)
        engine.api_key = _LEAKED_KEY
        engine.headers = {"Authorization": f"Bearer {_LEAKED_KEY}"}
        engine.api_base = "https://api.adsabs.harvard.edu/v1"
        engine.max_results = 10
        engine.sort_by = "relevance"
        engine.from_publication_date = None
        engine.min_citations = 0
        engine.include_arxiv = True
        engine.engine_type = "test_engine"
        engine.rate_tracker = _stub_rate_tracker()

        exc = requests.exceptions.ConnectionError(
            f"HTTPSConnectionPool: Max retries exceeded with url: "
            f"/v1/search/query?token={_LEAKED_KEY}"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            with patch.object(mod, "safe_get", side_effect=exc, create=True):
                try:
                    engine._get_previews("test query")
                except Exception:
                    pass

        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text, (
                "NASAADSSearchEngine leaked api_key as encoding "
                f"{encoding!r}. The except block must call "
                "redact_secrets(str(e), self.api_key) before logging."
            )
        # Sanity: the except branch must have actually run.
        assert "Error searching NASA ADS" in loguru_caplog_full.text, (
            "NASA ADS test did not exercise the except branch. "
            f"Captured logs: {loguru_caplog_full.text!r}"
        )

    def test_exa_get_previews_does_not_leak_header_key(
        self, loguru_caplog_full
    ):
        """Exa sends its key in the ``x-api-key`` header. After the
        _scrub_error centralization, the catch block relies on the literal
        pass (self.api_key via _secret_attrs) — the regex cannot scrub a raw
        header value. Use an OPAQUE key embedded header-style so only the
        literal pass can redact it; this pins that exa keeps its net.
        """
        from local_deep_research.web_search_engines.engines import (
            search_engine_exa as mod,
        )

        engine = mod.ExaSearchEngine.__new__(mod.ExaSearchEngine)
        engine.api_key = _LEAKED_OPAQUE_KEY
        engine.max_results = 10
        engine.search_type = "auto"
        engine.include_domains = []
        engine.exclude_domains = []
        engine.start_published_date = None
        engine.end_published_date = None
        engine.category = None
        engine.include_full_content = False
        engine.base_url = "https://api.exa.ai"
        engine.engine_type = "test_engine"
        engine.rate_tracker = _stub_rate_tracker()
        engine._raise_if_rate_limit = lambda *a, **k: None

        exc = requests.exceptions.ConnectionError(
            f"request failed; sent x-api-key: {_LEAKED_OPAQUE_KEY}"
        )
        with loguru_caplog_full.at_level("DEBUG"):
            with patch.object(mod, "safe_post", side_effect=exc, create=True):
                try:
                    engine._get_previews("test query")
                except Exception:
                    pass

        for encoding in _all_encodings_of(_LEAKED_OPAQUE_KEY):
            assert encoding not in loguru_caplog_full.text, (
                f"ExaSearchEngine leaked header api_key as {encoding!r}."
            )
        assert "Error getting Exa results" in loguru_caplog_full.text

    def test_serper_get_previews_does_not_leak_header_key(
        self, loguru_caplog_full
    ):
        """Serper sends its key in the ``X-API-KEY`` header — same literal-pass
        contract as exa. Opaque key embedded header-style."""
        from local_deep_research.web_search_engines.engines import (
            search_engine_serper as mod,
        )

        engine = mod.SerperSearchEngine.__new__(mod.SerperSearchEngine)
        engine.api_key = _LEAKED_OPAQUE_KEY
        engine.max_results = 10
        engine.region = "us"
        engine.search_language = "en"
        engine.time_period = None
        engine.base_url = "https://google.serper.dev/search"
        engine.engine_type = "test_engine"
        engine.rate_tracker = _stub_rate_tracker()
        engine._raise_if_rate_limit = lambda *a, **k: None

        exc = requests.exceptions.ConnectionError(
            f"request failed; sent X-API-KEY: {_LEAKED_OPAQUE_KEY}"
        )
        with loguru_caplog_full.at_level("DEBUG"):
            with patch.object(mod, "safe_post", side_effect=exc, create=True):
                try:
                    engine._get_previews("test query")
                except Exception:
                    pass

        for encoding in _all_encodings_of(_LEAKED_OPAQUE_KEY):
            assert encoding not in loguru_caplog_full.text, (
                f"SerperSearchEngine leaked header api_key as {encoding!r}."
            )
        assert "Error getting Serper API results" in loguru_caplog_full.text


class TestSearchEngineParamsKeyLeakage:
    """Pin no-leak behavior for search engines that pass the API key via
    the ``params=`` dict to ``safe_get``. The underlying ``requests``
    library assembles ``params`` into the request URL before the network
    call; when ``requests`` raises (``ConnectionError``, ``Timeout``,
    etc.), the exception's ``__str__()`` includes that assembled URL —
    and therefore the key.

    Each engine catches the exception in its own try/except and is
    protected today by either ``_sanitize_error_message()``
    (regex-based, defined on the base class) and/or
    ``redact_secrets()`` (literal-value substitution). This contract is
    "safe by convention" — every catch site must remember the wrap. A
    future refactor that consolidates error handling into a base method
    without one of those calls would silently leak. The tests below
    pin the no-leak property so such a regression flips the test.

    Note on Guardian specifically: its key parameter is ``api-key`` (with
    a dash), which the base-class regex at ``search_engine_base.py:984``
    does NOT match (the regex covers ``api_key|apikey|key|token|secret``).
    Guardian therefore relies entirely on ``redact_secrets()`` for the
    URL-leak path — making it the most fragile of the four and the most
    important to pin.
    """

    @staticmethod
    def _mojeek_engine():
        from local_deep_research.web_search_engines.engines import (
            search_engine_mojeek as mod,
        )

        engine = mod.MojeekSearchEngine.__new__(mod.MojeekSearchEngine)
        engine.api_key = _LEAKED_KEY
        engine.search_url = "https://api.mojeek.com/search"
        engine.max_results = 10
        engine.safe_search = True
        engine.language = None
        engine.region = None
        return (
            mod,
            engine,
            "_get_search_results",
            "/search?api_key=",
            "Error when searching using Mojeek",
        )

    @staticmethod
    def _scaleserp_engine():
        from local_deep_research.web_search_engines.engines import (
            search_engine_scaleserp as mod,
        )

        engine = mod.ScaleSerpSearchEngine.__new__(mod.ScaleSerpSearchEngine)
        engine.api_key = _LEAKED_KEY
        engine.base_url = "https://api.scaleserp.com/search"
        engine.max_results = 10
        engine.location = "United States"
        engine.language = "en"
        engine.device = "desktop"
        engine.safe_search = False
        engine.enable_cache = False
        engine.engine_type = "test_engine"
        engine._knowledge_graph = None
        engine._search_results = None
        engine.rate_tracker = _stub_rate_tracker()
        engine._raise_if_rate_limit = lambda *a, **k: None
        return (
            mod,
            engine,
            "_get_previews",
            "/search?api_key=",
            "Error getting ScaleSerp API results",
        )

    @staticmethod
    def _google_pse_engine():
        from local_deep_research.web_search_engines.engines import (
            search_engine_google_pse as mod,
        )

        engine = mod.GooglePSESearchEngine.__new__(mod.GooglePSESearchEngine)
        engine.api_key = _LEAKED_KEY
        engine.search_engine_id = "test-cx"
        engine.max_results = 10
        engine.safe = "off"
        engine.language = "en"
        engine.region = "us"
        engine.max_retries = 1  # single attempt — no real sleep, no real retry
        engine.retry_delay = 0
        engine.engine_type = "test_engine"
        engine.rate_tracker = _stub_rate_tracker()
        # _make_request will re-raise as RequestException after max_retries
        # — callers must catch that. Return the method name to invoke.
        # The retry loop logs each attempt before re-raising, so the log
        # marker is the per-attempt warning from line ~266.
        return (
            mod,
            engine,
            "_make_request",
            "/customsearch/v1?key=",
            "Request error on attempt",
        )

    @staticmethod
    def _guardian_engine():
        from local_deep_research.web_search_engines.engines import (
            search_engine_guardian as mod,
        )

        engine = mod.GuardianSearchEngine.__new__(mod.GuardianSearchEngine)
        engine.api_key = _LEAKED_KEY
        engine.api_url = "https://content.guardianapis.com/search"
        engine.max_results = 10
        engine.from_date = "2024-01-01"
        engine.to_date = "2024-12-31"
        engine.order_by = "relevance"
        engine.section = None
        engine.engine_type = "test_engine"
        engine.rate_tracker = _stub_rate_tracker()
        engine._raise_if_rate_limit = lambda *a, **k: None
        return (
            mod,
            engine,
            "_get_all_data",
            "/search?api-key=",
            "Error getting data from The Guardian API",
        )

    @staticmethod
    def _pubmed_engine():
        from local_deep_research.web_search_engines.engines import (
            search_engine_pubmed as mod,
        )

        engine = mod.PubMedSearchEngine.__new__(mod.PubMedSearchEngine)
        engine.api_key = _LEAKED_KEY
        engine.search_url = (
            "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esearch.fcgi"
        )
        engine.engine_type = "test_engine"
        engine.rate_tracker = _stub_rate_tracker()
        return (
            mod,
            engine,
            "_get_result_count",
            "?api_key=",
            "Error getting result count",
        )

    @pytest.mark.parametrize(
        "engine_factory_name",
        [
            "_mojeek_engine",
            "_scaleserp_engine",
            "_google_pse_engine",
            "_guardian_engine",
            "_pubmed_engine",
        ],
    )
    def test_no_leak_when_safe_get_raises_with_assembled_url(
        self, loguru_caplog_full, engine_factory_name
    ):
        """When ``safe_get`` raises a ``ConnectionError`` whose message
        embeds the assembled URL (base + query string with the key from
        ``params=``), the engine's catch block must scrub the key before
        logging.

        Mutation check: temporarily remove the per-engine
        ``redact_secrets()`` / ``_sanitize_error_message()`` call from
        the catch block and the relevant parametrized case will fail.
        """
        factory = getattr(self, engine_factory_name)
        mod, engine, method_name, url_path_with_key, marker = factory()

        # Simulate the exception that requests raises when the network
        # call fails — its message embeds the *assembled* URL with the
        # query string built from `params=`.
        exc = requests.exceptions.ConnectionError(
            f"HTTPSConnectionPool: Max retries exceeded with url: "
            f"{url_path_with_key}{_LEAKED_KEY}&q=test"
        )

        # No create=True: every engine under test imports safe_get at
        # module level, so a typo'd patch target (or a future refactor
        # that drops the import) must fail loudly rather than silently
        # no-op into a passing test.
        with loguru_caplog_full.at_level("DEBUG"):
            with patch.object(
                mod, "safe_get", side_effect=exc
            ) as mock_safe_get:
                # Only swallow the request-layer exception we injected.
                # Google PSE's _make_request re-raises as
                # RequestException after retries; the other three engines
                # catch internally and return []. AttributeError /
                # TypeError from a bad test stub must propagate so the
                # test fails loudly instead of silently skipping the
                # logging/redaction path.
                try:
                    getattr(engine, method_name)("test query")
                except requests.exceptions.RequestException:
                    pass

        # Sanity: the SUT must have actually called safe_get.
        assert mock_safe_get.called, (
            f"{engine_factory_name}: safe_get was not invoked — the "
            f"redaction path under test never ran. Check that "
            f"{mod.__name__} still imports safe_get at module level."
        )
        # Sanity: the SUT must have actually logged the catch-block warning.
        # Without this, the leak assertion above passes vacuously if the
        # catch block is removed or refactored to silently re-raise.
        assert marker in loguru_caplog_full.text, (
            f"{engine_factory_name}: catch-block log marker {marker!r} not "
            f"emitted — the redaction path under test never ran. Check that "
            f"{mod.__name__} still wraps the safe_get call with a "
            f"logger.warning(...) that includes the redacted message. "
            f"Captured logs: {loguru_caplog_full.text!r}"
        )

        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text, (
                f"{engine_factory_name}: api_key leaked as encoding "
                f"{encoding!r}. The catch block must scrub the key via "
                f"redact_secrets(str(e), self.api_key) and/or "
                f"self._sanitize_error_message(str(e)) before logging."
            )