security: harden P0 boundaries and recovery

CHANGELOG.md

@@ -5,6 +5,22 @@ Follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) · versioned wi
 
 ---
 
+## [Unreleased]
+
+### Added
+- **Protected Backup Recovery.** Added an authenticated backup inventory and restore API for character and rehearsal
+  snapshots, with path-safe file selection and atomic pre-mutation backups.
+- **Character and Voice Metadata Validation.** Added bounded type and size validation before records and voice sidecars
+  are persisted.
+
+### Changed
+- **API Boundary Hardening.** External API authentication is enabled by default, request and mutation limits are
+  enforced, and settings responses redact provider credentials for non-browser callers.
+
+### Fixed
+- **SSRF and Credential-Leak Paths.** User-supplied downloads are validated against private-network access settings,
+  and provider keys no longer travel in model-probe URL query strings.
+
 ## [1.22.3] — 2026-09-04
 
 ### Changed

core/config.py

@@ -46,11 +46,9 @@ _SETTINGS_KEYS = {
     # Inbound API key — gates non-browser callers (external scripts, MCP
     # clients) hitting this app's own /api/* and /mcp routes. Distinct from
     # every *outbound* key above, which are credentials this app sends to
-    # OTHER services. Off by default (external_api_key_required=False) —
+    # OTHER services. The server-side gate is enabled by default; deployments
-    # this is a live, actively-used app and the same-origin detection this
+    # behind another authentication layer may explicitly opt out with
-    # gate relies on has never been exercised against real browser traffic,
+    # REQUIRE_EXTERNAL_API_KEY=0.
-    # so enforcing it unconditionally risked locking out the working UI on
-    # an untested edge case. Turn it on deliberately once confirmed safe.
     "external_api_key", "external_api_key_required",
     # Browser-persistent UI state
     "refine_llm_url", "conv_llm_url", "seed_finder_text",
@@ -343,7 +341,7 @@ def _load_settings() -> dict:
         "comfyui_prompt_field":   "text",
         "comfyui_output_node_id": "",
         "external_api_key":       "",
-        "external_api_key_required": False,
+        "external_api_key_required": True,
         "refine_llm_url":         "",
         "conv_llm_url":           "",
         "seed_finder_text":       "",

core/constants.py

@@ -68,6 +68,9 @@ _BENCHMARK_SENTENCES = [
     ),
 ]
 _ALLOW_PRIVATE_DOWNLOADS = os.environ.get("ALLOW_PRIVATE_DOWNLOADS", "").lower() in {"1", "true", "yes"}
+# External callers must present the app's inbound key by default. Set this only
+# for a deliberately isolated local deployment that has another auth layer.
+_REQUIRE_EXTERNAL_API_KEY = os.environ.get("REQUIRE_EXTERNAL_API_KEY", "1").lower() not in {"0", "false", "no", "off"}
 
 # ── Config dir calculation ────────────────────────────────────────────────────
 

core/database.py

@@ -18,6 +18,7 @@ from pathlib import Path
 from typing import Any
 
 from core.constants import CONFIG_DIR
+from core.validation import _validate_character_record
 
 _DB_PATH = CONFIG_DIR / "tts_creator.db"
 
@@ -213,6 +214,7 @@ def char_get(char_id: str) -> dict | None:
 
 
 def char_put(rec: dict) -> dict:
+    _validate_character_record(rec)
     existing = char_get(rec.get("id", "")) if rec.get("id") else None
     image = rec.get("image")
     sheet = rec.get("sheet") or {}

core/validation.py

@@ -60,3 +60,31 @@ def _safe_child_path(root: Path, candidate: Path) -> Path:
     except ValueError:
         raise HTTPException(403, "Access denied")
     return candidate_resolved
+
+
+def _validate_character_record(record: dict) -> dict:
+    """Validate the persisted shape of a character before it reaches SQLite."""
+    if not isinstance(record, dict):
+        raise HTTPException(400, "Character record must be an object")
+    required_text = {
+        "id": 240, "book": 500, "name": 300, "tags": 4000,
+    }
+    for field, max_length in required_text.items():
+        value = record.get(field, "")
+        if not isinstance(value, str):
+            raise HTTPException(400, f"Character field '{field}' must be text")
+        if len(value) > max_length:
+            raise HTTPException(413, f"Character field '{field}' is too large")
+    if not record.get("id"):
+        raise HTTPException(400, "Character id is required")
+    for field in ("voice", "image"):
+        value = record.get(field)
+        if value is not None and not isinstance(value, str):
+            raise HTTPException(400, f"Character field '{field}' must be text or null")
+        if isinstance(value, str) and len(value) > 24 * 1024 * 1024:
+            raise HTTPException(413, f"Character field '{field}' is too large")
+    if not isinstance(record.get("sheet", {}), dict):
+        raise HTTPException(400, "Character sheet must be an object")
+    if "analysis" in record and record["analysis"] is not None and not isinstance(record["analysis"], (dict, list)):
+        raise HTTPException(400, "Character analysis must be an object, array, or null")
+    return record

