Coverage for src/local_deep_research/web/routes/settings_routes.py: 91%
939 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1"""
2Settings Routes Module
4This module handles all settings-related HTTP endpoints for the application.
6CHECKBOX HANDLING PATTERN:
7--------------------------
8This module supports TWO submission modes to handle checkboxes correctly:
10**MODE 1: AJAX/JSON Submission (Primary - /save_all_settings)**
11- JavaScript intercepts form submission with e.preventDefault()
12- Checkbox values read directly from DOM via checkbox.checked
13- Data sent as JSON: {"setting.key": true/false}
14- Hidden fallback inputs are managed but NOT used in this mode
15- Provides better UX with instant feedback and validation
17**MODE 2: Traditional POST Submission (Fallback - /save_settings)**
18- Used when JavaScript is disabled (accessibility/no-JS environments)
19- Browser submits form data naturally via request.form
20- Hidden fallback pattern CRITICAL here:
21 * Checked checkbox: Submits checkbox value, hidden input disabled
22 * Unchecked checkbox: Submits hidden input value "false"
23- Ensures unchecked checkboxes are captured (HTML limitation workaround)
25**Implementation Details:**
261. Each checkbox has `data-hidden-fallback` attribute → hidden input ID
272. checkbox_handler.js manages hidden input disabled state
283. AJAX mode: settings.js reads checkbox.checked directly (lines 2233-2240)
294. POST mode: Flask reads request.form including enabled hidden inputs
305. Both modes use convert_setting_value() for consistent boolean conversion
32**Why Both Patterns?**
33- AJAX: Better UX, immediate validation, no page reload
34- Traditional POST: Accessibility, progressive enhancement, JavaScript-free operation
35- Hidden inputs: Only meaningful for traditional POST, ignored in AJAX mode
37This dual-mode approach ensures the app works for all users while providing
38optimal experience when JavaScript is available.
39"""
41import platform
42import time
43from typing import Any, Optional
44from datetime import UTC, datetime, timedelta, timezone
46import requests
47from flask import (
48 Blueprint,
49 flash,
50 jsonify,
51 redirect,
52 request,
53 session,
54 url_for,
55)
56from flask_wtf.csrf import generate_csrf
57from loguru import logger
58from sqlalchemy.orm import Session
60from ...config.constants import DEFAULT_OLLAMA_URL
61from ...llm.providers.base import normalize_provider
62from ...config.paths import get_data_directory, get_encrypted_database_path
63from ...database.models import RateLimitEstimate, Setting, SettingType
64from ...database.session_context import get_user_db_session
65from ...database.encrypted_db import db_manager
66from ...utilities.db_utils import get_settings_manager
67from ...utilities.url_utils import normalize_url
68from ...security.decorators import require_json_body
69from ...security.egress.policy import DEFAULT_EGRESS_SCOPE
70from ...security.egress.validators import (
71 first_egress_validation_error,
72)
73from ..auth.decorators import login_required
74from ..utils.request_helpers import parse_bool_arg
75from ...security.rate_limiter import settings_limit
76from ...settings.manager import (
77 get_typed_setting_value,
78 is_valid_setting_key,
79 parse_boolean,
80)
81from ..services.settings_service import (
82 DYNAMIC_SETTINGS, # noqa: F401 — re-exported for tests that import from routes
83 create_or_update_setting,
84 invalidate_settings_caches,
85 reschedule_document_jobs_if_needed,
86 reschedule_zotero_jobs_if_needed,
87 set_setting,
88 validate_setting,
89)
90from ..utils.route_decorators import with_user_session
91from ..utils.templates import render_template_with_defaults
94from ...security import safe_get
95from ..warning_checks import calculate_warnings
96from ...constants import DEFAULT_SEARCH_TOOL
98# Settings whose changes trigger a warning recalculation.
99WARNING_AFFECTING_KEYS = frozenset(
100 [
101 "llm.provider",
102 "search.tool",
103 "search.iterations",
104 "search.questions_per_iteration",
105 "llm.local_context_window_size",
106 "llm.context_window_unrestricted",
107 "llm.context_window_size",
108 "policy.egress_scope",
109 "llm.require_local_endpoint",
110 "embeddings.require_local",
111 ]
112)
114# Create a Blueprint for settings
115settings_bp = Blueprint("settings", __name__, url_prefix="/settings")
118def _filter_editable_settings(form_data: dict, db_session: Session) -> dict:
119 """Remove non-editable keys from *form_data* in place.
121 Returns a dict of *all* ``{key: Setting}`` records from the database
122 so callers can reuse it for further validation (e.g. egress-policy checks).
123 """
124 all_db_settings = {
125 setting.key: setting for setting in db_session.query(Setting).all()
126 }
128 non_editable_keys = [
129 key
130 for key in form_data.keys()
131 if key in all_db_settings and not all_db_settings[key].editable
132 ]
133 if non_editable_keys:
134 logger.warning("Skipping non-editable settings: {}", non_editable_keys)
135 for key in non_editable_keys:
136 del form_data[key]
138 return all_db_settings
141# NOTE: Routes use session["username"] (not .get()) intentionally.
142# @login_required guarantees the key exists; direct access fails fast
143# if the decorator is ever removed.
145# Namespace validation for new setting creation via the web API.
146# Keys starting with any ALLOWED prefix may be created; any prefix in
147# BLOCKED takes precedence and is rejected even if it also matches an
148# allowed prefix. Existing keys (updates) bypass this check — it only
149# applies to creation of new DB rows through the three write routes.
150ALLOWED_SETTING_PREFIXES = frozenset(
151 {
152 "app.",
153 "backup.",
154 "benchmark.",
155 "chat.",
156 "database.",
157 "document_scheduler.",
158 "embeddings.",
159 "focused_iteration.",
160 "general.",
161 "langgraph_agent.",
162 "llm.",
163 "local_search_",
164 "news.",
165 "notifications.",
166 "rag.",
167 "rate_limiting.",
168 "report.",
169 "research_library.",
170 "search.",
171 "ui.",
172 "web.",
173 "zotero.",
174 }
175)
176BLOCKED_SETTING_PREFIXES = frozenset(
177 {
178 "auth.",
179 "bootstrap.",
180 "db_config.",
181 "security.",
182 "server.",
183 "testing.",
184 }
185)
188def _is_allowed_new_setting_key(key: str) -> bool:
189 """Return True if *key* is permitted to be created via the web API."""
190 # Reject malformed keys (blank, trailing/leading dot, empty ".." segment,
191 # stray whitespace) before the namespace check — a trailing-dot key such
192 # as ``local_search_chunk_size.`` otherwise passes the prefix allow-list
193 # and corrupts prefix lookups (see #4840).
194 if not is_valid_setting_key(key):
195 return False
196 key = key.lower()
197 for prefix in BLOCKED_SETTING_PREFIXES:
198 if key.startswith(prefix):
199 return False
200 for prefix in ALLOWED_SETTING_PREFIXES:
201 if key.startswith(prefix):
202 return True
203 return False
206def _new_key_rejection_reason(key) -> str:
207 """Explain why ``_is_allowed_new_setting_key`` rejected *key*.
209 Distinguishes a malformed key (bad syntax) from an allowed-namespace
210 violation so an API consumer with, say, a trailing-dot key (#4840) is
211 pointed at the real problem instead of being told it's a namespace issue.
212 """
213 if not is_valid_setting_key(key):
214 return f"Setting key is malformed: {key!r}"
215 return f"Creating settings under this namespace is not allowed: {key}"
218def _get_setting_from_session(key: str | None, default=None):
219 """Helper to get a setting using the current session context.
221 A ``None`` key returns ``default``. ``SettingsManager.get_setting``
222 treats ``key=None`` as "return all settings" (a real feature used by
223 ``get_all_settings`` for enumeration); this route helper is for
224 fetching one named setting and must not inherit that bulk-read
225 semantic. Without the guard, callers would receive a dict of every
226 setting (including other providers' API keys) when a provider
227 declares ``api_key_setting = None``.
229 NOTE: As of the LLM construction collapse (PR follow-up to #3957),
230 LM Studio and Llama.cpp now declare real ``api_key_setting`` values
231 with ``api_key_optional = True``, so this None guard fires only for
232 truly key-less providers (custom subclasses that don't use settings
233 at all).
234 """
235 if key is None:
236 return default
237 username = session.get("username")
238 with get_user_db_session(username) as db_session:
239 if db_session:
240 settings_manager = get_settings_manager(db_session, username)
241 return settings_manager.get_setting(key, default)
242 return default
245def _model_list_local_only() -> bool:
246 """True when the model-list endpoints must NOT call cloud provider APIs
247 (doing so sends the API key off-machine).
249 The effective local-only posture = the ``llm.require_local_endpoint``
250 toggle OR a private egress scope (``private_only``, or ``adaptive``
251 resolving to private). Checking only the raw toggle missed a PRIVATE_ONLY
252 user who never ticked it — the scope forces local at run time but the
253 stored toggle is still False. Best-effort: returns False (allow) on error
254 so a transient settings hiccup doesn't break the model dropdown.
255 """
256 username = session.get("username")
257 try:
258 with get_user_db_session(username) as db_session:
259 if not db_session: 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true
260 return False
261 sm = get_settings_manager(db_session, username)
262 if bool(sm.get_setting("llm.require_local_endpoint", False)): 262 ↛ 263line 262 didn't jump to line 263 because the condition on line 262 was never true
263 return True
264 scope = str(
265 sm.get_setting("policy.egress_scope", DEFAULT_EGRESS_SCOPE)
266 ).lower()
267 if scope == "private_only": 267 ↛ 268line 267 didn't jump to line 268 because the condition on line 267 was never true
268 return True
269 if scope == "adaptive": 269 ↛ 282line 269 didn't jump to line 282 because the condition on line 269 was always true
270 from ...security.egress.policy import context_from_snapshot
272 snap = sm.get_settings_snapshot()
273 if isinstance(snap, dict):
274 primary = sm.get_setting("search.tool", DEFAULT_SEARCH_TOOL)
275 return bool(
276 context_from_snapshot(
277 snap,
278 primary or DEFAULT_SEARCH_TOOL,
279 username=username,
280 ).require_local_llm
281 )
282 return False
283 except Exception:
284 logger.debug(
285 "model-list local-only check failed; allowing", exc_info=True
286 )
287 return False
290def coerce_setting_for_write(key: str, value: Any, ui_element: str) -> Any:
291 """Coerce an incoming value to the correct type before writing to the DB.
293 All web routes that save settings should use this function to ensure
294 consistent type conversion.
296 No JSON pre-parsing (``json.loads``) is needed here because:
297 - ``get_typed_setting_value`` already parses JSON strings internally
298 via ``_parse_json_value`` (for ``ui_element="json"``) and
299 ``_parse_multiselect`` (for ``ui_element="multiselect"``).
300 - For JSON API endpoints, ``request.get_json()`` already delivers
301 dicts/lists as native Python objects.
302 - For ``ui_element="text"``, pre-parsing would corrupt data: a JSON
303 string like ``'{"k": "v"}'`` would become a dict, then ``str()``
304 would produce ``"{'k': 'v'}"`` (Python repr, not valid JSON).
305 """
306 # check_env=False: we are persisting a user-supplied value, not reading
307 # from an environment variable override. check_env=True (the default)
308 # would silently replace the user's value with an env var, which is
309 # incorrect on the write path.
310 return get_typed_setting_value(
311 key=key,
312 value=value,
313 ui_element=ui_element,
314 default=None,
315 check_env=False,
316 )
319@settings_bp.route("/", methods=["GET"])
320@login_required
321def settings_page():
322 """Main settings dashboard with links to specialized config pages"""
323 return render_template_with_defaults("settings_dashboard.html")
326@settings_bp.route("/save_all_settings", methods=["POST"])
327@login_required
328@settings_limit
329@require_json_body(
330 error_format="status", error_message="No settings data provided"
331)
332@with_user_session()
333def save_all_settings(
334 db_session: Optional[Session] = None, settings_manager=None
335):
336 """Handle saving all settings at once from the unified settings page"""
337 try:
338 from ...security.data_sanitizer import DataSanitizer
340 # Process JSON data
341 form_data = request.get_json()
342 if not form_data:
343 return (
344 jsonify(
345 {
346 "status": "error",
347 "message": "No settings data provided",
348 }
349 ),
350 400,
351 )
353 # Track validation errors
354 validation_errors = []
355 settings_by_type: dict[str, Any] = {}
357 # Track changes for logging
358 updated_settings = []
359 created_settings = []
361 # Store original values for better messaging
362 original_values = {}
364 # Fetch all settings and remove non-editable keys
365 all_db_settings = _filter_editable_settings(form_data, db_session)
367 # Reject public hostnames being added to the local-hosts allowlist, and
368 # inherently-public engines being added to the trusted-engines list.
369 _egress_err = first_egress_validation_error(form_data, all_db_settings)
370 if _egress_err is not None: 370 ↛ 371line 370 didn't jump to line 371 because the condition on line 370 was never true
371 validation_errors.append(_egress_err)
372 return (
373 jsonify(
374 {
375 "status": "error",
376 "message": "Validation errors",
377 "errors": validation_errors,
378 }
379 ),
380 400,
381 )
383 # Update each setting
384 for key, value in form_data.items():
385 # Skip corrupted keys or empty strings as keys
386 if not key or not isinstance(key, str) or key.strip() == "":
387 continue
389 # Get the setting metadata from pre-fetched dict
390 current_setting = all_db_settings.get(key)
392 # SAFETY NET: an empty string OR the redaction sentinel for a
393 # password-typed setting is a no-op, never a "clear the value"
394 # request. Reasons:
395 # 1. The form templates (Jinja2 + JS-rendered) deliberately
396 # render password inputs empty so the saved value never
397 # enters the HTML source. A user who blurs the field
398 # without typing must not wipe their stored API key.
399 # 2. /settings/api redacts password values to the redaction
400 # sentinel ("[REDACTED]"). A stale browser tab could submit
401 # either "" or the sentinel — both must be idempotent on
402 # the DB, or the round-trip would persist the literal
403 # sentinel over the real secret.
404 # 3. Defense-in-depth against direct cURL or automation
405 # mistakes that POST an empty/sentinel value.
406 # To unset a password setting, clear the source env var or
407 # use settings import.
408 if (
409 current_setting
410 and DataSanitizer.is_sensitive_setting(
411 key, current_setting.ui_element
412 )
413 and isinstance(value, str)
414 and value in ("", DataSanitizer.REDACTION_TEXT)
415 ):
416 logger.debug(f"Skipping empty secret write for {key} (no-op)")
417 continue
419 # EARLY VALIDATION: Convert checkbox values BEFORE any other processing
420 # This prevents incorrect triggering of corrupted value detection
421 if current_setting and current_setting.ui_element == "checkbox":
422 if not isinstance(value, bool):
423 logger.debug(
424 f"Converting checkbox {key} from {type(value).__name__} to bool: {value}"
425 )
426 value = parse_boolean(value)
427 form_data[key] = (
428 value # Update the form_data with converted value
429 )
431 # Store original value for messaging
432 if current_setting:
433 original_values[key] = current_setting.value
435 # Determine setting type and category
436 if key.startswith("llm."):
437 setting_type = SettingType.LLM
438 category = "llm_general"
439 if (
440 "temperature" in key
441 or "max_tokens" in key
442 or "batch" in key
443 or "layers" in key
444 ):
445 category = "llm_parameters"
446 elif key.startswith("search."):
447 setting_type = SettingType.SEARCH
448 category = "search_general"
449 if (
450 "iterations" in key
451 or "results" in key
452 or "region" in key
453 or "questions" in key
454 or "section" in key
455 ):
456 category = "search_parameters"
457 elif key.startswith("report."):
458 setting_type = SettingType.REPORT
459 category = "report_parameters"
460 elif key.startswith("database."):
461 setting_type = SettingType.DATABASE
462 category = "database_parameters"
463 elif key.startswith("app."):
464 setting_type = SettingType.APP
465 category = "app_interface"
466 elif key.startswith("chat."): 466 ↛ 467line 466 didn't jump to line 467 because the condition on line 466 was never true
467 setting_type = SettingType.CHAT
468 category = "chat"
469 else:
470 setting_type = None
471 category = None
473 # Special handling for corrupted or empty values
474 if value == "[object Object]" or (
475 isinstance(value, str)
476 and value.strip() in ["{}", "[]", "{", "["]
477 ):
478 if key.startswith("report."):
479 value = {}
480 else:
481 # Use default or null for other types
482 if key == "llm.model":
483 value = ""
484 elif key == "llm.provider":
485 value = "ollama"
486 elif key == "search.tool":
487 value = DEFAULT_SEARCH_TOOL
488 elif key in ["app.theme", "app.default_theme"]:
489 value = "dark"
490 else:
491 value = None
493 logger.warning(f"Corrected corrupted value for {key}: {value}")
494 # NOTE: No JSON pre-parsing is done here. After the
495 # corruption replacement above, values are Python dicts
496 # (e.g. {}), hardcoded strings, or None — none are JSON
497 # strings that need parsing. Type conversion below via
498 # coerce_setting_for_write() handles everything; that
499 # function delegates to get_typed_setting_value() which
500 # already parses JSON internally for "json" and
501 # "multiselect" ui_elements.
503 if current_setting:
504 # Coerce to correct Python type (e.g. str "5" → int 5
505 # for number settings, str "true" → bool for checkboxes).
506 converted_value = coerce_setting_for_write(
507 key=current_setting.key,
508 value=value,
509 ui_element=current_setting.ui_element,
510 )
512 # Validate the setting
513 is_valid, error_message = validate_setting(
514 current_setting, converted_value
515 )
517 if is_valid:
518 # Save the converted setting using the same session
519 success = set_setting(
520 key, converted_value, db_session=db_session
521 )
522 if success: 522 ↛ 526line 522 didn't jump to line 526 because the condition on line 522 was always true
523 updated_settings.append(key)
525 # Track settings by type for exporting
526 if current_setting.type not in settings_by_type:
527 settings_by_type[current_setting.type] = []
528 settings_by_type[current_setting.type].append(
529 current_setting
530 )
531 else:
532 # Add to validation errors
533 validation_errors.append(
534 {
535 "key": key,
536 "name": current_setting.name,
537 "error": error_message,
538 }
539 )
540 else:
541 # Namespace validation: reject new keys outside allowed prefixes.
542 if not _is_allowed_new_setting_key(key):
543 logger.warning(
544 "Security: Rejected setting outside allowed namespaces: {!r} (user={!r})",
545 key,
546 session["username"],
547 )
548 validation_errors.append(
549 {
550 "key": key,
551 "name": key,
552 "error": _new_key_rejection_reason(key),
553 }
554 )
555 continue
557 # Create a new setting
558 new_setting = {
559 "key": key,
560 "value": value,
561 "type": setting_type.value.lower()
562 if setting_type is not None
563 else "app",
564 "name": key.split(".")[-1].replace("_", " ").title(),
565 "description": f"Setting for {key}",
566 "category": category,
567 "ui_element": "text", # Default UI element
568 }
570 # Determine better UI element based on value type
571 if isinstance(value, bool):
572 new_setting["ui_element"] = "checkbox"
573 elif isinstance(value, (int, float)) and not isinstance(
574 value, bool
575 ):
576 new_setting["ui_element"] = "number"
577 elif isinstance(value, (dict, list)):
578 new_setting["ui_element"] = "textarea"
580 # Create the setting
581 db_setting = create_or_update_setting(
582 new_setting, db_session=db_session
583 )
585 if db_setting:
586 created_settings.append(key)
587 # Track settings by type for exporting
588 if db_setting.type not in settings_by_type: 588 ↛ 590line 588 didn't jump to line 590 because the condition on line 588 was always true
589 settings_by_type[db_setting.type] = []
590 settings_by_type[db_setting.type].append(db_setting)
591 else:
592 validation_errors.append(
593 {
594 "key": key,
595 "name": new_setting["name"],
596 "error": "Failed to create setting",
597 }
598 )
600 # Report validation errors if any
601 if validation_errors:
602 return (
603 jsonify(
604 {
605 "status": "error",
606 "message": "Validation errors",
607 "errors": validation_errors,
608 }
609 ),
610 400,
611 )
613 # Get all settings to return to the client for proper state update
614 all_settings = {}
615 for setting in db_session.query(Setting).all():
616 # Convert enum to string if present
617 setting_type = setting.type
618 if hasattr(setting_type, "value"):
619 setting_type = setting_type.value
621 all_settings[setting.key] = {
622 "value": setting.value,
623 "name": setting.name,
624 "description": setting.description,
625 "type": setting_type,
626 "category": setting.category,
627 "ui_element": setting.ui_element,
628 "editable": setting.editable,
629 "options": setting.options,
630 "visible": setting.visible,
631 "min_value": setting.min_value,
632 "max_value": setting.max_value,
633 "step": setting.step,
634 }
636 # Customize the success message based on what changed
637 success_message = ""
638 if len(updated_settings) == 1:
639 # For a single update, provide more specific info about what changed
640 key = updated_settings[0]
641 # Reuse the already-fetched setting from our pre-fetched dict
642 updated_setting = all_db_settings.get(key)
643 name = (
644 updated_setting.name
645 if updated_setting
646 else key.split(".")[-1].replace("_", " ").title()
647 )
649 # Format the message
650 if key in original_values: 650 ↛ 664line 650 didn't jump to line 664 because the condition on line 650 was always true
651 new_value = updated_setting.value if updated_setting else None
653 # If it's a boolean, use "enabled/disabled" language
654 if isinstance(new_value, bool):
655 state = "enabled" if new_value else "disabled"
656 success_message = f"{name} {state}"
657 else:
658 # For non-boolean values
659 if isinstance(new_value, (dict, list)):
660 success_message = f"{name} updated"
661 else:
662 success_message = f"{name} updated"
663 else:
664 success_message = f"{name} updated"
665 else:
666 # Multiple settings or generic message
667 success_message = f"Settings saved successfully ({len(updated_settings)} updated, {len(created_settings)} created)"
669 # Check if any warning-affecting settings were changed and include warnings.
670 # Redact secret values in the echoed settings so a POST response never
671 # ships plaintext API keys back to the browser — matching the
672 # redaction the GET /settings/api endpoint already applies.
673 response_data = {
674 "status": "success",
675 "message": success_message,
676 "updated": updated_settings,
677 "created": created_settings,
678 "settings": DataSanitizer.redact_settings_snapshot(all_settings),
679 }
681 warning_affecting_keys = WARNING_AFFECTING_KEYS
683 # Check if any warning-affecting settings were changed
684 if any(
685 key in warning_affecting_keys
686 for key in updated_settings + created_settings
687 ):
688 warnings = calculate_warnings()
689 response_data["warnings"] = warnings
690 logger.info(
691 f"Bulk settings update affected warning keys, calculated {len(warnings)} warnings"
692 )
694 invalidate_settings_caches(session["username"])
695 reschedule_document_jobs_if_needed(
696 session["username"], updated_settings + created_settings
697 )
698 reschedule_zotero_jobs_if_needed(
699 session["username"], updated_settings + created_settings
700 )
701 return jsonify(response_data)
703 except Exception:
704 logger.exception("Error saving settings")
705 return (
706 jsonify(
707 {
708 "status": "error",
709 "message": "An internal error occurred while saving settings.",
710 }
711 ),
712 500,
713 )
716@settings_bp.route("/reset_to_defaults", methods=["POST"])
717@login_required
718@settings_limit
719@with_user_session()
720def reset_to_defaults(
721 db_session: Optional[Session] = None, settings_manager=None
722):
723 """Reset all settings to their default values"""
724 try:
725 settings_manager.load_from_defaults_file()
727 logger.info("Successfully imported settings from default files")
729 except Exception:
730 logger.exception("Error importing default settings")
731 return jsonify(
732 {
733 "status": "error",
734 "message": "Failed to reset settings to defaults",
735 }
736 ), 500
738 invalidate_settings_caches(session["username"])
739 return jsonify(
740 {
741 "status": "success",
742 "message": "All settings have been reset to default values",
743 }
744 )
747@settings_bp.route("/save_settings", methods=["POST"])
748@login_required
749@settings_limit
750@with_user_session()
751def save_settings(db_session: Optional[Session] = None, settings_manager=None):
752 """Save all settings from the form using POST method - fallback when JavaScript is disabled"""
753 try:
754 from ...security.data_sanitizer import DataSanitizer
756 # Get form data
757 form_data = request.form.to_dict()
759 # Remove CSRF token from the data
760 form_data.pop("csrf_token", None)
762 updated_count = 0
763 failed_count = 0
764 rejected_count = 0
766 # Fetch all settings and remove non-editable keys
767 all_db_settings = _filter_editable_settings(form_data, db_session)
769 # Egress-policy validators — the JSON route (save_all_settings) runs
770 # these; the POST fallback must too, or a JS-disabled client could
771 # whitelist a public hostname as "local", which the JSON route does
772 # not permit.
773 _policy_err = first_egress_validation_error(form_data, all_db_settings)
774 if _policy_err is not None: 774 ↛ 775line 774 didn't jump to line 775 because the condition on line 774 was never true
775 flash(_policy_err.get("error", "Invalid policy setting"), "error")
776 return redirect(url_for("settings.settings_page"))
778 # Process each setting
779 for key, value in form_data.items():
780 try:
781 # Get the setting from pre-fetched dict
782 db_setting = all_db_settings.get(key)
784 # Namespace validation: reject new keys outside allowed prefixes.
785 # Existing keys (updates) bypass this check — it only applies
786 # to creation of brand-new rows through this form-POST route.
787 if db_setting is None and not _is_allowed_new_setting_key(key):
788 logger.warning(
789 "Security: Rejected setting outside allowed namespaces: {!r} (user={!r})",
790 key,
791 session["username"],
792 )
793 rejected_count += 1
794 continue
796 # SAFETY NET: empty string OR the redaction sentinel for a
797 # password-typed setting is a no-op — never "clear my key".
798 # The no-JS form renders password inputs empty (and GET
799 # redacts them to the sentinel), so a plain form submit
800 # must not wipe the stored secret. Matches the guards in
801 # save_all_settings + api_update_setting.
802 if (
803 db_setting
804 and DataSanitizer.is_sensitive_setting(
805 key, db_setting.ui_element
806 )
807 and isinstance(value, str)
808 and value in ("", DataSanitizer.REDACTION_TEXT)
809 ):
810 logger.debug(
811 f"Skipping empty secret write for {key} via "
812 "save_settings (no-op)"
813 )
814 continue
816 # Coerce form POST string to correct Python type.
817 if db_setting:
818 value = coerce_setting_for_write(
819 key=db_setting.key,
820 value=value,
821 ui_element=db_setting.ui_element,
822 )
824 # Save the setting
825 if settings_manager.set_setting(key, value, commit=False):
826 updated_count += 1
827 else:
828 failed_count += 1
829 logger.warning(f"Failed to save setting {key}")
831 except Exception:
832 logger.exception(f"Error saving setting {key}")
833 failed_count += 1
835 # Commit all changes at once
836 try:
837 db_session.commit()
839 flash(
840 f"Settings saved successfully! Updated {updated_count} settings.",
841 "success",
842 )
843 if failed_count > 0:
844 flash(
845 f"Warning: {failed_count} settings failed to save.",
846 "warning",
847 )
848 if rejected_count > 0:
849 flash(
850 f"Rejected {rejected_count} settings (unknown namespace). "
851 "This may indicate a bug or an attempted injection.",
852 "error",
853 )
854 invalidate_settings_caches(session["username"])
856 except Exception:
857 db_session.rollback()
858 logger.exception("Failed to commit settings")
859 flash("Error saving settings. Please try again.", "error")
861 return redirect(url_for("settings.settings_page"))
863 except Exception:
864 logger.exception("Error in save_settings")
865 flash("An internal error occurred while saving settings.", "error")
866 return redirect(url_for("settings.settings_page"))
869# API Routes
870@settings_bp.route("/api", methods=["GET"])
871@login_required
872@with_user_session()
873def api_get_all_settings(
874 db_session: Optional[Session] = None, settings_manager=None
875):
876 """Get all settings.
878 The response is redacted via ``DataSanitizer.redact_settings_snapshot``
879 so password-typed values (API keys, OAuth tokens, etc.) come back as
880 ``"[REDACTED]"`` instead of plaintext. The previous implementation
881 leaked env-overridden API keys to anyone who could authenticate as a
882 user — even though the values render unmasked into the server-side
883 settings form too (a separate template-level concern), the JSON API
884 is its own surface area: it gets cached by clients, logged by
885 proxies, and copy/pasted into bug reports. Redaction here is
886 defense-in-depth.
888 No JS caller round-trips this endpoint's values back into a form
889 (the settings form is server-rendered from the template, which uses
890 the model directly), so redacting the response does not break the UI.
891 """
892 try:
893 from ...security.data_sanitizer import DataSanitizer
895 # Get query parameters
896 category = request.args.get("category")
898 # Get settings (nested-with-metadata shape:
899 # {key: {value, ui_element, type, ...}})
900 settings = settings_manager.get_all_settings()
902 # Filter by category if requested
903 if category:
904 # Need to get all setting details to check category
905 db_settings = db_session.query(Setting).all()
906 category_keys = [
907 s.key for s in db_settings if s.category == category
908 ]
910 # Filter settings by keys
911 settings = {
912 key: value
913 for key, value in settings.items()
914 if key in category_keys
915 }
917 settings = DataSanitizer.redact_settings_snapshot(settings)
919 return jsonify({"status": "success", "settings": settings})
920 except Exception:
921 logger.exception("Error getting settings")
922 return jsonify({"error": "Failed to retrieve settings"}), 500
925@settings_bp.route("/api/<path:key>", methods=["GET"])
926@login_required
927@with_user_session()
928def api_get_db_setting(
929 key, db_session: Optional[Session] = None, settings_manager=None
930):
931 """Get a specific setting by key from DB, falling back to defaults.
933 Password-typed values are redacted in the response so this endpoint
934 matches the redaction contract enforced on ``/settings/api`` (the bulk
935 GET) and ``/settings/api/bulk``. The same threat applies: any
936 authenticated caller could otherwise grab a plaintext API key with
937 a single GET. Metadata fields (``ui_element``, ``type``, ...) stay
938 intact so the front-end can still render the correct input control.
939 """
940 from ...security.data_sanitizer import DataSanitizer
942 try:
943 # Get setting from database using the same session
944 db_setting = (
945 db_session.query(Setting).filter(Setting.key == key).first()
946 )
948 if db_setting:
949 # Return full setting details from DB
950 setting_data = {
951 "key": db_setting.key,
952 "value": DataSanitizer.redact_value(
953 db_setting.key, db_setting.ui_element, db_setting.value
954 ),
955 "type": db_setting.type
956 if isinstance(db_setting.type, str)
957 else db_setting.type.value,
958 "name": db_setting.name,
959 "description": db_setting.description,
960 "category": db_setting.category,
961 "ui_element": db_setting.ui_element,
962 "options": db_setting.options,
963 "min_value": db_setting.min_value,
964 "max_value": db_setting.max_value,
965 "step": db_setting.step,
966 "visible": db_setting.visible,
967 "editable": db_setting.editable,
968 }
969 return jsonify(setting_data)
971 # Not in DB — check defaults so this endpoint is consistent
972 # with GET /settings/api which includes default settings
973 default_meta = settings_manager.default_settings.get(key)
974 if default_meta: 974 ↛ 975line 974 didn't jump to line 975 because the condition on line 974 was never true
975 default_ui = default_meta.get("ui_element", "text")
976 setting_data = {
977 "key": key,
978 "value": DataSanitizer.redact_value(
979 key, default_ui, default_meta.get("value")
980 ),
981 "type": default_meta.get("type", "APP"),
982 "name": default_meta.get("name", key),
983 "description": default_meta.get("description"),
984 "category": default_meta.get("category"),
985 "ui_element": default_ui,
986 "options": default_meta.get("options"),
987 "min_value": default_meta.get("min_value"),
988 "max_value": default_meta.get("max_value"),
989 "step": default_meta.get("step"),
990 "visible": default_meta.get("visible", True),
991 "editable": default_meta.get("editable", True),
992 }
993 return jsonify(setting_data)
995 return jsonify({"error": f"Setting not found: {key}"}), 404
996 except Exception:
997 logger.exception(f"Error getting setting {key}")
998 return jsonify({"error": "Failed to retrieve settings"}), 500
1001@settings_bp.route("/api/<path:key>", methods=["PUT"])
1002@login_required
1003@settings_limit
1004@require_json_body(error_message="No data provided")
1005@with_user_session(include_settings_manager=False)
1006def api_update_setting(key, db_session: Optional[Session] = None):
1007 """Update a setting"""
1008 try:
1009 from ...security.data_sanitizer import DataSanitizer
1011 # Get request data
1012 data = request.get_json()
1013 value = data.get("value")
1014 if value is None:
1015 return jsonify({"error": "No value provided"}), 400
1017 # Check if setting exists
1018 db_setting = (
1019 db_session.query(Setting).filter(Setting.key == key).first()
1020 )
1022 if db_setting:
1023 # Check if setting is editable
1024 if not db_setting.editable:
1025 return jsonify({"error": f"Setting {key} is not editable"}), 403
1027 # SAFETY NET: an empty string OR the redaction sentinel on a
1028 # password-typed setting is a no-op (never "clear my key").
1029 # Companion guard to the same check in save_all_settings — see
1030 # the longer rationale there. Returning 200 with a message keeps
1031 # the endpoint idempotent so client-side save indicators don't
1032 # error.
1033 if (
1034 DataSanitizer.is_sensitive_setting(key, db_setting.ui_element)
1035 and isinstance(value, str)
1036 and value in ("", DataSanitizer.REDACTION_TEXT)
1037 ):
1038 logger.debug(
1039 f"Skipping empty secret write for {key} via "
1040 "api_update_setting (no-op)"
1041 )
1042 return jsonify(
1043 {
1044 "message": (
1045 f"Setting {key} unchanged (empty password ignored)"
1046 )
1047 }
1048 ), 200
1050 # Coerce to correct Python type before saving.
1051 # Without this, values from JSON API requests are stored
1052 # as-is (e.g. string "5" instead of int 5 for number
1053 # settings, string "true" instead of bool for checkboxes).
1054 value = coerce_setting_for_write(
1055 key=db_setting.key,
1056 value=value,
1057 ui_element=db_setting.ui_element,
1058 )
1060 # Validate the setting (matches save_all_settings pattern)
1061 is_valid, error_message = validate_setting(db_setting, value)
1062 if not is_valid:
1063 logger.warning(
1064 f"Validation failed for setting {key}: {error_message}"
1065 )
1066 return jsonify(
1067 {"error": f"Invalid value for setting {key}"}
1068 ), 400
1070 # Cross-field egress-policy validation. The full-form saves
1071 # (save_all_settings / save_settings) run these guards, but this
1072 # single-key PUT endpoint did not — letting a client smuggle a
1073 # public hostname into llm.allowed_local_hostnames (which the host
1074 # classifier then trusts as local), one key at a time. Run the
1075 # same validators here for the keys they govern.
1076 if key in (
1077 "policy.egress_scope",
1078 "search.tool",
1079 "llm.allowed_local_hostnames",
1080 "policy.trusted_search_engines",
1081 ):
1082 _all_db_settings = {
1083 s.key: s for s in db_session.query(Setting).all()
1084 }
1085 _err = first_egress_validation_error(
1086 {key: value}, _all_db_settings
1087 )
1088 if _err is not None:
1089 logger.bind(policy_audit=True).warning(
1090 "egress-policy setting rejected at api_update_setting",
1091 key=key,
1092 reason=_err.get("error"),
1093 )
1094 return jsonify({"error": _err["error"]}), 400
1096 # Update setting
1097 # Pass the db_session to avoid session lookup issues
1098 success = set_setting(key, value, db_session=db_session)
1099 if success:
1100 response_data: dict[str, Any] = {
1101 "message": f"Setting {key} updated successfully"
1102 }
1104 # If this is a key that affects warnings, include warning calculations
1105 warning_affecting_keys = WARNING_AFFECTING_KEYS
1107 if key in warning_affecting_keys:
1108 warnings = calculate_warnings()
1109 response_data["warnings"] = warnings
1110 logger.debug(
1111 f"Setting {key} changed to {value}, calculated {len(warnings)} warnings"
1112 )
1114 invalidate_settings_caches(session["username"])
1115 # A document_scheduler.* toggle (e.g. sweep_library_collections
1116 # or generate_rag) must take effect without a re-login.
1117 reschedule_document_jobs_if_needed(session["username"], [key])
1118 reschedule_zotero_jobs_if_needed(session["username"], [key])
1119 return jsonify(response_data)
1120 return jsonify({"error": f"Failed to update setting {key}"}), 500
1122 # Namespace validation: reject new keys outside allowed prefixes.
1123 if not _is_allowed_new_setting_key(key):
1124 logger.warning(
1125 "Security: Rejected setting outside allowed namespaces: {!r} (user={!r})",
1126 key,
1127 session["username"],
1128 )
1129 return jsonify({"error": _new_key_rejection_reason(key)}), 400
1131 # Create new setting with default metadata
1132 setting_dict = {
1133 "key": key,
1134 "value": value,
1135 "name": key.split(".")[-1].replace("_", " ").title(),
1136 "description": f"Setting for {key}",
1137 }
1139 # Add additional metadata if provided.
1140 # 'visible' and 'editable' are system-controlled — not accepted from callers.
1141 for field in [
1142 "type",
1143 "name",
1144 "description",
1145 "category",
1146 "ui_element",
1147 "options",
1148 "min_value",
1149 "max_value",
1150 "step",
1151 ]:
1152 if field in data:
1153 setting_dict[field] = data[field]
1155 # Create setting
1156 db_setting = create_or_update_setting(
1157 setting_dict, db_session=db_session
1158 )
1160 if db_setting:
1161 invalidate_settings_caches(session["username"])
1162 reschedule_document_jobs_if_needed(session["username"], [key])
1163 reschedule_zotero_jobs_if_needed(session["username"], [key])
1164 return (
1165 jsonify(
1166 {
1167 "message": f"Setting {key} created successfully",
1168 "setting": {
1169 "key": db_setting.key,
1170 # Don't echo a freshly-created password value
1171 # back in plaintext — redact like every other
1172 # response that ships settings to the browser.
1173 "value": (
1174 DataSanitizer.REDACTION_TEXT
1175 if DataSanitizer.is_sensitive_setting(
1176 key, db_setting.ui_element
1177 )
1178 else db_setting.value
1179 ),
1180 "type": db_setting.type.value,
1181 "name": db_setting.name,
1182 },
1183 }
1184 ),
1185 201,
1186 )
1187 return jsonify({"error": f"Failed to create setting {key}"}), 500
1188 except Exception:
1189 logger.exception(f"Error updating setting {key}")
1190 return jsonify({"error": "Failed to update setting"}), 500
1193@settings_bp.route("/api/<path:key>", methods=["DELETE"])
1194@login_required
1195@with_user_session()
1196def api_delete_setting(
1197 key, db_session: Optional[Session] = None, settings_manager=None
1198):
1199 """Delete a setting"""
1200 try:
1201 # Check if setting exists
1202 db_setting = (
1203 db_session.query(Setting).filter(Setting.key == key).first()
1204 )
1205 if not db_setting:
1206 return jsonify({"error": f"Setting not found: {key}"}), 404
1208 # Check if setting is editable
1209 if not db_setting.editable:
1210 return jsonify({"error": f"Setting {key} is not editable"}), 403
1212 # Delete setting
1213 success = settings_manager.delete_setting(key)
1214 if success:
1215 invalidate_settings_caches(session["username"])
1216 return jsonify({"message": f"Setting {key} deleted successfully"})
1217 return jsonify({"error": f"Failed to delete setting {key}"}), 500
1218 except Exception:
1219 logger.exception(f"Error deleting setting {key}")
1220 return jsonify({"error": "Failed to delete setting"}), 500
1223@settings_bp.route("/api/import", methods=["POST"])
1224@login_required
1225@settings_limit
1226@with_user_session()
1227def api_import_settings(
1228 db_session: Optional[Session] = None, settings_manager=None
1229):
1230 """Import settings from defaults file"""
1231 try:
1232 settings_manager.load_from_defaults_file()
1234 invalidate_settings_caches(session["username"])
1235 return jsonify({"message": "Settings imported successfully"})
1236 except Exception:
1237 logger.exception("Error importing settings")
1238 return jsonify({"error": "Failed to import settings"}), 500
1241@settings_bp.route("/api/categories", methods=["GET"])
1242@login_required
1243@with_user_session(include_settings_manager=False)
1244def api_get_categories(db_session: Optional[Session] = None):
1245 """Get all setting categories"""
1246 try:
1247 # Get all distinct categories
1248 categories = db_session.query(Setting.category).distinct().all()
1249 category_list = [c[0] for c in categories if c[0] is not None]
1251 return jsonify({"categories": category_list})
1252 except Exception:
1253 logger.exception("Error getting categories")
1254 return jsonify({"error": "Failed to retrieve settings"}), 500
1257@settings_bp.route("/api/types", methods=["GET"])
1258@login_required
1259def api_get_types():
1260 """Get all setting types"""
1261 try:
1262 # Get all setting types
1263 types = [t.value for t in SettingType]
1264 return jsonify({"types": types})
1265 except Exception:
1266 logger.exception("Error getting types")
1267 return jsonify({"error": "Failed to retrieve settings"}), 500
1270@settings_bp.route("/api/ui_elements", methods=["GET"])
1271@login_required
1272def api_get_ui_elements():
1273 """Get all UI element types"""
1274 try:
1275 # Define supported UI element types
1276 ui_elements = [
1277 "text",
1278 "select",
1279 "checkbox",
1280 "slider",
1281 "number",
1282 "textarea",
1283 "color",
1284 "date",
1285 "file",
1286 "password",
1287 ]
1289 return jsonify({"ui_elements": ui_elements})
1290 except Exception:
1291 logger.exception("Error getting UI elements")
1292 return jsonify({"error": "Failed to retrieve settings"}), 500
1295@settings_bp.route("/api/available-models", methods=["GET"])
1296@login_required
1297def api_get_available_models():
1298 """Get available LLM models from various providers"""
1299 endpoint_start = time.perf_counter()
1300 try:
1301 from ...database.models import ProviderModel
1303 # Check if force_refresh is requested
1304 force_refresh = parse_bool_arg("force_refresh")
1306 # Get all auto-discovered providers (show all so users can discover
1307 # and configure providers they haven't set up yet)
1308 from ...llm.providers import get_discovered_provider_options
1310 # LlamaCppProvider is auto-discovered (provider_name="llama.cpp"), so
1311 # it is already included here. The previous hardcoded LLAMACPP append
1312 # produced a duplicate entry in the dropdown.
1313 provider_options = get_discovered_provider_options()
1315 # Available models by provider
1316 providers: dict[str, Any] = {}
1318 # Check database cache first (unless force_refresh is True)
1319 if not force_refresh:
1320 try:
1321 # Define cache expiration (24 hours)
1322 cache_expiry = datetime.now(UTC) - timedelta(hours=24)
1324 # Get cached models from database
1325 username = session["username"]
1326 with get_user_db_session(username) as db_session:
1327 cached_models = (
1328 db_session.query(ProviderModel)
1329 .filter(ProviderModel.last_updated > cache_expiry)
1330 .all()
1331 )
1333 if cached_models: 1333 ↛ 1334line 1333 didn't jump to line 1334 because the condition on line 1333 was never true
1334 logger.info(
1335 f"Found {len(cached_models)} cached models in database"
1336 )
1338 # Group models by provider
1339 for model in cached_models:
1340 provider_key = (
1341 f"{normalize_provider(model.provider)}_models"
1342 )
1343 if provider_key not in providers:
1344 providers[provider_key] = []
1346 providers[provider_key].append(
1347 {
1348 "value": model.model_key,
1349 "label": model.model_label,
1350 "provider": model.provider.upper(),
1351 }
1352 )
1354 # If we have cached data for all providers, return it
1355 if providers:
1356 _log_available_models_duration(
1357 endpoint_start, cache_hit=True
1358 )
1359 logger.info("Returning cached models from database")
1360 return jsonify(
1361 {
1362 "provider_options": provider_options,
1363 "providers": providers,
1364 }
1365 )
1367 except Exception:
1368 logger.warning("Error reading cached models from database")
1369 # Continue to fetch fresh data
1371 # Ollama / OpenAI / Anthropic model listing is handled by the
1372 # auto-discovery loop below (their provider classes' list_models_for_api),
1373 # which is the single fetch path. The previous hand-rolled per-provider
1374 # blocks here were dead in normal mode (overwritten by discovery) and an
1375 # inferior duplicate (no SSRF validation, no auth headers); local-only mode
1376 # now keeps the local providers via the LOCAL_PROVIDERS filter below.
1378 # Fetch models from auto-discovered providers
1379 from ...llm.providers import discover_providers
1381 discovered_providers = discover_providers()
1383 # Egress policy: when the effective posture is local-only the user
1384 # has opted into local-only inference, so don't list cloud providers
1385 # (OpenRouter, Google, XAI, IonOS, OpenAI, Anthropic, ...). We still
1386 # list the known-local providers (ollama/llamacpp/lmstudio) via their
1387 # provider classes — which validate the URL (SSRF) and support auth
1388 # headers — by filtering the discovered set to LOCAL_PROVIDERS rather
1389 # than skipping discovery entirely. Use the scope-aware helper so a
1390 # PRIVATE_ONLY / adaptive-private user is covered even if the raw
1391 # toggle is False.
1392 require_local_for_discovered = _model_list_local_only()
1393 if require_local_for_discovered:
1394 from ...llm.providers._helpers import LOCAL_PROVIDERS
1396 local_only_providers = {
1397 key: info
1398 for key, info in discovered_providers.items()
1399 if normalize_provider(key) in LOCAL_PROVIDERS
1400 }
1401 logger.bind(policy_audit=True).info(
1402 "local-only egress posture: limiting discovered model lists "
1403 "to local providers",
1404 kept=list(local_only_providers.keys()),
1405 skipped=[
1406 key
1407 for key in discovered_providers
1408 if key not in local_only_providers
1409 ],
1410 )
1411 discovered_providers = local_only_providers
1413 for provider_key, provider_info in discovered_providers.items():
1414 provider_models = []
1415 try:
1416 logger.info(
1417 f"Fetching models from {provider_info.provider_name}"
1418 )
1420 # Get the provider class
1421 provider_class = provider_info.provider_class
1423 # Get API key if configured
1424 api_key = _get_setting_from_session(
1425 provider_class.api_key_setting, ""
1426 )
1428 # Get base URL if provider has configurable URL
1429 provider_base_url: str | None = None
1430 if (
1431 hasattr(provider_class, "url_setting")
1432 and provider_class.url_setting
1433 ):
1434 provider_base_url = _get_setting_from_session(
1435 provider_class.url_setting, ""
1436 )
1438 # Use the provider's list_models_for_api method
1439 models = provider_class.list_models_for_api(
1440 api_key, provider_base_url
1441 )
1443 # Format models for the API response
1444 for model in models:
1445 provider_models.append(
1446 {
1447 "value": model["value"],
1448 "label": model[
1449 "label"
1450 ], # Use provider's label as-is
1451 "provider": provider_key,
1452 }
1453 )
1455 logger.info(
1456 f"Successfully fetched {len(provider_models)} models from {provider_info.provider_name}"
1457 )
1459 except Exception:
1460 logger.exception(
1461 f"Error getting {provider_info.provider_name} models"
1462 )
1464 # Set models in providers dict using lowercase key
1465 providers[f"{normalize_provider(provider_key)}_models"] = (
1466 provider_models
1467 )
1468 logger.info(
1469 f"Final {provider_key} models count: {len(provider_models)}"
1470 )
1472 # Save fetched models to database cache
1473 if force_refresh or providers: 1473 ↛ 1522line 1473 didn't jump to line 1522 because the condition on line 1473 was always true
1474 # We fetched fresh data, save it to database
1475 username = session["username"]
1476 with get_user_db_session(username) as db_session:
1477 try:
1478 if force_refresh:
1479 # When force refresh, clear ALL cached models to remove any stale data
1480 # from old code versions or deleted providers
1481 deleted_count = db_session.query(ProviderModel).delete()
1482 logger.info(
1483 f"Force refresh: cleared all {deleted_count} cached models"
1484 )
1485 else:
1486 # Clear old cache entries only for providers we're updating
1487 for provider_key in providers:
1488 provider_name = provider_key.replace(
1489 "_models", ""
1490 ).upper()
1491 db_session.query(ProviderModel).filter(
1492 ProviderModel.provider == provider_name
1493 ).delete()
1495 # Insert new models
1496 for provider_key, models in providers.items():
1497 provider_name = provider_key.replace(
1498 "_models", ""
1499 ).upper()
1500 for model in models:
1501 if ( 1501 ↛ 1500line 1501 didn't jump to line 1500 because the condition on line 1501 was always true
1502 isinstance(model, dict)
1503 and "value" in model
1504 and "label" in model
1505 ):
1506 new_model = ProviderModel(
1507 provider=provider_name,
1508 model_key=model["value"],
1509 model_label=model["label"],
1510 last_updated=datetime.now(UTC),
1511 )
1512 db_session.add(new_model)
1514 db_session.commit()
1515 logger.info("Successfully cached models to database")
1517 except Exception:
1518 logger.exception("Error saving models to database cache")
1519 db_session.rollback()
1521 # Return all options
1522 _log_available_models_duration(endpoint_start, cache_hit=False)
1523 return jsonify(
1524 {"provider_options": provider_options, "providers": providers}
1525 )
1527 except Exception:
1528 logger.exception("Error getting available models")
1529 _log_available_models_duration(
1530 endpoint_start, cache_hit=False, error=True
1531 )
1532 return jsonify(
1533 {
1534 "status": "error",
1535 "message": "Failed to retrieve available models",
1536 }
1537 ), 500
1540def _log_available_models_duration(
1541 start: float, cache_hit: bool, error: bool = False
1542) -> None:
1543 """Log /api/available-models endpoint duration.
1545 Uses INFO when the endpoint took > 1s (indicating a real provider fetch
1546 latency worth flagging), DEBUG otherwise. This is the likely culprit for
1547 Path C (LLM provider timeout masquerading as backend hang).
1548 """
1549 elapsed_ms = (time.perf_counter() - start) * 1000
1550 path = (
1551 "error"
1552 if error
1553 else ("cache hit" if cache_hit else "full provider fetch")
1554 )
1555 if elapsed_ms > 1000:
1556 logger.info(f"/api/available-models ({path}) took {elapsed_ms:.0f}ms")
1557 else:
1558 logger.debug(f"/api/available-models ({path}) took {elapsed_ms:.0f}ms")
1561def _get_engine_icon_and_category(
1562 engine_data: dict, engine_class=None
1563) -> tuple:
1564 """
1565 Get icon emoji and category label for a search engine based on its attributes.
1567 Args:
1568 engine_data: Engine configuration dictionary
1569 engine_class: Optional loaded engine class to check attributes
1571 Returns:
1572 Tuple of (icon, category) strings
1573 """
1574 # Check attributes from either the class or the engine data
1575 if engine_class:
1576 is_scientific = getattr(engine_class, "is_scientific", False)
1577 is_generic = getattr(engine_class, "is_generic", False)
1578 is_local = getattr(engine_class, "is_local", False)
1579 is_news = getattr(engine_class, "is_news", False)
1580 is_code = getattr(engine_class, "is_code", False)
1581 else:
1582 is_scientific = engine_data.get("is_scientific", False)
1583 is_generic = engine_data.get("is_generic", False)
1584 is_local = engine_data.get("is_local", False)
1585 is_news = engine_data.get("is_news", False)
1586 is_code = engine_data.get("is_code", False)
1588 # Check books attribute
1589 if engine_class:
1590 is_books = getattr(engine_class, "is_books", False)
1591 else:
1592 is_books = engine_data.get("is_books", False)
1594 # Return icon and category based on engine type
1595 # Priority: local > scientific > news > code > books > generic > default
1596 if is_local:
1597 return "📁", "Local RAG"
1598 if is_scientific:
1599 return "🔬", "Scientific"
1600 if is_news:
1601 return "📰", "News"
1602 if is_code:
1603 return "💻", "Code"
1604 if is_books:
1605 return "📚", "Books"
1606 if is_generic:
1607 return "🌐", "Web Search"
1608 return "🔍", "Search"
1611@settings_bp.route("/api/available-search-engines", methods=["GET"])
1612@login_required
1613@with_user_session()
1614def api_get_available_search_engines(
1615 db_session: Optional[Session] = None, settings_manager=None
1616):
1617 """Get available search engines"""
1618 try:
1619 # Get search engines using the same approach as search_engines_config.py
1620 from ...web_search_engines.search_engines_config import search_config
1621 from ...web_search_engines.engine_groups import (
1622 classify_engine_group,
1623 effective_group,
1624 group_label,
1625 group_order,
1626 )
1628 username = session["username"]
1629 search_engines = search_config(username=username, db_session=db_session)
1631 # Get user's favorites using SettingsManager
1632 favorites = settings_manager.get_setting("search.favorites", [])
1633 if not isinstance(favorites, list): 1633 ↛ 1634line 1633 didn't jump to line 1634 because the condition on line 1633 was never true
1634 favorites = []
1636 # Extract search engines from config
1637 engines_dict = {}
1638 engine_options = []
1640 if search_engines: 1640 ↛ 1723line 1640 didn't jump to line 1723 because the condition on line 1640 was always true
1641 # Format engines for API response with metadata
1642 from ...security.module_whitelist import (
1643 get_safe_module_class,
1644 SecurityError,
1645 )
1647 for engine_id, engine_data in search_engines.items():
1648 # Try to load the engine class to get metadata
1649 engine_class = None
1650 try:
1651 module_path = engine_data.get("module_path")
1652 class_name = engine_data.get("class_name")
1653 if module_path and class_name: 1653 ↛ 1668line 1653 didn't jump to line 1668 because the condition on line 1653 was always true
1654 # Use secure whitelist-validated import
1655 engine_class = get_safe_module_class(
1656 module_path, class_name
1657 )
1658 except SecurityError:
1659 logger.warning(
1660 f"Security: Blocked unsafe module for {engine_id}"
1661 )
1662 except Exception as e:
1663 logger.debug(
1664 f"Could not load engine class for {engine_id}: {e}"
1665 )
1667 # Get icon and category from engine attributes
1668 icon, category = _get_engine_icon_and_category(
1669 engine_data, engine_class
1670 )
1672 # Check if engine requires an API key
1673 requires_api_key = engine_data.get("requires_api_key", False)
1675 # Build display name with icon, category, and API key status
1676 base_name = engine_data.get("display_name", engine_id)
1677 if requires_api_key:
1678 label = f"{icon} {base_name} ({category}, API key)"
1679 else:
1680 label = f"{icon} {base_name} ({category}, Free)"
1682 # Check if engine is a favorite
1683 is_favorite = engine_id in favorites
1685 # Classify the engine into a selector band. base_group ignores
1686 # favorite status (used by the frontend to move an engine back
1687 # to its category when un-starred); the effective band is the
1688 # Favorites overlay when starred. See engine_groups.py.
1689 base_group = classify_engine_group(
1690 engine_id, category, requires_api_key
1691 )
1692 shown_group = effective_group(base_group, is_favorite)
1694 engines_dict[engine_id] = {
1695 "display_name": base_name,
1696 "description": engine_data.get("description", ""),
1697 "strengths": engine_data.get("strengths", []),
1698 "icon": icon,
1699 "category": category,
1700 "requires_api_key": requires_api_key,
1701 "is_favorite": is_favorite,
1702 }
1704 engine_options.append(
1705 {
1706 "value": engine_id,
1707 "label": label,
1708 "icon": icon,
1709 "category": category,
1710 "requires_api_key": requires_api_key,
1711 "is_favorite": is_favorite,
1712 "group": shown_group,
1713 "group_label": group_label(shown_group),
1714 "group_order": group_order(shown_group),
1715 "base_group": base_group,
1716 "base_group_label": group_label(base_group),
1717 "base_group_order": group_order(base_group),
1718 }
1719 )
1721 # Sort engine_options by band order (favorites band first), then
1722 # alphabetically by label within each band.
1723 engine_options.sort(
1724 key=lambda x: (
1725 x.get("group_order", 999),
1726 x.get("label", "").lower(),
1727 )
1728 )
1730 # If no engines found, log the issue but return empty list
1731 if not engine_options: 1731 ↛ 1732line 1731 didn't jump to line 1732 because the condition on line 1731 was never true
1732 logger.warning("No search engines found in configuration")
1734 return jsonify(
1735 {
1736 "engines": engines_dict,
1737 "engine_options": engine_options,
1738 "favorites": favorites,
1739 }
1740 )
1742 except Exception:
1743 logger.exception("Error getting available search engines")
1744 return jsonify({"error": "Failed to retrieve search engines"}), 500
1747@settings_bp.route("/api/search-favorites", methods=["GET"])
1748@login_required
1749@with_user_session()
1750def api_get_search_favorites(
1751 db_session: Optional[Session] = None, settings_manager=None
1752):
1753 """Get the list of favorite search engines for the current user"""
1754 try:
1755 favorites = settings_manager.get_setting("search.favorites", [])
1756 if not isinstance(favorites, list):
1757 favorites = []
1758 return jsonify({"favorites": favorites})
1760 except Exception:
1761 logger.exception("Error getting search favorites")
1762 return jsonify({"error": "Failed to retrieve favorites"}), 500
1765@settings_bp.route("/api/search-favorites", methods=["PUT"])
1766@login_required
1767@require_json_body(error_message="No data provided")
1768@with_user_session()
1769def api_update_search_favorites(
1770 db_session: Optional[Session] = None, settings_manager=None
1771):
1772 """Update the list of favorite search engines for the current user"""
1773 try:
1774 data = request.get_json()
1775 favorites = data.get("favorites")
1776 if favorites is None:
1777 return jsonify({"error": "No favorites provided"}), 400
1779 if not isinstance(favorites, list):
1780 return jsonify({"error": "Favorites must be a list"}), 400
1782 if settings_manager.set_setting("search.favorites", favorites):
1783 invalidate_settings_caches(session["username"])
1784 return jsonify(
1785 {
1786 "message": "Favorites updated successfully",
1787 "favorites": favorites,
1788 }
1789 )
1790 return jsonify({"error": "Failed to update favorites"}), 500
1792 except Exception:
1793 logger.exception("Error updating search favorites")
1794 return jsonify({"error": "Failed to update favorites"}), 500
1797@settings_bp.route("/api/search-favorites/toggle", methods=["POST"])
1798@login_required
1799@require_json_body(error_message="No data provided")
1800@with_user_session()
1801def api_toggle_search_favorite(
1802 db_session: Optional[Session] = None, settings_manager=None
1803):
1804 """Toggle a search engine as favorite"""
1805 try:
1806 data = request.get_json()
1807 engine_id = data.get("engine_id")
1808 if not engine_id:
1809 return jsonify({"error": "No engine_id provided"}), 400
1811 # Get current favorites
1812 favorites = settings_manager.get_setting("search.favorites", [])
1813 if not isinstance(favorites, list):
1814 favorites = []
1815 else:
1816 # Make a copy to avoid modifying the original
1817 favorites = list(favorites)
1819 # Toggle the engine
1820 is_favorite = engine_id in favorites
1821 if is_favorite:
1822 favorites.remove(engine_id)
1823 is_favorite = False
1824 else:
1825 favorites.append(engine_id)
1826 is_favorite = True
1828 # Update the setting
1829 if settings_manager.set_setting("search.favorites", favorites):
1830 invalidate_settings_caches(session["username"])
1831 return jsonify(
1832 {
1833 "message": "Favorite toggled successfully",
1834 "engine_id": engine_id,
1835 "is_favorite": is_favorite,
1836 "favorites": favorites,
1837 }
1838 )
1839 return jsonify({"error": "Failed to toggle favorite"}), 500
1841 except Exception:
1842 logger.exception("Error toggling search favorite")
1843 return jsonify({"error": "Failed to toggle favorite"}), 500
1846# Legacy routes for backward compatibility - these will redirect to the new routes
1847@settings_bp.route("/main", methods=["GET"])
1848@login_required
1849def main_config_page():
1850 """Redirect to app settings page"""
1851 return redirect(url_for("settings.settings_page"))
1854@settings_bp.route("/collections", methods=["GET"])
1855@login_required
1856def collections_config_page():
1857 """Redirect to app settings page"""
1858 return redirect(url_for("settings.settings_page"))
1861@settings_bp.route("/api_keys", methods=["GET"])
1862@login_required
1863def api_keys_config_page():
1864 """Redirect to LLM settings page"""
1865 return redirect(url_for("settings.settings_page"))
1868@settings_bp.route("/search_engines", methods=["GET"])
1869@login_required
1870def search_engines_config_page():
1871 """Redirect to search settings page"""
1872 return redirect(url_for("settings.settings_page"))
1875@settings_bp.route("/llm", methods=["GET"])
1876@login_required
1877def llm_config_page():
1878 """Redirect to LLM settings page"""
1879 return redirect(url_for("settings.settings_page"))
1882@settings_bp.route("/open_file_location", methods=["POST"])
1883@login_required
1884def open_file_location():
1885 """Open the location of a configuration file.
1887 Security: This endpoint is disabled for server deployments.
1888 It only makes sense for desktop usage where the server and client are on the same machine.
1889 """
1890 return jsonify(
1891 {
1892 "status": "error",
1893 "message": "This feature is disabled. It is only available in desktop mode.",
1894 }
1895 ), 403
1898@settings_bp.context_processor
1899def inject_csrf_token():
1900 """Inject CSRF token into the template context for all settings routes."""
1901 return {"csrf_token": generate_csrf}
1904@settings_bp.route("/fix_corrupted_settings", methods=["POST"])
1905@login_required
1906@settings_limit
1907@with_user_session(include_settings_manager=False)
1908def fix_corrupted_settings(db_session: Optional[Session] = None):
1909 """Fix corrupted settings in the database"""
1910 try:
1911 # Track fixed and removed settings
1912 fixed_settings = []
1913 removed_duplicate_settings = []
1914 # First, find and remove duplicate settings with the same key
1915 # This happens because of errors in settings import/export
1916 from sqlalchemy import func as sql_func
1918 # Find keys with duplicates
1919 duplicate_keys = (
1920 db_session.query(Setting.key)
1921 .group_by(Setting.key)
1922 .having(sql_func.count(Setting.key) > 1)
1923 .all()
1924 )
1925 duplicate_keys = [key[0] for key in duplicate_keys]
1927 # For each duplicate key, keep the latest updated one and remove others
1928 for key in duplicate_keys:
1929 dupe_settings = (
1930 db_session.query(Setting)
1931 .filter(Setting.key == key)
1932 .order_by(Setting.updated_at.desc())
1933 .all()
1934 )
1936 # Keep the first one (most recently updated) and delete the rest
1937 for i, setting in enumerate(dupe_settings):
1938 if i > 0: # Skip the first one (keep it)
1939 db_session.delete(setting)
1940 removed_duplicate_settings.append(key)
1942 # Check for settings with corrupted values
1943 all_settings = db_session.query(Setting).all()
1944 for setting in all_settings:
1945 # Check different types of corruption
1946 is_corrupted = False
1948 if (
1949 setting.value is None
1950 or (
1951 isinstance(setting.value, str)
1952 and setting.value
1953 in [
1954 "{",
1955 "[",
1956 "{}",
1957 "[]",
1958 "[object Object]",
1959 "null",
1960 "undefined",
1961 ]
1962 )
1963 or (isinstance(setting.value, dict) and len(setting.value) == 0)
1964 ):
1965 is_corrupted = True
1967 # Skip if not corrupted
1968 if not is_corrupted:
1969 continue
1971 default_value: Any = None
1973 # Try to find a matching default setting based on key
1974 if setting.key.startswith("llm."):
1975 if setting.key == "llm.model":
1976 default_value = ""
1977 elif setting.key == "llm.provider":
1978 default_value = "ollama"
1979 elif setting.key == "llm.temperature":
1980 default_value = 0.7
1981 elif setting.key == "llm.max_tokens": 1981 ↛ 2022line 1981 didn't jump to line 2022 because the condition on line 1981 was always true
1982 default_value = 1024
1983 elif setting.key.startswith("search."):
1984 if setting.key == "search.tool":
1985 default_value = DEFAULT_SEARCH_TOOL
1986 elif setting.key == "search.max_results":
1987 default_value = 10
1988 elif setting.key == "search.region":
1989 default_value = "us"
1990 elif setting.key == "search.questions_per_iteration":
1991 default_value = 3
1992 elif setting.key == "search.searches_per_section":
1993 default_value = 2
1994 elif setting.key == "search.skip_relevance_filter":
1995 default_value = False
1996 elif setting.key == "search.safe_search":
1997 default_value = True
1998 elif setting.key == "search.search_language": 1998 ↛ 2022line 1998 didn't jump to line 2022 because the condition on line 1998 was always true
1999 default_value = "English"
2000 elif setting.key.startswith("report."):
2001 if setting.key == "report.searches_per_section":
2002 default_value = 2
2003 elif setting.key.startswith("app."): 2003 ↛ 2022line 2003 didn't jump to line 2022 because the condition on line 2003 was always true
2004 if (
2005 setting.key == "app.theme"
2006 or setting.key == "app.default_theme"
2007 ):
2008 default_value = "dark"
2009 elif setting.key == "app.enable_notifications" or (
2010 setting.key == "app.enable_web"
2011 or setting.key == "app.web_interface"
2012 ):
2013 default_value = True
2014 elif setting.key == "app.host":
2015 default_value = "0.0.0.0"
2016 elif setting.key == "app.port":
2017 default_value = 5000
2018 elif setting.key == "app.debug": 2018 ↛ 2022line 2018 didn't jump to line 2022 because the condition on line 2018 was always true
2019 default_value = True
2021 # Update the setting with the default value if found
2022 if default_value is not None:
2023 setting.value = default_value
2024 fixed_settings.append(setting.key)
2025 else:
2026 # If no default found but it's a corrupted JSON, set to empty object
2027 if setting.key.startswith("report."): 2027 ↛ 1944line 2027 didn't jump to line 1944 because the condition on line 2027 was always true
2028 setting.value = {}
2029 fixed_settings.append(setting.key)
2031 # Commit changes
2032 if fixed_settings or removed_duplicate_settings:
2033 db_session.commit()
2034 logger.info(
2035 f"Fixed {len(fixed_settings)} corrupted settings: {', '.join(fixed_settings)}"
2036 )
2037 if removed_duplicate_settings:
2038 logger.info(
2039 f"Removed {len(removed_duplicate_settings)} duplicate settings"
2040 )
2041 invalidate_settings_caches(session["username"])
2043 # Return success
2044 return jsonify(
2045 {
2046 "status": "success",
2047 "message": f"Fixed {len(fixed_settings)} corrupted settings, removed {len(removed_duplicate_settings)} duplicates",
2048 "fixed_settings": fixed_settings,
2049 "removed_duplicates": removed_duplicate_settings,
2050 }
2051 )
2053 except Exception:
2054 logger.exception("Error fixing corrupted settings")
2055 db_session.rollback()
2056 return (
2057 jsonify(
2058 {
2059 "status": "error",
2060 "message": "An internal error occurred while fixing corrupted settings. Please try again later.",
2061 }
2062 ),
2063 500,
2064 )
2067@settings_bp.route("/api/warnings", methods=["GET"])
2068@login_required
2069def api_get_warnings():
2070 """Get current warnings based on settings"""
2071 try:
2072 warnings = calculate_warnings()
2073 return jsonify({"warnings": warnings})
2074 except Exception:
2075 logger.exception("Error getting warnings")
2076 return jsonify({"error": "Failed to retrieve warnings"}), 500
2079@settings_bp.route("/api/backup-status", methods=["GET"])
2080@login_required
2081def api_get_backup_status():
2082 """Get backup status for the current user."""
2083 try:
2084 from ...config.paths import get_user_backup_directory
2086 username = session.get("username")
2087 if not username: 2087 ↛ 2088line 2087 didn't jump to line 2088 because the condition on line 2087 was never true
2088 return jsonify({"error": "Not authenticated"}), 401
2090 from ...utilities.formatting import human_size
2091 from ...database.backup.backup_service import is_safe_glob_result
2093 backup_dir = get_user_backup_directory(username)
2095 # Sort by modification time (not filename) for robustness
2096 backup_list = []
2097 total_size = 0
2098 for b in backup_dir.glob("ldr_backup_*.db"):
2099 # Skip symlinks / entries resolving outside backup_dir so a
2100 # planted symlink can't have its (external) target's metadata
2101 # reported here — same glob-hardening applied in BackupService.
2102 if not is_safe_glob_result(b, backup_dir):
2103 continue
2104 try:
2105 stat = b.stat()
2106 total_size += stat.st_size
2107 backup_list.append(
2108 {
2109 "filename": b.name,
2110 "size_bytes": stat.st_size,
2111 "size_human": human_size(stat.st_size),
2112 "created_at": datetime.fromtimestamp(
2113 stat.st_mtime, tz=timezone.utc
2114 ).isoformat(),
2115 "_mtime": stat.st_mtime,
2116 }
2117 )
2118 except FileNotFoundError:
2119 continue
2121 # Sort newest first by mtime, then remove internal field
2122 backup_list.sort(key=lambda x: x["_mtime"], reverse=True)
2123 for entry in backup_list:
2124 del entry["_mtime"]
2126 backup_enabled = _get_setting_from_session("backup.enabled", True)
2128 return jsonify(
2129 {
2130 "enabled": bool(backup_enabled),
2131 "count": len(backup_list),
2132 "backups": backup_list,
2133 "total_size_bytes": total_size,
2134 "total_size_human": human_size(total_size),
2135 }
2136 )
2138 except Exception:
2139 logger.exception("Error getting backup status")
2140 return jsonify({"error": "Failed to retrieve backup status"}), 500
2143@settings_bp.route("/api/ollama-status", methods=["GET"])
2144@login_required
2145def check_ollama_status():
2146 """Check if Ollama is running and available"""
2147 try:
2148 # Get Ollama URL from settings
2149 raw_base_url = _get_setting_from_session(
2150 "llm.ollama.url", DEFAULT_OLLAMA_URL
2151 )
2152 base_url = (
2153 normalize_url(raw_base_url) if raw_base_url else DEFAULT_OLLAMA_URL
2154 )
2156 response = safe_get(
2157 f"{base_url}/api/version",
2158 timeout=2,
2159 allow_localhost=True,
2160 allow_private_ips=True,
2161 )
2163 if response.status_code == 200:
2164 return jsonify(
2165 {
2166 "running": True,
2167 "version": response.json().get("version", "unknown"),
2168 }
2169 )
2170 return jsonify(
2171 {
2172 "running": False,
2173 "error": f"Ollama returned status code {response.status_code}",
2174 }
2175 )
2176 except requests.exceptions.RequestException:
2177 logger.exception("Ollama check failed")
2178 return jsonify(
2179 {"running": False, "error": "Failed to check search engine status"}
2180 )
2183@settings_bp.route("/api/rate-limiting/status", methods=["GET"])
2184@login_required
2185def api_get_rate_limiting_status():
2186 """Get current rate limiting status and statistics"""
2187 try:
2188 username = session["username"]
2190 # exploration_rate / learning_rate are the *configured* settings, not
2191 # the effective operating values: AdaptiveRateLimitTracker._apply_profile
2192 # scales them by the active profile (conservative ~0.5x/0.7x, aggressive
2193 # ~1.5x/1.3x, each capped). The profile is reported alongside so a
2194 # consumer can tell which transform the running tracker applies; we
2195 # deliberately don't duplicate that scaling math here.
2196 status = {
2197 # Default True to match the schema default (default_settings.json)
2198 # and the tracker's web-mode default; the prior endpoint reported
2199 # the tracker's effective enabled state, which was on by default.
2200 "enabled": _get_setting_from_session("rate_limiting.enabled", True),
2201 "profile": _get_setting_from_session(
2202 "rate_limiting.profile", "balanced"
2203 ),
2204 "exploration_rate": _get_setting_from_session(
2205 "rate_limiting.exploration_rate", 0.1
2206 ),
2207 "learning_rate": _get_setting_from_session(
2208 "rate_limiting.learning_rate", 0.45
2209 ),
2210 "memory_window": _get_setting_from_session(
2211 "rate_limiting.memory_window", 100
2212 ),
2213 }
2215 with get_user_db_session(username) as db_session:
2216 estimates = (
2217 db_session.query(RateLimitEstimate)
2218 .order_by(RateLimitEstimate.engine_type)
2219 .all()
2220 )
2222 engines = []
2223 for est in estimates:
2224 engines.append(
2225 {
2226 "engine_type": est.engine_type,
2227 "base_wait_seconds": round(est.base_wait_seconds, 2),
2228 "min_wait_seconds": round(est.min_wait_seconds, 2),
2229 "max_wait_seconds": round(est.max_wait_seconds, 2),
2230 "last_updated": est.last_updated,
2231 "total_attempts": est.total_attempts,
2232 "success_rate": round(est.success_rate * 100, 1),
2233 }
2234 )
2236 return jsonify({"status": status, "engines": engines})
2238 except Exception:
2239 logger.exception("Error getting rate limiting status")
2240 return jsonify({"error": "An internal error occurred"}), 500
2243@settings_bp.route(
2244 "/api/rate-limiting/engines/<engine_type>/reset", methods=["POST"]
2245)
2246@login_required
2247def api_reset_engine_rate_limiting(engine_type):
2248 """Reset (forget) the learned rate-limit estimate for a specific engine.
2250 Deletes the engine's persisted ``RateLimitEstimate`` row from the user's
2251 database so the adaptive tracker re-learns it from scratch. The previous
2252 implementation called the per-request ``get_tracker()``, whose mutation
2253 path is gated on a research-session context that is absent in an analytics
2254 HTTP request — so it was a silent no-op that never cleared the persisted
2255 estimate the ``/status`` and ``/current`` endpoints display (#4721).
2256 """
2257 try:
2258 username = session["username"]
2259 with get_user_db_session(username) as db_session:
2260 db_session.query(RateLimitEstimate).filter_by(
2261 engine_type=engine_type
2262 ).delete(synchronize_session=False)
2263 db_session.commit()
2265 return jsonify(
2266 {"message": f"Rate limiting data reset for {engine_type}"}
2267 )
2269 except Exception:
2270 logger.exception(f"Error resetting rate limiting for {engine_type}")
2271 return jsonify({"error": "An internal error occurred"}), 500
2274@settings_bp.route("/api/rate-limiting/cleanup", methods=["POST"])
2275@login_required
2276def api_cleanup_rate_limiting():
2277 """Clean up old rate limiting data.
2279 Note: not using @require_json_body because the JSON body is optional
2280 here — the endpoint works with or without a payload (defaults to 30 days).
2281 """
2282 try:
2283 data = request.get_json() if request.is_json else None
2284 days = data.get("days", 30) if data is not None else 30
2286 try:
2287 days = int(days)
2288 except (TypeError, ValueError):
2289 return jsonify({"error": "'days' must be an integer"}), 400
2290 if days < 1 or days > 365:
2291 return jsonify({"error": "'days' must be between 1 and 365"}), 400
2293 # Delete persisted estimates not updated within the window. Mirrors the
2294 # read endpoints (#4721): operate on RateLimitEstimate rather than the
2295 # per-request get_tracker(), whose cleanup path is a no-op outside a
2296 # research-session context. last_updated is a unix timestamp (Float).
2297 username = session["username"]
2298 cutoff = time.time() - days * 86400
2299 with get_user_db_session(username) as db_session:
2300 db_session.query(RateLimitEstimate).filter(
2301 RateLimitEstimate.last_updated < cutoff
2302 ).delete(synchronize_session=False)
2303 db_session.commit()
2305 return jsonify(
2306 {"message": f"Cleaned up rate limiting data older than {days} days"}
2307 )
2309 except Exception:
2310 logger.exception("Error cleaning up rate limiting data")
2311 return jsonify({"error": "An internal error occurred"}), 500
2314@settings_bp.route("/api/bulk", methods=["GET"])
2315@login_required
2316def get_bulk_settings():
2317 """Get multiple settings at once for performance.
2319 The caller-supplied ``keys[]`` is unrestricted, so this endpoint is a
2320 candidate channel to exfiltrate password-typed settings — any
2321 authenticated client can request ``keys[]=llm.openai.api_key`` and
2322 would otherwise receive the plaintext value. Redact password fields
2323 by checking the leaf segment of the dotted key against
2324 ``DataSanitizer.DEFAULT_SENSITIVE_KEYS`` (e.g. ``api_key``,
2325 ``password``, ``access_token``, …). Same defense-in-depth predicate
2326 used in ``redact_settings_snapshot``; here we use the suffix-only
2327 check because the response shape is ``{value, exists}`` — no
2328 ``ui_element`` is fetched. ``exists`` is preserved so callers can
2329 still tell whether the key is configured.
2330 """
2331 try:
2332 from ...security.data_sanitizer import DataSanitizer
2334 # Get requested settings from query parameters
2335 requested = request.args.getlist("keys[]")
2336 if not requested:
2337 # Default to common settings if none specified
2338 requested = [
2339 "llm.provider",
2340 "llm.model",
2341 "search.tool",
2342 "search.iterations",
2343 "search.questions_per_iteration",
2344 "search.search_strategy",
2345 "benchmark.evaluation.provider",
2346 "benchmark.evaluation.model",
2347 "benchmark.evaluation.temperature",
2348 "benchmark.evaluation.endpoint_url",
2349 ]
2351 # Fetch all settings at once
2352 result = {}
2353 for key in requested:
2354 try:
2355 value = _get_setting_from_session(key)
2356 exists = value is not None
2357 # No ui_element metadata in the bulk path, so redact_value
2358 # falls back to the suffix-only arm of the shared predicate.
2359 value = DataSanitizer.redact_value(key, None, value)
2360 result[key] = {"value": value, "exists": exists}
2361 except Exception:
2362 logger.warning(f"Error getting setting {key}")
2363 result[key] = {
2364 "value": None,
2365 "exists": False,
2366 "error": "Failed to retrieve setting",
2367 }
2369 return jsonify({"success": True, "settings": result})
2371 except Exception:
2372 logger.exception("Error getting bulk settings")
2373 return jsonify(
2374 {"success": False, "error": "An internal error occurred"}
2375 ), 500
2378@settings_bp.route("/api/data-location", methods=["GET"])
2379@login_required
2380def api_get_data_location():
2381 """Get information about data storage location and security"""
2382 try:
2383 # Get the data directory path
2384 data_dir = get_data_directory()
2385 # Get the encrypted databases path
2386 encrypted_db_path = get_encrypted_database_path()
2388 # Check if LDR_DATA_DIR environment variable is set
2389 from local_deep_research.settings.manager import SettingsManager
2391 settings_manager = SettingsManager()
2392 custom_data_dir = settings_manager.get_setting("bootstrap.data_dir")
2394 # Get platform-specific default location info
2395 platform_info = {
2396 "Windows": "C:\\Users\\Username\\AppData\\Local\\local-deep-research",
2397 "macOS": "~/Library/Application Support/local-deep-research",
2398 "Linux": "~/.local/share/local-deep-research",
2399 }
2401 # Current platform
2402 current_platform = platform.system()
2403 if current_platform == "Darwin":
2404 current_platform = "macOS"
2406 # Get SQLCipher settings from environment
2407 from ...database.sqlcipher_utils import get_sqlcipher_settings
2409 # Debug logging
2410 logger.info(f"db_manager type: {type(db_manager)}")
2411 logger.info(
2412 f"db_manager.has_encryption: {getattr(db_manager, 'has_encryption', 'ATTRIBUTE NOT FOUND')}"
2413 )
2415 cipher_settings = (
2416 get_sqlcipher_settings() if db_manager.has_encryption else {}
2417 )
2419 return jsonify(
2420 {
2421 "data_directory": str(data_dir),
2422 "database_path": str(encrypted_db_path),
2423 "encrypted_database_path": str(encrypted_db_path),
2424 "is_custom": custom_data_dir is not None,
2425 "custom_env_var": "LDR_DATA_DIR",
2426 "custom_env_value": custom_data_dir,
2427 "platform": current_platform,
2428 "platform_default": platform_info.get(
2429 current_platform, str(data_dir)
2430 ),
2431 "platform_info": platform_info,
2432 "security_notice": {
2433 "encrypted": db_manager.has_encryption,
2434 "warning": "All data including API keys stored in the database are securely encrypted."
2435 if db_manager.has_encryption
2436 else "All data including API keys stored in the database are currently unencrypted. Please ensure appropriate file system permissions are set.",
2437 "recommendation": "Your data is protected with database encryption."
2438 if db_manager.has_encryption
2439 else "Consider using environment variables for sensitive API keys instead of storing them in the database.",
2440 },
2441 "encryption_settings": cipher_settings,
2442 }
2443 )
2445 except Exception:
2446 logger.exception("Error getting data location information")
2447 return jsonify({"error": "Failed to retrieve data location"}), 500
2450@settings_bp.route("/api/notifications/test-url", methods=["POST"])
2451@login_required
2452def api_test_notification_url():
2453 """
2454 Test a notification service URL.
2456 This endpoint creates a temporary NotificationService instance to test
2457 the provided URL. No database session or password is required because:
2458 - The service URL is provided directly in the request body
2459 - Test notifications use a temporary Apprise instance
2460 - No user settings or database queries are performed
2462 Security note: Rate limiting is not applied here because users need to
2463 test URLs when configuring notifications. Abuse is mitigated by the
2464 @login_required decorator and the fact that users can only spam their
2465 own notification services.
2466 """
2467 try:
2468 from ...notifications.service import NotificationService
2469 from ...settings.env_registry import get_env_setting
2471 data = request.get_json()
2472 if not data or "service_url" not in data:
2473 return jsonify(
2474 {"success": False, "error": "service_url is required"}
2475 ), 400
2477 service_url = data["service_url"]
2479 # Create notification service instance and test the URL.
2480 # Gate by the env-only master switch so the test endpoint cannot
2481 # bypass the operator's risk-acceptance decision (see SECURITY.md
2482 # "Notification Webhook SSRF").
2483 notification_service = NotificationService(
2484 allow_private_ips=bool(
2485 get_env_setting("notifications.allow_private_ips", False)
2486 ),
2487 outbound_allowed=bool(
2488 get_env_setting("notifications.allow_outbound", False)
2489 ),
2490 )
2491 result = notification_service.test_service(service_url)
2493 # Only return expected fields to prevent information leakage
2494 safe_response = {
2495 "success": result.get("success", False),
2496 "message": result.get("message", ""),
2497 "error": result.get("error", ""),
2498 }
2499 return jsonify(safe_response)
2501 except Exception:
2502 logger.exception("Error testing notification URL")
2503 return jsonify(
2504 {
2505 "success": False,
2506 "error": "Failed to test notification service. Check logs for details.",
2507 }
2508 ), 500