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

"""Regression tests for in-scope API-key redaction in factory exception logs.

A third-party review of issue #4183 found that ``create_search_engine`` and
``_create_full_search_wrapper`` called ``redact_secrets(str(e))`` with NO
secret arguments. ``redact_secrets`` does literal-substring replacement, so
with an empty secret list it's a no-op: an exception message echoing the
API key (e.g. a constructor validation error) survived into logs verbatim.

The fix passes the in-scope key(s) — ``api_key`` in ``create_search_engine``,
and ``api_key`` / ``serpapi_api_key`` from ``wrapper_params`` in
``_create_full_search_wrapper`` — and wraps with ``sanitize_error_message``
for the shape-based pass too.

These tests use a credential with NO distinctive prefix (not ``sk-``, not
``AIza``, etc.) so that ``sanitize_error_message``'s regexes won't catch it
on shape alone — proving the literal-substring pass is what closes the leak.

The factory logs these failures via the diagnose-gated
``security.secure_logging`` wrapper at ERROR level; the levelname
assertions below guard against a re-downgrade to WARNING.
"""

from unittest.mock import Mock, patch

import pytest

from local_deep_research.web_search_engines.search_engine_factory import (
    _create_full_search_wrapper,
)

FACTORY_MODULE = "local_deep_research.web_search_engines.search_engine_factory"

# 24 chars, no sk-/pk-/AIza/ghp_ prefix — survives sanitize_error_message,
# so only the literal redact_secrets(..., api_key) pass catches it.
_BARE_KEY = "test-brave-key-XYZ789"


class _RaisingWrapperBrave:
    """Mock full-search wrapper whose init echoes the brave api_key."""

    def __init__(self, api_key=None, **kwargs):
        raise ValueError(f"Brave wrapper rejected key: {api_key}")


class _RaisingWrapperSerpAPI:
    """Mock full-search wrapper whose init echoes the serpapi key."""

    def __init__(self, serpapi_api_key=None, **kwargs):
        raise ValueError(f"SerpAPI wrapper rejected key: {serpapi_api_key}")


@pytest.fixture(autouse=True)
def _bypass_engine_pdp():
    """Bypass the egress PEP so mock engine names aren't rejected.

    The PEP itself is exercised in tests/security/test_egress_policy.py.
    """
    from local_deep_research.security.egress.policy import Decision

    with (
        patch(
            "local_deep_research.security.egress.policy.evaluate_engine",
            return_value=Decision(True, "test_bypass"),
        ),
        patch(
            "local_deep_research.security.egress.policy.evaluate_retriever",
            return_value=Decision(True, "test_bypass"),
        ),
    ):
        yield


class TestCreateSearchEngineRedactsApiKey:
    def test_constructor_exception_does_not_leak_api_key(self, loguru_caplog):
        """``api_key`` resolved from settings must be literal-redacted."""
        from local_deep_research.web_search_engines.search_engine_factory import (
            create_search_engine,
        )

        class _RaisingEngine:
            def __init__(self, api_key=None, **kwargs):
                raise ValueError(f"Bad key received: {api_key}")

        settings_snapshot = {
            "search.engine.web.test_engine.api_key": {"value": _BARE_KEY}
        }

        with (
            patch(f"{FACTORY_MODULE}.retriever_registry") as mock_registry,
            patch(f"{FACTORY_MODULE}.search_config") as mock_config,
            patch(
                f"{FACTORY_MODULE}.get_safe_module_class",
                return_value=_RaisingEngine,
            ),
        ):
            mock_registry.get.return_value = None
            mock_config.return_value = {
                "test_engine": {
                    "module_path": ".engines.test",
                    "class_name": "TestEngine",
                    "requires_api_key": True,
                }
            }
            with loguru_caplog.at_level("ERROR"):
                result = create_search_engine(
                    engine_name="test_engine",
                    settings_snapshot=settings_snapshot,
                )

        assert result is None
        assert _BARE_KEY not in loguru_caplog.text
        assert "Bad key received" in loguru_caplog.text
        assert "***REDACTED***" in loguru_caplog.text
        # Regression guard: the factory failure must stay at ERROR, not be
        # re-downgraded to WARNING. Filter by message so an unrelated
        # record can't satisfy (or break) the assertion.
        matching = [
            r
            for r in loguru_caplog.records
            if "Bad key received" in r.getMessage()
        ]
        assert matching
        assert all(r.levelname == "ERROR" for r in matching)


