Coverage for src/local_deep_research/security/egress/validators.py: 88%
64 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""Cross-field egress-policy validators run at settings-save time.
3Pure functions: each takes the about-to-be-saved ``form_data`` plus the
4current ``all_db_settings`` (key -> setting row with a ``.value``) and returns
5a validation-error dict ``{"key", "error"}`` to surface in the save response,
6or ``None`` when the combination is fine.
8The settings write routes (``web/routes/settings_routes.py``) orchestrate these
9— they stay in the route layer because they also enforce non-egress concerns —
10but the egress rules themselves live here next to the policy they encode.
11"""
13from .policy import (
14 EgressContext,
15 EgressScope,
16 _classify_host,
17 _resolve_with_timeout,
18)
21def validate_allowed_local_hostnames(form_data, all_db_settings):
22 """Reject public hostnames being added to llm.allowed_local_hostnames.
24 The default-settings description for this key claims "Public hostnames
25 added here are rejected at save time", but until now no code actually
26 did the rejection. This guard resolves each entry via the same host
27 classifier the policy uses, and refuses any that resolve to public
28 addresses. A hostname that fails to resolve (DNS down) is accepted —
29 fail-open on transient lookup errors so the user can recover.
30 """
31 key = "llm.allowed_local_hostnames"
32 if key not in form_data:
33 return None
34 value = form_data[key]
35 # Setting is JSON-typed; the save pipeline may hand us a list or a
36 # JSON string. Decode defensively.
37 if isinstance(value, str):
38 try:
39 import json as _json
41 decoded = _json.loads(value) if value.strip() else []
42 except Exception:
43 return {
44 "key": key,
45 "error": "allowed_local_hostnames must be a JSON array of hostnames",
46 }
47 value = decoded
48 if not isinstance(value, list):
49 return {
50 "key": key,
51 "error": "allowed_local_hostnames must be a list",
52 }
54 # Build a minimal real context just for the resolver. Use the
55 # dataclass constructor — NOT EgressContext.__new__ + setattr, which
56 # raised FrozenInstanceError on this frozen dataclass and (separately)
57 # set a non-existent ``allowed_local_hostnames`` field instead of the
58 # real ``local_hostnames`` the classifier reads. The constructor
59 # initializes the init=False internals (_dns_cache, _lock as RLock)
60 # correctly. Empty local_hostnames => classify purely on IP class.
61 probe_ctx = EgressContext(
62 scope=EgressScope.BOTH,
63 primary_engine="searxng",
64 require_local_llm=False,
65 require_local_embeddings=False,
66 local_hostnames=(),
67 )
69 rejected = []
70 for entry in value:
71 if not isinstance(entry, str) or not entry.strip():
72 continue
73 hostname = entry.strip().lower()
74 try:
75 # Distinguish "could not resolve" from "resolved to a public IP".
76 # _classify_host collapses BOTH to False (its documented fail-safe
77 # treats an unresolvable host as public), so relying on it here
78 # would reject a legitimate intranet/VPN host on any DNS hiccup or
79 # split-horizon DNS — the exact use case this setting exists for.
80 # Only reject names that actually resolve to a public address;
81 # accept unresolvable ones (fail-open on save, as documented).
82 # _resolve_with_timeout returns the addrinfo for literal IPs too,
83 # so literal public/private IPs still flow through _classify_host.
84 if _resolve_with_timeout(hostname) is None:
85 continue
86 classification = _classify_host(hostname, probe_ctx)
87 except Exception:
88 # DNS or unknown error — allow (fail open) so the user can
89 # save when networking is flaky. Runtime classification will
90 # still gate egress.
91 continue
92 if classification is False:
93 rejected.append(hostname)
94 if rejected:
95 return {
96 "key": key,
97 "error": (
98 "These hostnames resolve to PUBLIC addresses and would "
99 "let the policy treat external hosts as local: "
100 f"{', '.join(rejected)}. Remove them, or use the SSRF "
101 "allowlist instead."
102 ),
103 }
104 return None
107def validate_trusted_search_engines(form_data, all_db_settings):
108 """Reject inherently-public engines from ``policy.trusted_search_engines``.
110 Trusting a search engine relaxes its Exposure to CONTAINED — meaningful only
111 for a local-nature store you host (Elasticsearch/Paperless). An inherently
112 public engine (google, searxng, brave, …) genuinely queries the internet, so
113 trusting it would launder a public sink past the two-axis rule. Reject those;
114 local / unknown names are accepted (runtime classification still gates them,
115 and ``engine_label`` also refuses to relax an ``is_public`` engine).
116 """
117 key = "policy.trusted_search_engines"
118 if key not in form_data:
119 return None
120 value = form_data[key]
121 if isinstance(value, str): 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true
122 import json
124 try:
125 value = json.loads(value) if value.strip() else []
126 except Exception:
127 return {
128 "key": key,
129 "error": "trusted_search_engines must be a JSON array",
130 }
131 if not isinstance(value, list): 131 ↛ 132line 131 didn't jump to line 132 because the condition on line 131 was never true
132 return {"key": key, "error": "trusted_search_engines must be a list"}
134 from .policy import _get_engine_class
136 rejected = []
137 for entry in value:
138 if not isinstance(entry, str) or not entry.strip(): 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true
139 continue
140 name = entry.strip().lower()
141 cls = _get_engine_class(name)
142 if cls is not None and getattr(cls, "is_public", False) is True:
143 rejected.append(name)
144 if rejected:
145 return {
146 "key": key,
147 "error": (
148 "These are inherently-public search engines and cannot be "
149 "trusted as contained: " + ", ".join(rejected) + "."
150 ),
151 }
152 return None
155# The cross-field egress validators every settings-write route must run. Keep
156# the *list* here (single source of truth) so adding one doesn't require a
157# lockstep edit at each save route; each route only supplies its own response
158# shaping around ``first_egress_validation_error``.
159EGRESS_SETTINGS_VALIDATORS = (
160 validate_allowed_local_hostnames,
161 validate_trusted_search_engines,
162)
165def first_egress_validation_error(form_data, all_db_settings):
166 """Run the egress validators; return the first error dict, or ``None``."""
167 for _validator in EGRESS_SETTINGS_VALIDATORS:
168 err = _validator(form_data, all_db_settings)
169 if err is not None:
170 return err
171 return None