Coverage for src/local_deep_research/security/egress/validators.py: 90%
306 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""Cross-field egress-policy validators run at settings-save time.
3Each validator takes the about-to-be-saved ``form_data`` plus the current
4``all_db_settings`` (key -> setting row with a ``.value``) and returns a
5validation-error dict ``{"key", "error"}`` to surface in the save response,
6or ``None`` when the combination is fine. Most are pure; the engine-URL
7guard machinery here (allowlist, operator-approval resolver, descriptor
8sweep) additionally reads operator environment state — never the DB — and
9the allowlist parser logs dropped entries once per distinct env value.
11The settings write routes (``web/routers/settings.py``) orchestrate these
12— they stay in the route layer because they also enforce non-egress concerns —
13but the egress rules themselves live here next to the policy they encode.
14"""
16import functools
17import ipaddress
18from dataclasses import dataclass
19from typing import Optional, Tuple
20from urllib.parse import unquote, urlsplit
22from ...utilities.type_utils import unwrap_setting
23from .policy import (
24 EgressContext,
25 EgressScope,
26 PolicyDeniedError,
27 _classify_host,
28 _resolve_with_timeout,
29 parse_user_egress_scope,
30)
32# The SearXNG instance URL is the canonical exploitable case and is always
33# guarded even if dynamic engine-class enumeration fails below.
34_SEARXNG_INSTANCE_URL_KEY = (
35 "search.engine.web.searxng.default_params.instance_url"
36)
39def validate_egress_scope(form_data, _all_db_settings):
40 """Validate the scope and enforce the operator-controlled escape hatch."""
41 key = "policy.egress_scope"
42 if key not in form_data:
43 return None
44 value = form_data[key]
45 if not isinstance(value, str): 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true
46 return {"key": key, "error": "Egress scope must be a string"}
47 try:
48 parse_user_egress_scope(value)
49 except PolicyDeniedError as exc:
50 if exc.decision.reason == "unprotected_egress_disabled":
51 return {
52 "key": key,
53 "error": "Unprotected egress is disabled by the server operator",
54 }
55 return {
56 "key": key,
57 "error": "Unknown egress scope",
58 }
59 return None
62def validate_allowed_local_hostnames(form_data, all_db_settings):
63 """Reject public hostnames being added to llm.allowed_local_hostnames.
65 The default-settings description for this key claims "Public hostnames
66 added here are rejected at save time", but until now no code actually
67 did the rejection. This guard resolves each entry via the same host
68 classifier the policy uses, and refuses any that resolve to public
69 addresses. A hostname that fails to resolve (DNS down) is accepted —
70 fail-open on transient lookup errors so the user can recover.
71 """
72 key = "llm.allowed_local_hostnames"
73 if key not in form_data:
74 return None
75 value = form_data[key]
76 # Setting is JSON-typed; the save pipeline may hand us a list or a
77 # JSON string. Decode defensively.
78 if isinstance(value, str):
79 try:
80 import json as _json
82 decoded = _json.loads(value) if value.strip() else []
83 except Exception:
84 return {
85 "key": key,
86 "error": "allowed_local_hostnames must be a JSON array of hostnames",
87 }
88 value = decoded
89 if not isinstance(value, list):
90 return {
91 "key": key,
92 "error": "allowed_local_hostnames must be a list",
93 }
95 # Build a minimal real context just for the resolver. Use the
96 # dataclass constructor — NOT EgressContext.__new__ + setattr, which
97 # raised FrozenInstanceError on this frozen dataclass and (separately)
98 # set a non-existent ``allowed_local_hostnames`` field instead of the
99 # real ``local_hostnames`` the classifier reads. The constructor
100 # initializes the init=False internals (_dns_cache, _lock as RLock)
101 # correctly. Empty local_hostnames => classify purely on IP class.
102 probe_ctx = EgressContext(
103 scope=EgressScope.BOTH,
104 primary_engine="searxng",
105 require_local_llm=False,
106 require_local_embeddings=False,
107 local_hostnames=(),
108 )
110 rejected = []
111 for entry in value:
112 if not isinstance(entry, str) or not entry.strip():
113 continue
114 hostname = entry.strip().lower()
115 try:
116 # Distinguish "could not resolve" from "resolved to a public IP".
117 # _classify_host collapses BOTH to False (its documented fail-safe
118 # treats an unresolvable host as public), so relying on it here
119 # would reject a legitimate intranet/VPN host on any DNS hiccup or
120 # split-horizon DNS — the exact use case this setting exists for.
121 # Only reject names that actually resolve to a public address;
122 # accept unresolvable ones (fail-open on save, as documented).
123 # _resolve_with_timeout returns the addrinfo for literal IPs too,
124 # so literal public/private IPs still flow through _classify_host.
125 if _resolve_with_timeout(hostname) is None:
126 continue
127 classification = _classify_host(hostname, probe_ctx)
128 except Exception:
129 # DNS or unknown error — allow (fail open) so the user can
130 # save when networking is flaky. Runtime classification will
131 # still gate egress.
132 continue
133 if classification is False:
134 rejected.append(hostname)
135 if rejected:
136 return {
137 "key": key,
138 "error": (
139 "These hostnames resolve to PUBLIC addresses and would "
140 "let the policy treat external hosts as local: "
141 f"{', '.join(rejected)}. Remove them, or use the SSRF "
142 "allowlist instead."
143 ),
144 }
145 return None
148def validate_trusted_search_engines(form_data, all_db_settings):
149 """Reject inherently-public engines from ``policy.trusted_search_engines``.
151 Trusting a search engine relaxes its Exposure to CONTAINED — meaningful only
152 for a local-nature store you host (Elasticsearch/Paperless). An inherently
153 public engine (google, searxng, brave, …) genuinely queries the internet, so
154 trusting it would launder a public sink past the two-axis rule. Reject those;
155 local / unknown names are accepted (runtime classification still gates them,
156 and ``engine_label`` also refuses to relax an ``is_public`` engine).
157 """
158 key = "policy.trusted_search_engines"
159 if key not in form_data:
160 return None
161 value = form_data[key]
162 if isinstance(value, str):
163 import json
165 try:
166 value = json.loads(value) if value.strip() else []
167 except Exception:
168 return {
169 "key": key,
170 "error": "trusted_search_engines must be a JSON array",
171 }
172 if not isinstance(value, list): 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true
173 return {"key": key, "error": "trusted_search_engines must be a list"}
175 from .policy import _get_engine_class
177 rejected = []
178 for entry in value:
179 if not isinstance(entry, str) or not entry.strip(): 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true
180 continue
181 name = entry.strip().lower()
182 cls = _get_engine_class(name)
183 if cls is not None and getattr(cls, "is_public", False) is True:
184 rejected.append(name)
185 if rejected:
186 return {
187 "key": key,
188 "error": (
189 "These are inherently-public search engines and cannot be "
190 "trusted as contained: " + ", ".join(rejected) + "."
191 ),
192 }
193 return None
196@dataclass(frozen=True)
197class GuardedEngineUrl:
198 """A search engine whose configurable URL participates in the egress
199 URL guards, derived from the engine class's own declarations.
201 ``is_public`` True → the engine proxies the public internet, so a
202 PRIVATE URL is the anomaly (save-time rejection + runtime refusal +
203 "private URL not approved" banner). ``is_public`` False → a LOCAL
204 document engine, where a PUBLIC-looking URL is the anomaly (advisory
205 banner only — a hosted endpoint can be legitimate).
207 Per-engine UI conventions derived from ``engine_name``:
208 display name setting ``search.engine.web.{name}.display_name``,
209 activation flags ``.enabled`` / ``.use_in_auto_search`` /
210 ``.agent_enabled``, and dismiss keys
211 ``app.warnings.dismiss_{name}_private_url`` (public) /
212 ``app.warnings.dismiss_{name}_public_url`` (local) — each dismiss key
213 must be registered in ``defaults/default_settings.json`` (pinned by a
214 registry-contract test). A future public engine that declares
215 ``url_setting`` must also call ``resolve_engine_allow_private_ips``
216 on its fetch path, as SearXNG does.
217 """
219 engine_name: str
220 url_setting: str
221 is_public: bool
224@functools.lru_cache(maxsize=1)
225def guarded_engine_url_descriptors() -> Tuple[GuardedEngineUrl, ...]:
226 """All engines participating in the URL guards, from the registry.
228 Derived once from ``ENGINE_REGISTRY`` + engine-class declarations
229 (``url_setting`` plus ``is_public``/``is_local``); seeded with SearXNG
230 so the canonical guard survives an engine-class import failure.
232 COST NOTE: the first call imports every engine module in the registry
233 (~1-3 s cold) to read the class declarations; results are cached for
234 the process lifetime (``.cache_clear()`` resets, for tests). Callers on
235 render paths pay this once per process.
236 """
237 found = {
238 "searxng": GuardedEngineUrl(
239 engine_name="searxng",
240 url_setting=_SEARXNG_INSTANCE_URL_KEY,
241 is_public=True,
242 )
243 }
244 try:
245 from ...web_search_engines.engine_registry import ENGINE_REGISTRY
246 from .policy import _get_engine_class
248 for name in ENGINE_REGISTRY:
249 cls = _get_engine_class(name)
250 if cls is None:
251 continue
252 url_setting = getattr(cls, "url_setting", None)
253 if not (isinstance(url_setting, str) and url_setting):
254 continue
255 if getattr(cls, "is_public", None) is True:
256 found[name] = GuardedEngineUrl(
257 engine_name=name, url_setting=url_setting, is_public=True
258 )
259 elif getattr(cls, "is_local", None) is True: 259 ↛ 248line 259 didn't jump to line 248 because the condition on line 259 was always true
260 found[name] = GuardedEngineUrl(
261 engine_name=name, url_setting=url_setting, is_public=False
262 )
263 except Exception: # noqa: silent-exception - seed still guards SearXNG
264 pass
265 return tuple(found[name] for name in sorted(found))
268@functools.lru_cache(maxsize=1)
269def _public_engine_url_settings() -> frozenset:
270 """URL-setting keys belonging to PUBLIC-nature search engines.
272 These are the engine URLs a save-time SSRF check must guard: a public
273 engine (SearXNG) proxies the public web, so a private/loopback
274 ``instance_url`` is only ever an internal-network probe target, never a
275 legitimate data source. LOCAL engines (Elasticsearch, Paperless) are
276 deliberately EXCLUDED — a private URL is their whole purpose, and the PDP's
277 fail-up URL override already reclassifies them public when pointed at an
278 external host.
280 Delegates to ``guarded_engine_url_descriptors`` so the save-time guard
281 and the banner/orchestrator machinery can never disagree about which
282 engines are covered (single registry sweep, single seed).
283 """
284 return frozenset(
285 d.url_setting for d in guarded_engine_url_descriptors() if d.is_public
286 )
289# Env-only operator allowlist of exact private engine-URL origins — the
290# finer-grained alternative to the blanket search.allow_private_engine_urls
291# gate (see env_definitions/security.py for the operator-facing contract).
292_ALLOWLIST_SETTING_KEY = "search.private_engine_url_allowlist"
294_DEFAULT_PORTS = {"http": 80, "https": 443}
297def _engine_url_origin(value):
298 """Parse ``value`` into a comparable ``(scheme, host, port)`` origin.
300 Normalization mirrors what actually denotes the same network
301 destination and nothing more: scheme and host are lowercased, the host
302 is stripped of a trailing dot, an IP-literal host is canonicalized
303 through :mod:`ipaddress` (so ``[::1]`` and its long form compare
304 equal), and a missing port takes the scheme default (80/443). Path,
305 query, fragment, and userinfo are ignored — they don't change the
306 destination host.
308 Two deliberate refusals keep this parser from ever disagreeing with
309 the HTTP client (urllib3) about which host a hostile URL names:
310 URLs containing RFC-forbidden characters (backslash, whitespace,
311 control chars — the parser-differential class) return ``None``, and
312 the host is NOT percent-decoded here (urllib3 before 2.8 did not
313 decode http(s) hosts and 2.8+ does, so this parser stays decode-free:
314 an escaped host matches only an allowlist entry written with the same
315 escapes, never a plain entry by decoding, and ``validate_url`` refuses
316 percent-encoded hosts at fetch time). Returns ``None`` for anything
317 that isn't a cleanly parseable
318 http(s) URL with a hostname; every failure mode therefore fails
319 CLOSED (no allowlist match), never open.
320 """
321 try:
322 text = str(value).strip()
323 from ..ssrf_validator import RFC_FORBIDDEN_URL_CHARS_RE
325 if RFC_FORBIDDEN_URL_CHARS_RE.search(text):
326 return None
327 parsed = urlsplit(text)
328 scheme = (parsed.scheme or "").lower()
329 if scheme not in _DEFAULT_PORTS:
330 return None
331 host = parsed.hostname
332 if not host: 332 ↛ 333line 332 didn't jump to line 333 because the condition on line 332 was never true
333 return None
334 host = host.rstrip(".").lower()
335 if not host: 335 ↛ 336line 335 didn't jump to line 336 because the condition on line 335 was never true
336 return None
337 try:
338 host = str(ipaddress.ip_address(host))
339 except ValueError:
340 pass # not an IP literal — keep the hostname text
341 port = parsed.port
342 if port is None:
343 port = _DEFAULT_PORTS[scheme]
344 return (scheme, host, port)
345 except Exception:
346 return None
349def engine_url_origin_text(origin) -> str:
350 """Render an ``_engine_url_origin`` tuple back to allowlist-entry text.
352 The inverse of ``_engine_url_origin`` for display: IPv6 hosts are
353 re-bracketed and a scheme-default port is omitted, so the returned
354 string, fed into ``LDR_SEARCH_PRIVATE_ENGINE_URL_ALLOWLIST``, is
355 guaranteed to match the origin it was rendered from (round-trip by
356 construction — pinned by the banner remedy tests). Lives next to the
357 parser so the pair can't drift apart.
358 """
359 scheme, host, port = origin
360 host_disp = f"[{host}]" if ":" in host else host
361 suffix = "" if port == _DEFAULT_PORTS[scheme] else f":{port}"
362 return f"{scheme}://{host_disp}{suffix}"
365# Raw allowlist value we last warned about — the dropped-entry WARNING is
366# emitted once per distinct env value, not on every consult (the matcher
367# runs on every settings save, engine init, and /api/warnings recompute).
368_last_warned_allowlist_raw: Optional[str] = None
371def _private_engine_url_allowlist() -> frozenset:
372 """Operator-approved private engine-URL origins from the environment.
374 Reads the env-only ``search.private_engine_url_allowlist`` setting and
375 parses each comma-separated entry with ``_engine_url_origin``.
376 Unparseable entries are dropped (fail-closed) and warned about once per
377 distinct env value. Any lookup error yields the empty set — the safe
378 posture.
379 """
380 try:
381 from ...settings.env_registry import get_env_setting
383 raw = get_env_setting(_ALLOWLIST_SETTING_KEY, None)
384 except Exception: # noqa: silent-exception - default to the safe posture
385 return frozenset()
386 if not raw or not isinstance(raw, str):
387 return frozenset()
388 origins = set()
389 dropped = []
390 for entry in raw.split(","):
391 entry = entry.strip()
392 if not entry:
393 continue
394 origin = _engine_url_origin(entry)
395 if origin is not None:
396 origins.add(origin)
397 else:
398 dropped.append(entry)
399 if dropped:
400 # An unparseable entry grants nothing (fail-closed), but silently
401 # dropping it leaves the operator staring at the same rejection
402 # message telling them to add the origin they think they already
403 # added — so name what was ignored (once per distinct env value,
404 # to keep a permanently-malformed entry from spamming the logs).
405 global _last_warned_allowlist_raw
406 if raw != _last_warned_allowlist_raw:
407 _last_warned_allowlist_raw = raw
408 try:
409 from ..secure_logging import logger
411 logger.warning(
412 "Ignoring unparseable entries in "
413 "LDR_SEARCH_PRIVATE_ENGINE_URL_ALLOWLIST (each entry "
414 "must be a full origin like http://localhost:8080): "
415 f"{dropped}"
416 )
417 except Exception: # noqa: silent-exception - best-effort logging
418 pass
419 return frozenset(origins)
422def engine_url_in_private_allowlist(url) -> bool:
423 """True when ``url``'s origin is in the operator's env-only allowlist.
425 Grants ONLY the private/loopback/link-local exemption a matching origin
426 was listed for — callers consult it strictly in place of the blanket
427 ``search.allow_private_engine_urls`` gate, so the cloud-metadata block
428 and the http(s)-scheme requirement are unaffected by a listing.
429 """
430 origin = _engine_url_origin(url)
431 if origin is None:
432 return False
433 return origin in _private_engine_url_allowlist()
436def resolve_engine_allow_private_ips(
437 instance_url: str, url_setting: str
438) -> bool:
439 """Whether a PUBLIC engine may target a private/loopback instance URL.
441 A public-nature engine proxies the public web, so a private instance
442 URL is only legitimate when an OPERATOR chose it, never when an
443 arbitrary authenticated user smuggled an internal host in as an SSRF
444 probe. Private egress to the configured instance is therefore
445 permitted only when:
447 1. the operator opt-in ``search.allow_private_engine_urls`` is set
448 (env-only; the blanket LAN / localhost self-host case), OR
449 2. ``instance_url``'s exact origin is listed in the env-only operator
450 allowlist ``search.private_engine_url_allowlist`` (the
451 finer-grained alternative to the blanket opt-in), OR
452 3. the engine's own ``url_setting`` is env-locked (its derived
453 ``LDR_…`` variable is set — Docker / operator-provisioned,
454 trusted).
456 All three are OPERATOR-controlled. The run's egress scope
457 (``policy.egress_scope``) deliberately does NOT grant it: that setting is a
458 user-editable dropdown (STRICT included), so relaxing on it would be a
459 self-service SSRF bypass. By default private IPs are refused, so a
460 tampered instance URL cannot reach an internal host. Cloud-metadata /
461 link-local metadata IPs stay blocked regardless
462 (ALWAYS_BLOCKED_METADATA_IPS in the SSRF validator's probe path). Fails
463 CLOSED (False) on any error: a genuinely-public instance is unaffected —
464 only private egress is withheld.
465 """
466 # 1. Operator opt-in (LAN self-host).
467 try:
468 from ...settings.env_registry import get_env_setting
470 if bool(get_env_setting("search.allow_private_engine_urls", False)):
471 return True
472 except Exception: # noqa: silent-exception - fall through to next check
473 pass
475 # 2. Operator allowlist: this exact origin was individually approved.
476 try:
477 if engine_url_in_private_allowlist(instance_url):
478 return True
479 except Exception: # noqa: silent-exception - fall through to next check
480 pass
482 # 3. Env-locked instance URL (Docker / operator-provisioned).
483 try:
484 from ...settings.manager import check_env_setting
486 if check_env_setting(url_setting) is not None:
487 return True
488 except Exception: # noqa: silent-exception - fall through to next check
489 pass
491 # Otherwise: withhold private egress. We deliberately DO NOT relax this
492 # based on the run's resolved egress scope. ``policy.egress_scope`` is a
493 # user-editable, self-service setting (STRICT is a plain dropdown option,
494 # not operator-gated), and a public engine always queries the internet.
495 # Letting a user-chosen scope grant private-IP access would be a
496 # self-service SSRF bypass (set scope=strict, then point the instance
497 # URL at an internal host). Private egress for a public engine is an
498 # OPERATOR decision only — conditions 1–3 above.
499 return False
502def resolve_searxng_allow_private_ips(instance_url: str) -> bool:
503 """SearXNG-keyed convenience wrapper over
504 ``resolve_engine_allow_private_ips`` (kept for the engine and tests)."""
505 return resolve_engine_allow_private_ips(
506 instance_url, _SEARXNG_INSTANCE_URL_KEY
507 )
510def resolve_engine_allow_private_result_fetch(
511 instance_url: str, url_setting: str
512) -> bool:
513 """Whether full-content fetches of a PUBLIC engine's RESULT URLs may
514 reach private / loopback hosts.
516 Distinct from, and strictly narrower than,
517 ``resolve_engine_allow_private_ips``: that grant answers "may this
518 engine reach its configured *instance* URL", and its env-locked-URL
519 condition is met by every Docker (compose) deployment. A public engine
520 proxies the public web, so extending the instance grant to the URLs
521 the engine *returns* would let a poisoned index turn result fetching
522 into SSRF against the operator's LAN (loopback included). This
523 returns True (the engine holds the grant) only when BOTH hold:
525 1. the operator set the env-only opt-in
526 ``search.allow_private_result_fetch``
527 (``LDR_SEARCH_ALLOW_PRIVATE_RESULT_FETCH=true``), which the
528 bundled ``docker-compose.yml`` does not set, AND
529 2. the engine is approved for its own instance URL
530 (``resolve_engine_allow_private_ips``).
532 Holding the grant is necessary but not sufficient: the full-content
533 wrapper applies it only when the run's egress scope is
534 ``unprotected`` (``FullSearchResults._effective_allow_private_ips``).
535 Under every other scope, the default adaptive scope included, private
536 result URLs stay blocked even when this returns True.
538 Link-local (``169.254.0.0/16``, ``fe80::/10``) and the cloud-metadata
539 literals in ``ALWAYS_BLOCKED_METADATA_IPS`` stay blocked even when
540 this returns True: the full-content fetch path validates with
541 ``block_link_local=True``. Fails CLOSED (False) on any error.
542 """
543 try:
544 from ...settings.env_registry import get_env_setting
546 if not bool(
547 get_env_setting("search.allow_private_result_fetch", False)
548 ):
549 return False
550 except Exception: # noqa: silent-exception - fail closed
551 return False
552 return resolve_engine_allow_private_ips(instance_url, url_setting)
555def resolve_searxng_allow_private_result_fetch(instance_url: str) -> bool:
556 """SearXNG-keyed convenience wrapper over
557 ``resolve_engine_allow_private_result_fetch``."""
558 return resolve_engine_allow_private_result_fetch(
559 instance_url, _SEARXNG_INSTANCE_URL_KEY
560 )
563def _engine_url_ssrf_error(key, value, *, allow_private: bool):
564 """Return an error dict for a public-engine URL that must be refused.
566 Rejects:
567 - a non-http(s) scheme (never a valid engine URL — refused even when the
568 operator opt-in is set),
569 - a host that IS or RESOLVES TO a cloud-metadata / always-blocked
570 address (refused regardless of the operator opt-in), and
571 - unless ``allow_private`` is set, a host that IS or RESOLVES TO a
572 private / loopback / link-local address.
574 Reuses ``ssrf_validator.is_ip_blocked`` for the classification: the
575 ``allow_localhost=True, allow_private_ips=True`` form isolates the
576 always-blocked cloud-metadata set, so metadata stays refused even under the
577 operator opt-in. Returns ``None`` when the value is not a URL string (type
578 validation handles that elsewhere).
580 Fail-open on transient resolution failure (DNS down / split-horizon): only
581 names that actually resolve to a blocked address are rejected, so a
582 legitimate public host stays saveable during a DNS hiccup — mirroring
583 ``validate_allowed_local_hostnames``.
584 """
585 if not isinstance(value, str) or not value.strip(): 585 ↛ 586line 585 didn't jump to line 586 because the condition on line 585 was never true
586 return None
587 url = value.strip()
589 from ..ssrf_validator import RFC_FORBIDDEN_URL_CHARS_RE, is_ip_blocked
591 if RFC_FORBIDDEN_URL_CHARS_RE.search(url):
592 return {
593 "key": key,
594 "error": f"{key} contains illegal URL characters",
595 }
596 try:
597 parsed = urlsplit(url)
598 except Exception:
599 return {"key": key, "error": f"{key} is not a valid URL"}
601 if (parsed.scheme or "").lower() not in ("http", "https"):
602 return {
603 "key": key,
604 "error": f"{key} must be an http:// or https:// URL",
605 }
607 host = parsed.hostname
608 if not host: 608 ↛ 609line 608 didn't jump to line 609 because the condition on line 608 was never true
609 return {"key": key, "error": f"{key} has no hostname"}
610 host = unquote(host)
611 if host.startswith("[") and host.endswith("]"): 611 ↛ 612line 611 didn't jump to line 612 because the condition on line 611 was never true
612 host = host[1:-1]
613 host = host.rstrip(".")
615 def _block_reason(ip_str) -> Optional[str]:
616 # Cloud-metadata / always-blocked set: refused regardless of opt-in.
617 if is_ip_blocked(ip_str, allow_localhost=True, allow_private_ips=True):
618 return "metadata"
619 # Private / loopback / link-local: refused unless operator opted in.
620 if not allow_private and is_ip_blocked(
621 ip_str, allow_localhost=False, allow_private_ips=False
622 ):
623 return "private"
624 return None
626 try:
627 ipaddress.ip_address(host)
628 is_literal_ip = True
629 except ValueError:
630 is_literal_ip = False
632 if is_literal_ip:
633 reason = _block_reason(host)
634 return _engine_url_error(key, reason) if reason else None
636 addr_info = _resolve_with_timeout(host)
637 if not addr_info:
638 return None # fail-open on unresolvable host
639 for entry in addr_info:
640 try:
641 ip_str = entry[4][0]
642 except (IndexError, TypeError):
643 continue
644 reason = _block_reason(ip_str)
645 if reason:
646 return _engine_url_error(key, reason)
647 return None
650def _engine_url_error(key, reason):
651 if reason == "metadata":
652 message = (
653 f"{key} points at a cloud-metadata / always-blocked address, "
654 "which is never a permitted destination."
655 )
656 else:
657 message = (
658 f"{key} points at a private, loopback, or link-local address. A "
659 "public search engine proxies the internet, so an internal URL is "
660 "refused to prevent it being used to reach your private network. "
661 "Self-hosted instances on localhost/LAN are still supported, but "
662 "the server operator must approve them in the server "
663 "environment: add this exact URL origin to "
664 "LDR_SEARCH_PRIVATE_ENGINE_URL_ALLOWLIST (comma-separated "
665 "scheme://host:port entries), or pin the URL via its LDR_ "
666 "environment variable (as the bundled docker-compose.yml does), "
667 "or set LDR_SEARCH_ALLOW_PRIVATE_ENGINE_URLS=true to allow all "
668 "private addresses. Only one of these is needed; restart after "
669 "changing. See docs/SearXNG-Setup.md."
670 )
671 return {"key": key, "error": message}
674# Sentinel marking a guarded key that has no stored value yet — treated as a
675# genuine change so a brand-new setting is still validated.
676_NO_STORED_VALUE = object()
679def _stored_engine_url_value(all_db_settings, key):
680 """Return the currently-stored value for ``key`` in ``all_db_settings``.
682 Entries are ``Setting`` ORM rows exposing the value via ``.value`` in the
683 live save routes (``_filter_editable_settings`` builds ``{key: Setting}``),
684 but some callers hand plain ``{"value": ...}`` dicts. Normalize both to the
685 scalar with ``unwrap_setting`` — the dict form directly, an ORM/attribute
686 row via its ``.value``. Returns ``_NO_STORED_VALUE`` when the key is absent.
687 """
688 if not all_db_settings or key not in all_db_settings:
689 return _NO_STORED_VALUE
690 entry = all_db_settings[key]
691 if not isinstance(entry, dict):
692 entry = getattr(entry, "value", _NO_STORED_VALUE)
693 if entry is _NO_STORED_VALUE: 693 ↛ 694line 693 didn't jump to line 694 because the condition on line 693 was never true
694 return _NO_STORED_VALUE
695 return unwrap_setting(entry)
698def _normalize_engine_url_for_comparison(value):
699 """Parse ``value`` into a tuple of URL parts for cosmetic-variant
700 comparison, or ``None`` if it cannot be parsed as a URL.
702 Collapses only variation that denotes the exact SAME url:
703 - scheme and host case (``HTTP://LOCALHOST`` == ``http://localhost``),
704 - any trailing slash(es) on the path (``rstrip("/")`` strips all of
705 them, so ``"/path//"`` and ``"/path"`` compare equal, and an
706 empty path and ``"/"`` compare equal too).
708 Everything that could denote a DIFFERENT destination is preserved
709 verbatim and NOT normalized away: the host itself, the port, any path
710 beyond a bare trailing slash, the query, the fragment, and any userinfo
711 (unusual in an ``instance_url``, but kept rather than risk folding two
712 distinct credentials together). In particular this never strips the
713 host, resolves DNS, or drops the port — so a genuine change to a
714 different host/port/scheme still fails to compare equal here.
716 Returns ``None`` on any parse failure (e.g. a non-numeric port, a
717 malformed IPv6 literal) so the caller falls back to the exact
718 string comparison — fail toward validating, never toward skipping.
719 """
720 try:
721 parsed = urlsplit(str(value).strip())
722 return (
723 (parsed.scheme or "").lower(),
724 parsed.username,
725 parsed.password,
726 (parsed.hostname or "").lower(),
727 parsed.port,
728 (parsed.path or "").rstrip("/"),
729 parsed.query,
730 parsed.fragment,
731 )
732 except Exception:
733 return None
736def _engine_url_is_unchanged(submitted_value, all_db_settings, key):
737 """True when the submitted engine URL equals the currently-stored one.
739 The frontend "All Settings" tab resubmits every rendered input, including a
740 shipped default the user never touched (e.g. the SearXNG
741 ``instance_url = http://localhost:8080`` loopback default). Validating an
742 unchanged value would 400 an otherwise-unrelated save, so only a genuine
743 change is validated. A brand-new key with no stored value is treated as
744 changed.
746 Both sides are first string-normalized so a metadata-wrapped vs scalar
747 representation of the same value is not misread as a change. If that
748 exact match fails, both sides are also compared via
749 ``_normalize_engine_url_for_comparison`` so a purely cosmetic resubmit —
750 different scheme/host case, or a trailing slash — of the SAME url is
751 still recognized as unchanged (e.g. resubmitting the stored
752 ``http://localhost:8080`` as ``HTTP://LOCALHOST:8080`` or
753 ``http://localhost:8080/``). This only ever widens what counts as
754 UNCHANGED (and is therefore skipped from validation); it never narrows
755 it, so it cannot make two genuinely different hosts compare equal. If
756 either side fails to parse, that comparison is skipped and the exact
757 match above is the only thing that can call it unchanged.
758 """
759 stored = _stored_engine_url_value(all_db_settings, key)
760 if stored is _NO_STORED_VALUE:
761 return False
762 submitted_str = str(submitted_value).strip()
763 stored_str = str(stored).strip()
764 if submitted_str == stored_str:
765 return True
766 submitted_norm = _normalize_engine_url_for_comparison(submitted_value)
767 stored_norm = _normalize_engine_url_for_comparison(stored)
768 if submitted_norm is None or stored_norm is None:
769 return False
770 return submitted_norm == stored_norm
773def validate_engine_instance_urls(form_data, all_db_settings):
774 """Reject a PUBLIC search-engine URL that targets a private address.
776 SearXNG (and any future public-nature engine with a configurable URL) is
777 user-editable through the settings API. Because such an engine proxies the
778 public web, a private / loopback / link-local ``instance_url`` is only ever
779 an internal-network probe target — so at save time we refuse it under the
780 default posture, closing the SSRF where any authenticated user turns a
781 research run into an internal port scan. Non-http(s) schemes are refused
782 outright.
784 The operator opt-in ``search.allow_private_engine_urls`` (env-only) lifts
785 the private-address rejection for genuine LAN self-hosting, and the
786 env-only allowlist ``search.private_engine_url_allowlist`` lifts it for
787 specific listed origins only; Docker deployments instead env-lock the
788 URL, which the route layer honors via ``check_env_setting`` before these
789 validators run. Neither mechanism lifts the cloud-metadata or
790 non-http(s)-scheme rejections.
791 """
792 guarded = _public_engine_url_settings()
793 present = [k for k in guarded if k in form_data]
794 if not present:
795 return None
797 allow_private = False
798 try:
799 from ...settings.env_registry import get_env_setting
801 allow_private = bool(
802 get_env_setting("search.allow_private_engine_urls", False)
803 )
804 except Exception: # noqa: silent-exception - default to the safe posture
805 allow_private = False
807 for key in present:
808 value = unwrap_setting(form_data[key])
809 # Validate-on-change: skip an UNCHANGED value (e.g. the shipped default
810 # loopback ``instance_url`` resubmitted untouched by the All-Settings
811 # tab) so it can't block an unrelated save. The runtime
812 # allow_private_ips=False backstop still blocks the actual fetch.
813 if _engine_url_is_unchanged(value, all_db_settings, key):
814 continue
815 err = _engine_url_ssrf_error(
816 key,
817 value,
818 allow_private=allow_private
819 or engine_url_in_private_allowlist(value),
820 )
821 if err is not None:
822 return err
823 return None
826# The cross-field egress validators every settings-write route must run. Keep
827# the *list* here (single source of truth) so adding one doesn't require a
828# lockstep edit at each save route; each route only supplies its own response
829# shaping around ``first_egress_validation_error``.
830EGRESS_SETTINGS_VALIDATORS = (
831 validate_egress_scope,
832 validate_allowed_local_hostnames,
833 validate_trusted_search_engines,
834 validate_engine_instance_urls,
835)
838def first_egress_validation_error(form_data, all_db_settings):
839 """Run the egress validators; return the first error dict, or ``None``."""
840 for _validator in EGRESS_SETTINGS_VALIDATORS:
841 err = _validator(form_data, all_db_settings)
842 if err is not None:
843 return err
844 return None