"""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}" )