feat(playground): Add auth gate, session persistence and token display to shareable playground (#12519)

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

Author: 2 people

src/backend/base/langflow/api/utils/flow_utils.py

@@ -104,33 +104,52 @@ async def cascade_delete_flow(session: AsyncSession, flow_id: uuid.UUID) -> None
         raise RuntimeError(msg, e) from e
 
 
-async def verify_public_flow_and_get_user(flow_id: uuid.UUID, client_id: str | None) -> tuple[User, uuid.UUID]:
+def compute_virtual_flow_id(identifier: str | uuid.UUID, flow_id: uuid.UUID) -> uuid.UUID:
+    """Compute a deterministic virtual flow ID for session/message isolation.
+
+    Args:
+        identifier: A unique identifier (user_id for authenticated users, client_id for anonymous).
+        flow_id: The original flow ID.
+
+    Returns:
+        A deterministic UUID v5 derived from the identifier and flow_id.
+    """
+    return uuid.uuid5(uuid.NAMESPACE_DNS, f"{identifier}_{flow_id}")
+
+
+async def verify_public_flow_and_get_user(
+    flow_id: uuid.UUID,
+    client_id: str | None,
+    authenticated_user_id: uuid.UUID | None = None,
+) -> tuple[User, uuid.UUID]:
     """Verify a public flow request and generate a deterministic flow ID.
 
     This utility function:
-    1. Checks that a client_id cookie is provided
+    1. Checks that a client_id cookie or authenticated_user_id is provided
     2. Verifies the flow exists and is marked as PUBLIC
-    3. Creates a deterministic UUID based on client_id and original flow_id
+    3. Creates a deterministic UUID based on the identifier and original flow_id
     4. Retrieves the flow owner user for permission purposes
 
-    This function is used to support public flow endpoints that don't require
-    authentication but still need to operate within the permission model.
+    When an authenticated_user_id is provided, it takes precedence over client_id
+    for UUID v5 generation. This enables DB-persisted sessions for logged-in users
+    on the shareable playground.
 
     Args:
         flow_id: The original flow ID to verify
         client_id: The client ID from the request cookie
+        authenticated_user_id: The authenticated user's ID (takes precedence over client_id)
 
     Returns:
         tuple: (flow owner user, deterministic flow ID for tracking)
 
     Raises:
         HTTPException:
-            - 400 if no client_id is provided
+            - 400 if neither client_id nor authenticated_user_id is provided
             - 403 if flow doesn't exist or isn't public
             - 403 if unable to retrieve the flow owner user
             - 403 if user is not found for public flow
     """
-    if not client_id:
+    if not client_id and not authenticated_user_id:
         raise HTTPException(status_code=400, detail="No client_id cookie found")
 
     # Check if the flow is public
@@ -143,9 +162,9 @@ async def verify_public_flow_and_get_user(flow_id: uuid.UUID, client_id: str | N
         if not flow or flow.access_type is not AccessTypeEnum.PUBLIC:
             raise HTTPException(status_code=403, detail="Flow is not public")
 
-    # Create a new flow ID using the client_id and flow_id
-    new_id = f"{client_id}_{flow_id}"
-    new_flow_id = uuid.uuid5(uuid.NAMESPACE_DNS, new_id)
+    # Use authenticated user_id for deterministic UUID when available, otherwise client_id
+    identifier = str(authenticated_user_id) if authenticated_user_id else client_id
+    new_flow_id = compute_virtual_flow_id(identifier, flow_id)
 
     # Get the user associated with the flow
     try:

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

@@ -42,12 +42,14 @@
     VerticesOrderResponse,
 )
 from langflow.exceptions.component import ComponentBuildError
-from langflow.services.auth.utils import get_current_active_user
+from langflow.services.auth.utils import get_current_active_user, get_current_user_optional
 from langflow.services.chat.service import ChatService
 from langflow.services.database.models.flow.model import AccessTypeEnum, Flow
+from langflow.services.database.models.user.model import User
 from langflow.services.deps import (
     get_chat_service,
     get_queue_service,
+    get_settings_service,
     get_telemetry_service,
     session_scope,
 )
@@ -648,6 +650,7 @@ async def build_public_tmp(
     flow_name: str | None = None,
     request: Request,
     queue_service: Annotated[JobQueueService, Depends(get_queue_service)],
+    authenticated_user: Annotated[User | None, Depends(get_current_user_optional)] = None,
     event_delivery: EventDeliveryType = EventDeliveryType.POLLING,
 ):
     """Build a public flow without requiring authentication.
@@ -676,6 +679,7 @@ async def build_public_tmp(
         flow_name: Optional name for the flow
         request: FastAPI request object (needed for cookie access)
         queue_service: Queue service for job management
+        authenticated_user: Optional authenticated user (resolved from cookie/token if present)
         event_delivery: Optional event delivery type - default is streaming
 
     Returns:
@@ -684,7 +688,16 @@ async def build_public_tmp(
     try:
         # Verify this is a public flow and get the associated user
         client_id = request.cookies.get("client_id")
-        owner_user, new_flow_id = await verify_public_flow_and_get_user(flow_id=flow_id, client_id=client_id)
+        # Only use authenticated user_id when auto-login is disabled.
+        # When AUTO_LOGIN=TRUE, the frontend uses client_id for UUID v5,
+        # so the backend must match to avoid flow_id mismatch.
+        auth_settings = get_settings_service().auth_settings
+        authenticated_user_id = authenticated_user.id if authenticated_user and not auth_settings.AUTO_LOGIN else None
+        owner_user, new_flow_id = await verify_public_flow_and_get_user(
+            flow_id=flow_id,
+            client_id=client_id,
+            authenticated_user_id=authenticated_user_id,
+        )
 
         # Validate the stored flow data after the public-access boundary.
         # Public flows never accept client-supplied data.
@@ -707,7 +720,7 @@ async def build_public_tmp(
             log_builds=log_builds or False,
             current_user=owner_user,
             queue_service=queue_service,
-            flow_name=flow_name or f"{client_id}_{flow_id}",
+            flow_name=flow_name or f"{authenticated_user_id or client_id}_{flow_id}",
         )
     except CustomComponentValidationError as exc:
         await logger.awarning(f"Public flow validation failed: {exc}")

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

@@ -7,6 +7,7 @@
 from sqlmodel import col, delete, select
 
 from langflow.api.utils import DbSession, custom_params
+from langflow.api.utils.flow_utils import compute_virtual_flow_id
 from langflow.schema.message import MessageResponse
 from langflow.services.auth.utils import get_current_active_user
 from langflow.services.database.models.flow.model import Flow
@@ -304,6 +305,164 @@ async def delete_messages_sessions(
     }
 
 
+@router.get("/messages/shared/sessions")
+async def get_shared_message_sessions(
+    session: DbSession,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    source_flow_id: Annotated[UUID, Query(description="The original public flow ID")],
+) -> list[str]:
+    """Get session IDs for a shared/public flow, scoped to the authenticated user.
+
+    Uses a deterministic virtual flow_id derived from the user's ID and the
+    original flow ID. Only messages stored under this virtual flow_id are returned.
+    """
+    try:
+        virtual_flow_id = compute_virtual_flow_id(current_user.id, source_flow_id)
+        stmt = select(MessageTable.session_id).distinct()
+        stmt = stmt.where(MessageTable.flow_id == virtual_flow_id)
+        stmt = stmt.where(col(MessageTable.session_id).isnot(None))
+
+        session_ids = await session.exec(stmt)
+        return list(session_ids)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e)) from e
+
+
+@router.get("/messages/shared")
+async def get_shared_messages(
+    session: DbSession,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    source_flow_id: Annotated[UUID, Query(description="The original public flow ID")],
+    session_id: Annotated[str | None, Query()] = None,
+    order_by: Annotated[str | None, Query()] = "timestamp",
+) -> list[MessageResponse]:
+    """Get messages for a shared/public flow, scoped to the authenticated user.
+
+    Uses a deterministic virtual flow_id derived from the user's ID and the
+    original flow ID. Only messages stored under this virtual flow_id are returned.
+    """
+    try:
+        virtual_flow_id = compute_virtual_flow_id(current_user.id, source_flow_id)
+        stmt = select(MessageTable)
+        stmt = stmt.where(MessageTable.flow_id == virtual_flow_id)
+
+        if session_id:
+            from urllib.parse import unquote
+
+            decoded_session_id = unquote(session_id)
+            stmt = stmt.where(MessageTable.session_id == decoded_session_id)
+        allowed_order_fields = {"timestamp", "sender", "sender_name", "session_id", "text"}
+        if order_by:
+            if order_by not in allowed_order_fields:
+                raise HTTPException(status_code=400, detail=f"Invalid order_by field: {order_by}")
+            order_col = getattr(MessageTable, order_by).asc()
+            stmt = stmt.order_by(order_col)
+
+        messages = await session.exec(stmt)
+        return [MessageResponse.model_validate(d, from_attributes=True) for d in messages]
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e)) from e
+
+
+@router.delete("/messages/shared/session/{session_id}", status_code=204)
+async def delete_shared_messages_session(
+    session_id: str,
+    session: DbSession,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    source_flow_id: Annotated[UUID, Query(description="The original public flow ID")],
+):
+    """Delete messages for a session on a shared/public flow, scoped to the authenticated user."""
+    try:
+        virtual_flow_id = compute_virtual_flow_id(current_user.id, source_flow_id)
+        stmt = (
+            delete(MessageTable)
+            .where(MessageTable.flow_id == virtual_flow_id)
+            .where(MessageTable.session_id == session_id)
+        )
+        await session.exec(stmt)
+    except Exception as e:
+        await session.rollback()
+        raise HTTPException(status_code=500, detail=str(e)) from e
+
+
+@router.put("/messages/shared/{message_id}", response_model=MessageRead)
+async def update_shared_message(
+    message_id: UUID,
+    message: MessageUpdate,
+    session: DbSession,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    source_flow_id: Annotated[UUID, Query(description="The original public flow ID")],
+):
+    """Update a message on a shared/public flow, scoped to the authenticated user."""
+    try:
+        virtual_flow_id = compute_virtual_flow_id(current_user.id, source_flow_id)
+        db_message = (
+            await session.exec(
+                select(MessageTable).where(
+                    MessageTable.id == message_id,
+                    MessageTable.flow_id == virtual_flow_id,
+                )
+            )
+        ).first()
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e)) from e
+
+    if not db_message:
+        raise HTTPException(status_code=404, detail="Message not found")
+
+    try:
+        message_dict = message.model_dump(exclude_unset=True, exclude_none=True)
+        if "text" in message_dict and message_dict["text"] != db_message.text:
+            message_dict["edit"] = True
+        db_message.sqlmodel_update(message_dict)
+        session.add(db_message)
+        await session.flush()
+        await session.refresh(db_message)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e)) from e
+    return db_message
+
+
+@router.patch("/messages/shared/session/{old_session_id}")
+async def rename_shared_session(
+    old_session_id: str,
+    new_session_id: Annotated[str, Query(description="The new session ID")],
+    session: DbSession,
+    current_user: Annotated[User, Depends(get_current_active_user)],
+    source_flow_id: Annotated[UUID, Query(description="The original public flow ID")],
+) -> list[MessageResponse]:
+    """Rename a session on a shared/public flow, scoped to the authenticated user."""
+    try:
+        virtual_flow_id = compute_virtual_flow_id(current_user.id, source_flow_id)
+        stmt = select(MessageTable).where(
+            MessageTable.flow_id == virtual_flow_id,
+            MessageTable.session_id == old_session_id,
+        )
+        messages = list(await session.exec(stmt))
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e)) from e
+
+    if not messages:
+        raise HTTPException(status_code=404, detail="No messages found with the given session ID")
+
+    try:
+        for message in messages:
+            message.session_id = new_session_id
+        session.add_all(messages)
+        await session.flush()
+
+        result = []
+        for message in messages:
+            await session.refresh(message)
+            result.append(MessageResponse.model_validate(message, from_attributes=True))
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e)) from e
+
+    return result
+
+
 @router.get("/transactions", dependencies=[Depends(get_current_active_user)])
 async def get_transactions(
     flow_id: Annotated[UUID, Query()],

src/backend/base/langflow/services/auth/utils.py

@@ -269,6 +269,30 @@ async def get_webhook_user(flow_id: str, request: Request) -> UserRead:
     return await _auth_service().get_webhook_user(flow_id, request)
 
 
+async def get_current_user_optional(
+    request: Request,
+    db: AsyncSession = Depends(injectable_session_scope),
+) -> User | None:
+    """Resolve the current user if authenticated, otherwise return None.
+
+    Checks HttpOnly cookie (access_token_lf), Authorization header, and API key.
+    Used by endpoints that support both authenticated and unauthenticated access.
+    """
+    token = request.cookies.get("access_token_lf")
+    api_key = request.query_params.get("x-api-key") or request.headers.get("x-api-key")
+    auth_header = request.headers.get("Authorization")
+    if auth_header and auth_header.startswith("Bearer "):
+        token = token or auth_header[len("Bearer ") :]
+
+    if not token and not api_key:
+        return None
+
+    try:
+        return await _auth_service().get_current_user_for_sse(token, api_key, db)
+    except (AuthenticationError, HTTPException):
+        return None
+
+
 async def get_current_active_user(user: User = Depends(get_current_user)) -> User | UserRead:
     result = await _auth_service().get_current_active_user(user)
     if result is None: