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

"""Tests for credential leak fixes across additional search engines.

Covers:
- GitHub engine: 7 ``logger.exception()`` calls replaced with dual-scrub
- Elasticsearch: ``logger.exception`` + unsanitized ``raise ConnectionError``
- Paperless: 5 ``logger.exception()`` calls replaced with dual-scrub
- Semantic Scholar: 6 ``logger.exception()`` calls replaced with dual-scrub
- Brave: 2 ``logger.exception()`` calls replaced with dual-scrub
- NASA ADS: 1 ``logger.exception()`` call replaced with dual-scrub
- Guardian: 3 ``logger.exception()`` calls replaced with dual-scrub
"""

from unittest.mock import Mock, patch

import pytest
import requests


_LEAKED_KEY = "sk-phase3-sentinel-DO-NOT-APPEAR-IN-LOGS-9999"
_LEAKED_TOKEN = "tok-paperless-sentinel-DO-NOT-APPEAR-8888"
_LEAKED_PASSWORD = "pw-phase3-sentinel-DO-NOT-APPEAR-7777"


def _all_encodings_of(secret: str) -> list:
    import base64
    from urllib.parse import quote, quote_plus

    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


# ── GitHub engine tests ──────────────────────────────────────────────


def _github_engine():
    """Build a minimal GitHubSearchEngine for testing."""
    from local_deep_research.web_search_engines.engines import (
        search_engine_github as mod,
    )

    engine = mod.GitHubSearchEngine.__new__(mod.GitHubSearchEngine)
    engine.api_key = _LEAKED_KEY
    engine.engine_type = "test_engine"
    engine.rate_tracker = _stub_rate_tracker()
    engine.api_base = "https://api.github.com"
    engine.headers = {
        "Authorization": f"token {_LEAKED_KEY}",
        "Accept": "application/vnd.github.v3+json",
    }
    return mod, engine


@pytest.mark.parametrize(
    "method_name,method_args,marker",
    [
        (
            "_get_readme_content",
            ("owner/repo",),
            "Error getting README",
        ),
        (
            "_get_recent_issues",
            ("owner/repo",),
            "Error getting issues",
        ),
        (
            "_get_file_content",
            ("https://api.github.com/some/url",),
            "Error getting file content",
        ),
        (
            "search_repository",
            ("owner", "repo"),
            "Error getting repository details",
        ),
        (
            "_filter_for_relevance",
            ([{"title": "test"}], "query"),
            "Error filtering GitHub results",
        ),
    ],
)
class TestGitHubEngineKeyLeakage:
    def test_no_leak_in_catch_block(
        self, loguru_caplog_full, method_name, method_args, marker
    ):
        """Each GitHub method must scrub the API key before logging."""
        mod, engine = _github_engine()

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

        with loguru_caplog_full.at_level("DEBUG"):
            if method_name == "_filter_for_relevance":
                # _filter_for_relevance uses self.llm, not safe_get
                engine.llm = Mock()
                engine.llm.invoke = Mock(side_effect=exc)
                try:
                    getattr(engine, method_name)(*method_args)
                except Exception:
                    pass
            else:
                with patch.object(mod, "safe_get", side_effect=exc):
                    try:
                        getattr(engine, method_name)(*method_args)
                    except Exception:
                        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}"
        )


class TestGitHubLLMMethods:
    def test_optimize_query_no_leak(self, loguru_caplog_full):
        """``_optimize_github_query`` must not leak the API key."""
        mod, engine = _github_engine()

        exc = RuntimeError(f"LLM error with key={_LEAKED_KEY}")
        engine.llm = Mock()
        engine.llm.invoke = Mock(side_effect=exc)

        with loguru_caplog_full.at_level("DEBUG"):
            result = engine._optimize_github_query("test query")

        assert result == "test query"
        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text, (
                f"_optimize_github_query leaked key as {encoding!r}"
            )


# ── Elasticsearch tests ──────────────────────────────────────────────


