fix(authz): support scoped project visibility (langflow-ai#14429)

* fix(authz): support scoped project visibility

* test: cover shared project edge cases

* fix(authz): address project visibility review

* fix(authz): enforce reserved project scope boundaries

Author: erichare

.secrets.baseline

@@ -84,6 +84,7 @@
     count_deployments_by_provider,
     delete_deployment_by_id,
     get_deployment_by_resource_key,
+    has_visible_deployment_for_provider,
 )
 from langflow.services.database.models.deployment.crud import (
     create_deployment_from_model as create_deployment_db,
@@ -792,25 +793,42 @@ async def list_deployments(
         )
     )
 
-    # OSS / no-plugin path keeps the strict owner gate (byte-for-byte the prior
-    # behavior). Relax it only when the prefilter actually lists ids to surface:
-    # an empty list means "no extra visibility", so there's nothing a cross-user
-    # reader could see under a provider they don't own — keep the strict 404 there
-    # rather than degrade it to an empty 200. A non-empty list resolves the
-    # provider account by id alone so a shared deployment under another user's
-    # provider account can be listed; the (owner ⊕ visible) union below still
-    # governs which rows actually surface.
-    if visibility_scope is not None and visibility_scope.has_cross_user_access:
+    # OSS / no-plugin path keeps the strict owner gate. A structured prefilter
+    # may relax it only after the same SQL visibility predicate used by the page
+    # proves that this specific provider owns at least one row visible to the
+    # caller. A coarse workspace/project/global grant alone is not enough:
+    # loading arbitrary foreign provider UUIDs would expose an existence oracle.
+    use_shared_provider_lookup = bool(
+        visibility_scope is not None
+        and visibility_scope.has_cross_user_access
+        and await has_visible_deployment_for_provider(
+            session,
+            user_id=current_user.id,
+            deployment_provider_account_id=provider_id,
+            visibility_scope=visibility_scope,
+        )
+    )
+    if use_shared_provider_lookup:
         provider_account = await get_shared_listing_provider_account_or_404(provider_id=provider_id, db=session)
     else:
         provider_account = await get_owned_provider_account_or_404(
             provider_id=provider_id, user_id=current_user.id, db=session
         )
-    await ensure_deployment_permission(
-        current_user,
-        DeploymentAction.READ,
-        project_id=project_id,
-    )
+    try:
+        await ensure_deployment_permission(
+            current_user,
+            DeploymentAction.READ,
+            project_id=project_id,
+        )
+    except HTTPException as exc:
+        if use_shared_provider_lookup:
+            # The relaxed provider-account lookup above loads by UUID alone so
+            # shared deployments can be listed. Once that path is active, mask
+            # a policy deny exactly like a missing account; otherwise callers
+            # could distinguish an existing foreign provider UUID (403) from a
+            # nonexistent UUID (404).
+            raise deny_to_404(exc, detail="Deployment provider account not found.") from exc
+        raise
     deployment_adapter = resolve_deployment_adapter(provider_account.provider_key)
     deployment_mapper = get_deployment_mapper(provider_account.provider_key)
     if load_from_provider:

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

