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

"""Truth-table tests for the two-axis egress classification core (ADR-0007).

Stage A: the core is pure and not yet wired into any PEP. These tests pin the
four-quadrant combination rule so later wiring can't silently change it.
"""

from __future__ import annotations

import pytest

from local_deep_research.security.egress.classification import (
    Component,
    Exposure,
    Label,
    Mode,
    Role,
    Sensitivity,
    evaluate_run,
)

S = Sensitivity.SENSITIVE
NS = Sensitivity.NON_SENSITIVE
EXP = Exposure.EXPOSING
CON = Exposure.CONTAINED


# ---------------------------------------------------------------------------
# Helpers — build the components a realistic run is made of
# ---------------------------------------------------------------------------


def source(name: str, sens: Sensitivity, exp: Exposure = CON) -> Component:
    return Component(name, Role.SOURCE, Label(sens, exp))


def search_sink(name: str, exp: Exposure, sens: Sensitivity = NS) -> Component:
    return Component(name, Role.SEARCH_SINK, Label(sens, exp))


def inference(name: str, exp: Exposure) -> Component:
    return Component(name, Role.INFERENCE_SINK, Label(NS, exp))


def web_engine(name: str) -> list[Component]:
    """A public web engine: a non-sensitive source AND an exposing search sink."""
    return [source(name, NS), search_sink(name, EXP)]


def dual_risk_store(name: str) -> list[Component]:
    """Quadrant 4: a sensitive source that is also an exposing search sink."""
    return [source(name, S, EXP), search_sink(name, EXP, S)]


LOCAL_LLM = inference("llm:ollama", CON)
CLOUD_LLM = inference("llm:anthropic", EXP)


# ---------------------------------------------------------------------------
# Quadrant 1 — non-sensitive + contained: combines with anything
# ---------------------------------------------------------------------------


def test_all_non_sensitive_allows_any_sink():
    run = [
        source("collection_public", NS),
        *web_engine("google"),
        CLOUD_LLM,
    ]
    d = evaluate_run(run)
    assert d.allowed
    assert d.reason == "no_sensitive_source"


# ---------------------------------------------------------------------------
# Quadrant 2 — sensitive + contained: only with other contained sources
# ---------------------------------------------------------------------------


def test_sensitive_collection_with_local_llm_allowed():
    run = [source("collection_private", S), LOCAL_LLM]
    d = evaluate_run(run)
    assert d.allowed
    assert d.reason == "sensitive_contained"


def test_two_sensitive_collections_with_local_llm_allowed():
    run = [source("collection_a", S), source("collection_b", S), LOCAL_LLM]
    assert evaluate_run(run).allowed


def test_sensitive_collection_with_cloud_llm_denied():
    run = [source("collection_private", S), CLOUD_LLM]
    d = evaluate_run(run)
    assert not d.allowed
    assert d.reason == "sensitive_to_exposing_inference"
    assert d.offending == ("llm:anthropic",)


def test_sensitive_collection_with_public_engine_denied():
    run = [source("collection_private", S), *web_engine("arxiv"), LOCAL_LLM]
    d = evaluate_run(run)
    assert not d.allowed
    assert d.reason == "sensitive_to_exposing_search"
    assert d.offending == ("arxiv",)


# ---------------------------------------------------------------------------
# Quadrant 3 — non-sensitive + exposing: only with non-sensitive sources
# ---------------------------------------------------------------------------


def test_public_engine_with_public_collection_and_cloud_llm_allowed():
    run = [
        source("collection_public", NS),
        *web_engine("google"),
        CLOUD_LLM,
    ]
    assert evaluate_run(run).allowed


# ---------------------------------------------------------------------------
# Quadrant 4 — sensitive + exposing: solo, with contained inference
# ---------------------------------------------------------------------------


def test_dual_risk_store_solo_with_local_inference_allowed():
    run = [*dual_risk_store("elasticsearch"), LOCAL_LLM]
    d = evaluate_run(run)
    assert d.allowed
    assert d.reason == "sensitive_contained"