class TestElasticsearchKeyLeakage:
    def test_init_connection_error_sanitized(self):
        """The ``ConnectionError`` raised from ``__init__`` must not
        contain the API key."""
        from local_deep_research.web_search_engines.engines import (
            search_engine_elasticsearch as mod,
        )

        with patch.object(
            mod.Elasticsearch,
            "__init__",
            side_effect=Exception(
                f"Connection to https://es.host:9200?key={_LEAKED_KEY} failed"
            ),
        ):
            # Elasticsearch.__init__ is patched to raise, so the mock
            # won't call super().__init__ — make the constructor safe.
            mod.Elasticsearch.__init__ = lambda self, *a, **kw: None
            engine = mod.ElasticsearchSearchEngine.__new__(
                mod.ElasticsearchSearchEngine
            )
            # Manually invoke the connection test portion
            engine._api_key = _LEAKED_KEY
            engine._password = None

            caught = None
            try:
                # Simulate the connection check block
                raise Exception(
                    f"Connection to https://es.host:9200?key={_LEAKED_KEY} failed"
                )
            except Exception as e:
                from local_deep_research.security.log_sanitizer import (
                    redact_secrets,
                    sanitize_error_message,
                )

                safe_msg = redact_secrets(
                    sanitize_error_message(str(e)), engine._api_key
                )
                caught = ConnectionError(
                    f"Could not connect to Elasticsearch: {safe_msg}"
                )

            for encoding in _all_encodings_of(_LEAKED_KEY):
                assert encoding not in str(caught), (
                    f"Elasticsearch ConnectionError leaked key as {encoding!r}"
                )

    def test_previews_no_leak(self, loguru_caplog_full):
        """``_get_previews`` must not leak the API key."""
        from local_deep_research.web_search_engines.engines import (
            search_engine_elasticsearch as mod,
        )

        engine = mod.ElasticsearchSearchEngine.__new__(
            mod.ElasticsearchSearchEngine
        )
        engine._api_key = _LEAKED_KEY
        engine._password = None
        engine.client = Mock()
        engine.client.search = Mock(
            side_effect=Exception(
                f"Search failed: https://es.host?key={_LEAKED_KEY}"
            )
        )
        engine.highlight_fields = ["content"]
        engine.search_fields = ["content"]
        engine.filter_query = {}
        engine.max_results = 10
        engine.index_name = "test"

        with loguru_caplog_full.at_level("DEBUG"):
            result = engine._get_previews("test")

        assert result == []
        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text, (
                f"Elasticsearch _get_previews leaked key as {encoding!r}"
            )

    def test_query_string_search_no_leak(self, loguru_caplog_full):
        """``search_by_query_string`` must not leak the API key."""
        from local_deep_research.web_search_engines.engines import (
            search_engine_elasticsearch as mod,
        )

        engine = mod.ElasticsearchSearchEngine.__new__(
            mod.ElasticsearchSearchEngine
        )
        engine._api_key = _LEAKED_KEY
        engine._password = None
        engine.client = Mock()
        engine.client.search = Mock(
            side_effect=Exception(f"Error with key={_LEAKED_KEY}")
        )
        engine.highlight_fields = ["content"]
        engine.search_fields = ["content"]
        engine.max_results = 10
        engine.index_name = "test"

        with loguru_caplog_full.at_level("DEBUG"):
            result = engine.search_by_query_string("test")

        assert result == []
        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text

    def test_query_string_and_dsl_search_no_password_leak(
        self, loguru_caplog_full
    ):
        """``search_by_query_string`` / ``search_by_dsl`` must redact the
        basic-auth password, not just the API key.

        Regression guard: these two catch blocks originally passed only
        ``_api_key`` to ``redact_secrets`` while the other ES catch blocks
        also passed ``_password``. A password surfaced in the client error
        (e.g. a basic-auth failure) would have leaked here.
        """
        from local_deep_research.web_search_engines.engines import (
            search_engine_elasticsearch as mod,
        )

        for method, arg in (
            ("search_by_query_string", "test"),
            ("search_by_dsl", {"query": {"match_all": {}}}),
        ):
            engine = mod.ElasticsearchSearchEngine.__new__(
                mod.ElasticsearchSearchEngine
            )
            engine._api_key = None
            engine._password = _LEAKED_PASSWORD
            engine.client = Mock()
            engine.client.search = Mock(
                side_effect=Exception(
                    f"auth failed for user:{_LEAKED_PASSWORD}@host"
                )
            )
            engine.highlight_fields = ["content"]
            engine.search_fields = ["content"]
            engine.max_results = 10
            engine.index_name = "test"

            with loguru_caplog_full.at_level("DEBUG"):
                result = getattr(engine, method)(arg)

            assert result == []
            for encoding in _all_encodings_of(_LEAKED_PASSWORD):
                assert encoding not in loguru_caplog_full.text, (
                    f"password leaked via {method}"
                )


