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: name = Path(filename or fallback).name.replace("\x00", "") if not name or name in {".", ".."}: 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