def test_dual_risk_store_with_cloud_inference_denied():
    run = [*dual_risk_store("elasticsearch"), CLOUD_LLM]
    d = evaluate_run(run)
    assert not d.allowed
    assert d.reason == "sensitive_to_exposing_inference"


def test_dual_risk_store_with_other_sensitive_source_denied():
    run = [
        *dual_risk_store("elasticsearch"),
        source("collection_private", S),
        LOCAL_LLM,
    ]
    d = evaluate_run(run)
    assert not d.allowed
    assert d.reason == "sensitive_to_exposing_search"
    assert d.offending == ("elasticsearch",)


def test_two_dual_risk_stores_denied():
    run = [
        *dual_risk_store("es_a"),
        *dual_risk_store("es_b"),
        LOCAL_LLM,
    ]
    assert not evaluate_run(run).allowed


# ---------------------------------------------------------------------------
# Exposing embeddings count as an exposing inference sink
# ---------------------------------------------------------------------------


def test_sensitive_with_cloud_embeddings_denied():
    run = [
        source("collection_private", S),
        LOCAL_LLM,
        inference("embeddings:openai", EXP),
    ]
    d = evaluate_run(run)
    assert not d.allowed
    assert d.reason == "sensitive_to_exposing_inference"
    assert d.offending == ("embeddings:openai",)


# ---------------------------------------------------------------------------
# Permissive mode suspends the invariant (caller still warns)
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "run",
    [
        [source("collection_private", S), CLOUD_LLM],
        [source("collection_private", S), *web_engine("arxiv"), LOCAL_LLM],
        [*dual_risk_store("es"), CLOUD_LLM],
    ],
)
def test_permissive_mode_always_allows(run):
    enforced = evaluate_run(run)
    assert not enforced.allowed  # denied when enforcing
    d = evaluate_run(run, mode=Mode.PERMISSIVE)
    assert d.allowed
    # The would-be-enforced diagnostic is preserved for the stage-C banner.
    assert d.reason == f"permissive:{enforced.reason}"
    assert d.offending == enforced.offending


def test_quadrant4_store_plus_second_exposing_engine_denied():
    # A dual-risk store is only allowed SOLO. Add a public web engine and the
    # store's sensitive data can now reach that foreign exposing sink -> deny.
    # The store's OWN sink is not offending (returning its data to itself is
    # not a new leak); only the foreign "arxiv" sink is reported.
    run = [*dual_risk_store("es"), *web_engine("arxiv"), LOCAL_LLM]
    d = evaluate_run(run)
    assert not d.allowed
    assert d.reason == "sensitive_to_exposing_search"
    assert d.offending == ("arxiv",)


def test_name_collision_does_not_exempt_non_sensitive_sink():
    # A NON-sensitive exposing search sink that merely shares a name with an
    # unrelated sensitive source must NOT be treated as the solo dual-risk case.
    run = [
        Component("x", Role.SOURCE, Label(S, CON)),  # sensitive source "x"
        Component("x", Role.SEARCH_SINK, Label(NS, EXP)),  # unrelated sink "x"
        LOCAL_LLM,
    ]
    d = evaluate_run(run)
    assert not d.allowed
    assert d.reason == "sensitive_to_exposing_search"


def test_multi_violation_inference_takes_precedence():
    run = [
        source("collection_private", S),
        *web_engine("google"),
        CLOUD_LLM,
    ]
    # Inference leak is checked first; the exposing search engine is masked.
    d = evaluate_run(run)
    assert not d.allowed
    assert d.reason == "sensitive_to_exposing_inference"


# ---------------------------------------------------------------------------
# Degenerate inputs
# ---------------------------------------------------------------------------


def test_empty_run_allowed():
    assert evaluate_run([]).allowed


def test_only_contained_inference_no_sources_allowed():
    assert evaluate_run([LOCAL_LLM]).allowed