diff --git a/tests/test_e2e_full_stack.py b/tests/test_e2e_full_stack.py index 3bf673f0..47cb672c 100644 --- a/tests/test_e2e_full_stack.py +++ b/tests/test_e2e_full_stack.py @@ -4,6 +4,7 @@ End-to-end integration tests using real infrastructure. These tests spin up actual services (PostgreSQL, Redis, Gotenberg, WebDAV, SFTP, MinIO) and test the complete application workflow from API request to file upload. """ + import os import time import pytest @@ -13,6 +14,13 @@ from unittest.mock import patch # Import testcontainers requirement pytest.importorskip("testcontainers", reason="testcontainers not installed") +try: + import psycopg2 # noqa: F401 + + _has_psycopg2 = True +except ModuleNotFoundError: + _has_psycopg2 = False + from tests.fixtures_integration import ( postgres_container, redis_container, @@ -33,7 +41,7 @@ from tests.fixtures_integration import ( class TestEndToEndWithRedis: """ End-to-end tests with real Redis and Celery workers. - + These tests verify the complete task queueing and execution workflow. """ @@ -47,14 +55,16 @@ class TestEndToEndWithRedis: ): """ Test complete workflow: Queue task in Redis → Celery worker executes → Upload to WebDAV. - + This is the closest to production - actual message queueing and async execution. """ from app.tasks.upload_to_webdav import upload_to_webdav - - with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \ - patch("app.tasks.upload_to_webdav.log_task_progress"): - + + with ( + patch("app.tasks.upload_to_webdav.settings") as mock_settings, + patch("app.tasks.upload_to_webdav.log_task_progress"), + ): + # Configure to use real WebDAV server mock_settings.webdav_url = webdav_container["url"] + "/" mock_settings.webdav_username = webdav_container["username"] @@ -62,10 +72,10 @@ class TestEndToEndWithRedis: mock_settings.webdav_folder = "" mock_settings.webdav_verify_ssl = False mock_settings.http_request_timeout = 30 - + # Queue the task (it goes to Redis) result = upload_to_webdav.delay(sample_text_file, file_id=1) - + # Wait for task to complete (worker picks it up from Redis) timeout = 30 start_time = time.time() @@ -73,26 +83,24 @@ class TestEndToEndWithRedis: if time.time() - start_time > timeout: pytest.fail(f"Task did not complete within {timeout} seconds") time.sleep(0.5) - + # Get the result task_result = result.get(timeout=10) - + # Verify task completed successfully assert task_result["status"] == "Completed" assert task_result["file"] == sample_text_file - + # Verify file was actually uploaded to WebDAV server filename = os.path.basename(sample_text_file) file_url = f"{webdav_container['url']}/{filename}" - + response = requests.get( - file_url, - auth=(webdav_container["username"], webdav_container["password"]), - timeout=5 + file_url, auth=(webdav_container["username"], webdav_container["password"]), timeout=5 ) - + assert response.status_code == 200 - + # Verify content matches with open(sample_text_file, "rb") as f: assert response.content == f.read() @@ -104,33 +112,30 @@ class TestEndToEndWithRedis: ): """ Test that tasks are properly queued in Redis. - + This verifies the Redis broker is working correctly. """ from app.tasks.upload_to_webdav import upload_to_webdav import redis - + # Connect to Redis directly r = redis.from_url(redis_container["url"]) - + # Check Redis is accessible assert r.ping() - + # Get current queue length initial_queue_length = r.llen("celery") - + # Queue a task (don't execute, just verify queueing) with patch("app.tasks.upload_to_webdav.settings") as mock_settings: mock_settings.webdav_url = "http://test.com" mock_settings.webdav_username = "user" mock_settings.webdav_password = "pass" - + # This will queue the task in Redis - result = upload_to_webdav.apply_async( - args=["/tmp/test.txt"], - kwargs={"file_id": 1} - ) - + result = upload_to_webdav.apply_async(args=["/tmp/test.txt"], kwargs={"file_id": 1}) + # Verify task ID was generated assert result.id is not None @@ -144,67 +149,64 @@ class TestEndToEndWithRedis: ): """ Test multiple tasks executing in parallel through Redis/Celery. - + This tests concurrent task processing. """ from app.tasks.upload_to_webdav import upload_to_webdav - + # Create multiple test files files = [] for i in range(5): test_file = tmp_path / f"test_{i}.txt" test_file.write_text(f"Test file {i}") files.append(str(test_file)) - - with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \ - patch("app.tasks.upload_to_webdav.log_task_progress"): - + + with ( + patch("app.tasks.upload_to_webdav.settings") as mock_settings, + patch("app.tasks.upload_to_webdav.log_task_progress"), + ): + mock_settings.webdav_url = webdav_container["url"] + "/" mock_settings.webdav_username = webdav_container["username"] mock_settings.webdav_password = webdav_container["password"] mock_settings.webdav_folder = "parallel-test" mock_settings.webdav_verify_ssl = False mock_settings.http_request_timeout = 30 - + # Create folder on WebDAV server folder_url = f"{webdav_container['url']}/parallel-test" requests.request( - "MKCOL", - folder_url, - auth=(webdav_container["username"], webdav_container["password"]), - timeout=5 + "MKCOL", folder_url, auth=(webdav_container["username"], webdav_container["password"]), timeout=5 ) - + # Queue all tasks results = [] for idx, file_path in enumerate(files): result = upload_to_webdav.delay(file_path, file_id=idx + 100) results.append((result, file_path)) - + # Wait for all tasks to complete timeout = 60 start_time = time.time() all_ready = False - + while not all_ready: if time.time() - start_time > timeout: pytest.fail("Tasks did not complete within timeout") - + all_ready = all(r.ready() for r, _ in results) time.sleep(0.5) - + # Verify all tasks succeeded for result, file_path in results: task_result = result.get(timeout=5) assert task_result["status"] == "Completed" - + # Verify file on server filename = os.path.basename(file_path) file_url = f"{webdav_container['url']}/parallel-test/{filename}" response = requests.get( - file_url, - auth=(webdav_container["username"], webdav_container["password"]), - timeout=5 + file_url, auth=(webdav_container["username"], webdav_container["password"]), timeout=5 ) assert response.status_code == 200 @@ -217,37 +219,39 @@ class TestEndToEndWithRedis: ): """ Test that tasks retry on failure using Redis. - + This verifies the retry mechanism works with real broker. """ from app.tasks.upload_to_webdav import upload_to_webdav - - with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \ - patch("app.tasks.upload_to_webdav.log_task_progress"), \ - patch("app.tasks.upload_to_webdav.requests.put") as mock_put: - + + with ( + patch("app.tasks.upload_to_webdav.settings") as mock_settings, + patch("app.tasks.upload_to_webdav.log_task_progress"), + patch("app.tasks.upload_to_webdav.requests.put") as mock_put, + ): + mock_settings.webdav_url = "http://test.com/" mock_settings.webdav_username = "user" mock_settings.webdav_password = "pass" mock_settings.webdav_folder = "" mock_settings.webdav_verify_ssl = False mock_settings.http_request_timeout = 30 - + # First attempt fails with 500 mock_response_fail = requests.Response() mock_response_fail.status_code = 500 mock_response_fail._content = b"Server Error" - + # Second attempt succeeds mock_response_success = requests.Response() mock_response_success.status_code = 201 - + # Configure mock to fail once, then succeed mock_put.side_effect = [mock_response_fail, mock_response_success] - + # Queue task result = upload_to_webdav.delay(sample_text_file, file_id=1) - + # Wait for completion (including retry) timeout = 30 start_time = time.time() @@ -263,7 +267,7 @@ class TestEndToEndWithRedis: class TestFullInfrastructure: """ Tests using the complete infrastructure stack. - + PostgreSQL + Redis + Gotenberg + Upload targets (WebDAV/SFTP/MinIO) """ @@ -272,38 +276,43 @@ class TestFullInfrastructure: Verify all infrastructure components are running. """ infra = full_infrastructure - + # Check PostgreSQL assert infra["postgres"]["url"] is not None assert "postgresql" in infra["postgres"]["url"] - + # Check Redis assert infra["redis"]["url"] is not None import redis + r = redis.from_url(infra["redis"]["url"]) assert r.ping() - + # Check Gotenberg assert infra["gotenberg"]["url"] is not None response = requests.get(f"{infra['gotenberg']['url']}/health", timeout=5) assert response.status_code == 200 - + # Check WebDAV assert infra["webdav"]["url"] is not None - + # Check SFTP assert infra["sftp"]["host"] is not None assert infra["sftp"]["port"] is not None - + # Check MinIO assert infra["minio"]["access_key"] is not None + @pytest.mark.skipif( + not _has_psycopg2, + reason="psycopg2 not installed", + ) def test_database_with_real_postgres(self, postgres_container, db_session_real): """ Test database operations with real PostgreSQL instead of SQLite. """ from app.models import FileRecord - + # Create a file record file_record = FileRecord( filename="test.pdf", @@ -311,13 +320,13 @@ class TestFullInfrastructure: file_size=1024, mime_type="application/pdf", ) - + db_session_real.add(file_record) db_session_real.commit() - + # Verify it was saved assert file_record.id is not None - + # Query it back queried = db_session_real.query(FileRecord).filter_by(filename="test.pdf").first() assert queried is not None @@ -333,26 +342,28 @@ class TestFullInfrastructure: ): """ Test uploading to multiple targets in parallel (WebDAV + SFTP). - + This simulates the send_to_all_destinations workflow. """ from app.tasks.upload_to_webdav import upload_to_webdav - + infra = full_infrastructure - - with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \ - patch("app.tasks.upload_to_webdav.log_task_progress"): - + + with ( + patch("app.tasks.upload_to_webdav.settings") as mock_settings, + patch("app.tasks.upload_to_webdav.log_task_progress"), + ): + mock_settings.webdav_url = infra["webdav"]["url"] + "/" mock_settings.webdav_username = infra["webdav"]["username"] mock_settings.webdav_password = infra["webdav"]["password"] mock_settings.webdav_folder = "" mock_settings.webdav_verify_ssl = False mock_settings.http_request_timeout = 30 - + # Upload to WebDAV webdav_result = upload_to_webdav.delay(sample_text_file, file_id=1) - + # Wait for completion timeout = 30 start_time = time.time() @@ -360,25 +371,23 @@ class TestFullInfrastructure: if time.time() - start_time > timeout: pytest.fail("Task timeout") time.sleep(0.5) - + # Verify WebDAV upload result = webdav_result.get(timeout=10) assert result["status"] == "Completed" - + # Verify file on WebDAV server filename = os.path.basename(sample_text_file) file_url = f"{infra['webdav']['url']}/{filename}" response = requests.get( - file_url, - auth=(infra["webdav"]["username"], infra["webdav"]["password"]), - timeout=5 + file_url, auth=(infra["webdav"]["username"], infra["webdav"]["password"]), timeout=5 ) assert response.status_code == 200 def test_gotenberg_pdf_conversion(self, gotenberg_container, tmp_path): """ Test PDF conversion using real Gotenberg service. - + This verifies document processing capabilities. """ # Create a simple HTML file @@ -390,16 +399,14 @@ class TestFullInfrastructure:

