chore: merge main into web search fix

Author: Oxygen56

.secrets.baseline

@@ -1088,15 +1088,15 @@
         "filename": "src/backend/tests/conftest.py",
         "hashed_secret": "8bb6118f8fd6935ad0876a3be34a717d32708ffd",
         "is_verified": false,
-        "line_number": 559,
+        "line_number": 608,
         "is_secret": false
       },
       {
         "type": "Secret Keyword",
         "filename": "src/backend/tests/conftest.py",
         "hashed_secret": "61fbb5a12cd7b1f1fe1624120089efc0cd299e43",
         "is_verified": false,
-        "line_number": 769,
+        "line_number": 818,
         "is_secret": false
       }
     ],
@@ -1833,7 +1833,7 @@
         "filename": "src/backend/tests/unit/components/files_and_knowledge/test_file_component.py",
         "hashed_secret": "72cb70dbbafe97e5ea13ad88acd65d08389439b0",
         "is_verified": false,
-        "line_number": 774,
+        "line_number": 817,
         "is_secret": false
       }
     ],
@@ -7248,5 +7248,5 @@
       }
     ]
   },
-  "generated_at": "2026-08-04T17:46:52Z"
+  "generated_at": "2026-08-08T15:38:52Z"
 }

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "langflow"
-version = "1.11.2"
+version = "1.11.3"
 description = "A Python package with a built-in web application"
 requires-python = ">=3.10,<3.15"
 license = "MIT"
@@ -17,7 +17,7 @@ maintainers = [
 ]
 # Define your main dependencies here
 dependencies = [
-    "langflow-base[complete]>=0.11.2",
+    "langflow-base[complete]>=0.11.3",
     # langflow-extensions:bundle-deps-start
     # Release coordination: langflow declares BOUNDED ranges (>=A,<B) for every
     # curated lfx-* package, never exact pins -- exact pins in reusable library
@@ -249,6 +249,9 @@ ignore-regex = '.*(Stati Uniti|Tense=Pres).*'
 # 50-min step wall. A 90s cap fires below the ~127s stall, and timeout_method="thread" dumps ALL
 # thread stacks (faulthandler) — naming the exact blocking call even when it is a leaked background
 # task. Revert to timeout=150 / timeout_method="signal" once the stall is identified and bounded.
+# NOTE: the thread method's timer needs the GIL, so a C-level hang that holds the GIL defeats it
+# (observed: release-1.11.3 py3.13 Group 5 froze 25+ min with no dump). A second, GIL-proof
+# faulthandler watchdog (120s, dump + exit) backstops it in src/backend/tests/conftest.py.
 timeout = 90
 timeout_method = "thread"
 minversion = "6.0"

src/backend/base/langflow/api/build.py

@@ -13,6 +13,7 @@
 from lfx.log.logger import logger
 from lfx.schema.legacy_render import project_payload_to_v1
 from lfx.schema.schema import InputValueRequest
+from lfx.utils.file_path_security import LocalFileAccessError
 from sqlmodel import select
 
 from langflow.api.disconnect import DisconnectHandlerStreamingResponse
@@ -571,7 +572,7 @@ async def build_graph_and_get_order() -> tuple[list[str], list[str], Graph]:
                 error_message=str(exc),
             )
 
-            if "stream or streaming set to True" in str(exc):
+            if isinstance(exc, LocalFileAccessError) or "stream or streaming set to True" in str(exc):
                 raise HTTPException(status_code=400, detail=str(exc)) from exc
             await logger.aexception("Error checking build status: " + str(exc))
             raise HTTPException(status_code=500, detail=str(exc)) from exc
@@ -614,22 +615,33 @@ async def create_graph(fresh_session, flow_id_str: str, flow_name: str | None) -
                 session_id=effective_session_id,
                 run_id=str(job_id) if job_id is not None else run_id,
             )
