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

"""Regression tests for URL-context redaction in engine log lines.

Comparison review of issue #4183 found that even after ``_scrub_error(e)``
scrubs the exception message, the URL interpolated into the *same* log line
(``url``, ``wayback_url``, ``self.instance_url``) was written to logs raw.
URLs can carry HTTP basic-auth userinfo (``https://user:pass@host``) or
credential-bearing query params (``?api_key=...``), both of which leaked.

These tests pin the fix: every URL interpolation in a log line goes through
``redact_url_for_log`` so only ``scheme://host:port`` reaches the log sink.

Uses ``loguru_caplog`` (not ``loguru_caplog_full``) because these engines
log exceptions through the diagnose-gated ``security.secure_logging``
wrapper: at ERROR level, but with no exception attached to the record
outside diagnose mode — so there is no rendered exception block to
capture, only the formatted message. (The wrapper's own gating is pinned
in ``test_secure_logging.py``.) The levelname assertions guard against a
re-downgrade to WARNING.
"""

from unittest.mock import Mock, patch

import requests

from local_deep_research.web_search_engines.engines.search_engine_paperless import (
    PaperlessSearchEngine,
)
from local_deep_research.web_search_engines.engines.search_engine_searxng import (
    SearXNGSearchEngine,
)
from local_deep_research.web_search_engines.engines.search_engine_wayback import (
    WaybackSearchEngine,
)

WAYBACK_MODULE = (
    "local_deep_research.web_search_engines.engines.search_engine_wayback"
)
SEARXNG_MODULE = (
    "local_deep_research.web_search_engines.engines.search_engine_searxng"
)
PAPERLESS_MODULE = (
    "local_deep_research.web_search_engines.engines.search_engine_paperless"
)
OLLAMA_MODULE = (
    "local_deep_research.embeddings.providers.implementations.ollama"
)


def _mock_response(status=200, json_body=None, text=""):
    resp = Mock()
    resp.status_code = status
    resp.json.return_value = json_body if json_body is not None else {}
    resp.text = text
    return resp


class TestWaybackUrlRedaction:
    def test_snapshots_warning_redacts_userinfo(self, loguru_caplog):
        """``url`` with ``user:pass@`` must not leak the password."""
        engine = WaybackSearchEngine()
        cred_url = "https://user:hunter2@example.com/page"
        with patch(f"{WAYBACK_MODULE}.safe_get", side_effect=OSError("boom")):
            with loguru_caplog.at_level("ERROR"):
                engine._get_wayback_snapshots(cred_url)
        assert "hunter2" not in loguru_caplog.text
        assert "user:hunter2@" not in loguru_caplog.text
        # The error log fired AND carries the redacted form — tie the two
        # together so the assertion can't false-pass on an unrelated log line
        # that happens to mention the host.
        assert any(
            "Error getting Wayback snapshots for" in line
            and "https://example.com" in line
            for line in loguru_caplog.text.splitlines()
        )
        matching = [
            r
            for r in loguru_caplog.records
            if "Error getting Wayback snapshots for" in r.getMessage()
        ]
        assert matching
        assert all(r.levelname == "ERROR" for r in matching)

    def test_snapshots_warning_redacts_query_token(self, loguru_caplog):
        """Line 236: credential-bearing query params must not leak."""
        engine = WaybackSearchEngine()
        cred_url = "https://example.com/page?api_key=xyz12345abcde"
        with patch(f"{WAYBACK_MODULE}.safe_get", side_effect=OSError("boom")):
            with loguru_caplog.at_level("ERROR"):
                engine._get_wayback_snapshots(cred_url)
        assert "xyz12345abcde" not in loguru_caplog.text
        assert "api_key=" not in loguru_caplog.text

    def test_full_content_redacts_wayback_url(self, loguru_caplog):
        """``wayback_url`` carrying embedded creds must not leak.

        Covers the info log on the happy path ("Retrieving content from …")
        and the error log when ``_get_wayback_content`` then raises ("Error
        processing …"). The wayback_url embeds the original target in its
        path; if that target had basic-auth userinfo, the raw log line
        would leak it.
        """
        engine = WaybackSearchEngine()
        cred_wayback_url = (
            "https://web.archive.org/web/20240101/"
            "https://user:hunter2@example.com/page"
        )
        items = [{"link": cred_wayback_url, "title": "X"}]
        with patch(f"{WAYBACK_MODULE}.safe_get", side_effect=OSError("boom")):
            with loguru_caplog.at_level("INFO"):
                engine._get_full_content(items)
        assert "hunter2" not in loguru_caplog.text
        assert "user:hunter2@" not in loguru_caplog.text
        matching = [
            r
            for r in loguru_caplog.records
            if "Error retrieving content from" in r.getMessage()
        ]
        assert matching
        assert all(r.levelname == "ERROR" for r in matching)