@@ -173,7 +173,9 @@ async def read_flows(
         default_folder = (await session.exec(select(Folder).where(Folder.name == DEFAULT_FOLDER_NAME))).first()
         default_folder_id = default_folder.id if default_folder else None
 
-        starter_folder = (await session.exec(select(Folder).where(Folder.name == STARTER_FOLDER_NAME))).first()
+        starter_folder = (
+            await session.exec(select(Folder).where(Folder.name == STARTER_FOLDER_NAME, Folder.user_id.is_(None)))
+        ).first()
         starter_folder_id = starter_folder.id if starter_folder else None
 
         if not starter_folder and not default_folder:
@@ -1060,7 +1062,11 @@ async def read_basic_examples(
         cached_flow_reads = _starter_flows_cache.get("starter_flows")
         if cached_flow_reads is CACHE_MISS:
             try:
-                starter_folder = (await session.exec(select(Folder).where(Folder.name == STARTER_FOLDER_NAME))).first()
+                starter_folder = (
+                    await session.exec(
+                        select(Folder).where(Folder.name == STARTER_FOLDER_NAME, Folder.user_id.is_(None))
+                    )
+                ).first()
 
                 if not starter_folder:
                     return compress_response([])

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

@@ -8,7 +8,7 @@
 from lfx.log.logger import logger
 from lfx.services.mcp_composer.service import MCPComposerService
 from lfx.utils.util_strings import escape_like_pattern
-from sqlalchemy import literal, or_, update
+from sqlalchemy import literal, null, or_, update
 from sqlalchemy.orm import selectinload
 from sqlmodel import select
 
@@ -60,11 +60,13 @@
 from langflow.services.database.models.folder.model import (
     Folder,
     FolderCreate,
+    FolderListRead,
     FolderRead,
     FolderReadWithFlows,
     FolderUpdate,
 )
 from langflow.services.database.models.folder.pagination_model import FolderWithPaginatedFlows
+from langflow.services.database.models.user.model import User
 from langflow.services.deps import get_service, get_settings_service
 from langflow.services.schema import ServiceType
 
@@ -220,7 +222,7 @@ async def _move_flows_into_project() -> None:
     return folder_read
 
 
-@router.get("/", response_model=list[FolderRead], status_code=200)
+@router.get("/", response_model=list[FolderListRead], status_code=200)
 async def read_projects(
     *,
     session: DbSession,
@@ -254,26 +256,45 @@ async def read_projects(
         else:
             stmt = select(Folder).where(or_(owned_clause, Folder.user_id == None))  # noqa: E711
         projects = (await session.exec(stmt)).all()
-        projects = [project for project in projects if project.name != STARTER_FOLDER_NAME]
+        projects = [
+            project for project in projects if not (project.name == STARTER_FOLDER_NAME and project.user_id is None)
+        ]
         # When no DB prefilter is available (OSS pass-through), drop projects the
         # user can't read in memory. ``domain_extractor`` groups requests by
-        # workspace so each batch is evaluated against the right policy tuple
-        # (projects are the resource itself, so the domain falls back to
-        # workspace or ``*``). When the prefilter is active the SQL union is
-        # already authoritative — skip the per-row enforce to avoid an N+1.
+        # concrete project so each batch is evaluated against the same policy
+        # tuple as the single-resource guard. When the prefilter is active the
+        # SQL union is already authoritative — skip the per-row enforce to
+        # avoid an N+1.
         if visibility_scope is None:
             projects = await filter_visible_resources(
                 current_user,
                 resource_type="project",
                 candidates=list(projects),
-                domain_extractor=lambda project: _resolve_authz_domain(project.workspace_id, None),
+                domain_extractor=lambda project: _resolve_authz_domain(project.workspace_id, project.id),
                 owner_extractor=lambda project: project.user_id,
                 act=ProjectAction.READ,
             )
         sorted_projects = sorted(projects, key=lambda x: x.name != DEFAULT_FOLDER_NAME)
 
-        # Convert to FolderRead while session is still active to avoid detached instance errors
-        return [FolderRead.model_validate(project, from_attributes=True) for project in sorted_projects]
+        owner_ids = {project.user_id for project in sorted_projects if project.user_id is not None}
+        owners_by_id: dict[str, str] = {}
+        if owner_ids:
+            owner_rows = (await session.exec(select(User.id, User.username).where(User.id.in_(owner_ids)))).all()
+            owners_by_id = {str(owner_id): username for owner_id, username in owner_rows}
+
+        # Convert while the session is active so owner-qualified project lists
+        # do not trigger lazy loads after the request-scoped session closes.
+        return [
+            FolderListRead.model_validate(
+                project,
+                from_attributes=True,
+                update={
+                    "owner_username": owners_by_id.get(str(project.user_id)) if project.user_id is not None else None,
+                    "is_owner": str(project.user_id) == str(current_user.id),
+                },
+            )
+            for project in sorted_projects
+        ]
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e)) from e
 
@@ -365,7 +386,7 @@ async def read_project(
                     stmt,
                     id_column=Flow.id,
                     owner_clause=Flow.user_id == current_user.id,
-                    workspace_expression=literal(project.workspace_id),
+                    workspace_expression=null() if project.workspace_id is None else literal(project.workspace_id),
                     project_column=Flow.folder_id,
                     visibility=visibility_scope,
                 )
@@ -490,6 +511,17 @@ async def update_project(
     except HTTPException as exc:
         raise deny_to_404(exc, detail="Project not found") from exc
 
+    if (
+        project.name is not None
+        and project.name != existing_project.name
+        and existing_project.name == STARTER_FOLDER_NAME
+        and existing_project.user_id is None
+    ):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=f"The system-managed '{STARTER_FOLDER_NAME}' project cannot be renamed.",
+        )
+
     # Flow rollup uses the project owner — a non-owner editing a shared
     # project must touch the owner's flows, not the actor's same-folder
     # flows (which would be empty for a non-owner anyway).
@@ -723,10 +755,12 @@ async def _load_project() -> Folder | None:
     except HTTPException as exc:
         raise deny_to_404(exc, detail="Project not found") from exc
 
-    # Prevent deletion of the Langflow Assistant folder
-    if project.name == ASSISTANT_FOLDER_NAME:
-        msg = f"Cannot delete the '{ASSISTANT_FOLDER_NAME}' folder, that contains pre-built flows."
-        await logger.adebug("Cannot delete the '%s' folder, that contains pre-built flows.", ASSISTANT_FOLDER_NAME)
+    # Prevent deletion of projects managed by Langflow. The ownerless Starter
+    # Project is also a stable authorization boundary for bundled examples.
+    is_system_starter = project.name == STARTER_FOLDER_NAME and project.user_id is None
+    if project.name == ASSISTANT_FOLDER_NAME or is_system_starter:
+        msg = f"Cannot delete the '{project.name}' folder, which contains pre-built flows."
+        await logger.adebug("Cannot delete the '%s' folder, which contains pre-built flows.", project.name)
         raise HTTPException(
             status_code=403,
             detail=msg,

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

@@ -845,7 +845,7 @@ async def delete_starter_projects(session, folder_id) -> None:
 
 
 async def folder_exists(session, folder_name):
-    stmt = select(Folder).where(Folder.name == folder_name)
+    stmt = select(Folder).where(Folder.name == folder_name, Folder.user_id.is_(None))
     folder = (await session.exec(stmt)).first()
     return folder is not None
 
@@ -858,7 +858,7 @@ async def get_or_create_starter_folder(session):
         await session.flush()
         await session.refresh(db_folder)
         return db_folder
-    stmt = select(Folder).where(Folder.name == STARTER_FOLDER_NAME)
+    stmt = select(Folder).where(Folder.name == STARTER_FOLDER_NAME, Folder.user_id.is_(None))
     return (await session.exec(stmt)).first()
 
 

src/backend/base/langflow/initial_setup/setup.py

@@ -321,7 +321,11 @@ class _ResourceSpec:
         owner_kw="project_user_id",
         id_kw="project_id",
         workspace_kw="workspace_id",
-        scope_kw=None,
+        # Existing-project checks use the concrete project domain so plugins
+        # can distinguish reserved projects from their parent workspace.
+        # CREATE has no project id and still resolves to the workspace/global
+        # domain as before.
+        scope_kw="project_id",
     ),
     "knowledge_base": _ResourceSpec(
         resource_type="knowledge_base",

src/backend/base/langflow/services/authorization/guards.py

@@ -5,7 +5,7 @@
 from typing import TYPE_CHECKING, Any, TypeVar
 
 from lfx.services.authorization.base import ResourceVisibilityScope
-from sqlalchemy import Select, false
+from sqlalchemy import Select, and_, false
 from sqlmodel import col, or_
 
 from langflow.services.authorization.actions import FlowAction
@@ -259,16 +259,56 @@ def restrict_to_owned_or_visible_scope(
 ) -> StatementT:
     """Apply owner, concrete-ID, workspace, and project visibility before pagination."""
     if visibility.all_resources:
-        return stmt
+        if project_column is None or not visibility.excluded_global_project_ids:
+            return stmt
+        # Global role access can exclude reserved projects without enumerating
+        # every visible resource. Ownership and concrete grants remain additive,
+        # so users retain their own resources and directly shared resources in
+        # an otherwise excluded project. Folderless resources remain global.
+        global_clauses: list[ColumnElement[bool]] = [
+            owner_clause,
+            col(project_column).is_(None),
+            col(project_column).not_in(visibility.excluded_global_project_ids),
+        ]
+        if visibility.resource_ids:
+            global_clauses.append(col(id_column).in_(visibility.resource_ids))
+        return stmt.where(or_(*global_clauses))
 
     clauses: list[ColumnElement[bool]] = [owner_clause]
     if visibility.resource_ids:
         clauses.append(col(id_column).in_(visibility.resource_ids))
     resolved_workspace = workspace_expression
     if resolved_workspace is None and workspace_column is not None:
         resolved_workspace = col(workspace_column)
+    workspace_project_allowed: ColumnElement[bool] | None = None
+    if project_column is not None and visibility.excluded_workspace_project_ids:
+        # A workspace-only resource has no project to exclude. Keep it visible
+        # for an explicit workspace grant while excluding resources attached to
+        # reserved projects. The explicit ``IS NULL`` branch also keeps SQL's
+        # three-valued NULL semantics aligned with ``resource_visible_in_scope``.
+        workspace_project_allowed = or_(
+            col(project_column).is_(None),
+            col(project_column).not_in(visibility.excluded_workspace_project_ids),
+        )
     if resolved_workspace is not None and visibility.workspace_ids:
-        clauses.append(resolved_workspace.in_(visibility.workspace_ids))
+        workspace_clause = resolved_workspace.in_(visibility.workspace_ids)
+        if workspace_project_allowed is not None:
+            workspace_clause = and_(workspace_clause, workspace_project_allowed)
+        clauses.append(workspace_clause)
+    if resolved_workspace is not None and project_column is not None and visibility.include_unassigned_workspace:
+        # The logical unassigned workspace contains projects whose stored
+        # workspace is NULL; it does not contain folderless/workspace-less
+        # resources. Requiring a concrete project keeps list filtering aligned
+        # with direct authorization, which resolves this scope through the
+        # resource's project relation.
+        unassigned_project_allowed = col(project_column).is_not(None)
+        if visibility.excluded_workspace_project_ids:
+            unassigned_project_allowed = and_(
+                unassigned_project_allowed,
+                col(project_column).not_in(visibility.excluded_workspace_project_ids),
+            )
+        workspace_clause = and_(resolved_workspace.is_(None), unassigned_project_allowed)
+        clauses.append(workspace_clause)
     if project_column is not None and visibility.project_ids:
         clauses.append(col(project_column).in_(visibility.project_ids))
     return stmt.where(or_(*clauses))
@@ -305,9 +345,15 @@ def resource_visible_in_scope(
     project_id: UUID | None = None,
 ) -> bool:
     """Evaluate a compact visibility scope for an already-loaded resource."""
+    globally_visible = visibility.all_resources and (
+        project_id is None or project_id not in visibility.excluded_global_project_ids
+    )
+    workspace_project_allowed = project_id is None or project_id not in visibility.excluded_workspace_project_ids
+    unassigned_project_allowed = project_id is not None and project_id not in visibility.excluded_workspace_project_ids
     return bool(
-        visibility.all_resources
+        globally_visible
         or resource_id in visibility.resource_ids
-        or (workspace_id is not None and workspace_id in visibility.workspace_ids)
+        or (workspace_project_allowed and workspace_id is not None and workspace_id in visibility.workspace_ids)
+        or (unassigned_project_allowed and workspace_id is None and visibility.include_unassigned_workspace)
         or (project_id is not None and project_id in visibility.project_ids)
     )

src/backend/base/langflow/services/authorization/listing.py

@@ -397,16 +397,51 @@ async def _scope_to_owner_or_allowed(
             visible_ids=allowed_ids or (),
         )
 
+    from langflow.services.database.models.folder.model import Folder
+
+    # ``Deployment.workspace_id`` was introduced as denormalized plumbing and
+    # is absent on legacy rows (and on rows created through the current CRUD
+    # helper). Resolve the workspace from the authoritative project relation so
+    # Default-workspace access cannot absorb deployments whose project belongs
+    # to an explicit workspace.
+    project_workspace = (
+        select(Folder.workspace_id).where(Folder.id == Deployment.project_id).correlate(Deployment).scalar_subquery()
+    )
     return await apply_owned_or_visible_scope_prefilter(
         stmt,
         id_column=Deployment.id,
         owner_clause=Deployment.user_id == user_id,
-        workspace_column=Deployment.workspace_id,
+        workspace_expression=project_workspace,
         project_column=Deployment.project_id,
         visibility=visibility_scope,
     )
 
 
+async def has_visible_deployment_for_provider(
+    db: AsyncSession,
+    *,
+    user_id: UUID,
+    deployment_provider_account_id: UUID,
+    visibility_scope: ResourceVisibilityScope,
+) -> bool:
+    """Return whether this provider owns a deployment visible to the caller.
+
+    Cross-user deployment listing must resolve a foreign provider account to
+    select its adapter, but loading an arbitrary account by UUID creates an
+    existence oracle. Bind that relaxed lookup to the same owner/visibility
+    predicate used by the page and count queries, scoped to the requested
+    provider, before any provider metadata is loaded.
+    """
+    stmt = select(Deployment.id).where(Deployment.deployment_provider_account_id == deployment_provider_account_id)
+    stmt = await _scope_to_owner_or_allowed(
+        stmt,
+        user_id=user_id,
+        allowed_ids=None,
+        visibility_scope=visibility_scope,
+    )
+    return (await db.exec(stmt.limit(1))).first() is not None
+
+
 async def list_deployments_page(
     db: AsyncSession,
     *,