Coverage for src/local_deep_research/security/egress/warnings.py: 93%
145 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""Egress-policy warning checks.
3Check functions: take primitive values, return a warning dict or ``None``.
4These surface on the research-form banner so users can see at a glance
5when their current policy lets data leave the machine — or silently
6blocks something they configured. Most checks are pure; the engine-URL
7checks additionally read operator environment state (never the DB) and
8never resolve DNS, so all of them stay cheap enough for the
9page-render hot path.
11The banners, each with its own dismiss flag:
131. **Public search egress enabled** — the active scope permits any public
14 engine to fire. The default scope ``adaptive`` can resolve to a
15 public-allowing posture, so this fires on a fresh install too; suppressed
16 via the dismiss flag so first-launch isn't a wall of red.
172. **Cloud LLM enabled** — the user hasn't opted into
18 ``llm.require_local_endpoint`` and the configured provider is one of
19 the unambiguously-cloud providers. Critical-severity because the
20 user's full prompt content leaks on every research run.
213. **Cloud embeddings enabled** — even worse: indexing a corpus with
22 OpenAI embeddings POSTs every chunk to OpenAI. Critical-severity.
234. **Unprotected egress / effective scope / trusted destinations** —
24 posture-transparency banners for the egress-scope machinery.
255. **SearXNG private URL not approved** — the selected public engine's
26 private instance URL lacks operator approval, so the engine will
27 silently self-disable at run time.
286. **Local engine public URL** — the mirror image: a LOCAL document
29 engine (Paperless, Elasticsearch) points at a host that does not look
30 private, so document queries would leave the box and a private-only
31 scope excludes the engine.
32"""
34from typing import Optional
36# Single source of truth — these warning checks now live in the same
37# egress package as the PDP, so import the cloud-provider set directly
38# instead of maintaining a hand-synced copy (which previously risked
39# drift: a provider added to one list but not the other).
40from .policy import _CLOUD_LLM_PROVIDERS
43def check_public_egress_enabled(
44 egress_scope: str,
45 acknowledged: bool,
46) -> Optional[dict]:
47 """Banner when the active scope permits public-internet search engines.
49 Fires for ``adaptive`` (which can resolve to a public posture), ``both``,
50 and ``public_only`` — the scopes that can allow public engines. Suppressed
51 when the user has acknowledged the
52 egress-policy warnings via the fresh-install flag — otherwise every
53 new install would face three loud banners before doing anything.
54 """
55 if acknowledged:
56 return None
57 # "adaptive" can resolve to a public-allowing scope (public or
58 # unclassifiable primary), so warn conservatively — better a dismissible
59 # banner than a silent public-egress path the user didn't expect.
60 if egress_scope not in ("both", "public_only", "adaptive"):
61 return None
63 return {
64 "type": "public_egress_enabled",
65 "icon": "🌐",
66 "title": "Public search egress enabled",
67 "message": (
68 "This run can reach external search engines. Set the Egress "
69 "Scope to 'Private only' below if you want to keep all "
70 "research traffic on-machine."
71 ),
72 "dismissKey": "app.warnings.dismiss_egress_policy",
73 "actionUrl": "#policy_egress_scope",
74 "actionLabel": "Adjust scope",
75 }
78def check_unprotected_egress(egress_scope: str) -> Optional[dict]:
79 """Loud, non-dismissible banner when egress protection is turned off.
81 The ``unprotected`` escape hatch disables all egress-scope restrictions
82 for the run — any engine / URL / LLM / embeddings provider is permitted
83 (only the hard SSRF and cloud-metadata blocks remain). Deliberately not
84 dismissible: the user should never forget protection is off.
85 """
86 if (egress_scope or "").lower() != "unprotected":
87 return None
89 return {
90 "type": "egress_unprotected",
91 "icon": "⚠️",
92 "title": "Egress protection is disabled",
93 "message": (
94 "Egress Scope is set to 'Unprotected' — this run may reach any "
95 "search engine, URL, and LLM/embeddings provider. Only hard SSRF "
96 "and cloud-metadata blocking still applies. Pick another scope "
97 "below to re-enable protection."
98 ),
99 "dismissKey": None,
100 "actionUrl": "#policy_egress_scope",
101 "actionLabel": "Adjust scope",
102 }
105def check_trusted_destinations(
106 trusted_inference, trusted_engines
107) -> Optional[dict]:
108 """Banner when the user has vouched for off-machine destinations.
110 Trusting an inference provider or search engine relaxes the two-axis
111 classification (treats an off-machine sink as contained), so sensitive
112 sources may flow to it. Surfaced so a stale trust entry can't silently keep
113 sending data off-box.
114 """
115 from .policy import coerce_str_list
117 names = sorted(
118 {
119 n
120 for n in coerce_str_list(trusted_inference)[1]
121 + coerce_str_list(trusted_engines)[1]
122 if n.strip()
123 }
124 )
125 if not names:
126 return None
128 return {
129 "type": "egress_trusted_destinations",
130 "icon": "🤝",
131 "title": "Trusted destinations configured",
132 "message": (
133 "These off-machine destinations are marked trusted, so sensitive "
134 "sources may be combined with them: "
135 + ", ".join(names)
136 + ". Remove any you did not intend."
137 ),
138 "dismissKey": None,
139 "actionUrl": "#policy_egress_scope",
140 "actionLabel": "Review",
141 }
144def check_effective_scope(
145 egress_scope: str,
146 effective_scope: str,
147 primary_engine: str,
148 acknowledged: bool,
149) -> Optional[dict]:
150 """Informational banner stating what the ADAPTIVE scope actually resolves
151 to for the current primary engine.
153 Adaptive is opaque on its own ("follows your primary"); this makes the
154 effective posture explicit so the user knows whether THIS config means
155 public searches, private/local-only, or both. Only fires for ``adaptive``
156 (the explicit scopes are self-describing in the dropdown). Has its own
157 dismiss flag so dismissing it doesn't hide the risk banners.
158 """
159 if acknowledged:
160 return None
161 if (egress_scope or "").lower() != "adaptive":
162 return None
163 eff = (effective_scope or "").lower()
164 primary = primary_engine or "your primary engine"
165 base = {
166 "type": "egress_effective_scope",
167 "dismissKey": "app.warnings.dismiss_adaptive_scope_info",
168 "actionUrl": "#policy_egress_scope",
169 "actionLabel": "Change mode",
170 }
171 if eff == "private_only":
172 return {
173 **base,
174 "icon": "🔒",
175 "title": "Adaptive → Private only (stays local)",
176 "message": (
177 f"Your primary engine ('{primary}') is private, so this run "
178 "stays on your machine: only local engines run, and LLM + "
179 "embeddings are forced local — nothing leaves the box."
180 ),
181 }
182 if eff == "public_only":
183 return {
184 **base,
185 "icon": "🌐",
186 "title": "Adaptive → Public searches enabled",
187 "message": (
188 f"Your primary engine ('{primary}') is public, so this run "
189 "uses public web/academic engines. Your local collections are "
190 "not queried."
191 ),
192 }
193 # both (unclassifiable primary)
194 return {
195 **base,
196 "icon": "🔀",
197 "title": "Adaptive → Public + private searches enabled",
198 "message": (
199 f"Your primary ('{primary}') could not be classified as public "
200 "or private, so this run can use both public engines and your "
201 "local collections."
202 ),
203 }
206def check_private_engine_url_blocked(
207 display_name: str,
208 engine_name: str,
209 url_setting: str,
210 instance_url: str,
211 active: bool,
212 acknowledged: bool,
213) -> Optional[dict]:
214 """Banner when an active PUBLIC engine's instance URL is a
215 private/loopback address the egress guard will refuse to fetch.
217 Generic over any engine in ``guarded_engine_url_descriptors()`` with
218 ``is_public=True`` — the banner type, dismiss key, env-lock variable,
219 and settings anchor are all derived from ``engine_name`` /
220 ``url_setting``, so a future public engine with a configurable URL is
221 covered by declaration, not by new code.
223 Without this, the failure mode is quiet: the engine disables itself at
224 init with only a log-panel ERROR, and research runs simply return no
225 results from it. This banner surfaces the misconfiguration on the
226 research form with the exact operator remedies.
228 Cheap by design (no DNS on the page-render hot path): a literal-IP host
229 is classified with the SSRF validator's own ``is_ip_blocked`` (so CGNAT
230 and every other range the gate blocks fire the banner, and metadata /
231 always-blocked addresses — which the env remedies would not help — stay
232 silent); a hostname is classified with the ``is_private_ip`` known-name
233 check. A private-DNS hostname (``searx.internal`` → 10.x) and a
234 short-form literal (``127.1``) are therefore missed here — the runtime
235 gate still blocks them and logs the remediation; this advisory covers
236 the dominant localhost / LAN-literal case, mirroring the
237 cloud-embeddings banner's no-DNS tradeoff. Approval state is read via
238 the operator-approval resolver (blanket gate / origin allowlist /
239 env-locked URL), which reads only environment state. The displayed
240 remedy is derived from the SAME parser the allowlist uses
241 (``_engine_url_origin``), so the printed line round-trips by
242 construction; a URL that parser refuses (malformed port, forbidden
243 characters) yields no banner rather than an unusable remedy. Fails
244 toward NO banner on any error — advisory, never load-bearing.
245 """
246 if acknowledged or not active:
247 return None
248 if not instance_url or not isinstance(instance_url, str):
249 return None
250 try:
251 import ipaddress
253 from ..network_utils import is_private_ip
254 from ..ssrf_validator import is_ip_blocked
255 from .validators import (
256 _engine_url_origin,
257 engine_url_origin_text,
258 resolve_engine_allow_private_ips,
259 )
261 parsed_origin = _engine_url_origin(instance_url)
262 if parsed_origin is None:
263 return None
264 _scheme, host, _port = parsed_origin
266 try:
267 ipaddress.ip_address(host)
268 is_literal = True
269 except ValueError:
270 is_literal = False
271 if is_literal:
272 # Fire only for addresses the gate blocks SOLELY as private —
273 # metadata/always-blocked fall out of the loose call.
274 strict = is_ip_blocked(
275 host, allow_localhost=False, allow_private_ips=False
276 )
277 loose = is_ip_blocked(
278 host, allow_localhost=True, allow_private_ips=True
279 )
280 if not strict or loose:
281 return None
282 elif not is_private_ip(host):
283 return None
285 if resolve_engine_allow_private_ips(instance_url, url_setting):
286 return None
288 # Origin for the copy-paste remedy — rendered by the allowlist's own
289 # inverse formatter, so feeding it back into the allowlist is
290 # guaranteed to match; never the raw URL, so userinfo/path/query
291 # never appear.
292 origin = engine_url_origin_text(parsed_origin)
293 env_lock_var = "LDR_" + url_setting.upper().replace(".", "_")
294 anchor = "setting-" + url_setting.replace(".", "-")
295 except Exception: # noqa: silent-exception - advisory only
296 return None
298 return {
299 "type": f"{engine_name}_private_url_blocked",
300 "icon": "🚫",
301 "title": f"{display_name} is disabled: private URL not approved",
302 "message": (
303 f"{display_name} is enabled as a search engine, but its "
304 f"instance URL ({origin}) is a private/localhost address, "
305 "which is not fetched by default — research runs will return "
306 f"no {display_name} results. The server operator must approve "
307 "it in the server environment and restart LDR: set "
308 f"LDR_SEARCH_PRIVATE_ENGINE_URL_ALLOWLIST={origin} "
309 f"(recommended), or env-lock the URL via {env_lock_var}, or "
310 "LDR_SEARCH_ALLOW_PRIVATE_ENGINE_URLS=true to allow all "
311 "private URLs. Only one of these is needed."
312 ),
313 "dismissKey": f"app.warnings.dismiss_{engine_name}_private_url",
314 "actionUrl": f"/settings#{anchor}",
315 "actionLabel": f"Open {display_name} settings",
316 }
319def check_searxng_private_url_blocked(
320 primary_engine: str,
321 instance_url: str,
322 acknowledged: bool,
323) -> Optional[dict]:
324 """SearXNG-keyed convenience wrapper over
325 ``check_private_engine_url_blocked``. Test-facing only — production
326 emission goes through the orchestrator's descriptor loop; this wrapper
327 exists so focused tests can exercise the SearXNG parameterization
328 (active = SearXNG is the selected primary engine) without duplicating
329 the declaration constants."""
330 return check_private_engine_url_blocked(
331 "SearXNG",
332 "searxng",
333 "search.engine.web.searxng.default_params.instance_url",
334 instance_url,
335 (primary_engine or "").lower() == "searxng",
336 acknowledged,
337 )
340def _extract_engine_hosts(urls) -> list:
341 """Best-effort host extraction from an engine URL setting value.
343 Accepts a single URL string, a list of URL strings, or a JSON/comma
344 string of them (settings saved via the web UI may arrive as raw JSON
345 text — same class of input ``SearXNGSearchEngine._normalize_list``
346 handles). Entries that cannot be parsed are skipped: this feeds an
347 advisory banner, so failure means silence, never breakage.
348 """
349 import json as _json
350 from urllib.parse import urlsplit
352 if urls is None:
353 return []
354 if isinstance(urls, str):
355 stripped = urls.strip()
356 if not stripped:
357 return []
358 try:
359 parsed = _json.loads(stripped)
360 urls = parsed if isinstance(parsed, list) else [stripped]
361 except (ValueError, RecursionError):
362 urls = [u.strip() for u in stripped.split(",") if u.strip()]
363 if not isinstance(urls, list): 363 ↛ 364line 363 didn't jump to line 364 because the condition on line 363 was never true
364 return []
365 hosts = []
366 for entry in urls:
367 if not isinstance(entry, str) or not entry.strip():
368 continue
369 text = entry.strip()
370 try:
371 host = urlsplit(text).hostname
372 if not host and "://" not in text:
373 # Bare "host:9200" / "host" forms (Elasticsearch accepts
374 # them): urlsplit reads the host as the scheme, so re-parse
375 # as a network location.
376 host = urlsplit("//" + text).hostname
377 if host: 377 ↛ 366line 377 didn't jump to line 366 because the condition on line 377 was always true
378 hosts.append(host)
379 except Exception: # noqa: silent-exception - advisory only
380 continue
381 return hosts
384def check_local_engine_public_url(
385 display_name: str,
386 engine_name: str,
387 urls,
388 active: bool,
389 acknowledged: bool,
390 extra: str = "",
391 url_setting: str = "",
392) -> Optional[dict]:
393 """Banner when a LOCAL document engine (Paperless, Elasticsearch) points
394 at a host that does not look private.
396 The mirror image of the SearXNG private-URL banner: for a public engine
397 a private URL is the anomaly, for a local document store a PUBLIC one
398 is — search queries about the user's private documents are sent to that
399 host, and the PDP's fail-up reclassifies the engine as exposing, so a
400 Private-only egress scope silently excludes it from runs. Both
401 consequences are surprising, so surface them.
403 Advisory only, never a block: a genuinely-public endpoint can be a
404 deliberate choice (e.g. Elastic Cloud). Classification is cheap and
405 DNS-free: a host counts as local when the ``is_private_ip`` known-name
406 check says so, when it is a dotless single label (a Docker/compose
407 service name like ``paperless`` can never be public DNS), or when it
408 is an IP literal in a range the egress gate itself treats as private
409 (CGNAT and friends, via ``is_ip_blocked``). A multi-label private-DNS
410 hostname (``paperless.internal`` → 10.x) is still flagged even though
411 nothing would leave the box — the banner over-warns in the safe
412 direction and is dismissible. Fails toward silence on any error.
413 """
414 if acknowledged or not active:
415 return None
416 try:
417 import ipaddress
419 from ..network_utils import is_private_ip
420 from ..ssrf_validator import is_ip_blocked
422 def _looks_local(host: str) -> bool:
423 # IP literals FIRST: an IPv6 literal contains no dot, so the
424 # dotless-service-name shortcut below would otherwise swallow
425 # every IPv6 address, public ones included.
426 try:
427 ipaddress.ip_address(host)
428 except ValueError:
429 # Not an IP literal: known-private/localhost names, or a
430 # dotless single label (a Docker/compose service name can
431 # never be public DNS). Multi-label hostnames can't be
432 # verified without DNS and stay flagged.
433 return is_private_ip(host) or "." not in host
434 # IP literal: defer to the gate's own private classification
435 # (covers CGNAT etc. that is_private_ip doesn't know about).
436 return is_ip_blocked(
437 host, allow_localhost=False, allow_private_ips=False
438 ) and not is_ip_blocked(
439 host, allow_localhost=True, allow_private_ips=True
440 )
442 flagged = sorted(
443 {h for h in _extract_engine_hosts(urls) if not _looks_local(h)}
444 )
445 except Exception: # noqa: silent-exception - advisory only
446 return None
447 labels = flagged + ([extra] if extra else [])
448 if not labels:
449 return None
451 subject = ", ".join(labels)
452 action_url = (
453 f"/settings#setting-{url_setting.replace('.', '-')}"
454 if url_setting
455 else "/settings"
456 )
457 return {
458 "type": f"{engine_name}_public_url",
459 "icon": "📤",
460 "title": f"{display_name} points at a non-private host",
461 "message": (
462 f"Your {display_name} engine is configured for {subject}, "
463 "which does not look like a private/local address. Searches "
464 "send queries about your private documents to that host, and "
465 "under a Private-only egress scope the engine is excluded from "
466 "runs entirely. If this is a deliberate, trusted hosted "
467 "endpoint, dismiss this warning. (Multi-label private-DNS "
468 "hostnames cannot be verified without a lookup and may be "
469 "flagged incorrectly.)"
470 ),
471 "dismissKey": f"app.warnings.dismiss_{engine_name}_public_url",
472 "actionUrl": action_url,
473 "actionLabel": f"Review {display_name} settings",
474 }
477def check_cloud_llm_enabled(
478 provider: str,
479 require_local_endpoint: bool,
480 acknowledged: bool,
481) -> Optional[dict]:
482 """Banner when the configured LLM provider is cloud-only and the
483 require-local-endpoint toggle is off.
485 Critical-severity: the user's full prompt content (including the
486 research query and all retrieved context) is sent to the provider on
487 every call.
488 """
489 if acknowledged: 489 ↛ 490line 489 didn't jump to line 490 because the condition on line 489 was never true
490 return None
491 if require_local_endpoint:
492 return None
493 if not provider or provider.lower() not in _CLOUD_LLM_PROVIDERS:
494 return None
496 return {
497 "type": "cloud_llm_enabled",
498 "icon": "☁️",
499 "title": "LLM provider is cloud-hosted",
500 "message": (
501 f"Your LLM provider ({provider}) is cloud-hosted. Query "
502 "content will be sent off-machine on every research run, "
503 "independent of the Egress Scope setting. Tick 'Require local "
504 "LLM endpoint' below if you want fully local inference."
505 ),
506 "dismissKey": "app.warnings.dismiss_cloud_llm",
507 "actionUrl": "#llm_require_local_endpoint",
508 "actionLabel": "Require local LLM",
509 }
512def check_cloud_embeddings_enabled(
513 embeddings_provider: str,
514 embeddings_base_url: str,
515 require_local_embeddings: bool,
516 acknowledged: bool,
517) -> Optional[dict]:
518 """Banner when the embeddings provider sends data off-machine on
519 indexing.
521 Highest-severity of the three: indexing a private corpus with OpenAI
522 embeddings POSTs every chunk to OpenAI's API. A user who is unaware
523 of this loses their corpus.
525 Suppressed when ``base_url`` is set to a local URL (LM Studio,
526 vLLM, llama.cpp), since the OpenAI provider type is then pointed at
527 a local endpoint.
528 """
529 if acknowledged: 529 ↛ 530line 529 didn't jump to line 530 because the condition on line 529 was never true
530 return None
531 if require_local_embeddings:
532 return None
533 if (embeddings_provider or "").lower() != "openai":
534 return None
536 # If base_url is set and points to a local hostname, suppress the
537 # warning — the user has configured OpenAI-compatible-but-local
538 # (LM Studio, vLLM, etc.). Uses is_private_ip so RFC1918 ranges
539 # (10.x, 172.16-31.x, 192.168.x), CGNAT, link-local, IPv6 private,
540 # and .local mDNS hosts are all recognised. Substring-matching
541 # against a small list missed legitimate private-network endpoints.
542 #
543 # NOTE: this is a LITERAL-IP / .local check only — it does NOT resolve
544 # DNS, unlike the enforcing evaluate_embeddings (which DNS-resolves via
545 # _classify_host). So a private-DNS base_url (e.g. my-llm.internal → 10.x)
546 # is treated as LOCAL by enforcement (nothing leaves) yet still shows this
547 # banner. That divergence is intentional: this advisory runs on every
548 # settings-page render and must stay synchronous/cheap, so it deliberately
549 # over-warns (the safe direction) rather than add a blocking DNS lookup to
550 # page load. The banner never blocks anything; the PDP is the source of
551 # truth for what actually egresses.
552 if embeddings_base_url:
553 try:
554 from urllib.parse import urlsplit
556 from ..network_utils import is_private_ip
558 parsed = urlsplit(embeddings_base_url)
559 hostname = parsed.hostname
560 if hostname and is_private_ip(hostname):
561 return None
562 except Exception: # noqa: silent-exception
563 # Defensive: if URL parsing fails, fall through to issuing
564 # the banner — the failure itself is a sign of misconfig
565 # the user should see, not a reason to mask the warning.
566 pass
568 return {
569 "type": "cloud_embeddings_enabled",
570 "icon": "📤",
571 "title": "Document chunks will be sent to OpenAI",
572 "message": (
573 "Your embeddings provider is OpenAI. Indexing a collection "
574 "will POST every chunk to OpenAI's API — the entire corpus "
575 "leaves the machine. Tick 'Require local embeddings' below "
576 "to switch to sentence-transformers or a local OpenAI-"
577 "compatible endpoint."
578 ),
579 "dismissKey": "app.warnings.dismiss_cloud_embeddings",
580 "actionUrl": "#embeddings_require_local",
581 "actionLabel": "Require local embeddings",
582 }