From a8bea9007e911af3387a58b5c84bd716e9ae0dcb Mon Sep 17 00:00:00 2001 From: Jamo Date: Mon, 25 May 2026 00:25:28 +0200 Subject: [PATCH] fix(uploads): cap attachment size and refuse browser-executable MIME types --- backend/testjam/core/config.py | 23 +++ backend/testjam/routers/bugs.py | 5 +- backend/testjam/routers/cases.py | 6 +- .../testjam/routers/executions/attachments.py | 10 +- backend/testjam/routers/versions.py | 5 +- backend/testjam/services/uploads.py | 21 +++ backend/tests/security/test_upload_caps.py | 151 ++++++++++++++++++ 7 files changed, 210 insertions(+), 11 deletions(-) create mode 100644 backend/tests/security/test_upload_caps.py diff --git a/backend/testjam/core/config.py b/backend/testjam/core/config.py index 7b9c270..8db79f5 100644 --- a/backend/testjam/core/config.py +++ b/backend/testjam/core/config.py @@ -37,6 +37,29 @@ class Settings(BaseSettings): # without holding the whole payload in a single Python bytes object. EXPORT_STREAM_CHUNK_BYTES: int = 64 * 1024 + # Per-file cap on attachments uploaded to bugs / cases / executions / + # results / versions. The upload helper rejects anything larger with a + # 413 before it ever hits disk. Default of 25 MiB is generous enough + # for screenshots, junit bundles, and short screen recordings without + # opening an obvious DOS by disk fill. + MAX_ATTACHMENT_BYTES: int = 25 * 1024 * 1024 + + # MIME types refused on attachment upload (415). Defaults block the + # vectors that turn a download endpoint into a stored-XSS or RCE + # primitive: anything the victim's browser will execute by itself + # (HTML / JS / SVG with script support) and the obvious native + # executable types. The list is a CSV so it can be overridden per + # deployment via the BLOCKED_ATTACHMENT_MIME_TYPES env var. + BLOCKED_ATTACHMENT_MIME_TYPES: str = ( + "text/html,application/xhtml+xml,application/javascript," + "text/javascript,image/svg+xml,application/x-msdownload," + "application/x-httpd-php,text/x-php" + ) + + @property + def blocked_attachment_mime_types(self) -> set[str]: + return {item.strip().lower() for item in self.BLOCKED_ATTACHMENT_MIME_TYPES.split(",") if item.strip()} + @property def cors_origins_list(self) -> list[str]: return [origin.strip() for origin in self.CORS_ORIGINS.split(",") if origin.strip()] diff --git a/backend/testjam/routers/bugs.py b/backend/testjam/routers/bugs.py index 469f529..26eb954 100644 --- a/backend/testjam/routers/bugs.py +++ b/backend/testjam/routers/bugs.py @@ -694,17 +694,18 @@ def upload_attachment( bug = _get_bug(db, id) _require_writer(db, current, bug.project_id) _ensure_project_writable(db, bug.project_id) + payload = uploads.read_attachment(file) destination_directory = os.path.join(BUG_UPLOAD_DIR, str(id)) os.makedirs(destination_directory, exist_ok=True) stored_filename = uploads.safe_filename(file.filename) destination_path = uploads.safe_join(destination_directory, file.filename) with open(destination_path, "wb") as handle: - shutil.copyfileobj(file.file, handle) + handle.write(payload) attachment = BugAttachment( bug_id=id, filename=stored_filename, content_type=file.content_type, - size_bytes=os.path.getsize(destination_path), + size_bytes=len(payload), file_path=destination_path, uploaded_by_id=current.id, ) diff --git a/backend/testjam/routers/cases.py b/backend/testjam/routers/cases.py index 32ff3ba..ca3e992 100644 --- a/backend/testjam/routers/cases.py +++ b/backend/testjam/routers/cases.py @@ -413,18 +413,18 @@ def upload_attachment( if not case: raise HTTPException(status_code=404, detail="Not found") _require_case_writer(db, current, _project_id_for_case(case, db)) + payload = uploads.read_attachment(file) dest_dir = os.path.join(UPLOAD_DIR, str(id)) os.makedirs(dest_dir, exist_ok=True) stored_filename = uploads.safe_filename(file.filename) dest_path = uploads.safe_join(dest_dir, file.filename) with open(dest_path, "wb") as f: - shutil.copyfileobj(file.file, f) - size = os.path.getsize(dest_path) + f.write(payload) attachment = Attachment( test_case_id=id, filename=stored_filename, content_type=file.content_type, - size_bytes=size, + size_bytes=len(payload), file_path=dest_path, uploaded_by=current.username, ) diff --git a/backend/testjam/routers/executions/attachments.py b/backend/testjam/routers/executions/attachments.py index 92a729a..ad710f5 100644 --- a/backend/testjam/routers/executions/attachments.py +++ b/backend/testjam/routers/executions/attachments.py @@ -52,17 +52,18 @@ def upload_execution_attachment( ): if not db.get(TestExecution, id): raise HTTPException(status_code=404, detail="Not found") + payload = uploads.read_attachment(file) dest_dir = os.path.join(EXECUTION_UPLOAD_DIR, str(id)) os.makedirs(dest_dir, exist_ok=True) stored_filename = uploads.safe_filename(file.filename) dest_path = uploads.safe_join(dest_dir, file.filename) with open(dest_path, "wb") as f: - shutil.copyfileobj(file.file, f) + f.write(payload) att = ExecutionAttachment( execution_id=id, filename=stored_filename, content_type=file.content_type, - size_bytes=os.path.getsize(dest_path), + size_bytes=len(payload), file_path=dest_path, uploaded_by=current.username, ) @@ -126,17 +127,18 @@ def upload_result_attachment( ): if not db.get(TestResult, id): raise HTTPException(status_code=404, detail="Not found") + payload = uploads.read_attachment(file) dest_dir = os.path.join(UPLOAD_DIR, str(id)) os.makedirs(dest_dir, exist_ok=True) stored_filename = uploads.safe_filename(file.filename) dest_path = uploads.safe_join(dest_dir, file.filename) with open(dest_path, "wb") as f: - shutil.copyfileobj(file.file, f) + f.write(payload) att = ResultAttachment( result_id=id, filename=stored_filename, content_type=file.content_type, - size_bytes=os.path.getsize(dest_path), + size_bytes=len(payload), file_path=dest_path, uploaded_by=current.username, ) diff --git a/backend/testjam/routers/versions.py b/backend/testjam/routers/versions.py index a1e18ae..7b8f1e3 100644 --- a/backend/testjam/routers/versions.py +++ b/backend/testjam/routers/versions.py @@ -150,17 +150,18 @@ def upload_version_attachment( ): version = _version_or_404(db, id) _require_project_role(db, version.project_id, current, UPLOAD_ROLES) + payload = uploads.read_attachment(file) destination_directory = os.path.join(VERSION_UPLOAD_DIR, str(id)) os.makedirs(destination_directory, exist_ok=True) stored_filename = uploads.safe_filename(file.filename) destination_path = uploads.safe_join(destination_directory, file.filename) with open(destination_path, "wb") as handle: - shutil.copyfileobj(file.file, handle) + handle.write(payload) attachment = VersionAttachment( version_id=id, filename=stored_filename, content_type=file.content_type, - size_bytes=os.path.getsize(destination_path), + size_bytes=len(payload), file_path=destination_path, uploaded_by_id=current.id, ) diff --git a/backend/testjam/services/uploads.py b/backend/testjam/services/uploads.py index 1f5846a..479871e 100644 --- a/backend/testjam/services/uploads.py +++ b/backend/testjam/services/uploads.py @@ -10,6 +10,8 @@ from fastapi import HTTPException, UploadFile +from testjam.core.config import settings + FALLBACK_FILENAME = "attachment" DEFAULT_CHUNK = 64 * 1024 @@ -59,3 +61,22 @@ def read_bounded(file: UploadFile, max_bytes: int) -> bytes: ) chunks.append(chunk) return b"".join(chunks) + + +def read_attachment(file: UploadFile) -> bytes: + """Validate and read an attachment upload. + + Rejects MIME types in ``settings.blocked_attachment_mime_types`` with + 415 — those are vectors the victim browser would render inline. + Reads bytes through :func:`read_bounded` so anything above + ``settings.MAX_ATTACHMENT_BYTES`` short-circuits with 413 before it + hits disk. Callers receive the bytes and are responsible for writing + them to the sanitized destination path. + """ + content_type = (file.content_type or "").lower().strip() + if content_type in settings.blocked_attachment_mime_types: + raise HTTPException( + status_code=415, + detail=f"Attachment content type '{content_type}' is not allowed", + ) + return read_bounded(file, settings.MAX_ATTACHMENT_BYTES) diff --git a/backend/tests/security/test_upload_caps.py b/backend/tests/security/test_upload_caps.py new file mode 100644 index 0000000..175d26f --- /dev/null +++ b/backend/tests/security/test_upload_caps.py @@ -0,0 +1,151 @@ +"""Attachment upload size + MIME enforcement. + +The shared `uploads.read_attachment` helper short-circuits oversize +uploads with 413 and rejects MIME types the victim's browser would +render inline (HTML, SVG, JS, executables) with 415. These tests +exercise both rails through every attachment endpoint so a regression +in any one handler — they're independent code paths — is caught. +""" +import io + +import pytest + +from testjam.core.config import settings + + +OVERSIZE_CAP = 64 + + +@pytest.fixture +def small_cap(monkeypatch): + monkeypatch.setattr(settings, "MAX_ATTACHMENT_BYTES", OVERSIZE_CAP) + + +@pytest.fixture +def case_id(auth_client, project_id): + suite_id = auth_client.post( + f"/api/v1/projects/{project_id}/suites", json={"name": "S"}, + ).json()["id"] + return auth_client.post( + f"/api/v1/suites/{suite_id}/cases", json={"name": "TC", "suite_id": suite_id}, + ).json()["id"] + + +@pytest.fixture +def bug_id(auth_client, project_id): + return auth_client.post( + f"/api/v1/projects/{project_id}/bugs", json={"title": "B"}, + ).json()["id"] + + +@pytest.fixture +def execution_id(auth_client, project_id): + return auth_client.post( + f"/api/v1/projects/{project_id}/executions", + json={"title": "Run", "type": "manual", "test_case_ids": []}, + ).json()["id"] + + +@pytest.fixture +def result_id(auth_client, execution_id, project_id): + suite_id = auth_client.post( + f"/api/v1/projects/{project_id}/suites", json={"name": "S2"}, + ).json()["id"] + case_id = auth_client.post( + f"/api/v1/suites/{suite_id}/cases", json={"name": "TC2", "suite_id": suite_id}, + ).json()["id"] + result = auth_client.post( + f"/api/v1/executions/{execution_id}/results", + json={"test_case_id": case_id, "status": "passed"}, + ) + return result.json()["id"] + + +@pytest.fixture +def version_id(auth_client, project_id): + return auth_client.post( + f"/api/v1/projects/{project_id}/versions", json={"name": "v1"}, + ).json()["id"] + + +def _upload(client, url, *, payload=b"hello", mime="text/plain", filename="note.txt"): + return client.post( + url, + files={"file": (filename, io.BytesIO(payload), mime)}, + ) + + +ATTACHMENT_ENDPOINTS = [ + ("case", "/api/v1/cases/{case_id}/attachments"), + ("bug", "/api/v1/bugs/{bug_id}/attachments"), + ("execution", "/api/v1/executions/{execution_id}/attachments"), + ("result", "/api/v1/results/{result_id}/attachments"), + ("version", "/api/v1/versions/{version_id}/attachments"), +] + + +@pytest.mark.parametrize("kind,url_template", ATTACHMENT_ENDPOINTS) +def test_oversize_attachment_returns_413( + auth_client, small_cap, kind, url_template, + case_id, bug_id, execution_id, result_id, version_id, +): + url = url_template.format( + case_id=case_id, bug_id=bug_id, execution_id=execution_id, + result_id=result_id, version_id=version_id, + ) + response = _upload(auth_client, url, payload=b"x" * (OVERSIZE_CAP + 1)) + + assert response.status_code == 413 + + +@pytest.mark.parametrize("kind,url_template", ATTACHMENT_ENDPOINTS) +def test_blocked_mime_returns_415( + auth_client, kind, url_template, + case_id, bug_id, execution_id, result_id, version_id, +): + url = url_template.format( + case_id=case_id, bug_id=bug_id, execution_id=execution_id, + result_id=result_id, version_id=version_id, + ) + response = _upload( + auth_client, url, + payload=b"", + mime="text/html", + filename="exploit.html", + ) + + assert response.status_code == 415 + assert "text/html" in response.json()["detail"] + + +@pytest.mark.parametrize("kind,url_template", ATTACHMENT_ENDPOINTS) +def test_blocked_svg_returns_415( + auth_client, kind, url_template, + case_id, bug_id, execution_id, result_id, version_id, +): + url = url_template.format( + case_id=case_id, bug_id=bug_id, execution_id=execution_id, + result_id=result_id, version_id=version_id, + ) + response = _upload( + auth_client, url, + payload=b"", + mime="image/svg+xml", + filename="exploit.svg", + ) + + assert response.status_code == 415 + + +@pytest.mark.parametrize("kind,url_template", ATTACHMENT_ENDPOINTS) +def test_allowed_mime_under_cap_succeeds( + auth_client, kind, url_template, + case_id, bug_id, execution_id, result_id, version_id, +): + url = url_template.format( + case_id=case_id, bug_id=bug_id, execution_id=execution_id, + result_id=result_id, version_id=version_id, + ) + response = _upload(auth_client, url, payload=b"hello", mime="text/plain") + + assert response.status_code == 201