Coverage for src/local_deep_research/security/egress/policy.py: 92%
460 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""Egress policy module: central PDP for search engines, LLM endpoints,
2embeddings, and URL fetches.
4This is an in-process correctness guardrail, NOT a hard security boundary.
5It defends against honest misconfiguration, prompt-injection-induced
6URL fetches, accidental egress, and the LangGraph silent-expansion bug.
7It does NOT defend against compromised dependencies, code-execution
8in the LDR process, or a determined adversary who can modify the policy
9module itself. Operators needing a hard boundary should layer OS-level
10controls (network namespaces, firewall rules, restricted Docker).
12Vocabulary borrowed from XACML / zero-trust:
13 PDP — Policy Decision Point (the evaluate_* functions in this module)
14 PEP — Policy Enforcement Point (the call sites that consult the PDP)
15"""
17from __future__ import annotations
19import json
20import socket
21import threading
22from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout
23from dataclasses import dataclass, field
24from enum import Enum
25from typing import Optional
26from urllib.parse import unquote, urlsplit
28from loguru import logger
30from ..network_utils import is_private_ip
31from ...utilities.type_utils import unwrap_setting
34# Bounded DNS lookup timeout; getaddrinfo has no native timeout kwarg, so
35# we run it inside a single-shot ThreadPoolExecutor and abandon the
36# Future after the timeout elapses. This avoids the previous
37# socket.setdefaulttimeout() approach, which mutated a process-global
38# setting and could corrupt unrelated network code under concurrency.
39_DNS_TIMEOUT_SEC = 2.0
42# Hard limit: after this many denied fetches in a single run, fail closed
43# even on otherwise-legal fetches. Prevents an exhaust attack via a malicious
44# indexed document looping the agent through hundreds of denied URLs.
45MAX_DENIED_FETCHES_PER_RUN = 50
47# Only SECURITY-relevant denials count toward the per-run quota. Benign parse
48# failures (a mailto:/ftp:/tel: link or a malformed href scraped from a page)
49# are common in legitimate documents and must NOT exhaust the budget — doing
50# so would make a long PUBLIC_ONLY run start refusing legitimate public URLs
51# mid-run. These reasons are still audit-logged; they just don't tick the quota.
52# ``dangerous_scheme`` (javascript:/data:/file:/… hrefs) belongs here too: these
53# are non-fetchable, non-network schemes scraped from ordinary HTML (onclick
54# handlers, inline data: URIs), so — like ``unsupported_scheme`` — they can't
55# cause egress and shouldn't let a doc full of data: URIs exhaust the budget.
56_NON_QUOTA_DENIAL_REASONS = frozenset(
57 {"url_malformed", "unsupported_scheme", "no_hostname", "dangerous_scheme"}
58)
61class EgressScope(str, Enum):
62 """User-declared egress boundary for a research run.
64 - STRICT: only the user's primary engine; no expansion at all.
65 - PUBLIC_ONLY: any public (external web/academic) engine.
66 - PRIVATE_ONLY: any private (local collection / library) engine.
67 - BOTH: INTERNAL ONLY — the resolved scope for an unclassifiable ADAPTIVE
68 primary (allow any classified engine, no local-inference coupling). It is
69 NO LONGER a user-selectable scope: `both` was the "muddy middle" (ADR-0007
70 Problem 3), retired in favour of `adaptive` + per-collection `is_public`.
71 A residual user `both` (stored / env / queued) is coerced to `adaptive`
72 by ``context_from_snapshot`` and migration 0019.
73 - ADAPTIVE: scope FOLLOWS the primary engine — a concrete private
74 primary behaves as PRIVATE_ONLY, a concrete public primary as
75 PUBLIC_ONLY, and an unclassifiable primary as BOTH.
76 The default: most users never touch scope and "it just matches my
77 main engine." Resolved to a concrete scope at context construction;
78 the stored EgressContext carries the RESOLVED scope, not ADAPTIVE.
79 - UNPROTECTED: escape hatch — egress protection is DISABLED for the run.
80 Any engine / URL / provider is permitted; the hard SSRF and
81 cloud-metadata blocks in ``evaluate_url`` still apply. Not recommended.
82 """
84 STRICT = "strict"
85 PUBLIC_ONLY = "public_only"
86 PRIVATE_ONLY = "private_only"
87 BOTH = "both"
88 ADAPTIVE = "adaptive"
89 UNPROTECTED = "unprotected"
92# Code-side single source of truth for the default egress scope, used by
93# every reader that needs a fallback for a MISSING policy.egress_scope key
94# (partial snapshots from the programmatic API, un-bootstrapped settings
95# DBs). Import THIS instead of hardcoding a string literal: scattered
96# literals are how the registry default ("adaptive") and the code
97# fallbacks ("both") drifted apart in the first place. Must match the
98# registered default in defaults/default_settings.json — pinned by
99# tests/security/test_egress_policy.py::
100# test_default_scope_constant_matches_registry.
101DEFAULT_EGRESS_SCOPE: str = EgressScope.ADAPTIVE.value
104@dataclass(frozen=True)
105class EgressContext:
106 """Frozen per-run policy snapshot.
108 Constructed once via ``context_from_snapshot()`` at run-start.
109 The dataclass is frozen, but mutable internals (``_dns_cache``,
110 ``_fetch_denial_count``) use ``field(init=False, default_factory=...)``
111 so they can accumulate state during a run. Counters live inside a
112 dict because direct ``int`` field reassignment fails on ``frozen=True``.
113 """
115 scope: EgressScope
116 primary_engine: str
117 require_local_llm: bool
118 require_local_embeddings: bool
119 local_hostnames: tuple[str, ...] = ()
120 username: Optional[str] = None
121 # Mutable internal state (kept off the public surface).
122 _dns_cache: dict = field(init=False, default_factory=dict, repr=False)
123 _fetch_denial_count: dict = field(
124 init=False, default_factory=lambda: {"count": 0}, repr=False
125 )
126 # Guards _dns_cache + _fetch_denial_count mutations. Scope is the
127 # cache writes only — DNS I/O must NOT happen while holding the lock,
128 # or concurrent subagent threads serialize on each other's lookups.
129 _lock: threading.RLock = field(
130 init=False, default_factory=threading.RLock, repr=False
131 )
134@dataclass(frozen=True)
135class Decision:
136 """Result of a PDP evaluation. ``reason`` is a short machine code
137 (e.g. ``"unclassified"``, ``"scope_mismatch"``) — never the user's
138 rejected query or URL content.
139 """
141 allowed: bool
142 reason: str
145class PolicyDeniedError(RuntimeError):
146 """Raised by PEPs when a policy decision is hard-stop denial.
148 Raising (rather than returning a graceful empty result) ensures
149 consistent denial latency, which mitigates the LangGraph timing-leak
150 pattern where an LLM could infer policy state from how fast a denied
151 tool call returns.
152 """
154 def __init__(self, decision: Decision, target: str = ""):
155 self.decision = decision
156 self.target = target
157 super().__init__(f"policy_denied: {decision.reason}")
160def _is_nat64_wrapped_metadata(hostname: str) -> bool:
161 """True iff ``hostname`` parses as an IPv6 NAT64 address wrapping a
162 cloud-metadata IPv4 (AWS / Azure / GCE / etc).
164 The wrapping passes ``is_private_ip`` (because IPv6 link-local
165 ranges match) but the embedded IPv4 actually reaches the metadata
166 endpoint. We classify these as PUBLIC so STRICT and PRIVATE_ONLY
167 refuse to fetch them.
168 """
169 try:
170 import ipaddress
172 from ..ssrf_validator import is_nat64_wrapped_metadata_ip
174 candidate = hostname
175 if candidate.startswith("[") and candidate.endswith("]"):
176 candidate = candidate[1:-1]
177 ip = ipaddress.ip_address(candidate)
178 return is_nat64_wrapped_metadata_ip(ip)
179 except (ValueError, TypeError):
180 return False
181 except Exception: # pragma: no cover - defensive
182 return False
185def _resolve_with_timeout(hostname: str) -> Optional[list]:
186 """Resolve ``hostname`` via getaddrinfo with a bounded timeout.
188 Returns the addrinfo list on success or None on timeout / lookup
189 failure. getaddrinfo has no native timeout, so we drive it from a
190 single-shot worker thread and abandon the Future on timeout — far
191 safer than socket.setdefaulttimeout(), which mutates a
192 process-global setting and races with unrelated network code.
193 """
195 def _do_lookup():
196 return socket.getaddrinfo(hostname, None, type=socket.SOCK_STREAM)
198 # NB: do NOT use ``with ThreadPoolExecutor(...)`` here. The context
199 # manager's __exit__ calls shutdown(wait=True), which blocks until the
200 # worker thread finishes — so a hung getaddrinfo would defeat the whole
201 # point of the timeout (the call would return only after the OS DNS
202 # timeout, not after _DNS_TIMEOUT_SEC). Instead we abandon the worker on
203 # timeout via shutdown(wait=False): the caller returns promptly and the
204 # orphaned thread dies on its own when getaddrinfo eventually returns.
205 executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="ldr-dns")
206 future = executor.submit(_do_lookup)
207 try:
208 return future.result(timeout=_DNS_TIMEOUT_SEC)
209 except (FutureTimeout, socket.gaierror, socket.timeout, OSError):
210 return None
211 except Exception: # pragma: no cover - defensive
212 return None
213 finally:
214 # wait=False so a timed-out lookup does not re-block here; a
215 # completed lookup leaves an idle thread that shuts down at once.
216 executor.shutdown(wait=False)
219_UNCACHED = object()
222def _cache_classification(
223 ctx: EgressContext, hostname: str, value: bool
224) -> bool:
225 """First-writer-wins cache set; returns the value now in the cache.
227 Concurrent subagent threads may classify the same hostname outside the
228 lock (the DNS lookup is intentionally unsynchronized). If a round-robin
229 DNS name resolves to a private IP for one thread and a public IP for
230 another, last-writer-wins would let a later call flip an earlier call's
231 result — cache *incoherence* that could relax PRIVATE_ONLY/STRICT on a
232 subsequent fetch. Pinning the first writer's value makes the per-run
233 classification stable and deterministic regardless of completion order.
234 """
235 with ctx._lock:
236 existing = ctx._dns_cache.get(hostname, _UNCACHED)
237 if existing is not _UNCACHED:
238 return existing
239 ctx._dns_cache[hostname] = value
240 return value
243def _classify_host(
244 hostname: str, ctx: EgressContext, allow_dns: bool = True
245) -> Optional[bool]:
246 """Classify a hostname as local (True) / public (False) / unknown (None).
248 Uses the run-scoped DNS cache to avoid repeated ``getaddrinfo`` lookups.
249 Falls back to public on DNS timeout (fail-safe).
251 Lock discipline: cache reads and writes are guarded by ``ctx._lock``;
252 the DNS lookup itself runs OUTSIDE the lock so concurrent subagent
253 threads don't serialize on each other's DNS calls. Cache writes are
254 first-writer-wins (``_cache_classification``) so a hostname's
255 classification is stable for the run even under concurrent disagreeing
256 lookups.
257 """
258 if not hostname: 258 ↛ 259line 258 didn't jump to line 259 because the condition on line 258 was never true
259 return None
261 # Cache hit — read under the lock so concurrent writers don't tear
262 # the dict.
263 with ctx._lock:
264 if hostname in ctx._dns_cache:
265 return ctx._dns_cache[hostname]
267 # User-declared local hostnames override DNS classification.
268 if hostname in ctx.local_hostnames: 268 ↛ 269line 268 didn't jump to line 269 because the condition on line 268 was never true
269 return _cache_classification(ctx, hostname, True)
271 # NAT64-wrapped metadata IPs (e.g. 64:ff9b::169.254.169.254) classify
272 # as link-local by is_private_ip but actually wrap AWS/GCE instance
273 # metadata. Force them to PUBLIC so STRICT/PRIVATE_ONLY don't allow
274 # them. Check happens BEFORE is_private_ip so the wrapping wins.
275 if _is_nat64_wrapped_metadata(hostname):
276 return _cache_classification(ctx, hostname, False)
278 # String-literal check first (no DNS needed for IPs and known literals).
279 # If the helper itself raises (malformed hostname, broken stdlib edge case),
280 # fall through to DNS resolution rather than fail the whole evaluation —
281 # the DNS path has its own bounded timeout and error handling.
282 from ..ssrf_validator import is_ip_blocked
284 try:
285 # A literal cloud-metadata IP (169.254.169.254, fd00:ec2::254, …) passes
286 # is_private_ip (link-local) but must NOT be treated as local — otherwise
287 # STRICT/PRIVATE_ONLY would accept an IMDS SSRF target as a "local"
288 # inference/search host. Mirror the DNS-branch metadata block below.
289 # is_ip_blocked returns False for non-IP hostnames (ValueError → False),
290 # so legitimate hostnames still fall through to is_private_ip / DNS.
291 if is_ip_blocked(
292 hostname, allow_localhost=True, allow_private_ips=True
293 ):
294 return _cache_classification(ctx, hostname, False)
295 if is_private_ip(hostname):
296 return _cache_classification(ctx, hostname, True)
297 except Exception as exc:
298 logger.debug(
299 "is_private_ip raised on hostname, falling back to DNS",
300 error=str(exc),
301 )
303 # DNS-resolved classification — runs OUTSIDE the lock.
304 if not allow_dns: 304 ↛ 311line 304 didn't jump to line 311 because the condition on line 304 was never true
305 # The caller (the advisory warning-banner render path) opted OUT of
306 # the synchronous getaddrinfo so a settings-page render never blocks
307 # up to _DNS_TIMEOUT_SEC on a lookup. Return "unknown" so adaptive
308 # scope resolution falls back to the engine's static classification.
309 # Deliberately NOT cached — a later enforcement-path call (allow_dns
310 # default True) must still resolve this host for real.
311 return None
312 addr_info = _resolve_with_timeout(hostname)
313 if addr_info is None:
314 # Fail-safe: unresolvable / timeout → treat as public.
315 return _cache_classification(ctx, hostname, False)
317 # If any resolved IP is private (and not a NAT64 metadata wrap),
318 # classify as local.
319 for entry in addr_info:
320 ip_str = entry[4][0]
321 try:
322 # A hostname that resolves to a cloud-metadata IP must NOT be
323 # treated as local: link-local metadata IPs (169.254.169.254, …)
324 # pass is_private_ip, which would classify the host local and let
325 # STRICT/PRIVATE_ONLY fetch it. Classify as public so those scopes
326 # refuse it — mirrors the literal-IP metadata block in evaluate_url
327 # and the NAT64 handling just below. (Under PUBLIC_ONLY/BOTH the
328 # SSRF validator at the actual fetch is the metadata backstop.)
329 if is_ip_blocked(
330 ip_str, allow_localhost=True, allow_private_ips=True
331 ):
332 return _cache_classification(ctx, hostname, False)
333 if _is_nat64_wrapped_metadata(ip_str): 333 ↛ 334line 333 didn't jump to line 334 because the condition on line 333 was never true
334 return _cache_classification(ctx, hostname, False)
335 if is_private_ip(ip_str):
336 return _cache_classification(ctx, hostname, True)
337 except Exception: # pragma: no cover - defensive
338 continue
339 return _cache_classification(ctx, hostname, False)
342def _get_engine_class(engine_name: str):
343 """Lazy load the engine class from the registry to avoid circular imports."""
344 # Import is lazy to break the dependency cycle:
345 # security/* → engines/* → security/*
346 from ...web_search_engines.engine_registry import ENGINE_REGISTRY
347 from ..module_whitelist import get_safe_module_class
349 entry = ENGINE_REGISTRY.get(engine_name)
350 if entry is None:
351 return None
352 try:
353 return get_safe_module_class(entry.module_path, entry.class_name)
354 except Exception:
355 return None
358def _engine_flags(engine_cls):
359 """Read the (is_public, is_local, url_setting) triple from an engine class.
361 Uses ``is True`` / ``is False`` semantics: a missing attribute is treated
362 as "not declared", which is distinct from an attribute explicitly set
363 to ``False``.
364 """
365 is_public = getattr(engine_cls, "is_public", None)
366 is_local = getattr(engine_cls, "is_local", None)
367 url_setting = getattr(engine_cls, "url_setting", None)
368 return is_public, is_local, url_setting
371def _resolve_collection_is_public(
372 engine_name: str, username: Optional[str]
373) -> bool:
374 """Look up a collection engine's per-collection ``is_public`` flag.
376 ``engine_name`` is ``"collection_<uuid>"`` or ``"library"``. The
377 aggregate ``library`` is always private (it spans all local docs).
378 Fails closed to private (False) on any lookup error or missing row —
379 a collection we can't classify must NOT be treated as public.
380 """
381 if engine_name == "library" or not engine_name.startswith("collection_"):
382 return False
383 collection_id = engine_name[len("collection_") :]
384 try:
385 from ...database.models.library import Collection
386 from ...database.session_context import get_user_db_session
388 with get_user_db_session(username) as session:
389 row = (
390 session.query(Collection)
391 .filter(Collection.id == collection_id)
392 .first()
393 )
394 return bool(getattr(row, "is_public", False)) if row else False
395 except Exception: # noqa: silent-exception - fail closed to private
396 logger.debug(
397 "could not resolve collection is_public; treating as private",
398 engine=engine_name,
399 )
400 return False
403def _engine_bucket(
404 engine_name: str,
405 ctx: EgressContext,
406 settings_snapshot: dict,
407 metadata: Optional[dict] = None,
408 allow_dns: bool = True,
409):
410 """Return the ``(is_public, is_local)`` classification for an engine based
411 on its **Python class flags**, with a fail-up URL override.
413 The class-level ``is_public`` / ``is_local`` flags express the engine's
414 *nature* — whether it queries the public internet or searches local data —
415 regardless of where the engine's server happens to be hosted. This is used
416 by ``evaluate_engine`` and ``_resolve_adaptive_scope`` so that e.g. a
417 locally-hosted SearXNG (which proxies to Google/Bing) is still treated as
418 a public engine.
420 The URL override is asymmetric and can only make the classification
421 MORE restrictive: a local-nature engine whose ``url_setting`` resolves
422 to a public host is reclassified public (queries would leave the box),
423 while a public-nature engine on a local URL stays public.
425 - Static engines: read declared ``is_public``/``is_local`` flags directly.
426 - ``library`` / ``collection_<uuid>``: per-collection ``is_public``
427 flag (from ``metadata`` when the caller has the engine config, else a
428 direct DB lookup), defaulting private.
429 - Anything else (unknown): ``(None, None)``.
430 """
431 engine_cls = _get_engine_class(engine_name)
432 if engine_cls is None:
433 if engine_name == "library" or engine_name.startswith("collection_"):
434 if metadata is not None and metadata.get("is_public") is not None:
435 is_public = bool(metadata.get("is_public"))
436 else:
437 is_public = _resolve_collection_is_public(
438 engine_name, ctx.username
439 )
440 # A collection is ALWAYS a local knowledge base — it lives on this
441 # machine — so it stays searchable under PRIVATE_ONLY regardless of
442 # its public flag. ``is_public`` is ADDITIVE, not exclusive: marking
443 # a collection public (non-sensitive content — papers you fetched,
444 # public publications) ALSO makes it eligible under PUBLIC_ONLY and
445 # OK to process with cloud inference. A private collection is
446 # local-only and excluded from public-scope runs. Hence
447 # ``(is_public, is_local=True)`` rather than the old mutually-
448 # exclusive ``(is_public, not is_public)`` which wrongly hid a
449 # public collection from private runs.
450 return is_public, True
451 return None, None
452 is_public, is_local, url_setting = _engine_flags(engine_cls)
453 # Fail-up URL override (asymmetric, only ever MORE restrictive): a
454 # local-data engine (Elasticsearch, Paperless) whose configured URL
455 # resolves to a PUBLIC host is reclassified as public — querying it
456 # sends the user's queries off the box, so PRIVATE_ONLY must deny it
457 # at selection time (matching pre-static-flags behavior). The reverse
458 # direction is deliberately NOT applied: a public-nature engine
459 # (SearXNG) hosted on localhost stays public, because its data source
460 # is the internet regardless of where the proxy runs.
461 if url_setting and is_local is True and is_public is not True:
462 host_classification = _classify_engine_url(
463 url_setting, settings_snapshot, ctx, allow_dns=allow_dns
464 )
465 if host_classification is False:
466 # False from _classify_engine_url == "connect target is a
467 # PUBLIC host" -> fail up to public.
468 return True, False
469 return is_public, is_local
472# Denial reasons that justify dropping an engine from an LLM candidate
473# list. Only ACTIVE policy denials qualify — see filter_engines_by_egress.
474_PREFILTER_STRIP_REASONS = frozenset(
475 {
476 "scope_mismatch_public_only",
477 "scope_mismatch_private_only",
478 "strict_not_primary",
479 # Registered engine without classification flags: the factory's
480 # evaluate_engine call fails it closed too, so stripping matches
481 # enforcement exactly.
482 "unclassified",
483 }
484)
487def filter_engines_by_egress(
488 engine_names: list[str],
489 ctx: EgressContext,
490 settings_snapshot: dict,
491) -> list[str]:
492 """ADVISORY pre-filter for a candidate engine list: remove engines the
493 egress scope actively denies so the LLM doesn't waste selection slots
494 on them. Uses the engine's static class flags (``is_public`` /
495 ``is_local``) — a locally-hosted SearXNG is still treated as public
496 because it queries the internet.
498 This is a UX optimization in FRONT of the factory PEP, not an
499 enforcement point — so it must never be STRICTER than the factory.
500 Names unknown to the static registry (``engine_unknown``) are KEPT
501 (except under STRICT, where the not-the-primary gate fires before the
502 registry lookup — exactly as it does in the factory):
503 they may be retriever-backed or dynamically injected engines that the
504 factory evaluates via its own retriever path; dropping them here would
505 silently hide legitimate engines from LLM selection. The factory
506 still denies anything truly disallowed at instantiation.
508 Callers: the LangGraph tool builder, or any code that builds a
509 selection set before presenting it to an LLM.
510 """
511 allowed: list[str] = []
512 for name in engine_names:
513 decision = evaluate_engine(
514 name, ctx, settings_snapshot=settings_snapshot
515 )
516 if decision.allowed or decision.reason not in _PREFILTER_STRIP_REASONS:
517 allowed.append(name)
518 return allowed
521def filter_candidates_by_egress(
522 engine_names: list[str],
523 settings_snapshot: Optional[dict],
524) -> list[str]:
525 """Best-effort scope pre-filter for an LLM candidate list.
527 Wraps ``filter_engines_by_egress`` with the snapshot plumbing
528 LLM-selection callers share: read ``policy.egress_scope`` /
529 ``search.tool`` from the snapshot (flat or ``{"value": ...}`` shaped),
530 build the context, filter. Returns the input list unchanged when the
531 snapshot is missing, the scope is BOTH (nothing to strip), or anything
532 fails — the factory PEP still enforces at child instantiation, so this
533 helper must never break engine selection.
534 """
535 if not settings_snapshot:
536 return engine_names
537 try:
538 scope_raw = str(
539 _get_setting_value(
540 settings_snapshot, "policy.egress_scope", DEFAULT_EGRESS_SCOPE
541 )
542 or DEFAULT_EGRESS_SCOPE
543 ).lower()
544 # BOTH (or an unrecognized value) strips nothing at selection time;
545 # adaptive is included because it can RESOLVE to a restrictive scope.
546 if scope_raw not in (
547 "private_only",
548 "public_only",
549 "strict",
550 "adaptive",
551 ):
552 return engine_names
553 primary = resolve_run_primary_engine(settings_snapshot)
554 ctx = context_from_snapshot(
555 settings_snapshot,
556 primary,
557 username=settings_snapshot.get("_username"),
558 )
559 return filter_engines_by_egress(engine_names, ctx, settings_snapshot)
560 except Exception: # noqa: silent-exception - advisory pre-filter only
561 logger.bind(policy_audit=True).debug(
562 "egress candidate pre-filter failed; "
563 "factory PEP still enforces at child instantiation"
564 )
565 return engine_names
568def evaluate_engine(
569 engine_name: str,
570 ctx: EgressContext,
571 *,
572 settings_snapshot: dict,
573 metadata: Optional[dict] = None,
574) -> Decision:
575 """Decide whether an engine may be instantiated under the current policy.
577 Allows iff (a) the engine's classification is compatible with the scope's
578 bucket AND (b) if scope == STRICT, the engine name equals the primary.
579 Unclassified engines always fail closed.
581 ``metadata`` (optional) carries the engine's config entry (e.g. the
582 per-collection ``is_public`` from search_config) so callers that already
583 have it avoid a redundant DB lookup; when absent it's resolved directly.
584 """
585 if settings_snapshot is None:
586 return Decision(False, "no_snapshot")
587 try:
588 # UNPROTECTED: egress protection disabled for this run (escape hatch)
589 # — any engine is permitted. SSRF/metadata blocks still apply at
590 # evaluate_url; this only lifts the egress-scope engine gate.
591 if ctx.scope == EgressScope.UNPROTECTED:
592 return Decision(True, "egress_unprotected")
594 # STRICT: only the primary engine is permitted.
595 if (
596 ctx.scope == EgressScope.STRICT
597 and engine_name != ctx.primary_engine
598 ):
599 return Decision(False, "strict_not_primary")
601 # A truly unknown engine (not in the static registry and not a
602 # collection) fails closed as engine_unknown. ``library`` /
603 # ``collection_*`` are classified by _engine_bucket below.
604 is_collection = engine_name == "library" or engine_name.startswith(
605 "collection_"
606 )
607 if _get_engine_class(engine_name) is None and not is_collection:
608 return Decision(False, "engine_unknown")
610 is_public, is_local = _engine_bucket(
611 engine_name, ctx, settings_snapshot, metadata
612 )
614 # Fail-closed for unclassified engines.
615 if is_public is not True and is_local is not True: 615 ↛ 616line 615 didn't jump to line 616 because the condition on line 615 was never true
616 return Decision(False, "unclassified")
618 # Scope/bucket compatibility.
619 if ctx.scope == EgressScope.PUBLIC_ONLY and is_public is not True:
620 return Decision(False, "scope_mismatch_public_only")
621 if ctx.scope == EgressScope.PRIVATE_ONLY and is_local is not True:
622 return Decision(False, "scope_mismatch_private_only")
624 return Decision(True, "allowed")
625 except Exception: # pragma: no cover - defensive
626 logger.bind(policy_audit=True).exception(
627 "evaluate_engine internal error", engine=engine_name
628 )
629 return Decision(False, "internal_error")
632def _classify_engine_url(
633 url_setting: str,
634 settings_snapshot: dict,
635 ctx: EgressContext,
636 allow_dns: bool = True,
637) -> Optional[bool]:
638 """Classify an engine's configured URL via DNS resolution.
640 For list-typed settings (Elasticsearch ``hosts``), returns False (public)
641 if ANY entry classifies as public — safer fail-up.
642 Returns True if local, False if public, None if undetermined.
643 """
644 value = _get_setting_value(settings_snapshot, url_setting, None)
645 if not value:
646 return None
648 entries = value if isinstance(value, list) else [value]
649 any_public = False
650 any_local = False
651 for entry in entries:
652 if not isinstance(entry, str): 652 ↛ 653line 652 didn't jump to line 653 because the condition on line 652 was never true
653 continue
654 try:
655 parsed = urlsplit(entry if "://" in entry else f"http://{entry}")
656 hostname = parsed.hostname
657 except Exception:
658 continue
659 if not hostname: 659 ↛ 660line 659 didn't jump to line 660 because the condition on line 659 was never true
660 continue
661 # Decode percent-encoding before classifying, identically to
662 # evaluate_url: the HTTP client decodes the host before connecting,
663 # so "192%2e168%2e1%2e1" must be classified as 192.168.1.1, not as
664 # an unresolvable literal.
665 hostname = unquote(hostname)
666 result = _classify_host(hostname, ctx, allow_dns=allow_dns)
667 if result is True:
668 any_local = True
669 elif result is False: 669 ↛ 651line 669 didn't jump to line 651 because the condition on line 669 was always true
670 any_public = True
672 if any_public:
673 return False # "any public" wins under list-typed settings
674 if any_local: 674 ↛ 676line 674 didn't jump to line 676 because the condition on line 674 was always true
675 return True
676 return None
679_CLOUD_LLM_PROVIDERS = frozenset(
680 {
681 "openai",
682 "anthropic",
683 "google",
684 "openrouter",
685 "requesty",
686 "orcarouter",
687 "deepseek",
688 "xai",
689 "ionos",
690 }
691)
693# Providers that default to a localhost endpoint and ship without any
694# remote routing. The LLM gate at ``config/llm_config.py`` uses this set
695# as a SNAPSHOT-LESS ALLOW-LIST: when ``get_llm`` is invoked with no
696# settings snapshot (background helpers, scaffolding paths, tests), only
697# these providers may proceed; everything else — including ambiguous
698# providers like ``openai_endpoint`` and any future cloud provider not
699# yet enumerated in ``_CLOUD_LLM_PROVIDERS`` — fails closed at the gate.
700# Keeping the set tight is the point: anything new defaults to refused.
701_LOCAL_DEFAULT_LLM_PROVIDERS = frozenset(
702 {
703 "ollama",
704 "lmstudio",
705 "llamacpp",
706 }
707)
709# Embeddings analogue of ``_LOCAL_DEFAULT_LLM_PROVIDERS``. Used by the
710# embeddings gate as a SNAPSHOT-LESS ALLOW-LIST: when ``get_embeddings``
711# runs without a settings snapshot (the ``embeddings.require_local`` toggle
712# is then unreadable), only these localhost-default providers may proceed.
713# ``openai`` (and any future cloud embeddings provider) fails closed —
714# matching the LLM gate so a snapshot-less background caller can't ship the
715# local corpus to a cloud embedder. ``sentence_transformers`` runs fully
716# in-process; ``ollama`` defaults to a localhost endpoint.
717_LOCAL_DEFAULT_EMBEDDING_PROVIDERS = frozenset(
718 {
719 "sentence_transformers",
720 "ollama",
721 }
722)
725def _is_user_registered_llm(provider: str) -> bool:
726 """True when ``provider`` is a user-registered in-process LLM.
728 The programmatic API (``quick_summary(llms={"mock": ...})``) and plugins
729 register LLM objects in the LLM registry. Built-in providers are ALSO
730 auto-registered there by ``discover_providers()``, so registry membership
731 alone cannot discriminate — a name that is registered but NOT one of the
732 auto-discovered built-ins is user-supplied in-process code. Such objects
733 carry no configurable endpoint the PDP could classify (denying them with
734 ``provider_url_unset`` just breaks the documented offline workflow), the
735 operator injected them deliberately at code level, and under
736 PRIVATE_ONLY/STRICT the PEP-578 audit hook still blocks any stray
737 outbound socket they might open.
739 A user registration that *shadows* a built-in name (e.g. "openai") is NOT
740 treated as user code: the discovered-name check keeps it on the strict
741 path, and the ``_CLOUD_LLM_PROVIDERS`` gate fires first anyway.
743 Fails closed (False) on any error.
744 """
745 try:
746 from ...llm.llm_registry import is_llm_registered
747 from ...llm.providers import discover_providers
748 from ...llm.providers.base import normalize_provider
750 if not provider or not is_llm_registered(provider):
751 return False
752 discovered = {normalize_provider(key) for key in discover_providers()}
753 return normalize_provider(provider) not in discovered
754 except Exception: # pragma: no cover - defensive
755 logger.bind(policy_audit=True).exception(
756 "user-registered LLM check failed", provider=provider
757 )
758 return False
761def evaluate_llm_endpoint(
762 provider: str,
763 ctx: EgressContext,
764 *,
765 settings_snapshot: dict,
766) -> Decision:
767 """Decide whether an LLM provider may be instantiated.
769 Only meaningful when ``ctx.require_local_llm`` is True; otherwise allow.
770 """
771 if settings_snapshot is None:
772 return Decision(False, "no_snapshot")
773 try:
774 if not ctx.require_local_llm:
775 return Decision(True, "no_local_requirement")
777 # Hard-cloud providers always fail under require_local_llm.
778 if provider in _CLOUD_LLM_PROVIDERS:
779 return Decision(False, "provider_cloud_only")
781 # User-registered in-process LLMs (programmatic API / plugins) have
782 # no endpoint to classify; allow them — the audit hook still
783 # backstops stray sockets under PRIVATE_ONLY/STRICT.
784 if _is_user_registered_llm(provider):
785 return Decision(True, "user_registered_llm")
787 # Providers with a configurable URL: classify the URL.
788 url_key = f"llm.{provider}.url"
789 url_value = _get_setting_value(settings_snapshot, url_key, None)
790 if not url_value:
791 # Local-default providers (ollama, lmstudio, llamacpp) without
792 # an override fall back to their localhost defaults. Use the
793 # shared constant so this can't drift from the snapshot-less
794 # allow-list.
795 if provider in _LOCAL_DEFAULT_LLM_PROVIDERS:
796 return Decision(True, "provider_local_default")
797 return Decision(False, "provider_url_unset")
799 try:
800 parsed = urlsplit(url_value)
801 # Percent-decode the host before classifying — the HTTP client
802 # decodes it before connect, so "http://127%2e0%2e0%2e1:11434"
803 # must read as the local 127.0.0.1 (matches evaluate_url and
804 # _classify_engine_url). Without this a legitimate percent-encoded
805 # local endpoint is wrongly classified public and denied.
806 hostname = unquote(parsed.hostname) if parsed.hostname else None
807 except Exception:
808 return Decision(False, "url_malformed")
810 classification = _classify_host(hostname, ctx) if hostname else None
811 if classification is True:
812 return Decision(True, "provider_local")
813 return Decision(False, "provider_remote")
814 except Exception: # pragma: no cover - defensive
815 logger.bind(policy_audit=True).exception(
816 "evaluate_llm_endpoint internal error", provider=provider
817 )
818 return Decision(False, "internal_error")
821def evaluate_embeddings(
822 provider: str,
823 ctx: EgressContext,
824 *,
825 settings_snapshot: dict,
826) -> Decision:
827 """Decide whether an embeddings provider may be instantiated.
829 Only meaningful when ``ctx.require_local_embeddings`` is True.
830 OpenAI is treated as unambiguously cloud unless ``embeddings.openai.base_url``
831 is configured to a local host.
832 """
833 if settings_snapshot is None:
834 return Decision(False, "no_snapshot")
835 try:
836 if not ctx.require_local_embeddings:
837 return Decision(True, "no_local_requirement")
839 if provider == "sentence_transformers":
840 return Decision(True, "provider_local")
842 if provider == "ollama":
843 url = _get_setting_value(
844 settings_snapshot, "embeddings.ollama.url", None
845 ) or _get_setting_value(settings_snapshot, "llm.ollama.url", None)
846 if not url:
847 return Decision(True, "provider_local_default")
848 parsed = urlsplit(url)
849 host = unquote(parsed.hostname) if parsed.hostname else None
850 classification = _classify_host(host, ctx) if host else None
851 return (
852 Decision(True, "provider_local")
853 if classification is True
854 else Decision(False, "provider_remote")
855 )
857 if provider == "openai":
858 base_url = _get_setting_value(
859 settings_snapshot, "embeddings.openai.base_url", None
860 )
861 if base_url:
862 parsed = urlsplit(base_url)
863 host = unquote(parsed.hostname) if parsed.hostname else None
864 classification = _classify_host(host, ctx) if host else None
865 if classification is True:
866 return Decision(True, "provider_local_endpoint")
867 return Decision(False, "provider_cloud")
869 return Decision(False, "provider_unknown")
870 except Exception: # pragma: no cover - defensive
871 logger.bind(policy_audit=True).exception(
872 "evaluate_embeddings internal error", provider=provider
873 )
874 return Decision(False, "internal_error")
877_DANGEROUS_SCHEMES = frozenset(
878 {"javascript", "data", "file", "vbscript", "about"}
879)
881# Cloud-metadata endpoints reachable by HOSTNAME rather than literal IP. GCP's
882# IMDS answers on metadata.google.internal / metadata.goog (resolving to
883# 169.254.169.254); is_ip_blocked only catches the IP literal, so these names
884# need an explicit block to honor the "metadata is NEVER permitted, regardless
885# of scope" invariant in evaluate_url. AWS/Azure/Alibaba IMDS have no such
886# hostname (IP only) and are covered by the is_ip_blocked / alt-encoding path.
887_METADATA_HOSTNAMES = frozenset({"metadata.google.internal", "metadata.goog"})
890def _normalize_alt_ipv4(host: str) -> Optional[str]:
891 """Return the canonical dotted-quad for an IPv4 host written in an
892 alternate encoding the libc resolver accepts but ``ipaddress.ip_address``
893 rejects — octal (``0251.0376.0251.0376``), hex (``0xa9fea9fe``), integer
894 (``2852039166``) and short forms (``169.254.43518``). Returns ``None`` for
895 anything ``inet_aton`` won't parse (real hostnames, IPv6, junk).
897 ``socket.getaddrinfo`` resolves these numeric forms to the same address the
898 HTTP client will ``connect()`` to, so the metadata-IP block in
899 ``evaluate_url`` must classify them by their *real* target — otherwise
900 ``http://0251.0376.0251.0376/`` slips past ``is_ip_blocked`` (which only
901 parses canonical notation) and reads as an allowed public host.
902 """
903 try:
904 import socket
906 return socket.inet_ntoa(socket.inet_aton(host))
907 except (OSError, UnicodeError, TypeError):
908 return None
911def _quota_ctx(ctx: EgressContext) -> EgressContext:
912 """Return the EgressContext whose counter backs the per-RUN denied-fetch
913 quota.
915 Each call site (ContentFetcher, full_search, download_service, the audit
916 hook) builds its OWN EgressContext from the snapshot, so a per-context
917 counter would reset the budget every time a new engine/fetcher is built —
918 letting a malicious document evade ``MAX_DENIED_FETCHES_PER_RUN`` by
919 spreading denied fetches across contexts. The run's ARMED active context is
920 set once at run start and re-armed identically on pool workers, so it is
921 shared across the whole run; anchoring the counter there makes the quota
922 truly per-run. Falls back to the passed ctx when no run context is armed
923 (snapshot-less / programmatic callers, settings-page calls), keeping those
924 isolated calls self-contained.
925 """
926 try:
927 from .audit_hook import get_active_context
929 active = get_active_context()
930 except Exception: # pragma: no cover - defensive
931 active = None
932 return active if active is not None else ctx
935def evaluate_url(url: str, ctx: EgressContext) -> Decision:
936 """Decide whether an arbitrary URL may be fetched (e.g., by ``fetch_content``).
938 Enforces the run's denied-fetch quota to prevent exhaustion attacks
939 via malicious indexed documents that loop the agent through hundreds
940 of denied fetches.
941 """
942 try:
943 # Quota check: hard fail after MAX_DENIED_FETCHES_PER_RUN. Anchor the
944 # counter to the run's active context (``_quota_ctx``) so the budget is
945 # per-RUN, not per-EgressContext. Read under that ctx's lock so
946 # concurrent subagents see a consistent count.
947 qctx = _quota_ctx(ctx)
948 with qctx._lock:
949 if qctx._fetch_denial_count["count"] >= MAX_DENIED_FETCHES_PER_RUN:
950 return Decision(False, "denial_quota_exceeded")
952 if not isinstance(url, str) or not url:
953 return _record_denial(ctx, "url_malformed")
955 try:
956 parsed = urlsplit(url)
957 except Exception:
958 return _record_denial(ctx, "url_malformed")
960 if parsed.scheme.lower() in _DANGEROUS_SCHEMES:
961 return _record_denial(ctx, "dangerous_scheme")
962 if parsed.scheme.lower() not in ("http", "https"):
963 return _record_denial(ctx, "unsupported_scheme")
964 if not parsed.hostname:
965 return _record_denial(ctx, "no_hostname")
967 # HTTP client libraries (requests/urllib3) percent-DECODE the host
968 # in the netloc before the socket connect, but urlsplit().hostname
969 # preserves the encoding. Classifying the encoded form lets
970 # "http://192%2e168%2e1%2e1/" read as an unresolvable public host
971 # (DNS fails) while the client actually connects to the private
972 # 192.168.1.1 — a scope bypass under PUBLIC_ONLY. Classify the
973 # DECODED host so the policy sees the real connect target.
974 host = unquote(parsed.hostname)
975 # Trailing dots are insignificant to the resolver (getaddrinfo strips
976 # them), so "169.254.169.254." and "metadata.google.internal." must be
977 # classified identically to their bare forms — otherwise a trailing
978 # dot dodges the metadata checks below. Mirrors the SSRF validator's
979 # rstrip(".").
980 host = host.rstrip(".")
982 # Cloud-metadata endpoints (AWS/GCE/Azure IMDS at 169.254.169.254
983 # etc.) are NEVER permitted, regardless of scope. They classify as
984 # link-local/private, so STRICT and PRIVATE_ONLY would otherwise
985 # ALLOW them — a credential-theft path for prompt-injected fetches
986 # and, more importantly, for the audit-hook net which calls
987 # evaluate_url directly on raw socket.connect targets (bypassing the
988 # SSRF validator that the explicit fetch PEPs run first). Reuse
989 # is_ip_blocked(allow_private_ips=True), which returns True only for
990 # the always-blocked metadata set (+ NAT64 wraps) and also unwraps
991 # IPv4-mapped IPv6 forms.
992 try:
993 from ..ssrf_validator import is_ip_blocked
995 if is_ip_blocked(
996 host, allow_localhost=True, allow_private_ips=True
997 ):
998 return _record_denial(ctx, "blocked_metadata_ip")
999 # is_ip_blocked only parses canonical IP notation, so an
1000 # alternate-encoded metadata literal (octal/hex/integer) slips
1001 # through above. Normalize it to the dotted-quad the resolver
1002 # would connect to and re-check, so the "metadata IPs are NEVER
1003 # permitted, regardless of scope" invariant holds for those forms
1004 # too — and they get the explicit blocked_metadata_ip reason
1005 # rather than leaking into the scope-mismatch bucket.
1006 alt = _normalize_alt_ipv4(host)
1007 if (
1008 alt is not None
1009 and alt != host
1010 and is_ip_blocked(
1011 alt, allow_localhost=True, allow_private_ips=True
1012 )
1013 ):
1014 return _record_denial(ctx, "blocked_metadata_ip")
1015 # Metadata endpoints reachable by hostname (GCP) — is_ip_blocked
1016 # can't see these because they aren't IP literals. Block by name so
1017 # the invariant holds under PUBLIC_ONLY/BOTH too (the SSRF validator
1018 # backstops the actual fetch, but evaluate_url's own guarantee must
1019 # not depend on it).
1020 if host.lower() in _METADATA_HOSTNAMES:
1021 return _record_denial(ctx, "blocked_metadata_ip")
1022 except Exception: # noqa: silent-exception - defensive, see below
1023 # Non-IP host or helper error: fall through to normal
1024 # classification (DNS path has its own handling).
1025 pass
1027 # UNPROTECTED: egress protection disabled — any host is allowed. This
1028 # gate is deliberately AFTER the SSRF / cloud-metadata block above
1029 # (which always applies), so the escape hatch only lifts the
1030 # egress-scope host restriction, never the hard SSRF invariant.
1031 if ctx.scope == EgressScope.UNPROTECTED:
1032 return Decision(True, "egress_unprotected")
1034 classification = _classify_host(host, ctx)
1036 if ctx.scope == EgressScope.STRICT:
1037 # Under STRICT, URLs whose host is private are allowed; public
1038 # hosts (DOI redirects, citations) are not. This is a deliberate
1039 # trade-off: prompt-injection spoofing of provenance tags is a
1040 # bigger risk than false-positives on DOI redirects.
1041 if classification is True:
1042 return Decision(True, "allowed_private_host_under_strict")
1043 return _record_denial(ctx, "strict_public_host")
1045 if ctx.scope == EgressScope.PUBLIC_ONLY:
1046 if classification is False:
1047 return Decision(True, "allowed_public_host")
1048 return _record_denial(ctx, "scope_mismatch_public_only")
1050 if ctx.scope == EgressScope.PRIVATE_ONLY:
1051 if classification is True:
1052 return Decision(True, "allowed_private_host")
1053 return _record_denial(ctx, "scope_mismatch_private_only")
1055 # BOTH: any classified host is fine.
1056 if classification is None: 1056 ↛ 1057line 1056 didn't jump to line 1057 because the condition on line 1056 was never true
1057 return _record_denial(ctx, "host_unclassified")
1058 return Decision(True, "allowed_both_scope")
1059 except Exception: # pragma: no cover - defensive
1060 logger.bind(policy_audit=True).exception("evaluate_url internal error")
1061 return Decision(False, "internal_error")
1064def _record_denial(ctx: EgressContext, reason: str) -> Decision:
1065 """Increment the denial counter inside the frozen context's mutable dict
1066 and emit a redacted audit log line. Lock guards the read-modify-write
1067 against concurrent subagent threads.
1068 """
1069 counts_toward_quota = reason not in _NON_QUOTA_DENIAL_REASONS
1070 # Increment the per-RUN counter (the run's active context, shared across
1071 # every call-site context) rather than this one context's — see _quota_ctx.
1072 qctx = _quota_ctx(ctx)
1073 with qctx._lock:
1074 if counts_toward_quota:
1075 qctx._fetch_denial_count["count"] += 1
1076 count = qctx._fetch_denial_count["count"]
1077 logger.bind(policy_audit=True).warning(
1078 "policy denied URL fetch",
1079 reason=reason,
1080 scope=ctx.scope.value,
1081 denial_count=count,
1082 counted=counts_toward_quota,
1083 )
1084 return Decision(False, reason)
1087def evaluate_retriever(
1088 retriever_name: str,
1089 ctx: EgressContext,
1090 *,
1091 metadata: Optional[dict] = None,
1092) -> Decision:
1093 """Decide whether a registered retriever may be invoked.
1095 Reads classification (``{"is_local": bool}``) set at registration
1096 time. Unclassified retrievers fail closed. ``metadata`` may be
1097 supplied by the caller (e.g. the search-engine factory, which has
1098 already looked up the retriever in its own registry reference) to
1099 avoid a second registry lookup and to honor a test-patched registry;
1100 when ``None`` the global registry is consulted.
1101 """
1102 try:
1103 # UNPROTECTED: egress protection disabled — any retriever is permitted.
1104 if ctx.scope == EgressScope.UNPROTECTED:
1105 return Decision(True, "egress_unprotected")
1107 if metadata is None:
1108 from ...web_search_engines.retriever_registry import (
1109 retriever_registry,
1110 )
1112 try:
1113 metadata = retriever_registry.get_metadata(
1114 retriever_name, username=ctx.username
1115 )
1116 except AttributeError:
1117 # Older registry without metadata API; treat as unclassified.
1118 return Decision(False, "retriever_unclassified")
1120 if not metadata:
1121 return Decision(False, "retriever_unknown")
1123 is_local = metadata.get("is_local")
1124 if is_local is None: 1124 ↛ 1125line 1124 didn't jump to line 1125 because the condition on line 1124 was never true
1125 return Decision(False, "retriever_unclassified")
1127 if ctx.scope == EgressScope.STRICT:
1128 # STRICT permits only the user's primary engine. A retriever
1129 # IS allowed under STRICT when it is itself the primary (the
1130 # common "research only against my private KB" setup);
1131 # otherwise it's an expansion STRICT forbids.
1132 if retriever_name == ctx.primary_engine:
1133 return Decision(True, "allowed_primary_retriever")
1134 return Decision(False, "strict_not_primary")
1135 if ctx.scope == EgressScope.PUBLIC_ONLY and is_local:
1136 return Decision(False, "scope_mismatch_public_only")
1137 if ctx.scope == EgressScope.PRIVATE_ONLY and not is_local: 1137 ↛ 1138line 1137 didn't jump to line 1138 because the condition on line 1137 was never true
1138 return Decision(False, "scope_mismatch_private_only")
1139 return Decision(True, "allowed")
1140 except Exception: # pragma: no cover - defensive
1141 logger.bind(policy_audit=True).exception(
1142 "evaluate_retriever internal error", retriever=retriever_name
1143 )
1144 return Decision(False, "internal_error")
1147def _retriever_is_local(
1148 retriever_name: str, username: Optional[str]
1149) -> Optional[bool]:
1150 """Return the ``is_local`` classification (True/False) of a registered
1151 retriever, or ``None`` when it is unknown/unclassified.
1153 Mirrors the registry lookup in ``evaluate_retriever`` so adaptive-scope
1154 resolution and enforcement agree on a retriever primary's classification.
1155 """
1156 try:
1157 from ...web_search_engines.retriever_registry import (
1158 retriever_registry,
1159 )
1161 metadata = retriever_registry.get_metadata(
1162 retriever_name, username=username
1163 )
1164 except Exception: # noqa: silent-exception - unknown → caller falls back
1165 return None
1166 if not metadata:
1167 return None
1168 is_local = metadata.get("is_local")
1169 return is_local if isinstance(is_local, bool) else None
1172def _resolve_adaptive_scope(
1173 primary_engine: str,
1174 settings_snapshot: dict,
1175 *,
1176 username: Optional[str],
1177 local_hostnames: tuple,
1178 allow_dns: bool = True,
1179) -> EgressScope:
1180 """Map ADAPTIVE to a concrete scope by classifying the primary engine.
1182 Uses the engine's **static class flags** (plus ``_engine_bucket``'s
1183 fail-up URL override) so that e.g. a locally-hosted SearXNG is still
1184 treated as a public engine, and a remote-hosted Elasticsearch resolves
1185 public rather than pulling the run into PRIVATE_ONLY.
1187 private primary => PRIVATE_ONLY, public primary => PUBLIC_ONLY,
1188 unknown / classification-error => BOTH (adaptive's documented
1189 permissive fallback, equivalent to pre-policy behavior, so a
1190 classification hiccup never hard-fails the run).
1192 A registered LangChain retriever (private KB) is not in ENGINE_REGISTRY,
1193 so ``_engine_bucket`` returns ``(None, None)`` for it; we then consult the
1194 retriever registry so a local retriever primary resolves to PRIVATE_ONLY
1195 (forcing local inference) rather than leaking the corpus to cloud models.
1196 """
1197 if not primary_engine:
1198 return EgressScope.BOTH
1199 try:
1200 probe_ctx = EgressContext(
1201 scope=EgressScope.BOTH,
1202 primary_engine=primary_engine,
1203 require_local_llm=False,
1204 require_local_embeddings=False,
1205 local_hostnames=local_hostnames,
1206 username=username,
1207 )
1208 is_public, is_local = _engine_bucket(
1209 primary_engine, probe_ctx, settings_snapshot, allow_dns=allow_dns
1210 )
1211 except Exception: # noqa: silent-exception - adaptive falls back to BOTH
1212 logger.bind(policy_audit=True).debug(
1213 "adaptive scope classification failed; falling back to BOTH",
1214 primary=primary_engine,
1215 )
1216 return EgressScope.BOTH
1218 if is_local is True and is_public is not True:
1219 return EgressScope.PRIVATE_ONLY
1220 if is_public is True and is_local is not True:
1221 return EgressScope.PUBLIC_ONLY
1223 # Unknown to _engine_bucket — it may be a registered retriever (private
1224 # KB). Classify via the retriever registry before falling back to BOTH.
1225 if is_public is None and is_local is None:
1226 retriever_local = _retriever_is_local(primary_engine, username)
1227 if retriever_local is True:
1228 return EgressScope.PRIVATE_ONLY
1229 if retriever_local is False:
1230 return EgressScope.PUBLIC_ONLY
1232 return EgressScope.BOTH
1235def context_from_snapshot(
1236 settings_snapshot: dict,
1237 primary_engine: str,
1238 *,
1239 username: Optional[str] = None,
1240 allow_dns: bool = True,
1241) -> EgressContext:
1242 """Construct the frozen ``EgressContext`` for a research run.
1244 Reads policy settings out of the snapshot exactly once at run-start.
1245 A missing ``policy.egress_scope`` yields ``DEFAULT_EGRESS_SCOPE``
1246 (``adaptive`` — the protective default that follows the primary engine),
1247 NOT a permissive fallback. A residual ``both`` value is coerced to
1248 ``adaptive`` below.
1250 ``allow_dns=False`` makes ADAPTIVE resolution skip the synchronous
1251 getaddrinfo when classifying a URL-configurable primary engine — used by
1252 the advisory warning-banner render path so a settings-page load never
1253 blocks on DNS. Enforcement callers leave it True so classification is
1254 accurate. (Has no effect for non-ADAPTIVE scopes, which never resolve
1255 a URL.)
1256 """
1257 if settings_snapshot is None:
1258 raise ValueError("settings_snapshot is required")
1259 if not isinstance(settings_snapshot, dict):
1260 # Fail closed with the same ValueError contract callers already
1261 # handle (llm_config / research_service convert it to a hard policy
1262 # stop). A non-dict snapshot would otherwise crash deeper in
1263 # _get_setting_value with a bare AttributeError, which a broad
1264 # caller-side except could swallow into a permissive default.
1265 raise ValueError(
1266 f"settings_snapshot must be a dict, got "
1267 f"{type(settings_snapshot).__name__}"
1268 )
1270 scope_raw = _get_setting_value(
1271 settings_snapshot, "policy.egress_scope", DEFAULT_EGRESS_SCOPE
1272 )
1273 # `both` is retired (ADR-0007): the blanket-permissive middle scope is
1274 # coerced to the protective `adaptive` default so a residual value — an
1275 # un-migrated DB, a queued snapshot, or an env-var override that bypasses
1276 # settings validation — can never silently keep the old no-protection
1277 # behaviour. Migration 0019 rewrites stored values; this is the read-time
1278 # backstop. (EgressScope.BOTH still exists as the INTERNAL resolution result
1279 # for an unclassifiable ADAPTIVE primary — it is never a user-facing scope.)
1280 if str(scope_raw).strip().lower() == EgressScope.BOTH.value:
1281 scope_raw = EgressScope.ADAPTIVE.value
1282 try:
1283 scope = EgressScope(str(scope_raw).lower())
1284 except ValueError as exc:
1285 # An unrecognised scope means the saved setting was corrupted
1286 # or tampered with — silently falling back to BOTH (the most
1287 # permissive scope) would mask policy violations rather than
1288 # surface them. Refuse the run instead so the operator notices.
1289 logger.bind(policy_audit=True).warning(
1290 "refusing to construct EgressContext from unknown scope",
1291 value=scope_raw,
1292 )
1293 raise PolicyDeniedError(
1294 Decision(False, "unknown_egress_scope"),
1295 target=str(scope_raw),
1296 ) from exc
1298 require_local_llm = _coerce_bool(
1299 _get_setting_value(
1300 settings_snapshot, "llm.require_local_endpoint", False
1301 )
1302 )
1303 require_local_embeddings = _coerce_bool(
1304 _get_setting_value(settings_snapshot, "embeddings.require_local", False)
1305 )
1307 raw_hostnames = _get_setting_value(
1308 settings_snapshot, "llm.allowed_local_hostnames", ()
1309 )
1310 if isinstance(raw_hostnames, (list, tuple)): 1310 ↛ 1313line 1310 didn't jump to line 1313 because the condition on line 1310 was always true
1311 local_hostnames = tuple(h for h in raw_hostnames if isinstance(h, str))
1312 else:
1313 local_hostnames = ()
1315 # ADAPTIVE resolves to a concrete scope by classifying the primary
1316 # engine: a concrete private primary => PRIVATE_ONLY, a concrete public
1317 # primary => PUBLIC_ONLY, an unclassifiable primary => BOTH.
1318 # The resolved scope is what the EgressContext stores and what
1319 # every downstream PEP enforces. Classification reuses _engine_bucket
1320 # via a throwaway BOTH-scoped context (its DNS cache is discarded).
1321 if scope == EgressScope.ADAPTIVE:
1322 scope = _resolve_adaptive_scope(
1323 primary_engine,
1324 settings_snapshot,
1325 username=username,
1326 local_hostnames=local_hostnames,
1327 allow_dns=allow_dns,
1328 )
1330 # PRIVATE_ONLY means "my data stays on this box." That guarantee only
1331 # holds if BOTH inference paths are local — a cloud LLM receives the
1332 # query + retrieved local chunks, and a cloud embedder receives the
1333 # whole corpus at ingest. So under PRIVATE_ONLY the require_local_*
1334 # flags are IMPLIED: scope overrides them, so a user who left them at
1335 # their default (False) can't silently exfiltrate via inference. This
1336 # also fires when ADAPTIVE resolved to PRIVATE_ONLY above.
1337 #
1338 # STRICT is deliberately NOT coupled here: it restricts the search
1339 # engine set to the primary only and is orthogonal to where inference
1340 # runs (a user may legitimately want single-engine search + a cloud
1341 # LLM). Callers that gate on ctx.require_local_* therefore get
1342 # scope-correct behaviour without a separate flag read.
1343 if scope == EgressScope.PRIVATE_ONLY:
1344 require_local_llm = True
1345 require_local_embeddings = True
1346 elif scope == EgressScope.UNPROTECTED:
1347 # Escape hatch: egress protection is off, so the local-inference
1348 # requirements are lifted too (any provider is permitted).
1349 require_local_llm = False
1350 require_local_embeddings = False
1352 return EgressContext(
1353 scope=scope,
1354 primary_engine=primary_engine,
1355 require_local_llm=require_local_llm,
1356 require_local_embeddings=require_local_embeddings,
1357 local_hostnames=local_hostnames,
1358 username=username,
1359 )
1362def _get_setting_value(snapshot: dict, key: str, default):
1363 """Read a setting from the snapshot.
1365 Accepts both flat ({key: value}) and nested ({key: {"value": value}})
1366 schemas — both shapes appear in LDR's settings infrastructure.
1367 """
1368 raw = snapshot.get(key, default)
1369 return unwrap_setting(raw)
1372def coerce_str_list(raw) -> tuple[bool, list[str]]:
1373 """Decode a JSON-list-ish setting value into a list of strings.
1375 Returns ``(ok, values)``. Accepts a real list/tuple or a JSON-string list
1376 and filters to string entries. ``ok`` is False only when a non-empty JSON
1377 string failed to parse — callers that must *reject* malformed input branch
1378 on it; fail-safe callers ignore it and use the (empty) list. Single home for
1379 the decode shared by the trust resolver, the trust banner, and the
1380 settings-save validators.
1381 """
1382 if isinstance(raw, str):
1383 if not raw.strip(): 1383 ↛ 1384line 1383 didn't jump to line 1384 because the condition on line 1383 was never true
1384 return True, []
1385 try:
1386 raw = json.loads(raw)
1387 except Exception:
1388 return False, []
1389 if isinstance(raw, (list, tuple)): 1389 ↛ 1391line 1389 didn't jump to line 1391 because the condition on line 1389 was always true
1390 return True, [x for x in raw if isinstance(x, str)]
1391 return True, []
1394def resolve_run_primary_engine(
1395 settings_snapshot: Optional[dict], default: Optional[str] = None
1396) -> str:
1397 """Resolve a research run's primary search engine id from a snapshot.
1399 Single source of truth for "which engine drives this run's egress
1400 scope". Under the default ADAPTIVE scope the concrete scope is derived by
1401 classifying THIS primary, so run-scoped callers that build an
1402 ``EgressContext`` should derive the primary here. Otherwise two layers can
1403 resolve ADAPTIVE to different scopes and one silently under-enforces — the
1404 LangGraph tool-list filter once derived the primary from the engine CLASS
1405 name (a collection -> ``"libraryrag"`` -> unclassified -> BOTH) while the
1406 factory PEP derived it from ``search.tool`` (the real collection key ->
1407 PRIVATE_ONLY), so a public engine reached the agent and was then
1408 hard-denied mid-run (``scope_mismatch_private_only``).
1410 Reads ``search.tool`` (flat or ``{"value": ...}`` shaped). A blank-after-
1411 strip (``" "``) or non-string (``5``, list/dict) value is treated as
1412 MISSING — it must not slip through the truthiness check and classify to the
1413 permissive BOTH scope.
1415 A run with NO configured primary is a wiring error, not a normal state.
1416 Silently substituting a default would set the egress scope from an engine
1417 the user never chose — and since the system default is public (``searxng``),
1418 that fails OPEN. So a missing/empty ``search.tool`` raises ``ValueError``
1419 unless the caller passes an explicit ``default``:
1421 * run-level callers (research_service, the strategy tool-list filter,
1422 ``filter_candidates_by_egress``) pass none and fail closed — the worker
1423 refuses the run; advisory filters degrade to unfiltered with the factory
1424 PEP still enforcing.
1425 * the search-engine factory passes ``default=engine_name`` because it is
1426 evaluating that one specific engine, so deriving the scope from it is
1427 meaningful rather than arbitrary.
1429 Raises:
1430 ValueError: ``search.tool`` is missing/blank/non-string and no truthy
1431 ``default`` is given.
1432 """
1433 primary = (
1434 _get_setting_value(settings_snapshot, "search.tool", None)
1435 if settings_snapshot
1436 else None
1437 )
1438 # Blank-after-strip or non-string is not a usable engine id → missing.
1439 primary = primary.strip() if isinstance(primary, str) else None
1440 if primary:
1441 return primary
1442 if default:
1443 return default
1444 raise ValueError(
1445 "no primary search engine configured: settings 'search.tool' is "
1446 "missing, blank, or not a string, so the run's egress scope cannot "
1447 "be resolved"
1448 )
1451def _coerce_bool(value) -> bool:
1452 """Coerce a setting value to bool, defensively.
1454 Strings ``"true"`` / ``"True"`` are True; anything else string-shaped
1455 is False. Real booleans pass through. The strict coercion prevents
1456 type confusion (e.g., the string ``"false"`` being truthy).
1457 """
1458 if isinstance(value, bool):
1459 return value
1460 if isinstance(value, str): 1460 ↛ 1462line 1460 didn't jump to line 1462 because the condition on line 1460 was always true
1461 return value.strip().lower() in ("true", "1", "yes", "on")
1462 return bool(value)