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

"""Tests for credential leak fixes that address findings from the PR #4452 audit.

Covers:
- PubMed engine: 5 ``logger.exception()`` calls replaced with dual-scrub
- Base class ``run()``: unredacted ``error_message`` persisted to database
- LLM config ``_log_llm_error``: ``logger.exception()`` replaced with ``logger.warning``
- Google PSE ``_validate_connection``: bare re-raise replaced with sanitized exception
- OpenAI compat errors: ``{exc!s}`` replaced with ``sanitize_error_message``
"""

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

import pytest
import requests


_LEAKED_KEY = "sk-leaked-sentinel-DO-NOT-APPEAR-12345"


def _all_encodings_of(secret: str) -> list:
    return [
        secret,
        quote(secret, safe=""),
        quote_plus(secret),
        repr(secret)[1:-1],
        base64.b64encode(secret.encode()).decode(),
        secret[:8],
    ]


def _stub_rate_tracker() -> Mock:
    tracker = Mock()
    tracker.enabled = False
    tracker.apply_rate_limit.return_value = 0
    return tracker


# ── PubMed engine tests ──────────────────────────────────────────────


def _pubmed_engine():
    """Build a minimal PubMedSearchEngine for testing."""
    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.engine_type = "test_engine"
    engine.rate_tracker = _stub_rate_tracker()
    engine.summary_url = (
        "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esummary.fcgi"
    )
    engine.fetch_url = (
        "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/efetch.fcgi"
    )
    engine.link_url = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/elink.fcgi"
    engine.get_full_text = True
    return mod, engine


