fix: prevent path traversal in knowledge base create endpoint (#12337)

* fix: prevent path traversal in knowledge base create endpoint

Add _validate_kb_path_containment() helper that applies Path.resolve()
and is_relative_to() before any directory creation in create_knowledge_base.
Replaces duplicated containment-check logic in _resolve_kb_path() with a
single shared helper to prevent security-critical divergence.

* test: add path traversal tests for knowledge base create endpoint

Add four tests covering the create endpoint attack surface:
- single-level traversal (../victim_user/evil_kb)
- absolute path injection (/tmp/evil)
- prefix-ambiguity bypass (../activeuser_evil/secret_kb)
- logger.warning observability on traversal attempt

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

Author: AntonioABLima

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

@@ -35,6 +35,24 @@
 router = APIRouter(tags=["Knowledge Bases"], prefix="/knowledge_bases", include_in_schema=False)
 
 
+def _validate_kb_path_containment(kb_user_path: Path, kb_path: Path, kb_name: str, username: str) -> None:
+    """Raise 403 if kb_path is not contained within kb_user_path.
+
+    Uses is_relative_to() instead of startswith() to prevent path traversal attacks.
+    startswith() has a prefix-ambiguity bug: a user named "alice" would incorrectly allow
+    paths under "alice_evil/" because the string starts with "alice". is_relative_to() performs
+    proper path containment checking.
+    """
+    if not kb_path.is_relative_to(kb_user_path):
+        logger.warning(
+            "Path traversal attempt blocked: user=%s kb_name=%r resolved_path=%s",
+            username,
+            kb_name,
+            kb_path,
+        )
+        raise HTTPException(status_code=403, detail=f"Access denied for knowledge base '{kb_name}'.")
+
+
 def _resolve_kb_path(kb_name: str, current_user: CurrentActiveUser) -> Path:
     """Resolve and validate KB path.
 
@@ -49,18 +67,7 @@ def _resolve_kb_path(kb_name: str, current_user: CurrentActiveUser) -> Path:
     kb_user_path = (kb_root_path / kb_user).resolve()
     kb_path = (kb_user_path / kb_name).resolve()
 
-    # Security: use is_relative_to() instead of startswith() to prevent path traversal attacks.
-    # startswith() has a prefix-ambiguity bug: a user named "alice" would incorrectly allow
-    # paths under "alice_evil/" because the string starts with "alice". is_relative_to() performs
-    # proper path containment checking.
-    if not kb_path.is_relative_to(kb_user_path):
-        logger.warning(
-            "Path traversal attempt blocked: user=%s kb_name=%r resolved_path=%s",
-            kb_user,
-            kb_name,
-            kb_path,
-        )
-        raise HTTPException(status_code=403, detail=f"Access denied for knowledge base '{kb_name}'.")
+    _validate_kb_path_containment(kb_user_path, kb_path, kb_name, kb_user)
 
     if not kb_path.exists() or not kb_path.is_dir():
         raise HTTPException(status_code=404, detail=f"Knowledge base '{kb_name}' not found")
@@ -78,11 +85,17 @@ async def create_knowledge_base(
         kb_root_path = KBStorageHelper.get_root_path()
         kb_user = current_user.username
         kb_name = request.name.strip().replace(" ", "_")
-        kb_path = kb_root_path / kb_user / kb_name
         # Validate KB name
         if not kb_name or len(kb_name) < MIN_KB_NAME_LENGTH:
             raise HTTPException(status_code=400, detail="Knowledge base name must be at least 3 characters")
 
+        # Security: resolve paths and validate containment to prevent path traversal attacks.
+        # A crafted kb_name like "../victim/evil" or an absolute path like "/tmp/evil" must be
+        # rejected before any directory is created.
+        kb_user_path = (kb_root_path / kb_user).resolve()
+        kb_path = (kb_user_path / kb_name).resolve()
+        _validate_kb_path_containment(kb_user_path, kb_path, kb_name, kb_user)
+
         # Check if KB already exists
         if kb_path.exists():
             raise HTTPException(status_code=409, detail=f"Knowledge base '{kb_name}' already exists")

src/backend/tests/unit/test_knowledge_bases_api.py

@@ -170,6 +170,128 @@ async def test_create_knowledge_base(
         data = response.json()
         assert data["name"] == "New KB"
 
+    @patch("langflow.api.v1.knowledge_bases.KBStorageHelper.get_root_path")
+    async def test_create_kb_path_traversal_single_level(
+        self, mock_root, client: AsyncClient, logged_in_headers, tmp_path
+    ):
+        """Single-level traversal '../victim_user/evil_kb' in POST must be blocked with 400/403.
+
+        VULNERABILITY: the create endpoint builds kb_path = kb_root_path / kb_user / kb_name
+        without resolve() or is_relative_to(), so '../victim_user/evil_kb' escapes the user dir.
+        """
+        mock_root.return_value = tmp_path
+        (tmp_path / "activeuser").mkdir(parents=True)
+        victim_dir = tmp_path / "victim_user" / "evil_kb"
+
+        response = await client.post(
+            "api/v1/knowledge_bases",
+            headers=logged_in_headers,
+            json={
+                "name": "../victim_user/evil_kb",
+                "embedding_provider": "OpenAI",
+                "embedding_model": "text-embedding-3-small",
+            },
+        )
+
+        assert response.status_code in (400, 403), (
+            f"VULNERABILITY CONFIRMED: create endpoint accepted traversal payload with status {response.status_code}"
+        )
+        assert not victim_dir.exists(), (
+            "VULNERABILITY CONFIRMED: path traversal created a directory outside the user's KB root"
+        )
+
+    @patch("langflow.api.v1.knowledge_bases.KBStorageHelper.get_root_path")
+    async def test_create_kb_path_traversal_absolute_path(
+        self, mock_root, client: AsyncClient, logged_in_headers, tmp_path
+    ):
+        """Absolute path in kb_name must be blocked — e.g. '/tmp/evil'.
+
+        VULNERABILITY: kb_root_path / kb_user / '/tmp/evil' resolves to '/tmp/evil' in Python
+        because Path drops all previous components when a segment starts with '/'.
+        """
+        mock_root.return_value = tmp_path
+        (tmp_path / "activeuser").mkdir(parents=True)
+        evil_dir = tmp_path / "evil_absolute"
+
+        response = await client.post(
+            "api/v1/knowledge_bases",
+            headers=logged_in_headers,
+            json={
+                "name": str(evil_dir),
+                "embedding_provider": "OpenAI",
+                "embedding_model": "text-embedding-3-small",
+            },
+        )
+
+        assert response.status_code in (400, 403), (
+            f"VULNERABILITY CONFIRMED: create endpoint accepted absolute path payload "
+            f"with status {response.status_code}"
+        )
+        assert not evil_dir.exists(), (
+            "VULNERABILITY CONFIRMED: absolute path in kb_name created a directory outside the KB root"
+        )
+
+    @patch("langflow.api.v1.knowledge_bases.KBStorageHelper.get_root_path")
+    async def test_create_kb_path_traversal_prefix_ambiguity(
+        self, mock_root, client: AsyncClient, logged_in_headers, tmp_path
+    ):
+        """Prefix-ambiguity attack on create: user='activeuser', target dir='activeuser_evil'.
+
+        With startswith('/root/activeuser'), the path '/root/activeuser_evil/secret_kb'
+        incorrectly passes because the string starts with '/root/activeuser'.
+        is_relative_to() closes this gap and must block the request with 400/403.
+        """
+        mock_root.return_value = tmp_path
+
+        (tmp_path / "activeuser").mkdir(parents=True)
+        victim_kb = tmp_path / "activeuser_evil" / "secret_kb"
+        victim_kb.mkdir(parents=True)
+
+        response = await client.post(
+            "api/v1/knowledge_bases",
+            headers=logged_in_headers,
+            json={
+                "name": "../activeuser_evil/secret_kb",
+                "embedding_provider": "OpenAI",
+                "embedding_model": "text-embedding-3-small",
+            },
+        )
+
+        assert response.status_code in (400, 403), (
+            "VULNERABILITY CONFIRMED: prefix-ambiguity bypass succeeded on create endpoint — "
+            "startswith() may still be in use instead of is_relative_to()"
+        )
+        assert not (tmp_path / "activeuser_evil" / "secret_kb_new").exists(), (
+            "VULNERABILITY CONFIRMED: prefix-ambiguity attack created a directory outside the user's KB root"
+        )
+
+    @patch("langflow.api.v1.knowledge_bases.logger.warning")
+    @patch("langflow.api.v1.knowledge_bases.KBStorageHelper.get_root_path")
+    async def test_create_kb_path_traversal_logs_warning(
+        self, mock_root, mock_warning, client: AsyncClient, logged_in_headers, tmp_path
+    ):
+        """A traversal attempt on create must emit a warning log with user= and kb_name= context."""
+        mock_root.return_value = tmp_path
+
+        (tmp_path / "activeuser").mkdir(parents=True)
+        (tmp_path / "victim_user" / "secret_kb").mkdir(parents=True)
+
+        await client.post(
+            "api/v1/knowledge_bases",
+            headers=logged_in_headers,
+            json={
+                "name": "../victim_user/secret_kb",
+                "embedding_provider": "OpenAI",
+                "embedding_model": "text-embedding-3-small",
+            },
+        )
+
+        mock_warning.assert_called_once()
+        warning_args = mock_warning.call_args[0]
+        all_args_str = str(warning_args)
+        assert "user=" in all_args_str, "Warning log must contain 'user=' in the format string"
+        assert "kb_name=" in all_args_str, "Warning log must contain 'kb_name=' in the format string"
+
     async def test_create_kb_name_too_short(self, client: AsyncClient, logged_in_headers):
         response = await client.post(
             "api/v1/knowledge_bases",