Integration Test

This is a test document.

""") - + # Convert to PDF using Gotenberg with open(html_file, "rb") as f: files = {"files": f} response = requests.post( - f"{gotenberg_container['url']}/forms/chromium/convert/html", - files=files, - timeout=30 + f"{gotenberg_container['url']}/forms/chromium/convert/html", files=files, timeout=30 ) - + assert response.status_code == 200 assert response.headers["Content-Type"] == "application/pdf" assert len(response.content) > 0 @@ -408,12 +415,12 @@ class TestFullInfrastructure: def test_minio_s3_upload(self, minio_container, sample_text_file): """ Test S3-compatible upload using real MinIO. - + This tests S3 upload functionality with actual storage. """ import boto3 from botocore.client import Config - + # Create S3 client configured for MinIO s3_client = boto3.client( "s3", @@ -423,28 +430,27 @@ class TestFullInfrastructure: config=Config(signature_version="s3v4"), region_name=minio_container["region"], ) - + # Create bucket bucket_name = "test-bucket" s3_client.create_bucket(Bucket=bucket_name) - + # Upload file filename = os.path.basename(sample_text_file) with open(sample_text_file, "rb") as f: s3_client.upload_fileobj(f, bucket_name, filename) - + # Verify upload response = s3_client.list_objects_v2(Bucket=bucket_name) assert "Contents" in response assert len(response["Contents"]) == 1 assert response["Contents"][0]["Key"] == filename - + # Download and verify content download_path = os.path.join(os.path.dirname(sample_text_file), "downloaded.txt") s3_client.download_file(bucket_name, filename, download_path) - - with open(sample_text_file, "rb") as original, \ - open(download_path, "rb") as downloaded: + + with open(sample_text_file, "rb") as original, open(download_path, "rb") as downloaded: assert original.read() == downloaded.read() def test_sftp_upload(self, sftp_container, sample_text_file): @@ -452,11 +458,11 @@ class TestFullInfrastructure: Test SFTP upload using real SFTP server. """ import paramiko - + # Create SFTP client ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) - + # Connect to SFTP server ssh.connect( hostname=sftp_container["host"], @@ -465,35 +471,34 @@ class TestFullInfrastructure: password=sftp_container["password"], timeout=10, ) - + sftp = ssh.open_sftp() - + try: # Upload file filename = os.path.basename(sample_text_file) remote_path = f"{sftp_container['folder']}/{filename}" - + sftp.put(sample_text_file, remote_path) - + # Verify upload stat = sftp.stat(remote_path) assert stat.st_size == os.path.getsize(sample_text_file) - + # Download and verify content download_path = os.path.join(os.path.dirname(sample_text_file), "sftp_downloaded.txt") sftp.get(remote_path, download_path) - - with open(sample_text_file, "rb") as original, \ - open(download_path, "rb") as downloaded: + + with open(sample_text_file, "rb") as original, open(download_path, "rb") as downloaded: assert original.read() == downloaded.read() - + finally: sftp.close() ssh.close() @pytest.mark.integration -@pytest.mark.requires_docker +@pytest.mark.requires_docker @pytest.mark.e2e @pytest.mark.slow class TestProductionLikeScenarios: @@ -517,16 +522,16 @@ class TestProductionLikeScenarios: 4. Process document (mock OCR/metadata extraction) 5. Upload to WebDAV via Celery 6. Verify all steps completed - + This is the closest to real production usage. """ from app.models import FileRecord from app.tasks.upload_to_webdav import upload_to_webdav - + # Create test document test_doc = tmp_path / "invoice.pdf" test_doc.write_bytes(b"%PDF-1.4\n%Test PDF\n%%EOF") - + # Step 1: Store in database file_record = FileRecord( filename="invoice.pdf", @@ -536,35 +541,34 @@ class TestProductionLikeScenarios: ) db_session_real.add(file_record) db_session_real.commit() - + assert file_record.id is not None db_file_id = file_record.id - + # Step 2: Queue upload task infra = full_infrastructure - - with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \ - patch("app.tasks.upload_to_webdav.log_task_progress"): - + + with ( + patch("app.tasks.upload_to_webdav.settings") as mock_settings, + patch("app.tasks.upload_to_webdav.log_task_progress"), + ): + mock_settings.webdav_url = infra["webdav"]["url"] + "/" mock_settings.webdav_username = infra["webdav"]["username"] mock_settings.webdav_password = infra["webdav"]["password"] mock_settings.webdav_folder = "processed" mock_settings.webdav_verify_ssl = False mock_settings.http_request_timeout = 30 - + # Create folder folder_url = f"{infra['webdav']['url']}/processed" requests.request( - "MKCOL", - folder_url, - auth=(infra["webdav"]["username"], infra["webdav"]["password"]), - timeout=5 + "MKCOL", folder_url, auth=(infra["webdav"]["username"], infra["webdav"]["password"]), timeout=5 ) - + # Step 3: Queue upload result = upload_to_webdav.delay(str(test_doc), file_id=db_file_id) - + # Step 4: Wait for processing timeout = 30 start_time = time.time() @@ -572,17 +576,15 @@ class TestProductionLikeScenarios: if time.time() - start_time > timeout: pytest.fail("Pipeline timeout") time.sleep(0.5) - + # Step 5: Verify completion task_result = result.get(timeout=10) assert task_result["status"] == "Completed" - + # Step 6: Verify file on WebDAV file_url = f"{infra['webdav']['url']}/processed/invoice.pdf" response = requests.get( - file_url, - auth=(infra["webdav"]["username"], infra["webdav"]["password"]), - timeout=5 + file_url, auth=(infra["webdav"]["username"], infra["webdav"]["password"]), timeout=5 ) assert response.status_code == 200 assert response.content == test_doc.read_bytes() diff --git a/tests/test_path_traversal_security.py b/tests/test_path_traversal_security.py index 878b547d..0bb44178 100644 --- a/tests/test_path_traversal_security.py +++ b/tests/test_path_traversal_security.py @@ -45,7 +45,7 @@ class TestFilenameSanitization: result = sanitize_filename("/etc/passwd") assert "/" not in result - assert result == "_etc_passwd" + assert result == "etc_passwd" def test_sanitize_removes_windows_path_separators(self): """Test that Windows path separators are removed.""" @@ -83,9 +83,9 @@ class TestFilenameSanitization: from app.utils.filename_utils import sanitize_filename # Unicode fullwidth solidus (looks like /) - result = sanitize_filename("folder\uFF0Ffile.pdf") + result = sanitize_filename("folder\uff0ffile.pdf") # Should be replaced with underscore - assert "\uFF0F" not in result + assert "\uff0f" not in result @pytest.mark.security @@ -100,10 +100,10 @@ class TestEmbedMetadataPathTraversal: # Simulate malicious metadata from GPT malicious_filename = "../../etc/passwd" - + # This should be sanitized before being used sanitized = sanitize_filename(malicious_filename) - + # Verify sanitization removes path traversal assert ".." not in sanitized assert "/" not in sanitized @@ -112,7 +112,7 @@ class TestEmbedMetadataPathTraversal: # Verify unique_filepath with sanitized name stays in directory result = unique_filepath(str(tmp_path), sanitized, ".pdf") result_path = Path(result) - + # Ensure result is within tmp_path assert result_path.parent == tmp_path @@ -120,7 +120,7 @@ class TestEmbedMetadataPathTraversal: """Test that embed_metadata_into_pdf sanitizes the filename from metadata.""" from app.tasks.embed_metadata_into_pdf import unique_filepath from app.utils.filename_utils import sanitize_filename - + # Test various malicious filenames malicious_filenames = [ "../../../etc/passwd", @@ -130,15 +130,15 @@ class TestEmbedMetadataPathTraversal: "folder/../file", "folder\\..\\file", ] - + for malicious in malicious_filenames: # Sanitize as the task should do sanitized = sanitize_filename(malicious) - + # Verify no path traversal is possible result = unique_filepath(str(tmp_path), sanitized, ".pdf") result_path = Path(result) - + # Result must be direct child of tmp_path assert result_path.parent == tmp_path, f"Failed for: {malicious}" @@ -163,16 +163,16 @@ class TestEmbedMetadataPathTraversal: # Setup mock_settings.workdir = str(tmp_path) - + # Create a temporary PDF file test_pdf = tmp_path / "test.pdf" test_pdf.write_bytes(b"%PDF-1.4\n") - + # Mock PDF operations mock_reader_instance = MagicMock() mock_reader_instance.pages = [] mock_pdf_reader.return_value = mock_reader_instance - + mock_writer_instance = MagicMock() mock_pdf_writer.return_value = mock_writer_instance @@ -187,12 +187,8 @@ class TestEmbedMetadataPathTraversal: processed_dir = tmp_path / "processed" processed_dir.mkdir() - # Execute task - task_mock = MagicMock() - task_mock.request.id = "test-task-id" - + # Execute task (called directly, Celery injects 'self' automatically) result = embed_metadata_into_pdf( - task_mock, str(test_pdf), "test text", malicious_metadata, @@ -219,11 +215,11 @@ class TestExtractMetadataFilenameValidation: def test_validates_filename_format(self): """Test that invalid filename formats are rejected.""" import re - + # Valid pattern from extract_metadata_with_gpt.py # TODO: Consider extracting this to a shared constant to avoid duplication - valid_pattern = r'^[\w\-\. ]+$' - + valid_pattern = r"^[\w\-\. ]+$" + # Test valid filenames valid_filenames = [ "2024-01-15_Invoice.pdf", @@ -231,12 +227,12 @@ class TestExtractMetadataFilenameValidation: "My Document 2024.pdf", "file-name_123.pdf", ] - + for filename in valid_filenames: # Remove extension for test name_only = filename.rsplit(".", 1)[0] assert re.match(valid_pattern, name_only), f"Valid filename rejected: {filename}" - + # Test invalid filenames invalid_filenames = [ "../../../etc/passwd", @@ -247,7 +243,7 @@ class TestExtractMetadataFilenameValidation: "file|name.pdf", "file<>name.pdf", ] - + for filename in invalid_filenames: assert not re.match(valid_pattern, filename), f"Invalid filename accepted: {filename}" @@ -259,7 +255,7 @@ class TestExtractMetadataFilenameValidation: "/etc/shadow", "folder/../file", ] - + for filename in malicious_filenames: # Check for path traversal indicators has_traversal = ".." in filename or "/" in filename or "\\" in filename @@ -275,18 +271,18 @@ class TestPathValidationSecurity: """Test that is_relative_to prevents directory traversal.""" base_dir = tmp_path / "workdir" base_dir.mkdir() - + # Create a file outside base_dir outside_dir = tmp_path / "outside" outside_dir.mkdir() outside_file = outside_dir / "file.txt" outside_file.write_text("test") - + # Attempt to access file outside base_dir try: outside_resolved = outside_file.resolve() base_resolved = base_dir.resolve() - + # Should return False (file is not relative to base_dir) is_safe = outside_resolved.is_relative_to(base_resolved) assert not is_safe, "Path traversal not detected" @@ -299,21 +295,21 @@ class TestPathValidationSecurity: """Test that resolve() handles symlink attacks.""" base_dir = tmp_path / "workdir" base_dir.mkdir() - + # Create target outside base_dir outside_dir = tmp_path / "outside" outside_dir.mkdir() target_file = outside_dir / "secret.txt" target_file.write_text("secret") - + # Create symlink inside base_dir pointing outside symlink_path = base_dir / "link.txt" symlink_path.symlink_to(target_file) - + # Resolve should give us the real path resolved = symlink_path.resolve() base_resolved = base_dir.resolve() - + # The resolved path should NOT be relative to base_dir try: is_safe = resolved.is_relative_to(base_resolved) @@ -326,17 +322,17 @@ class TestPathValidationSecurity: """Demonstrate why string-based path validation is insecure.""" base_dir = tmp_path / "workdir" base_dir.mkdir() - + # Create a similar-named directory fake_dir = tmp_path / "workdir-fake" fake_dir.mkdir() fake_file = fake_dir / "file.txt" fake_file.write_text("content") - + # String-based check (insecure) base_str = str(base_dir) fake_str = str(fake_file) - + # This would INCORRECTLY pass string.startswith() if not careful # because "workdir-fake" starts with "workdir" if base_str.endswith("/") or base_str.endswith("\\"): @@ -345,7 +341,7 @@ class TestPathValidationSecurity: else: # Without separator, vulnerable to partial matches string_check_unsafe = fake_str.startswith(base_str) - + # Pathlib-based check (secure) try: pathlib_check = fake_file.resolve().is_relative_to(base_dir.resolve()) @@ -364,39 +360,50 @@ class TestFileUploadSecurity: def test_ui_upload_uses_basename(self): """Test that ui_upload extracts basename to prevent path traversal.""" import os - + + from app.utils.filename_utils import sanitize_filename + # Simulate malicious filenames malicious_filenames = [ "../../../etc/passwd", - "..\\..\\windows\\system32", "/etc/shadow", "folder/../file.pdf", ] - + for malicious in malicious_filenames: # os.path.basename should extract just the filename basename = os.path.basename(malicious) - + # Verify no path traversal remains in basename assert ".." not in basename, f"Path traversal not removed: {malicious} -> {basename}" assert "/" not in basename, f"Path separator not removed: {malicious} -> {basename}" - assert "\\" not in basename, f"Path separator not removed: {malicious} -> {basename}" + + # Windows-style backslash paths: os.path.basename on Linux does NOT + # split on backslash, so the application also uses sanitize_filename + # to handle these. Verify the combined approach is safe. + windows_paths = [ + "..\\..\\windows\\system32", + ] + for malicious in windows_paths: + sanitized = sanitize_filename(os.path.basename(malicious)) + assert ".." not in sanitized, f"Path traversal not removed after sanitize: {malicious} -> {sanitized}" + assert "\\" not in sanitized, f"Backslash not removed after sanitize: {malicious} -> {sanitized}" def test_sanitize_after_basename(self): """Test that sanitization happens after basename extraction.""" from app.utils.filename_utils import sanitize_filename import os - + malicious = "../../../passwd.pdf" - + # Step 1: Extract basename (as ui_upload does) basename = os.path.basename(malicious) assert basename == "passwd.pdf" - + # Step 2: Sanitize (as ui_upload does) sanitized = sanitize_filename(basename) assert sanitized == "passwd.pdf" - + # Final result is safe assert ".." not in sanitized assert "/" not in sanitized @@ -410,11 +417,11 @@ class TestFileHashSecurity: def test_hash_file_with_absolute_path_only(self, tmp_path): """Test that hash_file should only accept absolute paths.""" from app.utils.file_operations import hash_file - + # Create a test file test_file = tmp_path / "test.pdf" test_file.write_bytes(b"test content") - + # Should work with absolute path result = hash_file(str(test_file)) assert isinstance(result, str) @@ -423,7 +430,7 @@ class TestFileHashSecurity: def test_hash_file_rejects_path_traversal(self): """Test that hash_file doesn't allow path traversal.""" from app.utils.file_operations import hash_file - + # Attempt to hash a file using path traversal # This should fail because the file doesn't exist with pytest.raises(FileNotFoundError): @@ -440,25 +447,25 @@ class TestEndToEndPathTraversal: from app.utils.filename_utils import sanitize_filename import os import uuid - + # Simulate ui_upload flow malicious_upload_filename = "../../../etc/passwd" - + # Step 1: Extract basename base_filename = os.path.basename(malicious_upload_filename) assert base_filename == "passwd" - + # Step 2: Sanitize safe_filename = sanitize_filename(base_filename) assert safe_filename == "passwd" - + # Step 3: Add UUID (as ui_upload does) unique_id = str(uuid.uuid4()) target_filename = f"{unique_id}.{safe_filename}" - + # Step 4: Join with workdir target_path = os.path.join(str(tmp_path), target_filename) - + # Verify final path is safe final_path = Path(target_path) assert final_path.parent == tmp_path @@ -469,27 +476,27 @@ class TestEndToEndPathTraversal: """Test metadata embedding flow prevents path traversal.""" from app.utils.filename_utils import sanitize_filename import os - + # Simulate GPT returning malicious filename gpt_metadata = { "filename": "../../../etc/shadow", "document_type": "Invoice", } - + # Step 1: Extract filename from metadata suggested_filename = gpt_metadata.get("filename", "fallback") - + # Step 2: Sanitize (as embed_metadata_into_pdf should do) suggested_filename = sanitize_filename(suggested_filename) - + # Step 3: Remove extension suggested_filename = os.path.splitext(suggested_filename)[0] - + # Step 4: Build final path processed_dir = tmp_path / "processed" processed_dir.mkdir() final_path = os.path.join(str(processed_dir), f"{suggested_filename}.pdf") - + # Verify final path is safe result_path = Path(final_path) assert result_path.parent == processed_dir diff --git a/tests/test_rate_limiting.py b/tests/test_rate_limiting.py index 8b1827a6..05421764 100644 --- a/tests/test_rate_limiting.py +++ b/tests/test_rate_limiting.py @@ -32,7 +32,6 @@ def test_rate_limit_configuration(): assert hasattr(settings, "rate_limiting_enabled") assert hasattr(settings, "rate_limit_default") assert hasattr(settings, "rate_limit_upload") - assert hasattr(settings, "rate_limit_process") assert hasattr(settings, "rate_limit_auth") # Verify that settings are strings in correct format @@ -40,8 +39,6 @@ def test_rate_limit_configuration(): assert "/" in settings.rate_limit_default # Should be like "100/minute" assert isinstance(settings.rate_limit_upload, str) assert "/" in settings.rate_limit_upload - assert isinstance(settings.rate_limit_process, str) - assert "/" in settings.rate_limit_process assert isinstance(settings.rate_limit_auth, str) assert "/" in settings.rate_limit_auth @@ -224,7 +221,6 @@ def test_rate_limit_format_validation(): assert validate_rate_limit(settings.rate_limit_default) assert validate_rate_limit(settings.rate_limit_upload) - assert validate_rate_limit(settings.rate_limit_process) assert validate_rate_limit(settings.rate_limit_auth)