135 lines
6.2 KiB
Python
135 lines
6.2 KiB
Python
"""URL validation and safe path helpers."""
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import re
|
|
import socket
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit
|
|
|
|
from fastapi import HTTPException
|
|
|
|
|
|
def _normalize_service_url(url: str) -> str:
|
|
"""Replace 0.0.0.0 with host.docker.internal so server-side probes reach the host."""
|
|
return re.sub(r"(https?://)0\.0\.0\.0([\/:$])", r"\1host.docker.internal\2", str(url or ""))
|
|
|
|
|
|
def _validate_http_url(raw: str, *, allow_private: bool = True) -> str:
|
|
raw = _normalize_service_url(str(raw or "")).strip()
|
|
if not raw:
|
|
raise HTTPException(400, "URL is required")
|
|
parts = urlsplit(raw)
|
|
if parts.scheme not in {"http", "https"}:
|
|
raise HTTPException(400, "Only http:// and https:// URLs are allowed")
|
|
if not parts.hostname:
|
|
raise HTTPException(400, "URL must include a hostname")
|
|
if parts.username or parts.password:
|
|
raise HTTPException(400, "URLs with embedded credentials are not allowed")
|
|
if allow_private:
|
|
return raw
|
|
try:
|
|
infos = socket.getaddrinfo(parts.hostname, parts.port or (443 if parts.scheme == "https" else 80), type=socket.SOCK_STREAM)
|
|
except socket.gaierror:
|
|
raise HTTPException(400, "URL hostname could not be resolved")
|
|
for info in infos:
|
|
ip = ipaddress.ip_address(info[4][0])
|
|
if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved:
|
|
raise HTTPException(400, "Private, local, and reserved network URLs are not allowed for downloads")
|
|
return raw
|
|
|
|
|
|
def _copy_limited(src, dest, limit: int) -> int:
|
|
total = 0
|
|
while True:
|
|
chunk = src.read(1024 * 1024)
|
|
if not chunk:
|
|
break
|
|
total += len(chunk)
|
|
if total > limit:
|
|
raise HTTPException(413, "Uploaded file is too large")
|
|
dest.write(chunk)
|
|
return total
|
|
|
|
|
|
def _validated_upload_suffix(filename: str | None, *, default: str, allowed: set[str]) -> str:
|
|
"""Validate a browser upload name and return its normalized suffix."""
|
|
name = str(filename or default)
|
|
if not name or len(name) > 255 or any(ord(char) < 32 or ord(char) == 127 for char in name):
|
|
raise HTTPException(400, "Invalid upload filename")
|
|
if "/" in name or "\\" in name or name in {".", ".."}:
|
|
raise HTTPException(400, "Upload filename must not contain a path")
|
|
suffix = Path(name).suffix.lower() or default
|
|
if suffix not in allowed:
|
|
raise HTTPException(400, "Unsupported file type")
|
|
return suffix
|
|
|
|
|
|
def _safe_child_path(root: Path, candidate: Path) -> Path:
|
|
root_resolved = root.resolve()
|
|
candidate_resolved = candidate.resolve()
|
|
try:
|
|
candidate_resolved.relative_to(root_resolved)
|
|
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")
|
|
sheet = record.get("sheet") or {}
|
|
forms = sheet.get("forms", [])
|
|
if forms is not None and not isinstance(forms, list):
|
|
raise HTTPException(400, "Character sheet forms must be an array")
|
|
if isinstance(forms, list) and len(forms) > 8:
|
|
raise HTTPException(413, "A character may have at most 8 forms")
|
|
for form in forms or []:
|
|
if not isinstance(form, dict):
|
|
raise HTTPException(400, "Each character form must be an object")
|
|
for field, max_length in {
|
|
"id": 120, "name": 160, "physical": 12000, "clothing": 12000,
|
|
"image_prompt": 8000, "concept_art_prompt": 8000,
|
|
"image": 24 * 1024 * 1024, "concept_art_image": 24 * 1024 * 1024,
|
|
"consistent_prompt_version": 120, "consistent_identity_fingerprint": 120,
|
|
}.items():
|
|
value = form.get(field, "")
|
|
if value is not None and not isinstance(value, str):
|
|
raise HTTPException(400, f"Character form field '{field}' must be text")
|
|
if isinstance(value, str) and len(value) > max_length:
|
|
raise HTTPException(413, f"Character form field '{field}' is too large")
|
|
for field in ("inventory", "visual_props"):
|
|
value = form.get(field, [])
|
|
if value is not None and not isinstance(value, list):
|
|
raise HTTPException(400, f"Character form field '{field}' must be an array")
|
|
if isinstance(value, list) and len(value) > 12:
|
|
raise HTTPException(413, f"Character form field '{field}' has too many items")
|
|
if isinstance(value, list) and any(not isinstance(item, str) or len(item) > 300 for item in value):
|
|
raise HTTPException(400, f"Character form field '{field}' contains an invalid item")
|
|
active_form_id = sheet.get("active_form_id", "")
|
|
if active_form_id is not None and not isinstance(active_form_id, str):
|
|
raise HTTPException(400, "Character active form id must be text")
|
|
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
|