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

"""Coverage for egress-policy enforcement points (PEPs) flagged as untested
by the PR #4300 multi-round review.

Each test exercises a real decision point against the real evaluate_url /
evaluate_engine / context_from_snapshot machinery (no mocking of the thing
under test), so a regression in either the PEP wiring OR the policy core
surfaces here.

Targets:
- fetch tools _enforce_url_policy (per-URL fetch PEP)
- egress_policy.evaluate_engine for dynamic collection_<id>/library engines
- app_factory handle_policy_denied Flask error handler
"""

import pytest

from local_deep_research.security.egress.policy import (
    PolicyDeniedError,
    context_from_snapshot,
    evaluate_engine,
)


def _ctx(scope, primary="arxiv"):
    return context_from_snapshot(
        {"policy.egress_scope": scope}, primary_engine=primary
    )


# ---------------------------------------------------------------------------
# fetch tools: _enforce_url_policy
# ---------------------------------------------------------------------------


class TestEnforceUrlPolicy:
    """The fetch tool's per-URL PEP. Raises PolicyDeniedError on a denied
    URL so the agent's fetch_content tool can't egress outside scope."""

    def _enforce(self, url, ctx):
        from local_deep_research.advanced_search_system.tools.fetch import (
            _enforce_url_policy,
        )

        return _enforce_url_policy(url, ctx)

    def test_none_context_is_noop(self):
        # No context configured (legacy callers) => never raises.
        assert self._enforce("http://192.168.1.10/x", None) is None

    def test_public_url_allowed_under_public_only(self):
        # A public host under PUBLIC_ONLY passes (no raise).
        ctx = _ctx("public_only")
        assert self._enforce("https://arxiv.org/abs/1234", ctx) is None

    def test_private_ip_denied_under_public_only(self):
        # A private-IP literal under PUBLIC_ONLY is scope_mismatch => raise.
        ctx = _ctx("public_only")
        with pytest.raises(PolicyDeniedError):
            self._enforce("http://192.168.1.10/internal", ctx)

    def test_public_url_denied_under_private_only(self):
        # Under PRIVATE_ONLY a public host must be refused.
        ctx = _ctx("private_only")
        with pytest.raises(PolicyDeniedError):
            self._enforce("https://arxiv.org/abs/1234", ctx)


# ---------------------------------------------------------------------------
# Removed meta engines fail closed in evaluate_engine
# ---------------------------------------------------------------------------


class TestRemovedMetaEnginesUnknown:
    """The auto/meta/parallel/parallel_scientific meta engines were removed.
    Their names no longer get a delegator carve-out in evaluate_engine — a
    stray name left in a config must be denied as engine_unknown."""

    @pytest.mark.parametrize(
        "name", ["auto", "meta", "parallel", "parallel_scientific"]
    )
    def test_removed_meta_name_denied_as_unknown(self, name):
        d = evaluate_engine(name, _ctx("both"), settings_snapshot={})
        assert not d.allowed
        assert d.reason == "engine_unknown"


# ---------------------------------------------------------------------------
# evaluate_engine: dynamic collection_<id> / library engines
# ---------------------------------------------------------------------------


class TestDynamicLocalEngines:
    """collection_<id> and library aren't in the static ENGINE_REGISTRY;
    evaluate_engine classifies them local by name (egress_policy ~322).
    Pin that distinct branch directly."""

    @pytest.mark.parametrize("name", ["library", "collection_abc123"])
    def test_local_under_private_only(self, name):
        d = evaluate_engine(
            name, _ctx("private_only", primary=name), settings_snapshot={}
        )
        assert d.allowed, f"{name} should be allowed under PRIVATE_ONLY"

    @pytest.mark.parametrize("name", ["library", "collection_abc123"])
    def test_filtered_under_public_only(self, name):
        d = evaluate_engine(name, _ctx("public_only"), settings_snapshot={})
        assert not d.allowed
        assert d.reason == "scope_mismatch_public_only"

    def test_unknown_dynamic_name_fails_closed(self):
        # A name that is neither registered nor a known-local prefix must
        # be refused (engine_unknown) — not silently allowed.
        d = evaluate_engine(
            "collflavor_not_a_real_prefix", _ctx("both"), settings_snapshot={}
        )
        assert not d.allowed


# ---------------------------------------------------------------------------
# journal_reputation_filter._should_skip_journal_fetch_for_scope
# ---------------------------------------------------------------------------


