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

"""End-to-end "nothing leaves the box" tripwire for the egress policy.

This is the marquee integration test: it arms a REAL ``EgressContext`` via
``set_active_context`` (the same call the run orchestrator makes) and then
asserts that *every* layer of the stack agrees on the same decision under one
armed context:

  * the installed PEP 578 audit hook gating raw ``socket.connect`` (the
    last-line backstop that fires BEFORE the SYN is sent), and
  * the PDP call-site evaluators (``evaluate_url`` / ``evaluate_engine`` /
    ``evaluate_llm_endpoint``) that the explicit fetch / engine-factory /
    LLM-factory PEPs consult.

Unlike the focused hook test (``test_egress_audit_hook.py``) and the PDP unit
tests (``test_egress_policy.py``), this file does not pick the layers apart — it
checks that with a single armed PRIVATE_ONLY (and STRICT) context the *whole*
stack refuses public egress while still permitting the user's own loopback /
local services, and that the permissive scopes (PUBLIC_ONLY / BOTH) do NOT arm
the socket-level hook.

No outbound network is ever required: under an armed enforcing context the audit
hook raises ``PolicyDeniedError`` before ``connect`` reaches the kernel, so even
a non-blocking connect to a public IP returns the policy denial with no packets
on the wire.
"""

from __future__ import annotations

import socket

import pytest

from local_deep_research.security import (
    clear_active_context,
    get_active_context,
    install_audit_hook,
    is_audit_hook_installed,
    set_active_context,
)
from local_deep_research.security.egress.policy import (
    EgressContext,
    EgressScope,
    PolicyDeniedError,
    evaluate_engine,
    evaluate_llm_endpoint,
    evaluate_url,
)

# A public IP (Google DNS) and a public engine / cloud LLM provider used as the
# "should be denied" probes, plus a loopback IP used as the "should be allowed"
# probe. None of these need a live connection — see module docstring.
_PUBLIC_IP = "8.8.8.8"
_LOOPBACK_IP = "127.0.0.1"
_PUBLIC_ENGINE = "arxiv"  # statically classified public in the engine registry
_CLOUD_LLM = "openai"  # in _CLOUD_LLM_PROVIDERS
_METADATA_IP = "169.254.169.254"  # AWS/GCE IMDS — never permitted, any scope
_METADATA_IP_OCTAL = "0251.0376.0251.0376"  # alternate encoding of the above


@pytest.fixture(autouse=True)
def _ensure_hook_and_clean_context():
    """The audit hook is install-once-and-keep-forever (PEP 578 hooks cannot be
    removed). Make sure it is installed for the socket-layer assertions, and
    that no context leaks into or out of any test in this file — a leaked
    enforcing context would make an unrelated test's public connect raise.
    """
    install_audit_hook()
    clear_active_context()
    yield
    clear_active_context()


def _make_ctx(
    scope: EgressScope,
    *,
    require_local_llm: bool = False,
    require_local_embeddings: bool = False,
) -> EgressContext:
    """Build a fresh minimal context for ``scope``. Fresh per test so the
    per-run denial quota / DNS cache never carries across tests.
    """
    return EgressContext(
        scope=scope,
        primary_engine="wikipedia",
        require_local_llm=require_local_llm,
        require_local_embeddings=require_local_embeddings,
    )


def _attempt_connect(host: str, port: int = 80, timeout: float = 0.5):
    """Open a raw AF_INET socket and connect; return the raised exception (or
    None). We only care whether the audit hook raised ``PolicyDeniedError`` —
    the kernel-level outcome (refused / timeout) is irrelevant because the hook
    fires first.
    """
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    try:
        sock.connect((host, port))
        return None
    except PolicyDeniedError as exc:
        return exc
    except OSError as exc:
        return exc
    finally:
        sock.close()


# ---------------------------------------------------------------------------
# PRIVATE_ONLY: the full stack agrees
# ---------------------------------------------------------------------------


def test_private_only_armed_blocks_public_socket_allows_loopback():
    """With a real armed PRIVATE_ONLY context, a raw socket.connect to a public
    IP is refused by the audit hook (no network), while a connect to loopback
    is NOT refused by the hook. Allow+deny pair: if the hook stopped gating,
    the public probe would no longer raise PolicyDeniedError.
    """
    assert is_audit_hook_installed() is True
    ctx = _make_ctx(EgressScope.PRIVATE_ONLY)
    set_active_context(ctx)
    try:
        public_err = _attempt_connect(_PUBLIC_IP, 53)
        loopback_err = _attempt_connect(_LOOPBACK_IP, 1)
    finally:
        clear_active_context()

    assert isinstance(public_err, PolicyDeniedError), (
        f"armed PRIVATE_ONLY did not block public socket: {public_err!r}"
    )
    assert public_err.decision.reason == "scope_mismatch_private_only"
    assert public_err.target == _PUBLIC_IP
    assert not isinstance(loopback_err, PolicyDeniedError), (
        f"armed PRIVATE_ONLY wrongly blocked loopback: {loopback_err!r}"
    )


