fix(security): prevent arbitrary file read via public flow build endpoint (GHSA-rcjh-r59h-gq37) (#13029)

* fix(security): prevent arbitrary file read via public flow build endpoint (GHSA-rcjh-r59h-gq37)

The unauthenticated `POST /api/v1/build_public_tmp/{flow_id}/flow`
endpoint accepted a `files` array whose paths were resolved by
`LocalStorageService` and read off disk before being attached to the
LLM prompt. A path like `/etc/hosts` was split into `flow_id="/etc"`
and `file_name="hosts"`. `data_dir / "/etc"` collapsed to `/etc`
because joining an absolute segment resets the prefix, and the
containment check was anchored at that attacker-controlled folder
so it passed. The file contents were then echoed back through the
chat reply.

Fix layers:

1. Boundary validation in `build_public_tmp`: each `files` entry
   must match `{source_flow_id}/{basename}` with no traversal,
   backslash, or null bytes. Foreign flow_ids are rejected with
   HTTP 400.
2. `LocalStorageService._validated_path` now also rejects
   `flow_id` values containing path separators, traversal, or
   null bytes, and anchors the containment check at the data
   directory rather than the per-flow folder.

Adds regression tests for both layers covering `/etc/hosts`,
traversal, foreign-namespace and null-byte payloads, plus an
acceptance test that legitimate `{flow_id}/{filename}` references
still work for AUTO_LOGIN playground uploads.

* add identifier validation

---------

Co-authored-by: cristhianzl <cristhian.lousa@gmail.com>

src/backend/base/langflow/api/v1/chat.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import re
 import time
 import traceback
 import uuid
@@ -637,6 +638,36 @@ async def build_flow_and_stream(flow_id, inputs, background_tasks, current_user)
     )
 
 
