fix(security): enforce admin-only component builds (#14436)

* fix(security): enforce admin-only component builds

* test(lfx): isolate component loader failure test

Author: erichare

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

@@ -15,6 +15,7 @@
 from lfx.services.cache.utils import CacheMiss
 from lfx.utils.flow_validation import (
     CustomComponentValidationError,
+    prepare_flow_build_for_user,
     prepare_public_flow_build,
     validate_flow_for_current_settings,
     validate_public_flow_no_code_execution,
@@ -177,6 +178,12 @@ async def retrieve_vertices_order(
         if not data:
             graph = await build_graph_from_db(flow_id=flow_id, session=session, chat_service=chat_service)
         else:
+            sanitized_data = await prepare_flow_build_for_user(
+                data.model_dump(),
+                is_superuser=current_user.is_superuser,
+            )
+            if sanitized_data is not None:
+                data = FlowDataRequest.model_validate(sanitized_data)
             graph = await build_and_cache_graph_from_data(
                 flow_id=flow_id, graph_data=data.model_dump(), chat_service=chat_service
             )
@@ -214,6 +221,8 @@ async def retrieve_vertices_order(
             raise HTTPException(status_code=400, detail=str(exc)) from exc
         if isinstance(exc, CustomComponentValidationError):
             raise HTTPException(status_code=400, detail=str(exc)) from exc
+        if isinstance(exc, RuntimeError):
+            raise HTTPException(status_code=503, detail=str(exc)) from exc
         await logger.aexception("Error checking build status")
         raise HTTPException(status_code=500, detail=str(exc)) from exc
 
@@ -308,7 +317,13 @@ async def build_flow(
 
     try:
         if data:
-            validate_flow_for_current_settings(data.model_dump())
+            raw_data = data.model_dump()
+            sanitized_data = await prepare_flow_build_for_user(
+                raw_data,
+                is_superuser=current_user.is_superuser,
+            )
+            if sanitized_data is not None:
+                data = FlowDataRequest.model_validate(sanitized_data)
         elif flow and flow.data:
             validate_flow_for_current_settings(flow.data)
     except CustomComponentValidationError as exc:

src/backend/base/langflow/api/v2/workflow.py

@@ -221,8 +221,8 @@ async def authorize_flow_action(
         ) from err
 
 
-def _apply_execution_gates(parsed, flow, current_user: UserRead) -> None:
-    """The langflow request gates that run before a flow executes."""
+def _apply_execution_gates(parsed, flow, current_user: UserRead):
+    """Run request gates and return any server-sanitized execution payload."""
     _reject_unsupported_sync_fields(parsed)
     _reject_sync_only_fields(parsed)
     try:
@@ -234,7 +234,7 @@ def _apply_execution_gates(parsed, flow, current_user: UserRead) -> None:
         if exc.status_code == status.HTTP_404_NOT_FOUND:
             raise _flow_not_found_http_exception(str(parsed.flow_id)) from exc
         raise
-    _validate_flow_data_for_execution(parsed, flow)
+    return _validate_flow_data_for_execution(parsed, flow, current_user)
 
 
 async def run_sync_with_mapping(
@@ -246,7 +246,7 @@ async def run_sync_with_mapping(
     background_tasks: BackgroundTasks,
 ) -> WorkflowExecutionResponse:
     """Inline sync run with the langflow timeout/validation error mapping."""
-    _apply_execution_gates(parsed, flow, current_user)
+    parsed = _apply_execution_gates(parsed, flow, current_user)
     job_id = uuid4()
     try:
         return await execute_sync_workflow_with_timeout(
@@ -321,7 +321,7 @@ def build_stream_response(
     side-channel, and vertex-build persistence all survive. Validation gates run
     before the response is constructed so a bad request fails before streaming.
     """
-    _apply_execution_gates(parsed, flow, current_user)
+    parsed = _apply_execution_gates(parsed, flow, current_user)
     adapter = get_stream_adapter(
         stream_protocol,
         StreamAdapterContext(
@@ -346,7 +346,7 @@ async def submit_background_with_mapping(
     stream_protocol: str,
 ) -> WorkflowJobResponse:
     """Queue a durable background run with the langflow service error mapping."""
-    _apply_execution_gates(parsed, flow, current_user)
+    parsed = _apply_execution_gates(parsed, flow, current_user)
     try:
         return await execute_workflow_background(
             parsed=parsed,

src/backend/base/langflow/api/v2/workflow_validation.py

@@ -8,8 +8,14 @@
 
 from __future__ import annotations
 
+from dataclasses import replace
+
 from fastapi import HTTPException, status
-from lfx.utils.flow_validation import CustomComponentValidationError, validate_flow_for_current_settings
+from lfx.utils.flow_validation import (
+    CustomComponentValidationError,
+    prepare_flow_build_for_user_from_cache,
+    validate_flow_for_current_settings,
+)
 from lfx.workflow.converters import ParsedWorkflowRun
 
 from langflow.services.authorization.fetch import deny_to_404
@@ -85,17 +91,27 @@ def _enforce_flow_data_override_owner(parsed: ParsedWorkflowRun, flow: FlowRead,
     )
 
 
-def _validate_flow_data_for_execution(parsed: ParsedWorkflowRun, flow: FlowRead) -> None:
-    """Apply the same server-side component policy gate used by v1/public runs."""
+def _validate_flow_data_for_execution(
+    parsed: ParsedWorkflowRun,
+    flow: FlowRead,
+    current_user: UserRead,
+) -> ParsedWorkflowRun:
+    """Apply component policies and return sanitized caller-supplied graph data."""
     try:
         if parsed.data is not None:
-            validate_flow_for_current_settings(parsed.data)
+            sanitized_data = prepare_flow_build_for_user_from_cache(
+                parsed.data,
+                is_superuser=current_user.is_superuser,
+            )
+            if sanitized_data is not None:
+                return replace(parsed, data=sanitized_data)
         elif flow.data:
             validate_flow_for_current_settings(flow.data)
     except CustomComponentValidationError as exc:
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
     except RuntimeError as exc:
         raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)) from exc
+    return parsed
 
 
 def _validate_output_ids(output_ids: list[str] | None, terminal_node_ids: list[str]) -> None:

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

@@ -14,6 +14,7 @@
 from contextlib import asynccontextmanager
 from types import SimpleNamespace
 from typing import Any
+from unittest.mock import AsyncMock, MagicMock
 from uuid import UUID, uuid4
 
 import pytest
@@ -62,7 +63,12 @@ def patch_build_flow(monkeypatch):
     """Install fakes for session_scope, _read_flow, ensure_flow_permission, start_flow_build."""
     from langflow.api.v1 import chat as chat_module
 
-    state: dict[str, Any] = {"session_exec": [], "read_flow": None, "ensure_raises": None}
+    state: dict[str, Any] = {
+        "session_exec": [],
+        "read_flow": None,
+        "ensure_raises": None,
+        "start_kwargs": None,
+    }
 
     @asynccontextmanager
     async def fake_session_scope():
@@ -75,7 +81,8 @@ async def fake_ensure(*_args, **_kwargs):
         if state["ensure_raises"] is not None:
             raise state["ensure_raises"]
 
-    async def fake_start_build(**_kwargs):
+    async def fake_start_build(**kwargs):
+        state["start_kwargs"] = kwargs
         return "fake-job-id"
 
     monkeypatch.setattr(chat_module, "session_scope", fake_session_scope)
@@ -171,7 +178,10 @@ async def test_build_flow_owner_can_override_flow_data(patch_build_flow, monkeyp
     from langflow.api.v1 import chat as chat_module
     from langflow.api.v1.schemas import FlowDataRequest
 
-    monkeypatch.setattr(chat_module, "validate_flow_for_current_settings", lambda _data: None)
+    async def allow_inline(_data, *, is_superuser):
+        assert is_superuser is False
+
+    monkeypatch.setattr(chat_module, "prepare_flow_build_for_user", allow_inline)
 
     owner = _make_user()
     flow = _make_flow(owner_id=owner.id, public=False)
@@ -188,6 +198,152 @@ async def test_build_flow_owner_can_override_flow_data(patch_build_flow, monkeyp
     assert result == {"job_id": "fake-job-id"}
 
 
+@pytest.mark.asyncio
+async def test_build_flow_admin_only_blocks_non_superuser_inline_custom_code(patch_build_flow, monkeypatch):
+    """Admin-only policy is applied before owner-supplied inline graph data reaches the build worker."""
+    from langflow.api.v1 import chat as chat_module
+    from langflow.api.v1.schemas import FlowDataRequest
+    from lfx.utils.flow_validation import CustomComponentValidationError
+
+    owner = _make_user()
+    flow = _make_flow(owner_id=owner.id, public=False)
+    patch_build_flow["read_flow"] = flow
+    override = FlowDataRequest(nodes=[{"id": "custom"}], edges=[])
+
+    async def reject_custom_code(_data, *, is_superuser):
+        assert is_superuser is False
+        message = "custom components are restricted to administrators"
+        raise CustomComponentValidationError(message)
+
+    monkeypatch.setattr(chat_module, "prepare_flow_build_for_user", reject_custom_code)
+
+    with pytest.raises(HTTPException) as excinfo:
+        await chat_module.build_flow(
+            flow_id=flow.id,
+            background_tasks=None,
+            current_user=owner,
+            queue_service=_make_queue_service(),
+            data=override,
+        )
+
+    assert excinfo.value.status_code == 400
+    assert "restricted to administrators" in excinfo.value.detail
+    assert patch_build_flow["start_kwargs"] is None
+
+
+@pytest.mark.asyncio
+async def test_build_flow_admin_only_passes_sanitized_template_to_worker(patch_build_flow, monkeypatch):
+    """A known template remains usable, but the worker receives the server-trusted payload."""
+    from langflow.api.v1 import chat as chat_module
+    from langflow.api.v1.schemas import FlowDataRequest
+
+    owner = _make_user()
+    flow = _make_flow(owner_id=owner.id, public=False)
+    patch_build_flow["read_flow"] = flow
+    override = FlowDataRequest(nodes=[{"id": "known", "data": {"source": "request"}}], edges=[])
+    sanitized = {"nodes": [{"id": "known", "data": {"source": "server"}}], "edges": [], "viewport": None}
+
+    async def sanitize(_data, *, is_superuser):
+        assert is_superuser is False
+        return sanitized
+
+    monkeypatch.setattr(chat_module, "prepare_flow_build_for_user", sanitize)
+
+    result = await chat_module.build_flow(
+        flow_id=flow.id,
+        background_tasks=None,
+        current_user=owner,
+        queue_service=_make_queue_service(),
+        data=override,
+    )
+
+    assert result == {"job_id": "fake-job-id"}
+    assert patch_build_flow["start_kwargs"]["data"].model_dump() == sanitized
+    assert override.nodes[0]["data"]["source"] == "request"
+
+
+@pytest.mark.asyncio
+async def test_build_flow_admin_only_keeps_superuser_inline_custom_code(patch_build_flow, monkeypatch):
+    """The operator's admin-only setting preserves the documented superuser exception."""
+    from langflow.api.v1 import chat as chat_module
+    from langflow.api.v1.schemas import FlowDataRequest
+
+    owner = _make_user(is_superuser=True)
+    flow = _make_flow(owner_id=owner.id, public=False)
+    patch_build_flow["read_flow"] = flow
+    override = FlowDataRequest(nodes=[{"id": "custom"}], edges=[])
+    seen: dict[str, Any] = {}
+
+    async def preserve_superuser_data(data, *, is_superuser):
+        assert is_superuser is True
+        seen["validated"] = data
+
+    monkeypatch.setattr(chat_module, "prepare_flow_build_for_user", preserve_superuser_data)
+
+    result = await chat_module.build_flow(
+        flow_id=flow.id,
+        background_tasks=None,
+        current_user=owner,
+        queue_service=_make_queue_service(),
+        data=override,
+    )
+
+    assert result == {"job_id": "fake-job-id"}
+    assert seen["validated"] == override.model_dump()
+    assert patch_build_flow["start_kwargs"]["data"] is override
+
+
+@pytest.mark.asyncio
+async def test_legacy_vertices_route_passes_only_sanitized_inline_data(monkeypatch):
+    """The still-routed legacy graph cache cannot bypass the inline-data policy."""
+    from langflow.api.v1 import chat as chat_module
+    from langflow.api.v1.schemas import FlowDataRequest
+
+    owner = _make_user()
+    flow = _make_flow(owner_id=owner.id)
+    original = FlowDataRequest(nodes=[{"id": "known", "data": {"source": "request"}}], edges=[])
+    sanitized = {"nodes": [{"id": "known", "data": {"source": "server"}}], "edges": [], "viewport": None}
+    seen: dict[str, Any] = {}
+
+    async def ensure_allowed(*_args, **_kwargs):
+        return None
+
+    async def sanitize(_data, *, is_superuser):
+        assert is_superuser is False
+        return sanitized
+
+    graph = MagicMock()
+    graph.prepare.return_value = graph
+    graph.vertices = []
+    graph.vertices_to_run = set()
+    graph.first_layer = []
+    graph.run_id = None
+    graph.set_run_id.side_effect = lambda value: setattr(graph, "run_id", value)
+
+    async def build_from_data(*, graph_data, **_kwargs):
+        seen["graph_data"] = graph_data
+        return graph
+
+    chat_service = SimpleNamespace(set_cache=AsyncMock())
+    monkeypatch.setattr(chat_module, "ensure_flow_permission", ensure_allowed)
+    monkeypatch.setattr(chat_module, "prepare_flow_build_for_user", sanitize)
+    monkeypatch.setattr(chat_module, "build_and_cache_graph_from_data", build_from_data)
+    monkeypatch.setattr(chat_module, "get_chat_service", lambda: chat_service)
+    monkeypatch.setattr(chat_module, "get_telemetry_service", MagicMock)
+    monkeypatch.setattr(chat_module, "get_top_level_vertices", lambda *_args: [])
+
+    await chat_module.retrieve_vertices_order(
+        flow_id=flow.id,
+        background_tasks=SimpleNamespace(add_task=lambda *_args, **_kwargs: None),
+        data=original,
+        session=_FakeSession([[flow]]),
+        current_user=owner,
+    )
+
+    assert seen["graph_data"] == sanitized
+    assert original.nodes[0]["data"]["source"] == "request"
+
+
 @pytest.mark.asyncio
 async def test_build_flow_plugin_deny_returns_404_not_403(patch_build_flow):
     """ensure_flow_permission raising 403 must surface as 404 (UUID privacy)."""