Merge pull request #1224 from rommapp/fix/escape-download-urls-when-s…

…erving-nginx

fix: Escape URLs when files are served by nginx

backend/endpoints/rom.py

@@ -29,7 +29,7 @@
 from streaming_form_data.targets import FileTarget, NullTarget
 from utils.filesystem import sanitize_filename
 from utils.hashing import crc32_to_hex
-from utils.nginx import ZipContentLine, ZipResponse
+from utils.nginx import FileRedirectResponse, ZipContentLine, ZipResponse
 from utils.router import APIRouter
 
 router = APIRouter()
@@ -197,21 +197,14 @@ async def head_rom_content(
                 },
             )
 
-        return Response(
-            media_type="application/octet-stream",
-            headers={
-                "Content-Disposition": f'attachment; filename="{quote(rom.file_name)}"',
-                "X-Accel-Redirect": f"/library/{rom.full_path}",
-            },
+        return FileRedirectResponse(
+            download_path=Path(f"/library/{rom.full_path}"),
+            filename=rom.file_name,
         )
 
     if len(files_to_check) == 1:
-        return Response(
-            media_type="application/octet-stream",
-            headers={
-                "Content-Disposition": f'attachment; filename="{quote(files_to_check[0])}"',
-                "X-Accel-Redirect": f"/library/{rom.full_path}/{files_to_check[0]}",
-            },
+        return FileRedirectResponse(
+            download_path=Path(f"/library/{rom.full_path}/{files_to_check[0]}"),
         )
 
     return Response(
@@ -256,21 +249,14 @@ async def get_rom_content(
     files_to_download = sorted(files or [r["filename"] for r in rom.files])
 
     if not rom.multi:
-        return Response(
-            media_type="application/octet-stream",
-            headers={
-                "Content-Disposition": f'attachment; filename="{quote(rom.file_name)}"',
-                "X-Accel-Redirect": f"/library/{rom.full_path}",
-            },
+        return FileRedirectResponse(
+            download_path=Path(f"/library/{rom.full_path}"),
+            filename=rom.file_name,
         )
 
     if len(files_to_download) == 1:
-        return Response(
-            media_type="application/octet-stream",
-            headers={
-                "Content-Disposition": f'attachment; filename="{quote(files_to_download[0])}"',
-                "X-Accel-Redirect": f"/library/{rom.full_path}/{files_to_download[0]}",
-            },
+        return FileRedirectResponse(
+            download_path=Path(f"/library/{rom.full_path}/{files_to_download[0]}"),
         )
 
     content_lines = [

backend/utils/filesystem.py

@@ -3,6 +3,11 @@
 from collections.abc import Iterator
 from pathlib import Path
 
+from anyio import Path as AnyIOPath
+
+# Type alias for a path that can be either a `pathlib.Path` or an `anyio.Path`.
+AnyPath = Path | AnyIOPath
+
 
 def iter_files(path: str, recursive: bool = False) -> Iterator[tuple[Path, str]]:
     """List files in a directory.

backend/utils/nginx.py

@@ -1,8 +1,10 @@
 import dataclasses
 from collections.abc import Collection
 from typing import Any
+from urllib.parse import quote
 
 from fastapi.responses import Response
+from utils.filesystem import AnyPath
 
 
 @dataclasses.dataclass(frozen=True)
@@ -47,3 +49,27 @@ def __init__(
         )
 
         super().__init__(**kwargs)
+
+
+class FileRedirectResponse(Response):
+    """Response class for serving a file download by using the X-Accel-Redirect header."""
+
+    def __init__(
+        self, *, download_path: AnyPath, filename: str | None = None, **kwargs: Any
+    ):
+        """
+        Arguments:
+          - download_path: Path to the file to be served.
+          - filename: Name of the file to be served. If not provided, the file name from the
+              download_path is used.
+        """
+        media_type = kwargs.pop("media_type", "application/octet-stream")
+        filename = filename or download_path.name
+        kwargs.setdefault("headers", {}).update(
+            {
+                "Content-Disposition": f'attachment; filename="{quote(filename)}"',
+                "X-Accel-Redirect": quote(str(download_path)),
+            }
+        )
+
+        super().__init__(media_type=media_type, **kwargs)