@pytest.mark.parametrize(
    "method_name,url_path,marker",
    [
        (
            "_get_article_summaries",
            "/esummary.fcgi?api_key=",
            "Error getting article summaries",
        ),
        (
            "_get_article_abstracts",
            "/efetch.fcgi?api_key=",
            "Error getting article abstracts",
        ),
        (
            "_get_article_detailed_metadata",
            "/efetch.fcgi?api_key=",
            "Error getting detailed article metadata",
        ),
        (
            "_find_pmc_ids",
            "/elink.fcgi?api_key=",
            "Error finding PMC IDs",
        ),
        (
            "_get_pmc_full_text",
            "/efetch.fcgi?api_key=",
            "Error getting PMC full text",
        ),
    ],
)
class TestPubMedEngineKeyLeakage:
    def test_no_leak_in_catch_block(
        self, loguru_caplog_full, method_name, url_path, marker
    ):
        """Each PubMed method must scrub the API key before logging."""
        mod, engine = _pubmed_engine()

        exc = requests.exceptions.ConnectionError(
            f"HTTPSConnectionPool: Max retries exceeded with url: "
            f"{url_path}{_LEAKED_KEY}&id=12345"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            with patch.object(mod, "safe_get", side_effect=exc):
                try:
                    method = getattr(engine, method_name)
                    if method_name == "_get_article_summaries":
                        method(["12345"])
                    elif method_name == "_find_pmc_ids":
                        method(["12345"])
                    elif method_name == "_get_pmc_full_text":
                        method("PMC12345")
                    else:
                        method(["12345"])
                except requests.exceptions.RequestException:
                    pass

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

        assert marker in loguru_caplog_full.text, (
            f"{method_name}: catch-block marker {marker!r} not in logs — "
            f"the redaction path may not have run. "
            f"Captured: {loguru_caplog_full.text!r}"
        )


# ── Base class database persistence tests ────────────────────────────


class TestBaseClassDatabasePersistence:
    def test_error_message_sanitized_before_db_persist(self, loguru_caplog):
        """The ``error_message`` passed to ``SearchTracker.record_search``
        must be sanitized so credentials don't persist to the database."""
        from local_deep_research.web_search_engines.search_engine_base import (
            BaseSearchEngine,
        )
        from local_deep_research.metrics.search_tracker import SearchTracker

        class _TestEngine(BaseSearchEngine):
            engine_type = "test_engine"
            api_key = _LEAKED_KEY

            def _get_previews(self, query, *args, **kwargs):
                raise requests.exceptions.ConnectionError(
                    f"HTTPSConnectionPool: url: /search?api_key={_LEAKED_KEY}"
                )

        engine = _TestEngine.__new__(_TestEngine)
        engine.api_key = _LEAKED_KEY
        engine.engine_type = "test_engine"
        engine.rate_tracker = _stub_rate_tracker()
        engine.programmatic_mode = False
        # __init__ always sets this (settings_snapshot or {}); run() and the
        # egress-verification path read it. Bypassing __init__ via __new__
        # omits it, which AttributeErrors only when a sibling test arms the
        # global egress audit net — a parallel-run flake. Set it explicitly.
        engine.settings_snapshot = {}

        captured = {}

        def _capture_record(**kwargs):
            captured["error_message"] = kwargs.get("error_message")

        with loguru_caplog.at_level("DEBUG"):
            with (
                patch.object(
                    SearchTracker, "record_search", side_effect=_capture_record
                ),
                patch(
                    "local_deep_research.web_search_engines.search_engine_base.set_search_context"
                ),
            ):
                engine.run("test query")

        assert "error_message" in captured, "record_search was never called"
        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in captured["error_message"], (
                f"api_key leaked into DB error_message as {encoding!r}: "
                f"{captured['error_message']!r}"
            )

    def test_log_path_sanitizes_foreign_url_credentials(self, loguru_caplog):
        """The logged ``safe_msg`` in run() must run the regex sanitizer,
        not just literal redaction.

        Regression guard: a *foreign* credential (not the engine's own
        api_key) embedded as ``https://user:pass@host`` is only caught by
        sanitize_error_message's regex, not by redact_secrets(literal). The
        log path previously used redact_secrets alone, so it leaked into the
        log even though the DB path was dual-scrubbed.
        """
        from local_deep_research.web_search_engines.search_engine_base import (
            BaseSearchEngine,
        )
        from local_deep_research.metrics.search_tracker import SearchTracker

        foreign_secret = "hunter2-foreign-pw-DO-NOT-APPEAR"

        class _TestEngine(BaseSearchEngine):
            engine_type = "test_engine"
            api_key = "the-engines-own-key-unrelated"

            def _get_previews(self, query, *args, **kwargs):
                raise requests.exceptions.ConnectionError(
                    f"connect failed: https://admin:{foreign_secret}@es.local/_search"
                )

        engine = _TestEngine.__new__(_TestEngine)
        engine.api_key = "the-engines-own-key-unrelated"
        engine.engine_type = "test_engine"
        engine.rate_tracker = _stub_rate_tracker()
        engine.programmatic_mode = False
        engine.settings_snapshot = {}

        with loguru_caplog.at_level("DEBUG"):
            with (
                patch.object(SearchTracker, "record_search"),
                patch(
                    "local_deep_research.web_search_engines.search_engine_base.set_search_context"
                ),
            ):
                engine.run("test query")

        assert foreign_secret not in loguru_caplog.text, (
            f"foreign URL credential leaked into log: {loguru_caplog.text!r}"
        )


# ── LLM config _log_llm_error tests ─────────────────────────────────


class TestLLMConfigLogError:
    def test_no_leak_with_url_in_exception(self, loguru_caplog_full):
        """``_log_llm_error`` must not leak credentials from the exception."""
        from local_deep_research.config.llm_config import _log_llm_error

        exc = RuntimeError(
            f"Connection to https://api.openai.com/v1?key={_LEAKED_KEY} failed"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            _log_llm_error(exc)

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

    def test_uses_warning_not_exception(self, loguru_caplog):
        """``_log_llm_error`` should use ``logger.warning``, not
        ``logger.exception``, to avoid writing the traceback chain."""
        from local_deep_research.config.llm_config import _log_llm_error

        exc = RuntimeError("test error")

        with loguru_caplog.at_level("DEBUG"):
            _log_llm_error(exc)

        assert "LLM Request - Failed with error" in loguru_caplog.text
        assert "Traceback" not in loguru_caplog.text


# ── Google PSE _validate_connection tests ────────────────────────────


class TestGooglePSEValidateConnection:
    def test_rethrown_exception_is_sanitized(self):
        """The re-thrown exception from ``_validate_connection`` must not
        contain the API key in its string representation."""
        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

        exc = requests.exceptions.ConnectionError(
            f"HTTPSConnectionPool: url: /v1?key={_LEAKED_KEY}"
        )
        engine._make_request = Mock(side_effect=exc)

        caught = None
        try:
            engine._validate_connection()
        except Exception as e:
            caught = e

        assert caught is not None
        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in str(caught), (
                f"Re-thrown exception leaks key as {encoding!r}: {str(caught)!r}"
            )

    def test_exception_chain_suppressed(self):
        """``from None`` should suppress the original exception chain."""
        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

        exc = ValueError("original")
        engine._make_request = Mock(side_effect=exc)

        caught = None
        try:
            engine._validate_connection()
        except Exception as e:
            caught = e

        assert caught is not None
        assert caught.__cause__ is None

    def test_log_output_sanitized(self, loguru_caplog_full):
        """The warning log from ``_validate_connection`` must not leak."""
        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

        exc = requests.exceptions.ConnectionError(
            f"HTTPSConnectionPool: url: /v1?key={_LEAKED_KEY}"
        )
        engine._make_request = Mock(side_effect=exc)

        with loguru_caplog_full.at_level("DEBUG"):
            try:
                engine._validate_connection()
            except Exception:
                pass

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


# ── OpenAI compat errors tests ───────────────────────────────────────


class TestOpenAICompatErrorDetails:
    def test_details_with_url_key_redacted(self):
        """The ``Details:`` suffix must not contain URL query credentials."""
        from local_deep_research.error_handling.openai_compat_errors import (
            friendly_openai_compatible_error,
        )

        exc = RuntimeError(
            f"Connection to https://api.host/v1?key={_LEAKED_KEY} failed"
        )
        result = friendly_openai_compatible_error(
            exc, provider="test", base_url="https://api.host/v1", model="gpt-4"
        )

        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in result, (
                f"Details leaked key as {encoding!r}: {result!r}"
            )

    def test_details_with_bearer_redacted(self):
        """Bearer tokens must be scrubbed from the Details suffix."""
        from local_deep_research.error_handling.openai_compat_errors import (
            friendly_openai_compatible_error,
        )

        exc = RuntimeError(f"Bearer {_LEAKED_KEY} rejected")
        result = friendly_openai_compatible_error(
            exc, provider="test", base_url="https://api.host/v1", model="gpt-4"
        )

        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in result

    def test_no_secrets_preserves_details(self):
        """Non-sensitive details should be preserved."""
        from local_deep_research.error_handling.openai_compat_errors import (
            friendly_openai_compatible_error,
        )

        exc = RuntimeError("connection refused on port 443")
        result = friendly_openai_compatible_error(
            exc, provider="test", base_url="https://api.host/v1", model="gpt-4"
        )

        assert "connection refused on port 443" in result