def test_private_only_armed_pdp_denies_public_url_engine_and_cloud_llm():
    """Under the SAME armed PRIVATE_ONLY context, every PDP call site agrees:
    evaluate_url(public) / evaluate_engine(public) / evaluate_llm_endpoint(cloud)
    all deny, while their private/local mirrors are allowed. The mirror halves
    make each assertion fail if the scope check were reverted to allow-all.
    """
    ctx = _make_ctx(
        EgressScope.PRIVATE_ONLY,
        require_local_llm=True,
        require_local_embeddings=True,
    )
    set_active_context(ctx)
    try:
        armed = get_active_context()
        assert armed is ctx

        # evaluate_url: public denied, loopback allowed.
        url_public = evaluate_url(f"http://{_PUBLIC_IP}/", armed)
        url_private = evaluate_url(f"http://{_LOOPBACK_IP}/", armed)

        # evaluate_engine: public engine denied, local engine allowed.
        engine_public = evaluate_engine(
            _PUBLIC_ENGINE, armed, settings_snapshot={}
        )
        engine_local = evaluate_engine("paperless", armed, settings_snapshot={})

        # evaluate_llm_endpoint: cloud denied, local-default allowed.
        llm_cloud = evaluate_llm_endpoint(
            _CLOUD_LLM, armed, settings_snapshot={}
        )
        llm_local = evaluate_llm_endpoint("ollama", armed, settings_snapshot={})
    finally:
        clear_active_context()

    assert not url_public.allowed
    assert url_public.reason == "scope_mismatch_private_only"
    assert url_private.allowed, "loopback URL wrongly denied under PRIVATE_ONLY"

    assert not engine_public.allowed
    assert engine_public.reason == "scope_mismatch_private_only"
    assert engine_local.allowed, (
        "local engine wrongly denied under PRIVATE_ONLY"
    )

    assert not llm_cloud.allowed
    assert llm_cloud.reason == "provider_cloud_only"
    assert llm_local.allowed, "local LLM wrongly denied under require_local_llm"


def test_private_only_armed_blocks_metadata_ip_canonical_and_octal():
    """Cloud-metadata IPs (IMDS) are NEVER permitted regardless of scope, and
    classify with the explicit blocked_metadata_ip reason — including the
    alternate octal encoding the libc resolver accepts. PRIVATE_ONLY would
    otherwise ALLOW these (they look link-local/private), so this guards the
    credential-theft path. Mirror: a real loopback URL is still allowed, proving
    the denial is the metadata guard, not a blanket refusal.
    """
    ctx = _make_ctx(EgressScope.PRIVATE_ONLY)
    set_active_context(ctx)
    try:
        armed = get_active_context()
        canonical = evaluate_url(
            f"http://{_METADATA_IP}/latest/meta-data/", armed
        )
        octal = evaluate_url(f"http://{_METADATA_IP_OCTAL}/", armed)
        loopback = evaluate_url(f"http://{_LOOPBACK_IP}/", armed)
    finally:
        clear_active_context()

    assert not canonical.allowed
    assert canonical.reason == "blocked_metadata_ip"
    assert not octal.allowed
    assert octal.reason == "blocked_metadata_ip"
    assert loopback.allowed, "loopback wrongly denied — guard is over-broad"


# ---------------------------------------------------------------------------
# STRICT: same socket-layer contract as PRIVATE_ONLY
# ---------------------------------------------------------------------------


def test_strict_armed_socket_layer_matches_private_only():
    """STRICT must behave like PRIVATE_ONLY at the socket layer: public host
    refused (with the strict_public_host reason), loopback permitted. Allow+deny
    pair against the public/loopback probes.
    """
    ctx = _make_ctx(EgressScope.STRICT)
    set_active_context(ctx)
    try:
        public_err = _attempt_connect(_PUBLIC_IP, 53)
        loopback_err = _attempt_connect(_LOOPBACK_IP, 1)
    finally:
        clear_active_context()

    assert isinstance(public_err, PolicyDeniedError), (
        f"armed STRICT did not block public socket: {public_err!r}"
    )
    assert public_err.decision.reason == "strict_public_host"
    assert not isinstance(loopback_err, PolicyDeniedError), (
        f"armed STRICT wrongly blocked loopback: {loopback_err!r}"
    )


# ---------------------------------------------------------------------------
# PUBLIC_ONLY / BOTH: do NOT arm the socket-level hook
# ---------------------------------------------------------------------------


def test_public_only_armed_does_not_arm_socket_hook():
    """PUBLIC_ONLY governs ENGINE selection, not which hosts the process may
    reach at the socket level (local Ollama / settings DB use private IPs). The
    hook must therefore stay a no-op for ALL host classes under PUBLIC_ONLY —
    otherwise legitimate local services would be false-positived. Contrast with
    the PRIVATE_ONLY tripwire above, which DOES raise on the public probe.
    """
    ctx = _make_ctx(EgressScope.PUBLIC_ONLY)
    set_active_context(ctx)
    try:
        results = {
            host: _attempt_connect(host, 1)
            for host in (_PUBLIC_IP, "192.168.42.1", _LOOPBACK_IP)
        }
    finally:
        clear_active_context()

    for host, err in results.items():
        assert not isinstance(err, PolicyDeniedError), (
            f"PUBLIC_ONLY wrongly armed the hook for {host}: {err!r}"
        )


def test_both_armed_does_not_arm_socket_hook():
    """BOTH is the permissive default — the audit hook has nothing to enforce,
    so a public connect must pass through. This is the deny-side contrast to the
    PRIVATE_ONLY public probe: same address, opposite outcome, proving the hook
    keys off the armed scope and not the address alone.
    """
    ctx = _make_ctx(EgressScope.BOTH)
    set_active_context(ctx)
    try:
        err = _attempt_connect(_PUBLIC_IP, 53)
    finally:
        clear_active_context()

    assert not isinstance(err, PolicyDeniedError), (
        f"BOTH scope wrongly armed the hook: {err!r}"
    )