fix: prevent arbitrary file write via path traversal in files endpoint (#12227)

* fix(security): prevent arbitrary file write via path traversal in file uploads

* [autofix.ci] apply automated fixes

---------

Co-authored-by: Janardan S Kavia <janardanskavia@Janardans-MacBook-Pro.local>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

src/backend/base/langflow/api/v2/files.py

@@ -158,8 +158,75 @@ async def upload_user_file(
 
     # Create a new database record for the uploaded file.
     try:
+        # SECURITY FIX: Validate and sanitize multipart upload filename to prevent path traversal attacks
+        # First, validate the original filename to reject obvious malicious attempts
+        if not file.filename:
+            raise HTTPException(status_code=400, detail="No filename provided")
+
+        # Reject filenames containing directory traversal sequences or path separators
+        # This prevents attackers from using directory traversal in the Content-Disposition header
+        # Check for: path separators (/, \), traversal (..), null bytes, and other dangerous chars
+        dangerous_chars = ["..", "/", "\\", "\x00", "\n", "\r"]
+        if any(char in file.filename for char in dangerous_chars):
+            raise HTTPException(
+                status_code=400,
+                detail=(
+                    "Invalid file name. Filename must not contain directory paths, "
+                    "'..' sequences, or control characters."
+                ),
+            )
+
+        # Additional check: reject filenames that are too long (prevent DoS)
+        # Most filesystems have a 255 byte limit for filenames
+        MAX_FILENAME_BYTES = 255  # noqa: N806
+        if len(file.filename.encode("utf-8")) > MAX_FILENAME_BYTES:
+            raise HTTPException(
+                status_code=400,
+                detail="File name is too long. Maximum 255 bytes allowed.",
+            )
+
+        # Extract only the basename as an additional safety measure
+        # This provides defense-in-depth in case the above checks are bypassed
+        new_filename = Path(file.filename).name
+
+        # Final validation: ensure the sanitized filename is valid and not empty
+        if not new_filename or new_filename in (".", ".."):
+            raise HTTPException(status_code=400, detail="Invalid file name after sanitization")
+
+        # Reject reserved filenames on Windows (CON, PRN, AUX, NUL, COM1-9, LPT1-9)
+        # This prevents issues when code runs on Windows systems
+        reserved_names = {
+            "CON",
+            "PRN",
+            "AUX",
+            "NUL",
+            "COM1",
+            "COM2",
+            "COM3",
+            "COM4",
+            "COM5",
+            "COM6",
+            "COM7",
+            "COM8",
+            "COM9",
+            "LPT1",
+            "LPT2",
+            "LPT3",
+            "LPT4",
+            "LPT5",
+            "LPT6",
+            "LPT7",
+            "LPT8",
+            "LPT9",
+        }
+        name_without_ext = new_filename.rsplit(".", 1)[0].upper()
+        if name_without_ext in reserved_names:
+            raise HTTPException(
+                status_code=400,
+                detail=f"Invalid file name. '{name_without_ext}' is a reserved system name.",
+            )
+
         # Enforce unique constraint on name, except for the special _mcp_servers file
-        new_filename = file.filename
         try:
             root_filename, file_extension = new_filename.rsplit(".", 1)
         except ValueError:

src/backend/base/langflow/services/storage/local.py

@@ -110,11 +110,63 @@ class LocalStorageService(StorageService):
             FileNotFoundError: If the specified flow does not exist.
             IsADirectoryError: If the file name is a directory.
             PermissionError: If there is no permission to write the file.
+            ValueError: If path traversal is detected (security violation).
         """
+        # SECURITY FIX: Defense-in-depth path containment check to prevent directory traversal
+        # Validate BEFORE creating any directories to prevent race conditions
+        # This ensures the resolved absolute path stays within the intended flow directory
+        # even if the API layer validation is bypassed or has bugs.
         folder_path = self.data_dir / flow_id
-        await folder_path.mkdir(parents=True, exist_ok=True)
         file_path = folder_path / file_name
 
+        try:
+            # Resolve paths to their absolute canonical forms
+            # Note: resolve() works even if the path doesn't exist yet
+            resolved_folder_path = await folder_path.resolve()
+            resolved_file_path = await file_path.resolve()
+
+            # Check if the resolved file path is actually within the flow directory
+            # Using is_relative_to() ensures no path traversal can escape the boundary
+            # Example attack: file_name="../../etc/passwd" would resolve outside folder_path
+            if not resolved_file_path.is_relative_to(resolved_folder_path):
+                # SECURITY: Don't log the actual paths to avoid information disclosure
+                await logger.aerror(
+                    f"Path traversal attempt detected for flow_id='{flow_id}'. "
+                    "File path would escape flow directory boundary."
+                )
+                msg = "Invalid file path: path traversal detected"
+                raise ValueError(msg)
+
+            # Additional check: ensure file_name doesn't contain path separators
+            # This catches cases where Path() might normalize away dangerous sequences
+            if "/" in file_name or "\\" in file_name or ".." in file_name:
+                await logger.aerror(
+                    f"Invalid file_name contains path separators or traversal sequences: flow_id='{flow_id}'"
+                )
+                msg = "Invalid file name: contains path separators"
+                raise ValueError(msg)
+
+        except ValueError:
+            # Re-raise ValueError (our security checks)
+            raise
+        except AttributeError:
+            # is_relative_to() not available (Python < 3.9), fall back to string comparison
+            # This should never happen in production but provides compatibility
+            resolved_folder_str = str(await folder_path.resolve())
+            resolved_file_str = str(await file_path.resolve())
+            if not resolved_file_str.startswith(resolved_folder_str):
+                await logger.aerror(f"Path traversal attempt detected for flow_id='{flow_id}' (fallback check)")
+                msg = "Invalid file path: path traversal detected"
+                raise ValueError(msg) from None
+        except Exception as e:
+            # Log unexpected errors during path resolution but don't expose details to user
+            await logger.aerror(f"Error validating file path for flow_id='{flow_id}': {type(e).__name__}")
+            msg = "Invalid file path"
+            raise ValueError(msg) from e
+
+        # Only create directory after validation passes
+        await folder_path.mkdir(parents=True, exist_ok=True)
+
         try:
             mode = "ab" if append else "wb"
             async with async_open(str(file_path), mode) as f:

src/backend/tests/unit/test_path_traversal_security.py

@@ -0,0 +1,248 @@
+"""Unit tests for path traversal security fixes.
+
+Tests the security fixes for CVE-2025-XXXXX (Arbitrary File Write vulnerability).
+Verifies both API-layer filename sanitization and storage-layer path containment checks.
+"""
+
+from pathlib import Path
+
+import pytest
+from httpx import AsyncClient
+
+
+async def test_upload_file_rejects_directory_traversal(client: AsyncClient, logged_in_headers):
+    """Test that directory traversal sequences in multipart filename are rejected."""
+    malicious_filename = "../../etc/passwd"
+    files = {"file": (malicious_filename, b"malicious content", "text/plain")}
+
+    response = await client.post(
+        "api/v2/files/",
+        files=files,
+        headers=logged_in_headers,
+    )
+
+    assert response.status_code == 400
+    assert "Invalid file name" in response.json()["detail"]
+
+
+async def test_upload_file_rejects_backslash_traversal(client: AsyncClient, logged_in_headers):
+    """Test that Windows-style directory traversal is rejected."""
+    malicious_filename = "..\\..\\windows\\system32\\config\\sam"
+    files = {"file": (malicious_filename, b"malicious content", "text/plain")}
+
+    response = await client.post(
+        "api/v2/files/",
+        files=files,
+        headers=logged_in_headers,
+    )
+
+    assert response.status_code == 400
+    assert "Invalid file name" in response.json()["detail"]
+
+
+async def test_upload_file_rejects_absolute_path(client: AsyncClient, logged_in_headers):
+    """Test that absolute paths in filename are rejected."""
+    malicious_filename = "/etc/passwd"
+    files = {"file": (malicious_filename, b"malicious content", "text/plain")}
+
+    response = await client.post(
+        "api/v2/files/",
+        files=files,
+        headers=logged_in_headers,
+    )
+
+    assert response.status_code == 400
+    assert "Invalid file name" in response.json()["detail"]
+
+
+async def test_upload_file_rejects_complex_traversal(client: AsyncClient, logged_in_headers):
+    """Test that complex traversal patterns are rejected."""
+    malicious_filenames = [
+        "....//....//etc/shadow",
+        "normal/../../../evil.py",
+        "file.txt/../../../root/.ssh/id_rsa",
+    ]
+
+    for filename in malicious_filenames:
+        files = {"file": (filename, b"malicious content", "text/plain")}
+        response = await client.post(
+            "api/v2/files/",
+            files=files,
+            headers=logged_in_headers,
+        )
+
+        assert response.status_code == 400, f"Failed to reject: {filename}"
+        assert "Invalid file name" in response.json()["detail"]
+
+
+async def test_upload_file_accepts_valid_filename(client: AsyncClient, logged_in_headers):
+    """Test that legitimate filenames are accepted."""
+    valid_filenames = [
+        "document.pdf",
+        "image.png",
+        "data_file.csv",
+        "my-file_v2.txt",
+    ]
+
+    for filename in valid_filenames:
+        files = {"file": (filename, b"legitimate content", "text/plain")}
+        response = await client.post(
+            "api/v2/files/",
+            files=files,
+            headers=logged_in_headers,
+        )
+
+        assert response.status_code == 201, f"Rejected valid filename: {filename}"
+        data = response.json()
+        assert data["name"] is not None
+
+
+async def test_upload_file_rejects_path_with_slashes(client: AsyncClient, logged_in_headers):
+    """Test that filenames containing slashes are rejected (security fix)."""
+    # Filenames with slashes should be rejected to prevent path traversal
+    files = {"file": ("subdir/nested/file.txt", b"content", "text/plain")}
+    response = await client.post(
+        "api/v2/files/",
+        files=files,
+        headers=logged_in_headers,
+    )
+
+    # Should be rejected because it contains slashes
+    assert response.status_code == 400
+    assert "Invalid file name" in response.json()["detail"]
+
+
+async def test_upload_file_rejects_empty_filename(client: AsyncClient, logged_in_headers):
+    """Test that empty filename is rejected."""
+    files = {"file": ("", b"content", "text/plain")}
+    response = await client.post(
+        "api/v2/files/",
+        files=files,
+        headers=logged_in_headers,
+    )
+
+    # Should be rejected with either 400 or 422 (both are valid error codes for invalid input)
+    assert response.status_code in (400, 422)
+
+
+async def test_upload_and_download_legitimate_file(client: AsyncClient, logged_in_headers):
+    """Test complete upload and download flow with legitimate file."""
+    filename = "test_document.pdf"
+    content = b"This is a legitimate PDF content"
+
+    # Upload
+    files = {"file": (filename, content, "application/pdf")}
+    upload_response = await client.post(
+        "api/v2/files/",
+        files=files,
+        headers=logged_in_headers,
+    )
+
+    assert upload_response.status_code == 201
+    data = upload_response.json()
+    assert "test_document" in data["name"]
+    assert data["size"] == len(content)
+
+    # Download
+    file_id = data["id"]
+    download_response = await client.get(
+        f"api/v2/files/{file_id}",
+        headers=logged_in_headers,
+    )
+
+    assert download_response.status_code == 200
+    assert download_response.content == content
+
+
+async def test_upload_file_no_path_escape_via_null_bytes(client: AsyncClient, logged_in_headers):
+    """Test that null bytes and other tricks don't bypass validation."""
+    # Null bytes and other special characters should not bypass validation
+    malicious_filenames = [
+        "file.txt\x00../../etc/passwd",
+        "../../etc/passwd\x00.txt",
+    ]
+
+    for filename in malicious_filenames:
+        files = {"file": (filename, b"malicious", "text/plain")}
+        response = await client.post(
+            "api/v2/files/",
+            files=files,
+            headers=logged_in_headers,
+        )
+
+        # Should be rejected (either 400 for invalid filename or sanitized to safe name)
+        # The important thing is it doesn't write to /etc/passwd
+        assert response.status_code in (400, 201)
+        if response.status_code == 201:
+            # If accepted, verify the filename was sanitized
+            data = response.json()
+            assert "etc" not in data["path"]
+            assert "passwd" not in data["path"] or "passwd" in data["name"]
+
+
+@pytest.mark.usefixtures("active_user")
+async def test_storage_layer_path_containment_check(client: AsyncClient, logged_in_headers):
+    """Test that storage layer has defense-in-depth path containment.
+
+    This test verifies that even if API layer validation is bypassed,
+    the storage layer will reject path traversal attempts.
+    """
+    # This test verifies the storage layer protection exists
+    # by attempting an upload that should be caught by API layer
+    # but would also be caught by storage layer if API layer failed
+
+    malicious_filename = "../../../../../../tmp/pwned.txt"
+    files = {"file": (malicious_filename, b"you got pwned", "text/plain")}
+
+    response = await client.post(
+        "api/v2/files/",
+        files=files,
+        headers=logged_in_headers,
+    )
+
+    # Should be rejected at API layer
+    assert response.status_code == 400
+
+    # Verify no file was created in /tmp (S108: this is intentional for security testing)
+    import tempfile
+
+    tmp_dir = tempfile.gettempdir()
+    pwned_path = Path(tmp_dir) / "pwned.txt"
+    if pwned_path.exists():
+        # If it exists, verify it wasn't created by us
+        content = pwned_path.read_bytes()
+        assert b"you got pwned" not in content
+
+
+async def test_upload_file_with_unicode_filename(client: AsyncClient, logged_in_headers):
+    """Test that Unicode filenames are handled correctly."""
+    unicode_filenames = [
+        "文档.pdf",
+        "документ.txt",
+        "αρχείο.csv",
+    ]
+
+    for filename in unicode_filenames:
+        files = {"file": (filename, b"content", "text/plain")}
+        response = await client.post(
+            "api/v2/files/",
+            files=files,
+            headers=logged_in_headers,
+        )
+
+        # Should accept valid Unicode filenames
+        assert response.status_code == 201, f"Rejected Unicode filename: {filename}"
+
+
+async def test_upload_file_preserves_extension(client: AsyncClient, logged_in_headers):
+    """Test that file extensions are preserved correctly."""
+    files = {"file": ("document.pdf", b"PDF content", "application/pdf")}
+    response = await client.post(
+        "api/v2/files/",
+        files=files,
+        headers=logged_in_headers,
+    )
+
+    assert response.status_code == 201
+    data = response.json()
+    assert data["path"].endswith(".pdf")