/tests/security/test_egress_fullsearch.py

"""Integration tests for the ``full_search.py`` egress Policy Enforcement Point:

``web_search_engines/engines/full_search.py`` — per-URL egress gating in the
full-content fetch path: when an ``egress_context`` is present, URLs that
``evaluate_url`` denies for the scope are dropped before any network fetch,
allowed ones are kept, and the denial audit log is URL-redacted.

These drive the REAL call-site code. Only unavoidable heavy deps are mocked
(the network ``batch_fetch_and_extract`` and the SSRF ``validate_url``, so the
SSRF axis is isolated from the orthogonal egress-scope axis under test). The
egress decision itself (``evaluate_url``) is exercised for real against IP
literals, so no DNS / network is required.
"""

from __future__ import annotations

import pytest
from loguru import logger

from local_deep_research.security.egress.policy import (
    EgressContext,
    EgressScope,
)
from local_deep_research.web_search_engines.engines import full_search as fs


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _make_ctx(scope: EgressScope) -> EgressContext:
    return EgressContext(
        scope=scope,
        primary_engine="library",
        require_local_llm=False,
        require_local_embeddings=False,
    )


class _FakeWebSearch:
    """Minimal stand-in for the inner web search engine (``.invoke``)."""

    def __init__(self, results):
        self._results = results

    def invoke(self, query):  # noqa: D401 - protocol method
        return list(self._results)


# Public + private IP literals resolve locally (no DNS / network) and match
# the literal-classification path the existing edge-case tests use.
_PUBLIC_URL = "http://93.184.216.34/page"
_PRIVATE_URL = "http://10.0.0.5/api"


# ===========================================================================
# full_search per-URL egress gating
# ===========================================================================


@pytest.fixture
def patched_fetch(monkeypatch):
    """Capture the URLs that survive the gate and reach the fetcher; isolate
    the SSRF axis by forcing ``validate_url`` True so only egress scope
    decides which URLs pass."""
    captured = {"urls": None, "calls": 0}

    def fake_batch(urls, **kwargs):
        captured["calls"] += 1
        captured["urls"] = list(urls)
        return {u: "content-for-" + u for u in urls}

    monkeypatch.setattr(fs, "batch_fetch_and_extract", fake_batch)
    monkeypatch.setattr(fs, "validate_url", lambda u: True)
    return captured


def _engine(results, ctx):
    return fs.FullSearchResults(
        llm=None,  # llm None -> check_urls returns results unfiltered
        web_search=_FakeWebSearch(results),
        egress_context=ctx,
    )


def test_run_private_only_drops_public_keeps_private(patched_fetch):
    results = [
        {"title": "pub", "link": _PUBLIC_URL},
        {"title": "priv", "link": _PRIVATE_URL},
    ]
    engine = _engine(results, _make_ctx(EgressScope.PRIVATE_ONLY))
    out = engine.run("q")

    # Only the private URL reached the fetcher.
    assert patched_fetch["urls"] == [_PRIVATE_URL]
    # And only its result carries full content; the denied public one is None.
    by_link = {r["link"]: r for r in out}
    assert (
        by_link[_PRIVATE_URL]["full_content"] == "content-for-" + _PRIVATE_URL
    )
    assert by_link[_PUBLIC_URL]["full_content"] is None


def test_run_public_only_drops_private_keeps_public(patched_fetch):
    results = [
        {"title": "pub", "link": _PUBLIC_URL},
        {"title": "priv", "link": _PRIVATE_URL},
    ]
    engine = _engine(results, _make_ctx(EgressScope.PUBLIC_ONLY))
    out = engine.run("q")

    assert patched_fetch["urls"] == [_PUBLIC_URL]
    by_link = {r["link"]: r for r in out}
    assert by_link[_PUBLIC_URL]["full_content"] == "content-for-" + _PUBLIC_URL
    assert by_link[_PRIVATE_URL]["full_content"] is None


def test_run_all_denied_skips_fetch_entirely(patched_fetch):
    """PRIVATE_ONLY with only public URLs -> nothing passes the gate, the
    network fetcher is never invoked, all results get null content."""
    results = [{"title": "pub", "link": _PUBLIC_URL}]
    engine = _engine(results, _make_ctx(EgressScope.PRIVATE_ONLY))
    out = engine.run("q")

    assert patched_fetch["calls"] == 0
    assert out[0]["full_content"] is None


def test_run_without_egress_context_does_not_gate(patched_fetch):
    """Control: with no egress_context the per-URL scope gate is inactive, so
    both URLs reach the fetcher. This proves the gate above is driven by the
    egress context and not by some unrelated filter."""
    results = [
        {"title": "pub", "link": _PUBLIC_URL},
        {"title": "priv", "link": _PRIVATE_URL},
    ]
    engine = fs.FullSearchResults(
        llm=None, web_search=_FakeWebSearch(results), egress_context=None
    )
    engine.run("q")

    assert patched_fetch["urls"] == [_PUBLIC_URL, _PRIVATE_URL]


def test_get_full_content_gates_per_url(patched_fetch):
    """The secondary ``_get_full_content`` path enforces the same per-URL
    scope gate."""
    items = [
        {"title": "pub", "link": _PUBLIC_URL},
        {"title": "priv", "link": _PRIVATE_URL},
    ]
    engine = _engine(items, _make_ctx(EgressScope.PRIVATE_ONLY))
    out = engine._get_full_content(items)

    assert patched_fetch["urls"] == [_PRIVATE_URL]
    by_link = {r["link"]: r for r in out}
    assert (
        by_link[_PRIVATE_URL]["full_content"] == "content-for-" + _PRIVATE_URL
    )
    assert by_link[_PUBLIC_URL]["full_content"] is None


def test_denied_url_audit_log_is_redacted(patched_fetch):
    """The denial audit record must carry only scheme://host (no path / query /
    token), proving sensitive URL components are not leaked into logs."""
    sensitive_url = "http://93.184.216.34/secret/report?token=SUPERSECRET"
    results = [{"title": "pub", "link": sensitive_url}]

    records = []

    def sink(message):
        records.append(message.record)

    # The package disables its own loguru namespace in __init__; enable it so
    # the in-module audit warnings actually reach our sink. Restore in finally.
    logger.enable("local_deep_research")
    sink_id = logger.add(sink, level="WARNING")
    try:
        engine = _engine(results, _make_ctx(EgressScope.PRIVATE_ONLY))
        engine.run("q")
    finally:
        logger.remove(sink_id)
        logger.disable("local_deep_research")

    # Find the policy-audit denial record for this URL.
    audit = [
        r
        for r in records
        if r["extra"].get("policy_audit") and "url" in r["extra"]
    ]
    assert audit, "expected a policy_audit denial record"
    rec = audit[0]
    logged_url = rec["extra"]["url"]
    assert logged_url == "http://93.184.216.34"
    # The token / path must appear nowhere in the logged URL or message.
    assert "SUPERSECRET" not in logged_url
    assert "SUPERSECRET" not in rec["message"]
    assert "/secret" not in logged_url
    assert rec["extra"]["reason"] == "scope_mismatch_private_only"