+# Public flow file paths must be `{source_flow_id}/{safe_basename}` — uploads
+# under that namespace are the only legitimate inputs for an unauthenticated
+# build. Anything else (absolute paths, traversal, foreign flow_ids) is a
+# probe at the arbitrary-file-read class of bug.
+_PUBLIC_FILE_PATH_RE = re.compile(
+    r"^([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})/([^/\\]+)$"
+)
+_PUBLIC_FILE_REJECTED_SUBSTRINGS = ("\x00", "..", "\\")
+
+
+def _validate_public_files(files: list[str] | None, source_flow_id: uuid.UUID) -> None:
+    """Reject file references that aren't `{source_flow_id}/{basename}`."""
+    if not files:
+        return
+    expected_flow_id = str(source_flow_id).lower()
+    for entry in files:
+        if not isinstance(entry, str) or not entry:
+            raise HTTPException(status_code=400, detail="Invalid file entry")
+        if any(token in entry for token in _PUBLIC_FILE_REJECTED_SUBSTRINGS):
+            raise HTTPException(status_code=400, detail="Invalid file path")
+        match = _PUBLIC_FILE_PATH_RE.match(entry)
+        if not match:
+            raise HTTPException(status_code=400, detail="Invalid file path format")
+        flow_id_segment, basename = match.group(1), match.group(2)
+        if flow_id_segment.lower() != expected_flow_id:
+            raise HTTPException(status_code=400, detail="File not in this flow's namespace")
+        if basename in (".", ".."):
+            raise HTTPException(status_code=400, detail="Invalid filename")
+
+
 @router.post("/build_public_tmp/{flow_id}/flow")
 async def build_public_tmp(
     *,
@@ -694,6 +725,11 @@ async def build_public_tmp(
         Dict with job_id that can be used to poll for build status
     """
     try:
+        # Reject caller-supplied file references that aren't scoped to this
+        # public flow's own storage namespace. Done before any flow lookup so
+        # malformed requests fail fast and don't touch the DB.
+        _validate_public_files(files, flow_id)
+
         # Verify this is a public flow and get the associated user
         client_id = request.cookies.get("client_id")
         # Only use authenticated user_id when auto-login is disabled.

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

@@ -45,6 +45,9 @@ class LocalStorageService(StorageService):
 
         Centralizes the path-containment check used across every file operation so
         a single bug cannot re-introduce the traversal issue on a subset of methods.
+        Both flow_id and file_name must be free of separators / traversal so a
+        caller-supplied flow_id like "/etc" cannot collapse data_dir via Path
+        joining (absolute segment resets the join).
         """
         if not isinstance(file_name, str) or "/" in file_name or "\\" in file_name or ".." in file_name:
             await logger.aerror(
@@ -53,25 +56,40 @@ class LocalStorageService(StorageService):
             msg = "Invalid file name: contains path separators"
             raise ValueError(msg)
 
+        if (
+            not isinstance(flow_id, str)
+            or not flow_id
+            or "/" in flow_id
+            or "\\" in flow_id
+            or ".." in flow_id
+            or "\x00" in flow_id
+        ):
+            await logger.aerror("Invalid flow_id contains path separators or traversal sequences")
+            msg = "Invalid flow_id: contains path separators"
+            raise ValueError(msg)
+
         folder_path = self.data_dir / flow_id
         file_path = folder_path / file_name
 
         try:
-            resolved_folder = await folder_path.resolve()
+            resolved_data_dir = await self.data_dir.resolve()
             resolved_file = await file_path.resolve()
-            if not resolved_file.is_relative_to(resolved_folder):
+            # Containment check is anchored at data_dir, not folder_path: an
+            # attacker-controlled flow_id segment must not be allowed to define
+            # the boundary it is then compared against.
+            if not resolved_file.is_relative_to(resolved_data_dir):
                 await logger.aerror(
                     f"Path traversal attempt detected for flow_id='{flow_id}'. "
-                    "File path would escape flow directory boundary."
+                    "File path would escape data directory boundary."
                 )
                 msg = "Invalid file path: path traversal detected"
                 raise ValueError(msg)
         except ValueError:
             raise
         except AttributeError:
-            resolved_folder_str = str(await folder_path.resolve())
+            resolved_data_dir_str = str(await self.data_dir.resolve())
             resolved_file_str = str(await file_path.resolve())
-            if not resolved_file_str.startswith(resolved_folder_str):
+            if not resolved_file_str.startswith(resolved_data_dir_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

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

@@ -65,6 +65,39 @@ class S3StorageService(StorageService):
             f"region={os.getenv('AWS_DEFAULT_REGION', 'default')}"
         )
 
+    def _validate_identifiers(self, flow_id: str, file_name: str | None = None) -> None:
+        """Reject flow_id / file_name values that could escape the flow namespace.
+
+        Defense in depth at the S3 backend for GHSA-rcjh-r59h-gq37: the public-flow
+        boundary in chat.py is the primary gate, but any caller reaching this
+        backend with untrusted identifiers must still fail safely. Validation is
+        synchronous and runs before any AWS call so a malformed input cannot be
+        turned into a get_object on an arbitrary bucket key.
+        """
+        if (
+            not isinstance(flow_id, str)
+            or not flow_id
+            or "/" in flow_id
+            or "\\" in flow_id
+            or ".." in flow_id
+            or "\x00" in flow_id
+        ):
+            logger.error("Invalid flow_id contains path separators or traversal sequences")
+            msg = "Invalid flow_id: contains path separators"
+            raise ValueError(msg)
+
+        if file_name is not None and (
+            not isinstance(file_name, str)
+            or not file_name
+            or "/" in file_name
+            or "\\" in file_name
+            or ".." in file_name
+            or "\x00" in file_name
+        ):
+            logger.error("Invalid file_name contains path separators or traversal sequences")
+            msg = "Invalid file name: contains path separators"
+            raise ValueError(msg)
+
     def build_full_path(self, flow_id: str, file_name: str) -> str:
         """Build the full S3 key for a file.
 
@@ -143,6 +176,7 @@ class S3StorageService(StorageService):
             msg = "Append mode is not supported for S3 storage"
             raise NotImplementedError(msg)
 
+        self._validate_identifiers(flow_id, file_name)
         key = self.build_full_path(flow_id, file_name)
 
         try:
@@ -197,6 +231,7 @@ class S3StorageService(StorageService):
         Raises:
             FileNotFoundError: If the file does not exist in S3
         """
+        self._validate_identifiers(flow_id, file_name)
         key = self.build_full_path(flow_id, file_name)
 
         try:
@@ -230,6 +265,7 @@ class S3StorageService(StorageService):
         Raises:
             FileNotFoundError: If the file does not exist in S3
         """
+        self._validate_identifiers(flow_id, file_name)
         key = self.build_full_path(flow_id, file_name)
 
         try:
@@ -266,8 +302,7 @@ class S3StorageService(StorageService):
         Raises:
             Exception: If there's an error listing files from S3
         """
-        if not isinstance(flow_id, str):
-            flow_id = str(flow_id)
+        self._validate_identifiers(flow_id)
 
         prefix = self.build_full_path(flow_id, "")
 
@@ -302,6 +337,7 @@ class S3StorageService(StorageService):
         Note:
             S3 delete_object doesn't raise an error if the object doesn't exist
         """
+        self._validate_identifiers(flow_id, file_name)
         key = self.build_full_path(flow_id, file_name)
 
         try:
@@ -325,6 +361,7 @@ class S3StorageService(StorageService):
         Raises:
             FileNotFoundError: If the file does not exist in S3
         """
+        self._validate_identifiers(flow_id, file_name)
         key = self.build_full_path(flow_id, file_name)
 
         try:

src/backend/tests/unit/services/storage/test_local_storage_service.py

@@ -200,18 +200,22 @@ class TestLocalStorageServiceListOperations:
         flow_id = "dir_test"
         file_name = "test.txt"
 
-        # Create a file
+        # Create a file via the service
         await local_storage_service.save_file(flow_id, file_name, b"content")
 
-        # Create a subdirectory (by creating a file in a subdirectory)
-        await local_storage_service.save_file(f"{flow_id}/subdir", "nested.txt", b"content")
+        # Create a subdirectory directly on disk so list_files has something to skip.
+        # Cannot route this through save_file: passing a slashed flow_id is the same
+        # shape as the public-build arbitrary-file-read regression, so the storage
+        # layer (correctly) rejects it.
+        subdir_path = local_storage_service.data_dir / flow_id / "subdir"
+        await subdir_path.mkdir(parents=True, exist_ok=True)
 
         # List files - should only return files in the flow_id directory, not subdirectories
         listed_files = await local_storage_service.list_files(flow_id)
 
-        # Should only return the file in the root, not the nested one
+        # Should only return the file in the root, not the subdirectory
         assert file_name in listed_files
-        assert "nested.txt" not in listed_files  # Nested file is in subdirectory
+        assert "subdir" not in listed_files
 
 
 @pytest.mark.asyncio
@@ -375,6 +379,31 @@ class TestLocalStorageServicePathTraversal:
         with pytest.raises(ValueError, match="Invalid file"):
             await local_storage_service.get_file("flow_a", "../flow_b/secret.txt")
 
+    @pytest.mark.parametrize(
+        "malicious_flow_id",
+        [
+            "/etc",
+            "..",
+            "../other",
+            "..\\other",
+            "flow/sub",
+            "flow\\sub",
+            "with\x00null",
+            "",
+        ],
+    )
+    async def test_get_file_rejects_malicious_flow_id(self, local_storage_service, malicious_flow_id):
+        # Pre-vuln: flow_id="/etc" + file_name="passwd" collapsed Path join to /etc/passwd
+        # and the containment check passed because it was anchored at folder_path.
+        with pytest.raises(ValueError, match="Invalid"):
+            await local_storage_service.get_file(malicious_flow_id, "passwd")
+
+    async def test_get_file_rejects_absolute_flow_id_collapse(self, local_storage_service):
+        # Direct regression for the public-build arbitrary-file-read: ensure the
+        # /etc/<file> shape is rejected even when the file would exist on disk.
+        with pytest.raises(ValueError, match="Invalid"):
+            await local_storage_service.get_file("/etc", "hosts")
+
 
 @pytest.mark.asyncio
 class TestLocalStorageServiceEdgeCases:

src/backend/tests/unit/services/storage/test_s3_storage_service.py

@@ -0,0 +1,124 @@
+"""Unit tests for S3StorageService input validation.
+
+These tests run offline. They assert that malformed flow_id / file_name values
+are rejected at the storage layer BEFORE any AWS call is attempted, so the
+boundary fix in chat.py is not the only line of defense for callers that reach
+the S3 backend with untrusted identifiers.
+
+Regression for GHSA-rcjh-r59h-gq37 (defense in depth at the S3 backend).
+"""
+
+from unittest.mock import Mock
+
+import pytest
+from langflow.services.storage.s3 import S3StorageService
+
+
+@pytest.fixture
+def mock_settings_service(tmp_path):
+    """Settings configured for S3 with a stable bucket / prefix."""
+    settings_service = Mock()
+    settings_service.settings.config_dir = str(tmp_path)
+    settings_service.settings.object_storage_bucket_name = "langflow-unit-test-bucket"
+    settings_service.settings.object_storage_prefix = "test-prefix"
+    settings_service.settings.object_storage_tags = {}
+    return settings_service
+
+
+@pytest.fixture
+def mock_session_service():
+    return Mock()
+
+
+@pytest.fixture
+def s3_service_offline(mock_session_service, mock_settings_service, monkeypatch):
+    """S3StorageService that fails loudly if any AWS call is attempted.
+
+    Validation MUST short-circuit before _get_client is invoked. If a test
+    reaches this assertion, the validation guard is missing or bypassed.
+    """
+    service = S3StorageService(mock_session_service, mock_settings_service)
+
+    def _no_aws_calls():
+        msg = "validation should have rejected this input before reaching S3"
+        raise AssertionError(msg)
+
+    monkeypatch.setattr(service, "_get_client", _no_aws_calls)
+    return service
+
+
+_MALICIOUS_FLOW_IDS = [
+    "/etc",
+    "..",
+    "../other",
+    "..\\other",
+    "flow/sub",
+    "flow\\sub",
+    "with\x00null",
+    "",
+]
+
+_MALICIOUS_FILE_NAMES = [
+    "../passwd",
+    "..\\passwd",
+    "sub/passwd",
+    "sub\\passwd",
+    "with\x00null",
+    "",
+]
+
+
+@pytest.mark.asyncio
+class TestS3StorageServicePathValidation:
+    """GHSA-rcjh-r59h-gq37: S3 backend must reject untrusted identifiers locally."""
+
+    @pytest.mark.parametrize("malicious_flow_id", _MALICIOUS_FLOW_IDS)
+    async def test_get_file_rejects_malicious_flow_id(self, s3_service_offline, malicious_flow_id):
+        with pytest.raises(ValueError, match="Invalid"):
+            await s3_service_offline.get_file(malicious_flow_id, "passwd")
+
+    @pytest.mark.parametrize("malicious_flow_id", _MALICIOUS_FLOW_IDS)
+    async def test_save_file_rejects_malicious_flow_id(self, s3_service_offline, malicious_flow_id):
+        with pytest.raises(ValueError, match="Invalid"):
+            await s3_service_offline.save_file(malicious_flow_id, "passwd", b"x")
+
+    @pytest.mark.parametrize("malicious_flow_id", _MALICIOUS_FLOW_IDS)
+    async def test_delete_file_rejects_malicious_flow_id(self, s3_service_offline, malicious_flow_id):
+        with pytest.raises(ValueError, match="Invalid"):
+            await s3_service_offline.delete_file(malicious_flow_id, "passwd")
+
+    @pytest.mark.parametrize("malicious_flow_id", _MALICIOUS_FLOW_IDS)
+    async def test_get_file_size_rejects_malicious_flow_id(self, s3_service_offline, malicious_flow_id):
+        with pytest.raises(ValueError, match="Invalid"):
+            await s3_service_offline.get_file_size(malicious_flow_id, "passwd")
+
+    @pytest.mark.parametrize("malicious_flow_id", _MALICIOUS_FLOW_IDS)
+    async def test_get_file_stream_rejects_malicious_flow_id(self, s3_service_offline, malicious_flow_id):
+        with pytest.raises(ValueError, match="Invalid"):
+            # AsyncIterator functions don't raise until first iteration.
+            async for _ in s3_service_offline.get_file_stream(malicious_flow_id, "passwd"):
+                pass
+
+    @pytest.mark.parametrize("malicious_flow_id", _MALICIOUS_FLOW_IDS)
+    async def test_list_files_rejects_malicious_flow_id(self, s3_service_offline, malicious_flow_id):
+        with pytest.raises(ValueError, match="Invalid"):
+            await s3_service_offline.list_files(malicious_flow_id)
+
+    @pytest.mark.parametrize("malicious_file_name", _MALICIOUS_FILE_NAMES)
+    async def test_get_file_rejects_malicious_file_name(self, s3_service_offline, malicious_file_name):
+        with pytest.raises(ValueError, match="Invalid"):
+            await s3_service_offline.get_file("legit_flow", malicious_file_name)
+
+    @pytest.mark.parametrize("malicious_file_name", _MALICIOUS_FILE_NAMES)
+    async def test_save_file_rejects_malicious_file_name(self, s3_service_offline, malicious_file_name):
+        with pytest.raises(ValueError, match="Invalid"):
+            await s3_service_offline.save_file("legit_flow", malicious_file_name, b"x")
+
+    async def test_get_file_rejects_absolute_flow_id_collapse(self, s3_service_offline):
+        """Direct regression for the public-build arbitrary-file-read at the S3 layer.
+
+        Pre-vuln: ``build_full_path("/etc", "hosts")`` produced a key that resolved
+        to an attacker-controlled S3 path. Validation must reject the shape.
+        """
+        with pytest.raises(ValueError, match="Invalid"):
+            await s3_service_offline.get_file("/etc", "hosts")

src/backend/tests/unit/test_chat_endpoint.py

@@ -562,6 +562,73 @@ async def test_build_public_tmp_ignores_data_parameter(client, json_memory_chatb
     assert "job_id" in response_data
 
 
+@pytest.mark.benchmark
+@pytest.mark.security
+@pytest.mark.parametrize(
+    "malicious_files",
+    [
+        ["/etc/hosts"],
+        ["/etc/passwd"],
+        ["../../etc/passwd"],
+        ["..\\..\\windows\\system32\\drivers\\etc\\hosts"],
+        ["s3://other-bucket/secret.txt"],
+        ["just_a_filename.txt"],
+        # foreign flow_id segment — looks well-formed but isn't this flow's namespace
+        ["00000000-0000-0000-0000-000000000000/file.png"],
+        # null byte smuggling
+        ["abc\x00/file.png"],
+    ],
+)
+async def test_build_public_tmp_rejects_malicious_files(
+    client, json_memory_chatbot_no_llm, logged_in_headers, malicious_files
+):
+    """Regression for GHSA-rcjh-r59h-gq37 — unauth public build must not accept arbitrary file paths."""
+    flow_id = await create_flow(client, json_memory_chatbot_no_llm, logged_in_headers)
+    response = await client.patch(
+        f"api/v1/flows/{flow_id}",
+        json={"access_type": "PUBLIC"},
+        headers=logged_in_headers,
+    )
+    assert response.status_code == codes.OK
+
+    client.cookies.set("client_id", "test-files-validation-client")
+    response = await client.post(
+        f"api/v1/build_public_tmp/{flow_id}/flow",
+        json={
+            "inputs": {"session": "test_session"},
+            "files": malicious_files,
+        },
+        headers={"Content-Type": "application/json"},
+    )
+
+    assert response.status_code == codes.BAD_REQUEST
+    assert "file" in response.json()["detail"].lower()
+
+
+@pytest.mark.benchmark
+async def test_build_public_tmp_accepts_files_in_own_namespace(client, json_memory_chatbot_no_llm, logged_in_headers):
+    """Files namespaced under the public flow's own UUID must still be accepted."""
+    flow_id = await create_flow(client, json_memory_chatbot_no_llm, logged_in_headers)
+    response = await client.patch(
+        f"api/v1/flows/{flow_id}",
+        json={"access_type": "PUBLIC"},
+        headers=logged_in_headers,
+    )
+    assert response.status_code == codes.OK
+
+    client.cookies.set("client_id", "test-files-allowed-client")
+    response = await client.post(
+        f"api/v1/build_public_tmp/{flow_id}/flow",
+        json={
+            "inputs": {"session": "test_session"},
+            "files": [f"{flow_id}/example_attachment.png"],
+        },
+        headers={"Content-Type": "application/json"},
+    )
+
+    assert response.status_code == codes.OK
+
+
 @pytest.mark.benchmark
 async def test_build_public_tmp_checks_public_access_before_validation(
     client, json_memory_chatbot_no_llm, logged_in_headers, monkeypatch

src/lfx/src/lfx/services/storage/local.py

@@ -43,6 +43,9 @@ class LocalStorageService(StorageService, Service):
 
         Centralizes the path-containment check used across every file operation so
         a single bug cannot re-introduce the traversal issue on a subset of methods.
+        Both flow_id and file_name must be free of separators / traversal so a
+        caller-supplied flow_id like "/etc" cannot collapse data_dir via Path
+        joining (absolute segment resets the join).
         """
         if not isinstance(file_name, str) or "/" in file_name or "\\" in file_name or ".." in file_name:
             await logger.aerror(
@@ -51,25 +54,40 @@ class LocalStorageService(StorageService, Service):
             msg = "Invalid file name: contains path separators"
             raise ValueError(msg)
 
+        if (
+            not isinstance(flow_id, str)
+            or not flow_id
+            or "/" in flow_id
+            or "\\" in flow_id
+            or ".." in flow_id
+            or "\x00" in flow_id
+        ):
+            await logger.aerror("Invalid flow_id contains path separators or traversal sequences")
+            msg = "Invalid flow_id: contains path separators"
+            raise ValueError(msg)
+
         folder_path = self.data_dir / flow_id
         file_path = folder_path / file_name
 
         try:
-            resolved_folder = await folder_path.resolve()
+            resolved_data_dir = await self.data_dir.resolve()
             resolved_file = await file_path.resolve()
-            if not resolved_file.is_relative_to(resolved_folder):
+            # Containment check is anchored at data_dir, not folder_path: an
+            # attacker-controlled flow_id segment must not be allowed to define
+            # the boundary it is then compared against.
+            if not resolved_file.is_relative_to(resolved_data_dir):
                 await logger.aerror(
                     f"Path traversal attempt detected for flow_id='{flow_id}'. "
-                    "File path would escape flow directory boundary."
+                    "File path would escape data directory boundary."
                 )
                 msg = "Invalid file path: path traversal detected"
                 raise ValueError(msg)
         except ValueError:
             raise
         except AttributeError:
-            resolved_folder_str = str(await folder_path.resolve())
+            resolved_data_dir_str = str(await self.data_dir.resolve())
             resolved_file_str = str(await file_path.resolve())
-            if not resolved_file_str.startswith(resolved_folder_str):
+            if not resolved_file_str.startswith(resolved_data_dir_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