import asyncio import json import re import tempfile from contextlib import suppress from io import BytesIO from pathlib import Path # we need to import tmpdir import anyio import pytest from asgi_lifespan import LifespanManager from fastapi import status from httpx import ASGITransport, AsyncClient from langflow.main import create_app from langflow.services.auth.utils import get_password_hash from langflow.services.database.models.api_key.model import ApiKey from langflow.services.database.models.flow.model import Flow, FlowCreate from langflow.services.database.models.user.model import User, UserRead from lfx.services.deps import session_scope from sqlalchemy.orm import selectinload from sqlmodel import select from tests.conftest import _delete_transactions_and_vertex_builds @pytest.fixture(name="files_created_api_key") async def files_created_api_key(files_client, files_active_user): # noqa: ARG001 hashed = get_password_hash("random_key") api_key = ApiKey( name="files_created_api_key", user_id=files_active_user.id, api_key="random_key", hashed_api_key=hashed, ) async with session_scope() as session: stmt = select(ApiKey).where(ApiKey.api_key == api_key.api_key) if existing_api_key := (await session.exec(stmt)).first(): api_key = existing_api_key else: session.add(api_key) await session.flush() await session.refresh(api_key) # Yield outside session scope to avoid database locks yield api_key # Clean up async with session_scope() as session: # Re-attach api_key to new session key_to_delete = await session.get(ApiKey, api_key.id) if key_to_delete: await session.delete(key_to_delete) @pytest.fixture(name="files_other_created_api_key") async def files_other_created_api_key(files_client, files_other_active_user): # noqa: ARG001 # Separate API key to emulate requests from a second tenant. hashed = get_password_hash("other_random_key") api_key = ApiKey( name="files_other_created_api_key", user_id=files_other_active_user.id, api_key="other_random_key", hashed_api_key=hashed, ) # Reuse an existing key when present, otherwise create one, then clean it up after the test. async with session_scope() as session: stmt = select(ApiKey).where(ApiKey.api_key == api_key.api_key) if existing_api_key := (await session.exec(stmt)).first(): api_key = existing_api_key else: session.add(api_key) await session.flush() await session.refresh(api_key) yield api_key async with session_scope() as session: key_to_delete = await session.get(ApiKey, api_key.id) if key_to_delete: await session.delete(key_to_delete) @pytest.fixture(name="files_active_user") async def files_active_user(files_client): # noqa: ARG001 async with session_scope() as session: user = User( username="files_active_user", password=get_password_hash("testpassword"), is_active=True, is_superuser=False, ) stmt = select(User).where(User.username == user.username) if active_user := (await session.exec(stmt)).first(): user = active_user else: session.add(user) await session.flush() await session.refresh(user) user = UserRead.model_validate(user, from_attributes=True) yield user # Clean up # Now cleanup transactions, vertex_build async with session_scope() as session: user = await session.get(User, user.id, options=[selectinload(User.flows)]) await _delete_transactions_and_vertex_builds(session, user.flows) await session.delete(user) @pytest.fixture(name="files_other_active_user") async def files_other_active_user(files_client): # noqa: ARG001 # Secondary user used to validate cross-user access restrictions. async with session_scope() as session: user = User( username="files_other_active_user", password=get_password_hash("testpassword"), is_active=True, is_superuser=False, ) stmt = select(User).where(User.username == user.username) if active_user := (await session.exec(stmt)).first(): user = active_user else: session.add(user) await session.flush() await session.refresh(user) user = UserRead.model_validate(user, from_attributes=True) yield user async with session_scope() as session: user = await session.get(User, user.id, options=[selectinload(User.flows)]) await _delete_transactions_and_vertex_builds(session, user.flows) await session.delete(user) @pytest.fixture(name="files_flow") async def files_flow( files_client, # noqa: ARG001 json_flow: str, files_active_user, ): loaded_json = json.loads(json_flow) flow_data = FlowCreate(name="test_flow", data=loaded_json.get("data"), user_id=files_active_user.id) flow = Flow.model_validate(flow_data.model_dump(exclude={"id"})) async with session_scope() as session: session.add(flow) await session.flush() await session.refresh(flow) # Yield outside session scope to avoid database locks yield flow # Clean up async with session_scope() as session: # Re-attach flow to new session flow_to_delete = await session.get(Flow, flow.id) if flow_to_delete: await session.delete(flow_to_delete) @pytest.fixture def max_file_size_upload_fixture(monkeypatch): monkeypatch.setenv("LANGFLOW_MAX_FILE_SIZE_UPLOAD", "1") yield monkeypatch.undo() @pytest.fixture def max_file_size_upload_10mb_fixture(monkeypatch): monkeypatch.setenv("LANGFLOW_MAX_FILE_SIZE_UPLOAD", "10") yield monkeypatch.undo() @pytest.fixture(name="files_client") async def files_client_fixture( monkeypatch, request, ): # Set the database url to a test database if "noclient" in request.keywords: yield else: def init_app(): db_dir = tempfile.mkdtemp() db_path = Path(db_dir) / "test.db" monkeypatch.setenv("LANGFLOW_DATABASE_URL", f"sqlite:///{db_path}") monkeypatch.setenv("LANGFLOW_AUTO_LOGIN", "false") from lfx.services.manager import get_service_manager get_service_manager().factories.clear() get_service_manager().services.clear() # Clear the services cache app = create_app() return app, db_path app, db_path = await asyncio.to_thread(init_app) async with ( LifespanManager(app, startup_timeout=None, shutdown_timeout=60) as manager, AsyncClient(transport=ASGITransport(app=manager.app), base_url="http://testserver/") as client, ): yield client # app.dependency_overrides.clear() monkeypatch.undo() # clear the temp db with suppress(FileNotFoundError): await anyio.Path(db_path).unlink() async def test_upload_file(files_client, files_created_api_key, files_flow): headers = {"x-api-key": files_created_api_key.api_key} response = await files_client.post( f"api/v1/files/upload/{files_flow.id}", files={"file": ("test.txt", b"test content")}, headers=headers, ) assert response.status_code == 201, f"Expected 201, got {response.status_code}: {response.json()}" response_json = response.json() assert response_json["flowId"] == str(files_flow.id) # Check that the file_path matches the expected pattern file_path_pattern = re.compile(rf"{files_flow.id}/\d{{4}}-\d{{2}}-\d{{2}}_\d{{2}}-\d{{2}}-\d{{2}}_test\.txt") assert file_path_pattern.match(response_json["file_path"]) async def test_download_file(files_client, files_created_api_key, files_flow): headers = {"x-api-key": files_created_api_key.api_key} # First upload a file response = await files_client.post( f"api/v1/files/upload/{files_flow.id}", files={"file": ("test.txt", b"test content")}, headers=headers, ) assert response.status_code == 201 # Get the actual filename from the response file_path = response.json()["file_path"] file_name = file_path.split("/")[-1] # Then try to download it response = await files_client.get(f"api/v1/files/download/{files_flow.id}/{file_name}", headers=headers) assert response.status_code == 200 assert response.content == b"test content" async def test_list_files(files_client, files_created_api_key, files_flow): headers = {"x-api-key": files_created_api_key.api_key} # First upload a file response = await files_client.post( f"api/v1/files/upload/{files_flow.id}", files={"file": ("test.txt", b"test content")}, headers=headers, ) assert response.status_code == 201 # Then list the files response = await files_client.get(f"api/v1/files/list/{files_flow.id}", headers=headers) assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.json()}" files = response.json()["files"] assert len(files) == 1 assert files[0].endswith("test.txt") async def test_delete_file(files_client, files_created_api_key, files_flow): headers = {"x-api-key": files_created_api_key.api_key} response = await files_client.delete(f"api/v1/files/delete/{files_flow.id}/test.txt", headers=headers) assert response.status_code == 200 assert response.json() == {"message": "File test.txt deleted successfully"} async def test_file_operations(files_client, files_created_api_key, files_flow): headers = {"x-api-key": files_created_api_key.api_key} flow_id = files_flow.id file_name = "test.txt" file_content = b"Hello, world!" # Step 1: Upload the file response = await files_client.post( f"api/v1/files/upload/{flow_id}", files={"file": (file_name, file_content)}, headers=headers, ) assert response.status_code == 201 response_json = response.json() assert response_json["flowId"] == str(flow_id) # Check that the file_path matches the expected pattern file_path_pattern = re.compile(rf"{flow_id}/\d{{4}}-\d{{2}}-\d{{2}}_\d{{2}}-\d{{2}}-\d{{2}}_{file_name}") assert file_path_pattern.match(response_json["file_path"]) # Extract the full file name with timestamp from the response full_file_name = response_json["file_path"].split("/")[-1] # Step 2: List files in the folder response = await files_client.get(f"api/v1/files/list/{files_flow.id}", headers=headers) assert response.status_code == 200 assert full_file_name in response.json()["files"] # Step 3: Download the file and verify its content response = await files_client.get(f"api/v1/files/download/{files_flow.id}/{full_file_name}", headers=headers) assert response.status_code == 200 assert response.content == file_content assert response.headers["content-type"] == "application/octet-stream" # Step 4: Delete the file response = await files_client.delete(f"api/v1/files/delete/{files_flow.id}/{full_file_name}", headers=headers) assert response.status_code == 200 assert response.json() == {"message": f"File {full_file_name} deleted successfully"} # Verify that the file is indeed deleted response = await files_client.get(f"api/v1/files/list/{files_flow.id}", headers=headers) assert full_file_name not in response.json()["files"] @pytest.mark.usefixtures("max_file_size_upload_fixture") async def test_upload_file_size_limit(files_client, files_created_api_key, files_flow): headers = {"x-api-key": files_created_api_key.api_key} # Test file under the limit (500KB) small_content = b"x" * (500 * 1024) small_file = ("small_file.txt", small_content, "application/octet-stream") headers["Content-Length"] = str(len(small_content)) response = await files_client.post( f"api/v1/files/upload/{files_flow.id}", files={"file": small_file}, headers=headers, ) assert response.status_code == 201, f"Expected 201, got {response.status_code}: {response.json()}" # Test file over the limit (1MB + 1KB) large_content = b"x" * (1024 * 1024 + 1024) bio = BytesIO(large_content) headers["Content-Length"] = str(len(large_content)) response = await files_client.post( f"api/v1/files/upload/{files_flow.id}", files={"file": ("large_file.txt", bio, "application/octet-stream")}, headers=headers, ) assert response.status_code == 413, f"Expected 413, got {response.status_code}: {response.json()}" assert "Content size limit exceeded. Maximum allowed is 1MB and got 1.001MB." in response.json()["detail"] @pytest.fixture async def setup_profile_pictures(monkeypatch): """Fixture to set up profile pictures in a temporary config directory. This fixture must run before files_client to set LANGFLOW_CONFIG_DIR before app initialization. Args: monkeypatch: For overriding environment variables """ # Create a temporary directory for profile pictures temp_dir = tempfile.mkdtemp() config_path = Path(temp_dir) # Create profile pictures directory structure people_dir = config_path / "profile_pictures" / "People" space_dir = config_path / "profile_pictures" / "Space" people_dir.mkdir(parents=True, exist_ok=True) space_dir.mkdir(parents=True, exist_ok=True) # Create test profile picture files (must be > 100 bytes for test assertions) rocket_svg = ( b'' b'' b'' ) person_svg = ( b'' b'' b'' b'' ) (space_dir / "046-rocket.svg").write_bytes(rocket_svg) (people_dir / "001-person.svg").write_bytes(person_svg) # Override the config_dir setting BEFORE app initialization monkeypatch.setenv("LANGFLOW_CONFIG_DIR", str(config_path)) yield config_path # Cleanup import shutil shutil.rmtree(temp_dir, ignore_errors=True) async def test_list_profile_pictures(setup_profile_pictures, files_client): # noqa: ARG001 """Test listing profile pictures from local filesystem. Args: files_client: HTTP client for making API requests setup_profile_pictures: Fixture that sets up profile pictures directory """ response = await files_client.get("api/v1/files/profile_pictures/list") assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.json()}" data = response.json() assert "files" in data files = data["files"] # The app copies all profile pictures during initialization, # so we should have many files including our test ones assert len(files) > 0, "Should have at least some profile pictures" # Check that files are properly formatted as "Folder/filename" assert any(file.startswith("Space/") for file in files), "Should have Space profile pictures" assert any(file.startswith("People/") for file in files), "Should have People profile pictures" # Check that the rocket file exists (either our test one or the real one) assert "Space/046-rocket.svg" in files, "Should have the rocket profile picture" async def test_download_profile_picture_space_rocket(setup_profile_pictures, files_client): # noqa: ARG001 """Test downloading the rocket profile picture from Space folder. Args: files_client: HTTP client for making API requests setup_profile_pictures: Fixture that sets up profile pictures directory """ response = await files_client.get("api/v1/files/profile_pictures/Space/046-rocket.svg") assert response.status_code == 200 # Verify content type assert "image/svg+xml" in response.headers["content-type"] # Verify content - check for SVG structure (real rocket.svg has elements) content = response.content assert b"" in content # Real rocket has path elements, not circles assert len(content) > 100, "SVG content should be substantial" async def test_download_profile_picture_people(setup_profile_pictures, files_client): # noqa: ARG001 """Test downloading a profile picture from People folder. Note: The actual people profile pictures are copied during app init, so we test with whatever profile picture exists. Args: files_client: HTTP client for making API requests setup_profile_pictures: Fixture that sets up profile pictures directory """ # List available people profile pictures first list_response = await files_client.get("api/v1/files/profile_pictures/list") assert list_response.status_code == 200 people_files = [f for f in list_response.json()["files"] if f.startswith("People/")] # Skip test if no people profile pictures are available if not people_files: import pytest pytest.skip("No people profile pictures available") # Test downloading the first available people profile picture first_people_file = people_files[0].replace("People/", "") response = await files_client.get(f"api/v1/files/profile_pictures/People/{first_people_file}") assert response.status_code == 200 # Verify content type assert "image/svg+xml" in response.headers["content-type"] # Verify content content = response.content assert b"" in content assert len(content) > 100, "SVG content should be substantial" async def test_download_profile_picture_not_found(setup_profile_pictures, files_client): # noqa: ARG001 """Test downloading a non-existent profile picture returns 404. Args: files_client: HTTP client for making API requests setup_profile_pictures: Fixture that sets up profile pictures directory """ response = await files_client.get("api/v1/files/profile_pictures/Space/nonexistent.svg") assert response.status_code == 404 data = response.json() assert "not found" in data["detail"].lower() async def test_profile_pictures_with_s3_storage(setup_profile_pictures, files_client, monkeypatch): # noqa: ARG001 """Test that profile pictures work with S3 storage type. Profile pictures should always be served from local filesystem, regardless of the storage_type setting. Args: files_client: HTTP client for making API requests setup_profile_pictures: Fixture that sets up profile pictures directory monkeypatch: For overriding environment variables """ # Set storage type to S3 (simulating S3 configuration) monkeypatch.setenv("LANGFLOW_STORAGE_TYPE", "s3") # List should still work (from local filesystem) response = await files_client.get("api/v1/files/profile_pictures/list") assert response.status_code == 200 data = response.json() # Should have profile pictures (app copies them during init) assert len(data["files"]) > 0, "Should have profile pictures even with S3 storage" assert "Space/046-rocket.svg" in data["files"], "Should have rocket profile picture" # Download should still work (from local filesystem) response = await files_client.get("api/v1/files/profile_pictures/Space/046-rocket.svg") assert response.status_code == 200 assert b"" in content async def test_list_profile_pictures_fallback_to_package(empty_config_dir, files_client): # noqa: ARG001 """Test that list endpoint fallbacks to package when config_dir is empty. This tests the list fallback from PR #10758 - when config_dir has no profile pictures, it should list from the package's bundled directory. """ response = await files_client.get("api/v1/files/profile_pictures/list") assert response.status_code == 200 data = response.json() assert "files" in data files = data["files"] # Should have files from the package assert len(files) > 0, "Should have profile pictures from package fallback" assert any(f.startswith("Space/") for f in files), "Should have Space profile pictures" assert any(f.startswith("People/") for f in files), "Should have People profile pictures" # The bundled rocket should be available assert "Space/046-rocket.svg" in files async def test_download_profile_picture_not_found_in_both_locations(empty_config_dir, files_client): # noqa: ARG001 """Test 404 when profile picture doesn't exist in config_dir OR package. This ensures the fallback logic correctly returns 404 when the file is not found in either location. """ response = await files_client.get("api/v1/files/profile_pictures/Space/nonexistent-file-xyz.svg") assert response.status_code == 404 data = response.json() assert "not found" in data["detail"].lower() assert "Space/nonexistent-file-xyz.svg" in data["detail"] async def test_download_profile_picture_invalid_folder(empty_config_dir, files_client): # noqa: ARG001 """Test 400 when using an invalid folder name. Only 'People' and 'Space' folders are whitelisted for security. Invalid folder names should be rejected with 400 Bad Request. """ response = await files_client.get("api/v1/files/profile_pictures/InvalidFolder/file.svg") assert response.status_code == 400 data = response.json() # Check for the new specific error message assert "folder must be one of" in data["detail"].lower() async def test_download_profile_picture_config_dir_takes_precedence(setup_profile_pictures, files_client): """Test that config_dir profile pictures take precedence over package. When a file exists in both config_dir and package, the config_dir version should be served. """ config_path = setup_profile_pictures # Create a custom rocket SVG in config_dir with identifiable content custom_svg = b'CUSTOM_CONFIG_DIR_VERSION' space_dir = config_path / "profile_pictures" / "Space" space_dir.mkdir(parents=True, exist_ok=True) (space_dir / "046-rocket.svg").write_bytes(custom_svg) response = await files_client.get("api/v1/files/profile_pictures/Space/046-rocket.svg") assert response.status_code == 200 # Should get the config_dir version, not the package version content = response.content assert b"CUSTOM_CONFIG_DIR_VERSION" in content async def test_list_profile_pictures_config_dir_takes_precedence(setup_profile_pictures, files_client): """Test that config_dir listing takes precedence over package. When config_dir has profile pictures, they should be listed instead of the package's bundled ones. """ config_path = setup_profile_pictures # Ensure we have a file in config_dir space_dir = config_path / "profile_pictures" / "Space" space_dir.mkdir(parents=True, exist_ok=True) (space_dir / "custom-test-file.svg").write_bytes(b"") response = await files_client.get("api/v1/files/profile_pictures/list") assert response.status_code == 200 data = response.json() files = data["files"] # Should include our custom file from config_dir assert "Space/custom-test-file.svg" in files async def test_download_profile_picture_path_traversal_attempt(empty_config_dir, files_client): # noqa: ARG001 """Test that path traversal attacks are prevented. Attempting to access files outside the profile_pictures directory using '../' should not work. """ # Try path traversal to access parent directories response = await files_client.get("api/v1/files/profile_pictures/../../../etc/passwd") # Should be 400 (invalid input), 404 (not found), or 422 (validation error) assert response.status_code in [400, 404, 422, 500] async def test_download_profile_picture_special_characters_in_filename(empty_config_dir, files_client): # noqa: ARG001 """Test handling of special characters in filename. Filenames with spaces or special characters should be handled properly. """ # Test with URL-encoded space (the real file has a space: "042-space shuttle.svg") response = await files_client.get("api/v1/files/profile_pictures/Space/042-space%20shuttle.svg") # Should work if the file exists with that name, or 404 if not assert response.status_code in [200, 404] async def test_list_profile_pictures_empty_response_format(empty_config_dir, files_client): # noqa: ARG001 """Test that the list response format is correct even with fallback. The response should always have the correct format: {"files": [...]} """ response = await files_client.get("api/v1/files/profile_pictures/list") assert response.status_code == 200 data = response.json() # Verify response structure assert isinstance(data, dict) assert "files" in data assert isinstance(data["files"], list) # Verify file path format (should be "Folder/filename") for file_path in data["files"]: assert "/" in file_path, f"File path should contain '/': {file_path}" folder, filename = file_path.split("/", 1) assert folder in ["People", "Space"], f"Invalid folder: {folder}" assert len(filename) > 0, "Filename should not be empty" async def test_download_profile_picture_content_is_valid_svg(empty_config_dir, files_client): # noqa: ARG001 """Test that downloaded profile pictures are valid SVG files. This ensures the fallback serves actual SVG content, not corrupted data. """ # Download the rocket from package fallback response = await files_client.get("api/v1/files/profile_pictures/Space/046-rocket.svg") assert response.status_code == 200 content = response.content # Verify it's valid XML/SVG assert content.startswith((b"<", b"" in content.lower(), "Should have closing svg tag" @pytest.fixture async def partial_config_dir(monkeypatch): """Fixture that sets up a config directory with only People folder (no Space). This tests the edge case where config_dir is partially populated. """ temp_dir = tempfile.mkdtemp() config_path = Path(temp_dir) # Create only People directory with a file people_dir = config_path / "profile_pictures" / "People" people_dir.mkdir(parents=True, exist_ok=True) (people_dir / "test-person.svg").write_bytes(b"") # Note: Space directory is NOT created intentionally # Override the config_dir setting BEFORE app initialization monkeypatch.setenv("LANGFLOW_CONFIG_DIR", str(config_path)) yield config_path # Cleanup import shutil shutil.rmtree(temp_dir, ignore_errors=True) async def test_profile_pictures_fallback_with_partial_config_dir(partial_config_dir, files_client): # noqa: ARG001 """Test fallback when config_dir has only People folder but not Space. This is an edge case where config_dir is partially populated. """ # For list: since we have at least one file in People, it should NOT fallback completely # The current implementation only falls back if BOTH people AND space are empty response = await files_client.get("api/v1/files/profile_pictures/list") assert response.status_code == 200 data = response.json() files = data["files"] # Should have our People file from config_dir assert "People/test-person.svg" in files # For download: Space files should still work via fallback to package response = await files_client.get("api/v1/files/profile_pictures/Space/046-rocket.svg") assert response.status_code == 200, "Space files should fallback to package" # ============================================================================ # Tests for image download endpoint # These tests ensure the /images endpoint works correctly for browser tags. # Regression tests for the fix that reverted commit 7ba8c73 changes to /images. # ============================================================================ async def test_download_image_for_browser(files_client, files_created_api_key, files_flow): """Test that images can be downloaded for browser tag rendering. Regression test: commit 7ba8c73 broke browser image display. """ headers = {"x-api-key": files_created_api_key.api_key} # First upload an image (this requires auth) # Create a minimal valid PNG (1x1 transparent pixel) png_content = ( b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01" b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00\x01" b"\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82" ) response = await files_client.post( f"api/v1/files/upload/{files_flow.id}", files={"file": ("test_image.png", png_content, "image/png")}, headers=headers, ) assert response.status_code == 201, f"Upload failed: {response.json()}" # Get the actual filename from the response file_path = response.json()["file_path"] file_name = file_path.split("/")[-1] # Download the image - simulates browser tag behavior response = await files_client.get( f"api/v1/files/images/{files_flow.id}/{file_name}", headers=headers, ) assert response.status_code == 200, ( f"Image download failed with {response.status_code}. This breaks browser tags in chat." ) # Verify content type is image assert "image" in response.headers.get("content-type", ""), "Response should be an image" async def test_download_image_returns_correct_content_type(files_client, files_created_api_key, files_flow): """Test that the /images endpoint returns correct content-type for images.""" headers = {"x-api-key": files_created_api_key.api_key} # Create a minimal valid PNG png_content = ( b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01" b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00\x01" b"\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82" ) response = await files_client.post( f"api/v1/files/upload/{files_flow.id}", files={"file": ("test.png", png_content, "image/png")}, headers=headers, ) assert response.status_code == 201 file_path = response.json()["file_path"] file_name = file_path.split("/")[-1] # Download image response = await files_client.get(f"api/v1/files/images/{files_flow.id}/{file_name}", headers=headers) assert response.status_code == 200 assert "image/png" in response.headers.get("content-type", "") async def test_download_image_rejects_non_image_files(files_client, files_created_api_key, files_flow): """Test that the /images endpoint rejects non-image files.""" headers = {"x-api-key": files_created_api_key.api_key} # Upload a text file response = await files_client.post( f"api/v1/files/upload/{files_flow.id}", files={"file": ("test.txt", b"not an image", "text/plain")}, headers=headers, ) assert response.status_code == 201 file_path = response.json()["file_path"] file_name = file_path.split("/")[-1] # Try to download via /images endpoint (should fail) response = await files_client.get(f"api/v1/files/images/{files_flow.id}/{file_name}", headers=headers) # Should reject non-image content types assert response.status_code == 500 assert "not an image" in response.json().get("detail", "").lower() async def test_download_image_requires_authentication(files_client, files_created_api_key, files_flow): """Test that /images rejects unauthenticated requests.""" headers = {"x-api-key": files_created_api_key.api_key} png_content = ( b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01" b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00\x01" b"\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82" ) response = await files_client.post( f"api/v1/files/upload/{files_flow.id}", files={"file": ("auth_required.png", png_content, "image/png")}, headers=headers, ) assert response.status_code == status.HTTP_201_CREATED file_path = response.json()["file_path"] file_name = file_path.split("/")[-1] # No auth header: browser-like unauthenticated request must be rejected. response = await files_client.get(f"api/v1/files/images/{files_flow.id}/{file_name}") assert response.status_code == status.HTTP_403_FORBIDDEN async def test_download_image_with_invalid_flow_id(files_client, files_created_api_key): """Test that /images returns 404 for non-existent flow_id when authenticated.""" import uuid fake_flow_id = uuid.uuid4() headers = {"x-api-key": files_created_api_key.api_key} response = await files_client.get(f"api/v1/files/images/{fake_flow_id}/nonexistent.png", headers=headers) assert response.status_code == status.HTTP_404_NOT_FOUND async def test_download_image_returns_404_for_other_users_flow( files_client, files_created_api_key, files_other_created_api_key, files_flow ): """Test that /images does not disclose images across tenants.""" owner_headers = {"x-api-key": files_created_api_key.api_key} other_headers = {"x-api-key": files_other_created_api_key.api_key} png_content = ( b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01" b"\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00\x01" b"\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82" ) upload_response = await files_client.post( f"api/v1/files/upload/{files_flow.id}", files={"file": ("browser_test.png", png_content, "image/png")}, headers=owner_headers, ) assert upload_response.status_code == status.HTTP_201_CREATED file_path = upload_response.json()["file_path"] file_name = file_path.split("/")[-1] # Hide resource existence across tenants by returning 404 for non-owners. download_response = await files_client.get( f"api/v1/files/images/{files_flow.id}/{file_name}", headers=other_headers, ) assert download_response.status_code == status.HTTP_404_NOT_FOUND # ============================================================================ # Tests for path traversal protection (ValidatedFileName) # ============================================================================ @pytest.mark.parametrize( "malicious_filename", [ # Backslash-based traversal (Windows style) "..\\..\\..\\etc\\passwd", "..\\secret.txt", # Double-dot embedded in filename "..txt", "test..secret", ], ) async def test_download_file_path_traversal_rejected( files_client, files_created_api_key, files_flow, malicious_filename ): """Test that path traversal attempts are rejected on /download endpoint.""" headers = {"x-api-key": files_created_api_key.api_key} response = await files_client.get( f"api/v1/files/download/{files_flow.id}/{malicious_filename}", headers=headers, ) assert response.status_code == 400, f"Path traversal should be rejected: {malicious_filename}" assert "invalid file name" in response.json()["detail"].lower() assert "simple file name" in response.json()["detail"].lower() @pytest.mark.parametrize( "malicious_filename", [ # Forward slashes in URL path are treated as path separators by FastAPI, # so the router returns 404 (route not found) - this is defense in depth "../../../etc/passwd", "../secret.txt", "subdir/file.txt", # URL-encoded forward slashes are also decoded by FastAPI and treated as path separators "..%2F..%2F..%2Fetc%2Fpasswd", "subdir%2Ffile.txt", ], ) async def test_download_file_forward_slash_traversal_blocked( files_client, files_created_api_key, files_flow, malicious_filename ): """Test that forward slash path traversal is blocked by FastAPI routing. FastAPI treats both literal and URL-encoded forward slashes as path separators, providing defense in depth before our validation layer is reached. """ headers = {"x-api-key": files_created_api_key.api_key} url = f"api/v1/files/download/{files_flow.id}/{malicious_filename}" response = await files_client.get(url, headers=headers) # FastAPI returns 404 because the path with slashes doesn't match any route assert response.status_code == 404, ( f"Forward slash traversal should be blocked by routing: {malicious_filename}, got {response.status_code}" ) @pytest.mark.parametrize( "malicious_filename", [ # Backslash-based traversal (Windows style) "..\\..\\..\\etc\\passwd.png", "..\\secret.png", # Double-dot embedded in filename "..png", ], ) async def test_download_image_path_traversal_rejected(files_client, files_flow, malicious_filename): """Test that path traversal attempts are rejected on /images endpoint.""" response = await files_client.get( f"api/v1/files/images/{files_flow.id}/{malicious_filename}", ) assert response.status_code == 400, f"Path traversal should be rejected: {malicious_filename}" assert "invalid file name" in response.json()["detail"].lower() assert "simple file name" in response.json()["detail"].lower() @pytest.mark.parametrize( "malicious_filename", [ # Forward slashes in URL path are treated as path separators by FastAPI, # so the router returns 404 (route not found) - this is defense in depth "../../../etc/passwd.png", "../secret.png", "subdir/image.png", # URL-encoded forward slashes are also decoded by FastAPI and treated as path separators "..%2F..%2Fimage.png", "subdir%2Fimage.png", ], ) async def test_download_image_forward_slash_traversal_blocked(files_client, files_flow, malicious_filename): """Test that forward slash path traversal is blocked by FastAPI routing on /images endpoint.""" url = f"api/v1/files/images/{files_flow.id}/{malicious_filename}" response = await files_client.get(url) # FastAPI returns 404 because the path with slashes doesn't match any route assert response.status_code == 404, ( f"Forward slash traversal should be blocked by routing: {malicious_filename}, got {response.status_code}" ) @pytest.mark.parametrize( "malicious_filename", [ # Backslash-based traversal (Windows style) "..\\..\\..\\etc\\passwd", "..\\secret.txt", # Double-dot embedded in filename "..txt", ], ) async def test_delete_file_path_traversal_rejected(files_client, files_created_api_key, files_flow, malicious_filename): """Test that path traversal attempts are rejected on /delete endpoint.""" headers = {"x-api-key": files_created_api_key.api_key} response = await files_client.delete( f"api/v1/files/delete/{files_flow.id}/{malicious_filename}", headers=headers, ) assert response.status_code == 400, f"Path traversal should be rejected: {malicious_filename}" assert "invalid file name" in response.json()["detail"].lower() assert "simple file name" in response.json()["detail"].lower() @pytest.mark.parametrize( "malicious_filename", [ # Forward slashes in URL path are treated as path separators by FastAPI, # so the router returns 404 (route not found) - this is defense in depth "../../../etc/passwd", "../secret.txt", "subdir/file.txt", # URL-encoded forward slashes are also decoded by FastAPI and treated as path separators "..%2F..%2F..%2Fetc%2Fpasswd", "subdir%2Ffile.txt", ], ) async def test_delete_file_forward_slash_traversal_blocked( files_client, files_created_api_key, files_flow, malicious_filename ): """Test that forward slash path traversal is blocked by FastAPI routing on /delete endpoint.""" headers = {"x-api-key": files_created_api_key.api_key} url = f"api/v1/files/delete/{files_flow.id}/{malicious_filename}" response = await files_client.delete(url, headers=headers) # FastAPI returns 404 because the path with slashes doesn't match any route assert response.status_code == 404, ( f"Forward slash traversal should be blocked by routing: {malicious_filename}, got {response.status_code}" )