Spaces:
Running
Running
File size: 4,705 Bytes
fba6023 8939f6f fba6023 8939f6f fba6023 8939f6f fba6023 8939f6f fba6023 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 | from __future__ import annotations
import mimetypes
import re
from pathlib import Path
from typing import Any
from app.core.config import Settings
from app.core.exceptions import InputError
MEDIA_EXTENSIONS = {
".3gp", ".aac", ".aiff", ".apng", ".avi", ".bin", ".bmp", ".flac",
".gif", ".heic", ".jpeg", ".jpg", ".m4a", ".m4v", ".mkv", ".m3u",
".m3u8", ".mov", ".mp3", ".mp4", ".mpeg", ".mpg", ".oga", ".ogg",
".opus", ".png", ".svg", ".tif", ".tiff", ".ts", ".vtt", ".wav",
".webm", ".webp", ".wmv", ".srt", ".ass", ".ssa",
}
class MediaValidator:
def __init__(self, settings: Settings) -> None:
self.settings = settings
def safe_filename(self, filename: str | None, fallback: str = "input.bin") -> str:
# Treat both POSIX and Windows separators as path delimiters regardless
# of the server OS; never allow a client path to become a stored name.
raw = str(filename or fallback).replace("\\", "/")
name = raw.rsplit("/", 1)[-1].replace("\x00", "")
if (
not name
or name in {".", ".."}
or any(ord(char) < 32 or ord(char) == 127 for char in name)
or '"' in name
):
raise InputError("Invalid filename")
return name[:240]
def validate_declared(self, filename: str, mime_type: str, size: int) -> None:
if size <= 0:
raise InputError("The uploaded media is empty")
if size > self.settings.max_upload_size:
raise InputError(
"The media exceeds MAX_UPLOAD_SIZE",
details={"size": size, "maximum": self.settings.max_upload_size},
)
suffix = Path(filename).suffix.lower()
if suffix and suffix not in MEDIA_EXTENSIONS:
raise InputError("Unsupported media file extension", details={"extension": suffix})
normalized_mime = mime_type.split(";", 1)[0].lower()
allowed = normalized_mime.startswith(("audio/", "video/", "image/", "text/")) or normalized_mime in {
"application/octet-stream", "application/x-subrip",
"application/vnd.apple.mpegurl", "application/x-mpegurl",
}
if normalized_mime and not allowed:
raise InputError("Unsupported media MIME type", details={"mime_type": mime_type})
def validate_probe(self, metadata: dict[str, Any]) -> None:
streams = [*(metadata.get("video_streams") or []), *(metadata.get("audio_streams") or [])]
if not streams:
raise InputError("The input has no decodable video or audio stream")
for stream in streams:
codec = stream.get("codec")
if not isinstance(codec, str) or not re.fullmatch(r"[a-zA-Z0-9_.-]{1,64}", codec):
raise InputError("The input contains an invalid or unsupported codec")
duration = metadata.get("duration")
if duration and float(duration) > self.settings.max_duration_seconds:
raise InputError(
"Media duration exceeds the configured limit",
details={"duration": duration, "maximum": self.settings.max_duration_seconds},
)
resolution = metadata.get("resolution") or {}
width, height = resolution.get("width", 0), resolution.get("height", 0)
if width and height and width * height > self.settings.max_resolution_pixels:
raise InputError(
"Media resolution exceeds the configured limit",
details={"width": width, "height": height},
)
@staticmethod
def infer_mime(path: Path, fallback: str = "application/octet-stream") -> str:
return mimetypes.guess_type(path.name)[0] or fallback
def as_bool(value: Any, default: bool = False) -> bool:
if value is None:
return default
if isinstance(value, bool):
return value
return str(value).strip().lower() in {"1", "true", "yes", "on"}
def bounded_number(
params: dict[str, Any], name: str, default: float, minimum: float, maximum: float
) -> float:
try:
value = float(params.get(name, default))
except (TypeError, ValueError) as exc:
raise InputError(f"'{name}' must be a number") from exc
if not minimum <= value <= maximum:
raise InputError(f"'{name}' must be between {minimum} and {maximum}")
return value
def positive_int(params: dict[str, Any], name: str, default: int) -> int:
try:
value = int(params.get(name, default))
except (TypeError, ValueError) as exc:
raise InputError(f"'{name}' must be an integer") from exc
if value <= 0:
raise InputError(f"'{name}' must be greater than zero")
return value
|