security: sanitize image evidence server-side (Closes #983)

tests/test_global_search.py

@@ -1,11 +1,19 @@
 import asyncio
+import io
 
 import httpx
 import pytest
+from PIL import Image
 
 from src import gitea_proxy, main
 
 
+def png_bytes(color="red"):
+    output = io.BytesIO()
+    Image.new("RGB", (2, 2), color).save(output, format="PNG")
+    return output.getvalue()
+
+
 @pytest.mark.anyio
 async def test_global_search_endpoint_scopes_every_page_to_an_exact_repository(monkeypatch):
     requested = []
@@ -343,7 +351,7 @@ async def test_global_search_preview_attachment_is_idempotent_for_exact_visible_
     monkeypatch.setattr(main.gitea_proxy, "upload_preview_attachment", upload)
     transport = httpx.ASGITransport(app=main.app)
     headers = {"Idempotency-Key": "search-photo-pull-42"}
-    image = b"\x89PNG\r\n\x1a\nsafe-pixels"
+    image = png_bytes()
     files = {"file": ("photo.png", image, "image/png")}
     async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
         first = await client.post(
@@ -381,7 +389,7 @@ async def test_global_search_preview_attachment_rejects_kind_mismatch_before_upl
     async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
         response = await client.post(
             "/api/v1/repos/stackchain/api/issues/42/preview/attachments?kind=pull",
-            files={"file": ("photo.png", b"\x89PNG\r\n\x1a\nsafe-pixels", "image/png")},
+            files={"file": ("photo.png", png_bytes(), "image/png")},
             headers={"Idempotency-Key": "search-photo-mismatch-42"},
         )
 

tests/test_image_sanitizer.py

@@ -0,0 +1,63 @@
+import io
+from pathlib import Path
+
+import pytest
+from PIL import Image
+
+from src.image_sanitizer import MAX_IMAGE_PIXELS, sanitize_image
+
+
+def test_runtime_dependencies_install_the_image_decoder():
+    requirements = (Path(__file__).parents[1] / "requirements.txt").read_text().splitlines()
+    assert any(line.startswith("Pillow==") for line in requirements)
+
+
+def encoded_image(format_name, size=(3, 2), *, exif=None):
+    image = Image.new("RGB", size)
+    image.putdata(
+        [(255, 0, 0), (0, 255, 0), (0, 0, 255), (255, 255, 0), (0, 255, 255), (255, 0, 255)]
+    )
+    output = io.BytesIO()
+    image.save(output, format=format_name, exif=exif or b"")
+    return output.getvalue()
+
+
+def test_sanitize_image_applies_orientation_and_removes_jpeg_metadata():
+    exif = Image.Exif()
+    exif[274] = 6
+    exif[ GPS_TAG := 34853 ] = {1: "N", 2: (1, 1, 1)}
+    source = encoded_image("JPEG", exif=exif)
+
+    sanitized = sanitize_image("image/jpeg", source)
+
+    with Image.open(io.BytesIO(sanitized)) as result:
+        assert result.size == (2, 3)
+        assert not result.getexif()
+        assert GPS_TAG not in result.info
+        assert result.format == "JPEG"
+
+
+def test_sanitize_image_rejects_bytes_that_do_not_match_claimed_type():
+    source = encoded_image("PNG")
+
+    with pytest.raises(ValueError, match="does not match"):
+        sanitize_image("image/jpeg", source)
+
+
+def test_sanitize_image_rejects_over_twelve_megapixels_before_decoding():
+    image = Image.new("RGB", (4096, 3073))
+    output = io.BytesIO()
+    image.save(output, format="PNG")
+
+    with pytest.raises(ValueError, match="12 megapixels"):
+        sanitize_image("image/png", output.getvalue())
+
+
+def test_sanitize_image_rejects_animated_webp():
+    first = Image.new("RGB", (2, 2), "red")
+    second = Image.new("RGB", (2, 2), "blue")
+    output = io.BytesIO()
+    first.save(output, format="WEBP", save_all=True, append_images=[second], duration=100)
+
+    with pytest.raises(ValueError, match="animated"):
+        sanitize_image("image/webp", output.getvalue())

tests/test_issue_attachments.py

@@ -1,13 +1,49 @@
 import asyncio
 import base64
+import io
+import time
 
 import httpx
 import pytest
+from PIL import Image, PngImagePlugin
 
 from src import gitea_proxy, main
 
 
-PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"mobile screenshot"
+def make_png(color="red", size=(2, 2)):
+    output = io.BytesIO()
+    Image.new("RGB", size, color).save(output, format="PNG")
+    return output.getvalue()
+
+
+PNG_BYTES = make_png()
+
+
+def png_with_private_metadata():
+    output = io.BytesIO()
+    metadata = PngImagePlugin.PngInfo()
+    metadata.add_text("Location", "private coordinates")
+    Image.new("RGB", (2, 2), "red").save(output, format="PNG", pnginfo=metadata)
+    return output.getvalue()
+
+
+@pytest.mark.anyio
+async def test_attachment_sanitization_keeps_event_loop_responsive(monkeypatch):
+    def slow_sanitizer(_content_type, content):
+        time.sleep(0.1)
+        return content
+
+    monkeypatch.setattr(main, "sanitize_image", slow_sanitizer)
+    started = time.monotonic()
+    ticks = []
+
+    async def tick():
+        await asyncio.sleep(0.01)
+        ticks.append(time.monotonic() - started)
+
+    await asyncio.gather(main._sanitize_attachment("image/png", b"image"), tick())
+
+    assert ticks[0] < 0.05
 
 
 @pytest.fixture(autouse=True)
@@ -58,6 +94,30 @@ async def test_attachment_endpoint_uploads_valid_screenshot_to_assigned_issue(mo
     ]
 
 
+@pytest.mark.anyio
+async def test_attachment_endpoint_strips_metadata_before_upstream(monkeypatch):
+    uploaded = []
+    source = png_with_private_metadata()
+
+    async def upload(_repository, _number, filename, content_type, content):
+        uploaded.append(content)
+        return {"name": filename, "url": "https://forge.example/a.png", "size": len(content)}
+
+    monkeypatch.setattr(main.gitea_proxy, "upload_assigned_issue_attachment", upload)
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.post(
+            "/api/v1/repos/stackchain/api/issues/17/attachments",
+            files={"file": ("evidence.png", source, "image/png")},
+        )
+
+    assert response.status_code == 201
+    assert uploaded and uploaded[0] != source
+    with Image.open(io.BytesIO(uploaded[0])) as sanitized:
+        assert sanitized.info.get("Location") is None
+        assert sanitized.getpixel((0, 0)) == (255, 0, 0)
+
+
 @pytest.mark.anyio
 async def test_attachment_endpoint_accepts_binary_multipart_without_base64_expansion(monkeypatch):
     calls = []
@@ -186,7 +246,7 @@ async def test_attachment_endpoint_rejects_changed_upload_for_used_key(monkeypat
             json={
                 "filename": "checkout.png",
                 "content_type": "image/png",
-                "data": base64.b64encode(PNG_BYTES + b" changed").decode("ascii"),
+                "data": base64.b64encode(make_png("blue")).decode("ascii"),
             },
             headers={"Idempotency-Key": "attachment-conflict-471"},
         )
@@ -198,7 +258,7 @@ async def test_attachment_endpoint_rejects_changed_upload_for_used_key(monkeypat
 
 @pytest.mark.anyio
 async def test_attachment_endpoint_admits_a_normal_phone_screenshot(monkeypatch):
-    screenshot = b"\x89PNG\r\n\x1a\n" + (b"x" * (100 * 1024))
+    screenshot = make_png(size=(390, 844))
 
     async def upload(_repository, _number, filename, _content_type, content):
         return {

tests/test_update_reply_attachments.py

@@ -1,9 +1,11 @@
+import io
 import json
 import subprocess
 from pathlib import Path
 
 import httpx
 import pytest
+from PIL import Image
 
 from src import gitea_proxy, main
 from tests.dashboard_bundle import dashboard
@@ -12,7 +14,9 @@ from tests.dashboard_bundle import dashboard
 ROOT = Path(__file__).parents[1]
 OUTBOX = ROOT / "frontend" / "authored-outbox.js"
 SYNC = ROOT / "frontend" / "background-issue-sync.js"
-PNG = b"\x89PNG\r\n\x1a\nmobile-update"
+_png = io.BytesIO()
+Image.new("RGB", (2, 2), "red").save(_png, format="PNG")
+PNG = _png.getvalue()
 
 
 def run_node(script: str):