Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 23 additions & 0 deletions backend/testjam/core/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()]
Expand Down
5 changes: 3 additions & 2 deletions backend/testjam/routers/bugs.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)
Expand Down
6 changes: 3 additions & 3 deletions backend/testjam/routers/cases.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)
Expand Down
10 changes: 6 additions & 4 deletions backend/testjam/routers/executions/attachments.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)
Expand Down Expand Up @@ -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,
)
Expand Down
5 changes: 3 additions & 2 deletions backend/testjam/routers/versions.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)
Expand Down
21 changes: 21 additions & 0 deletions backend/testjam/services/uploads.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@

from fastapi import HTTPException, UploadFile

from testjam.core.config import settings

FALLBACK_FILENAME = "attachment"
DEFAULT_CHUNK = 64 * 1024

Expand Down Expand Up @@ -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)
151 changes: 151 additions & 0 deletions backend/tests/security/test_upload_caps.py
Original file line number Diff line number Diff line change
@@ -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"<script>alert(1)</script>",
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"<svg onload=alert(1)/>",
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
Loading