Coverage for src/local_deep_research/security/egress/policy.py: 94%
663 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +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 hashlib
20import json
21import socket
22import threading
23from collections import OrderedDict
24from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout
25from dataclasses import dataclass, field
26from enum import Enum
27from typing import Literal, Optional
28from urllib.parse import unquote, urlsplit
30from loguru import logger
32from ..log_sanitizer import sanitize_for_log
33from ..network_utils import is_private_ip
34from ...utilities.type_utils import unwrap_setting
37# Bounded DNS lookup timeout; getaddrinfo has no native timeout kwarg, so
38# we run it inside a single-shot ThreadPoolExecutor and abandon the
39# Future after the timeout elapses. This avoids the previous
40# socket.setdefaulttimeout() approach, which mutated a process-global
41# setting and could corrupt unrelated network code under concurrency.
42_DNS_TIMEOUT_SEC = 2.0
45# Hard limit: after this many denied fetches in a single run, fail closed
46# even on otherwise-legal fetches. Prevents an exhaust attack via a malicious
47# indexed document looping the agent through hundreds of denied URLs.
48MAX_DENIED_FETCHES_PER_RUN = 50
50# Only SECURITY-relevant denials count toward the per-run quota. Benign parse
51# failures (a mailto:/ftp:/tel: link or a malformed href scraped from a page)
52# are common in legitimate documents and must NOT exhaust the budget — doing
53# so would make a long PUBLIC_ONLY run start refusing legitimate public URLs
54# mid-run. These reasons are still audit-logged; they just don't tick the quota.
55# ``dangerous_scheme`` (javascript:/data:/file:/… hrefs) belongs here too: these
56# are non-fetchable, non-network schemes scraped from ordinary HTML (onclick
57# handlers, inline data: URIs), so — like ``unsupported_scheme`` — they can't
58# cause egress and shouldn't let a doc full of data: URIs exhaust the budget.
59_NON_QUOTA_DENIAL_REASONS = frozenset(
60 {"url_malformed", "unsupported_scheme", "no_hostname", "dangerous_scheme"}
61)
64class EgressScope(str, Enum):
65 """User-declared egress boundary for a research run.
67 - STRICT: only the user's primary engine; no expansion at all.
68 - PUBLIC_ONLY: any public (external web/academic) engine.
69 - PRIVATE_ONLY: any private (local collection / library) engine.
70 - BOTH: INTERNAL ONLY — the resolved scope for an unclassifiable ADAPTIVE
71 primary (allow any classified engine, no local-inference coupling). It is
72 NO LONGER a user-selectable scope: `both` was the "muddy middle" (ADR-0007
73 Problem 3), retired in favour of `adaptive` + per-collection `is_public`.
74 A residual user `both` (stored / env / queued) is coerced to `adaptive`
75 by ``context_from_snapshot`` and migration 0019.
76 - ADAPTIVE: scope FOLLOWS the primary engine — a concrete private
77 primary behaves as PRIVATE_ONLY, a concrete public primary as
78 PUBLIC_ONLY, and an unclassifiable primary as BOTH.
79 The default: most users never touch scope and "it just matches my
80 main engine." Resolved to a concrete scope at context construction;
81 the stored EgressContext carries the RESOLVED scope, not ADAPTIVE.
82 - UNPROTECTED: escape hatch — egress protection is DISABLED for the run.
83 Any engine / URL / provider is permitted; the hard SSRF and
84 cloud-metadata blocks in ``evaluate_url`` still apply. Not recommended.
85 """
87 STRICT = "strict"
88 PUBLIC_ONLY = "public_only"
89 PRIVATE_ONLY = "private_only"
90 BOTH = "both"
91 ADAPTIVE = "adaptive"
92 UNPROTECTED = "unprotected"
95USER_SELECTABLE_PROTECTED_SCOPES = frozenset(
96 {
97 EgressScope.ADAPTIVE,
98 EgressScope.PUBLIC_ONLY,
99 EgressScope.PRIVATE_ONLY,
100 EgressScope.STRICT,
101 }
102)
105def unprotected_egress_allowed() -> bool:
106 """Return whether the operator enabled the unprotected escape hatch."""
107 from ...settings.env_registry import get_env_setting
109 return bool(get_env_setting("policy.allow_unprotected_egress", False))
112# Code-side single source of truth for the default egress scope, used by
113# every reader that needs a fallback for a MISSING policy.egress_scope key
114# (partial snapshots from the programmatic API, un-bootstrapped settings
115# DBs). Import THIS instead of hardcoding a string literal: scattered
116# literals are how the registry default ("adaptive") and the code
117# fallbacks ("both") drifted apart in the first place. Must match the
118# registered default in defaults/default_settings.json — pinned by
119# tests/security/test_egress_policy.py::
120# test_default_scope_constant_matches_registry.
121DEFAULT_EGRESS_SCOPE: str = EgressScope.ADAPTIVE.value
124@dataclass(frozen=True)
125class EgressContext:
126 """Frozen per-run policy snapshot.
128 Constructed once via ``context_from_snapshot()`` at run-start.
129 The dataclass is frozen, but mutable internals (``_dns_cache``,
130 ``_fetch_denial_count``) use ``field(init=False, default_factory=...)``
131 so they can accumulate state during a run. Counters live inside a
132 dict because direct ``int`` field reassignment fails on ``frozen=True``.
133 """
135 scope: EgressScope
136 primary_engine: str
137 require_local_llm: bool
138 require_local_embeddings: bool
139 local_hostnames: tuple[str, ...] = ()
140 username: Optional[str] = None
141 # Mutable internal state (kept off the public surface).
142 _dns_cache: dict = field(init=False, default_factory=dict, repr=False)
143 _fetch_denial_count: dict = field(
144 init=False, default_factory=lambda: {"count": 0}, repr=False
145 )
146 # Guards _dns_cache + _fetch_denial_count mutations. Scope is the
147 # cache writes only — DNS I/O must NOT happen while holding the lock,
148 # or concurrent subagent threads serialize on each other's lookups.
149 _lock: threading.RLock = field(
150 init=False, default_factory=threading.RLock, repr=False
151 )
154@dataclass(frozen=True)
155class Decision:
156 """Result of a PDP evaluation. ``reason`` is a short machine code
157 (e.g. ``"unclassified"``, ``"scope_mismatch"``) — never the user's
158 rejected query or URL content.
159 """
161 allowed: bool
162 reason: str
165@dataclass(frozen=True, slots=True)
166class EngineClassification:
167 """Public/local classification for one search engine."""
169 is_public: bool | None
170 is_local: bool | None
173class PolicyDeniedError(RuntimeError):
174 """Raised by PEPs when a policy decision is hard-stop denial.
176 Raising (rather than returning a graceful empty result) ensures
177 consistent denial latency, which mitigates the LangGraph timing-leak
178 pattern where an LLM could infer policy state from how fast a denied
179 tool call returns.
180 """
182 def __init__(self, decision: Decision, target: str = ""):
183 self.decision = decision
184 self.target = target
185 super().__init__(f"policy_denied: {decision.reason}")
188def parse_user_egress_scope(
189 raw: object,
190 *,
191 disabled_unprotected: Literal["reject", "adaptive"] = "reject",
192) -> EgressScope:
193 """Parse a user-facing scope and enforce the operator escape-hatch gate.
195 BOTH is deliberately absent from the protected allowlist: it remains an
196 internal adaptive-resolution result, not a value users may select. Runtime
197 readers can request the protective adaptive fallback for stale queued
198 unprotected values; request and settings writers use the default hard
199 rejection.
200 """
201 if disabled_unprotected not in {"reject", "adaptive"}: 201 ↛ 202line 201 didn't jump to line 202 because the condition on line 201 was never true
202 raise ValueError("disabled_unprotected must be 'reject' or 'adaptive'")
203 if not isinstance(raw, str):
204 raise PolicyDeniedError(
205 Decision(False, "unknown_egress_scope"), target=str(raw)
206 )
207 normalized = raw.strip().lower()
208 try:
209 scope = EgressScope(normalized)
210 except ValueError as exc:
211 raise PolicyDeniedError(
212 Decision(False, "unknown_egress_scope"), target=str(raw)
213 ) from exc
214 if scope in USER_SELECTABLE_PROTECTED_SCOPES:
215 return scope
216 if scope == EgressScope.UNPROTECTED:
217 if unprotected_egress_allowed():
218 return scope
219 if disabled_unprotected == "adaptive":
220 return EgressScope.ADAPTIVE
221 raise PolicyDeniedError(
222 Decision(False, "unprotected_egress_disabled"), target=str(raw)
223 )
224 raise PolicyDeniedError(
225 Decision(False, "unknown_egress_scope"), target=str(raw)
226 )
229def effective_scope_for_display(raw: object) -> str:
230 """Return a canonical, presentation-safe user scope value.
232 Invalid values, retired both, and operator-disabled unprotected selections
233 render as the protective default. This helper is presentation only;
234 enforcement remains in context_from_snapshot and its PEPs.
235 """
236 try:
237 return parse_user_egress_scope(
238 raw, disabled_unprotected="adaptive"
239 ).value
240 except PolicyDeniedError:
241 return DEFAULT_EGRESS_SCOPE
244def _is_nat64_wrapped_metadata(hostname: str) -> bool:
245 """True iff ``hostname`` parses as an IPv6 NAT64 address wrapping a
246 cloud-metadata IPv4 (AWS / Azure / GCE / etc).
248 The wrapping passes ``is_private_ip`` (because IPv6 link-local
249 ranges match) but the embedded IPv4 actually reaches the metadata
250 endpoint. We classify these as PUBLIC so STRICT and PRIVATE_ONLY
251 refuse to fetch them.
252 """
253 try:
254 import ipaddress
256 from ..ssrf_validator import is_nat64_wrapped_metadata_ip
258 candidate = hostname
259 if candidate.startswith("[") and candidate.endswith("]"):
260 candidate = candidate[1:-1]
261 ip = ipaddress.ip_address(candidate)
262 return is_nat64_wrapped_metadata_ip(ip)
263 except (ValueError, TypeError):
264 return False
265 except Exception: # pragma: no cover - defensive
266 return False
269def _resolve_with_timeout(hostname: str) -> Optional[list]:
270 """Resolve ``hostname`` via getaddrinfo with a bounded timeout.
272 Returns the addrinfo list on success or None on timeout / lookup
273 failure. getaddrinfo has no native timeout, so we drive it from a
274 single-shot worker thread and abandon the Future on timeout — far
275 safer than socket.setdefaulttimeout(), which mutates a
276 process-global setting and races with unrelated network code.
277 """
279 def _do_lookup():
280 return socket.getaddrinfo(hostname, None, type=socket.SOCK_STREAM)
282 # NB: do NOT use ``with ThreadPoolExecutor(...)`` here. The context
283 # manager's __exit__ calls shutdown(wait=True), which blocks until the
284 # worker thread finishes — so a hung getaddrinfo would defeat the whole
285 # point of the timeout (the call would return only after the OS DNS
286 # timeout, not after _DNS_TIMEOUT_SEC). Instead we abandon the worker on
287 # timeout via shutdown(wait=False): the caller returns promptly and the
288 # orphaned thread dies on its own when getaddrinfo eventually returns.
289 executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="ldr-dns")
290 future = executor.submit(_do_lookup)
291 try:
292 return future.result(timeout=_DNS_TIMEOUT_SEC)
293 except (FutureTimeout, socket.gaierror, socket.timeout, OSError):
294 return None
295 except Exception: # pragma: no cover - defensive
296 return None
297 finally:
298 # wait=False so a timed-out lookup does not re-block here; a
299 # completed lookup leaves an idle thread that shuts down at once.
300 executor.shutdown(wait=False)
303_UNCACHED = object()
306def _cache_classification(
307 ctx: EgressContext, hostname: str, value: bool
308) -> bool:
309 """First-writer-wins cache set; returns the value now in the cache.
311 Concurrent subagent threads may classify the same hostname outside the
312 lock (the DNS lookup is intentionally unsynchronized). If a round-robin
313 DNS name resolves to a private IP for one thread and a public IP for
314 another, last-writer-wins would let a later call flip an earlier call's
315 result — cache *incoherence* that could relax PRIVATE_ONLY/STRICT on a
316 subsequent fetch. Pinning the first writer's value makes the per-run
317 classification stable and deterministic regardless of completion order.
318 """
319 with ctx._lock:
320 existing = ctx._dns_cache.get(hostname, _UNCACHED)
321 if existing is not _UNCACHED:
322 return existing
323 ctx._dns_cache[hostname] = value
324 return value
327def _classify_host(
328 hostname: str, ctx: EgressContext, allow_dns: bool = True
329) -> Optional[bool]:
330 """Classify a hostname as local (True) / public (False) / unknown (None).
332 Uses the run-scoped DNS cache to avoid repeated ``getaddrinfo`` lookups.
333 Falls back to public on DNS timeout (fail-safe).
335 Lock discipline: cache reads and writes are guarded by ``ctx._lock``;
336 the DNS lookup itself runs OUTSIDE the lock so concurrent subagent
337 threads don't serialize on each other's DNS calls. Cache writes are
338 first-writer-wins (``_cache_classification``) so a hostname's
339 classification is stable for the run even under concurrent disagreeing
340 lookups.
341 """
342 if not hostname: 342 ↛ 343line 342 didn't jump to line 343 because the condition on line 342 was never true
343 return None
345 # Cache hit — read under the lock so concurrent writers don't tear
346 # the dict.
347 with ctx._lock:
348 if hostname in ctx._dns_cache:
349 return ctx._dns_cache[hostname]
351 # User-declared local hostnames override DNS classification.
352 if hostname in ctx.local_hostnames:
353 return _cache_classification(ctx, hostname, True)
355 # NAT64-wrapped metadata IPs (e.g. 64:ff9b::169.254.169.254) classify
356 # as link-local by is_private_ip but actually wrap AWS/GCE instance
357 # metadata. Force them to PUBLIC so STRICT/PRIVATE_ONLY don't allow
358 # them. Check happens BEFORE is_private_ip so the wrapping wins.
359 if _is_nat64_wrapped_metadata(hostname):
360 return _cache_classification(ctx, hostname, False)
362 # String-literal check first (no DNS needed for IPs and known literals).
363 # If the helper itself raises (malformed hostname, broken stdlib edge case),
364 # fall through to DNS resolution rather than fail the whole evaluation —
365 # the DNS path has its own bounded timeout and error handling.
366 from ..ssrf_validator import is_ip_blocked
368 try:
369 # A literal cloud-metadata IP (169.254.169.254, fd00:ec2::254, …) passes
370 # is_private_ip (link-local) but must NOT be treated as local — otherwise
371 # STRICT/PRIVATE_ONLY would accept an IMDS SSRF target as a "local"
372 # inference/search host. Mirror the DNS-branch metadata block below.
373 # is_ip_blocked returns False for non-IP hostnames (ValueError → False),
374 # so legitimate hostnames still fall through to is_private_ip / DNS.
375 # block_link_local extends the same classification from the literal
376 # metadata-address list to the whole IPv4 and IPv6 link-local ranges,
377 # which must not be admitted under STRICT/PRIVATE_ONLY.
378 #
379 # Regression evidence and local-backend controls:
380 # tests/security/test_llm_endpoint_link_local_hardening.py
381 # - test_link_local_does_not_classify_as_local_in_egress_policy
382 # - test_self_hosted_still_classifies_as_local
383 #
384 # Localhost and RFC1918 are untouched -- self-hosted backends live
385 # there and must keep working.
386 if is_ip_blocked(
387 hostname,
388 allow_localhost=True,
389 allow_private_ips=True,
390 block_link_local=True,
391 ):
392 return _cache_classification(ctx, hostname, False)
393 if is_private_ip(hostname):
394 return _cache_classification(ctx, hostname, True)
395 except Exception as exc:
396 logger.debug(
397 "is_private_ip raised on hostname, falling back to DNS",
398 error=str(exc),
399 )
401 # DNS-resolved classification — runs OUTSIDE the lock.
402 if not allow_dns:
403 # The caller (the advisory warning-banner render path) opted OUT of
404 # the synchronous getaddrinfo so a settings-page render never blocks
405 # up to _DNS_TIMEOUT_SEC on a lookup. Return "unknown" so adaptive
406 # scope resolution falls back to the engine's static classification.
407 # Deliberately NOT cached — a later enforcement-path call (allow_dns
408 # default True) must still resolve this host for real.
409 return None
410 addr_info = _resolve_with_timeout(hostname)
411 if addr_info is None:
412 # Fail-safe: unresolvable / timeout → treat as public.
413 return _cache_classification(ctx, hostname, False)
415 # A hostname is local only when every resolved address is local. A single
416 # public or metadata answer can be selected by the HTTP client, so it must
417 # prevent a local classification.
418 if not addr_info: 418 ↛ 419line 418 didn't jump to line 419 because the condition on line 418 was never true
419 return _cache_classification(ctx, hostname, False)
420 for entry in addr_info:
421 try:
422 ip_str = entry[4][0]
423 # A hostname that resolves to a cloud-metadata IP must NOT be
424 # treated as local: link-local metadata IPs (169.254.169.254, …)
425 # pass is_private_ip, which would classify the host local and let
426 # STRICT/PRIVATE_ONLY fetch it. Classify as public so those scopes
427 # refuse it — mirrors the literal-IP metadata block in evaluate_url
428 # and the NAT64 handling just below. (Under PUBLIC_ONLY/BOTH the
429 # SSRF validator at the actual fetch is the metadata backstop.)
430 if is_ip_blocked(
431 ip_str,
432 allow_localhost=True,
433 allow_private_ips=True,
434 block_link_local=True,
435 ):
436 return _cache_classification(ctx, hostname, False)
437 if _is_nat64_wrapped_metadata(ip_str): 437 ↛ 438line 437 didn't jump to line 438 because the condition on line 437 was never true
438 return _cache_classification(ctx, hostname, False)
439 if not is_private_ip(ip_str):
440 return _cache_classification(ctx, hostname, False)
441 except Exception: # pragma: no cover - defensive
442 return _cache_classification(ctx, hostname, False)
443 return _cache_classification(ctx, hostname, True)
446def _get_engine_class(engine_name: str):
447 """Lazy load the engine class from the registry to avoid circular imports."""
448 # Import is lazy to break the dependency cycle:
449 # security/* → engines/* → security/*
450 from ...web_search_engines.engine_registry import ENGINE_REGISTRY
451 from ..module_whitelist import get_safe_module_class
453 entry = ENGINE_REGISTRY.get(engine_name)
454 if entry is None:
455 return None
456 try:
457 return get_safe_module_class(entry.module_path, entry.class_name)
458 except Exception:
459 return None
462def _engine_flags(engine_cls):
463 """Read the (is_public, is_local, url_setting) triple from an engine class.
465 Uses ``is True`` / ``is False`` semantics: a missing attribute is treated
466 as "not declared", which is distinct from an attribute explicitly set
467 to ``False``.
468 """
469 is_public = getattr(engine_cls, "is_public", None)
470 is_local = getattr(engine_cls, "is_local", None)
471 url_setting = getattr(engine_cls, "url_setting", None)
472 return is_public, is_local, url_setting
475def _resolve_collection_is_public(
476 engine_name: str, username: Optional[str]
477) -> bool:
478 """Look up a collection engine's per-collection ``is_public`` flag.
480 ``engine_name`` is ``"collection_<uuid>"`` or ``"library"``. The
481 aggregate ``library`` is always private (it spans all local docs).
482 Fails closed to private (False) on any lookup error or missing row —
483 a collection we can't classify must NOT be treated as public.
484 """
485 if engine_name == "library" or not engine_name.startswith("collection_"):
486 return False
487 collection_id = engine_name[len("collection_") :]
488 try:
489 from ...database.models.library import Collection
490 from ...database.session_context import get_user_db_session
492 with get_user_db_session(username) as session:
493 row = (
494 session.query(Collection)
495 .filter(Collection.id == collection_id)
496 .first()
497 )
498 return bool(getattr(row, "is_public", False)) if row else False
499 except Exception: # noqa: silent-exception - fail closed to private
500 logger.debug(
501 "could not resolve collection is_public; treating as private",
502 engine=engine_name,
503 )
504 return False
507# ---------------------------------------------------------------------------
508# Vector-store (RAG sink) classification
509# ---------------------------------------------------------------------------
511# Provider keys known to keep every byte on this machine. This is a FALLBACK
512# ONLY — the authoritative signal is the provider class's ``is_local_file``
513# flag (``vector_stores/base.py``); this set is consulted just when the class
514# cannot be introspected (e.g. a partial install of an optional backend).
515# Like the snapshot-less LLM allow-list below, keeping it tight is the point:
516# a provider key that is not enumerated here and whose class cannot be loaded
517# is treated as REMOTE, not local.
518_LOCAL_FILE_VECTOR_STORE_PROVIDERS = frozenset({"faiss"})
520# Prefixes a *network host* can never start with, so a vector-store URI that
521# begins with one is unambiguously a local filesystem path (e.g. a Milvus-Lite
522# style "./store.db"). Deliberately conservative: anything not matched here — a
523# bare "store.db", a "host:port" — is NOT assumed to be a path and goes through
524# the DNS classifier, which fails up to public when it cannot resolve.
525#
526# NOTE: a bare "/" is intentionally NOT in this tuple. A URI authority (the
527# "//host[:port]" form used by e.g. a Milvus endpoint written without a
528# scheme) also starts with "/", so treating a single leading "/" as
529# sufficient would misclassify a network host as a local path. A genuine
530# absolute path is handled separately by ``_is_local_filesystem_uri`` — a
531# single leading "/" whose second character is NOT another "/".
532_LOCAL_PATH_URI_PREFIXES = ("./", "../", "~/", "~\\", ".\\", "..\\")
535def _vector_store_class(provider: str):
536 """Load a vector-store provider's class, or ``None`` if unavailable.
538 Import is lazy and failure-tolerant: an optional backend whose driver is
539 not installed (or an unknown provider key) must not break classification —
540 the caller treats an un-introspectable non-default provider as remote.
541 """
542 if not provider: 542 ↛ 543line 542 didn't jump to line 543 because the condition on line 542 was never true
543 return None
544 try:
545 from ...vector_stores.config import get_vector_store_class
547 return get_vector_store_class(provider)
548 except Exception:
549 logger.debug(
550 "vector-store provider class unavailable; "
551 "classifying from the provider key alone",
552 provider=sanitize_for_log(provider),
553 )
554 return None
557def _normalize_provider_key(raw) -> str:
558 """Normalize a provider key the way the vector-store factory does."""
559 if not isinstance(raw, str): 559 ↛ 560line 559 didn't jump to line 560 because the condition on line 559 was never true
560 return ""
561 return raw.strip().strip("\"'").strip().lower()
564def _resolve_index_vector_store_providers(
565 engine_name: str, username: Optional[str]
566) -> set:
567 """Providers recorded on the CURRENT ``RAGIndex`` rows for this engine.
569 A collection is queried through the store its index was built in, which
570 can differ from the currently configured default, so the recorded provider
571 is an additional (tightening) signal on top of the settings-resolved one.
572 ``library`` spans every collection, so every current row counts.
574 Fails SOFT (empty set) when the column does not exist (older schema), the
575 DB is unavailable, or anything else goes wrong: the settings-resolved
576 provider remains the primary signal, and the endpoint URI that decides
577 exposure comes from settings regardless.
578 """
579 try:
580 from ...database.models.library import RAGIndex
581 from ...database.session_context import get_user_db_session
583 column = getattr(RAGIndex, "vector_store_provider", None)
584 if column is None:
585 return set() # schema without a provider column: FAISS-only
586 with get_user_db_session(username) as session:
587 query = session.query(column).filter(RAGIndex.is_current.is_(True))
588 if engine_name != "library":
589 query = query.filter(RAGIndex.collection_name == engine_name)
590 return {
591 key
592 for key in (_normalize_provider_key(row[0]) for row in query)
593 if key
594 }
595 except Exception: # noqa: silent-exception - additive signal only
596 logger.debug(
597 "could not resolve RAG index vector-store providers",
598 engine=sanitize_for_log(engine_name),
599 )
600 return set()
603def _resolve_vector_store_providers(
604 engine_name: str, ctx: EgressContext, settings_snapshot: Optional[dict]
605) -> set:
606 """Every vector-store provider a library/collection run could query.
608 The union of the settings-resolved default and the providers recorded on
609 the engine's current index rows. An EMPTY set means the vector-store layer
610 itself could not be reached — in which case the run cannot construct a
611 store either, so nothing can leave the box and the caller keeps the
612 (contained) status quo.
613 """
614 providers = set()
615 try:
616 from ...vector_stores.config import resolve_provider
618 key = _normalize_provider_key(resolve_provider(settings_snapshot))
619 if key: 619 ↛ 623line 619 didn't jump to line 623 because the condition on line 619 was always true
620 providers.add(key)
621 except Exception: # noqa: silent-exception - see docstring
622 logger.debug("vector-store provider resolution unavailable")
623 providers |= _resolve_index_vector_store_providers(
624 engine_name, getattr(ctx, "username", None)
625 )
626 return providers
629def _starts_with_double_separator(s: str) -> bool:
630 """True when ``s`` opens with two path-separator characters.
632 Covers all four combinations of "/" and "\\" — "//", "/\\", "\\/", and
633 "\\\\" — since ``ntpath.normpath`` treats any of them as the start of a
634 UNC network-share prefix on Windows (e.g. ``ntpath.normpath("/\\host/x")
635 == "\\\\host\\x"``). A forward-slash-only check lets the mixed-slash
636 forms smuggle a remote share past the "unambiguously local" classifier.
637 """
638 return len(s) >= 2 and s[0] in "/\\" and s[1] in "/\\"
641# ``urllib.parse.urlsplit`` does NOT parse the string it is handed verbatim.
642# Following the WHATWG URL parser it first lstrips every C0 control character
643# and space, then DELETES every tab, CR and LF from ANYWHERE in the URL
644# (CPython's ``_WHATWG_C0_CONTROL_OR_SPACE`` and
645# ``_UNSAFE_URL_BYTES_TO_REMOVE``). ``str.strip()`` does neither: it leaves
646# "\x00" and the other non-whitespace C0 controls in place, and never touches
647# a tab in the MIDDLE of a string.
648_URL_PARSER_LEADING_STRIP = "".join(chr(code) for code in range(0x20)) + " "
649_URL_PARSER_REMOVED_CHARS = ("\t", "\r", "\n")
652def _normalize_uri_like_url_parser(value: str) -> str:
653 """``value`` reduced to the string ``urlsplit`` would actually parse.
655 Exists for the RAW string tests in ``_is_local_filesystem_uri``
656 (``startswith``, ``_starts_with_double_separator``, a drive-letter index),
657 whose verdict is consumed side by side with ``_classify_engine_url``,
658 which reads the host out of ``urlsplit``. Judging the raw string opens a
659 parser differential in which one value is simultaneously "a local
660 filesystem path" to ``str.startswith`` and "an authority" to every real
661 URL parser:
663 * ``"x:/\\t/attacker.example.com:19530"`` looks drive-absolute to a
664 literal ``"//"`` test (``uri[2:]`` is ``"/\\t/..."``), while
665 ``urlsplit`` deletes the tab and reports netloc
666 ``attacker.example.com:19530``.
667 * ``"/\\t/attacker.example.com/share/db"`` looks like an ordinary
668 single-slash absolute path to ``_starts_with_double_separator``, while
669 ``urlsplit`` deletes the tab and reports the same authority.
670 * A leading ``"\\x00"`` (or any other non-whitespace C0 control) survives
671 ``str.strip()`` and defeats every raw prefix test, while ``urlsplit``
672 lstrips it away.
674 Normalising once, here, is what keeps those predicates and ``urlsplit``
675 from drifting apart again.
677 A raw-string test is not the only caller that needs this. The final
678 ``strip()`` is BIDIRECTIONAL, and its LEADING half removes 19
679 Unicode-whitespace codepoints that ``urlsplit``'s WHATWG lstrip
680 (``\\x00``-``\\x20`` only) leaves in place: U+0085, U+00A0, U+1680,
681 U+2000-U+200A, U+2028, U+2029, U+202F, U+205F and U+3000. A leading one
682 of those makes ``urlsplit`` report an EMPTY scheme, so a function that
683 asks the parser for the scheme does NOT already get this preprocessing
684 for free. The call is therefore load-bearing in
685 ``_file_uri_names_a_remote_share``: without it,
686 "\\xa0file://localhost//attacker.example.com/share/db" has scheme "" and
687 is not a ``file://`` URI to that function, the network-share rejection
688 never fires, and the entry falls through to hostname classification —
689 where a sibling entry such as "127.0.0.1" answers LOCAL and the whole
690 setting classifies contained. Never drop this call on the ground that
691 ``urlsplit`` repeats it; for the leading-whitespace half it does not.
693 Only the TRAILING half of that ``strip()`` is a pure caller-compatibility
694 trim: ``urlsplit`` keeps trailing whitespace where this does not, but
695 trailing whitespace can never turn a local path into an authority, so the
696 extra trim cannot make a remote endpoint look local.
697 """
698 normalized = value.lstrip(_URL_PARSER_LEADING_STRIP)
699 for char in _URL_PARSER_REMOVED_CHARS:
700 normalized = normalized.replace(char, "")
701 return normalized.strip()
704def _is_local_filesystem_uri(value) -> bool:
705 """True when a vector-store URI is unambiguously a local file path."""
706 if not isinstance(value, str):
707 return False
708 # Judge the string ``urlsplit`` would parse, never the raw one — see
709 # ``_normalize_uri_like_url_parser``. Every prefix, separator and
710 # drive-letter test below is a RAW string test, so without this the
711 # "local drive path" / "single-slash absolute path" verdicts diverge from
712 # the host ``_classify_engine_url`` goes on to classify. This call is what
713 # stops "x:/\t/attacker.example.com:19530" and
714 # "/\t/attacker.example.com/share/db" short-circuiting LOCAL.
715 uri = _normalize_uri_like_url_parser(value)
716 if not uri:
717 return False
718 if uri.lower().startswith("file://"):
719 # RFC 8089 local forms: "file:///path" (empty authority),
720 # "file://localhost/path", or the Windows drive-letter form some
721 # resolvers write as "file://C:/data/store.db" — urlsplit reads the
722 # drive letter + colon as the netloc there (there is no real
723 # authority, just a colon that looks like one). Any OTHER authority
724 # is a remote-share reference some resolvers treat as a host (e.g.
725 # "file://attacker.example.com/share/store.db") and must NOT be
726 # classified as on-box — fall through to the remote/DNS path.
727 #
728 # ``urlsplit`` is NOT exception-safe: a malformed authority (e.g. an
729 # unbalanced IPv6-literal bracket, "file://[attacker.example.com/x"
730 # or "file://[::1/x") raises ``ValueError: Invalid IPv6 URL`` instead
731 # of returning a parse. Unlike its sibling in ``_classify_engine_url``
732 # (which already wraps this identical call), this call was bare, so
733 # the exception used to propagate all the way up through
734 # ``_vector_store_endpoint_is_local`` -> ``vector_store_is_contained``
735 # -> ``classify_engine`` into ``_resolve_adaptive_scope``'s catch-all,
736 # which resolves the PERMISSIVE ``EgressScope.BOTH`` — exactly the
737 # cloud-capable scope this module's fail-up rules exist to avoid.
738 # Fail toward not-local instead, matching every other unknown here.
739 try:
740 parsed = urlsplit(uri)
741 except Exception:
742 return False
743 netloc = parsed.netloc
744 # Windows drive letters are ASCII A-Z only. ``str.isalpha()`` is
745 # Unicode-wide (Cyrillic "с", fullwidth "Ｃ", accented "é" all
746 # match), so a Unicode lookalike authority (e.g.
747 # "file://с:/share/db") must NOT pass as an on-box drive path —
748 # the same hazard class as the single-letter-scheme lookalikes at
749 # the bottom of this function.
750 is_drive_letter_netloc = (
751 len(netloc) == 2
752 and netloc[0].isascii()
753 and netloc[0].isalpha()
754 and netloc[1] == ":"
755 )
756 if not (
757 netloc == ""
758 or netloc.lower() == "localhost"
759 or is_drive_letter_netloc
760 ):
761 return False
762 # A 4-plus-slash form ("file:////attacker.example.com/share/db")
763 # parses with an EMPTY netloc too — urlsplit only ever gives the
764 # first two slashes after the scheme authority meaning, so a third
765 # and fourth slash just start the PATH — but that path then begins
766 # with "//host/share", a UNC-style remote-share reference smuggled
767 # past the empty-netloc check above. Reject that form: a local file
768 # path never legitimately starts with two slashes. A mixed-slash
769 # variant ("file:///\\host/share/db" -> path "/\\host/share/db")
770 # smuggles the same UNC authority past a forward-slash-only check:
771 # ``ntpath.normpath`` treats ANY pair of leading separators — "//",
772 # "/\\", "\\/", or "\\\\" — as a UNC prefix on Windows, so reject all
773 # four combinations, not just the literal "//" one. The path may be
774 # percent-encoded (RFC 8089), so DECODE it before the separator
775 # check: "file:///%2fattacker.example.com/share/db" and the "%5c"
776 # twin read as single-separator paths while encoded, but unquote
777 # to exactly the "//host" / "/\\host" UNC prefixes the check exists
778 # to catch (``ntpath.normpath`` resolves both to a remote share).
779 # Decoding cannot smuggle an authority in: netloc was already
780 # validated above, and plain local paths ("/var/lib/x",
781 # "/C:/data") keep their single leading separator through unquote.
782 #
783 # DEFENCE IN DEPTH: this decode-and-check is a deliberate duplicate
784 # of ``_file_uri_names_a_remote_share``, which
785 # ``_vector_store_endpoint_is_local`` — the ONLY production caller of
786 # this function — already ran one frame earlier over every entry of
787 # the setting. At every production entry point the earlier check has
788 # therefore already returned "not contained", so reverting the
789 # ``unquote`` here changes no end-to-end outcome; it stays because
790 # this function is a general-purpose predicate that a future caller
791 # may reach directly, and it is pinned by its own unit test
792 # (``test_file_uri_percent_encoded_unc_in_path_is_not_a_local_file``).
793 return not _starts_with_double_separator(unquote(parsed.path))
794 if uri.startswith(_LOCAL_PATH_URI_PREFIXES):
795 return True
796 # A single leading "/" is a local absolute path ("/var/lib/x.db") ONLY
797 # when it is not the start of a "//host[:port]" URI authority (e.g. a
798 # Milvus endpoint written as "//attacker.example.com:19530"), including
799 # the mixed-slash smuggling form "/\\host/share" that ``ntpath.normpath``
800 # also resolves to a UNC network share on Windows. Reject any such
801 # authority-ish form here so it falls through to the DNS classifier
802 # instead of being silently treated as on-box.
803 if uri.startswith("/") and not _starts_with_double_separator(uri):
804 return True
805 # Windows drive-absolute path ("C:\\data\\store.db"). What follows the
806 # drive letter must not be a literal "//" — "<letter>://" is a
807 # "scheme://host" URL authority marker, not a path, so e.g.
808 # "x://attacker.example.com:19530" must NOT be misread as a local drive
809 # path and skip the DNS classifier. Note ``str.isalpha()`` is
810 # Unicode-wide (Cyrillic/fullwidth letters match too, e.g. "с://host" or
811 # "Ｃ://host"), so the drive letter must be ASCII here for the same
812 # reason the ``file://`` netloc test above requires it: "с:/x", "Ｃ:/x"
813 # and "é:/x" are not Windows drive-absolute paths, and short-circuiting
814 # them to on-box would skip the DNS classifier on a value no resolver
815 # reads as a local drive. The literal-"//" authority check below is a
816 # separate, independent guard covering the "<letter>://host" form.
817 #
818 # ONLY the literal "//" form is rejected, not every doubled separator:
819 # a drive letter always wins over a UNC prefix on Windows
820 # (``ntpath.normpath("C:\\\\data") == "C:\\data"``, never a network
821 # share), and — ON THE NORMALISED STRING this function tests —
822 # ``urlsplit`` gives an authority only to a literal "//", so
823 # "C:\\\\data\\store.db", "C:/\\data" and "C:\\/data" cannot name a
824 # remote host under either reading and are genuine local paths. That
825 # equivalence holds ONLY because ``_normalize_uri_like_url_parser`` has
826 # already deleted the tab/CR/LF and leading C0 controls that ``urlsplit``
827 # itself deletes: on the RAW string "x:/\t/attacker.example.com:19530"
828 # this literal-"//" test answers "no authority, local drive path" while
829 # ``urlsplit`` answers "netloc=attacker.example.com:19530". Never move
830 # this test back onto an un-normalised string. The
831 # forward-slash form "C://data" is irreducibly ambiguous — it parses as
832 # scheme "c" with hostname "data" — so it stays rejected and falls
833 # through to the DNS classifier; see changelog.d/5763.security.md.
834 return (
835 len(uri) > 2
836 and uri[0].isascii()
837 and uri[0].isalpha()
838 and uri[1] == ":"
839 and uri[2] in "\\/"
840 and not uri[2:].startswith("//")
841 )
844def _file_uri_names_a_remote_share(value) -> bool:
845 """True when ``value`` is a ``file://`` URI that does not stay on the box.
847 The DECODED path of a ``file://`` URI can name a UNC network share
848 (``//host/share``, or any of the mixed "/"/"\\" separator pairs
849 ``ntpath.normpath`` also resolves to one) regardless of what the URI's
850 nominal authority says — "file://localhost//attacker.example.com/share"
851 reads as localhost but resolves off-box. Such a URI must stay
852 non-contained even after ``_is_local_filesystem_uri`` has declined it,
853 so hostname classification cannot rescue it via the local-looking
854 authority.
856 The "is this a ``file://`` URI?" question is answered by the PARSED
857 scheme, not by a raw ``startswith`` on the string: ``urlsplit`` deletes
858 tab/CR/LF from anywhere in a URL and lstrips C0 controls, so
859 "fi\\tle://localhost//attacker.example.com/share/db" is not a ``file://``
860 URI to ``startswith`` but IS one to every real URL parser — and
861 ``_classify_engine_url``, the frame that decides the host, is a real URL
862 parser, which for that value reports the local-looking authority
863 ``localhost`` and would classify the endpoint contained. Asking
864 ``urlsplit`` for the scheme is what keeps this function from being
865 stepped around by such a value.
867 ``value`` is normalised by ``_normalize_uri_like_url_parser`` BEFORE it
868 reaches ``urlsplit``, and that call is load-bearing here, not a duplicate
869 of the parser's own preprocessing. ``urlsplit`` lstrips only
870 ``\\x00``-``\\x20``, while the normaliser's final ``strip()`` is
871 bidirectional and additionally removes the 19 leading Unicode-whitespace
872 codepoints ``str.strip()`` knows and the parser does not (U+0085, U+00A0,
873 U+1680, U+2000-U+200A, U+2028, U+2029, U+202F, U+205F, U+3000). Handed
874 the RAW value, "\\xa0file://localhost//attacker.example.com/share/db"
875 parses with an EMPTY scheme, this function answers False, the
876 network-share rejection never fires, and — inside a list-shaped endpoint
877 setting whose sibling entry is "127.0.0.1" — the whole setting classifies
878 CONTAINED. Normalise first; see
879 ``test_leading_unicode_whitespace_cannot_hide_a_network_share``.
881 Fails toward "remote" for a malformed authority: ``urlsplit`` raises
882 (``ValueError``: "Invalid IPv6 URL") on an unbalanced IPv6 bracket rather
883 than parsing, and an endpoint we cannot parse is one we cannot prove
884 local. The catch is broad, matching its sibling in
885 ``_is_local_filesystem_uri``, so a future parser error class cannot turn
886 a fail-closed answer into an exception that escapes into
887 ``_resolve_adaptive_scope``'s permissive catch-all.
889 Non-strings answer False here — they are not ``file://`` URIs, and every
890 other non-string shape already fails closed downstream by never proving
891 itself local.
892 """
893 if not isinstance(value, str):
894 return False
895 uri = _normalize_uri_like_url_parser(value)
896 try:
897 parsed = urlsplit(uri)
898 if parsed.scheme.lower() != "file":
899 return False
900 return _starts_with_double_separator(unquote(parsed.path))
901 except Exception:
902 return True
905def _vector_store_endpoint_is_local(
906 provider: str,
907 settings_snapshot: Optional[dict],
908 ctx: EgressContext,
909 allow_dns: bool = True,
910) -> bool:
911 """Whether a non-local-file store's configured endpoint stays on the box.
913 Mirrors the engine ``url_setting`` fail-up: the provider class names the
914 settings key holding its endpoint (``uri_setting``), that value is
915 classified by DNS, and ONLY a positively-local answer counts as contained.
916 A provider that declares no key, an unset/unparseable URI, or a host that
917 resolves publicly all yield False — for a store that is not a local file,
918 an endpoint we cannot prove local must be assumed to be off-machine.
919 """
920 cls = _vector_store_class(provider)
921 uri_setting = getattr(cls, "uri_setting", None) if cls else None
922 if not uri_setting:
923 return False
924 uri = _get_setting_value(settings_snapshot or {}, uri_setting, None)
925 # Reject a network-share file path PER ENTRY. A setting value is not
926 # necessarily a string: ``unwrap_setting`` peels the ``{"value": ...}``
927 # snapshot wrapper and returns whatever it wrapped, uncoerced, and
928 # ``_classify_engine_url`` below explicitly supports the LIST shape
929 # (Elasticsearch-style endpoint lists). A string-only guard here would
930 # leave a one-element list to fall through to hostname classification,
931 # where a nominally local authority ("localhost", "127.0.0.1") answers
932 # True and overrules the path rejection. Normalising to entries first
933 # keeps a single rejection site that covers every shape the setting can
934 # hold, and fails the WHOLE setting closed when ANY entry is a rejected
935 # path — matching the "any public entry wins" fail-up
936 # ``_classify_engine_url`` already applies to list-typed endpoints.
937 #
938 # COUPLED WITH ``_classify_engine_url``: this loop and that function must
939 # keep the SAME container test (``isinstance(..., list)``, not
940 # ``(list, tuple)``) and the same "skip non-``str`` entries" rule. If one
941 # side alone grows tuple support, a tuple-shaped setting reaches hostname
942 # classification with a rejected network-share path still in it — the
943 # exact bypass this per-entry loop exists to close. Change both or
944 # neither.
945 for entry in uri if isinstance(uri, list) else (uri,):
946 if _file_uri_names_a_remote_share(entry):
947 return False
948 if _is_local_filesystem_uri(uri):
949 return True
950 return (
951 _classify_engine_url(
952 uri_setting, settings_snapshot or {}, ctx, allow_dns=allow_dns
953 )
954 is True
955 )
958def vector_store_is_contained(
959 engine_name: str,
960 ctx: EgressContext,
961 settings_snapshot: Optional[dict],
962 allow_dns: bool = True,
963) -> bool:
964 """Whether the RAG vector store(s) behind ``engine_name`` stay on the box.
966 A library/collection engine is only as contained as the vector store its
967 embeddings are written to and searched in. A local-file store (FAISS) is
968 contained by construction — ``is_local_file`` is the discriminator, so
969 this is a strict no-op for it and no DNS lookup happens. A server-backed
970 store is contained only while its configured endpoint resolves local;
971 otherwise the collection's embeddings (and every query) leave the machine
972 and the engine must fail UP, exactly as Paperless/Elasticsearch do via
973 ``url_setting``.
975 Single home for the rule so the scope PEP (:func:`classify_engine`) and
976 the two-axis resolver (``run_classification.engine_label``) can never
977 diverge.
978 """
979 for provider in _resolve_vector_store_providers(
980 engine_name, ctx, settings_snapshot
981 ):
982 cls = _vector_store_class(provider)
983 if cls is not None:
984 if getattr(cls, "is_local_file", False) is True:
985 continue
986 elif provider in _LOCAL_FILE_VECTOR_STORE_PROVIDERS:
987 continue
988 if not _vector_store_endpoint_is_local(
989 provider, settings_snapshot, ctx, allow_dns=allow_dns
990 ):
991 return False
992 return True
995def classify_engine(
996 engine_name: str,
997 ctx: EgressContext,
998 *,
999 settings_snapshot: dict,
1000 metadata: Optional[dict] = None,
1001 allow_dns: bool = True,
1002) -> EngineClassification:
1003 """Classify an engine from its flags, collection metadata, and URL policy.
1005 The class-level ``is_public`` / ``is_local`` flags express the engine's
1006 *nature* — whether it queries the public internet or searches local data —
1007 regardless of where the engine's server happens to be hosted. This is used
1008 by ``evaluate_engine`` and ``_resolve_adaptive_scope`` so that e.g. a
1009 locally-hosted SearXNG (which proxies to Google/Bing) is still treated as
1010 a public engine.
1012 The URL override is asymmetric and can only make the classification
1013 MORE restrictive: a local-nature engine whose ``url_setting`` resolves
1014 to a public host is reclassified public (queries would leave the box),
1015 while a public-nature engine on a local URL stays public.
1017 - Static engines: read declared ``is_public``/``is_local`` flags directly.
1018 - ``library`` / ``collection_<uuid>``: per-collection ``is_public``
1019 flag (from ``metadata`` when the caller has the engine config, else a
1020 direct DB lookup), defaulting private.
1021 - Anything else (unknown): ``EngineClassification(None, None)``.
1022 """
1023 engine_cls = _get_engine_class(engine_name)
1024 if engine_cls is None:
1025 if engine_name == "library" or engine_name.startswith("collection_"):
1026 if metadata is not None and metadata.get("is_public") is not None:
1027 is_public = bool(metadata.get("is_public"))
1028 else:
1029 is_public = _resolve_collection_is_public(
1030 engine_name, ctx.username
1031 )
1032 # A collection is ALWAYS a local knowledge base — it lives on this
1033 # machine — so it stays searchable under PRIVATE_ONLY regardless of
1034 # its public flag. ``is_public`` is ADDITIVE, not exclusive: marking
1035 # a collection public (non-sensitive content — papers you fetched,
1036 # public publications) ALSO makes it eligible under PUBLIC_ONLY and
1037 # OK to process with cloud inference. A private collection is
1038 # local-only and excluded from public-scope runs. Hence
1039 # ``(is_public, is_local=True)`` rather than the old mutually-
1040 # exclusive ``(is_public, not is_public)`` which wrongly hid a
1041 # public collection from private runs.
1042 #
1043 # ... UNLESS the vector store behind it is not on this machine.
1044 # A collection is only local while its index is: a server-backed
1045 # store on a public endpoint ships the collection's embeddings and
1046 # every query to a third party, so it fails UP — but unlike the
1047 # Elasticsearch/Paperless case (a nature-private engine that
1048 # becomes nature-public once its endpoint leaves the box), a
1049 # collection's ``is_public`` is an independent, already-resolved
1050 # signal about the *content*, not a byproduct of containment.
1051 # Losing containment must only ever ADD restriction, never grant
1052 # new eligibility: it strips ``is_local`` (PRIVATE_ONLY denies it,
1053 # data no longer stays on the box) while PRESERVING the real
1054 # ``is_public`` flag (a private-flagged collection does NOT
1055 # become newly eligible under PUBLIC_ONLY just because its store
1056 # is remote — that would be a relaxation, not a tightening).
1057 # Strict no-op for a local-file store (FAISS):
1058 # ``vector_store_is_contained`` returns True without touching DNS.
1059 if not vector_store_is_contained(
1060 engine_name, ctx, settings_snapshot, allow_dns=allow_dns
1061 ):
1062 return EngineClassification(is_public=is_public, is_local=False)
1063 return EngineClassification(is_public=is_public, is_local=True)
1064 return EngineClassification(is_public=None, is_local=None)
1065 is_public, is_local, url_setting = _engine_flags(engine_cls)
1066 # Fail-up URL override (asymmetric, only ever MORE restrictive): a
1067 # local-data engine (Elasticsearch, Paperless) whose configured URL
1068 # resolves to a PUBLIC host is reclassified as public — querying it
1069 # sends the user's queries off the box, so PRIVATE_ONLY must deny it
1070 # at selection time (matching pre-static-flags behavior). The reverse
1071 # direction is deliberately NOT applied: a public-nature engine
1072 # (SearXNG) hosted on localhost stays public, because its data source
1073 # is the internet regardless of where the proxy runs.
1074 if url_setting and is_local is True and is_public is not True:
1075 host_classification = _classify_engine_url(
1076 url_setting, settings_snapshot, ctx, allow_dns=allow_dns
1077 )
1078 if host_classification is False:
1079 # False from _classify_engine_url == "connect target is a
1080 # PUBLIC host" -> fail up to public.
1081 return EngineClassification(is_public=True, is_local=False)
1082 return EngineClassification(is_public=is_public, is_local=is_local)
1085# Denial reasons that justify dropping an engine from an LLM candidate
1086# list. Only ACTIVE policy denials qualify — see filter_engines_by_egress.
1087_PREFILTER_STRIP_REASONS = frozenset(
1088 {
1089 "scope_mismatch_public_only",
1090 "scope_mismatch_private_only",
1091 "strict_not_primary",
1092 # Registered engine without classification flags: the factory's
1093 # evaluate_engine call fails it closed too, so stripping matches
1094 # enforcement exactly.
1095 "unclassified",
1096 }
1097)
1100def filter_engines_by_egress(
1101 engine_names: list[str],
1102 ctx: EgressContext,
1103 settings_snapshot: dict,
1104) -> list[str]:
1105 """ADVISORY pre-filter for a candidate engine list: remove engines the
1106 egress scope actively denies so the LLM doesn't waste selection slots
1107 on them. Uses the engine's static class flags (``is_public`` /
1108 ``is_local``) — a locally-hosted SearXNG is still treated as public
1109 because it queries the internet.
1111 This is a UX optimization in FRONT of the factory PEP, not an
1112 enforcement point — so it must never be STRICTER than the factory.
1113 Names unknown to the static registry (``engine_unknown``) are KEPT
1114 (except under STRICT, where the not-the-primary gate fires before the
1115 registry lookup — exactly as it does in the factory):
1116 they may be retriever-backed or dynamically injected engines that the
1117 factory evaluates via its own retriever path; dropping them here would
1118 silently hide legitimate engines from LLM selection. The factory
1119 still denies anything truly disallowed at instantiation.
1121 Callers: the LangGraph tool builder, or any code that builds a
1122 selection set before presenting it to an LLM.
1123 """
1124 allowed: list[str] = []
1125 for name in engine_names:
1126 decision = evaluate_engine(
1127 name, ctx, settings_snapshot=settings_snapshot
1128 )
1129 if decision.allowed or decision.reason not in _PREFILTER_STRIP_REASONS:
1130 allowed.append(name)
1131 return allowed
1134def filter_candidates_by_egress(
1135 engine_names: list[str],
1136 settings_snapshot: Optional[dict],
1137) -> list[str]:
1138 """Best-effort scope pre-filter for an LLM candidate list.
1140 Wraps ``filter_engines_by_egress`` with the snapshot plumbing
1141 LLM-selection callers share: read ``policy.egress_scope`` /
1142 ``search.tool`` from the snapshot (flat or ``{"value": ...}`` shaped),
1143 build the context, filter. Returns the input list unchanged when the
1144 snapshot is missing, the scope is BOTH (nothing to strip), or anything
1145 fails — the factory PEP still enforces at child instantiation, so this
1146 helper must never break engine selection.
1147 """
1148 if not settings_snapshot:
1149 return engine_names
1150 try:
1151 scope_raw = str(
1152 _get_setting_value(
1153 settings_snapshot, "policy.egress_scope", DEFAULT_EGRESS_SCOPE
1154 )
1155 or DEFAULT_EGRESS_SCOPE
1156 ).lower()
1157 # BOTH (or an unrecognized value) strips nothing at selection time;
1158 # adaptive is included because it can RESOLVE to a restrictive scope.
1159 if scope_raw not in (
1160 "private_only",
1161 "public_only",
1162 "strict",
1163 "adaptive",
1164 ):
1165 return engine_names
1166 primary = resolve_run_primary_engine(settings_snapshot)
1167 ctx = context_from_snapshot(
1168 settings_snapshot,
1169 primary,
1170 username=settings_snapshot.get("_username"),
1171 )
1172 return filter_engines_by_egress(engine_names, ctx, settings_snapshot)
1173 except Exception: # noqa: silent-exception - advisory pre-filter only
1174 logger.bind(policy_audit=True).debug(
1175 "egress candidate pre-filter failed; "
1176 "factory PEP still enforces at child instantiation"
1177 )
1178 return engine_names
1181def evaluate_engine(
1182 engine_name: str,
1183 ctx: EgressContext,
1184 *,
1185 settings_snapshot: dict,
1186 metadata: Optional[dict] = None,
1187) -> Decision:
1188 """Decide whether an engine may be instantiated under the current policy.
1190 Allows iff (a) the engine's classification is compatible with the scope's
1191 bucket AND (b) if scope == STRICT, the engine name equals the primary.
1192 Unclassified engines always fail closed.
1194 ``metadata`` (optional) carries the engine's config entry (e.g. the
1195 per-collection ``is_public`` from search_config) so callers that already
1196 have it avoid a redundant DB lookup; when absent it's resolved directly.
1197 """
1198 if settings_snapshot is None:
1199 return Decision(False, "no_snapshot")
1200 try:
1201 # UNPROTECTED: egress protection disabled for this run (escape hatch)
1202 # — any engine is permitted. SSRF/metadata blocks still apply at
1203 # evaluate_url; this only lifts the egress-scope engine gate.
1204 if ctx.scope == EgressScope.UNPROTECTED:
1205 return Decision(True, "egress_unprotected")
1207 # STRICT: only the primary engine is permitted.
1208 if (
1209 ctx.scope == EgressScope.STRICT
1210 and engine_name != ctx.primary_engine
1211 ):
1212 return Decision(False, "strict_not_primary")
1214 # A truly unknown engine (not in the static registry and not a
1215 # collection) fails closed as engine_unknown. ``library`` /
1216 # ``collection_*`` are classified by classify_engine below.
1217 is_collection = engine_name == "library" or engine_name.startswith(
1218 "collection_"
1219 )
1220 if _get_engine_class(engine_name) is None and not is_collection:
1221 return Decision(False, "engine_unknown")
1223 classification = classify_engine(
1224 engine_name,
1225 ctx,
1226 settings_snapshot=settings_snapshot,
1227 metadata=metadata,
1228 )
1230 # Fail-closed for unclassified engines.
1231 if (
1232 classification.is_public is not True
1233 and classification.is_local is not True
1234 ):
1235 return Decision(False, "unclassified")
1237 # Scope/bucket compatibility.
1238 if (
1239 ctx.scope == EgressScope.PUBLIC_ONLY
1240 and classification.is_public is not True
1241 ):
1242 return Decision(False, "scope_mismatch_public_only")
1243 if (
1244 ctx.scope == EgressScope.PRIVATE_ONLY
1245 and classification.is_local is not True
1246 ):
1247 return Decision(False, "scope_mismatch_private_only")
1249 return Decision(True, "allowed")
1250 except Exception: # pragma: no cover - defensive
1251 logger.bind(policy_audit=True).exception(
1252 "evaluate_engine internal error", engine=engine_name
1253 )
1254 return Decision(False, "internal_error")
1257def _classify_engine_url(
1258 url_setting: str,
1259 settings_snapshot: dict,
1260 ctx: EgressContext,
1261 allow_dns: bool = True,
1262) -> Optional[bool]:
1263 """Classify an engine's configured URL via DNS resolution.
1265 For list-typed settings (Elasticsearch ``hosts``), returns False (public)
1266 if ANY entry classifies as public — safer fail-up.
1267 Returns True if local, False if public, None if undetermined.
1268 """
1269 value = _get_setting_value(settings_snapshot, url_setting, None)
1270 if not value:
1271 return None
1273 # COUPLED WITH the per-entry network-share rejection in
1274 # ``_vector_store_endpoint_is_local``: both must keep the SAME container
1275 # test (``isinstance(..., list)``, not ``(list, tuple)``) and the same
1276 # "skip non-``str`` entries" rule. Widening only one side lets a
1277 # tuple-shaped setting reach this classifier with an entry the other side
1278 # never examined. Change both or neither.
1279 entries = value if isinstance(value, list) else [value]
1280 any_public = False
1281 any_local = False
1282 for entry in entries:
1283 if not isinstance(entry, str): 1283 ↛ 1284line 1283 didn't jump to line 1284 because the condition on line 1283 was never true
1284 continue
1285 try:
1286 if "://" in entry:
1287 to_parse = entry
1288 elif entry.startswith("//"):
1289 # Scheme-less "//host[:port]" authority form (e.g. a Milvus
1290 # endpoint written without a scheme). Prepending "http://"
1291 # here would yield "http:////host:port", which urlsplit
1292 # mis-parses as an empty netloc with the host landing in
1293 # ``path`` — silently dropping the hostname from
1294 # classification. Prepend only the scheme so the leading
1295 # "//" the entry already supplies becomes the URL's
1296 # authority marker.
1297 to_parse = f"http:{entry}"
1298 else:
1299 to_parse = f"http://{entry}"
1300 parsed = urlsplit(to_parse)
1301 hostname = parsed.hostname
1302 except Exception:
1303 continue
1304 if not hostname:
1305 continue
1306 # Decode percent-encoding before classifying, identically to
1307 # evaluate_url: the HTTP client decodes the host before connecting,
1308 # so "192%2e168%2e1%2e1" must be classified as 192.168.1.1, not as
1309 # an unresolvable literal.
1310 hostname = unquote(hostname)
1311 result = _classify_host(hostname, ctx, allow_dns=allow_dns)
1312 if result is True:
1313 any_local = True
1314 elif result is False:
1315 any_public = True
1317 if any_public:
1318 return False # "any public" wins under list-typed settings
1319 if any_local:
1320 return True
1321 return None
1324_CLOUD_LLM_PROVIDERS = frozenset(
1325 {
1326 "openai",
1327 "anthropic",
1328 "google",
1329 "openrouter",
1330 "requesty",
1331 "orcarouter",
1332 "atlascloud",
1333 "deepseek",
1334 "xai",
1335 "ionos",
1336 }
1337)
1339# Providers that default to a localhost endpoint and ship without any
1340# remote routing. The LLM gate at ``config/llm_config.py`` uses this set
1341# as a SNAPSHOT-LESS ALLOW-LIST: when ``get_llm`` is invoked with no
1342# settings snapshot (background helpers, scaffolding paths, tests), only
1343# these providers may proceed; everything else — including ambiguous
1344# providers like ``openai_endpoint`` and any future cloud provider not
1345# yet enumerated in ``_CLOUD_LLM_PROVIDERS`` — fails closed at the gate.
1346# Keeping the set tight is the point: anything new defaults to refused.
1347_LOCAL_DEFAULT_LLM_PROVIDERS = frozenset(
1348 {
1349 "ollama",
1350 "lmstudio",
1351 "llamacpp",
1352 }
1353)
1355# Embeddings analogue of ``_LOCAL_DEFAULT_LLM_PROVIDERS``. Used by the
1356# embeddings gate as a SNAPSHOT-LESS ALLOW-LIST: when ``get_embeddings``
1357# runs without a settings snapshot (the ``embeddings.require_local`` toggle
1358# is then unreadable), only these localhost-default providers may proceed.
1359# ``openai`` (and any future cloud embeddings provider) fails closed —
1360# matching the LLM gate so a snapshot-less background caller can't ship the
1361# local corpus to a cloud embedder. ``sentence_transformers`` runs fully
1362# in-process; ``ollama`` defaults to a localhost endpoint.
1363_LOCAL_DEFAULT_EMBEDDING_PROVIDERS = frozenset(
1364 {
1365 "sentence_transformers",
1366 "ollama",
1367 }
1368)
1371def _is_user_registered_llm(
1372 provider: str, username: Optional[str] = None
1373) -> bool:
1374 """True when ``provider`` is a user-registered in-process LLM.
1376 The programmatic API (``quick_summary(llms={"mock": ...})``) and plugins
1377 register LLM objects in the LLM registry. Built-in providers are ALSO
1378 auto-registered there by ``discover_providers()``, so registry membership
1379 alone cannot discriminate — a name that is registered but NOT one of the
1380 auto-discovered built-ins is user-supplied in-process code. Such objects
1381 carry no configurable endpoint the PDP could classify (denying them with
1382 ``provider_url_unset`` just breaks the documented offline workflow), the
1383 operator injected them deliberately at code level, and under
1384 PRIVATE_ONLY/STRICT the PEP-578 audit hook still blocks any stray
1385 outbound socket they might open.
1387 A user registration that *shadows* a built-in name (e.g. "openai") is NOT
1388 treated as user code: the discovered-name check keeps it on the strict
1389 path, and the ``_CLOUD_LLM_PROVIDERS`` gate fires first anyway.
1391 Fails closed (False) on any error.
1392 """
1393 try:
1394 from ...llm.llm_registry import is_llm_registered
1395 from ...llm.providers import discover_providers
1396 from ...llm.providers.base import normalize_provider
1398 if not provider or not is_llm_registered(provider, username=username):
1399 return False
1400 discovered = {normalize_provider(key) for key in discover_providers()}
1401 return normalize_provider(provider) not in discovered
1402 except Exception: # pragma: no cover - defensive
1403 logger.bind(policy_audit=True).exception(
1404 "user-registered LLM check failed", provider=provider
1405 )
1406 return False
1409def evaluate_llm_endpoint(
1410 provider: str,
1411 ctx: EgressContext,
1412 *,
1413 settings_snapshot: dict,
1414 username: Optional[str] = None,
1415) -> Decision:
1416 """Decide whether an LLM provider may be instantiated.
1418 Only meaningful when ``ctx.require_local_llm`` is True; otherwise allow.
1420 ``username`` scopes the user-registered-LLM exemption to the caller's own
1421 namespace so a per-user in-process LLM is recognized. It defaults to the
1422 context's username when the caller does not pass one explicitly.
1423 """
1424 if username is None:
1425 username = getattr(ctx, "username", None)
1426 if settings_snapshot is None:
1427 return Decision(False, "no_snapshot")
1428 try:
1429 if not ctx.require_local_llm:
1430 return Decision(True, "no_local_requirement")
1432 # Hard-cloud providers always fail under require_local_llm.
1433 if provider in _CLOUD_LLM_PROVIDERS:
1434 return Decision(False, "provider_cloud_only")
1436 # User-registered in-process LLMs (programmatic API / plugins) have
1437 # no endpoint to classify; allow them — the audit hook still
1438 # backstops stray sockets under PRIVATE_ONLY/STRICT.
1439 if _is_user_registered_llm(provider, username=username):
1440 return Decision(True, "user_registered_llm")
1442 # Providers with a configurable URL: classify the URL.
1443 url_key = f"llm.{provider}.url"
1444 url_value = _get_setting_value(settings_snapshot, url_key, None)
1445 if not url_value:
1446 # Local-default providers (ollama, lmstudio, llamacpp) without
1447 # an override fall back to their localhost defaults. Use the
1448 # shared constant so this can't drift from the snapshot-less
1449 # allow-list.
1450 if provider in _LOCAL_DEFAULT_LLM_PROVIDERS:
1451 return Decision(True, "provider_local_default")
1452 return Decision(False, "provider_url_unset")
1454 try:
1455 parsed = urlsplit(url_value)
1456 # Percent-decode the host before classifying — the HTTP client
1457 # decodes it before connect, so "http://127%2e0%2e0%2e1:11434"
1458 # must read as the local 127.0.0.1 (matches evaluate_url and
1459 # _classify_engine_url). Without this a legitimate percent-encoded
1460 # local endpoint is wrongly classified public and denied.
1461 hostname = unquote(parsed.hostname) if parsed.hostname else None
1462 except Exception:
1463 return Decision(False, "url_malformed")
1465 classification = _classify_host(hostname, ctx) if hostname else None
1466 if classification is True:
1467 return Decision(True, "provider_local")
1468 return Decision(False, "provider_remote")
1469 except Exception: # pragma: no cover - defensive
1470 logger.bind(policy_audit=True).exception(
1471 "evaluate_llm_endpoint internal error", provider=provider
1472 )
1473 return Decision(False, "internal_error")
1476def evaluate_embeddings(
1477 provider: str,
1478 ctx: EgressContext,
1479 *,
1480 settings_snapshot: dict,
1481) -> Decision:
1482 """Decide whether an embeddings provider may be instantiated.
1484 Only meaningful when ``ctx.require_local_embeddings`` is True.
1485 OpenAI is treated as unambiguously cloud unless ``embeddings.openai.base_url``
1486 is configured to a local host.
1487 """
1488 if settings_snapshot is None:
1489 return Decision(False, "no_snapshot")
1490 try:
1491 if not ctx.require_local_embeddings:
1492 return Decision(True, "no_local_requirement")
1494 if provider == "sentence_transformers":
1495 return Decision(True, "provider_local")
1497 if provider == "ollama":
1498 url = _get_setting_value(
1499 settings_snapshot, "embeddings.ollama.url", None
1500 ) or _get_setting_value(settings_snapshot, "llm.ollama.url", None)
1501 if not url:
1502 return Decision(True, "provider_local_default")
1503 parsed = urlsplit(url)
1504 host = unquote(parsed.hostname) if parsed.hostname else None
1505 classification = _classify_host(host, ctx) if host else None
1506 return (
1507 Decision(True, "provider_local")
1508 if classification is True
1509 else Decision(False, "provider_remote")
1510 )
1512 if provider == "openai":
1513 base_url = _get_setting_value(
1514 settings_snapshot, "embeddings.openai.base_url", None
1515 )
1516 if base_url:
1517 parsed = urlsplit(base_url)
1518 host = unquote(parsed.hostname) if parsed.hostname else None
1519 classification = _classify_host(host, ctx) if host else None
1520 if classification is True:
1521 return Decision(True, "provider_local_endpoint")
1522 return Decision(False, "provider_cloud")
1524 return Decision(False, "provider_unknown")
1525 except Exception: # pragma: no cover - defensive
1526 logger.bind(policy_audit=True).exception(
1527 "evaluate_embeddings internal error", provider=provider
1528 )
1529 return Decision(False, "internal_error")
1532_DANGEROUS_SCHEMES = frozenset(
1533 {"javascript", "data", "file", "vbscript", "about"}
1534)
1536# Cloud-metadata endpoints reachable by HOSTNAME rather than literal IP. GCP's
1537# IMDS answers on metadata.google.internal / metadata.goog (resolving to
1538# 169.254.169.254); is_ip_blocked only catches the IP literal, so these names
1539# need an explicit block to honor the "metadata is NEVER permitted, regardless
1540# of scope" invariant in evaluate_url. AWS/Azure/Alibaba IMDS have no such
1541# hostname (IP only) and are covered by the is_ip_blocked / alt-encoding path.
1542_METADATA_HOSTNAMES = frozenset({"metadata.google.internal", "metadata.goog"})
1545def _normalize_alt_ipv4(host: str) -> Optional[str]:
1546 """Return the canonical dotted-quad for an IPv4 host written in an
1547 alternate encoding the libc resolver accepts but ``ipaddress.ip_address``
1548 rejects — octal (``0251.0376.0251.0376``), hex (``0xa9fea9fe``), integer
1549 (``2852039166``) and short forms (``169.254.43518``). Returns ``None`` for
1550 anything ``inet_aton`` won't parse (real hostnames, IPv6, junk).
1552 ``socket.getaddrinfo`` resolves these numeric forms to the same address the
1553 HTTP client will ``connect()`` to, so the metadata-IP block in
1554 ``evaluate_url`` must classify them by their *real* target — otherwise
1555 ``http://0251.0376.0251.0376/`` slips past ``is_ip_blocked`` (which only
1556 parses canonical notation) and reads as an allowed public host.
1557 """
1558 try:
1559 import socket
1561 return socket.inet_ntoa(socket.inet_aton(host))
1562 except (OSError, UnicodeError, TypeError):
1563 return None
1566def _quota_ctx(ctx: EgressContext) -> EgressContext:
1567 """Return the EgressContext whose counter backs the per-RUN denied-fetch
1568 quota.
1570 Each call site (ContentFetcher, full_search, download_service, the audit
1571 hook) builds its OWN EgressContext from the snapshot, so a per-context
1572 counter would reset the budget every time a new engine/fetcher is built —
1573 letting a malicious document evade ``MAX_DENIED_FETCHES_PER_RUN`` by
1574 spreading denied fetches across contexts. The run's ARMED active context is
1575 set once at run start and re-armed identically on pool workers, so it is
1576 shared across the whole run; anchoring the counter there makes the quota
1577 truly per-run. Falls back to the passed ctx when no run context is armed
1578 (snapshot-less / programmatic callers, settings-page calls), keeping those
1579 isolated calls self-contained.
1580 """
1581 try:
1582 from .audit_hook import get_active_context
1584 active = get_active_context()
1585 except Exception: # pragma: no cover - defensive
1586 active = None
1587 return active if active is not None else ctx
1590def evaluate_url(url: str, ctx: EgressContext) -> Decision:
1591 """Decide whether an arbitrary URL may be fetched (e.g., by ``fetch_content``).
1593 Enforces the run's denied-fetch quota to prevent exhaustion attacks
1594 via malicious indexed documents that loop the agent through hundreds
1595 of denied fetches.
1596 """
1597 try:
1598 # Quota check: hard fail after MAX_DENIED_FETCHES_PER_RUN. Anchor the
1599 # counter to the run's active context (``_quota_ctx``) so the budget is
1600 # per-RUN, not per-EgressContext. Read under that ctx's lock so
1601 # concurrent subagents see a consistent count.
1602 qctx = _quota_ctx(ctx)
1603 with qctx._lock:
1604 if qctx._fetch_denial_count["count"] >= MAX_DENIED_FETCHES_PER_RUN:
1605 return Decision(False, "denial_quota_exceeded")
1607 if not isinstance(url, str) or not url:
1608 return _record_denial(ctx, url, "url_malformed")
1610 try:
1611 parsed = urlsplit(url)
1612 except Exception:
1613 return _record_denial(ctx, url, "url_malformed")
1615 if parsed.scheme.lower() in _DANGEROUS_SCHEMES:
1616 return _record_denial(ctx, url, "dangerous_scheme")
1617 if parsed.scheme.lower() not in ("http", "https"):
1618 return _record_denial(ctx, url, "unsupported_scheme")
1619 if not parsed.hostname:
1620 return _record_denial(ctx, url, "no_hostname")
1622 # HTTP client libraries (requests/urllib3) percent-DECODE the host
1623 # in the netloc before the socket connect, but urlsplit().hostname
1624 # preserves the encoding. Classifying the encoded form lets
1625 # "http://192%2e168%2e1%2e1/" read as an unresolvable public host
1626 # (DNS fails) while the client actually connects to the private
1627 # 192.168.1.1 — a scope bypass under PUBLIC_ONLY. Classify the
1628 # DECODED host so the policy sees the real connect target.
1629 host = unquote(parsed.hostname)
1630 # Trailing dots are insignificant to the resolver (getaddrinfo strips
1631 # them), so "169.254.169.254." and "metadata.google.internal." must be
1632 # classified identically to their bare forms — otherwise a trailing
1633 # dot dodges the metadata checks below. Mirrors the SSRF validator's
1634 # rstrip(".").
1635 host = host.rstrip(".")
1637 # Cloud-metadata endpoints (AWS/GCE/Azure IMDS at 169.254.169.254
1638 # etc.) are NEVER permitted, regardless of scope. They classify as
1639 # link-local/private, so STRICT and PRIVATE_ONLY would otherwise
1640 # ALLOW them — a credential-theft path for prompt-injected fetches
1641 # and, more importantly, for the audit-hook net which calls
1642 # evaluate_url directly on raw socket.connect targets (bypassing the
1643 # SSRF validator that the explicit fetch PEPs run first). Reuse
1644 # is_ip_blocked(allow_private_ips=True), which returns True only for
1645 # the always-blocked metadata set (+ NAT64 wraps) and also unwraps
1646 # IPv4-mapped IPv6 forms.
1647 try:
1648 from ..ssrf_validator import is_ip_blocked
1650 if is_ip_blocked(
1651 host, allow_localhost=True, allow_private_ips=True
1652 ):
1653 return _record_denial(ctx, url, "blocked_metadata_ip")
1654 # is_ip_blocked only parses canonical IP notation, so an
1655 # alternate-encoded metadata literal (octal/hex/integer) slips
1656 # through above. Normalize it to the dotted-quad the resolver
1657 # would connect to and re-check, so the "metadata IPs are NEVER
1658 # permitted, regardless of scope" invariant holds for those forms
1659 # too — and they get the explicit blocked_metadata_ip reason
1660 # rather than leaking into the scope-mismatch bucket.
1661 alt = _normalize_alt_ipv4(host)
1662 if (
1663 alt is not None
1664 and alt != host
1665 and is_ip_blocked(
1666 alt, allow_localhost=True, allow_private_ips=True
1667 )
1668 ):
1669 return _record_denial(ctx, url, "blocked_metadata_ip")
1670 # Metadata endpoints reachable by hostname (GCP) — is_ip_blocked
1671 # can't see these because they aren't IP literals. Block by name so
1672 # the invariant holds under PUBLIC_ONLY/BOTH too (the SSRF validator
1673 # backstops the actual fetch, but evaluate_url's own guarantee must
1674 # not depend on it).
1675 if host.lower() in _METADATA_HOSTNAMES:
1676 return _record_denial(ctx, url, "blocked_metadata_ip")
1677 except Exception: # noqa: silent-exception - defensive, see below
1678 # Non-IP host or helper error: fall through to normal
1679 # classification (DNS path has its own handling).
1680 pass
1682 # UNPROTECTED: egress protection disabled — any host is allowed. This
1683 # gate is deliberately AFTER the SSRF / cloud-metadata block above
1684 # (which always applies), so the escape hatch only lifts the
1685 # egress-scope host restriction, never the hard SSRF invariant.
1686 if ctx.scope == EgressScope.UNPROTECTED:
1687 return Decision(True, "egress_unprotected")
1689 classification = _classify_host(host, ctx)
1691 if ctx.scope == EgressScope.STRICT:
1692 # Under STRICT, URLs whose host is private are allowed; public
1693 # hosts (DOI redirects, citations) are not. This is a deliberate
1694 # trade-off: prompt-injection spoofing of provenance tags is a
1695 # bigger risk than false-positives on DOI redirects.
1696 if classification is True:
1697 return Decision(True, "allowed_private_host_under_strict")
1698 return _record_denial(ctx, url, "strict_public_host")
1700 if ctx.scope == EgressScope.PUBLIC_ONLY:
1701 if classification is False:
1702 return Decision(True, "allowed_public_host")
1703 return _record_denial(ctx, url, "scope_mismatch_public_only")
1705 if ctx.scope == EgressScope.PRIVATE_ONLY:
1706 if classification is True:
1707 return Decision(True, "allowed_private_host")
1708 return _record_denial(ctx, url, "scope_mismatch_private_only")
1710 # BOTH: any classified host is fine.
1711 if classification is None: 1711 ↛ 1712line 1711 didn't jump to line 1712 because the condition on line 1711 was never true
1712 return _record_denial(ctx, url, "host_unclassified")
1713 return Decision(True, "allowed_both_scope")
1714 except Exception: # pragma: no cover - defensive
1715 logger.bind(policy_audit=True).exception("evaluate_url internal error")
1716 return Decision(False, "internal_error")
1719def _record_denial(ctx: EgressContext, url: object, reason: str) -> Decision:
1720 """Increment the denial counter inside the frozen context's mutable dict
1721 and emit a redacted audit log line. Lock guards the read-modify-write
1722 against concurrent subagent threads.
1724 The offending URL is inlined into the WARNING message body rather than
1725 bound as a kwarg, matching the rest of the app's log convention (see
1726 ``[FILE_INTEGRITY]`` audit lines, successful-fetch ``[N] Title: ... URL: ...``
1727 observations). The ``database_sink`` persists ``record["message"]``
1728 verbatim into ``app_logs.message`` — and that is what the JSONL exporter
1729 emits — so this is the only layer at which a triager can read the URL.
1731 Note: after ``langgraph_agent_strategy._observation_event`` was taught
1732 to return ``None`` for ``fetch_content`` denial/error strings, the chat
1733 panel no longer emits a ``📄 From the page: Cannot fetch …`` MILESTONE
1734 for denials. The persisted WARNING is therefore the **only** record of
1735 the denial in the research log — if the URL is missing from this
1736 message body, there is nothing else to cross-reference. The URL still
1737 passes through ``redact_url_for_log`` so tokens, query strings, and
1738 paths never reach the audit log.
1739 """
1740 from ..ssrf_validator import redact_url_for_log
1742 counts_toward_quota = reason not in _NON_QUOTA_DENIAL_REASONS
1743 # Increment the per-RUN counter (the run's active context, shared across
1744 # every call-site context) rather than this one context's — see _quota_ctx.
1745 qctx = _quota_ctx(ctx)
1746 with qctx._lock:
1747 if counts_toward_quota:
1748 qctx._fetch_denial_count["count"] += 1
1749 count = qctx._fetch_denial_count["count"]
1750 # ``url_malformed`` and the ``urlsplit``-except path can hand us a
1751 # non-string (None, bytes, int) — urllib3.parse_url raises TypeError on
1752 # int/bytes and returns a Url with scheme=None for None, neither of which
1753 # we want on the audit line. Coerce to a stable, type-tagged placeholder
1754 # so the WARNING still carries a useful field even when the URL itself
1755 # was unparseable. The denial decision is unaffected — we already
1756 # decided to refuse.
1757 if not isinstance(url, str):
1758 audit_url = f"<unparseable:{type(url).__name__}>"
1759 else:
1760 audit_url = redact_url_for_log(url)
1761 # The URL / reason / scope / count / counted must ride on the message
1762 # body, NOT as logger.warning kwargs. database_sink persists
1763 # record["message"] into ``app_logs.message`` verbatim and only copies a
1764 # small fixed set of structured fields into the DB row. It *does* read
1765 # selected routing metadata from ``record["extra"]`` (for example
1766 # ``research_id`` / ``username``), but arbitrary extras or warning kwargs
1767 # are not serialized into ``app_logs.message`` (or the JSONL exporter that
1768 # reads it). The pinned
1769 # ``test_record_denial_warning_message_carries_redacted_url`` and
1770 # ``test_record_denial_warning_message_reaches_persisted_log`` tests
1771 # assert this invariant.
1772 logger.bind(policy_audit=True).warning(
1773 f"policy denied URL fetch url={audit_url} reason={reason} "
1774 f"scope={ctx.scope.value} count={count} counted={counts_toward_quota}"
1775 )
1776 return Decision(False, reason)
1779def evaluate_retriever(
1780 retriever_name: str,
1781 ctx: EgressContext,
1782 *,
1783 metadata: Optional[dict] = None,
1784) -> Decision:
1785 """Decide whether a registered retriever may be invoked.
1787 Reads classification (``{"is_local": bool}``) set at registration
1788 time. Unclassified retrievers fail closed. ``metadata`` may be
1789 supplied by the caller (e.g. the search-engine factory, which has
1790 already looked up the retriever in its own registry reference) to
1791 avoid a second registry lookup and to honor a test-patched registry;
1792 when ``None`` the global registry is consulted.
1793 """
1794 try:
1795 # UNPROTECTED: egress protection disabled — any retriever is permitted.
1796 if ctx.scope == EgressScope.UNPROTECTED:
1797 return Decision(True, "egress_unprotected")
1799 if metadata is None:
1800 from ...web_search_engines.retriever_registry import (
1801 retriever_registry,
1802 )
1804 try:
1805 metadata = retriever_registry.get_metadata(
1806 retriever_name, username=ctx.username
1807 )
1808 except AttributeError:
1809 # Older registry without metadata API; treat as unclassified.
1810 return Decision(False, "retriever_unclassified")
1812 if not metadata:
1813 return Decision(False, "retriever_unknown")
1815 is_local = metadata.get("is_local")
1816 if is_local is None: 1816 ↛ 1817line 1816 didn't jump to line 1817 because the condition on line 1816 was never true
1817 return Decision(False, "retriever_unclassified")
1819 if ctx.scope == EgressScope.STRICT:
1820 # STRICT permits only the user's primary engine. A retriever
1821 # IS allowed under STRICT when it is itself the primary (the
1822 # common "research only against my private KB" setup);
1823 # otherwise it's an expansion STRICT forbids.
1824 if retriever_name == ctx.primary_engine:
1825 return Decision(True, "allowed_primary_retriever")
1826 return Decision(False, "strict_not_primary")
1827 if ctx.scope == EgressScope.PUBLIC_ONLY and is_local:
1828 return Decision(False, "scope_mismatch_public_only")
1829 if ctx.scope == EgressScope.PRIVATE_ONLY and not is_local: 1829 ↛ 1830line 1829 didn't jump to line 1830 because the condition on line 1829 was never true
1830 return Decision(False, "scope_mismatch_private_only")
1831 return Decision(True, "allowed")
1832 except Exception: # pragma: no cover - defensive
1833 logger.bind(policy_audit=True).exception(
1834 "evaluate_retriever internal error", retriever=retriever_name
1835 )
1836 return Decision(False, "internal_error")
1839def _retriever_is_local(
1840 retriever_name: str, username: Optional[str]
1841) -> Optional[bool]:
1842 """Return the ``is_local`` classification (True/False) of a registered
1843 retriever, or ``None`` when it is unknown/unclassified.
1845 Mirrors the registry lookup in ``evaluate_retriever`` so adaptive-scope
1846 resolution and enforcement agree on a retriever primary's classification.
1847 """
1848 try:
1849 from ...web_search_engines.retriever_registry import (
1850 retriever_registry,
1851 )
1853 metadata = retriever_registry.get_metadata(
1854 retriever_name, username=username
1855 )
1856 except Exception: # noqa: silent-exception - unknown → caller falls back
1857 return None
1858 if not metadata:
1859 return None
1860 is_local = metadata.get("is_local")
1861 return is_local if isinstance(is_local, bool) else None
1864# Bounded, thread-safe, PER-USER dedup for the ADAPTIVE fail-open WARNING
1865# emitted below. ``context_from_snapshot`` runs at every research-run start
1866# AND at every per-URL fetch-gate check (``BaseSearchEngine.
1867# _build_full_search_egress_context``, called once per fetched URL), so an
1868# unclassifiable primary would otherwise emit one identical WARNING per
1869# fetched URL -- potentially hundreds per run. We remember, PER USER, which
1870# (sanitized) primary engines have already warned, so a genuinely new
1871# (user, primary) combination still logs once while repeats are silent.
1872#
1873# The dedup MUST be keyed on the actual user identity (a stable hash of the
1874# username, ``None`` bucket for an unauthenticated / un-threaded run) -- NOT
1875# merely on "was a username present". Keying on presence alone means a
1876# genuine fail-open for user B on a primary name already logged for user A is
1877# silently swallowed: the audit signal for B never fires. Keying on identity
1878# keeps each user's fail-opens independently detectable.
1879#
1880# ``primary_engine`` is user-controlled (``search.tool``), so the dedup state
1881# is bounded in BOTH dimensions -- mirroring the capped ``warned: set`` dedup
1882# in ``web/services/socketio_asgi.py::_install_origin_rejection_logging``:
1883# * each user's bucket caps the distinct primaries it remembers
1884# (``_ADAPTIVE_FAILOPEN_WARN_CAP_PER_USER``); once full it stops
1885# warning/recording, exactly as the old single global set did -- but now
1886# that ceiling is reached PER USER, so it only ever silences the ONE user
1887# who flooded it, and
1888# * the set of tracked users is a bounded LRU
1889# (``_ADAPTIVE_FAILOPEN_MAX_USERS``); the least-recently-seen user's whole
1890# bucket is evicted beyond the cap (which can only ever re-enable a
1891# warning, never suppress one).
1892# A single authenticated user flooding distinct junk primaries therefore only
1893# ever exhausts THEIR OWN bucket -- it can never suppress the fail-open audit
1894# warnings of any other user (the bug a single global set had: 512 junk
1895# primaries from one user permanently silenced everyone). Total memory is
1896# capped at MAX_USERS * WARN_CAP_PER_USER short strings.
1897_ADAPTIVE_FAILOPEN_WARNED: "OrderedDict[Optional[str], OrderedDict[str, None]]" = OrderedDict()
1898_ADAPTIVE_FAILOPEN_LOCK = threading.Lock()
1899# Max distinct (sanitized) primaries remembered per user; once a user's bucket
1900# is full we stop warning/recording for that user (memory-safe, and preserves
1901# the original anti-log-amplification behavior) -- but PER USER, so it never
1902# touches another user's audit budget.
1903_ADAPTIVE_FAILOPEN_WARN_CAP_PER_USER = 128
1904# Max distinct users tracked at once; the least-recently-seen user's whole
1905# bucket is LRU-evicted beyond this, keeping total memory bounded.
1906_ADAPTIVE_FAILOPEN_MAX_USERS = 256
1909def _adaptive_failopen_user_key(username: Optional[str]) -> Optional[str]:
1910 """Stable, bounded-length bucket key for the per-user fail-open dedup.
1912 ``None`` (unauthenticated / un-threaded run) is its own bucket. A present
1913 username is hashed rather than stored raw: it bounds the key length
1914 regardless of the username, and avoids retaining plaintext identities in
1915 this long-lived process-global audit structure (the warning itself only
1916 ever emits ``username_present``, never the name).
1917 """
1918 if username is None:
1919 return None
1920 return hashlib.sha256(username.encode("utf-8")).hexdigest()[:16]
1923def _should_warn_adaptive_failopen(
1924 sanitized_primary: str, username: Optional[str]
1925) -> bool:
1926 """True the first time this ``(user, sanitized_primary)`` pair is seen.
1928 Dedup is PER USER: the bucket key is a stable hash of ``username``
1929 (``None`` for an unauthenticated run), so a genuine fail-open for one user
1930 is never swallowed by an already-logged fail-open for a DIFFERENT user on
1931 the same primary name. Returns False for a repeat of an already-recorded
1932 pair, and for any NEW primary once *this* user's bucket is full (bounded,
1933 and only that user is affected). The set of tracked users is a bounded LRU
1934 -- see the ``_ADAPTIVE_FAILOPEN_WARNED`` docstring above for the
1935 memory-safety and cross-user-isolation rationale.
1936 """
1937 user_key = _adaptive_failopen_user_key(username)
1938 with _ADAPTIVE_FAILOPEN_LOCK:
1939 bucket = _ADAPTIVE_FAILOPEN_WARNED.get(user_key)
1940 if bucket is None:
1941 # First fail-open for this user. Enforce the users-tracked cap by
1942 # LRU-evicting the least-recently-seen user before inserting.
1943 if len(_ADAPTIVE_FAILOPEN_WARNED) >= _ADAPTIVE_FAILOPEN_MAX_USERS:
1944 _ADAPTIVE_FAILOPEN_WARNED.popitem(last=False)
1945 bucket = OrderedDict()
1946 _ADAPTIVE_FAILOPEN_WARNED[user_key] = bucket
1947 else:
1948 # Seen this user before -> mark most-recently-used so an active
1949 # user is never the one evicted by the users cap.
1950 _ADAPTIVE_FAILOPEN_WARNED.move_to_end(user_key)
1952 if sanitized_primary in bucket:
1953 return False
1954 # This user's per-user budget is full: stop warning AND recording, so
1955 # one user's junk-primary flood can neither grow memory nor amplify
1956 # logs. Bounded and scoped to this ONE user's bucket -- every other
1957 # user keeps their full audit budget.
1958 if len(bucket) >= _ADAPTIVE_FAILOPEN_WARN_CAP_PER_USER:
1959 return False
1960 bucket[sanitized_primary] = None
1961 return True
1964def _resolve_adaptive_scope(
1965 primary_engine: str,
1966 settings_snapshot: dict,
1967 *,
1968 username: Optional[str],
1969 local_hostnames: tuple,
1970 allow_dns: bool = True,
1971) -> EgressScope:
1972 """Map ADAPTIVE to a concrete scope by classifying the primary engine.
1974 Uses the engine's **static class flags** (plus ``classify_engine``'s
1975 fail-up URL override) so that e.g. a locally-hosted SearXNG is still
1976 treated as a public engine, and a remote-hosted Elasticsearch resolves
1977 public rather than pulling the run into PRIVATE_ONLY.
1979 private primary => PRIVATE_ONLY, public primary => PUBLIC_ONLY,
1980 unknown / classification-error => BOTH (adaptive's documented
1981 permissive fallback, equivalent to pre-policy behavior, so a
1982 classification hiccup never hard-fails the run).
1984 A registered LangChain retriever (private KB) is not in ENGINE_REGISTRY,
1985 so ``classify_engine`` returns no classification for it; we then consult the
1986 retriever registry so a local retriever primary resolves to PRIVATE_ONLY
1987 (forcing local inference) rather than leaking the corpus to cloud models.
1988 """
1989 if not primary_engine:
1990 return EgressScope.BOTH
1991 try:
1992 probe_ctx = EgressContext(
1993 scope=EgressScope.BOTH,
1994 primary_engine=primary_engine,
1995 require_local_llm=False,
1996 require_local_embeddings=False,
1997 local_hostnames=local_hostnames,
1998 username=username,
1999 )
2000 classification = classify_engine(
2001 primary_engine,
2002 probe_ctx,
2003 settings_snapshot=settings_snapshot,
2004 allow_dns=allow_dns,
2005 )
2006 except Exception: # noqa: silent-exception - adaptive falls back to BOTH
2007 logger.bind(policy_audit=True).debug(
2008 "adaptive scope classification failed; falling back to BOTH",
2009 primary=primary_engine,
2010 )
2011 return EgressScope.BOTH
2013 if classification.is_local is True and classification.is_public is not True:
2014 return EgressScope.PRIVATE_ONLY
2015 if classification.is_public is True and classification.is_local is not True:
2016 return EgressScope.PUBLIC_ONLY
2018 # SECURITY (fail-open regression): a PRIVATE non-contained collection —
2019 # its vector store lives off this machine — classifies as
2020 # (is_public=False, is_local=False). ``classify_engine``'s fail-up logic
2021 # (see its docstring) strips ``is_local`` when containment is lost while
2022 # PRESERVING the real ``is_public`` flag, so this is the mirror image of
2023 # a PUBLIC non-contained collection: that one correctly matches the
2024 # ``is_public is True`` branch above and resolves PUBLIC_ONLY, but a
2025 # PRIVATE one matches NEITHER exclusive branch and, without this rule,
2026 # fell through all the way to the permissive ``BOTH`` fallback below —
2027 # silently dropping the forced-local-LLM/embeddings guarantee
2028 # (``context_from_snapshot``'s PRIVATE_ONLY coupling) for exactly the
2029 # collection that most needs it, and resolving MORE permissively than
2030 # the public case despite being the more sensitive one. Resolve to the
2031 # RESTRICTIVE PRIVATE_ONLY instead — this preserves the pre-PR
2032 # forced-local posture. The collection engine itself still gets denied
2033 # by ``evaluate_engine`` as "unclassified" (neither flag is True)
2034 # regardless of the scope resolved here; PRIVATE_ONLY's role is to keep
2035 # every OTHER engine/inference path in the run local too, rather than
2036 # letting the run silently widen to cloud-capable BOTH. This branch is
2037 # reached via the private-non-contained-collection fail-up above, or by a
2038 # static engine whose class overrides NEITHER flag: ``BaseSearchEngine``
2039 # declares ``is_public = False`` and ``is_local = False`` as CLASS
2040 # DEFAULTS (``search_engine_base.py``), so such a subclass inherits
2041 # ``(False, False)`` from the base class — it need not explicitly write
2042 # either assignment itself — and lands HERE, not in the ``(None, None)``
2043 # branch below. ``_engine_flags`` only returns ``None`` for a flag that is
2044 # genuinely absent from the class's MRO, which no attribute on
2045 # ``BaseSearchEngine`` ever is; all 29 current registry engines override
2046 # exactly one flag to ``True`` and so don't reach this branch today, but
2047 # the next engine author who forgets to override either flag will, not
2048 # the retriever-lookup fallback. Dedup/log via the same per-(user, primary) mechanism
2049 # as the fail-open warning below — this fires on the identical hot path
2050 # (once per research-run start AND once per fetch-gate check).
2051 if classification.is_public is False and classification.is_local is False:
2052 safe_primary = sanitize_for_log(primary_engine)
2053 if _should_warn_adaptive_failopen(f"nonlocal:{safe_primary}", username):
2054 logger.bind(policy_audit=True).warning(
2055 f"ADAPTIVE egress: primary engine '{safe_primary}' classified "
2056 f"as neither public nor local-contained (a PRIVATE collection "
2057 f"whose vector store is not on this machine, or a "
2058 f"mis-declared static engine); resolving to the RESTRICTIVE "
2059 f"PRIVATE_ONLY scope (forcing local LLM/embeddings) rather "
2060 f"than the permissive BOTH."
2061 )
2062 return EgressScope.PRIVATE_ONLY
2064 # Unknown to classify_engine — it may be a registered retriever (private
2065 # KB). Classify via the retriever registry before falling back to BOTH.
2066 if classification.is_public is None and classification.is_local is None:
2067 retriever_local = _retriever_is_local(primary_engine, username)
2068 if retriever_local is True:
2069 return EgressScope.PRIVATE_ONLY
2070 if retriever_local is False:
2071 return EgressScope.PUBLIC_ONLY
2073 # SECURITY (detectable fail-open): the primary could not be classified
2074 # in ANY namespace visible here — it is not a static engine, not a
2075 # collection, and not a retriever resolvable for ``username``. ADAPTIVE
2076 # therefore resolves to the PERMISSIVE ``BOTH`` (cloud-capable) scope.
2077 #
2078 # The benign cause is a genuinely-unregistered / typo'd / removed
2079 # meta-engine primary. The DANGEROUS cause is a per-user PRIVATE
2080 # retriever that is invisible here because ``username=`` was not
2081 # threaded into this resolution (or the snapshot lacked ``_username``):
2082 # the retriever would classify as local (=> PRIVATE_ONLY, forcing local
2083 # inference) if its owner's namespace were consulted, but under the
2084 # wrong/absent username it disappears and the run silently fails open
2085 # to cloud-capable BOTH.
2086 #
2087 # We do NOT fail closed here: reaching this branch means classify_engine
2088 # returned (None, None), i.e. a *legitimately-public* primary can never
2089 # land here (a public engine classifies is_public=True => PUBLIC_ONLY
2090 # above). But we still cannot cleanly separate "benign unregistered
2091 # primary" from "invisible per-user retriever" without peeking across
2092 # other users' registry namespaces (breaking the isolation this PR
2093 # establishes, and misfiring PRIVATE_ONLY onto a legitimate
2094 # unregistered public primary that merely shares a name). Since the
2095 # distinction isn't clean, per the security-hardening guidance we keep
2096 # the non-breaking permissive behavior and make it DETECTABLE instead:
2097 # a missed ``username=`` threading now shows up as this WARNING rather
2098 # than silently permissive egress.
2099 # primary_engine is user-controlled (search.tool): scrub it before
2100 # interpolating into the log line so embedded newlines/control chars
2101 # can't forge additional log entries (log injection). Dedup on the
2102 # SANITIZED value too, per _should_warn_adaptive_failopen above, so
2103 # this fires at most once per unique (user, primary) pair in the
2104 # process rather than once per call. The dedup keys on the ACTUAL
2105 # username (hashed), not merely "was one present", so a genuine
2106 # fail-open for one user is never swallowed by another user's.
2107 safe_primary = sanitize_for_log(primary_engine)
2108 if _should_warn_adaptive_failopen(safe_primary, username):
2109 logger.bind(policy_audit=True).warning(
2110 f"ADAPTIVE egress fail-open: primary engine "
2111 f"'{safe_primary}' could not be classified in any visible "
2112 f"namespace (static engine, collection, or retriever visible to "
2113 f"this user); resolving PERMISSIVELY to BOTH (cloud-capable). If "
2114 f"'{safe_primary}' is a per-user PRIVATE retriever this is a "
2115 f"fail-open — verify username= is threaded into egress "
2116 f"resolution (username_present={username is not None})."
2117 )
2118 return EgressScope.BOTH
2120 return EgressScope.BOTH
2123def context_from_snapshot(
2124 settings_snapshot: dict,
2125 primary_engine: str,
2126 *,
2127 username: Optional[str] = None,
2128 allow_dns: bool = True,
2129) -> EgressContext:
2130 """Construct the frozen ``EgressContext`` for a research run.
2132 Reads policy settings out of the snapshot exactly once at run-start.
2133 A missing ``policy.egress_scope`` yields ``DEFAULT_EGRESS_SCOPE``
2134 (``adaptive`` — the protective default that follows the primary engine),
2135 NOT a permissive fallback. A residual ``both`` value is coerced to
2136 ``adaptive`` below.
2138 ``allow_dns=False`` makes ADAPTIVE resolution skip the synchronous
2139 getaddrinfo when classifying a URL-configurable primary engine — used by
2140 the advisory warning-banner render path so a settings-page load never
2141 blocks on DNS. Enforcement callers leave it True so classification is
2142 accurate. (Has no effect for non-ADAPTIVE scopes, which never resolve
2143 a URL.)
2144 """
2145 if settings_snapshot is None:
2146 raise ValueError("settings_snapshot is required")
2147 if not isinstance(settings_snapshot, dict):
2148 # Fail closed with the same ValueError contract callers already
2149 # handle (llm_config / research_service convert it to a hard policy
2150 # stop). A non-dict snapshot would otherwise crash deeper in
2151 # _get_setting_value with a bare AttributeError, which a broad
2152 # caller-side except could swallow into a permissive default.
2153 raise ValueError(
2154 f"settings_snapshot must be a dict, got "
2155 f"{type(settings_snapshot).__name__}"
2156 )
2158 scope_raw = _get_setting_value(
2159 settings_snapshot, "policy.egress_scope", DEFAULT_EGRESS_SCOPE
2160 )
2161 from ...settings.manager import check_env_setting
2163 env_scope = check_env_setting("policy.egress_scope")
2164 if env_scope is not None:
2165 scope_raw = env_scope
2166 # `both` is retired (ADR-0007): the blanket-permissive middle scope is
2167 # coerced to the protective `adaptive` default so a residual value — an
2168 # un-migrated DB, a queued snapshot, or an env-var override that bypasses
2169 # settings validation — can never silently keep the old no-protection
2170 # behaviour. Migration 0019 rewrites stored values; this is the read-time
2171 # backstop. (EgressScope.BOTH still exists as the INTERNAL resolution result
2172 # for an unclassifiable ADAPTIVE primary — it is never a user-facing scope.)
2173 if str(scope_raw).strip().lower() == EgressScope.BOTH.value:
2174 scope_raw = EgressScope.ADAPTIVE.value
2175 try:
2176 scope = parse_user_egress_scope(
2177 scope_raw, disabled_unprotected="adaptive"
2178 )
2179 except PolicyDeniedError:
2180 # An unrecognised scope means the saved setting was corrupted
2181 # or tampered with — silently falling back to BOTH (the most
2182 # permissive scope) would mask policy violations rather than
2183 # surface them. Refuse the run instead so the operator notices.
2184 logger.bind(policy_audit=True).warning(
2185 "refusing to construct EgressContext from unknown scope",
2186 value=scope_raw,
2187 )
2188 raise
2190 if (
2191 str(scope_raw).strip().lower() == EgressScope.UNPROTECTED.value
2192 and scope == EgressScope.ADAPTIVE
2193 ):
2194 logger.bind(policy_audit=True).warning(
2195 "unprotected egress is disabled; coercing scope to adaptive"
2196 )
2198 env_require_local_llm = check_env_setting("llm.require_local_endpoint")
2199 env_require_local_embeddings = check_env_setting("embeddings.require_local")
2200 require_local_llm = _coerce_bool(
2201 env_require_local_llm
2202 if env_require_local_llm is not None
2203 else _get_setting_value(
2204 settings_snapshot, "llm.require_local_endpoint", False
2205 )
2206 )
2207 require_local_embeddings = _coerce_bool(
2208 env_require_local_embeddings
2209 if env_require_local_embeddings is not None
2210 else _get_setting_value(
2211 settings_snapshot, "embeddings.require_local", False
2212 )
2213 )
2215 raw_hostnames = _get_setting_value(
2216 settings_snapshot, "llm.allowed_local_hostnames", ()
2217 )
2218 if isinstance(raw_hostnames, (list, tuple)): 2218 ↛ 2221line 2218 didn't jump to line 2221 because the condition on line 2218 was always true
2219 local_hostnames = tuple(h for h in raw_hostnames if isinstance(h, str))
2220 else:
2221 local_hostnames = ()
2223 # ADAPTIVE resolves to a concrete scope by classifying the primary
2224 # engine: a concrete private primary => PRIVATE_ONLY, a concrete public
2225 # primary => PUBLIC_ONLY, an unclassifiable primary => BOTH.
2226 # The resolved scope is what the EgressContext stores and what
2227 # every downstream PEP enforces. Classification reuses classify_engine
2228 # via a throwaway BOTH-scoped context (its DNS cache is discarded).
2229 if scope == EgressScope.ADAPTIVE:
2230 scope = _resolve_adaptive_scope(
2231 primary_engine,
2232 settings_snapshot,
2233 username=username,
2234 local_hostnames=local_hostnames,
2235 allow_dns=allow_dns,
2236 )
2238 # PRIVATE_ONLY means "my data stays on this box." That guarantee only
2239 # holds if BOTH inference paths are local — a cloud LLM receives the
2240 # query + retrieved local chunks, and a cloud embedder receives the
2241 # whole corpus at ingest. So under PRIVATE_ONLY the require_local_*
2242 # flags are IMPLIED: scope overrides them, so a user who left them at
2243 # their default (False) can't silently exfiltrate via inference. This
2244 # also fires when ADAPTIVE resolved to PRIVATE_ONLY above.
2245 #
2246 # STRICT is deliberately NOT coupled here: it restricts the search
2247 # engine set to the primary only and is orthogonal to where inference
2248 # runs (a user may legitimately want single-engine search + a cloud
2249 # LLM). Callers that gate on ctx.require_local_* therefore get
2250 # scope-correct behaviour without a separate flag read.
2251 if scope == EgressScope.PRIVATE_ONLY:
2252 require_local_llm = True
2253 require_local_embeddings = True
2254 elif scope == EgressScope.UNPROTECTED:
2255 # The escape hatch lifts user-level locality requirements, but cannot
2256 # weaken immutable operator environment locks.
2257 if env_require_local_llm is None:
2258 require_local_llm = False
2259 if env_require_local_embeddings is None:
2260 require_local_embeddings = False
2262 return EgressContext(
2263 scope=scope,
2264 primary_engine=primary_engine,
2265 require_local_llm=require_local_llm,
2266 require_local_embeddings=require_local_embeddings,
2267 local_hostnames=local_hostnames,
2268 username=username,
2269 )
2272def _get_setting_value(snapshot: dict, key: str, default):
2273 """Read a setting from the snapshot.
2275 Accepts both flat ({key: value}) and nested ({key: {"value": value}})
2276 schemas — both shapes appear in LDR's settings infrastructure.
2277 """
2278 raw = snapshot.get(key, default)
2279 return unwrap_setting(raw)
2282def coerce_str_list(raw) -> tuple[bool, list[str]]:
2283 """Decode a JSON-list-ish setting value into a list of strings.
2285 Returns ``(ok, values)``. Accepts a real list/tuple or a JSON-string list
2286 and filters to string entries. ``ok`` is False only when a non-empty JSON
2287 string failed to parse — callers that must *reject* malformed input branch
2288 on it; fail-safe callers ignore it and use the (empty) list. Single home for
2289 the decode shared by the trust resolver, the trust banner, and the
2290 settings-save validators.
2291 """
2292 if isinstance(raw, str):
2293 if not raw.strip(): 2293 ↛ 2294line 2293 didn't jump to line 2294 because the condition on line 2293 was never true
2294 return True, []
2295 try:
2296 raw = json.loads(raw)
2297 except Exception:
2298 return False, []
2299 if isinstance(raw, (list, tuple)): 2299 ↛ 2301line 2299 didn't jump to line 2301 because the condition on line 2299 was always true
2300 return True, [x for x in raw if isinstance(x, str)]
2301 return True, []
2304def resolve_run_primary_engine(
2305 settings_snapshot: Optional[dict], default: Optional[str] = None
2306) -> str:
2307 """Resolve a research run's primary search engine id from a snapshot.
2309 Single source of truth for "which engine drives this run's egress
2310 scope". Under the default ADAPTIVE scope the concrete scope is derived by
2311 classifying THIS primary, so run-scoped callers that build an
2312 ``EgressContext`` should derive the primary here. Otherwise two layers can
2313 resolve ADAPTIVE to different scopes and one silently under-enforces — the
2314 LangGraph tool-list filter once derived the primary from the engine CLASS
2315 name (a collection -> ``"libraryrag"`` -> unclassified -> BOTH) while the
2316 factory PEP derived it from ``search.tool`` (the real collection key ->
2317 PRIVATE_ONLY), so a public engine reached the agent and was then
2318 hard-denied mid-run (``scope_mismatch_private_only``).
2320 Reads ``search.tool`` (flat or ``{"value": ...}`` shaped). A blank-after-
2321 strip (``" "``) or non-string (``5``, list/dict) value is treated as
2322 MISSING — it must not slip through the truthiness check and classify to the
2323 permissive BOTH scope.
2325 A run with NO configured primary is a wiring error, not a normal state.
2326 Silently substituting a default would set the egress scope from an engine
2327 the user never chose — and since the system default is public (``searxng``),
2328 that fails OPEN. So a missing/empty ``search.tool`` raises ``ValueError``
2329 unless the caller passes an explicit ``default``:
2331 * run-level callers (research_service, the strategy tool-list filter,
2332 ``filter_candidates_by_egress``) pass none and fail closed — the worker
2333 refuses the run; advisory filters degrade to unfiltered with the factory
2334 PEP still enforcing.
2335 * the search-engine factory passes ``default=engine_name`` because it is
2336 evaluating that one specific engine, so deriving the scope from it is
2337 meaningful rather than arbitrary.
2339 Raises:
2340 ValueError: ``search.tool`` is missing/blank/non-string and no truthy
2341 ``default`` is given.
2342 """
2343 primary = (
2344 _get_setting_value(settings_snapshot, "search.tool", None)
2345 if settings_snapshot
2346 else None
2347 )
2348 # Blank-after-strip or non-string is not a usable engine id → missing.
2349 primary = primary.strip() if isinstance(primary, str) else None
2350 if primary:
2351 return primary
2352 if default:
2353 return default
2354 raise ValueError(
2355 "no primary search engine configured: settings 'search.tool' is "
2356 "missing, blank, or not a string, so the run's egress scope cannot "
2357 "be resolved"
2358 )
2361def build_run_egress_context(
2362 settings_snapshot: Optional[dict],
2363 search_engine: Optional[str],
2364 *,
2365 username: Optional[str] = None,
2366) -> EgressContext:
2367 """Resolve the research worker's per-run ``EgressContext``.
2369 Composes the policy setup ``run_research_process`` runs at startup:
2370 current operator env policy is reapplied on the snapshot, the primary
2371 engine is derived, and the ``EgressContext`` is constructed. This is the
2372 single production seam invoked by the worker; a stale queued snapshot
2373 carrying an operator-disabled ``unprotected`` scope is coerced to a
2374 protected scope here (via the read-time backstop in
2375 ``context_from_snapshot``), not at queue extraction time.
2376 """
2377 from ...settings.manager import (
2378 apply_environment_overrides_to_snapshot,
2379 )
2381 effective = apply_environment_overrides_to_snapshot(settings_snapshot or {})
2382 primary = resolve_run_primary_engine({"search.tool": search_engine})
2383 return context_from_snapshot(effective, primary, username=username)
2386def _coerce_bool(value) -> bool:
2387 """Coerce a setting value to bool, defensively.
2389 Strings ``"true"`` / ``"True"`` are True; anything else string-shaped
2390 is False. Real booleans pass through. The strict coercion prevents
2391 type confusion (e.g., the string ``"false"`` being truthy).
2392 """
2393 if isinstance(value, bool):
2394 return value
2395 if isinstance(value, str): 2395 ↛ 2397line 2395 didn't jump to line 2397 because the condition on line 2395 was always true
2396 return value.strip().lower() in ("true", "1", "yes", "on")
2397 return bool(value)