From 545d8598b74877f2bc7d2d25ee809f331dd9f373 Mon Sep 17 00:00:00 2001 From: Lucas Oliveira <62335616+lucaseduoli@users.noreply.github.com> Date: Fri, 15 May 2026 16:01:03 -0300 Subject: [PATCH] fix: added per-user endpoint for flows (#13072) * added per-user endpoint * fixed webhook test * added test * added cross user security * fixed nit * added depends wrapper * fix: update webhook_run_flow docstring to match new auth dependency signature * [autofix.ci] apply automated fixes * fixed flow 404 when running from second user * regenerated component index --------- Co-authored-by: ogabrielluiz Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> --- src/backend/base/langflow/api/v1/endpoints.py | 50 ++++++--- src/backend/base/langflow/helpers/user.py | 19 +++- .../base/langflow/services/auth/service.py | 16 +-- src/backend/tests/unit/test_webhook.py | 104 ++++++++++++++---- 4 files changed, 139 insertions(+), 50 deletions(-) diff --git a/src/backend/base/langflow/api/v1/endpoints.py b/src/backend/base/langflow/api/v1/endpoints.py index d991f21368..1a2175cfec 100644 --- a/src/backend/base/langflow/api/v1/endpoints.py +++ b/src/backend/base/langflow/api/v1/endpoints.py @@ -496,6 +496,35 @@ async def get_flow_for_current_user( return await get_flow_by_id_or_endpoint_name(flow_id_or_name, current_user.id) +async def get_flow_for_sse_user( + flow_id_or_name: str, + user: Annotated[User | UserRead, Depends(get_current_user_for_sse)], +) -> FlowRead: + """Auth-aware wrapper around ``get_flow_by_id_or_endpoint_name`` for SSE routes.""" + return await get_flow_by_id_or_endpoint_name(flow_id_or_name, user_id=user.id) + + +class WebhookAuth: + """Helper to carry both authenticated user and flow for webhook execution.""" + + def __init__(self, user: UserRead, flow: FlowRead): + self.user = user + self.flow = flow + + +async def get_webhook_auth( + flow_id_or_name: str, + request: Request, +) -> WebhookAuth: + """Auth-aware dependency that resolves both the webhook user and the flow. + + Centralizes the security logic for webhook run endpoints. + """ + webhook_user = await get_auth_service().get_webhook_user(flow_id_or_name, request) + flow = await get_flow_by_id_or_endpoint_name(flow_id_or_name, user_id=webhook_user.id) + return WebhookAuth(user=webhook_user, flow=flow) + + async def _run_flow_internal( *, background_tasks: BackgroundTasks, @@ -764,9 +793,7 @@ async def simplified_run_flow_session( @router.get("/webhook-events/{flow_id_or_name}", include_in_schema=False) async def webhook_events_stream( - flow_id_or_name: str, # noqa: ARG001 - Used by get_flow_by_id_or_endpoint_name dependency - flow: Annotated[Flow, Depends(get_flow_by_id_or_endpoint_name)], - user: Annotated[User | UserRead, Depends(get_current_user_for_sse)], + flow: Annotated[FlowRead, Depends(get_flow_for_sse_user)], request: Request, ): """Server-Sent Events (SSE) endpoint for real-time webhook build updates. @@ -777,12 +804,6 @@ async def webhook_events_stream( Authentication: Requires user to be logged in (via cookie) or provide API key. The user must own the flow to subscribe to its events. """ - # Verify user owns the flow - if str(flow.user_id) != str(user.id): - raise HTTPException( - status_code=HTTPStatus.FORBIDDEN, - detail="Access denied: You can only subscribe to events for flows you own", - ) async def event_generator() -> AsyncGenerator[str, None]: """Generate SSE events from the webhook event manager.""" @@ -823,15 +844,13 @@ async def webhook_events_stream( @router.post("/webhook/{flow_id_or_name}", response_model=dict, status_code=HTTPStatus.ACCEPTED) # noqa: RUF100, FAST003 async def webhook_run_flow( - flow_id_or_name: str, - flow: Annotated[Flow, Depends(get_flow_by_id_or_endpoint_name)], + auth: Annotated[WebhookAuth, Depends(get_webhook_auth)], request: Request, ): """Run a flow using a webhook request. Args: - flow_id_or_name: The flow ID or endpoint name (used by dependency). - flow: The flow to be executed. + auth: Resolved webhook user and flow, scoped to the authenticated caller. request: The incoming HTTP request. Returns: @@ -845,8 +864,9 @@ async def webhook_run_flow( await logger.adebug("Received webhook request") error_msg = "" - # Get the appropriate user for webhook execution based on auth settings - webhook_user = await get_auth_service().get_webhook_user(flow_id_or_name, request) + # Webhook user and flow are resolved by the dependency + webhook_user = auth.user + flow = auth.flow try: data = await request.body() diff --git a/src/backend/base/langflow/helpers/user.py b/src/backend/base/langflow/helpers/user.py index 34417acd50..8bc0ea4a07 100644 --- a/src/backend/base/langflow/helpers/user.py +++ b/src/backend/base/langflow/helpers/user.py @@ -8,15 +8,32 @@ from langflow.services.database.models.flow.model import Flow from langflow.services.database.models.user.model import User, UserRead -async def get_user_by_flow_id_or_endpoint_name(flow_id_or_name: str) -> UserRead | None: +async def get_user_by_flow_id_or_endpoint_name(flow_id_or_name: str, user_id: str | UUID | None = None) -> UserRead: async with session_scope_readonly() as session: + # Pre-resolve user_id and catch malformed IDs to prevent 500s + uuid_user_id: UUID | None = None + if user_id is not None: + try: + uuid_user_id = UUID(user_id) if isinstance(user_id, str) else user_id + except (ValueError, AttributeError) as exc: + raise HTTPException( + status_code=404, + detail=f"Flow identifier {flow_id_or_name} not found", + ) from exc + try: flow_id = UUID(flow_id_or_name) flow = await session.get(Flow, flow_id) except ValueError: stmt = select(Flow).where(Flow.endpoint_name == flow_id_or_name) + if uuid_user_id is not None: + stmt = stmt.where(Flow.user_id == uuid_user_id) flow = (await session.exec(stmt)).first() + # Enforce ownership check for both UUID and endpoint_name lookups + if flow and uuid_user_id and flow.user_id != uuid_user_id: + flow = None + if flow is None: raise HTTPException(status_code=404, detail=f"Flow identifier {flow_id_or_name} not found") diff --git a/src/backend/base/langflow/services/auth/service.py b/src/backend/base/langflow/services/auth/service.py index 7d20bd8361..6c7eb21d9c 100644 --- a/src/backend/base/langflow/services/auth/service.py +++ b/src/backend/base/langflow/services/auth/service.py @@ -428,20 +428,8 @@ class AuthService(BaseAuthService): logger.error(f"Webhook API key validation error: {exc}") raise HTTPException(status_code=403, detail="API key authentication failed") from exc - try: - flow_owner = await get_user_by_flow_id_or_endpoint_name(flow_id) - if flow_owner is None: - raise HTTPException(status_code=404, detail="Flow not found") - except HTTPException: - raise - except Exception as exc: - raise HTTPException(status_code=404, detail="Flow not found") from exc - - if flow_owner.id != authenticated_user.id: - raise HTTPException( - status_code=403, - detail="Access denied: You can only execute webhooks for flows you own", - ) + # The helper already enforces ownership and raises 404 if not found or not owned + await get_user_by_flow_id_or_endpoint_name(flow_id, user_id=authenticated_user.id) return authenticated_user diff --git a/src/backend/tests/unit/test_webhook.py b/src/backend/tests/unit/test_webhook.py index 13af7f58c8..c936f00657 100644 --- a/src/backend/tests/unit/test_webhook.py +++ b/src/backend/tests/unit/test_webhook.py @@ -837,31 +837,95 @@ class TestSimpleRunFlowTask: class TestWebhookEventsStreamAuth: - """Unit tests for webhook_events_stream authentication.""" + """Tests for webhook_events_stream authentication.""" - async def test_raises_403_when_user_does_not_own_flow(self): - """Should raise 403 when authenticated user doesn't own the flow.""" - from unittest.mock import Mock + async def test_raises_404_when_user_does_not_own_flow(self): + """Should raise 404 when authenticated user doesn't own the flow.""" + from unittest.mock import Mock, patch from fastapi import HTTPException - from langflow.api.v1.endpoints import webhook_events_stream - - flow = Mock() - flow.id = "test-flow-id" - flow.user_id = "owner-user-id" + from langflow.api.v1.endpoints import get_flow_for_sse_user mock_user = Mock() - mock_user.id = "different-user-id" # Different from flow owner + mock_user.id = "different-user-id" - request = Mock() + # Mock get_flow_by_id_or_endpoint_name to raise 404 as it would if user doesn't own it + with patch("langflow.api.v1.endpoints.get_flow_by_id_or_endpoint_name") as mock_get_flow: + mock_get_flow.side_effect = HTTPException(status_code=404, detail="Flow identifier test-flow-id not found") - with pytest.raises(HTTPException) as exc_info: - await webhook_events_stream( - flow_id_or_name="test-flow-id", - flow=flow, - user=mock_user, - request=request, - ) + with pytest.raises(HTTPException) as exc_info: + await get_flow_for_sse_user( + flow_id_or_name="test-flow-id", + user=mock_user, + ) - assert exc_info.value.status_code == 403 - assert "Access denied" in exc_info.value.detail + assert exc_info.value.status_code == 404 + assert "Flow identifier test-flow-id not found" in exc_info.value.detail + + async def test_webhook_events_stream_cross_user_access_returns_404( + self, client, added_webhook_test, user_two_api_key + ): + """Regression test: cross-user access to /webhook-events/{id} returns 404, not 403. + + This ensures no existence oracle (LE-639 compliance). + """ + flow_id = added_webhook_test["id"] + endpoint = f"api/v1/webhook-events/{flow_id}" + + # Clear cookies to ensure we don't use the identity from added_webhook_test/logged_in_headers + client.cookies.clear() + + # Access with User 2's API key + response = await client.get(endpoint, params={"x-api-key": user_two_api_key}) + + # Should return 404 because the flow doesn't belong to User 2 + assert response.status_code == 404 + assert f"Flow identifier {flow_id} not found" in response.json()["detail"] + + async def test_webhook_events_stream_cross_user_endpoint_name_returns_404( + self, client, added_webhook_test, user_two_api_key + ): + """Regression test: cross-user access to SSE via endpoint_name returns 404.""" + endpoint_name = added_webhook_test["endpoint_name"] + endpoint = f"api/v1/webhook-events/{endpoint_name}" + + # Clear cookies to ensure we don't use the identity from added_webhook_test/logged_in_headers + client.cookies.clear() + + # Access with User 2's API key + response = await client.get(endpoint, params={"x-api-key": user_two_api_key}) + + # Should return 404 because the flow doesn't belong to User 2 + assert response.status_code == 404 + assert f"Flow identifier {endpoint_name} not found" in response.json()["detail"] + + async def test_webhook_run_cross_user_uuid_returns_404(self, client, added_webhook_test, user_two_api_key): + """Regression test: cross-user access to run webhook via UUID returns 404.""" + flow_id = added_webhook_test["id"] + endpoint = f"api/v1/webhook/{flow_id}" + + # Clear cookies to ensure we don't use the identity from added_webhook_test/logged_in_headers + client.cookies.clear() + + # Access with User 2's API key + response = await client.post(endpoint, json={"test": "data"}, params={"x-api-key": user_two_api_key}) + + # Should return 404 because the flow doesn't belong to User 2 + # If it returns 403, it's an existence oracle + assert response.status_code == 404 + assert f"Flow identifier {flow_id} not found" in response.json()["detail"] + + async def test_webhook_run_cross_user_endpoint_name_returns_404(self, client, added_webhook_test, user_two_api_key): + """Regression test: cross-user access to run webhook via endpoint_name returns 404.""" + endpoint_name = added_webhook_test["endpoint_name"] + endpoint = f"api/v1/webhook/{endpoint_name}" + + # Clear cookies to ensure we don't use the identity from added_webhook_test/logged_in_headers + client.cookies.clear() + + # Access with User 2's API key + response = await client.post(endpoint, json={"test": "data"}, params={"x-api-key": user_two_api_key}) + + # Should return 404 because the flow doesn't belong to User 2 + assert response.status_code == 404 + assert f"Flow identifier {endpoint_name} not found" in response.json()["detail"]