core/voice.py

@@ -121,6 +121,7 @@ def _load_meta(wav: Path) -> dict:
 
 
 def _save_meta(wav: Path, meta: dict) -> None:
+    _validate_voice_meta(meta)
     # Write beside the final file and replace it atomically. A process exit or
     # host/container interruption during write must never leave a truncated
     # meta.json that makes ratings/settings appear to have vanished.
@@ -141,6 +142,34 @@ def _save_meta(wav: Path, meta: dict) -> None:
             pass
 
 
+def _validate_voice_meta(meta: dict) -> None:
+    """Validate the small, user-editable portion of a voice sidecar."""
+    if not isinstance(meta, dict):
+        raise ValueError("Voice metadata must be an object")
+    string_limits = {
+        "note": 4000, "flag": 16, "gender": 16, "age": 64, "lang": 32,
+        "language": 64, "persona": 8000, "origin": 64, "group": 300,
+        "name": 300, "tag": 300, "source": 2000, "avatar": 2000,
+        "voice_design_prompt": 12000, "reference_transcript_status": 64,
+        "reference_transcript_backend": 128, "import_preparation": 64,
+        "import_preparation_error": 1000,
+    }
+    for field, limit in string_limits.items():
+        value = meta.get(field)
+        if value is not None and (not isinstance(value, str) or len(value) > limit):
+            raise ValueError(f"Invalid voice metadata field: {field}")
+    if "rating" in meta:
+        rating = meta["rating"]
+        if isinstance(rating, bool) or not isinstance(rating, (int, float)) or not 0 <= rating <= 5:
+            raise ValueError("Voice rating must be a number from 0 to 5")
+    for field in ("enabled", "needs_tts_restart"):
+        if field in meta and not isinstance(meta[field], bool):
+            raise ValueError(f"Voice metadata field '{field}' must be boolean")
+    for field in ("loudness", "benchmark"):
+        if field in meta and meta[field] is not None and not isinstance(meta[field], dict):
+            raise ValueError(f"Voice metadata field '{field}' must be an object")
+
+
 def _picture_path(wav: Path) -> Path | None:
     for ext in _PICTURE_EXTS:
         p = wav.with_suffix(ext)

routes/characters.py

@@ -16,6 +16,7 @@ from fastapi import APIRouter, HTTPException, Request
 from fastapi.responses import Response
 
 from core.database import char_get_all, char_get, char_put, char_delete
+from core.validation import _validate_character_record
 
 router = APIRouter()
 
@@ -112,6 +113,7 @@ async def characters_put(char_id: str, request: Request):
     except Exception:
         raise HTTPException(400, "Invalid JSON")
     body["id"] = char_id
+    _validate_character_record(body)
     return char_put(body)
 
 
@@ -133,6 +135,7 @@ async def characters_migrate(request: Request):
     n = 0
     for rec in records:
         if rec.get("id"):
+            _validate_character_record(rec)
             char_put(rec)
             n += 1
     return {"imported": n}

routes/recovery.py

@@ -0,0 +1,76 @@
+"""Protected inventory and restoration of automatic recovery snapshots."""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from fastapi import APIRouter, HTTPException, Request
+
+from core.config import CONFIG_DIR
+from core.database import char_put, reh_put
+from core.validation import _safe_child_path
+
+router = APIRouter()
+_BACKUP_ROOT = CONFIG_DIR / "backups"
+_BACKUP_KINDS = {"characters", "rehearsals"}
+
+
+def _backup_path(kind: str, filename: str) -> Path:
+    if kind not in _BACKUP_KINDS or not filename or Path(filename).name != filename or not filename.endswith(".json"):
+        raise HTTPException(400, "Invalid backup reference")
+    directory = _BACKUP_ROOT / kind
+    path = _safe_child_path(directory, directory / filename)
+    if not path.is_file():
+        raise HTTPException(404, "Backup not found")
+    return path
+
+
+def _read_backup(path: Path) -> dict:
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError) as exc:
+        raise HTTPException(422, f"Backup cannot be read: {exc}")
+    if not isinstance(data, dict) or not isinstance(data.get("record"), dict):
+        raise HTTPException(422, "Invalid backup format")
+    return data
+
+
+@router.get("/api/backups")
+async def list_backups():
+    items: list[dict] = []
+    for kind in sorted(_BACKUP_KINDS):
+        directory = _BACKUP_ROOT / kind
+        if not directory.is_dir():
+            continue
+        for path in sorted(directory.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True):
+            try:
+                data = _read_backup(path)
+                record = data["record"]
+            except HTTPException:
+                continue
+            items.append({
+                "kind": kind,
+                "file": path.name,
+                "saved_at": data.get("saved_at", ""),
+                "reason": data.get("reason", ""),
+                "id": record.get("id"),
+                "name": record.get("name") or record.get("title", ""),
+            })
+    return {"backups": items[:500]}
+
+
+@router.post("/api/backups/restore")
+async def restore_backup(request: Request):
+    try:
+        data = await request.json()
+    except Exception:
+        raise HTTPException(400, "Invalid JSON")
+    kind = str(data.get("kind") or "").strip().lower()
+    path = _backup_path(kind, str(data.get("file") or "").strip())
+    payload = _read_backup(path)
+    record = payload["record"]
+    if kind == "characters":
+        restored = char_put(record)
+        return {"ok": True, "kind": kind, "id": restored.get("id"), "restored_from": path.name}
+    restored = reh_put(record)
+    return {"ok": True, "kind": kind, "id": restored.get("id"), "restored_from": path.name}