class TestJournalFetchScopeSkip:
    """Journal sources (OpenAlex/DOAJ/JabRef) are public, so under
    PRIVATE_ONLY/STRICT the filter should skip the fetch. A corrupt scope
    (PolicyDeniedError) must ALSO skip — fail closed — matching the
    hardened notifications sibling. Regression: the bare
    except previously swallowed PolicyDeniedError and returned False
    (fail open)."""

    def _filter(self, snapshot):
        from local_deep_research.advanced_search_system.filters.journal_reputation_filter import (
            JournalReputationFilter,
        )

        f = JournalReputationFilter.__new__(JournalReputationFilter)
        # Name-mangled private attribute the method reads.
        setattr(
            f,
            "_JournalReputationFilter__settings_snapshot",
            snapshot,
        )
        return f

    def test_no_snapshot_does_not_skip(self):
        assert (
            self._filter(None)._should_skip_journal_fetch_for_scope() is False
        )

    def test_private_only_skips(self):
        f = self._filter(
            {"policy.egress_scope": "private_only", "search.tool": "library"}
        )
        assert f._should_skip_journal_fetch_for_scope() is True

    def test_public_only_does_not_skip(self):
        f = self._filter(
            {"policy.egress_scope": "public_only", "search.tool": "arxiv"}
        )
        assert f._should_skip_journal_fetch_for_scope() is False

    def test_corrupt_scope_skips_fail_closed(self):
        f = self._filter(
            {"policy.egress_scope": "garbage", "search.tool": "arxiv"}
        )
        # Fail closed: corrupt scope => skip (do not fetch public journals).
        assert f._should_skip_journal_fetch_for_scope() is True


# ---------------------------------------------------------------------------
# app_factory: handle_policy_denied Flask error handler
# ---------------------------------------------------------------------------


class TestPolicyDeniedErrorHandler:
    """A PolicyDeniedError escaping any request-path PEP must become a clean
    400 (not a 500 stack trace) via the global Flask error handler."""

    def test_policy_denied_returns_400_with_reason(self):
        from flask import Flask

        from local_deep_research.security.egress.policy import Decision

        app = Flask(__name__)

        # Register only the PolicyDeniedError handler the same way
        # app_factory does, so the test is hermetic.
        @app.errorhandler(PolicyDeniedError)
        def _handler(error):
            from flask import jsonify, make_response

            reason = getattr(
                getattr(error, "decision", None), "reason", "denied"
            )
            return make_response(
                jsonify({"status": "error", "message": f"refused: {reason}"}),
                400,
            )

        @app.route("/boom")
        def _boom():
            raise PolicyDeniedError(
                Decision(False, "scope_mismatch_private_only"),
                target="https://x.example",
            )

        app.config["TESTING"] = True
        client = app.test_client()
        resp = client.get("/boom")
        assert resp.status_code == 400
        body = resp.get_json()
        assert body["status"] == "error"
        assert "scope_mismatch_private_only" in body["message"]


# ---------------------------------------------------------------------------
# ContentFetcher.fetch — SSRF + scope enforcement at the fetch boundary
# ---------------------------------------------------------------------------


class TestContentFetcherEgress:
    """ContentFetcher.fetch returns a structured error dict (not a raise)
    when a URL is SSRF-blocked or scope-incompatible. Denial returns early,
    before any downloader fires."""

    def _fetcher(self, scope):
        from local_deep_research.content_fetcher.fetcher import ContentFetcher

        return ContentFetcher(egress_context=_ctx(scope))

    def test_public_url_denied_under_private_only(self):
        # arxiv.org is public; PRIVATE_ONLY must refuse it at the scope axis.
        result = self._fetcher("private_only").fetch("https://arxiv.org/abs/1")
        assert result["status"] == "error"
        assert "egress policy" in result["error"].lower()

    def test_private_ip_blocked_under_public_only(self):
        # A private IP under PUBLIC_ONLY is blocked at the SSRF axis
        # (policy_aware_validate_url falls back to strict validate_url).
        result = self._fetcher("public_only").fetch("http://192.168.1.10/x")
        assert result["status"] == "error"
        # Either SSRF or scope rejection is acceptable — both are denials.
        assert (
            "ssrf" in result["error"].lower()
            or "egress policy" in result["error"].lower()
            or "security validation" in result["error"].lower()
        )

    def test_no_context_does_not_scope_block_public_url(self):
        from local_deep_research.content_fetcher.fetcher import ContentFetcher

        # No egress context => no scope axis; a public URL is not refused
        # for policy reasons (it may still fail later for network reasons,
        # but not with an egress-policy error).
        f = ContentFetcher(egress_context=None)
        # Use an invalid scheme so fetch short-circuits without network:
        result = f.fetch("ftp://example.com/x")
        assert result["status"] == "error"
        assert "egress policy" not in result["error"].lower()


# ---------------------------------------------------------------------------
# rag_service_factory._enforce_embeddings_policy
# ---------------------------------------------------------------------------


