#553 - File type validation on upload through web browser

Author: alitheg

atr/detection.py

@@ -65,32 +65,32 @@ def validate_directory(directory: pathlib.Path) -> list[str]:
             errors.append(f"{path.name}: Symbolic links are not allowed")
             continue
         if path.is_file():
-            if error := _validate_file(path):
+            if error := validate_file(path):
                 errors.append(error)
     return errors
 
 
-def _suffix(filename: str) -> str:
-    name = filename.lower()
-    for compound in _COMPOUND_SUFFIXES:
-        if name.endswith(compound):
-            return compound
-    return pathlib.Path(name).suffix
-
-
-def _validate_file(path: pathlib.Path) -> str | None:
+def validate_file(path: pathlib.Path) -> str | None:
     # TODO: Report errors using the whole relative path, not just the filename
     suffix = _suffix(path.name)
     if suffix not in _EXPECTED:
-        return None
+        return f"{path.name}: Unsupported file format ({suffix})"
     if path.stat().st_size == 0:
         return f"{path.name}: Empty file"
     try:
         results = puremagic.magic_file(path)
     except puremagic.PureError:
         return f"{path.name}: Unidentified file format (expected {suffix})"
-    detected_types = {r.mime_type for r in results}
+    detected_types = {r.mime_type for r in results if r.mime_type != ""}
     if not (detected_types & _EXPECTED[suffix]):
-        primary = results[0].mime_type if results else "unknown"
+        primary = next(iter(detected_types)) if len(detected_types) > 0 else "unknown"
         return f"{path.name}: Content mismatch (expected {suffix}, detected {primary})"
     return None
+
+
+def _suffix(filename: str) -> str:
+    name = filename.lower()
+    for compound in _COMPOUND_SUFFIXES:
+        if name.endswith(compound):
+            return compound
+    return pathlib.Path(name).suffix

atr/post/upload.py

@@ -28,6 +28,7 @@
 
 import atr.blueprints.post as post
 import atr.db as db
+import atr.detection as detection
 import atr.form as form
 import atr.get as get
 import atr.log as log
@@ -136,6 +137,9 @@ async def stage(
             # 1 MiB chunks
             while chunk := await asyncio.to_thread(file.stream.read, 1024 * 1024):
                 await f.write(chunk)
+        content_error = detection.validate_file(path=target_path)
+        if content_error:
+            raise ValueError(content_error)
     except Exception as e:
         log.exception("Error staging file:")
         if await aiofiles.os.path.exists(target_path):