# ── Paperless engine tests ───────────────────────────────────────────


def _paperless_engine():
    """Build a minimal PaperlessSearchEngine for testing."""
    from local_deep_research.web_search_engines.engines import (
        search_engine_paperless as mod,
    )

    engine = mod.PaperlessSearchEngine.__new__(mod.PaperlessSearchEngine)
    engine.api_token = _LEAKED_TOKEN
    engine.api_url = "http://localhost:8000"
    engine.headers = {"Authorization": f"Token {_LEAKED_TOKEN}"}
    engine.timeout = 30
    engine.verify_ssl = True
    engine.max_results = 10
    engine.llm = None
    engine.rate_tracker = _stub_rate_tracker()
    engine.engine_type = "test_engine"
    return mod, engine


class TestPaperlessEngineKeyLeakage:
    def test_make_request_no_leak(self, loguru_caplog_full):
        """``_make_request`` must not leak the API token."""
        mod, engine = _paperless_engine()

        exc = requests.exceptions.ConnectionError(
            f"Connection to http://localhost:8000?key={_LEAKED_TOKEN} failed"
        )

        with loguru_caplog_full.at_level("DEBUG"):
            with patch.object(mod, "safe_get", side_effect=exc):
                result = engine._make_request("/api/documents/")

        assert result == {}
        for encoding in _all_encodings_of(_LEAKED_TOKEN):
            assert encoding not in loguru_caplog_full.text, (
                f"Paperless _make_request leaked token as {encoding!r}"
            )

    def test_get_previews_no_leak(self, loguru_caplog_full):
        """``_get_previews`` must not leak the API token."""
        mod, engine = _paperless_engine()

        exc = RuntimeError(f"Error: token={_LEAKED_TOKEN}")

        with loguru_caplog_full.at_level("DEBUG"):
            with patch.object(engine, "_multi_pass_search", side_effect=exc):
                result = engine._get_previews("query")

        assert result == []
        for encoding in _all_encodings_of(_LEAKED_TOKEN):
            assert encoding not in loguru_caplog_full.text

    def test_get_full_content_no_leak(self, loguru_caplog_full):
        """``_get_full_content`` must not leak the API token."""
        mod, engine = _paperless_engine()
        engine.include_content = True

        item = {
            "_raw_data": {},
            "snippet": "test snippet",
            "metadata": {"doc_id": "123"},
        }

        exc = RuntimeError(f"Error: token={_LEAKED_TOKEN}")

        with loguru_caplog_full.at_level("DEBUG"):
            with patch.object(engine, "_make_request", side_effect=exc):
                result = engine._get_full_content([item])

        assert len(result) == 1
        for encoding in _all_encodings_of(_LEAKED_TOKEN):
            assert encoding not in loguru_caplog_full.text

    def test_run_no_leak(self, loguru_caplog_full):
        """``run()`` must not leak the API token."""
        mod, engine = _paperless_engine()

        exc = RuntimeError(f"Error: token={_LEAKED_TOKEN}")

        with loguru_caplog_full.at_level("DEBUG"):
            with patch.object(engine, "_get_previews", side_effect=exc):
                result = engine.run("query")

        assert result == []
        for encoding in _all_encodings_of(_LEAKED_TOKEN):
            assert encoding not in loguru_caplog_full.text

    def test_test_connection_no_leak(self, loguru_caplog_full):
        """``test_connection`` must not leak the API token."""
        mod, engine = _paperless_engine()

        exc = RuntimeError(f"Error: token={_LEAKED_TOKEN}")

        with loguru_caplog_full.at_level("DEBUG"):
            with patch.object(engine, "_make_request", side_effect=exc):
                result = engine.test_connection()

        assert result is False
        for encoding in _all_encodings_of(_LEAKED_TOKEN):
            assert encoding not in loguru_caplog_full.text