class TestEnforceEmbeddingsPolicy:
    """Pre-flight embeddings gate built from a SettingsManager (no full
    snapshot). Pins the llm.ollama.url fallback fix: ollama configured only
    via the shared llm.ollama.url must classify as local."""

    def _mgr(self, settings):
        from unittest.mock import Mock

        m = Mock()
        m.get_setting.side_effect = lambda key, default=None: settings.get(
            key, default
        )
        return m

    def _enforce(self, provider, settings):
        from local_deep_research.research_library.services.rag_service_factory import (
            _enforce_embeddings_policy,
        )

        return _enforce_embeddings_policy(provider, self._mgr(settings), "u")

    def test_retired_both_coerces_to_adaptive_forcing_local(self):
        # `both` is retired (ADR-0007): it coerces to ADAPTIVE, and the RAG gate
        # classifies against the library corpus (sensitive) -> PRIVATE_ONLY ->
        # local embeddings forced -> a cloud provider is DENIED even with the
        # require_local flag left False.
        from local_deep_research.security.egress.policy import (
            PolicyDeniedError,
        )

        with pytest.raises(PolicyDeniedError):
            self._enforce(
                "openai",
                {
                    "policy.egress_scope": "both",
                    "embeddings.require_local": False,
                },
            )

    def test_adaptive_default_denies_cloud_for_library(self):
        # A missing scope falls back to the registered default ADAPTIVE.
        # The gate classifies against primary_engine="library" (local
        # nature), so adaptive resolves PRIVATE_ONLY and forces local
        # embeddings — cloud providers are denied even with the
        # require_local flag at its default False.
        from local_deep_research.security.egress.policy import (
            PolicyDeniedError,
        )

        with pytest.raises(PolicyDeniedError):
            self._enforce("openai", {"embeddings.require_local": False})

    def test_ollama_via_llm_url_allowed(self):
        # Regression: ollama configured ONLY via the shared llm.ollama.url
        # (embeddings.ollama.url unset) must be classified local. Before the
        # fix the minimal snapshot omitted llm.ollama.url -> wrongly denied.
        assert (
            self._enforce(
                "ollama",
                {
                    "embeddings.require_local": True,
                    "llm.ollama.url": "http://127.0.0.1:11434",
                },
            )
            is None
        )

    def test_openai_cloud_denied_under_require_local(self):
        with pytest.raises(PolicyDeniedError):
            self._enforce("openai", {"embeddings.require_local": True})


# ---------------------------------------------------------------------------
# Scope -> require_local coupling (120-agent audit root fix)
# ---------------------------------------------------------------------------


class TestScopeForcesLocalInference:
    """PRIVATE_ONLY means "my data stays on this box" — that only holds if
    BOTH inference paths are local. context_from_snapshot forces
    require_local_llm + require_local_embeddings under PRIVATE_ONLY even
    when the raw flags are at their default False. STRICT is search-only
    and is intentionally NOT coupled."""

    def test_private_only_forces_both_require_local(self):
        ctx = _ctx("private_only", primary="library")
        assert ctx.require_local_llm is True
        assert ctx.require_local_embeddings is True

    def test_strict_does_not_force_require_local(self):
        ctx = _ctx("strict", primary="arxiv")
        assert ctx.require_local_llm is False
        assert ctx.require_local_embeddings is False

    def test_both_does_not_force_require_local(self):
        ctx = _ctx("both", primary="arxiv")
        assert ctx.require_local_llm is False
        assert ctx.require_local_embeddings is False

    def test_explicit_flags_still_honored_under_both(self):
        # The flags remain an independent opt-in for the non-private scopes.
        ctx = context_from_snapshot(
            {
                "policy.egress_scope": "both",
                "llm.require_local_endpoint": True,
                "embeddings.require_local": True,
            },
            primary_engine="arxiv",
        )
        assert ctx.require_local_llm is True
        assert ctx.require_local_embeddings is True

    def test_embeddings_gate_fires_under_private_only_without_flag(self):
        """A PRIVATE_ONLY run with embeddings.require_local at its default
        False must STILL refuse a cloud embedder — the gate reads the
        scope-aware ctx.require_local_embeddings, not the raw flag.
        Regression: previously the gate branched on the raw flag and a
        PRIVATE_ONLY corpus could be embedded by openai."""
        from local_deep_research.embeddings.embeddings_config import (
            get_embeddings,
        )

        with pytest.raises(PolicyDeniedError):
            get_embeddings(
                provider="openai",
                settings_snapshot={"policy.egress_scope": "private_only"},
            )


# ---------------------------------------------------------------------------
# Percent-encoded hostname bypass (evaluate_url)
# ---------------------------------------------------------------------------


class TestPercentEncodedHostBypass:
    """evaluate_url must classify the DECODED host. HTTP clients decode
    percent-encoding in the netloc before connecting, so the encoded form
    must not read as a different (public/unresolvable) host than the one
    the socket actually reaches."""

    def test_encoded_private_ip_denied_under_public_only(self):
        from local_deep_research.security.egress.policy import evaluate_url

        ctx = _ctx("public_only")
        d = evaluate_url("http://192%2e168%2e1%2e1/internal", ctx)
        assert not d.allowed
        assert d.reason == "scope_mismatch_public_only"

    def test_encoded_loopback_denied_under_public_only(self):
        from local_deep_research.security.egress.policy import evaluate_url

        ctx = _ctx("public_only")
        d = evaluate_url("http://127%2e0%2e0%2e1/x", ctx)
        assert not d.allowed

    def test_encoded_private_ip_allowed_under_private_only(self):
        # The decoded host IS private, so PRIVATE_ONLY should permit it
        # (matching where the client actually connects).
        from local_deep_research.security.egress.policy import evaluate_url

        ctx = _ctx("private_only", primary="library")
        d = evaluate_url("http://192%2e168%2e1%2e1/x", ctx)
        assert d.allowed