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