pengjunjie/langflow
1
0
Fork 1
mirror of https://github.com/langflow-ai/langflow.git synced 2026-07-24 09:49:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: added per-user endpoint for flows (#13072)

Browse Source 
* 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 <gabriel@langflow.org>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Lucas Oliveira
2026-05-15 16:01:03 -03:00
committed by GitHub
parent ea559d7971
commit 545d8598b7
4 changed files with 139 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
src/backend/base/langflow/api/v1/endpoints.py
View File

@ -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()

19
src/backend/base/langflow/helpers/user.py
View File

@ -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")

16
src/backend/base/langflow/services/auth/service.py
View File

@ -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

104
src/backend/tests/unit/test_webhook.py
View File

@ -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"]