fix: validate uploaded media files

zyrarbot/audio/media_probe.py

@@ -29,3 +29,24 @@ class MediaProbe:
         except (OSError, ValueError) as error:
             logger.error("ffprobe nie odczytał długości pliku %s: %s", filepath, error)
         return 0.0
+
+    async def is_media(self, filepath: str) -> bool:
+        """Return whether ffprobe recognizes the file as an audio/video container."""
+        try:
+            process = await asyncio.create_subprocess_exec(
+                "ffprobe",
+                "-v",
+                "error",
+                "-show_entries",
+                "format=format_name",
+                "-of",
+                "default=noprint_wrappers=1:nokey=1",
+                filepath,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            stdout, _ = await process.communicate()
+            return process.returncode == 0 and bool(stdout.decode().strip())
+        except OSError as error:
+            logger.error("Nie udało się uruchomić ffprobe dla %s: %s", filepath, error)
+            return False

zyrarbot/storage/uploads_repository.py

@@ -6,6 +6,30 @@ from typing import BinaryIO
 
 from zyrarbot.storage.upload_file_info import UploadFileInfo
 
+SUPPORTED_MEDIA_EXTENSIONS = frozenset(
+    {
+        ".aac",
+        ".avi",
+        ".flac",
+        ".flv",
+        ".m4a",
+        ".m4v",
+        ".mkv",
+        ".mov",
+        ".mp3",
+        ".mp4",
+        ".mpeg",
+        ".mpg",
+        ".oga",
+        ".ogg",
+        ".opus",
+        ".ts",
+        ".wav",
+        ".webm",
+        ".wma",
+    }
+)
+
 
 class UploadsRepository:
     """Katalog z plikami wgranymi na serwer; nazwy zawsze sprowadzane do basename."""
@@ -18,6 +42,10 @@ class UploadsRepository:
     def directory(self) -> str:
         return self._directory
 
+    @staticmethod
+    def is_supported_filename(filename: str) -> bool:
+        return os.path.splitext(filename)[1].lower() in SUPPORTED_MEDIA_EXTENSIONS
+
     def _guild_directory(self, guild_id: int | None) -> str:
         if guild_id is None:
             return self._directory

zyrarbot/ui/api/routers/uploads_router.py

@@ -1,10 +1,13 @@
 from __future__ import annotations
 
+import shutil
+import tempfile
 from typing import Any
 
 import discord
 from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
 
+from zyrarbot.audio.media_probe import MediaProbe
 from zyrarbot.storage.uploads_repository import UploadsRepository
 from zyrarbot.ui.api.dependencies import get_active_guild, get_uploads
 
@@ -25,8 +28,22 @@ async def upload(
     guild: discord.Guild = Depends(get_active_guild),
     uploads: UploadsRepository = Depends(get_uploads),
 ) -> dict[str, str]:
+    filename = file.filename or ""
+    if not uploads.is_supported_filename(filename):
+        raise HTTPException(
+            status_code=415,
+            detail="Unsupported file type. Upload an audio or video file.",
+        )
+
+    suffix = filename[filename.rfind(".") :]
+    with tempfile.NamedTemporaryFile(suffix=suffix) as temporary:
+        shutil.copyfileobj(file.file, temporary)
+        temporary.flush()
+        if not await MediaProbe().is_media(temporary.name):
+            raise HTTPException(status_code=415, detail="The uploaded file is not valid media")
+    file.file.seek(0)
     try:
-        filename = uploads.save(file.filename or "plik", file.file, guild.id)
+        filename = uploads.save(filename, file.file, guild.id)
     except OSError as error:
         raise HTTPException(status_code=500, detail=f"Błąd zapisu pliku: {error}") from error
     return {"status": "success", "filename": filename}