Coverage for src/local_deep_research/security/egress/audit_hook.py: 86%
81 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""Process-wide ``socket.connect`` audit hook for defense-in-depth
2egress control.
4Why this exists
5---------------
6Every explicit PEP we wrote gates a single known caller — ContentFetcher,
7evaluate_llm_endpoint, MCP download_content, and so on. They are the
8primary line of defense and they remain the primary line of defense.
9This module is the secondary line — the "what about the code paths we
10haven't reviewed yet?" net. It fires on EVERY outbound socket.connect
11inside the process, so a langchain tool registered by an MCP server, a
12new contributor reaching for ``requests.get`` directly, or a third-party
13library that opens its own connection all get caught by the same rule
14the user configured.
16Threat model
17------------
18- Catches: forgotten PEP coverage, new code paths that don't know about
19 the policy, third-party library egress, prompt-injection that steers a
20 tool into raw HTTP, ``urllib.request.urlopen``, ``httpx``, ``aiohttp``,
21 ``socket.socket().connect()`` — anything that ultimately calls
22 ``socket.connect`` (almost everything network-bound in CPython).
23- Does NOT catch: a determined adversary with code execution in the LDR
24 process. They can ``clear_active_context()`` from inside the process,
25 monkey-patch the hook, ``addaudithook`` a passthrough, or open a file
26 descriptor out-of-band. For a real boundary, layer OS-level controls
27 (network namespaces, firewall rules, restricted Docker) — see
28 SECURITY.md.
30Activation model
31----------------
32- ``install_audit_hook()`` is INSTALLED ONCE at app/library startup
33 (idempotent). PEP 578 audit hooks cannot be removed — that is the
34 design, not a limitation. Once it is in, it stays.
35- The hook is INACTIVE by default: when no EgressContext is registered
36 for the current thread, every connect passes through unmodified. This
37 is the right default — random scripts importing the package and
38 pytest collectors that touch a socket would otherwise raise
39 ``PolicyDeniedError`` mid-test.
40- The hook becomes ACTIVE for a thread when a worker calls
41 ``set_active_context(ctx)``. Clear with ``clear_active_context()`` in
42 a finally to prevent leak across thread-pool reuse. The
43 ``database.thread_local_session._ThreadCleanup`` exit handler already
44 does this for the standard research worker lifecycle.
45"""
47from __future__ import annotations
49import socket
50import sys
51import threading
52from contextlib import contextmanager
53from typing import Optional
55from loguru import logger
58# ---------------------------------------------------------------------------
59# Thread-local state
60# ---------------------------------------------------------------------------
62# Per-thread "currently active" EgressContext. The audit hook reads this; a
63# worker (research run, news scheduler, library indexer, etc.) sets it at
64# task entry and clears it in finally.
65_thread_local = threading.local()
67# Re-entrancy guard. The hook itself may indirectly trigger a socket call
68# (e.g., if a future logger ever ships a record through a network sink)
69# and without this we would recurse on every connect.
70_hook_reentry = threading.local()
73def set_active_context(ctx) -> None:
74 """Register ``ctx`` as the active EgressContext for the current
75 thread. Subsequent ``socket.connect`` calls from this thread are
76 gated through the policy until ``clear_active_context()`` is called.
78 Idempotent: setting None is equivalent to clearing.
79 """
80 if ctx is None:
81 clear_active_context()
82 return
83 # An ADAPTIVE scope must be resolved to a concrete scope BEFORE it is
84 # stored: the audit hook only gates PRIVATE_ONLY/STRICT, so an ADAPTIVE
85 # context reaching the hook would make the backstop a silent no-op.
86 # Production always builds contexts via context_from_snapshot (which
87 # resolves ADAPTIVE); this fail-fast guard stops a future direct
88 # construction or test from disarming the net unnoticed.
89 try:
90 from .policy import EgressScope
92 if getattr(ctx, "scope", None) == EgressScope.ADAPTIVE:
93 raise ValueError(
94 "EgressContext.scope must be resolved to a concrete scope "
95 "before activation; got ADAPTIVE. Build the context via "
96 "context_from_snapshot(), which resolves ADAPTIVE."
97 )
98 except ImportError: # pragma: no cover - policy import always available
99 pass
100 _thread_local.egress_context = ctx
103def clear_active_context() -> None:
104 """Clear the active EgressContext for the current thread. Safe to
105 call when no context is set.
106 """
107 # ``hasattr`` + ``del`` rather than ``setattr(None)`` so the
108 # ``getattr(..., None)`` branch in the hook short-circuits without
109 # touching the policy module at all.
110 if hasattr(_thread_local, "egress_context"):
111 del _thread_local.egress_context
114def get_active_context():
115 """Return the active EgressContext for this thread, or ``None``."""
116 return getattr(_thread_local, "egress_context", None)
119@contextmanager
120def active_egress_context(ctx):
121 """Context manager: set the active EgressContext for the duration of
122 the ``with`` block and clear it on exit (even on exception).
124 Useful for unit tests, ad-hoc scripts, and any worker that wants
125 automatic cleanup without writing a try/finally.
126 """
127 # Save/restore the previous context rather than unconditionally clearing,
128 # so nesting an inner ``active_egress_context`` doesn't wipe an outer
129 # one's context on exit (the single thread-local slot otherwise loses the
130 # parent — a nested chat-in-research or test would leave the thread with
131 # NO active context, silently disarming the backstop for the parent's
132 # remaining work).
133 previous = get_active_context()
134 set_active_context(ctx)
135 try:
136 yield
137 finally:
138 if previous is not None:
139 set_active_context(previous)
140 else:
141 clear_active_context()
144# ---------------------------------------------------------------------------
145# The audit hook
146# ---------------------------------------------------------------------------
148_INSTALLED = False
149_INSTALL_LOCK = threading.Lock()
152def install_audit_hook() -> None:
153 """Install the ``sys.audit`` hook. Idempotent — calling multiple
154 times is a no-op after the first. Per PEP 578 audit hooks cannot be
155 uninstalled, so this is install-once-and-keep-forever; the activation
156 gate is the per-thread context, not the hook itself.
157 """
158 global _INSTALLED
159 with _INSTALL_LOCK:
160 if _INSTALLED:
161 return
162 sys.addaudithook(_audit_hook)
163 _INSTALLED = True
166def is_installed() -> bool:
167 return _INSTALLED
170def _extract_host(address) -> Optional[str]:
171 """Extract the destination host from a ``socket.connect`` address
172 tuple. Returns ``None`` for shapes we don't recognise (caller passes
173 through — failing to extract is never a reason to block).
175 AF_INET addresses are ``(host, port)``. AF_INET6 addresses are
176 ``(host, port, flowinfo, scope_id)``. ``host`` is a string in both.
177 """
178 if not isinstance(address, tuple) or not address: 178 ↛ 179line 178 didn't jump to line 179 because the condition on line 178 was never true
179 return None
180 host = address[0]
181 # CPython accepts a bytes/bytearray host in ``socket.connect`` and fires
182 # the ``socket.connect`` audit event with the raw bytes (verified). Decode
183 # so a bytes-encoded IP/host literal is classified instead of silently
184 # passing through the hook — otherwise a library that connects with a
185 # bytes host bypasses the PRIVATE_ONLY/STRICT backstop entirely.
186 if isinstance(host, (bytes, bytearray)):
187 try:
188 host = bytes(host).decode("ascii")
189 except (UnicodeDecodeError, ValueError):
190 return None
191 if not isinstance(host, str): 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true
192 return None
193 return host
196def _audit_hook(event: str, args: tuple) -> None:
197 """sys.audit hook. Only watches ``socket.connect``; ignores every
198 other event. Raises ``PolicyDeniedError`` when the active context
199 refuses the destination.
201 Failure mode: this hook deliberately does NOT swallow exceptions
202 raised by the policy lookup. If the policy module itself errors,
203 the connect fails loudly so the operator notices — the
204 catch-and-continue alternative would degrade the secondary line of
205 defense into "silently disabled when there's a bug." That is the
206 opposite of what a defense-in-depth net is for.
207 """
208 if event != "socket.connect": 208 ↛ 209line 208 didn't jump to line 209 because the condition on line 208 was never true
209 return
211 # No active policy → fast path. This is the common case for any
212 # thread that is not running a research worker (e.g., the Flask
213 # request handler, sqlalchemy connection pools, the news scheduler
214 # in between jobs). Skipping here keeps the hook overhead at
215 # roughly two attribute lookups per connect.
216 ctx = getattr(_thread_local, "egress_context", None)
217 if ctx is None: 217 ↛ 218line 217 didn't jump to line 218 because the condition on line 217 was never true
218 return
220 # Re-entrancy guard. Anything that runs inside the hook (logger
221 # binding, exception construction, attribute access) might
222 # indirectly trigger another connect (e.g. via a future remote log
223 # sink). Without this, the hook would recurse and stack-overflow.
224 if getattr(_hook_reentry, "active", False):
225 return
227 # Arm the re-entrancy guard BEFORE touching ``args``/``sock``/``address``
228 # or importing the policy module: any of those attribute accesses could
229 # (under instrumentation or a future remote log sink) trigger another
230 # socket.connect, and the guard must already be set to break the
231 # recursion. The try/finally guarantees the guard is cleared on EVERY
232 # exit path — the early returns below and the PolicyDeniedError raise
233 # included — so a blocked connect doesn't wedge the hook off for the
234 # rest of the thread's life.
235 _hook_reentry.active = True
236 try:
237 sock, address = args
239 # Only gate network sockets. AF_UNIX, AF_NETLINK, AF_PACKET, etc.
240 # are off-network and should not pay the policy cost.
241 family = getattr(sock, "family", None)
242 if family not in (socket.AF_INET, socket.AF_INET6): 242 ↛ 243line 242 didn't jump to line 243 because the condition on line 242 was never true
243 return
245 host = _extract_host(address)
246 if not host: 246 ↛ 247line 246 didn't jump to line 247 because the condition on line 246 was never true
247 return
249 # Scope-restriction: this hook is a "no-public-egress" net, NOT a
250 # general re-implementation of every PEP at the socket layer.
251 # PRIVATE_ONLY and STRICT are the two scopes that explicitly say
252 # "no public host should be reached." PUBLIC_ONLY governs search-
253 # ENGINE selection (the user wants public sources only) and is NOT
254 # supposed to block legitimate local infrastructure traffic — local
255 # Ollama, local embeddings, the user's own settings DB. Same logic
256 # for BOTH. Applying evaluate_url under those scopes would refuse
257 # the local LLM connect with ``scope_mismatch_public_only``, which
258 # is a false positive against the user's actual intent.
259 from .policy import EgressScope
261 if ctx.scope not in (EgressScope.PRIVATE_ONLY, EgressScope.STRICT): 261 ↛ 262line 261 didn't jump to line 262 because the condition on line 261 was never true
262 return
264 # Imported lazily to avoid a hard dependency at hook-install
265 # time. The hook is installed early during app startup and the
266 # egress_policy module pulls in network_utils, ip_ranges, etc.
267 # — none of which are problematic, but the lazy import keeps
268 # the install path minimal.
269 from .policy import (
270 PolicyDeniedError,
271 evaluate_url,
272 )
274 # Synthesize a URL so evaluate_url's existing host classification
275 # logic does the work (private-IP check, NAT64, local_hostnames
276 # match, DNS resolution if the host is not already an IP). The
277 # scheme is fixed at http:// because evaluate_url only inspects
278 # the host part — the value here is irrelevant. IPv6 literals
279 # MUST be bracketed (``[::1]``) or urlsplit's hostname parser
280 # returns None and the policy then refuses on ``no_hostname``.
281 url_host = f"[{host}]" if family == socket.AF_INET6 else host
282 decision = evaluate_url(f"http://{url_host}", ctx)
283 if not decision.allowed: 283 ↛ 292line 283 didn't jump to line 292 because the condition on line 283 was always true
284 logger.bind(policy_audit=True).warning(
285 "audit hook blocked socket.connect",
286 host=host,
287 reason=decision.reason,
288 scope=ctx.scope.value,
289 )
290 raise PolicyDeniedError(decision, target=host)
291 finally:
292 _hook_reentry.active = False