Skip to content

Commit 394cf10

Browse files
authored
Merge pull request #27 from CompassSecurity/fix/file-upload-handling
fix(backend):improved upload filter against polyglot files
2 parents a1ed486 + b970854 commit 394cf10

2 files changed

Lines changed: 89 additions & 3 deletions

File tree

backend/app/services/file/file.py

Lines changed: 22 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -110,6 +110,25 @@ def validate_file_content(file: UploadFile) -> FileType:
110110
)
111111

112112

113+
_CANONICAL_EXTENSION = {
114+
FileType.PNG: ".png",
115+
FileType.JPEG: ".jpg",
116+
FileType.JPG: ".jpg",
117+
FileType.TXT: ".txt",
118+
}
119+
120+
121+
def _normalize_filename(filename: str, file_type: FileType) -> str:
122+
"""
123+
Force the filename extension to match the detected content type so that a
124+
dangerous extension (e.g. .php) cannot be preserved by uploading a polyglot
125+
file with valid image magic bytes.
126+
"""
127+
stem = filename.rsplit(".", 1)[0] if "." in filename else filename
128+
stem = stem or "unnamed"
129+
return f"{stem}{_CANONICAL_EXTENSION[file_type]}"
130+
131+
113132
def upload_file_service(
114133
activity_id: uuid.UUID,
115134
file: UploadFile,
@@ -126,9 +145,9 @@ def upload_file_service(
126145
detected_file_type = validate_file_content(file)
127146
file_content = file.file.read()
128147

129-
final_filename = sanitize_filename(file.filename) or "unnamed"
130-
if detected_file_type == FileType.TXT and not final_filename.endswith(".txt"):
131-
final_filename += ".txt"
148+
final_filename = _normalize_filename(
149+
sanitize_filename(file.filename) or "unnamed", detected_file_type
150+
)
132151

133152
category = (
134153
FileCategory.RED

backend/tests/api/v1/assessments/test_activity_acl.py

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -301,6 +301,73 @@ def test_blue_user_cannot_upload_when_not_waiting(
301301
assert response.status_code == 403
302302

303303

304+
_PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"\x00" * 16
305+
306+
307+
def test_upload_polyglot_keeps_safe_extension(
308+
client: TestClient,
309+
session: Session,
310+
test_assessment: Assessment,
311+
test_activity: Activity,
312+
auth_headers_regular: dict[str, str],
313+
test_acl_red: Acl,
314+
):
315+
"""A PNG-magic file named shell.php is stored with a .png extension, not .php."""
316+
response = client.post(
317+
f"/api/v1/assessments/{test_assessment.id}/activity/{test_activity.id}/upload",
318+
files={"file": ("shell.php", _PNG_BYTES, "image/png")},
319+
headers=auth_headers_regular,
320+
)
321+
assert response.status_code == 200
322+
323+
file_id = uuid.UUID(response.json()["file_id"])
324+
stored = session.get(File, file_id)
325+
assert stored.filename == "shell.png"
326+
assert ".php" not in stored.filename
327+
assert stored.content_type == FileType.PNG
328+
329+
330+
def test_upload_text_normalized_to_txt(
331+
client: TestClient,
332+
session: Session,
333+
test_assessment: Assessment,
334+
test_activity: Activity,
335+
auth_headers_regular: dict[str, str],
336+
test_acl_red: Acl,
337+
):
338+
"""UTF-8 content with a non-txt extension is stored as .txt."""
339+
response = client.post(
340+
f"/api/v1/assessments/{test_assessment.id}/activity/{test_activity.id}/upload",
341+
files={"file": ("notes.md", b"just some notes", "text/plain")},
342+
headers=auth_headers_regular,
343+
)
344+
assert response.status_code == 200
345+
346+
stored = session.get(File, uuid.UUID(response.json()["file_id"]))
347+
assert stored.filename == "notes.txt"
348+
assert stored.content_type == FileType.TXT
349+
350+
351+
def test_upload_matching_extension_unchanged(
352+
client: TestClient,
353+
session: Session,
354+
test_assessment: Assessment,
355+
test_activity: Activity,
356+
auth_headers_regular: dict[str, str],
357+
test_acl_red: Acl,
358+
):
359+
"""A real png keeps its name (no double extension)."""
360+
response = client.post(
361+
f"/api/v1/assessments/{test_assessment.id}/activity/{test_activity.id}/upload",
362+
files={"file": ("photo.png", _PNG_BYTES, "image/png")},
363+
headers=auth_headers_regular,
364+
)
365+
assert response.status_code == 200
366+
367+
stored = session.get(File, uuid.UUID(response.json()["file_id"]))
368+
assert stored.filename == "photo.png"
369+
370+
304371
def test_blue_user_fail_hidden(
305372
client: TestClient,
306373
session: Session,

0 commit comments

Comments
 (0)