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