SA-010 Avatar validation

backend/app/api/user_config.py

@@ -2,10 +2,13 @@
 ## SPDX-License-Identifier: GPL-3.0-or-later
 
 import json
+import warnings
+from io import BytesIO
 from datetime import datetime, timedelta, timezone
 from uuid import uuid4
 
 from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
+from PIL import Image, UnidentifiedImageError
 from pydantic import BaseModel
 
 from backend.app.api.dependencies import get_current_user
@@ -19,6 +22,9 @@ from backend.app.services.secret_store import decrypt_secret, encrypt_secret
 
 router = APIRouter()
 
+MAX_AVATAR_BYTES = 2 * 1024 * 1024
+MAX_AVATAR_PIXELS = 25_000_000
+
 
 class UserConfigUpdate(BaseModel):
     bio: str | None = None
@@ -320,25 +326,33 @@ async def upload_avatar(
     avatar: UploadFile = File(...),
     user: dict = Depends(get_current_user),
 ):
-    allowed_types = {
-        'image/gif': '.gif',
-        'image/jpeg': '.jpg',
-        'image/png': '.png',
-        'image/webp': '.webp',
-    }
-    suffix = allowed_types.get(avatar.content_type or '')
-    if suffix is None:
+    if avatar.content_type not in {'image/gif', 'image/jpeg', 'image/png', 'image/webp'}:
         raise HTTPException(status_code=415, detail='Avatar must be a PNG, JPEG, GIF, or WebP image')
 
-    contents = await avatar.read(2 * 1024 * 1024 + 1)
-    if len(contents) > 2 * 1024 * 1024:
+    contents = await avatar.read(MAX_AVATAR_BYTES + 1)
+    if len(contents) > MAX_AVATAR_BYTES:
         raise HTTPException(status_code=413, detail='Avatar must be 2 MB or smaller')
 
-    avatar_path = AVATARS_DIR / f'{user["id"]}{suffix}'
+    try:
+        with warnings.catch_warnings():
+            warnings.simplefilter('error', Image.DecompressionBombWarning)
+            with Image.open(BytesIO(contents)) as image:
+                if image.width * image.height > MAX_AVATAR_PIXELS:
+                    raise HTTPException(status_code=413, detail='Avatar dimensions are too large')
+                image.verify()
+            with Image.open(BytesIO(contents)) as image:
+                image.load()
+                normalized = image.convert('RGBA')
+    except HTTPException:
+        raise
+    except (Image.DecompressionBombError, Image.DecompressionBombWarning, UnidentifiedImageError, OSError, ValueError) as error:
+        raise HTTPException(status_code=415, detail='Avatar content is not a valid image') from error
+
+    avatar_path = AVATARS_DIR / f'{user["id"]}.png'
+    normalized.save(avatar_path, format='PNG', optimize=True)
     for existing_path in AVATARS_DIR.glob(f'{user["id"]}.*'):
         if existing_path != avatar_path:
             existing_path.unlink(missing_ok=True)
-    avatar_path.write_bytes(contents)
     avatar_url = f'/media/{avatar_path.name}'
 
     with get_connection() as conn:

backend/requirements.txt

@@ -3,6 +3,7 @@ uvicorn==0.52.4
 pydantic==2.13.4
 jinja2==3.1.6
 python-multipart==0.0.20
+Pillow==11.3.0
 pytest==9.1.1
 httpx==0.28.1
 httpx2==2.12.0

backend/tests/test_user_config.py

@@ -1,7 +1,10 @@
 ## Copyright © 2026 Olaf Kolkman
 ## SPDX-License-Identifier: GPL-3.0-or-later
 
+from io import BytesIO
+
 from fastapi.testclient import TestClient
+from PIL import Image
 from unittest.mock import patch
 
 from backend.app.main import app
@@ -68,15 +71,29 @@ def test_user_config_api_and_profile_page():
         'new_password': 'secret123',
     }, headers=bob_headers).status_code == 200
 
+    image_buffer = BytesIO()
+    Image.new('RGB', (2, 2), 'red').save(image_buffer, format='JPEG')
     upload_response = client.post(
         '/api/user/avatar',
         headers=headers,
-        files={'avatar': ('avatar.png', b'fake-png-data', 'image/png')},
+        files={'avatar': ('avatar.jpg', image_buffer.getvalue(), 'image/jpeg')},
     )
     assert upload_response.status_code == 200
     avatar_url = upload_response.json()['avatar_url']
     assert avatar_url.startswith('/media/user-1.png')
-    assert client.get(avatar_url).content == b'fake-png-data'
+    stored_avatar = client.get(avatar_url)
+    assert stored_avatar.status_code == 200
+    assert stored_avatar.headers['content-type'] == 'image/png'
+    with Image.open(BytesIO(stored_avatar.content)) as image:
+        assert image.format == 'PNG'
+        assert image.size == (2, 2)
+
+    rejected_upload = client.post(
+        '/api/user/avatar',
+        headers=headers,
+        files={'avatar': ('avatar.png', b'fake-png-data', 'image/png')},
+    )
+    assert rejected_upload.status_code == 415
 
     updated_profile = client.get('/api/user/me', headers=headers).json()
     assert updated_profile['avatar_url'] == avatar_url