|
|
|
@@ -0,0 +1,278 @@ |
|
|
|
import io |
|
|
|
from unittest.mock import patch |
|
|
|
|
|
|
|
import pytest |
|
|
|
from werkzeug.exceptions import Forbidden |
|
|
|
|
|
|
|
from controllers.common.errors import FilenameNotExistsError |
|
|
|
from controllers.console.error import ( |
|
|
|
FileTooLargeError, |
|
|
|
NoFileUploadedError, |
|
|
|
TooManyFilesError, |
|
|
|
UnsupportedFileTypeError, |
|
|
|
) |
|
|
|
from services.errors.file import FileTooLargeError as ServiceFileTooLargeError |
|
|
|
from services.errors.file import UnsupportedFileTypeError as ServiceUnsupportedFileTypeError |
|
|
|
|
|
|
|
|
|
|
|
class TestFileUploadSecurity: |
|
|
|
"""Test file upload security logic without complex framework setup""" |
|
|
|
|
|
|
|
# Test 1: Basic file validation |
|
|
|
def test_should_validate_file_presence(self): |
|
|
|
"""Test that missing file is detected""" |
|
|
|
from flask import Flask, request |
|
|
|
|
|
|
|
app = Flask(__name__) |
|
|
|
|
|
|
|
with app.test_request_context(method="POST", data={}): |
|
|
|
# Simulate the check in FileApi.post() |
|
|
|
if "file" not in request.files: |
|
|
|
with pytest.raises(NoFileUploadedError): |
|
|
|
raise NoFileUploadedError() |
|
|
|
|
|
|
|
def test_should_validate_multiple_files(self): |
|
|
|
"""Test that multiple files are rejected""" |
|
|
|
from flask import Flask, request |
|
|
|
|
|
|
|
app = Flask(__name__) |
|
|
|
|
|
|
|
file_data = { |
|
|
|
"file": (io.BytesIO(b"content1"), "file1.txt", "text/plain"), |
|
|
|
"file2": (io.BytesIO(b"content2"), "file2.txt", "text/plain"), |
|
|
|
} |
|
|
|
|
|
|
|
with app.test_request_context(method="POST", data=file_data, content_type="multipart/form-data"): |
|
|
|
# Simulate the check in FileApi.post() |
|
|
|
if len(request.files) > 1: |
|
|
|
with pytest.raises(TooManyFilesError): |
|
|
|
raise TooManyFilesError() |
|
|
|
|
|
|
|
def test_should_validate_empty_filename(self): |
|
|
|
"""Test that empty filename is rejected""" |
|
|
|
from flask import Flask, request |
|
|
|
|
|
|
|
app = Flask(__name__) |
|
|
|
|
|
|
|
file_data = {"file": (io.BytesIO(b"content"), "", "text/plain")} |
|
|
|
|
|
|
|
with app.test_request_context(method="POST", data=file_data, content_type="multipart/form-data"): |
|
|
|
file = request.files["file"] |
|
|
|
if not file.filename: |
|
|
|
with pytest.raises(FilenameNotExistsError): |
|
|
|
raise FilenameNotExistsError |
|
|
|
|
|
|
|
# Test 2: Security - Filename sanitization |
|
|
|
def test_should_detect_path_traversal_in_filename(self): |
|
|
|
"""Test protection against directory traversal attacks""" |
|
|
|
dangerous_filenames = [ |
|
|
|
"../../../etc/passwd", |
|
|
|
"..\\..\\windows\\system32\\config\\sam", |
|
|
|
"../../../../etc/shadow", |
|
|
|
"./../../../sensitive.txt", |
|
|
|
] |
|
|
|
|
|
|
|
for filename in dangerous_filenames: |
|
|
|
# Any filename containing .. should be considered dangerous |
|
|
|
assert ".." in filename, f"Filename {filename} should be detected as path traversal" |
|
|
|
|
|
|
|
def test_should_detect_null_byte_injection(self): |
|
|
|
"""Test protection against null byte injection""" |
|
|
|
dangerous_filenames = [ |
|
|
|
"file.jpg\x00.php", |
|
|
|
"document.pdf\x00.exe", |
|
|
|
"image.png\x00.sh", |
|
|
|
] |
|
|
|
|
|
|
|
for filename in dangerous_filenames: |
|
|
|
# Null bytes should be detected |
|
|
|
assert "\x00" in filename, f"Filename {filename} should be detected as null byte injection" |
|
|
|
|
|
|
|
def test_should_sanitize_special_characters(self): |
|
|
|
"""Test that special characters in filenames are handled safely""" |
|
|
|
# Characters that could be problematic in various contexts |
|
|
|
dangerous_chars = ["/", "\\", ":", "*", "?", '"', "<", ">", "|", "\x00"] |
|
|
|
|
|
|
|
for char in dangerous_chars: |
|
|
|
filename = f"file{char}name.txt" |
|
|
|
# These characters should be detected or sanitized |
|
|
|
assert any(c in filename for c in dangerous_chars) |
|
|
|
|
|
|
|
# Test 3: Permission validation |
|
|
|
def test_should_validate_dataset_permissions(self): |
|
|
|
"""Test dataset upload permission logic""" |
|
|
|
|
|
|
|
class MockUser: |
|
|
|
is_dataset_editor = False |
|
|
|
|
|
|
|
user = MockUser() |
|
|
|
source = "datasets" |
|
|
|
|
|
|
|
# Simulate the permission check in FileApi.post() |
|
|
|
if source == "datasets" and not user.is_dataset_editor: |
|
|
|
with pytest.raises(Forbidden): |
|
|
|
raise Forbidden() |
|
|
|
|
|
|
|
def test_should_allow_general_upload_without_permission(self): |
|
|
|
"""Test general upload doesn't require dataset permission""" |
|
|
|
|
|
|
|
class MockUser: |
|
|
|
is_dataset_editor = False |
|
|
|
|
|
|
|
user = MockUser() |
|
|
|
source = None # General upload |
|
|
|
|
|
|
|
# This should not raise an exception |
|
|
|
if source == "datasets" and not user.is_dataset_editor: |
|
|
|
raise Forbidden() |
|
|
|
# Test passes if no exception is raised |
|
|
|
|
|
|
|
# Test 4: Service error handling |
|
|
|
@patch("services.file_service.FileService.upload_file") |
|
|
|
def test_should_handle_file_too_large_error(self, mock_upload): |
|
|
|
"""Test that service FileTooLargeError is properly converted""" |
|
|
|
mock_upload.side_effect = ServiceFileTooLargeError("File too large") |
|
|
|
|
|
|
|
try: |
|
|
|
mock_upload(filename="test.txt", content=b"data", mimetype="text/plain", user=None, source=None) |
|
|
|
except ServiceFileTooLargeError as e: |
|
|
|
# Simulate the error conversion in FileApi.post() |
|
|
|
with pytest.raises(FileTooLargeError): |
|
|
|
raise FileTooLargeError(e.description) |
|
|
|
|
|
|
|
@patch("services.file_service.FileService.upload_file") |
|
|
|
def test_should_handle_unsupported_file_type_error(self, mock_upload): |
|
|
|
"""Test that service UnsupportedFileTypeError is properly converted""" |
|
|
|
mock_upload.side_effect = ServiceUnsupportedFileTypeError() |
|
|
|
|
|
|
|
try: |
|
|
|
mock_upload( |
|
|
|
filename="test.exe", content=b"data", mimetype="application/octet-stream", user=None, source=None |
|
|
|
) |
|
|
|
except ServiceUnsupportedFileTypeError: |
|
|
|
# Simulate the error conversion in FileApi.post() |
|
|
|
with pytest.raises(UnsupportedFileTypeError): |
|
|
|
raise UnsupportedFileTypeError() |
|
|
|
|
|
|
|
# Test 5: File type security |
|
|
|
def test_should_identify_dangerous_file_extensions(self): |
|
|
|
"""Test detection of potentially dangerous file extensions""" |
|
|
|
dangerous_extensions = [ |
|
|
|
".php", |
|
|
|
".PHP", |
|
|
|
".pHp", # PHP files (case variations) |
|
|
|
".exe", |
|
|
|
".EXE", # Executables |
|
|
|
".sh", |
|
|
|
".SH", # Shell scripts |
|
|
|
".bat", |
|
|
|
".BAT", # Batch files |
|
|
|
".cmd", |
|
|
|
".CMD", # Command files |
|
|
|
".ps1", |
|
|
|
".PS1", # PowerShell |
|
|
|
".jar", |
|
|
|
".JAR", # Java archives |
|
|
|
".vbs", |
|
|
|
".VBS", # VBScript |
|
|
|
] |
|
|
|
|
|
|
|
safe_extensions = [".txt", ".pdf", ".jpg", ".png", ".doc", ".docx"] |
|
|
|
|
|
|
|
# Just verify our test data is correct |
|
|
|
for ext in dangerous_extensions: |
|
|
|
assert ext.lower() in [".php", ".exe", ".sh", ".bat", ".cmd", ".ps1", ".jar", ".vbs"] |
|
|
|
|
|
|
|
for ext in safe_extensions: |
|
|
|
assert ext.lower() not in [".php", ".exe", ".sh", ".bat", ".cmd", ".ps1", ".jar", ".vbs"] |
|
|
|
|
|
|
|
def test_should_detect_double_extensions(self): |
|
|
|
"""Test detection of double extension attacks""" |
|
|
|
suspicious_filenames = [ |
|
|
|
"image.jpg.php", |
|
|
|
"document.pdf.exe", |
|
|
|
"photo.png.sh", |
|
|
|
"file.txt.bat", |
|
|
|
] |
|
|
|
|
|
|
|
for filename in suspicious_filenames: |
|
|
|
# Check that these have multiple extensions |
|
|
|
parts = filename.split(".") |
|
|
|
assert len(parts) > 2, f"Filename {filename} should have multiple extensions" |
|
|
|
|
|
|
|
# Test 6: Configuration validation |
|
|
|
def test_upload_configuration_structure(self): |
|
|
|
"""Test that upload configuration has correct structure""" |
|
|
|
# Simulate the configuration returned by FileApi.get() |
|
|
|
config = { |
|
|
|
"file_size_limit": 15, |
|
|
|
"batch_count_limit": 5, |
|
|
|
"image_file_size_limit": 10, |
|
|
|
"video_file_size_limit": 500, |
|
|
|
"audio_file_size_limit": 50, |
|
|
|
"workflow_file_upload_limit": 10, |
|
|
|
} |
|
|
|
|
|
|
|
# Verify all required fields are present |
|
|
|
required_fields = [ |
|
|
|
"file_size_limit", |
|
|
|
"batch_count_limit", |
|
|
|
"image_file_size_limit", |
|
|
|
"video_file_size_limit", |
|
|
|
"audio_file_size_limit", |
|
|
|
"workflow_file_upload_limit", |
|
|
|
] |
|
|
|
|
|
|
|
for field in required_fields: |
|
|
|
assert field in config, f"Missing required field: {field}" |
|
|
|
assert isinstance(config[field], int), f"Field {field} should be an integer" |
|
|
|
assert config[field] > 0, f"Field {field} should be positive" |
|
|
|
|
|
|
|
# Test 7: Source parameter handling |
|
|
|
def test_source_parameter_normalization(self): |
|
|
|
"""Test that source parameter is properly normalized""" |
|
|
|
test_cases = [ |
|
|
|
("datasets", "datasets"), |
|
|
|
("other", None), |
|
|
|
("", None), |
|
|
|
(None, None), |
|
|
|
] |
|
|
|
|
|
|
|
for input_source, expected in test_cases: |
|
|
|
# Simulate the source normalization in FileApi.post() |
|
|
|
source = "datasets" if input_source == "datasets" else None |
|
|
|
if source not in ("datasets", None): |
|
|
|
source = None |
|
|
|
assert source == expected |
|
|
|
|
|
|
|
# Test 8: Boundary conditions |
|
|
|
def test_should_handle_edge_case_file_sizes(self): |
|
|
|
"""Test handling of boundary file sizes""" |
|
|
|
test_cases = [ |
|
|
|
(0, "Empty file"), # 0 bytes |
|
|
|
(1, "Single byte"), # 1 byte |
|
|
|
(15 * 1024 * 1024 - 1, "Just under limit"), # Just under 15MB |
|
|
|
(15 * 1024 * 1024, "At limit"), # Exactly 15MB |
|
|
|
(15 * 1024 * 1024 + 1, "Just over limit"), # Just over 15MB |
|
|
|
] |
|
|
|
|
|
|
|
for size, description in test_cases: |
|
|
|
# Just verify our test data |
|
|
|
assert isinstance(size, int), f"{description}: Size should be integer" |
|
|
|
assert size >= 0, f"{description}: Size should be non-negative" |
|
|
|
|
|
|
|
def test_should_handle_special_mime_types(self): |
|
|
|
"""Test handling of various MIME types""" |
|
|
|
mime_type_tests = [ |
|
|
|
("application/octet-stream", "Generic binary"), |
|
|
|
("text/plain", "Plain text"), |
|
|
|
("image/jpeg", "JPEG image"), |
|
|
|
("application/pdf", "PDF document"), |
|
|
|
("", "Empty MIME type"), |
|
|
|
(None, "None MIME type"), |
|
|
|
] |
|
|
|
|
|
|
|
for mime_type, description in mime_type_tests: |
|
|
|
# Verify test data structure |
|
|
|
if mime_type is not None: |
|
|
|
assert isinstance(mime_type, str), f"{description}: MIME type should be string or None" |