fix: Add defense in depth user_id filter to handle list tools MCP handler (#13373)

* feat(mcp): add defense-in-depth user_id filter to handle_list_tools for consistency with other MCP handlers

* [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/v1/mcp_utils.py

@@ -382,7 +382,7 @@ async def handle_call_tool(
         raise
 
 
-async def handle_list_tools(project_id=None, *, mcp_enabled_only=False):
+async def handle_list_tools(project_id: UUID | None = None, *, mcp_enabled_only: bool = False):
     """Handle listing tools for MCP.
 
     Args:
@@ -401,8 +401,20 @@ async def handle_list_tools(project_id=None, *, mcp_enabled_only=False):
         async with session_scope() as session:
             # Build query based on parameters
             if project_id:
-                # Filter flows by project and optionally by MCP enabled status
-                flows_query = select(Flow).where(Flow.folder_id == project_id, Flow.is_component == False)  # noqa: E712
+                # SECURITY (defense-in-depth): Filter by both folder_id AND user_id.
+                # While verify_project_auth_conditional already ensures the user owns the project,
+                # this query-level filter provides an additional safety layer and maintains
+                # consistency with handle_list_resources() and handle_read_resource().
+                if current_user is None:
+                    await logger.awarning(
+                        "handle_list_tools called with project_id but no current user; returning empty list"
+                    )
+                    return tools
+                flows_query = select(Flow).where(
+                    Flow.folder_id == project_id,
+                    Flow.user_id == current_user.id,
+                    Flow.is_component == False,  # noqa: E712
+                )
                 if mcp_enabled_only:
                     flows_query = flows_query.where(Flow.mcp_enabled == True)  # noqa: E712
             elif current_user is not None:

src/backend/tests/unit/api/v1/test_mcp_projects.py

@@ -1550,4 +1550,114 @@ async def test_should_report_available_true_when_config_file_has_corrupt_json(
 
     for entry in results:
         assert entry["available"] is True, f"{entry['name']} should be available (directory exists)"
-        assert entry["installed"] is False, f"{entry['name']} should not be installed (JSON is corrupt)"
+
+
+async def test_handle_list_tools_filters_by_user_id_for_defense_in_depth(
+    user_test_project,
+    other_test_project,
+    other_test_user,
+    active_user,
+):
+    """Test that handle_list_tools filters by BOTH folder_id AND user_id for defense-in-depth.
+
+    This test verifies the query-level ownership validation in handle_list_tools() when
+    project_id is provided. While the endpoint-level authentication (verify_project_auth_conditional)
+    already ensures the user owns the project, the database query should ALSO filter by user_id
+    for consistency with other MCP handlers (handle_list_resources, handle_read_resource).
+
+    GIVEN: Two projects owned by different users, each with flows
+    WHEN:  handle_list_tools is called with a project_id and a specific user context
+    THEN:  Only flows from that project AND owned by that user are returned
+    """
+    from langflow.api.v1.mcp_utils import current_user_ctx, handle_list_tools
+
+    # Create flows for both users in their respective projects
+    user_flow_id = uuid4()
+    other_user_flow_id = uuid4()
+
+    async with session_scope() as session:
+        # Create flow for active_user in user_test_project
+        # Include minimal valid data structure to pass json_schema_from_flow validation
+        user_flow = Flow(
+            id=user_flow_id,
+            name="User Flow",
+            description="Flow owned by active user",
+            data={"nodes": [], "edges": []},  # Minimal valid flow structure
+            mcp_enabled=True,
+            action_name="user_action",
+            action_description="User action",
+            folder_id=user_test_project.id,
+            user_id=active_user.id,
+            is_component=False,
+        )
+        session.add(user_flow)
+
+        # Create flow for other_test_user in other_test_project
+        other_user_flow = Flow(
+            id=other_user_flow_id,
+            name="Other User Flow",
+            description="Flow owned by other user",
+            data={"nodes": [], "edges": []},  # Minimal valid flow structure
+            mcp_enabled=True,
+            action_name="other_action",
+            action_description="Other action",
+            folder_id=other_test_project.id,
+            user_id=other_test_user.id,
+            is_component=False,
+        )
+        session.add(other_user_flow)
+        await session.commit()
+
+    try:
+        # Test 1: Active user queries their own project - should see their flow
+        token = current_user_ctx.set(active_user)
+        try:
+            tools = await handle_list_tools(project_id=user_test_project.id, mcp_enabled_only=True)
+            assert len(tools) == 1, "Active user should see exactly 1 tool in their project"
+            assert tools[0].name == "user_action", "Should see the user's own flow"
+        finally:
+            current_user_ctx.reset(token)
+
+        # Test 2: Other user queries their own project - should see their flow
+        token = current_user_ctx.set(other_test_user)
+        try:
+            tools = await handle_list_tools(project_id=other_test_project.id, mcp_enabled_only=True)
+            assert len(tools) == 1, "Other user should see exactly 1 tool in their project"
+            assert tools[0].name == "other_action", "Should see the other user's flow"
+        finally:
+            current_user_ctx.reset(token)
+
+        # Test 3: Defense-in-depth verification - Active user context with other user's project_id
+        # This simulates a hypothetical scenario where endpoint auth is bypassed (shouldn't happen,
+        # but the query-level filter should still protect against it)
+        token = current_user_ctx.set(active_user)
+        try:
+            tools = await handle_list_tools(project_id=other_test_project.id, mcp_enabled_only=True)
+            assert len(tools) == 0, (
+                "Active user should see NO tools when querying other user's project_id. "
+                "The query-level user_id filter provides defense-in-depth protection."
+            )
+        finally:
+            current_user_ctx.reset(token)
+
+        # Test 4: Verify the same defense-in-depth for the other user
+        token = current_user_ctx.set(other_test_user)
+        try:
+            tools = await handle_list_tools(project_id=user_test_project.id, mcp_enabled_only=True)
+            assert len(tools) == 0, (
+                "Other user should see NO tools when querying active user's project_id. "
+                "The query-level user_id filter provides defense-in-depth protection."
+            )
+        finally:
+            current_user_ctx.reset(token)
+
+    finally:
+        # Cleanup
+        async with session_scope() as session:
+            user_flow = await session.get(Flow, user_flow_id)
+            if user_flow:
+                await session.delete(user_flow)
+            other_user_flow = await session.get(Flow, other_user_flow_id)
+            if other_user_flow:
+                await session.delete(other_user_flow)
+            await session.commit()