class TestSearXNGUrlRedaction:
    def test_init_redacts_basic_auth_instance_url(self, loguru_caplog):
        """All ``self.instance_url`` log sites must redact basic-auth creds.

        Self-hosted SearXNG behind HTTP basic auth is a common deployment
        (``https://user:pass@host``); the init path logs the instance URL at
        info, and on connection failure the warning fires with it too.
        """
        cred_url = "https://admin:s3cret-pw@searxng.example.com"
        with patch(
            f"{SEARXNG_MODULE}.safe_get",
            side_effect=requests.ConnectionError("nope"),
        ):
            with loguru_caplog.at_level("INFO"):
                SearXNGSearchEngine(instance_url=cred_url)
        assert "s3cret-pw" not in loguru_caplog.text
        assert "admin:s3cret-pw@" not in loguru_caplog.text
        # The connection-failure error log fired AND carries the redacted
        # host — tie the two together so the assertion can't false-pass on
        # the init info log alone.
        assert any(
            "Error while trying to access SearXNG instance at" in line
            and "searxng.example.com" in line
            for line in loguru_caplog.text.splitlines()
        )
        matching = [
            r
            for r in loguru_caplog.records
            if "Error while trying to access SearXNG instance at"
            in r.getMessage()
        ]
        assert matching
        assert all(r.levelname == "ERROR" for r in matching)

    def test_debug_logs_redact_scraped_result_url(self, loguru_caplog):
        """Result-href debug logs (extracted + filtered) must redact too.

        Scraped result hrefs come from the internet, not operator config, so
        the leak vector is a malicious result with a credential-bearing URL —
        lower severity than the instance_url vector but same leak class.
        """
        cred_result_url = "https://user:hunter2@evil.example.com/path"
        # Build a result element whose href carries basic-auth creds. The
        # ``_is_valid_search_result`` check rejects URLs starting with the
        # instance URL, so use an instance URL that doesn't prefix-match.
        html = (
            '<html><body><article class="result">'
            f'<a class="result-title" href="{cred_result_url}">title</a>'
            '<a class="result-url" href="{cred}">link</a>'
            '<div class="result-content">snippet</div>'
            "</article></body></html>"
        ).format(cred=cred_result_url)
        resp = _mock_response(status=200, text=html)
        with patch(f"{SEARXNG_MODULE}.safe_get", return_value=resp):
            engine = SearXNGSearchEngine(
                instance_url="http://searxng-invalid-host-99999.local:8080",
            )
            engine.is_available = True
            engine.search_url = (
                "http://searxng-invalid-host-99999.local:8080/search"
            )
            with loguru_caplog.at_level("DEBUG"):
                engine._get_search_results("anything")
        assert "hunter2" not in loguru_caplog.text
        assert "user:hunter2@" not in loguru_caplog.text


class TestPaperlessUrlRedaction:
    """Paperless-ngx ``api_url`` carries HTTP basic-auth creds in many
    self-hosted deployments (``https://user:pass@host``). Every log site
    that interpolates it — init info, per-request debug, failed-request
    debug — must go through ``redact_url_for_log``."""

    def test_init_and_request_logs_redact_basic_auth(self, loguru_caplog):
        cred_url = "https://admin:s3cret-pw@paperless.example.com"
        with (
            patch(f"{PAPERLESS_MODULE}.safe_get") as mock_get,
            patch(f"{PAPERLESS_MODULE}.requests.get") as mock_requests_get,
        ):
            mock_get.side_effect = requests.ConnectionError("nope")
            mock_requests_get.side_effect = requests.ConnectionError("nope")
            with loguru_caplog.at_level("DEBUG"):
                engine = PaperlessSearchEngine(api_url=cred_url, api_key="x")
                engine._make_request("/api/documents/")

        assert "s3cret-pw" not in loguru_caplog.text
        assert "admin:s3cret-pw@" not in loguru_caplog.text
        # The init info log fired AND carries the redacted host — tie them
        # together so the assertion can't false-pass on an unrelated log.
        assert any(
            "Initialized Paperless-ngx search engine for" in line
            and "paperless.example.com" in line
            for line in loguru_caplog.text.splitlines()
        )
        # The per-request debug log also fired with the redacted URL.
        assert any(
            "Making request to:" in line and "paperless.example.com" in line
            for line in loguru_caplog.text.splitlines()
        )


class TestOllamaBaseUrlRedaction:
    """Ollama ``base_url`` is operator-configured (settings/env) and may
    sit behind a reverse proxy with HTTP basic auth. The init info log
    that announces the resolved base_url must redact it."""

    def test_create_embeddings_info_redacts_basic_auth(self, loguru_caplog):
        cred_url = "https://user:hunter2@ollama.example.com"
        with patch(f"{OLLAMA_MODULE}.OllamaEmbeddings") as mock_inner:
            mock_inner.return_value = Mock()
            from local_deep_research.embeddings.providers.implementations.ollama import (
                OllamaEmbeddingsProvider,
            )

            with loguru_caplog.at_level("INFO"):
                OllamaEmbeddingsProvider.create_embeddings(
                    model="nomic-embed-text", base_url=cred_url
                )
        assert "hunter2" not in loguru_caplog.text
        assert "user:hunter2@" not in loguru_caplog.text
        assert any(
            "Creating OllamaEmbeddings with model=" in line
            and "ollama.example.com" in line
            for line in loguru_caplog.text.splitlines()
        )