Spaces:
Running
Running
| 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}, | |
| ) | |
| 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 | |