Coverage for src/local_deep_research/security/egress/warnings.py: 96%
57 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""Egress-policy warning checks.
3Pure 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.
7The audit identified three independent vectors that each deserve their
8own banner:
101. **Public search egress enabled** — the active scope permits any public
11 engine to fire. The default scope ``adaptive`` can resolve to a
12 public-allowing posture, so this fires on a fresh install too; suppressed
13 via the dismiss flag so first-launch isn't a wall of red.
142. **Cloud LLM enabled** — the user hasn't opted into
15 ``llm.require_local_endpoint`` and the configured provider is one of
16 the unambiguously-cloud providers. Critical-severity because the
17 user's full prompt content leaks on every research run.
183. **Cloud embeddings enabled** — even worse: indexing a corpus with
19 OpenAI embeddings POSTs every chunk to OpenAI. Critical-severity.
20"""
22from typing import Optional
24# Single source of truth — these warning checks now live in the same
25# egress package as the PDP, so import the cloud-provider set directly
26# instead of maintaining a hand-synced copy (which previously risked
27# drift: a provider added to one list but not the other).
28from .policy import _CLOUD_LLM_PROVIDERS
31def check_public_egress_enabled(
32 egress_scope: str,
33 acknowledged: bool,
34) -> Optional[dict]:
35 """Banner when the active scope permits public-internet search engines.
37 Fires for ``adaptive`` (which can resolve to a public posture), ``both``,
38 and ``public_only`` — the scopes that can allow public engines. Suppressed
39 when the user has acknowledged the
40 egress-policy warnings via the fresh-install flag — otherwise every
41 new install would face three loud banners before doing anything.
42 """
43 if acknowledged:
44 return None
45 # "adaptive" can resolve to a public-allowing scope (public or
46 # unclassifiable primary), so warn conservatively — better a dismissible
47 # banner than a silent public-egress path the user didn't expect.
48 if egress_scope not in ("both", "public_only", "adaptive"):
49 return None
51 return {
52 "type": "public_egress_enabled",
53 "icon": "🌐",
54 "title": "Public search egress enabled",
55 "message": (
56 "This run can reach external search engines. Set the Egress "
57 "Scope to 'Private only' below if you want to keep all "
58 "research traffic on-machine."
59 ),
60 "dismissKey": "app.warnings.dismiss_egress_policy",
61 "actionUrl": "#policy_egress_scope",
62 "actionLabel": "Adjust scope",
63 }
66def check_unprotected_egress(egress_scope: str) -> Optional[dict]:
67 """Loud, non-dismissible banner when egress protection is turned off.
69 The ``unprotected`` escape hatch disables all egress-scope restrictions
70 for the run — any engine / URL / LLM / embeddings provider is permitted
71 (only the hard SSRF and cloud-metadata blocks remain). Deliberately not
72 dismissible: the user should never forget protection is off.
73 """
74 if (egress_scope or "").lower() != "unprotected":
75 return None
77 return {
78 "type": "egress_unprotected",
79 "icon": "⚠️",
80 "title": "Egress protection is disabled",
81 "message": (
82 "Egress Scope is set to 'Unprotected' — this run may reach any "
83 "search engine, URL, and LLM/embeddings provider. Only hard SSRF "
84 "and cloud-metadata blocking still applies. Pick another scope "
85 "below to re-enable protection."
86 ),
87 "dismissKey": None,
88 "actionUrl": "#policy_egress_scope",
89 "actionLabel": "Adjust scope",
90 }
93def check_trusted_destinations(
94 trusted_inference, trusted_engines
95) -> Optional[dict]:
96 """Banner when the user has vouched for off-machine destinations.
98 Trusting an inference provider or search engine relaxes the two-axis
99 classification (treats an off-machine sink as contained), so sensitive
100 sources may flow to it. Surfaced so a stale trust entry can't silently keep
101 sending data off-box.
102 """
103 from .policy import coerce_str_list
105 names = sorted(
106 {
107 n
108 for n in coerce_str_list(trusted_inference)[1]
109 + coerce_str_list(trusted_engines)[1]
110 if n.strip()
111 }
112 )
113 if not names:
114 return None
116 return {
117 "type": "egress_trusted_destinations",
118 "icon": "🤝",
119 "title": "Trusted destinations configured",
120 "message": (
121 "These off-machine destinations are marked trusted, so sensitive "
122 "sources may be combined with them: "
123 + ", ".join(names)
124 + ". Remove any you did not intend."
125 ),
126 "dismissKey": None,
127 "actionUrl": "#policy_egress_scope",
128 "actionLabel": "Review",
129 }
132def check_effective_scope(
133 egress_scope: str,
134 effective_scope: str,
135 primary_engine: str,
136 acknowledged: bool,
137) -> Optional[dict]:
138 """Informational banner stating what the ADAPTIVE scope actually resolves
139 to for the current primary engine.
141 Adaptive is opaque on its own ("follows your primary"); this makes the
142 effective posture explicit so the user knows whether THIS config means
143 public searches, private/local-only, or both. Only fires for ``adaptive``
144 (the explicit scopes are self-describing in the dropdown). Has its own
145 dismiss flag so dismissing it doesn't hide the risk banners.
146 """
147 if acknowledged:
148 return None
149 if (egress_scope or "").lower() != "adaptive":
150 return None
151 eff = (effective_scope or "").lower()
152 primary = primary_engine or "your primary engine"
153 base = {
154 "type": "egress_effective_scope",
155 "dismissKey": "app.warnings.dismiss_adaptive_scope_info",
156 "actionUrl": "#policy_egress_scope",
157 "actionLabel": "Change mode",
158 }
159 if eff == "private_only":
160 return {
161 **base,
162 "icon": "🔒",
163 "title": "Adaptive → Private only (stays local)",
164 "message": (
165 f"Your primary engine ('{primary}') is private, so this run "
166 "stays on your machine: only local engines run, and LLM + "
167 "embeddings are forced local — nothing leaves the box."
168 ),
169 }
170 if eff == "public_only":
171 return {
172 **base,
173 "icon": "🌐",
174 "title": "Adaptive → Public searches enabled",
175 "message": (
176 f"Your primary engine ('{primary}') is public, so this run "
177 "uses public web/academic engines. Your local collections are "
178 "not queried."
179 ),
180 }
181 # both (unclassifiable primary)
182 return {
183 **base,
184 "icon": "🔀",
185 "title": "Adaptive → Public + private searches enabled",
186 "message": (
187 f"Your primary ('{primary}') could not be classified as public "
188 "or private, so this run can use both public engines and your "
189 "local collections."
190 ),
191 }
194def check_cloud_llm_enabled(
195 provider: str,
196 require_local_endpoint: bool,
197 acknowledged: bool,
198) -> Optional[dict]:
199 """Banner when the configured LLM provider is cloud-only and the
200 require-local-endpoint toggle is off.
202 Critical-severity: the user's full prompt content (including the
203 research query and all retrieved context) is sent to the provider on
204 every call.
205 """
206 if acknowledged:
207 return None
208 if require_local_endpoint:
209 return None
210 if not provider or provider.lower() not in _CLOUD_LLM_PROVIDERS:
211 return None
213 return {
214 "type": "cloud_llm_enabled",
215 "icon": "☁️",
216 "title": "LLM provider is cloud-hosted",
217 "message": (
218 f"Your LLM provider ({provider}) is cloud-hosted. Query "
219 "content will be sent off-machine on every research run, "
220 "independent of the Egress Scope setting. Tick 'Require local "
221 "LLM endpoint' below if you want fully local inference."
222 ),
223 "dismissKey": "app.warnings.dismiss_cloud_llm",
224 "actionUrl": "#llm_require_local_endpoint",
225 "actionLabel": "Require local LLM",
226 }
229def check_cloud_embeddings_enabled(
230 embeddings_provider: str,
231 embeddings_base_url: str,
232 require_local_embeddings: bool,
233 acknowledged: bool,
234) -> Optional[dict]:
235 """Banner when the embeddings provider sends data off-machine on
236 indexing.
238 Highest-severity of the three: indexing a private corpus with OpenAI
239 embeddings POSTs every chunk to OpenAI's API. A user who is unaware
240 of this loses their corpus.
242 Suppressed when ``base_url`` is set to a local URL (LM Studio,
243 vLLM, llama.cpp), since the OpenAI provider type is then pointed at
244 a local endpoint.
245 """
246 if acknowledged:
247 return None
248 if require_local_embeddings: 248 ↛ 249line 248 didn't jump to line 249 because the condition on line 248 was never true
249 return None
250 if (embeddings_provider or "").lower() != "openai":
251 return None
253 # If base_url is set and points to a local hostname, suppress the
254 # warning — the user has configured OpenAI-compatible-but-local
255 # (LM Studio, vLLM, etc.). Uses is_private_ip so RFC1918 ranges
256 # (10.x, 172.16-31.x, 192.168.x), CGNAT, link-local, IPv6 private,
257 # and .local mDNS hosts are all recognised. Substring-matching
258 # against a small list missed legitimate private-network endpoints.
259 #
260 # NOTE: this is a LITERAL-IP / .local check only — it does NOT resolve
261 # DNS, unlike the enforcing evaluate_embeddings (which DNS-resolves via
262 # _classify_host). So a private-DNS base_url (e.g. my-llm.internal → 10.x)
263 # is treated as LOCAL by enforcement (nothing leaves) yet still shows this
264 # banner. That divergence is intentional: this advisory runs on every
265 # settings-page render and must stay synchronous/cheap, so it deliberately
266 # over-warns (the safe direction) rather than add a blocking DNS lookup to
267 # page load. The banner never blocks anything; the PDP is the source of
268 # truth for what actually egresses.
269 if embeddings_base_url:
270 try:
271 from urllib.parse import urlsplit
273 from ..network_utils import is_private_ip
275 parsed = urlsplit(embeddings_base_url)
276 hostname = parsed.hostname
277 if hostname and is_private_ip(hostname):
278 return None
279 except Exception: # noqa: silent-exception
280 # Defensive: if URL parsing fails, fall through to issuing
281 # the banner — the failure itself is a sign of misconfig
282 # the user should see, not a reason to mask the warning.
283 pass
285 return {
286 "type": "cloud_embeddings_enabled",
287 "icon": "📤",
288 "title": "Document chunks will be sent to OpenAI",
289 "message": (
290 "Your embeddings provider is OpenAI. Indexing a collection "
291 "will POST every chunk to OpenAI's API — the entire corpus "
292 "leaves the machine. Tick 'Require local embeddings' below "
293 "to switch to sentence-transformers or a local OpenAI-"
294 "compatible endpoint."
295 ),
296 "dismissKey": "app.warnings.dismiss_cloud_embeddings",
297 "actionUrl": "#embeddings_require_local",
298 "actionLabel": "Require local embeddings",
299 }