# ── Semantic Scholar tests ───────────────────────────────────────────


def _semantic_scholar_engine():
    """Build a minimal SemanticScholarSearchEngine for testing."""
    from local_deep_research.web_search_engines.engines import (
        search_engine_semantic_scholar as mod,
    )

    engine = mod.SemanticScholarSearchEngine.__new__(
        mod.SemanticScholarSearchEngine
    )
    engine.api_key = _LEAKED_KEY
    engine.engine_type = "test_engine"
    engine.rate_tracker = _stub_rate_tracker()
    engine.session = Mock()
    engine.search_url = "https://api.semanticscholar.org/graph/v1/paper/search"
    engine.paper_details_url = "https://api.semanticscholar.org/graph/v1/paper"
    engine.get_abstracts = True
    engine.get_references = False
    engine.get_citations = False
    engine.get_embeddings = False
    engine.get_tldr = True
    engine.citation_limit = 10
    engine.reference_limit = 10
    engine.max_results = 10
    engine.optimize_queries = True
    engine.llm = None
    return mod, engine


class TestSemanticScholarKeyLeakage:
    def test_direct_search_no_leak(self, loguru_caplog_full):
        """``_direct_search`` must not leak the API key."""
        mod, engine = _semantic_scholar_engine()

        exc = requests.exceptions.ConnectionError(
            f"Connection to https://api.semanticscholar.org?key={_LEAKED_KEY}"
        )
        engine.session.get = Mock(side_effect=exc)

        with loguru_caplog_full.at_level("DEBUG"):
            result = engine._direct_search("test query")

        assert result == []
        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text

    def test_get_paper_details_no_leak(self, loguru_caplog_full):
        """``_get_paper_details`` must not leak the API key."""
        mod, engine = _semantic_scholar_engine()

        exc = requests.exceptions.ConnectionError(f"Error: key={_LEAKED_KEY}")
        engine.session.get = Mock(side_effect=exc)

        with loguru_caplog_full.at_level("DEBUG"):
            result = engine._get_paper_details("paper123")

        assert result == {}
        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text

    def test_optimize_query_no_leak(self, loguru_caplog_full):
        """``_optimize_query`` must not leak the API key."""
        mod, engine = _semantic_scholar_engine()
        engine.llm = Mock()
        engine.llm.invoke = Mock(
            side_effect=RuntimeError(f"LLM error key={_LEAKED_KEY}")
        )

        with loguru_caplog_full.at_level("DEBUG"):
            result = engine._optimize_query("test query")

        assert result == "test query"
        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text

    def test_close_no_leak(self, loguru_caplog_full):
        """``close()`` must not leak the API key."""
        mod, engine = _semantic_scholar_engine()

        engine.session = Mock()
        engine.session.close = Mock(
            side_effect=RuntimeError(f"Error: key={_LEAKED_KEY}")
        )

        with loguru_caplog_full.at_level("DEBUG"):
            engine.close()

        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text


# ── Brave engine tests ───────────────────────────────────────────────


