91 lines
3.6 KiB
Python
91 lines
3.6 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 _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")
|
|
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
|