-            if source_flow_id is not None:
-                graph.flow_id = str(flow_id)
-            return graph
-
-        if not flow_name:
-            result = await fresh_session.exec(select(Flow.name).where(Flow.id == flow_id))
-            flow_name = result.first()
-
-        return await build_graph_from_data(
-            flow_id=flow_id_str,
-            payload=data.model_dump(),
-            user_id=str(current_user.id),
-            flow_name=flow_name,
-            session_id=effective_session_id,
-            run_id=str(job_id) if job_id is not None else run_id,
-        )
+        else:
+            if not flow_name:
+                lookup_flow_id = source_flow_id if source_flow_id is not None else flow_id
+                result = await fresh_session.exec(select(Flow.name).where(Flow.id == lookup_flow_id))
+                flow_name = result.first()
+
+            # Sanitized public data still contains FileInput references under
+            # the real source-flow namespace. Build under that trusted scope;
+            # after parameter containment completes, switch the graph to the
+            # visitor-virtual execution ID below.
+            graph_build_flow_id = str(source_flow_id) if source_flow_id is not None else flow_id_str
+            graph = await build_graph_from_data(
+                flow_id=graph_build_flow_id,
+                payload=data.model_dump(),
+                user_id=str(current_user.id),
+                flow_name=flow_name,
+                session_id=effective_session_id,
+                run_id=str(job_id) if job_id is not None else run_id,
+            )
+
+        if source_flow_id is not None:
+            # This value comes from the server-resolved public flow, never from
+            # request data. FileInput containment and ChatInput attachment
+            # validation use it only after flow_id becomes visitor-virtual.
+            graph.source_flow_id = str(source_flow_id)
+            graph.flow_id = str(flow_id)
+        return graph
 
     def sort_vertices(graph: Graph) -> list[str]:
         try:

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/v1/flows.py

@@ -8,10 +8,11 @@
 from uuid import UUID
 
 import orjson
-from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
+from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile, status
 from fastapi.encoders import jsonable_encoder
 from fastapi_pagination import Page, Params
 from fastapi_pagination.ext.sqlmodel import apaginate
+from lfx.log.logger import logger
 from lfx.services.cache.utils import CACHE_MISS
 from pydantic import ValidationError
 from sqlmodel import and_, col, select
@@ -56,6 +57,10 @@
 from langflow.services.authorization.fetch import deny_to_404
 from langflow.services.authorization.utils import _resolve_authz_domain
 from langflow.services.cache.service import ThreadingInMemoryCache
+from langflow.services.database.lock_retry import (
+    is_database_lock_error,
+    run_with_lock_retry,
+)
 from langflow.services.database.models.deployment.exceptions import (
     araise_if_deployment_guard_error_or_skip,
 )
@@ -74,6 +79,7 @@
 # and FlowVersionError from the flow_version modules.
 from langflow.services.database.models.folder.constants import DEFAULT_FOLDER_NAME
 from langflow.services.database.models.folder.model import Folder
+from langflow.services.database.models.user.model import UserRead
 from langflow.services.deps import get_settings_service, get_storage_service
 from langflow.services.storage.service import StorageService
 from langflow.utils.compression import compress_response
@@ -103,6 +109,9 @@ def _handle_unique_constraint_error(exc: Exception, *, status_code: int = 400) -
 # build router
 router = APIRouter(prefix="/flows", tags=["Flows"])
 