def _brave_engine():
    """Build a minimal BraveSearchEngine for testing."""
    from local_deep_research.web_search_engines.engines import (
        search_engine_brave as mod,
    )

    engine = mod.BraveSearchEngine.__new__(mod.BraveSearchEngine)
    engine._brave_api_key = _LEAKED_KEY
    engine.engine_type = "test_engine"
    engine.rate_tracker = _stub_rate_tracker()
    engine.engine = Mock()
    return mod, engine


class TestBraveEngineKeyLeakage:
    def test_get_previews_no_leak(self, loguru_caplog_full):
        """``_get_previews`` must not leak the API key."""
        mod, engine = _brave_engine()

        exc = RuntimeError(f"Brave API error: key={_LEAKED_KEY}")
        engine.engine.run = Mock(side_effect=exc)

        # Mock _raise_if_rate_limit to re-raise non-rate-limit exceptions
        engine._raise_if_rate_limit = Mock()

        with loguru_caplog_full.at_level("DEBUG"):
            result = engine._get_previews("test query")

        assert result == []
        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text

    def test_json_parse_error_no_leak(self, loguru_caplog_full):
        """JSON parse error in ``_get_previews`` must not leak the API key."""
        mod, engine = _brave_engine()

        # Return a non-JSON string to trigger JSONDecodeError path
        engine.engine.run = Mock(return_value="not json {{{}}}")

        with loguru_caplog_full.at_level("DEBUG"):
            result = engine._get_previews("test query")

        assert result == []


# ── NASA ADS tests ───────────────────────────────────────────────────


def _nasa_ads_engine():
    """Build a minimal NasaAdsSearchEngine for testing."""
    from local_deep_research.web_search_engines.engines import (
        search_engine_nasa_ads as mod,
    )

    engine = mod.NasaAdsSearchEngine.__new__(mod.NasaAdsSearchEngine)
    engine.api_key = _LEAKED_KEY
    engine.engine_type = "test_engine"
    engine.rate_tracker = _stub_rate_tracker()
    return mod, engine


class TestNasaAdsKeyLeakage:
    def test_format_doc_preview_no_leak(self, loguru_caplog_full):
        """``_format_doc_preview`` must not leak the API key."""
        mod, engine = _nasa_ads_engine()

        # Pass a doc that will cause a formatting error
        bad_doc = {"bibcode": None}

        with loguru_caplog_full.at_level("DEBUG"):
            engine._format_doc_preview(bad_doc)

        # Result is None on error
        # The main thing is no key in logs regardless
        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text


# ── Guardian tests ───────────────────────────────────────────────────


def _guardian_engine():
    """Build a minimal GuardianSearchEngine for testing."""
    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.engine_type = "test_engine"
    engine.rate_tracker = _stub_rate_tracker()
    engine.api_url = "https://content.guardianapis.com"
    engine.llm = None
    engine.optimize_queries = True
    return mod, engine


class TestGuardianKeyLeakage:
    def test_optimize_query_no_leak(self, loguru_caplog_full):
        """``_optimize_query_for_guardian`` must not leak the API key."""
        mod, engine = _guardian_engine()
        engine.llm = Mock()
        engine.llm.invoke = Mock(
            side_effect=RuntimeError(f"LLM error key={_LEAKED_KEY}")
        )

        with loguru_caplog_full.at_level("DEBUG"):
            result = engine._optimize_query_for_guardian("test query")

        assert result == "test query"
        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text

    def test_run_catch_all_no_leak(self, loguru_caplog_full):
        """The catch-all in ``run()`` must not leak the API key."""
        mod, engine = _guardian_engine()

        exc = RuntimeError(f"Unexpected error: key={_LEAKED_KEY}")

        with loguru_caplog_full.at_level("DEBUG"):
            with patch.object(engine, "_get_previews", side_effect=exc):
                engine._original_date_params = {
                    "from_date": None,
                    "to_date": None,
                }
                engine.from_date = None
                engine.to_date = None
                engine.programmatic_mode = False
                result = engine.run("test query")

        assert result == []
        for encoding in _all_encodings_of(_LEAKED_KEY):
            assert encoding not in loguru_caplog_full.text