Coverage for src/local_deep_research/security/egress/validators.py: 90%
296 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +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: 250 ↛ 251line 250 didn't jump to line 251 because the condition on line 250 was never true
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 (urllib3 doesn't decode either, so
313 decoding here would match a destination the client never connects
314 to). Returns ``None`` for anything that isn't a cleanly parseable
315 http(s) URL with a hostname; every failure mode therefore fails
316 CLOSED (no allowlist match), never open.
317 """
318 try:
319 text = str(value).strip()
320 from ..ssrf_validator import RFC_FORBIDDEN_URL_CHARS_RE
322 if RFC_FORBIDDEN_URL_CHARS_RE.search(text):
323 return None
324 parsed = urlsplit(text)
325 scheme = (parsed.scheme or "").lower()
326 if scheme not in _DEFAULT_PORTS:
327 return None
328 host = parsed.hostname
329 if not host: 329 ↛ 330line 329 didn't jump to line 330 because the condition on line 329 was never true
330 return None
331 host = host.rstrip(".").lower()
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 try:
335 host = str(ipaddress.ip_address(host))
336 except ValueError:
337 pass # not an IP literal — keep the hostname text
338 port = parsed.port
339 if port is None:
340 port = _DEFAULT_PORTS[scheme]
341 return (scheme, host, port)
342 except Exception:
343 return None
346def engine_url_origin_text(origin) -> str:
347 """Render an ``_engine_url_origin`` tuple back to allowlist-entry text.
349 The inverse of ``_engine_url_origin`` for display: IPv6 hosts are
350 re-bracketed and a scheme-default port is omitted, so the returned
351 string, fed into ``LDR_SEARCH_PRIVATE_ENGINE_URL_ALLOWLIST``, is
352 guaranteed to match the origin it was rendered from (round-trip by
353 construction — pinned by the banner remedy tests). Lives next to the
354 parser so the pair can't drift apart.
355 """
356 scheme, host, port = origin
357 host_disp = f"[{host}]" if ":" in host else host
358 suffix = "" if port == _DEFAULT_PORTS[scheme] else f":{port}"
359 return f"{scheme}://{host_disp}{suffix}"
362# Raw allowlist value we last warned about — the dropped-entry WARNING is
363# emitted once per distinct env value, not on every consult (the matcher
364# runs on every settings save, engine init, and /api/warnings recompute).
365_last_warned_allowlist_raw: Optional[str] = None
368def _private_engine_url_allowlist() -> frozenset:
369 """Operator-approved private engine-URL origins from the environment.
371 Reads the env-only ``search.private_engine_url_allowlist`` setting and
372 parses each comma-separated entry with ``_engine_url_origin``.
373 Unparseable entries are dropped (fail-closed) and warned about once per
374 distinct env value. Any lookup error yields the empty set — the safe
375 posture.
376 """
377 try:
378 from ...settings.env_registry import get_env_setting
380 raw = get_env_setting(_ALLOWLIST_SETTING_KEY, None)
381 except Exception: # noqa: silent-exception - default to the safe posture
382 return frozenset()
383 if not raw or not isinstance(raw, str):
384 return frozenset()
385 origins = set()
386 dropped = []
387 for entry in raw.split(","):
388 entry = entry.strip()
389 if not entry:
390 continue
391 origin = _engine_url_origin(entry)
392 if origin is not None:
393 origins.add(origin)
394 else:
395 dropped.append(entry)
396 if dropped:
397 # An unparseable entry grants nothing (fail-closed), but silently
398 # dropping it leaves the operator staring at the same rejection
399 # message telling them to add the origin they think they already
400 # added — so name what was ignored (once per distinct env value,
401 # to keep a permanently-malformed entry from spamming the logs).
402 global _last_warned_allowlist_raw
403 if raw != _last_warned_allowlist_raw:
404 _last_warned_allowlist_raw = raw
405 try:
406 from ..secure_logging import logger
408 logger.warning(
409 "Ignoring unparseable entries in "
410 "LDR_SEARCH_PRIVATE_ENGINE_URL_ALLOWLIST (each entry "
411 "must be a full origin like http://localhost:8080): "
412 f"{dropped}"
413 )
414 except Exception: # noqa: silent-exception - best-effort logging
415 pass
416 return frozenset(origins)
419def engine_url_in_private_allowlist(url) -> bool:
420 """True when ``url``'s origin is in the operator's env-only allowlist.
422 Grants ONLY the private/loopback/link-local exemption a matching origin
423 was listed for — callers consult it strictly in place of the blanket
424 ``search.allow_private_engine_urls`` gate, so the cloud-metadata block
425 and the http(s)-scheme requirement are unaffected by a listing.
426 """
427 origin = _engine_url_origin(url)
428 if origin is None:
429 return False
430 return origin in _private_engine_url_allowlist()
433def resolve_engine_allow_private_ips(
434 instance_url: str, url_setting: str
435) -> bool:
436 """Whether a PUBLIC engine may target a private/loopback instance URL.
438 A public-nature engine proxies the public web, so a private instance
439 URL is only legitimate when an OPERATOR chose it, never when an
440 arbitrary authenticated user smuggled an internal host in as an SSRF
441 probe. Private egress to the configured instance is therefore
442 permitted only when:
444 1. the operator opt-in ``search.allow_private_engine_urls`` is set
445 (env-only; the blanket LAN / localhost self-host case), OR
446 2. ``instance_url``'s exact origin is listed in the env-only operator
447 allowlist ``search.private_engine_url_allowlist`` (the
448 finer-grained alternative to the blanket opt-in), OR
449 3. the engine's own ``url_setting`` is env-locked (its derived
450 ``LDR_…`` variable is set — Docker / operator-provisioned,
451 trusted).
453 All three are OPERATOR-controlled. The run's egress scope
454 (``policy.egress_scope``) deliberately does NOT grant it: that setting is a
455 user-editable dropdown (STRICT included), so relaxing on it would be a
456 self-service SSRF bypass. By default private IPs are refused, so a
457 tampered instance URL cannot reach an internal host. Cloud-metadata /
458 link-local metadata IPs stay blocked regardless
459 (ALWAYS_BLOCKED_METADATA_IPS in the SSRF validator's probe path). Fails
460 CLOSED (False) on any error: a genuinely-public instance is unaffected —
461 only private egress is withheld.
462 """
463 # 1. Operator opt-in (LAN self-host).
464 try:
465 from ...settings.env_registry import get_env_setting
467 if bool(get_env_setting("search.allow_private_engine_urls", False)):
468 return True
469 except Exception: # noqa: silent-exception - fall through to next check
470 pass
472 # 2. Operator allowlist: this exact origin was individually approved.
473 try:
474 if engine_url_in_private_allowlist(instance_url):
475 return True
476 except Exception: # noqa: silent-exception - fall through to next check
477 pass
479 # 3. Env-locked instance URL (Docker / operator-provisioned).
480 try:
481 from ...settings.manager import check_env_setting
483 if check_env_setting(url_setting) is not None:
484 return True
485 except Exception: # noqa: silent-exception - fall through to next check
486 pass
488 # Otherwise: withhold private egress. We deliberately DO NOT relax this
489 # based on the run's resolved egress scope. ``policy.egress_scope`` is a
490 # user-editable, self-service setting (STRICT is a plain dropdown option,
491 # not operator-gated), and a public engine always queries the internet.
492 # Letting a user-chosen scope grant private-IP access would be a
493 # self-service SSRF bypass (set scope=strict, then point the instance
494 # URL at an internal host). Private egress for a public engine is an
495 # OPERATOR decision only — conditions 1–3 above.
496 return False
499def resolve_searxng_allow_private_ips(instance_url: str) -> bool:
500 """SearXNG-keyed convenience wrapper over
501 ``resolve_engine_allow_private_ips`` (kept for the engine and tests)."""
502 return resolve_engine_allow_private_ips(
503 instance_url, _SEARXNG_INSTANCE_URL_KEY
504 )
507def _engine_url_ssrf_error(key, value, *, allow_private: bool):
508 """Return an error dict for a public-engine URL that must be refused.
510 Rejects:
511 - a non-http(s) scheme (never a valid engine URL — refused even when the
512 operator opt-in is set),
513 - a host that IS or RESOLVES TO a cloud-metadata / always-blocked
514 address (refused regardless of the operator opt-in), and
515 - unless ``allow_private`` is set, a host that IS or RESOLVES TO a
516 private / loopback / link-local address.
518 Reuses ``ssrf_validator.is_ip_blocked`` for the classification: the
519 ``allow_localhost=True, allow_private_ips=True`` form isolates the
520 always-blocked cloud-metadata set, so metadata stays refused even under the
521 operator opt-in. Returns ``None`` when the value is not a URL string (type
522 validation handles that elsewhere).
524 Fail-open on transient resolution failure (DNS down / split-horizon): only
525 names that actually resolve to a blocked address are rejected, so a
526 legitimate public host stays saveable during a DNS hiccup — mirroring
527 ``validate_allowed_local_hostnames``.
528 """
529 if not isinstance(value, str) or not value.strip(): 529 ↛ 530line 529 didn't jump to line 530 because the condition on line 529 was never true
530 return None
531 url = value.strip()
533 from ..ssrf_validator import RFC_FORBIDDEN_URL_CHARS_RE, is_ip_blocked
535 if RFC_FORBIDDEN_URL_CHARS_RE.search(url):
536 return {
537 "key": key,
538 "error": f"{key} contains illegal URL characters",
539 }
540 try:
541 parsed = urlsplit(url)
542 except Exception:
543 return {"key": key, "error": f"{key} is not a valid URL"}
545 if (parsed.scheme or "").lower() not in ("http", "https"):
546 return {
547 "key": key,
548 "error": f"{key} must be an http:// or https:// URL",
549 }
551 host = parsed.hostname
552 if not host: 552 ↛ 553line 552 didn't jump to line 553 because the condition on line 552 was never true
553 return {"key": key, "error": f"{key} has no hostname"}
554 host = unquote(host)
555 if host.startswith("[") and host.endswith("]"): 555 ↛ 556line 555 didn't jump to line 556 because the condition on line 555 was never true
556 host = host[1:-1]
557 host = host.rstrip(".")
559 def _block_reason(ip_str) -> Optional[str]:
560 # Cloud-metadata / always-blocked set: refused regardless of opt-in.
561 if is_ip_blocked(ip_str, allow_localhost=True, allow_private_ips=True):
562 return "metadata"
563 # Private / loopback / link-local: refused unless operator opted in.
564 if not allow_private and is_ip_blocked(
565 ip_str, allow_localhost=False, allow_private_ips=False
566 ):
567 return "private"
568 return None
570 try:
571 ipaddress.ip_address(host)
572 is_literal_ip = True
573 except ValueError:
574 is_literal_ip = False
576 if is_literal_ip:
577 reason = _block_reason(host)
578 return _engine_url_error(key, reason) if reason else None
580 addr_info = _resolve_with_timeout(host)
581 if not addr_info:
582 return None # fail-open on unresolvable host
583 for entry in addr_info:
584 try:
585 ip_str = entry[4][0]
586 except (IndexError, TypeError):
587 continue
588 reason = _block_reason(ip_str)
589 if reason:
590 return _engine_url_error(key, reason)
591 return None
594def _engine_url_error(key, reason):
595 if reason == "metadata":
596 message = (
597 f"{key} points at a cloud-metadata / always-blocked address, "
598 "which is never a permitted destination."
599 )
600 else:
601 message = (
602 f"{key} points at a private, loopback, or link-local address. A "
603 "public search engine proxies the internet, so an internal URL is "
604 "refused to prevent it being used to reach your private network. "
605 "Self-hosted instances on localhost/LAN are still supported, but "
606 "the server operator must approve them in the server "
607 "environment: add this exact URL origin to "
608 "LDR_SEARCH_PRIVATE_ENGINE_URL_ALLOWLIST (comma-separated "
609 "scheme://host:port entries), or pin the URL via its LDR_ "
610 "environment variable (as the bundled docker-compose.yml does), "
611 "or set LDR_SEARCH_ALLOW_PRIVATE_ENGINE_URLS=true to allow all "
612 "private addresses. Only one of these is needed; restart after "
613 "changing. See docs/SearXNG-Setup.md."
614 )
615 return {"key": key, "error": message}
618# Sentinel marking a guarded key that has no stored value yet — treated as a
619# genuine change so a brand-new setting is still validated.
620_NO_STORED_VALUE = object()
623def _stored_engine_url_value(all_db_settings, key):
624 """Return the currently-stored value for ``key`` in ``all_db_settings``.
626 Entries are ``Setting`` ORM rows exposing the value via ``.value`` in the
627 live save routes (``_filter_editable_settings`` builds ``{key: Setting}``),
628 but some callers hand plain ``{"value": ...}`` dicts. Normalize both to the
629 scalar with ``unwrap_setting`` — the dict form directly, an ORM/attribute
630 row via its ``.value``. Returns ``_NO_STORED_VALUE`` when the key is absent.
631 """
632 if not all_db_settings or key not in all_db_settings:
633 return _NO_STORED_VALUE
634 entry = all_db_settings[key]
635 if not isinstance(entry, dict):
636 entry = getattr(entry, "value", _NO_STORED_VALUE)
637 if entry is _NO_STORED_VALUE: 637 ↛ 638line 637 didn't jump to line 638 because the condition on line 637 was never true
638 return _NO_STORED_VALUE
639 return unwrap_setting(entry)
642def _normalize_engine_url_for_comparison(value):
643 """Parse ``value`` into a tuple of URL parts for cosmetic-variant
644 comparison, or ``None`` if it cannot be parsed as a URL.
646 Collapses only variation that denotes the exact SAME url:
647 - scheme and host case (``HTTP://LOCALHOST`` == ``http://localhost``),
648 - any trailing slash(es) on the path (``rstrip("/")`` strips all of
649 them, so ``"/path//"`` and ``"/path"`` compare equal, and an
650 empty path and ``"/"`` compare equal too).
652 Everything that could denote a DIFFERENT destination is preserved
653 verbatim and NOT normalized away: the host itself, the port, any path
654 beyond a bare trailing slash, the query, the fragment, and any userinfo
655 (unusual in an ``instance_url``, but kept rather than risk folding two
656 distinct credentials together). In particular this never strips the
657 host, resolves DNS, or drops the port — so a genuine change to a
658 different host/port/scheme still fails to compare equal here.
660 Returns ``None`` on any parse failure (e.g. a non-numeric port, a
661 malformed IPv6 literal) so the caller falls back to the exact
662 string comparison — fail toward validating, never toward skipping.
663 """
664 try:
665 parsed = urlsplit(str(value).strip())
666 return (
667 (parsed.scheme or "").lower(),
668 parsed.username,
669 parsed.password,
670 (parsed.hostname or "").lower(),
671 parsed.port,
672 (parsed.path or "").rstrip("/"),
673 parsed.query,
674 parsed.fragment,
675 )
676 except Exception:
677 return None
680def _engine_url_is_unchanged(submitted_value, all_db_settings, key):
681 """True when the submitted engine URL equals the currently-stored one.
683 The frontend "All Settings" tab resubmits every rendered input, including a
684 shipped default the user never touched (e.g. the SearXNG
685 ``instance_url = http://localhost:8080`` loopback default). Validating an
686 unchanged value would 400 an otherwise-unrelated save, so only a genuine
687 change is validated. A brand-new key with no stored value is treated as
688 changed.
690 Both sides are first string-normalized so a metadata-wrapped vs scalar
691 representation of the same value is not misread as a change. If that
692 exact match fails, both sides are also compared via
693 ``_normalize_engine_url_for_comparison`` so a purely cosmetic resubmit —
694 different scheme/host case, or a trailing slash — of the SAME url is
695 still recognized as unchanged (e.g. resubmitting the stored
696 ``http://localhost:8080`` as ``HTTP://LOCALHOST:8080`` or
697 ``http://localhost:8080/``). This only ever widens what counts as
698 UNCHANGED (and is therefore skipped from validation); it never narrows
699 it, so it cannot make two genuinely different hosts compare equal. If
700 either side fails to parse, that comparison is skipped and the exact
701 match above is the only thing that can call it unchanged.
702 """
703 stored = _stored_engine_url_value(all_db_settings, key)
704 if stored is _NO_STORED_VALUE:
705 return False
706 submitted_str = str(submitted_value).strip()
707 stored_str = str(stored).strip()
708 if submitted_str == stored_str:
709 return True
710 submitted_norm = _normalize_engine_url_for_comparison(submitted_value)
711 stored_norm = _normalize_engine_url_for_comparison(stored)
712 if submitted_norm is None or stored_norm is None:
713 return False
714 return submitted_norm == stored_norm
717def validate_engine_instance_urls(form_data, all_db_settings):
718 """Reject a PUBLIC search-engine URL that targets a private address.
720 SearXNG (and any future public-nature engine with a configurable URL) is
721 user-editable through the settings API. Because such an engine proxies the
722 public web, a private / loopback / link-local ``instance_url`` is only ever
723 an internal-network probe target — so at save time we refuse it under the
724 default posture, closing the SSRF where any authenticated user turns a
725 research run into an internal port scan. Non-http(s) schemes are refused
726 outright.
728 The operator opt-in ``search.allow_private_engine_urls`` (env-only) lifts
729 the private-address rejection for genuine LAN self-hosting, and the
730 env-only allowlist ``search.private_engine_url_allowlist`` lifts it for
731 specific listed origins only; Docker deployments instead env-lock the
732 URL, which the route layer honors via ``check_env_setting`` before these
733 validators run. Neither mechanism lifts the cloud-metadata or
734 non-http(s)-scheme rejections.
735 """
736 guarded = _public_engine_url_settings()
737 present = [k for k in guarded if k in form_data]
738 if not present:
739 return None
741 allow_private = False
742 try:
743 from ...settings.env_registry import get_env_setting
745 allow_private = bool(
746 get_env_setting("search.allow_private_engine_urls", False)
747 )
748 except Exception: # noqa: silent-exception - default to the safe posture
749 allow_private = False
751 for key in present:
752 value = unwrap_setting(form_data[key])
753 # Validate-on-change: skip an UNCHANGED value (e.g. the shipped default
754 # loopback ``instance_url`` resubmitted untouched by the All-Settings
755 # tab) so it can't block an unrelated save. The runtime
756 # allow_private_ips=False backstop still blocks the actual fetch.
757 if _engine_url_is_unchanged(value, all_db_settings, key):
758 continue
759 err = _engine_url_ssrf_error(
760 key,
761 value,
762 allow_private=allow_private
763 or engine_url_in_private_allowlist(value),
764 )
765 if err is not None:
766 return err
767 return None
770# The cross-field egress validators every settings-write route must run. Keep
771# the *list* here (single source of truth) so adding one doesn't require a
772# lockstep edit at each save route; each route only supplies its own response
773# shaping around ``first_egress_validation_error``.
774EGRESS_SETTINGS_VALIDATORS = (
775 validate_egress_scope,
776 validate_allowed_local_hostnames,
777 validate_trusted_search_engines,
778 validate_engine_instance_urls,
779)
782def first_egress_validation_error(form_data, all_db_settings):
783 """Run the egress validators; return the first error dict, or ``None``."""
784 for _validator in EGRESS_SETTINGS_VALIDATORS:
785 err = _validator(form_data, all_db_settings)
786 if err is not None:
787 return err
788 return None