+FLOW_UPDATE_FAILED = "Could not update the flow."
+FLOW_UPDATE_BUSY = "The database is busy. Please retry the request."
+
 
 @router.post("/", response_model=FlowRead, status_code=201)
 async def create_flow(
@@ -328,6 +337,7 @@ async def update_flow(
     storage_service: Annotated[StorageService, Depends(get_storage_service)],
 ):
     """Update a flow."""
+    actor = UserRead.model_validate(current_user, from_attributes=True)
     try:
         # Destination check: if the payload moves the flow into a new
         # workspace/folder, the caller must also be authorized to write at the
@@ -339,7 +349,7 @@ async def update_flow(
         if target_workspace_id != db_flow.workspace_id or target_folder_id != db_flow.folder_id:
             try:
                 await ensure_flow_permission(
-                    current_user,
+                    actor,
                     FlowAction.WRITE,
                     flow_id=flow_id,
                     flow_user_id=db_flow.user_id,
@@ -357,7 +367,7 @@ async def update_flow(
 
         async def operation() -> FlowRead:
             # Re-load inside each attempt so retry after nested rollback never uses an expired ORM instance.
-            db_flow_for_attempt = await _read_flow(session=session, flow_id=flow_id, user_id=current_user.id)
+            db_flow_for_attempt = await _read_flow(session=session, flow_id=flow_id, user_id=actor.id)
             if not db_flow_for_attempt:
                 raise HTTPException(status_code=404, detail="Flow not found")
             # TOCTOU: a concurrent PATCH could have moved this flow to a
@@ -367,7 +377,7 @@ async def operation() -> FlowRead:
             # stale check across a race.
             try:
                 await ensure_flow_permission(
-                    current_user,
+                    actor,
                     FlowAction.WRITE,
                     flow_id=flow_id,
                     flow_user_id=db_flow_for_attempt.user_id,
@@ -386,7 +396,7 @@ async def operation() -> FlowRead:
             ):
                 try:
                     await ensure_flow_permission(
-                        current_user,
+                        actor,
                         FlowAction.WRITE,
                         flow_id=flow_id,
                         flow_user_id=db_flow_for_attempt.user_id,
@@ -399,26 +409,47 @@ async def operation() -> FlowRead:
                 session=session,
                 db_flow=db_flow_for_attempt,
                 flow=flow,
-                user_id=current_user.id,
+                user_id=actor.id,
                 storage_service=storage_service,
             )
 
-        if folder_id_will_change:
-            return await retry_flow_operation_on_deployment_guard(
-                db=session,
-                user_id=current_user.id,
-                flow_ids=[flow_id],
-                operation=operation,
-            )
-        return await operation()
+        async def update_attempt(_attempt: int) -> FlowRead:
+            if folder_id_will_change:
+                return await retry_flow_operation_on_deployment_guard(
+                    db=session,
+                    user_id=actor.id,
+                    flow_ids=[flow_id],
+                    operation=operation,
+                )
+            return await operation()
+
+        return await run_with_lock_retry(
+            update_attempt,
+            session=session,
+            description=f"update_flow {flow_id}",
+        )
     except HTTPException:
         raise
     except Exception as e:
         await araise_if_deployment_guard_error_or_skip(
             e,
             log_message=f"op=update_flow flow_id={flow_id}",
         )
-        raise _handle_unique_constraint_error(e) from e
+        if is_database_lock_error(e):
+            await logger.awarning("op=update_flow flow_id=%s exhausted lock retries", flow_id)
+            raise HTTPException(
+                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                detail=FLOW_UPDATE_BUSY,
+                headers={"Retry-After": "1"},
+            ) from e
+        handled_error = _handle_unique_constraint_error(e)
+        if handled_error.status_code != status.HTTP_500_INTERNAL_SERVER_ERROR:
+            raise handled_error from e
+        await logger.aerror("op=update_flow flow_id=%s failed with %s", flow_id, type(e).__name__)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=FLOW_UPDATE_FAILED,
+        ) from e
 
 
 @router.put("/{flow_id}", response_model=FlowRead)

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

@@ -83,6 +83,7 @@
 from langflow.services.database.models.user.crud import get_user_by_username
 from langflow.services.database.models.user.model import User
 from langflow.services.deps import get_service
+from langflow.services.rate_limit.service import get_last_forwarded_for_hop
 
 # Constants
 ALL_INTERFACES_HOST = "0.0.0.0"  # noqa: S104
@@ -792,6 +793,9 @@ def get_client_ip(request: Request) -> str:
     (``rate_limit_trust_proxy``) do we consult ``X-Forwarded-For``, and then we
     take the rightmost entry — the last hop added by the trusted proxy, which a
     client cannot forge — mirroring ``langflow.services.rate_limit.service.get_client_ip``.
+    Every occurrence of the header is joined first, so a proxy that appends its
+    own line rather than extending the client's cannot leave the attacker's line
+    as the one we read.
 
     Args:
         request: FastAPI Request object
@@ -802,11 +806,9 @@ def get_client_ip(request: Request) -> str:
     # Only consult X-Forwarded-For when an operator has explicitly declared a
     # trusted proxy; otherwise the header is attacker-controlled.
     if get_settings_service().settings.rate_limit_trust_proxy:
-        forwarded_for = request.headers.get("X-Forwarded-For")
-        if forwarded_for:
-            # Rightmost entry = last hop added by the trusted proxy (unspoofable);
-            # the leftmost entry is client-supplied and must never be trusted.
-            return forwarded_for.split(",")[-1].strip()
+        last_hop = get_last_forwarded_for_hop(request)
+        if last_hop:
+            return last_hop
 
     # Default: trust only the real TCP peer.
     if request.client:

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,