feat: use binary screenshot transport (Closes #491)

src/main.py

@@ -19,7 +19,8 @@ from fastapi import FastAPI, Header, HTTPException, Path as PathParam, Query, Re
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse, RedirectResponse
 from fastapi.staticfiles import StaticFiles
-from pydantic import BaseModel, Field, PositiveInt, field_validator, model_validator
+from pydantic import BaseModel, Field, PositiveInt, ValidationError, field_validator, model_validator
+from starlette.datastructures import UploadFile
 
 from src import dashboard_auth, gitea_proxy, passkeys
 from src.available_issue_snapshot_store import AvailableIssueSnapshotStore
@@ -359,6 +360,34 @@ class IssueComment(BaseModel):
         return value
 
 
+def _validate_attachment_metadata(filename: str, content_type: str) -> None:
+    expected = {
+        "image/png": {"png"},
+        "image/jpeg": {"jpg", "jpeg"},
+        "image/webp": {"webp"},
+    }
+    if content_type not in expected:
+        raise ValueError("attachment content type must be PNG, JPEG, or WebP")
+    if "/" in filename or "\\" in filename or any(ord(character) < 32 for character in filename):
+        raise ValueError("attachment filename must be a plain file name")
+    extension = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
+    if extension not in expected[content_type]:
+        raise ValueError("attachment filename must match the selected image type")
+
+
+def _validate_attachment_content(content_type: str, content: bytes) -> bytes:
+    if len(content) > 2 * 1024 * 1024:
+        raise ValueError("screenshot must be 2 MB or smaller")
+    signatures = {
+        "image/png": content.startswith(b"\x89PNG\r\n\x1a\n"),
+        "image/jpeg": content.startswith(b"\xff\xd8\xff"),
+        "image/webp": content.startswith(b"RIFF") and content[8:12] == b"WEBP",
+    }
+    if not signatures.get(content_type, False):
+        raise ValueError("file contents do not match the selected image type")
+    return content
+
+
 class IssueAttachment(BaseModel):
     filename: str = Field(min_length=1, max_length=255)
     content_type: Literal["image/png", "image/jpeg", "image/webp"]
@@ -366,20 +395,7 @@ class IssueAttachment(BaseModel):
 
     @model_validator(mode="after")
     def validate_filename(self):
-        if (
-            "/" in self.filename
-            or "\\" in self.filename
-            or any(ord(character) < 32 for character in self.filename)
-        ):
-            raise ValueError("attachment filename must be a plain file name")
-        extension = self.filename.rsplit(".", 1)[-1].lower() if "." in self.filename else ""
-        expected = {
-            "image/png": {"png"},
-            "image/jpeg": {"jpg", "jpeg"},
-            "image/webp": {"webp"},
-        }
-        if extension not in expected[self.content_type]:
-            raise ValueError("attachment filename must match the selected image type")
+        _validate_attachment_metadata(self.filename, self.content_type)
         return self
 
     def content(self) -> bytes:
@@ -387,16 +403,12 @@ class IssueAttachment(BaseModel):
             content = base64.b64decode(self.data, validate=True)
         except (ValueError, binascii.Error) as exc:
             raise ValueError("attachment data must be valid base64") from exc
-        if len(content) > 2 * 1024 * 1024:
-            raise ValueError("screenshot must be 2 MB or smaller")
-        signatures = {
-            "image/png": content.startswith(b"\x89PNG\r\n\x1a\n"),
-            "image/jpeg": content.startswith(b"\xff\xd8\xff"),
-            "image/webp": content.startswith(b"RIFF") and content[8:12] == b"WEBP",
-        }
-        if not signatures[self.content_type]:
-            raise ValueError("file contents do not match the selected image type")
-        return content
+        return _validate_attachment_content(self.content_type, content)
+
+
+def _validate_binary_attachment(filename: str, content_type: str, content: bytes) -> bytes:
+    _validate_attachment_metadata(filename, content_type)
+    return _validate_attachment_content(content_type, content)
 
 
 class IssueCreation(BaseModel):
@@ -3107,7 +3119,7 @@ async def comment_on_assigned_issue(
 
 @app.post("/api/v1/repos/{owner}/{repo}/issues/{number}/attachments", status_code=201)
 async def attach_to_assigned_issue(
-    attachment: IssueAttachment,
+    request: Request,
     owner: str,
     repo: str,
     number: int = PathParam(gt=0),
@@ -3115,15 +3127,27 @@ async def attach_to_assigned_issue(
 ):
     repository = f"{owner}/{repo}"
     try:
-        content = attachment.content()
-    except ValueError as exc:
+        if request.headers.get("content-type", "").lower().startswith("multipart/form-data"):
+            form = await request.form()
+            uploaded = form.get("file")
+            if not isinstance(uploaded, UploadFile):
+                raise ValueError("screenshot file is required")
+            filename = str(uploaded.filename or "")
+            content_type = str(uploaded.content_type or "")
+            content = _validate_binary_attachment(filename, content_type, await uploaded.read())
+        else:
+            attachment = IssueAttachment.model_validate(await request.json())
+            filename = attachment.filename
+            content_type = attachment.content_type
+            content = attachment.content()
+    except (ValueError, ValidationError) as exc:
         raise HTTPException(status_code=422, detail=str(exc)) from exc
     async def upload_attachment():
         result = await gitea_proxy.upload_assigned_issue_attachment(
             repository,
             number,
-            attachment.filename,
-            attachment.content_type,
+            filename,
+            content_type,
             content,
         )
         safe_name = (
@@ -3142,8 +3166,8 @@ async def attach_to_assigned_issue(
                 "issue-attachment",
                 repository,
                 number,
-                attachment.filename,
-                attachment.content_type,
+                filename,
+                content_type,
                 hashlib.sha256(content).hexdigest(),
             ),
             timeout=ISSUE_ACTION_TIMEOUT_SECONDS,

src/request_boundary.py

@@ -5,7 +5,8 @@ from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
 SESSION_BODY_LIMIT = 16 * 1024
 API_MUTATION_BODY_LIMIT = 64 * 1024
-ISSUE_ATTACHMENT_BODY_LIMIT = 3 * 1024 * 1024
+ISSUE_ATTACHMENT_BODY_LIMIT = 2 * 1024 * 1024 + 64 * 1024
+LEGACY_JSON_ATTACHMENT_BODY_LIMIT = 3 * 1024 * 1024
 MUTATION_METHODS = frozenset({"POST", "PUT", "PATCH"})
 
 
@@ -50,6 +51,11 @@ class RequestBodyLimitMiddleware:
         if limit is None:
             await self.app(scope, receive, send)
             return
+        if (
+            limit == ISSUE_ATTACHMENT_BODY_LIMIT
+            and self._header(scope, b"content-type").startswith(b"application/json")
+        ):
+            limit = LEGACY_JSON_ATTACHMENT_BODY_LIMIT
 
         declared_length = self._content_length(scope)
         if declared_length is not None and declared_length > limit:
@@ -77,6 +83,13 @@ class RequestBodyLimitMiddleware:
 
         await self.app(scope, replay, send)
 
+    @staticmethod
+    def _header(scope: Scope, wanted: bytes) -> bytes:
+        for name, value in scope.get("headers", []):
+            if name.lower() == wanted:
+                return value.lower()
+        return b""
+
     @staticmethod
     def _content_length(scope: Scope) -> int | None:
         for name, value in scope.get("headers", []):