routes/settings.py

@@ -11,13 +11,31 @@ from core.presets import _load_design_presets, _save_design_presets
 router = APIRouter()
 
 
+_SECRET_SETTING_KEYS = {
+    "groq_api_key", "whisper_api_key", "tts_api_key", "voice_design_api_key",
+    "elevenlabs_api_key", "llm_api_key", "engine_api_keys", "external_api_key",
+}
+
+
 @router.get("/api/settings")
-async def get_settings():
+async def get_settings(request: Request):
     s = _load_settings()
     # Lazily generated on first read, not at server startup — a fresh
     # install shows a real usable key in Settings immediately without a
     # migration step, whether or not the gate is actually turned on yet.
     s["external_api_key"] = _ensure_external_api_key()
+    host = request.headers.get("host", "")
+    origin = request.headers.get("origin", "") or request.headers.get("referer", "")
+    try:
+        from urllib.parse import urlparse
+        same_origin = (
+            request.headers.get("sec-fetch-site", "").lower() == "same-origin"
+            or bool(origin and host and urlparse(origin).netloc == host)
+        )
+    except Exception:
+        same_origin = False
+    if not same_origin:
+        s = {key: value for key, value in s.items() if key not in _SECRET_SETTING_KEYS}
     return s
 
 

server.py

@@ -15,10 +15,10 @@ from fastapi.staticfiles import StaticFiles
 from fastapi.middleware.gzip import GZipMiddleware
 from starlette.datastructures import MutableHeaders
 
-from core.constants import STATIC_DIR, _BufferHandler, _MAX_JSON_BYTES, _MAX_UPLOAD_BYTES, _MAX_RAW_REQUEST_BYTES
+from core.constants import STATIC_DIR, _BufferHandler, _MAX_JSON_BYTES, _MAX_UPLOAD_BYTES, _MAX_RAW_REQUEST_BYTES, _REQUIRE_EXTERNAL_API_KEY
 from core.config import _load_settings, _ensure_external_api_key
 from core.voice_index import refresh_voice_index_background
-from routes import admin, settings, library, stt, sources, docker, tts, conversation, reader, characters, rehearsals_db
+from routes import admin, settings, library, stt, sources, docker, tts, conversation, reader, characters, rehearsals_db, recovery
 
 logger = logging.getLogger("uvicorn.error")
 
@@ -233,11 +233,10 @@ class ApiKeyGateMiddleware:
         if not (path.startswith("/api/") or path == "/mcp"):
             await self.app(scope, receive, send)
             return
-        if not _load_settings().get("external_api_key_required"):
+        if not _REQUIRE_EXTERNAL_API_KEY and not _load_settings().get("external_api_key_required"):
-            # Off by default — see the settings-key comment in core/config.py
+            # Explicit opt-out is intended only for an isolated deployment
-            # for why. /mcp still needs SOME signal it's being used
+            # protected by another authentication layer. Same-origin browser
-            # deliberately even while the general gate is off, but that's a
+            # requests still bypass the external caller gate below.
-            # judgment call for whoever enables it, not a silent bypass.
             await self.app(scope, receive, send)
             return
 
@@ -254,7 +253,9 @@ class ApiKeyGateMiddleware:
                 return False
 
         is_browser_same_origin = path != "/mcp" and (
-            _same_origin(headers.get("origin", "")) or _same_origin(headers.get("referer", ""))
+            headers.get("sec-fetch-site", "").lower() == "same-origin"
+            or _same_origin(headers.get("origin", ""))
+            or _same_origin(headers.get("referer", ""))
         )
         if is_browser_same_origin:
             await self.app(scope, receive, send)
@@ -293,6 +294,7 @@ app.include_router(conversation.router)
 app.include_router(reader.router)
 app.include_router(characters.router)
 app.include_router(rehearsals_db.router)
+app.include_router(recovery.router)
 
 # ── Static files ──────────────────────────────────────────────────────────────