class TestCreateFullSearchWrapperRedactsKeys:
    def test_brave_wrapper_exception_does_not_leak_api_key(self, loguru_caplog):
        base_engine = Mock()
        engine_config = {
            "full_search_module": ".engines.full_search",
            "full_search_class": "FullSearchResults",
        }
        settings_snapshot = {
            "search.engine.web.brave.api_key": {"value": _BARE_KEY}
        }

        with patch(
            f"{FACTORY_MODULE}.get_safe_module_class",
            return_value=_RaisingWrapperBrave,
        ):
            with loguru_caplog.at_level("ERROR"):
                result = _create_full_search_wrapper(
                    "brave",
                    base_engine,
                    engine_config,
                    Mock(),
                    {},
                    settings_snapshot=settings_snapshot,
                )

        assert result is base_engine
        assert _BARE_KEY not in loguru_caplog.text
        assert "Brave wrapper rejected key" in loguru_caplog.text
        assert "***REDACTED***" in loguru_caplog.text
        matching = [
            r
            for r in loguru_caplog.records
            if "Brave wrapper rejected key" in r.getMessage()
        ]
        assert matching
        assert all(r.levelname == "ERROR" for r in matching)

    def test_serpapi_wrapper_exception_does_not_leak_key(self, loguru_caplog):
        base_engine = Mock()
        engine_config = {
            "full_search_module": ".engines.full_search",
            "full_search_class": "FullSearchResults",
        }
        settings_snapshot = {
            "search.engine.web.serpapi.api_key": {"value": _BARE_KEY}
        }

        with patch(
            f"{FACTORY_MODULE}.get_safe_module_class",
            return_value=_RaisingWrapperSerpAPI,
        ):
            with loguru_caplog.at_level("ERROR"):
                result = _create_full_search_wrapper(
                    "serpapi",
                    base_engine,
                    engine_config,
                    Mock(),
                    {},
                    settings_snapshot=settings_snapshot,
                )

        assert result is base_engine
        assert _BARE_KEY not in loguru_caplog.text
        assert "SerpAPI wrapper rejected key" in loguru_caplog.text
        assert "***REDACTED***" in loguru_caplog.text
        matching = [
            r
            for r in loguru_caplog.records
            if "SerpAPI wrapper rejected key" in r.getMessage()
        ]
        assert matching
        assert all(r.levelname == "ERROR" for r in matching)


class TestHoistedApiKeyHandlesNoKeyRequiredEngines:
    def test_no_key_required_engine_exception_does_not_nameerror(
        self, loguru_caplog
    ):
        """``api_key = None`` hoist must keep the scrub call safe on the
        no-key path.

        Before the hoist, ``api_key`` was only bound inside
        ``if requires_api_key:``. The except handler at the end of
        ``create_search_engine`` references ``api_key`` unconditionally —
        so a constructor exception for a ``requires_api_key=False`` engine
        would have raised ``NameError`` instead of logging the fallback
        error. This test pins the hoist: the error log fires and the
        engine returns ``None``.
        """

        class _RaisingEngineNoKey:
            def __init__(self, **kwargs):
                raise RuntimeError("engine init blew up")

        from local_deep_research.web_search_engines.search_engine_factory import (
            create_search_engine,
        )

        # Non-empty snapshot — the factory rejects empty/falsy snapshots in
        # thread context. The engine_config mock below is what actually
        # drives the requires_api_key=False path; the snapshot just needs to
        # be truthy to clear the thread-context guard.
        settings_snapshot = {"_test_marker": True}
        with (
            patch(f"{FACTORY_MODULE}.retriever_registry") as mock_registry,
            patch(f"{FACTORY_MODULE}.search_config") as mock_config,
            patch(
                f"{FACTORY_MODULE}.get_safe_module_class",
                return_value=_RaisingEngineNoKey,
            ),
        ):
            mock_registry.get.return_value = None
            mock_config.return_value = {
                "test_engine_no_key": {
                    "module_path": ".engines.test",
                    "class_name": "TestEngineNoKey",
                    "requires_api_key": False,
                }
            }
            with loguru_caplog.at_level("ERROR"):
                result = create_search_engine(
                    engine_name="test_engine_no_key",
                    settings_snapshot=settings_snapshot,
                )

        assert result is None
        # The error log fired (no NameError swallowed it) and carries the
        # exception message — tie them together so the assertion can't
        # false-pass on an unrelated ERROR line.
        assert any(
            "Failed to create search engine" in line
            and "engine init blew up" in line
            for line in loguru_caplog.text.splitlines()
        )
        matching = [
            r
            for r in loguru_caplog.records
            if "engine init blew up" in r.getMessage()
        ]
        assert matching
        assert all(r.levelname == "ERROR" for r in matching)