Coverage for src/local_deep_research/web/warning_checks/__init__.py: 89%
151 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""Warning calculation for the settings UI.
3Thin orchestrator: reads settings from a single DB session,
4delegates to pure check functions in hardware.py and context.py.
5"""
7from typing import List, Optional
9from loguru import logger
11from ...database.session_context import get_user_db_session
12from ...utilities.db_utils import get_settings_manager
13from .context import (
14 check_context_below_history,
15 check_context_truncation_history,
16)
17from .backup import (
18 check_backup_disabled,
19 check_backup_healthy,
20 check_no_backups_exist,
21)
22from .hardware import (
23 LOCAL_PROVIDERS,
24 check_high_context,
25 check_legacy_server_config,
26 check_model_mismatch,
27)
28from ...security.egress.policy import DEFAULT_EGRESS_SCOPE
29from ...security.egress.warnings import (
30 check_cloud_embeddings_enabled,
31 check_cloud_llm_enabled,
32 check_effective_scope,
33 check_local_engine_public_url,
34 check_private_engine_url_blocked,
35 check_public_egress_enabled,
36 check_trusted_destinations,
37 check_unprotected_egress,
38)
39from ...constants import DEFAULT_SEARCH_TOOL
42def _safe_check(check_fn, *args, **kwargs):
43 """Run a single warning check, returning None on failure."""
44 try:
45 return check_fn(*args, **kwargs)
46 except Exception:
47 name = getattr(check_fn, "__name__", repr(check_fn))
48 logger.exception(f"Warning check {name} failed")
49 return None
52def calculate_warnings(username: Optional[str] = None) -> List[dict]:
53 """Calculate current warning conditions based on settings.
55 Uses a single DB session for all setting reads and history queries.
57 Args:
58 username: User whose settings/history to inspect. Callers in
59 FastAPI routes should pass this explicitly (from Depends(
60 require_auth)). When omitted, returns [] since we no longer
61 have a Flask session to read it from.
62 """
63 warnings: List[dict] = []
65 try:
66 if not username:
67 return []
68 with get_user_db_session(username) as db_session:
69 if not db_session:
70 return []
72 settings_manager = get_settings_manager(db_session, username)
74 # Read all needed settings in one session
75 provider = settings_manager.get_setting(
76 "llm.provider", "ollama"
77 ).lower()
78 local_context = settings_manager.get_setting(
79 "llm.local_context_window_size", 8192
80 )
81 current_model = settings_manager.get_setting("llm.model", "")
82 dismiss_high_context = settings_manager.get_setting(
83 "app.warnings.dismiss_high_context", False
84 )
85 dismiss_model_mismatch = settings_manager.get_setting(
86 "app.warnings.dismiss_model_mismatch", False
87 )
88 dismiss_context_below_history = settings_manager.get_setting(
89 "app.warnings.dismiss_context_below_history", False
90 )
91 dismiss_context_truncation_history = settings_manager.get_setting(
92 "app.warnings.dismiss_context_truncation_history", False
93 )
94 dismiss_legacy_config = settings_manager.get_setting(
95 "app.warnings.dismiss_legacy_config", False
96 )
97 backup_enabled = settings_manager.get_setting(
98 "backup.enabled", True
99 )
100 dismiss_backup_disabled = settings_manager.get_setting(
101 "app.warnings.dismiss_backup_disabled", False
102 )
103 dismiss_no_backups = settings_manager.get_setting(
104 "app.warnings.dismiss_no_backups", False
105 )
107 logger.debug(f"Starting warning calculation - provider={provider}")
109 is_local = provider in LOCAL_PROVIDERS
111 # --- Hardware / settings checks (pure functions) ---
112 w = _safe_check(
113 check_high_context,
114 provider,
115 local_context,
116 dismiss_high_context,
117 )
118 if w:
119 warnings.append(w)
121 w = _safe_check(
122 check_model_mismatch,
123 provider,
124 current_model,
125 local_context,
126 dismiss_model_mismatch,
127 )
128 if w:
129 warnings.append(w)
131 w = _safe_check(check_legacy_server_config, dismiss_legacy_config)
132 if w:
133 warnings.append(w)
135 # --- Egress policy checks ---
136 egress_scope = settings_manager.get_setting(
137 "policy.egress_scope", DEFAULT_EGRESS_SCOPE
138 )
139 require_local_endpoint = bool(
140 settings_manager.get_setting(
141 "llm.require_local_endpoint", False
142 )
143 )
144 embeddings_provider = settings_manager.get_setting(
145 "embeddings.provider", ""
146 )
147 embeddings_base_url = settings_manager.get_setting(
148 "embeddings.openai.base_url", ""
149 )
150 require_local_embeddings = bool(
151 settings_manager.get_setting("embeddings.require_local", False)
152 )
153 primary_engine = settings_manager.get_setting(
154 "search.tool", DEFAULT_SEARCH_TOOL
155 )
156 trusted_inference = settings_manager.get_setting(
157 "policy.trusted_inference_providers", []
158 )
159 trusted_search = settings_manager.get_setting(
160 "policy.trusted_search_engines", []
161 )
163 # Resolve the EFFECTIVE posture so the banners are accurate. For
164 # `adaptive`, this turns the opaque "follows the primary" into a
165 # concrete scope; it also applies the PRIVATE_ONLY -> force-local
166 # coupling, so a private-resolving run doesn't falsely show the
167 # "cloud LLM enabled" banner. Best-effort: any failure falls back
168 # to the raw values (the page must never break on this).
169 effective_scope = str(egress_scope).lower()
170 effective_require_local_endpoint = require_local_endpoint
171 effective_require_local_embeddings = require_local_embeddings
172 try:
173 from ...security.egress.policy import context_from_snapshot
175 _snap = settings_manager.get_settings_snapshot()
176 if isinstance(_snap, dict):
177 # allow_dns=False: this runs on the /api/warnings page-
178 # render hot path; skip the synchronous getaddrinfo that
179 # ADAPTIVE resolution would otherwise do for a URL-engine
180 # primary (could block the render up to _DNS_TIMEOUT_SEC).
181 # The banner is advisory and falls back to static
182 # classification — accuracy here is best-effort by design.
183 _eff_ctx = context_from_snapshot(
184 _snap,
185 primary_engine or DEFAULT_SEARCH_TOOL,
186 username=username,
187 allow_dns=False,
188 )
189 effective_scope = _eff_ctx.scope.value
190 effective_require_local_endpoint = (
191 _eff_ctx.require_local_llm
192 )
193 effective_require_local_embeddings = (
194 _eff_ctx.require_local_embeddings
195 )
196 except Exception:
197 logger.debug(
198 "could not resolve effective egress scope for warnings",
199 exc_info=True,
200 )
202 adaptive_info_dismissed = bool(
203 settings_manager.get_setting(
204 "app.warnings.dismiss_adaptive_scope_info", False
205 )
206 )
208 # Each egress banner has its OWN dismiss flag. Previously all
209 # three shared app.warnings.dismiss_egress_policy, so dismissing
210 # the fresh-install "public egress" notice ALSO permanently hid
211 # the critical cloud-LLM / cloud-embeddings warnings — a
212 # false-safety trap (switch to OpenAI later, never warned).
213 public_egress_dismissed = bool(
214 settings_manager.get_setting(
215 "app.warnings.dismiss_egress_policy", False
216 )
217 )
218 cloud_llm_dismissed = bool(
219 settings_manager.get_setting(
220 "app.warnings.dismiss_cloud_llm", False
221 )
222 )
223 cloud_embeddings_dismissed = bool(
224 settings_manager.get_setting(
225 "app.warnings.dismiss_cloud_embeddings", False
226 )
227 )
229 # Informational: state what ADAPTIVE actually resolves to.
230 w = _safe_check(
231 check_effective_scope,
232 egress_scope,
233 effective_scope,
234 primary_engine,
235 adaptive_info_dismissed,
236 )
237 if w:
238 warnings.append(w)
240 w = _safe_check(
241 check_public_egress_enabled,
242 effective_scope,
243 public_egress_dismissed,
244 )
245 if w:
246 warnings.append(w)
248 # Loud, non-dismissible banner when protection is turned off.
249 w = _safe_check(check_unprotected_egress, effective_scope)
250 if w:
251 warnings.append(w)
253 # Trusted off-machine destinations (stage D) relax classification.
254 w = _safe_check(
255 check_trusted_destinations, trusted_inference, trusted_search
256 )
257 if w: 257 ↛ 258line 257 didn't jump to line 258 because the condition on line 257 was never true
258 warnings.append(w)
260 w = _safe_check(
261 check_cloud_llm_enabled,
262 provider,
263 effective_require_local_endpoint,
264 cloud_llm_dismissed,
265 )
266 if w:
267 warnings.append(w)
269 w = _safe_check(
270 check_cloud_embeddings_enabled,
271 embeddings_provider,
272 embeddings_base_url,
273 effective_require_local_embeddings,
274 cloud_embeddings_dismissed,
275 )
276 if w: 276 ↛ 277line 276 didn't jump to line 277 because the condition on line 276 was never true
277 warnings.append(w)
279 # Engine-URL misconfiguration banners, driven entirely by the
280 # engine declarations (url_setting + is_public/is_local) that
281 # guarded_engine_url_descriptors() collects from the registry:
282 # - a PUBLIC engine (SearXNG) whose private instance URL lacks
283 # operator approval silently self-disables at run time;
284 # - a LOCAL document engine (Paperless, Elasticsearch) whose
285 # URL looks public would send document queries off-box and
286 # is excluded under a private-only scope (fail-up).
287 # Display names, activation flags, and dismiss keys follow the
288 # per-engine settings conventions, so a future engine is covered
289 # by declaring url_setting — not by editing this file.
290 # A failure in this loop (settings read, descriptor sweep) must
291 # not swallow the later backup/history banners, so it gets its
292 # own guard rather than relying on the function-wide except.
293 try:
294 primary_lower = str(primary_engine or "").lower()
295 from ...security.egress.validators import (
296 guarded_engine_url_descriptors,
297 )
299 for desc in guarded_engine_url_descriptors():
300 name = desc.engine_name
301 base = f"search.engine.web.{name}"
302 display = (
303 settings_manager.get_setting(f"{base}.display_name", "")
304 or name.replace("_", " ").title()
305 )
306 url_value = settings_manager.get_setting(
307 desc.url_setting, ""
308 )
309 engine_active = primary_lower == name or any(
310 bool(
311 settings_manager.get_setting(
312 f"{base}.{flag}", False
313 )
314 )
315 for flag in (
316 "enabled",
317 "use_in_auto_search",
318 "agent_enabled",
319 )
320 )
321 if desc.is_public:
322 w = _safe_check(
323 check_private_engine_url_blocked,
324 display_name=display,
325 engine_name=name,
326 url_setting=desc.url_setting,
327 instance_url=url_value,
328 active=engine_active,
329 acknowledged=bool(
330 settings_manager.get_setting(
331 f"app.warnings.dismiss_{name}_private_url",
332 False,
333 )
334 ),
335 )
336 else:
337 # Engine-specific public-endpoint indicator: an
338 # Elasticsearch cloud_id always targets Elastic
339 # Cloud, regardless of what the hosts list says.
340 extra = ""
341 if name == "elasticsearch":
342 cloud_id = settings_manager.get_setting(
343 f"{base}.default_params.cloud_id", ""
344 )
345 if str(cloud_id or "").strip():
346 extra = "Elastic Cloud (cloud_id)"
347 w = _safe_check(
348 check_local_engine_public_url,
349 display_name=display,
350 engine_name=name,
351 urls=url_value,
352 active=engine_active,
353 acknowledged=bool(
354 settings_manager.get_setting(
355 f"app.warnings.dismiss_{name}_public_url",
356 False,
357 )
358 ),
359 extra=extra,
360 url_setting=desc.url_setting,
361 )
362 if w:
363 warnings.append(w)
364 except Exception:
365 # logger.exception, not debug: if the descriptor sweep
366 # breaks (e.g. an engine-import regression) the whole
367 # banner feature goes dark — that must be visible in
368 # production logs, matching _safe_check's severity bar.
369 logger.exception("engine-URL warning banners skipped")
371 # --- Backup checks ---
372 w = _safe_check(
373 check_backup_disabled, backup_enabled, dismiss_backup_disabled
374 )
375 if w: 375 ↛ 376line 375 didn't jump to line 376 because the condition on line 375 was never true
376 warnings.append(w)
378 # Check backup file status (lightweight filesystem glob)
379 dismiss_backup_info = settings_manager.get_setting(
380 "app.warnings.dismiss_backup_info", False
381 )
382 try:
383 from ...config.paths import get_user_backup_directory
384 from ...utilities.formatting import human_size
385 from ...database.backup.backup_service import (
386 is_safe_glob_result,
387 )
389 if username: 389 ↛ 427line 389 didn't jump to line 427 because the condition on line 389 was always true
390 backup_dir = get_user_backup_directory(username)
391 total_size = 0
392 backup_count = 0
393 for f in backup_dir.glob("ldr_backup_*.db"):
394 # Skip symlinks / entries resolving outside backup_dir
395 # so a planted symlink can't inflate the count/size
396 # shown in warnings — same hardening as BackupService.
397 if not is_safe_glob_result(f, backup_dir): 397 ↛ 399line 397 didn't jump to line 399 because the condition on line 397 was always true
398 continue
399 try:
400 total_size += f.stat().st_size
401 backup_count += 1
402 except FileNotFoundError:
403 continue
405 w = _safe_check(
406 check_no_backups_exist,
407 backup_enabled,
408 backup_count,
409 dismiss_no_backups,
410 )
411 if w: 411 ↛ 414line 411 didn't jump to line 414 because the condition on line 411 was always true
412 warnings.append(w)
414 w = _safe_check(
415 check_backup_healthy,
416 backup_enabled,
417 backup_count,
418 human_size(total_size),
419 dismiss_backup_info,
420 )
421 if w: 421 ↛ 422line 421 didn't jump to line 422 because the condition on line 421 was never true
422 warnings.append(w)
423 except Exception:
424 logger.debug("Backup status check skipped")
426 # --- History-based checks (need DB queries) ---
427 if is_local and not dismiss_context_below_history:
428 w = _safe_check(
429 check_context_below_history, db_session, local_context
430 )
431 if w:
432 warnings.append(w)
434 if is_local and not dismiss_context_truncation_history:
435 w = _safe_check(
436 check_context_truncation_history, db_session, local_context
437 )
438 if w:
439 warnings.append(w)
441 except Exception:
442 logger.exception("Error calculating warnings")
444 return warnings