gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(security): add request size limits to API endpoints

Browse Source 
- Add RequestSizeLimitMiddleware that checks Content-Length header
  before request body is read: non-multipart requests capped at
  MAX_REQUEST_BODY_SIZE (default 1 MB), multipart uploads capped at
  MAX_UPLOAD_SIZE (default 1 GB). Returns HTTP 413 on violation.
- Register middleware in app/main.py
- Add max_request_body_size setting to app/config.py
- Fix ui_upload in files.py to check Content-Length early and read
  in 64 KB chunks (bounded memory usage), removing the post-write
  os.path.getsize check
- Document MAX_REQUEST_BODY_SIZE in .env.demo and ConfigurationGuide.md
- Mark SECURITY_AUDIT.md item #4 as resolved
- Add 9 tests in test_request_size_limit.py
- Update test_upload_file_too_large to use patch.object instead of
  the now-unused os.path.getsize mock

Closes #173

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-22 14:41:04 +00:00
parent c166bfa506
commit ca27a0b687
9 changed files with 872 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_file_upload.py
+155 -49
View File
@@ -32,17 +32,24 @@ def mock_celery_tasks():
mock_process_task.delay.return_value = mock_task
mock_convert_task.delay.return_value = mock_task
yield {"process_document": mock_process_task.delay, "convert_to_pdf": mock_convert_task.delay}
yield {
"process_document": mock_process_task.delay,
"convert_to_pdf": mock_convert_task.delay,
}
@pytest.mark.integration
class TestValidFileUploads:
"""Tests for successful file uploads with various valid file types."""
def test_upload_valid_pdf(self, client: TestClient, sample_pdf_path: str, mock_celery_tasks):
def test_upload_valid_pdf(
self, client: TestClient, sample_pdf_path: str, mock_celery_tasks
):
"""Test uploading a valid PDF file."""
with open(sample_pdf_path, "rb") as f:
response = client.post("/api/ui-upload", files={"file": ("document.pdf", f, "application/pdf")})
response = client.post(
"/api/ui-upload", files={"file": ("document.pdf", f, "application/pdf")}
)
assert response.status_code == 200
data = response.json()
@@ -67,7 +74,8 @@ class TestValidFileUploads:
"""Test uploading a valid text file."""
text_content = b"This is a test text file.\nWith multiple lines."
response = client.post(
"/api/ui-upload", files={"file": ("document.txt", io.BytesIO(text_content), "text/plain")}
"/api/ui-upload",
files={"file": ("document.txt", io.BytesIO(text_content), "text/plain")},
)
assert response.status_code == 200
@@ -87,7 +95,10 @@ class TestValidFileUploads:
b"\xff\xd9"
)
response = client.post("/api/ui-upload", files={"file": ("image.jpg", io.BytesIO(jpeg_content), "image/jpeg")})
response = client.post(
"/api/ui-upload",
files={"file": ("image.jpg", io.BytesIO(jpeg_content), "image/jpeg")},
)
assert response.status_code == 200
data = response.json()
@@ -106,7 +117,8 @@ class TestValidFileUploads:
)
response = client.post(
"/api/ui-upload", files={"file": ("screenshot.png", io.BytesIO(png_content), "image/png")}
"/api/ui-upload",
files={"file": ("screenshot.png", io.BytesIO(png_content), "image/png")},
)
assert response.status_code == 200
@@ -143,7 +155,10 @@ class TestValidFileUploads:
"""Test uploading a CSV file."""
csv_content = b"name,age,city\nJohn,30,NYC\nJane,25,LA\n"
response = client.post("/api/ui-upload", files={"file": ("data.csv", io.BytesIO(csv_content), "text/csv")})
response = client.post(
"/api/ui-upload",
files={"file": ("data.csv", io.BytesIO(csv_content), "text/csv")},
)
assert response.status_code == 200
data = response.json()
@@ -159,27 +174,36 @@ class TestInvalidFileUploads:
"""Test that files exceeding MAX_UPLOAD_SIZE are rejected."""
from app.config import settings
# Create a large file content (mock it to avoid memory issues)
large_content = b"x" * 1024 # 1KB for testing
with patch("os.path.getsize") as mock_getsize:
# Mock the file size to be over the configured limit
mock_getsize.return_value = settings.max_upload_size + 1
# Temporarily lower the upload limit so a tiny file exceeds it,
# avoiding the need to allocate a real 1 GB payload in memory.
small_limit = 100 # 100 bytes
small_content = b"x" * (small_limit + 1)
with patch.object(settings, "max_upload_size", small_limit):
response = client.post(
"/api/ui-upload", files={"file": ("huge.pdf", io.BytesIO(large_content), "application/pdf")}
"/api/ui-upload",
files={
"file": ("huge.pdf", io.BytesIO(small_content), "application/pdf")
},
)
assert response.status_code == 413 # Request Entity Too Large
assert "too large" in response.json()["detail"].lower()
assert "SECURITY_AUDIT.md" in response.json()["detail"]
assert response.status_code == 413 # Request Entity Too Large
assert "too large" in response.json()["detail"].lower()
assert "SECURITY_AUDIT.md" in response.json()["detail"]
def test_upload_executable_file(self, client: TestClient, mock_celery_tasks):
"""Test that executable files are handled (attempted conversion)."""
exe_content = b"MZ\x90\x00" # PE header
response = client.post(
"/api/ui-upload", files={"file": ("program.exe", io.BytesIO(exe_content), "application/x-msdownload")}
"/api/ui-upload",
files={
"file": (
"program.exe",
io.BytesIO(exe_content),
"application/x-msdownload",
)
},
)
# Per the code, unsupported types get a warning but are still processed
@@ -189,7 +213,10 @@ class TestInvalidFileUploads:
def test_upload_empty_file(self, client: TestClient, mock_celery_tasks):
"""Test uploading an empty file."""
response = client.post("/api/ui-upload", files={"file": ("empty.txt", io.BytesIO(b""), "text/plain")})
response = client.post(
"/api/ui-upload",
files={"file": ("empty.txt", io.BytesIO(b""), "text/plain")},
)
# Empty files are accepted and queued for processing
assert response.status_code == 200
@@ -209,14 +236,19 @@ class TestInvalidFileUploads:
class TestUploadSecurity:
"""Tests for security aspects of file uploads."""
def test_path_traversal_prevention_dotdot(self, client: TestClient, mock_celery_tasks):
def test_path_traversal_prevention_dotdot(
self, client: TestClient, mock_celery_tasks
):
"""Test that path traversal attempts are prevented."""
# Try to upload a file with path traversal in filename
malicious_filename = "../../etc/passwd.pdf"
pdf_content = b"%PDF-1.4\n%EOF"
response = client.post(
"/api/ui-upload", files={"file": (malicious_filename, io.BytesIO(pdf_content), "application/pdf")}
"/api/ui-upload",
files={
"file": (malicious_filename, io.BytesIO(pdf_content), "application/pdf")
},
)
assert response.status_code == 200
@@ -227,13 +259,18 @@ class TestUploadSecurity:
assert ".." not in data["stored_filename"]
assert "/" not in data["stored_filename"]
def test_path_traversal_prevention_absolute(self, client: TestClient, mock_celery_tasks):
def test_path_traversal_prevention_absolute(
self, client: TestClient, mock_celery_tasks
):
"""Test that absolute path attempts are prevented."""
malicious_filename = "/etc/shadow.pdf"
pdf_content = b"%PDF-1.4\n%EOF"
response = client.post(
"/api/ui-upload", files={"file": (malicious_filename, io.BytesIO(pdf_content), "application/pdf")}
"/api/ui-upload",
files={
"file": (malicious_filename, io.BytesIO(pdf_content), "application/pdf")
},
)
assert response.status_code == 200
@@ -243,13 +280,18 @@ class TestUploadSecurity:
assert data["original_filename"] == "shadow.pdf"
assert not data["stored_filename"].startswith("/")
def test_filename_with_special_characters(self, client: TestClient, mock_celery_tasks):
def test_filename_with_special_characters(
self, client: TestClient, mock_celery_tasks
):
"""Test handling of filenames with special characters."""
special_filename = "file name with spaces & special!@#chars.pdf"
pdf_content = b"%PDF-1.4\n%EOF"
response = client.post(
"/api/ui-upload", files={"file": (special_filename, io.BytesIO(pdf_content), "application/pdf")}
"/api/ui-upload",
files={
"file": (special_filename, io.BytesIO(pdf_content), "application/pdf")
},
)
assert response.status_code == 200
@@ -265,14 +307,19 @@ class TestUploadSecurity:
# Stored filename should have UUID and extension
assert data["stored_filename"].endswith(".pdf")
def test_path_traversal_prevention_windows_style(self, client: TestClient, mock_celery_tasks):
def test_path_traversal_prevention_windows_style(
self, client: TestClient, mock_celery_tasks
):
"""Test that Windows-style path traversal attempts are prevented."""
# Try Windows-style path with backslashes
malicious_filename = "..\\..\\..\\windows\\system32\\config.pdf"
pdf_content = b"%PDF-1.4\n%EOF"
response = client.post(
"/api/ui-upload", files={"file": (malicious_filename, io.BytesIO(pdf_content), "application/pdf")}
"/api/ui-upload",
files={
"file": (malicious_filename, io.BytesIO(pdf_content), "application/pdf")
},
)
assert response.status_code == 200
@@ -291,13 +338,18 @@ class TestUploadSecurity:
assert ".." not in data["original_filename"]
assert "/" not in data["original_filename"]
def test_path_traversal_prevention_mixed_separators(self, client: TestClient, mock_celery_tasks):
def test_path_traversal_prevention_mixed_separators(
self, client: TestClient, mock_celery_tasks
):
"""Test handling of filenames with mixed path separators."""
malicious_filename = "../path\\to/file.pdf"
pdf_content = b"%PDF-1.4\n%EOF"
response = client.post(
"/api/ui-upload", files={"file": (malicious_filename, io.BytesIO(pdf_content), "application/pdf")}
"/api/ui-upload",
files={
"file": (malicious_filename, io.BytesIO(pdf_content), "application/pdf")
},
)
assert response.status_code == 200
@@ -323,7 +375,10 @@ class TestUploadErrorHandling:
pdf_content = b"%PDF-1.4\n%EOF"
response = client.post(
"/api/ui-upload", files={"file": ("test.pdf", io.BytesIO(pdf_content), "application/pdf")}
"/api/ui-upload",
files={
"file": ("test.pdf", io.BytesIO(pdf_content), "application/pdf")
},
)
assert response.status_code == 500
@@ -331,14 +386,21 @@ class TestUploadErrorHandling:
def test_upload_celery_task_failure(self, client: TestClient, mock_celery_tasks):
"""Test handling when Celery task queueing fails."""
mock_celery_tasks["process_document"].side_effect = Exception("Celery connection failed")
mock_celery_tasks["process_document"].side_effect = Exception(
"Celery connection failed"
)
pdf_content = b"%PDF-1.4\n%EOF"
# The endpoint should still handle the error gracefully
# In this case, the exception will propagate
with pytest.raises(Exception):
client.post("/api/ui-upload", files={"file": ("test.pdf", io.BytesIO(pdf_content), "application/pdf")})
client.post(
"/api/ui-upload",
files={
"file": ("test.pdf", io.BytesIO(pdf_content), "application/pdf")
},
)
@pytest.mark.integration
@@ -351,11 +413,13 @@ class TestUploadFilenameHandling:
# Upload same file twice
response1 = client.post(
"/api/ui-upload", files={"file": ("same.pdf", io.BytesIO(pdf_content), "application/pdf")}
"/api/ui-upload",
files={"file": ("same.pdf", io.BytesIO(pdf_content), "application/pdf")},
)
response2 = client.post(
"/api/ui-upload", files={"file": ("same.pdf", io.BytesIO(pdf_content), "application/pdf")}
"/api/ui-upload",
files={"file": ("same.pdf", io.BytesIO(pdf_content), "application/pdf")},
)
assert response1.status_code == 200
@@ -375,14 +439,20 @@ class TestUploadFilenameHandling:
content = b"Some content"
response = client.post(
"/api/ui-upload", files={"file": ("NOEXTENSION", io.BytesIO(content), "application/octet-stream")}
"/api/ui-upload",
files={
"file": ("NOEXTENSION", io.BytesIO(content), "application/octet-stream")
},
)
assert response.status_code == 200
data = response.json()
assert data["original_filename"] == "NOEXTENSION"
# Stored filename should be just the UUID without extension
assert "." not in data["stored_filename"] or data["stored_filename"].count(".") == 0
assert (
"." not in data["stored_filename"]
or data["stored_filename"].count(".") == 0
)
@pytest.mark.integration
@@ -394,7 +464,10 @@ class TestUploadMimeTypeDetection:
pdf_content = b"%PDF-1.4\n%EOF"
response = client.post(
"/api/ui-upload", files={"file": ("doc.pdf", io.BytesIO(pdf_content), "application/octet-stream")}
"/api/ui-upload",
files={
"file": ("doc.pdf", io.BytesIO(pdf_content), "application/octet-stream")
},
)
assert response.status_code == 200
@@ -407,7 +480,14 @@ class TestUploadMimeTypeDetection:
image_content = b"\x00\x01\x02\x03"
response = client.post(
"/api/ui-upload", files={"file": ("photo.jpg", io.BytesIO(image_content), "application/octet-stream")}
"/api/ui-upload",
files={
"file": (
"photo.jpg",
io.BytesIO(image_content),
"application/octet-stream",
)
},
)
assert response.status_code == 200
@@ -419,7 +499,9 @@ class TestUploadMimeTypeDetection:
class TestFileSplitting:
"""Tests for file splitting functionality when MAX_SINGLE_FILE_SIZE is configured."""
def test_pdf_splitting_when_configured(self, client: TestClient, sample_pdf_path: str, mock_celery_tasks):
def test_pdf_splitting_when_configured(
self, client: TestClient, sample_pdf_path: str, mock_celery_tasks
):
"""Test that PDFs are split when they exceed MAX_SINGLE_FILE_SIZE."""
from app.config import settings
@@ -434,9 +516,14 @@ class TestFileSplitting:
]
# Mock should_split_file to return True
with patch("app.utils.file_splitting.should_split_file", return_value=True):
with patch(
"app.utils.file_splitting.should_split_file", return_value=True
):
with open(sample_pdf_path, "rb") as f:
response = client.post("/api/ui-upload", files={"file": ("large.pdf", f, "application/pdf")})
response = client.post(
"/api/ui-upload",
files={"file": ("large.pdf", f, "application/pdf")},
)
assert response.status_code == 200
data = response.json()
@@ -452,14 +539,19 @@ class TestFileSplitting:
# Verify each split file was queued for processing
assert mock_celery_tasks["process_document"].call_count == 3
def test_no_splitting_when_not_configured(self, client: TestClient, sample_pdf_path: str, mock_celery_tasks):
def test_no_splitting_when_not_configured(
self, client: TestClient, sample_pdf_path: str, mock_celery_tasks
):
"""Test that PDFs are not split when MAX_SINGLE_FILE_SIZE is None."""
from app.config import settings
# Ensure max_single_file_size is None (default)
with patch.object(settings, "max_single_file_size", None):
with open(sample_pdf_path, "rb") as f:
response = client.post("/api/ui-upload", files={"file": ("document.pdf", f, "application/pdf")})
response = client.post(
"/api/ui-upload",
files={"file": ("document.pdf", f, "application/pdf")},
)
assert response.status_code == 200
data = response.json()
@@ -472,14 +564,19 @@ class TestFileSplitting:
# Verify file was processed directly without splitting
mock_celery_tasks["process_document"].assert_called_once()
def test_no_splitting_for_small_files(self, client: TestClient, sample_pdf_path: str, mock_celery_tasks):
def test_no_splitting_for_small_files(
self, client: TestClient, sample_pdf_path: str, mock_celery_tasks
):
"""Test that small PDFs are not split even when MAX_SINGLE_FILE_SIZE is configured."""
from app.config import settings
# Configure a very large limit
with patch.object(settings, "max_single_file_size", 1000000000): # 1GB limit
with open(sample_pdf_path, "rb") as f:
response = client.post("/api/ui-upload", files={"file": ("small.pdf", f, "application/pdf")})
response = client.post(
"/api/ui-upload",
files={"file": ("small.pdf", f, "application/pdf")},
)
assert response.status_code == 200
data = response.json()
@@ -491,16 +588,24 @@ class TestFileSplitting:
# Verify file was processed directly
mock_celery_tasks["process_document"].assert_called_once()
def test_splitting_fallback_on_error(self, client: TestClient, sample_pdf_path: str, mock_celery_tasks):
def test_splitting_fallback_on_error(
self, client: TestClient, sample_pdf_path: str, mock_celery_tasks
):
"""Test that if splitting fails, the file is processed as a whole."""
from app.config import settings
with patch.object(settings, "max_single_file_size", 100): # Small limit
with patch("app.utils.file_splitting.should_split_file", return_value=True):
# Mock split_pdf_by_size to raise an exception
with patch("app.utils.file_splitting.split_pdf_by_size", side_effect=Exception("Split failed")):
with patch(
"app.utils.file_splitting.split_pdf_by_size",
side_effect=Exception("Split failed"),
):
with open(sample_pdf_path, "rb") as f:
response = client.post("/api/ui-upload", files={"file": ("document.pdf", f, "application/pdf")})
response = client.post(
"/api/ui-upload",
files={"file": ("document.pdf", f, "application/pdf")},
)
# Should still succeed, falling back to processing the whole file
assert response.status_code == 200
@@ -526,7 +631,8 @@ class TestFileSplitting:
)
response = client.post(
"/api/ui-upload", files={"file": ("image.png", io.BytesIO(image_content), "image/png")}
"/api/ui-upload",
files={"file": ("image.png", io.BytesIO(image_content), "image/png")},
)
assert response.status_code == 200