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

"""Escape-hatch (UNPROTECTED scope) tests — ADR-0007 stage C.

UNPROTECTED disables the egress-scope restrictions (the opt-out) but must NOT
lift the hard SSRF / cloud-metadata invariant, and must lift the forced
local-inference requirements.
"""

from __future__ import annotations

from local_deep_research.security.egress.policy import (
    EgressContext,
    EgressScope,
    context_from_snapshot,
    evaluate_engine,
    evaluate_llm_endpoint,
    evaluate_retriever,
    evaluate_url,
)


def unprot_ctx(primary: str = "arxiv") -> EgressContext:
    return EgressContext(
        scope=EgressScope.UNPROTECTED,
        primary_engine=primary,
        require_local_llm=False,
        require_local_embeddings=False,
    )


def test_engine_any_allowed_under_unprotected():
    # A normally-unclassified/unknown engine is permitted by the escape hatch.
    d = evaluate_engine(
        "some_unknown_engine", unprot_ctx(), settings_snapshot={}
    )
    assert d.allowed
    assert d.reason == "egress_unprotected"


def test_retriever_allowed_under_unprotected():
    assert evaluate_retriever("anything", unprot_ctx()).allowed


def test_url_public_allowed_under_unprotected():
    assert evaluate_url("http://example.com/", unprot_ctx()).allowed


def test_metadata_ssrf_still_blocked_under_unprotected():
    # The hard invariant survives the escape hatch.
    d = evaluate_url("http://169.254.169.254/latest/meta-data/", unprot_ctx())
    assert not d.allowed


def test_cloud_llm_allowed_under_unprotected():
    # require_local_llm is False under the hatch -> cloud provider allowed.
    assert evaluate_llm_endpoint(
        "openai", unprot_ctx(), settings_snapshot={}
    ).allowed


def test_context_unprotected_lifts_local_requirements():
    # Even with the require-local flags saved on, UNPROTECTED lifts them.
    snap = {
        "policy.egress_scope": "unprotected",
        "llm.require_local_endpoint": True,
        "embeddings.require_local": True,
        "search.tool": "arxiv",
    }
    ctx = context_from_snapshot(snap, "arxiv")
    assert ctx.scope is EgressScope.UNPROTECTED
    assert ctx.require_local_llm is False
    assert ctx.require_local_embeddings is False


def test_unprotected_warning_fires_and_is_non_dismissible():
    from local_deep_research.security.egress.warnings import (
        check_unprotected_egress,
    )

    w = check_unprotected_egress("unprotected")
    assert w is not None
    assert w["type"] == "egress_unprotected"
    assert w["dismissKey"] is None  # non-dismissible
    assert check_unprotected_egress("adaptive") is None
    assert check_unprotected_egress("private_only") is None


def test_trusted_destinations_warning():
    from local_deep_research.security.egress.warnings import (
        check_trusted_destinations,
    )

    assert check_trusted_destinations([], []) is None
    w = check_trusted_destinations(["anthropic"], [])
    assert w is not None
    assert w["type"] == "egress_trusted_destinations"
    assert "anthropic" in w["message"]
    # tolerates the JSON-string shape a snapshot may carry
    w2 = check_trusted_destinations('["openai"]', "[]")
    assert w2 is not None and "openai" in w2["message"]


def test_validate_trusted_search_engines_rejects_public():
    from local_deep_research.security.egress.validators import (
        validate_trusted_search_engines,
    )

    # An inherently-public engine cannot be trusted (would launder a sink).
    err = validate_trusted_search_engines(
        {"policy.trusted_search_engines": ["searxng"]}, {}
    )
    assert err is not None and "searxng" in err["error"]
    # A local-nature store / unknown name is accepted (runtime still gates).
    assert (
        validate_trusted_search_engines(
            {"policy.trusted_search_engines": ["elasticsearch"]}, {}
        )
        is None
    )
    assert validate_trusted_search_engines({}, {}) is None