Files
FabledScribe/src/fabledassistant/routes/settings.py
T
bvandeusen 00643c778e security: fix 10 vulnerabilities from security audit
- SSRF: block private/internal URLs in image cache fetch
- SSRF: block private/internal URLs in RSS feed fetch (scheme guard)
- SSRF: block private/internal URLs in CalDAV URL setting
- Auth: require login for GET /api/images/<id> (was unauthenticated)
- Auth: restrict Ollama model pull/delete to admin users only
- Info disclosure: remove email from /api/users/search response
- OAuth: skip email-based account linking when email_verified is false
- Config: raise hard error on default SECRET_KEY when SECURE_COOKIES=true
- Rate limit: document proxy header requirement; add startup warning
- XSS: remove src/alt from global DOMPurify ADD_ATTR allowlist

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-29 00:37:13 -04:00

122 lines
4.2 KiB
Python

from quart import Blueprint, jsonify, request
from fabledassistant.auth import login_required, get_current_user_id
from fabledassistant.config import Config
from fabledassistant.services.caldav import CALDAV_SETTING_KEYS, get_caldav_config, test_connection
from fabledassistant.services.llm import get_installed_models, _is_private_url
from fabledassistant.services.settings import delete_setting, get_all_settings, set_settings_batch
settings_bp = Blueprint("settings", __name__, url_prefix="/api/settings")
@settings_bp.route("", methods=["GET"])
@login_required
async def get_settings_route():
uid = get_current_user_id()
settings = await get_all_settings(uid)
return jsonify(settings)
@settings_bp.route("", methods=["PUT"])
@login_required
async def update_settings_route():
uid = get_current_user_id()
data = await request.get_json()
if not isinstance(data, dict):
return jsonify({"error": "Expected a JSON object"}), 400
if "default_model" in data:
installed = await get_installed_models()
if data["default_model"]:
model = str(data["default_model"])
if installed and model not in installed:
return jsonify({"error": f"Model '{model}' is not installed"}), 400
# Empty string for default_model means "reset to system default".
# Delete the DB row so get_setting() falls back to Config defaults
# rather than returning "" and breaking model resolution everywhere.
_MODEL_KEYS = frozenset({"default_model"})
to_save = {}
for k, v in data.items():
str_v = str(v)
if k in _MODEL_KEYS and not str_v:
await delete_setting(uid, k)
else:
to_save[k] = str_v
if to_save:
await set_settings_batch(uid, to_save)
settings = await get_all_settings(uid)
return jsonify(settings)
@settings_bp.route("/models", methods=["GET"])
@login_required
async def get_models_route():
"""Return installed Ollama models and the configured defaults."""
models = sorted(await get_installed_models())
return jsonify({
"models": models,
"default_chat_model": Config.OLLAMA_MODEL,
})
@settings_bp.route("/caldav", methods=["GET"])
@login_required
async def get_caldav():
uid = get_current_user_id()
config = await get_caldav_config(uid)
if config.get("caldav_password"):
config["caldav_password"] = "********"
return jsonify(config)
@settings_bp.route("/caldav", methods=["PUT"])
@login_required
async def update_caldav():
uid = get_current_user_id()
data = await request.get_json()
# Validate CalDAV URL before saving — block internal/private addresses
if "caldav_url" in data:
url = str(data.get("caldav_url") or "").strip()
if url:
parsed_scheme = url.split("://")[0].lower() if "://" in url else ""
if parsed_scheme not in ("http", "https"):
return jsonify({"error": "CalDAV URL must use http or https"}), 400
if _is_private_url(url):
return jsonify({"error": "CalDAV URL must not point to an internal or private address"}), 400
settings_to_save = {}
for key in CALDAV_SETTING_KEYS:
if key in data:
if key == "caldav_password" and data[key] == "********":
continue
settings_to_save[key] = str(data[key])
if settings_to_save:
await set_settings_batch(uid, settings_to_save)
return jsonify({"status": "ok"})
@settings_bp.route("/caldav/test", methods=["POST"])
@login_required
async def test_caldav():
uid = get_current_user_id()
result = await test_connection(uid)
return jsonify(result)
@settings_bp.route("/search", methods=["GET"])
@login_required
async def test_search():
"""Test SearXNG connectivity and preview results for a query."""
if not Config.searxng_enabled():
return jsonify({"configured": False, "results": [], "searxng_url": ""})
q = request.args.get("q", "").strip()
if not q:
return jsonify({"configured": True, "results": [], "searxng_url": Config.SEARXNG_URL})
from fabledassistant.services.research import _search_searxng
results = await _search_searxng(q)
return jsonify({"configured": True, "results": results, "query": q, "searxng_url": Config.SEARXNG_URL})