fix(gallery): validate album rename and image_ids input (#6822) (#6823)

routes/gallery/gallery_routes.py

@@ -306,6 +306,23 @@ def _b64_bytes(value: str, *, label: str = "image") -> bytes:
         raise HTTPException(400, f"Invalid {label}") from exc
 
 
+async def _json_object(request: Request) -> dict:
+    try:
+        data = await request.json()
+    except ValueError:
+        raise HTTPException(400, "Invalid JSON body")
+    if not isinstance(data, dict):
+        raise HTTPException(400, "JSON object body required")
+    return data
+
+
+async def _image_ids(request: Request) -> list:
+    ids = (await _json_object(request)).get("image_ids", [])
+    if not isinstance(ids, list) or not all(isinstance(i, str) for i in ids):
+        raise HTTPException(400, "image_ids must be a list of image ids")
+    return ids
+
+
 def _pil_image_to_b64(img, *, fmt: str = "PNG") -> str:
     buf = io.BytesIO()
     img.save(buf, format=fmt)
@@ -2944,14 +2961,22 @@ def setup_gallery_routes() -> APIRouter:
     @router.put("/api/gallery/albums/{album_id}")
     async def update_album(request: Request, album_id: str):
         user = get_current_user(request)
-        data = await request.json()
+        data = await _json_object(request)
+        name = data.get("name")
+        if name is not None:
+            if not isinstance(name, str) or not name.strip():
+                raise HTTPException(400, "Album name required")
+            name = name.strip()
+        description = data.get("description")
+        if description is not None and not isinstance(description, str):
+            raise HTTPException(400, "Album description must be a string")
         db = SessionLocal()
         try:
             album = _get_or_404_album(db, album_id, user)
-            if data.get("name") is not None:
-                album.name = data["name"]
-            if data.get("description") is not None:
-                album.description = data["description"]
+            if name is not None:
+                album.name = name
+            if description is not None:
+                album.description = description
             if data.get("cover_id") is not None:
                 cover_id = data["cover_id"] or None
                 if cover_id:
@@ -2981,8 +3006,7 @@ def setup_gallery_routes() -> APIRouter:
     @router.post("/api/gallery/albums/{album_id}/add")
     async def add_to_album(request: Request, album_id: str):
         user = get_current_user(request)
-        data = await request.json()
-        ids = data.get("image_ids", [])
+        ids = await _image_ids(request)
         db = SessionLocal()
         try:
             _get_or_404_album(db, album_id, user)
@@ -2999,8 +3023,7 @@ def setup_gallery_routes() -> APIRouter:
     @router.post("/api/gallery/albums/{album_id}/remove")
     async def remove_from_album(request: Request, album_id: str):
         user = get_current_user(request)
-        data = await request.json()
-        ids = data.get("image_ids", [])
+        ids = await _image_ids(request)
         db = SessionLocal()
         try:
             _get_or_404_album(db, album_id, user)

tests/test_gallery_album_input.py

@@ -0,0 +1,49 @@
+import asyncio
+
+import pytest
+from fastapi import HTTPException
+
+import routes.gallery_routes as gallery
+
+
+def route(path, method):
+    return next(r for r in gallery.setup_gallery_routes().routes if r.path == path and method in r.methods)
+
+
+def call(path, method, monkeypatch, body):
+    monkeypatch.setattr(gallery, 'get_current_user', lambda request: 'alice')
+
+    class Request:
+        async def json(self):
+            return body
+
+    return asyncio.run(route(path, method).endpoint(Request(), 'album-1'))
+
+
+@pytest.mark.parametrize('body', [
+    {'name': ''},
+    {'name': '   '},
+    {'name': 123},
+    {'name': ['x']},
+    {'description': {'a': 1}},
+    [1],
+])
+def test_album_update_rejects_bad_input(body, monkeypatch):
+    with pytest.raises(HTTPException) as exc:
+        call('/api/gallery/albums/{album_id}', 'PUT', monkeypatch, body)
+    assert exc.value.status_code == 400
+
+
+@pytest.mark.parametrize('op', ['add', 'remove'])
+@pytest.mark.parametrize('body', [
+    {'image_ids': 'abc'},
+    {'image_ids': 5},
+    {'image_ids': None},
+    {'image_ids': [{'a': 1}]},
+    {'image_ids': ['ok', 1]},
+    [1],
+])
+def test_album_image_ids_must_be_a_list_of_strings(op, body, monkeypatch):
+    with pytest.raises(HTTPException) as exc:
+        call(f'/api/gallery/albums/{{album_id}}/{op}', 'POST', monkeypatch, body)
+    assert exc.value.status_code == 400