"""Unit tests for flow filesystem path validation security.""" from unittest.mock import MagicMock from uuid import uuid4 import anyio import pytest from fastapi import HTTPException from langflow.api.v1.flows import _get_safe_flow_path from langflow.services.storage.service import StorageService @pytest.fixture def mock_storage_service(tmp_path): """Create a mock storage service with a temporary data directory.""" service = MagicMock(spec=StorageService) service.data_dir = tmp_path return service @pytest.fixture def user_id(): """Create a test user ID.""" return uuid4() class TestPathValidation: """Test cases for path validation security.""" def test_rejects_absolute_path_outside_allowed_directory(self, mock_storage_service, user_id): """Test that absolute paths outside the allowed directory are rejected.""" with pytest.raises(HTTPException) as exc_info: _get_safe_flow_path("/etc/passwd", user_id, mock_storage_service) assert exc_info.value.status_code == 400 assert "within" in exc_info.value.detail.lower() or "outside" in exc_info.value.detail.lower() @pytest.mark.asyncio async def test_accepts_absolute_path_within_allowed_directory(self, mock_storage_service, user_id, tmp_path): """Test that absolute paths within the user's flows directory are accepted.""" from pathlib import Path as StdlibPath import anyio mock_storage_service.data_dir = anyio.Path(tmp_path) base_dir = mock_storage_service.data_dir / "flows" / str(user_id) await base_dir.mkdir(parents=True, exist_ok=True) # Create an absolute path within the allowed directory (resolve to get actual absolute path) base_dir_stdlib = StdlibPath(str(base_dir)).resolve() allowed_absolute = str(base_dir_stdlib / "my_flow.json") path = _get_safe_flow_path(allowed_absolute, user_id, mock_storage_service) assert path is not None # Verify the returned path matches what we expect assert str(path) == allowed_absolute or str(path).endswith("my_flow.json") def test_rejects_directory_traversal(self, mock_storage_service, user_id): """Test that directory traversal sequences are rejected.""" with pytest.raises(HTTPException) as exc_info: _get_safe_flow_path("../../etc/passwd", user_id, mock_storage_service) assert exc_info.value.status_code == 400 assert "directory traversal" in exc_info.value.detail.lower() def test_rejects_backslash_directory_traversal(self, mock_storage_service, user_id): """Test that backslash directory traversal is caught after normalization.""" # Backslash traversal should be normalized first, then caught with pytest.raises(HTTPException) as exc_info: _get_safe_flow_path("sub\\..\\etc\\passwd", user_id, mock_storage_service) assert exc_info.value.status_code == 400 assert "directory traversal" in exc_info.value.detail.lower() def test_rejects_windows_backslash_traversal(self, mock_storage_service, user_id): """Test that Windows-style backslash traversal is rejected.""" with pytest.raises(HTTPException) as exc_info: _get_safe_flow_path("folder\\..\\..\\etc\\passwd", user_id, mock_storage_service) assert exc_info.value.status_code == 400 assert "directory traversal" in exc_info.value.detail.lower() def test_rejects_multiple_traversal(self, mock_storage_service, user_id): """Test that multiple directory traversals are rejected.""" with pytest.raises(HTTPException) as exc_info: _get_safe_flow_path("../../../etc/passwd", user_id, mock_storage_service) assert exc_info.value.status_code == 400 def test_rejects_traversal_in_subpath(self, mock_storage_service, user_id): """Test that traversal in subpaths is rejected.""" with pytest.raises(HTTPException) as exc_info: _get_safe_flow_path("subfolder/../../etc/passwd", user_id, mock_storage_service) assert exc_info.value.status_code == 400 def test_rejects_null_bytes(self, mock_storage_service, user_id): """Test that null bytes are rejected.""" with pytest.raises(HTTPException) as exc_info: _get_safe_flow_path("file\x00name.json", user_id, mock_storage_service) assert exc_info.value.status_code == 400 assert "null bytes" in exc_info.value.detail.lower() def test_rejects_empty_path(self, mock_storage_service, user_id): """Test that empty paths are rejected.""" with pytest.raises(HTTPException) as exc_info: _get_safe_flow_path("", user_id, mock_storage_service) assert exc_info.value.status_code == 400 def test_accepts_simple_relative_path(self, mock_storage_service, user_id): """Test that simple relative paths are accepted.""" path = _get_safe_flow_path("my_flow.json", user_id, mock_storage_service) assert path is not None # Verify it's within the user's flows directory assert str(user_id) in str(path) assert "flows" in str(path) def test_accepts_nested_relative_path(self, mock_storage_service, user_id): """Test that nested relative paths are accepted.""" path = _get_safe_flow_path("subfolder/my_flow.json", user_id, mock_storage_service) assert path is not None assert str(user_id) in str(path) assert "flows" in str(path) assert "subfolder" in str(path) def test_accepts_deeply_nested_path(self, mock_storage_service, user_id): """Test that deeply nested relative paths are accepted.""" path = _get_safe_flow_path("a/b/c/d/e/flow.json", user_id, mock_storage_service) assert path is not None assert str(user_id) in str(path) def test_path_is_user_isolated(self, mock_storage_service, user_id): # noqa: ARG002 """Test that paths are isolated per user.""" user1_id = uuid4() user2_id = uuid4() path1 = _get_safe_flow_path("flow.json", user1_id, mock_storage_service) path2 = _get_safe_flow_path("flow.json", user2_id, mock_storage_service) # Paths should be different and contain their respective user IDs assert str(path1) != str(path2) assert str(user1_id) in str(path1) assert str(user2_id) in str(path2) assert str(user1_id) not in str(path2) assert str(user2_id) not in str(path1) def test_handles_leading_slash_in_relative_path(self, mock_storage_service, user_id): """Test that leading slashes in relative paths are handled correctly.""" _get_safe_flow_path("flow.json", user_id, mock_storage_service) # Leading slash makes it absolute, so it will be checked against base directory # For a simple "/flow.json", it's not within the base dir, so it should be rejected with pytest.raises(HTTPException): _get_safe_flow_path("/flow.json", user_id, mock_storage_service) def test_accepts_paths_with_double_slash(self, mock_storage_service, user_id): """Test that paths with double slashes are normalized (not rejected, but normalized by Path).""" # Double slashes are normalized by the Path library, so they're acceptable # The security concern is directory traversal, not double slashes path = _get_safe_flow_path("sub//folder/file.json", user_id, mock_storage_service) assert path is not None def test_accepts_valid_filename_characters(self, mock_storage_service, user_id): """Test that valid filename characters are accepted.""" valid_paths = [ "flow.json", "my-flow.json", "flow_123.json", "flow.name.json", "subfolder/flow.json", "flow (1).json", ] for valid_path in valid_paths: path = _get_safe_flow_path(valid_path, user_id, mock_storage_service) assert path is not None @pytest.mark.asyncio async def test_path_resolves_within_base_directory(self, mock_storage_service, user_id, tmp_path): """Test that resolved paths stay within the base directory.""" from pathlib import Path as StdlibPath # Create a real path structure to test resolution mock_storage_service.data_dir = anyio.Path(tmp_path) path = _get_safe_flow_path("flow.json", user_id, mock_storage_service) # Use stdlib Path for synchronous resolution check resolved = StdlibPath(str(path)).resolve() base_dir_str = str(mock_storage_service.data_dir / "flows" / str(user_id)) resolved_base = StdlibPath(base_dir_str).resolve() # Resolved path should start with resolved base assert str(resolved).startswith(str(resolved_base)) @pytest.mark.asyncio async def test_rejects_symlink_to_prefix_collision_sibling(self, tmp_path): """Regression for CWE-22: symlink resolving to a sibling dir that is a prefix match. If base_dir is /tmp/flows/, a symlink inside it that resolves to /tmp/flows/-evil/stolen.json must be rejected. Without the trailing separator in the startswith check, the old code would accept it. """ from uuid import UUID mock_service = MagicMock(spec=StorageService) mock_service.data_dir = anyio.Path(tmp_path) user_id = UUID("00000000-0000-0000-0000-000000000001") base_dir = tmp_path / "flows" / str(user_id) base_dir.mkdir(parents=True, exist_ok=True) # Create a sibling directory whose name is a prefix-match of base_dir evil_dir = tmp_path / "flows" / (str(user_id) + "-evil") evil_dir.mkdir(parents=True, exist_ok=True) evil_file = evil_dir / "stolen.json" evil_file.write_text("stolen") # Symlink inside user's dir that points to the evil sibling file symlink = base_dir / "link" symlink.symlink_to(evil_file) with pytest.raises(HTTPException) as exc_info: _get_safe_flow_path("link", user_id, mock_service) assert exc_info.value.status_code == 400