fix: add missing ownership checks in projects API (GHSA-rpf3-3973-4gjr) (#12462)

* fix: enforce ownership check on flow assignment and paginated project read (GHSA-rpf3-3973-4gjr)

Prevent authenticated users from reassigning flows they don't own by adding
`Flow.user_id == current_user.id` to the UPDATE statements in `create_project`.
Also fix the paginated path in `read_project` which was missing the same filter,
allowing cross-user flow exfiltration via the `?page=&size=` query parameters.

* test: add security regression tests for GHSA-rpf3-3973-4gjr

- test_create_project_cannot_steal_other_users_flow: asserts that
  flows_list in create_project does not move flows owned by another user
- test_read_project_paginated_does_not_leak_other_users_flows: asserts
  that paginated GET /projects/{id} only returns flows owned by the
  requesting user

* test: improve security test naming and style for project ownership checks

Remove advisory IDs from test names and docstrings, rename helper and
test functions to match the existing codebase conventions.

* [autofix.ci] apply automated fixes

* fix: address ruff linting errors in ownership security tests

* test: add coverage for legitimate flows_list and paginated read assignments

---------

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

Author: 3 people

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

@@ -116,13 +116,17 @@ async def create_project(
 
         if project.components_list:
             update_statement_components = (
-                update(Flow).where(Flow.id.in_(project.components_list)).values(folder_id=new_project.id)  # type: ignore[attr-defined]
+                update(Flow)
+                .where(Flow.id.in_(project.components_list), Flow.user_id == current_user.id)  # type: ignore[attr-defined]
+                .values(folder_id=new_project.id)
             )
             await session.exec(update_statement_components)
 
         if project.flows_list:
             update_statement_flows = (
-                update(Flow).where(Flow.id.in_(project.flows_list)).values(folder_id=new_project.id)  # type: ignore[attr-defined]
+                update(Flow)
+                .where(Flow.id.in_(project.flows_list), Flow.user_id == current_user.id)  # type: ignore[attr-defined]
+                .values(folder_id=new_project.id)
             )
             await session.exec(update_statement_flows)
 
@@ -192,7 +196,7 @@ async def read_project(
     try:
         # Check if pagination is explicitly requested by the user (both page and size provided)
         if page is not None and size is not None:
-            stmt = select(Flow).where(Flow.folder_id == project_id)
+            stmt = select(Flow).where(Flow.folder_id == project_id, Flow.user_id == current_user.id)
 
             if Flow.updated_at is not None:
                 stmt = stmt.order_by(Flow.updated_at.desc())  # type: ignore[attr-defined]

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

@@ -1795,3 +1795,134 @@ async def test_download_project_sanitizes_windows_path_characters(
         assert "\\" not in file_names[0]
         assert ".." not in file_names[0]
         assert file_names[0].endswith(".json")
+
+
+async def _create_other_user(client: AsyncClient) -> tuple[str, dict]:
+    from langflow.services.auth.utils import get_password_hash
+    from langflow.services.database.models.user.model import User
+
+    user_id = str(uuid4())
+    username = f"other_user_{user_id[:8]}"
+    async with session_scope() as session:
+        user = User(
+            username=username,
+            password=get_password_hash("testpassword"),  # pragma: allowlist secret
+            is_active=True,
+            is_superuser=False,
+        )
+        session.add(user)
+        await session.commit()
+        await session.refresh(user)
+        created_id = str(user.id)
+
+    response = await client.post(
+        "api/v1/login", data={"username": username, "password": "testpassword"}
+    )  # pragma: allowlist secret
+    assert response.status_code == 200
+    token = response.json()["access_token"]
+    return created_id, {"Authorization": f"Bearer {token}"}
+
+
+async def test_create_project_does_not_reassign_other_users_flows(
+    client: AsyncClient,
+    logged_in_headers: dict,
+):
+    """Test that flows_list in create_project only moves flows owned by the requesting user."""
+    _, other_user_headers = await _create_other_user(client)
+
+    flow_resp = await client.post(
+        "api/v1/flows/",
+        json={"name": "user-flow", "data": {}},
+        headers=logged_in_headers,
+    )
+    assert flow_resp.status_code == status.HTTP_201_CREATED
+    flow_id = flow_resp.json()["id"]
+    original_folder_id = flow_resp.json()["folder_id"]
+
+    proj_resp = await client.post(
+        "api/v1/projects/",
+        json={"name": "other-project", "flows_list": [flow_id]},
+        headers=other_user_headers,
+    )
+    assert proj_resp.status_code == status.HTTP_201_CREATED
+    other_project_id = proj_resp.json()["id"]
+
+    flow_after = await client.get(f"api/v1/flows/{flow_id}", headers=logged_in_headers)
+    assert flow_after.status_code == status.HTTP_200_OK
+    assert flow_after.json()["folder_id"] == original_folder_id
+
+    proj_detail = await client.get(f"api/v1/projects/{other_project_id}", headers=other_user_headers)
+    assert proj_detail.status_code == status.HTTP_200_OK
+    assert len(proj_detail.json().get("flows", [])) == 0
+
+
+async def test_read_project_paginated_only_returns_current_users_flows(
+    client: AsyncClient,
+    logged_in_headers: dict,
+):
+    """Test that paginated GET /projects/{id} does not return flows owned by other users."""
+    _, other_user_headers = await _create_other_user(client)
+
+    flow_resp = await client.post(
+        "api/v1/flows/",
+        json={"name": "user-flow-paginated", "data": {}},
+        headers=logged_in_headers,
+    )
+    assert flow_resp.status_code == status.HTTP_201_CREATED
+    flow_id = flow_resp.json()["id"]
+    original_folder_id = flow_resp.json()["folder_id"]
+
+    proj_resp = await client.post(
+        "api/v1/projects/",
+        json={"name": "other-project-paginated", "flows_list": [flow_id]},
+        headers=other_user_headers,
+    )
+    assert proj_resp.status_code == status.HTTP_201_CREATED
+    other_project_id = proj_resp.json()["id"]
+
+    paginated = await client.get(
+        f"api/v1/projects/{other_project_id}",
+        params={"page": 1, "size": 50},
+        headers=other_user_headers,
+    )
+    assert paginated.status_code == status.HTTP_200_OK
+    items = paginated.json().get("flows", {}).get("items", [])
+    assert all(item["id"] != flow_id for item in items)
+
+    flow_after = await client.get(f"api/v1/flows/{flow_id}", headers=logged_in_headers)
+    assert flow_after.json()["folder_id"] == original_folder_id
+
+
+async def test_create_project_with_own_flows_assigns_them_correctly(
+    client: AsyncClient,
+    logged_in_headers: dict,
+):
+    """Test that flows_list in create_project correctly assigns flows owned by the requesting user."""
+    flow_resp = await client.post(
+        "api/v1/flows/",
+        json={"name": "my-flow", "data": {}},
+        headers=logged_in_headers,
+    )
+    assert flow_resp.status_code == status.HTTP_201_CREATED
+    flow_id = flow_resp.json()["id"]
+
+    proj_resp = await client.post(
+        "api/v1/projects/",
+        json={"name": "my-project", "flows_list": [flow_id]},
+        headers=logged_in_headers,
+    )
+    assert proj_resp.status_code == status.HTTP_201_CREATED
+    project_id = proj_resp.json()["id"]
+
+    flow_after = await client.get(f"api/v1/flows/{flow_id}", headers=logged_in_headers)
+    assert flow_after.status_code == status.HTTP_200_OK
+    assert flow_after.json()["folder_id"] == project_id
+
+    paginated = await client.get(
+        f"api/v1/projects/{project_id}",
+        params={"page": 1, "size": 50},
+        headers=logged_in_headers,
+    )
+    assert paginated.status_code == status.HTTP_200_OK
+    items = paginated.json().get("flows", {}).get("items", [])
+    assert any(item["id"] == flow_id for item in items)