FabledScribe/src/fabledassistant/routes/settings.py

"""User settings + integrations (CalDAV, SearXNG status).

Chat-model picker endpoints (/models), KV-cache priming, and journal-schedule
hooks were removed in Phase 8 alongside the chat/journal subsystems.
"""
import ipaddress
import logging
import socket
from urllib.parse import urlparse

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.settings import delete_setting, get_all_settings, get_setting, set_settings_batch

logger = logging.getLogger(__name__)


def _is_private_url(url: str) -> bool:
    """SSRF-blocking helper: returns True for URLs that resolve to private,
    loopback, or link-local addresses. Inlined here after services/llm.py
    (the original home) was removed in Phase 8."""
    try:
        host = urlparse(url).hostname
        if not host:
            return True
        # Resolve to all addresses; reject if any is private/loopback/link-local.
        infos = socket.getaddrinfo(host, None)
        for family, *_rest, sockaddr in infos:
            ip_str = sockaddr[0]
            try:
                ip = ipaddress.ip_address(ip_str)
            except ValueError:
                continue
            if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:
                return True
    except Exception:
        # Conservative: if we can't resolve, treat as private (reject).
        return True
    return False


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

    to_save = {}
    for k, v in data.items():
        str_v = str(v)
        if 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("/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():
    """Report SearXNG configuration status (used by the Integrations tab)."""
    if not Config.searxng_enabled():
        return jsonify({"configured": False, "results": [], "searxng_url": ""})
    return jsonify({"configured": True, "results": [], "searxng_url": Config.SEARXNG_URL})