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

fix: Route guards and unit tests

Browse Source
This commit is contained in:
Eric Hare
2026-05-21 15:27:36 -07:00
parent 373d10ca81
commit 5087977fa2
19 changed files with 1263 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
src/backend/base/langflow/alembic/versions/7c8d9e0f1a2b_authz_foundations.py
View File

@ -115,6 +115,18 @@ def upgrade() -> None:
batch_op.create_index(batch_op.f("ix_authz_role_assignment_role_id"), ["role_id"], unique=False)
batch_op.create_index(batch_op.f("ix_authz_role_assignment_domain_id"), ["domain_id"], unique=False)
# Hot-path lookup index: "all assignments for user X scoped to domain
# (type, id)" — the canonical query an enterprise enforcer issues on
# every request to compute effective roles. Single-column indexes
# leave the planner with a choice between two non-covering scans;
# this composite gives a single index seek.
op.create_index(
"ix_authz_role_assignment_user_domain",
"authz_role_assignment",
["user_id", "domain_type", "domain_id"],
unique=False,
)
# Partial unique indexes — NULL domain_id is never-equal in SQL, so
# split the constraint into scoped and unscoped buckets. The unscoped
# bucket filters on domain_id IS NULL only (NOT also on
@ -194,16 +206,16 @@ def upgrade() -> None:
sa.PrimaryKeyConstraint("id"),
sa.CheckConstraint(
"scope IN ('private', 'team', 'user', 'public')",
name="scope_enum",
name="ck_authz_share_scope_enum",
),
sa.CheckConstraint(
"permission_level IN ('read', 'write', 'execute', 'admin')",
name="permission_enum",
name="ck_authz_share_permission_enum",
),
sa.CheckConstraint(
"(scope IN ('team', 'user') AND target_id IS NOT NULL) "
"OR (scope IN ('private', 'public') AND target_id IS NULL)",
name="scope_target_consistency",
name="ck_authz_share_scope_target_consistency",
),
)
with op.batch_alter_table("authz_share", schema=None) as batch_op:
@ -249,6 +261,14 @@ def upgrade() -> None:
)
with op.batch_alter_table("authz_edit_lock", schema=None) as batch_op:
batch_op.create_index(batch_op.f("ix_authz_edit_lock_flow_id"), ["flow_id"], unique=True)
# Expired-lock sweeper queries WHERE expires_at < now() — without
# this index that's a full table scan. Cheaper to add here than
# to retrofit once a cleanup job lands and starts hurting.
batch_op.create_index(
batch_op.f("ix_authz_edit_lock_expires_at"),
["expires_at"],
unique=False,
)
# ------------------------------------------------------------------
# authz_audit_log — append-only authorization audit
@ -266,6 +286,15 @@ def upgrade() -> None:
sa.Column("timestamp", sa.DateTime(timezone=True), nullable=False),
sa.ForeignKeyConstraint(["user_id"], ["user.id"], ondelete="SET NULL"),
sa.PrimaryKeyConstraint("id"),
# ``owner_override`` is the third value the framework writes
# (see ``_AUDIT_OWNER_OVERRIDE`` in
# ``services/authorization/utils.py``) — without it in the CHECK
# set, an enterprise plugin's owner-shortcircuit audit row would
# silently violate the constraint.
sa.CheckConstraint(
"result IN ('allow', 'deny', 'owner_override')",
name="ck_authz_audit_log_result_enum",
),
)
with op.batch_alter_table("authz_audit_log", schema=None) as batch_op:
batch_op.create_index(batch_op.f("ix_authz_audit_log_user_id"), ["user_id"], unique=False)

75
src/backend/base/langflow/alembic/versions/8d3a1f9c2e0b_seed_authz_system_roles.py
View File

@ -120,34 +120,55 @@ def upgrade() -> None:
)
timestamp = datetime.now(timezone.utc)
# ``sa.Uuid`` adapts ``UUID`` objects per-dialect (CHAR(32) on SQLite,
# native UUID on Postgres). Passing a bare string skips that path and
# fails at bind time on SQLite (``'str' object has no attribute 'hex'``),
# so we pass real UUID objects here.
# ``sa.JSON`` serializes Python lists natively on both SQLite and Postgres.
# ``json.dumps`` here would write the JSON-encoded string *inside* the
# JSON column, which downstream readers (Enterprise PolicySync expects a
# list) would have to peel a second time.
for name, description, permissions in _SYSTEM_ROLES:
already_present = conn.execute(
sa.select(sa.literal(1)).select_from(authz_role).where(authz_role.c.name == name)
).scalar()
if already_present:
continue
# ``sa.Uuid`` adapts ``UUID`` objects per-dialect (CHAR(32) on SQLite,
# native UUID on Postgres). Passing a bare string skips that path and
# fails at bind time on SQLite (``'str' object has no attribute
# 'hex'``), so we pass real UUID objects here.
# ``sa.JSON`` serializes Python lists natively on both SQLite and
# Postgres. ``json.dumps`` here would write the JSON-encoded string
# *inside* the JSON column, which downstream readers (Enterprise
# PolicySync expects a list) would have to peel a second time.
conn.execute(
authz_role.insert().values(
id=uuid4(),
name=name,
description=description,
is_system=True,
permissions=list(permissions),
parent_role_id=None,
workspace_id=None,
created_at=timestamp,
updated_at=timestamp,
created_by=None,
)
)
# Use an atomic check-then-insert to harden against concurrent
# rollouts (two pods running migrations against the same DB). The
# previous implementation did a SELECT then an INSERT in two
# statements: under concurrency both could pass the SELECT, both
# could attempt the INSERT, and one would crash on the unique
# constraint. ``ON CONFLICT DO NOTHING`` / ``OR IGNORE`` makes this
# idempotent under any number of concurrent writers without a lock.
dialect = conn.dialect.name
values = {
"id": uuid4(),
"name": name,
"description": description,
"is_system": True,
"permissions": list(permissions),
"parent_role_id": None,
"workspace_id": None,
"created_at": timestamp,
"updated_at": timestamp,
"created_by": None,
}
if dialect == "postgresql":
from sqlalchemy.dialects.postgresql import insert as pg_insert
stmt = pg_insert(authz_role).values(**values).on_conflict_do_nothing(index_elements=["name"])
conn.execute(stmt)
elif dialect == "sqlite":
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
stmt = sqlite_insert(authz_role).values(**values).on_conflict_do_nothing(index_elements=["name"])
conn.execute(stmt)
else:
# Fallback for other dialects (e.g. unit test stand-ins): use the
# original check-then-insert. Not fully race-safe but matches the
# previous behavior and never raises on a clean fresh DB.
already_present = conn.execute(
sa.select(sa.literal(1)).select_from(authz_role).where(authz_role.c.name == name)
).scalar()
if already_present:
continue
conn.execute(authz_role.insert().values(**values))
def downgrade() -> None:

9
src/backend/base/langflow/api/utils/flow_utils.py
View File

@ -13,6 +13,7 @@ from sqlalchemy import delete
from sqlmodel import col, select
from sqlmodel.ext.asyncio.session import AsyncSession
from langflow.services.database.models.auth.authz import AuthzShare
from langflow.services.database.models.deployment.exceptions import (
araise_if_deployment_guard_error_or_skip,
)
@ -112,6 +113,14 @@ async def cascade_delete_flow(session: AsyncSession, flow_id: uuid.UUID) -> None
if trace_ids:
await session.exec(delete(SpanTable).where(col(SpanTable.trace_id).in_(trace_ids)))
await session.exec(delete(TraceTable).where(col(TraceTable.id).in_(trace_ids)))
# authz_share is polymorphic over resource_type/resource_id with no
# FK, so DB cascades cannot remove stale share rows when the flow is
# deleted. Clean them up here so a deleted flow's grants do not
# silently survive — that would let an enterprise plugin keep
# honoring share rows that point at a tombstoned resource.
await session.exec(
delete(AuthzShare).where(AuthzShare.resource_type == "flow").where(AuthzShare.resource_id == flow_id)
)
await session.exec(delete(Flow).where(Flow.id == flow_id))
except Exception as e:
await araise_if_deployment_guard_error_or_skip(

74
src/backend/base/langflow/api/v1/authz_shares.py
View File

@ -28,6 +28,7 @@ from typing import Annotated
from uuid import UUID
from fastapi import APIRouter, HTTPException, Query, status
from lfx.log.logger import logger
from sqlmodel import select
from langflow.api.utils import CurrentActiveUser, DbSession
@ -200,10 +201,14 @@ async def create_share(
try:
await session.flush()
except Exception as exc:
# Log the raw exception server-side; the client gets a fixed string
# so we don't leak table/column/constraint names through the 409
# body (security rule: error messages don't disclose schema).
await session.rollback()
logger.warning("authz_share insert rejected: %s", exc)
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"Share could not be created: {exc}",
detail="Share could not be created: it may already exist or conflict with an existing share.",
) from exc
await session.refresh(row)
@ -227,6 +232,10 @@ async def create_share(
return ShareRead.model_validate(row, from_attributes=True)
_LIST_SHARES_MAX_LIMIT = 200
_LIST_SHARES_DEFAULT_LIMIT = 100
@router.get("", response_model=list[ShareRead])
@router.get("/", response_model=list[ShareRead])
async def list_shares(
@ -236,12 +245,19 @@ async def list_shares(
resource_id: Annotated[UUID | None, Query()] = None,
target_id: Annotated[UUID | None, Query()] = None,
scope: Annotated[str | None, Query()] = None,
limit: Annotated[int, Query(ge=1, le=_LIST_SHARES_MAX_LIMIT)] = _LIST_SHARES_DEFAULT_LIMIT,
offset: Annotated[int, Query(ge=0)] = 0,
) -> list[ShareRead]:
"""List share rows.
Resource owners and superusers see every matching row. Non-owners only
see rows whose ``target_id`` is their own user id — they need to know
what's shared with them, but not the full grant ledger.
see rows whose ``target_id`` is their own user id, rows scoped to a team
they belong to, public rows, or rows on resources they own.
Always paginated — ``limit`` is capped at 200 to bound DB load. The
previous unbounded implementation issued an N+1 ``select(AuthzTeamMember)``
per row; this version pre-fetches the caller's team memberships once and
uses an in-memory set lookup.
"""
await ensure_share_permission(
current_user,
@ -264,12 +280,22 @@ async def list_shares(
raise HTTPException(status_code=400, detail=f"Unknown scope {scope!r}") from exc
stmt = stmt.where(AuthzShare.scope == scope_value)
# Stable ordering + bounded fetch. ``created_at`` is monotonically
# increasing for a given operator so this is effectively cursor-like;
# callers wanting strict cursor pagination can layer ``id`` ties.
stmt = stmt.order_by(AuthzShare.created_at.desc(), AuthzShare.id).offset(offset).limit(limit)
rows = list(await session.exec(stmt))
is_superuser = getattr(current_user, "is_superuser", False)
if is_superuser:
return [ShareRead.model_validate(row, from_attributes=True) for row in rows]
# Pre-fetch the caller's team memberships once — the previous per-row
# ``select(AuthzTeamMember)`` was an N+1 that scaled with the page size.
team_membership_stmt = select(AuthzTeamMember.team_id).where(AuthzTeamMember.user_id == current_user.id)
caller_team_ids: set[UUID] = set(await session.exec(team_membership_stmt))
# Non-superuser visibility: owner / creator / direct user target / team
# member / public. See ``_user_can_see_share`` for the full rule set.
visible: list[ShareRead] = []
@ -282,16 +308,41 @@ async def list_shares(
resource_type=row.resource_type,
resource_id=row.resource_id,
)
if await _user_can_see_share(
session,
if _row_visible_to(
row=row,
user_id=current_user.id,
resource_owner_id=owner_cache[key],
caller_team_ids=caller_team_ids,
):
visible.append(ShareRead.model_validate(row, from_attributes=True))
return visible
def _row_visible_to(
*,
row: AuthzShare,
user_id: UUID,
resource_owner_id: UUID | None,
caller_team_ids: set[UUID],
) -> bool:
"""In-memory variant of ``_user_can_see_share`` for the batch list path.
Same rules as ``_user_can_see_share`` but team membership is checked via
the pre-fetched ``caller_team_ids`` set so the list loop is O(N) instead
of issuing one ``select(AuthzTeamMember)`` per row.
"""
if user_id in {resource_owner_id, row.created_by}:
return True
scope = row.scope
if scope == ShareScope.PUBLIC.value:
return True
if scope == ShareScope.USER.value and row.target_id == user_id:
return True
if scope == ShareScope.TEAM.value and row.target_id is not None:
return row.target_id in caller_team_ids
return False
@router.get("/{share_id}", response_model=ShareRead)
async def get_share(
share_id: UUID,
@ -363,7 +414,18 @@ async def update_share(
detail=f"Unknown permission_level {payload.permission_level!r}",
) from exc
session.add(row)
await session.flush()
# Match the create_share error contract: rollback + fixed-string 409 +
# server-side log so a constraint violation on update doesn't return a
# raw SQLAlchemy stacktrace to the client.
try:
await session.flush()
except Exception as exc:
await session.rollback()
logger.warning("authz_share update rejected: %s", exc)
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Share could not be updated: it may conflict with an existing share.",
) from exc
await session.refresh(row)
await _invalidate_for_share(row.scope, row.target_id)

179
src/backend/base/langflow/api/v1/flows.py
View File

@ -195,14 +195,19 @@ async def read_flows(
# Apply the same authz filter the get_all branch uses so both modes of
# this endpoint behave consistently. OSS pass-through returns the page
# items unchanged; an enterprise plugin filters per-flow. ``page.total``
# may overcount denied items — a fully accurate count requires
# SQL-level prefiltering via authz_share (Phase 3 work).
# items unchanged; an enterprise plugin filters per-flow. The
# ``owner_extractor`` lets the helper short-circuit owner-allow without
# round-tripping through the enforcer — without it, an enterprise
# plugin lacking an explicit owner-allow policy would hide the caller's
# own flows from paginated results. ``page.total`` may overcount denied
# items — a fully accurate count requires SQL-level prefiltering via
# authz_share (Phase 3 work).
page.items = await filter_visible_resources(
current_user,
resource_type="flow",
candidates=list(page.items),
domain_extractor=lambda flow: _resolve_casbin_domain(flow.workspace_id, flow.folder_id),
owner_extractor=lambda flow: flow.user_id,
act=FlowAction.READ,
)
return page # noqa: TRY300 — final return inside try matches the existing style of this handler
@ -339,14 +344,17 @@ async def update_flow(
target_workspace_id = flow.workspace_id if flow.workspace_id is not None else db_flow.workspace_id
target_folder_id = flow.folder_id if flow.folder_id is not None else db_flow.folder_id
if target_workspace_id != db_flow.workspace_id or target_folder_id != db_flow.folder_id:
await ensure_flow_permission(
current_user,
FlowAction.WRITE,
flow_id=flow_id,
flow_user_id=db_flow.user_id,
workspace_id=target_workspace_id,
folder_id=target_folder_id,
)
try:
await ensure_flow_permission(
current_user,
FlowAction.WRITE,
flow_id=flow_id,
flow_user_id=db_flow.user_id,
workspace_id=target_workspace_id,
folder_id=target_folder_id,
)
except HTTPException as exc:
raise deny_to_404(exc, detail="Flow not found") from exc
# Explicit folder_id=None is ignored here because _patch_flow builds
# update_data with exclude_none=True, so null folder_id is a no-op.
@ -359,6 +367,41 @@ async def update_flow(
db_flow_for_attempt = await _read_flow(session=session, flow_id=flow_id, user_id=current_user.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
# different workspace/folder between the destination check above
# and this retry attempt. Re-authorize against the freshly
# reloaded source AND destination so the writer cannot ride a
# stale check across a race.
try:
await ensure_flow_permission(
current_user,
FlowAction.WRITE,
flow_id=flow_id,
flow_user_id=db_flow_for_attempt.user_id,
workspace_id=db_flow_for_attempt.workspace_id,
folder_id=db_flow_for_attempt.folder_id,
)
except HTTPException as exc:
raise deny_to_404(exc, detail="Flow not found") from exc
attempt_target_workspace_id = (
flow.workspace_id if flow.workspace_id is not None else db_flow_for_attempt.workspace_id
)
attempt_target_folder_id = flow.folder_id if flow.folder_id is not None else db_flow_for_attempt.folder_id
if (
attempt_target_workspace_id != db_flow_for_attempt.workspace_id
or attempt_target_folder_id != db_flow_for_attempt.folder_id
):
try:
await ensure_flow_permission(
current_user,
FlowAction.WRITE,
flow_id=flow_id,
flow_user_id=db_flow_for_attempt.user_id,
workspace_id=attempt_target_workspace_id,
folder_id=attempt_target_folder_id,
)
except HTTPException as exc:
raise deny_to_404(exc, detail="Flow not found") from exc
return await _patch_flow(
session=session,
db_flow=db_flow_for_attempt,
@ -405,18 +448,33 @@ async def upsert_flow(
existing_flow = (await session.exec(select(Flow).where(Flow.id == flow_id))).first()
if existing_flow is not None:
# Flow exists - check ownership (return 404 to avoid leaking resource existence)
if existing_flow.user_id != current_user.id:
# OSS floor: when the registered service cannot widen fetch (OSS
# pass-through, or any path where AUTHZ_ENABLED is off), non-owners
# must not be able to upsert by id. Without this guard, the
# ``Flow.id == flow_id``-only fetch above would let a non-owner
# overwrite another user's flow because ``ensure_flow_permission``
# is a no-op when AUTHZ_ENABLED=false. With an enterprise plugin
# registered, ``ensure_flow_permission`` below is authoritative —
# a valid WRITE share grant must be honored, and a deny converts
# to 404 to preserve UUID privacy.
from langflow.services.deps import get_authorization_service
authz = get_authorization_service()
can_widen = await authz.supports_cross_user_fetch() and await authz.is_enabled()
if not can_widen and existing_flow.user_id != current_user.id:
raise HTTPException(status_code=404, detail="Flow not found")
await ensure_flow_permission(
current_user,
FlowAction.WRITE,
flow_id=flow_id,
flow_user_id=existing_flow.user_id,
workspace_id=existing_flow.workspace_id,
folder_id=existing_flow.folder_id,
)
try:
await ensure_flow_permission(
current_user,
FlowAction.WRITE,
flow_id=flow_id,
flow_user_id=existing_flow.user_id,
workspace_id=existing_flow.workspace_id,
folder_id=existing_flow.folder_id,
)
except HTTPException as exc:
raise deny_to_404(exc, detail="Flow not found") from exc
# Destination check (see update_flow above): if the payload moves
# the flow into a new workspace/folder, also authorize WRITE at the
@ -426,14 +484,17 @@ async def upsert_flow(
target_workspace_id = flow.workspace_id if flow.workspace_id is not None else existing_flow.workspace_id
target_folder_id = flow.folder_id if flow.folder_id is not None else existing_flow.folder_id
if target_workspace_id != existing_flow.workspace_id or target_folder_id != existing_flow.folder_id:
await ensure_flow_permission(
current_user,
FlowAction.WRITE,
flow_id=flow_id,
flow_user_id=existing_flow.user_id,
workspace_id=target_workspace_id,
folder_id=target_folder_id,
)
try:
await ensure_flow_permission(
current_user,
FlowAction.WRITE,
flow_id=flow_id,
flow_user_id=existing_flow.user_id,
workspace_id=target_workspace_id,
folder_id=target_folder_id,
)
except HTTPException as exc:
raise deny_to_404(exc, detail="Flow not found") from exc
# Sync deployment state before folder changes
# Explicit folder_id=None is ignored here because _update_existing_flow
@ -687,10 +748,27 @@ async def delete_multiple_flows(
async def _delete_operation() -> int:
if not flow_ids:
return 0
flows_to_delete = (
await db.exec(select(Flow).where(col(Flow.id).in_(flow_ids)).where(Flow.user_id == user.id))
).all()
# Widen the fetch when an enterprise plugin can authorize cross-user
# DELETE via share grants. Without this, pre-scoping by
# ``Flow.user_id == user.id`` silently drops non-owned flows from
# the working set so a valid DELETE share is never honored and
# ``ensure_flow_permission`` never emits a deny audit row. In OSS
# default the query stays owner-scoped — the no-op pass-through
# cannot widen visibility.
from langflow.services.deps import get_authorization_service
authz = get_authorization_service()
base_stmt = select(Flow).where(col(Flow.id).in_(flow_ids))
if await authz.supports_cross_user_fetch() and await authz.is_enabled():
stmt = base_stmt
else:
stmt = base_stmt.where(Flow.user_id == user.id)
flows_to_delete = (await db.exec(stmt)).all()
for flow in flows_to_delete:
# A deny from the enterprise plugin raises 403. We let it
# propagate so the audit row is written and the caller knows
# the bulk delete was rejected for at least one flow — the
# alternative (silent skip) would mask policy violations.
await ensure_flow_permission(
user,
FlowAction.DELETE,
@ -733,20 +811,39 @@ async def download_multiple_file(
# TODO: Full-version download (include_version parameter) is planned as a follow-up feature.
# When implemented, add an include_version: bool = False parameter and embed version
# entries in each flow dict using get_flow_versions_with_provider_status and strip_version_data.
flows = (await db.exec(select(Flow).where(and_(Flow.user_id == user.id, Flow.id.in_(flow_ids))))).all() # type: ignore[attr-defined]
# Widen the fetch when the enterprise plugin can authorize cross-user READ
# via share grants. Pre-scoping by ``Flow.user_id == user.id`` here would
# silently drop shared flows from the working set so a valid READ share is
# never honored and no audit row is written. OSS default keeps the query
# owner-scoped so the no-op pass-through cannot widen visibility.
from langflow.services.deps import get_authorization_service
authz = get_authorization_service()
base_stmt = select(Flow).where(col(Flow.id).in_(flow_ids)) # type: ignore[attr-defined]
if await authz.supports_cross_user_fetch() and await authz.is_enabled():
stmt = base_stmt
else:
stmt = base_stmt.where(and_(Flow.user_id == user.id))
flows = (await db.exec(stmt)).all()
if not flows:
raise HTTPException(status_code=404, detail="No flows found.")
for flow in flows:
await ensure_flow_permission(
user,
FlowAction.READ,
flow_id=flow.id,
flow_user_id=flow.user_id,
workspace_id=flow.workspace_id,
folder_id=flow.folder_id,
)
# Convert deny to 404 to preserve UUID privacy — a caller probing
# arbitrary IDs cannot distinguish "doesn't exist" from "exists but
# not shared with me".
try:
await ensure_flow_permission(
user,
FlowAction.READ,
flow_id=flow.id,
flow_user_id=flow.user_id,
workspace_id=flow.workspace_id,
folder_id=flow.folder_id,
)
except HTTPException as exc:
raise deny_to_404(exc, detail="No flows found.") from exc
return _build_flows_download_response(flows)

7
src/backend/base/langflow/api/v1/knowledge_bases.py
View File

@ -167,6 +167,13 @@ async def _guard_kb_action(
# and we skip the DB roundtrip.
if kb_user_id == current_user.id:
return _KbGuardResult(record=resolved_record, owner_user=current_user)
# ``kb_user_id`` comes from a SQLAlchemy column in production — always a
# UUID. Defensive guard: skip the DB lookup if the value isn't a UUID
# (e.g. a test mock where ``record.user_id`` is an auto-generated
# ``MagicMock``). Without this, the bind fails with
# ``Error binding parameter 1: type 'MagicMock' is not supported``.
if not isinstance(kb_user_id, uuid.UUID):
return _KbGuardResult(record=resolved_record, owner_user=current_user)
async with session_scope() as session:
owner = await get_user_by_id(session, kb_user_id)
if owner is None:

47
src/backend/base/langflow/api/v1/projects.py
View File

@ -32,7 +32,12 @@ from langflow.api.v1.projects_mcp_helpers import (
)
from langflow.initial_setup.constants import ASSISTANT_FOLDER_NAME, STARTER_FOLDER_NAME
from langflow.services.auth.mcp_encryption import encrypt_auth_settings
from langflow.services.authorization import ProjectAction, ensure_project_permission, filter_visible_resources
from langflow.services.authorization import (
FlowAction,
ProjectAction,
ensure_project_permission,
filter_visible_resources,
)
from langflow.services.authorization.fetch import authorized_or_owner_scoped, deny_to_404
from langflow.services.authorization.utils import _resolve_casbin_domain
from langflow.services.database.models.deployment.exceptions import (
@ -320,7 +325,20 @@ async def read_project(
# If no pagination requested, return flows visible to the caller.
if treat_as_shared:
visible_flows = list(project.flows)
# A project share grant implies access to the project itself, but
# per-flow policy (deny rules, lower scopes) still applies. Without
# this call, ``list(project.flows)`` would leak every flow in the
# project regardless of finer-grained Casbin rules the plugin may
# have. OSS pass-through returns the input list unchanged, so this
# has no effect on default OSS installs.
visible_flows = await filter_visible_resources(
current_user,
resource_type="flow",
candidates=list(project.flows),
domain_extractor=lambda flow: _resolve_casbin_domain(flow.workspace_id, flow.folder_id),
owner_extractor=lambda flow: flow.user_id,
act=FlowAction.READ,
)
else:
visible_flows = [flow for flow in project.flows if flow.user_id == current_user.id]
project.flows = visible_flows
@ -468,12 +486,19 @@ async def update_project(
flow_ids_for_sync = list(dict.fromkeys(excluded_flows + concat_project_components))
async def _move_flows_for_project_update() -> None:
# Both SELECT and UPDATE must scope to the project owner — a
# non-owner editing a shared project must touch the *owner's*
# flows, not the actor's. The previous code filtered the SELECT
# by ``current_user.id`` (returning zero rows for non-owners) but
# then ran an UPDATE without any owner filter, so an id collision
# would have moved cross-user flows without per-flow authz.
# Scoping both statements to ``project_owner_id`` closes the gap.
if my_collection_project:
excluded_flow_rows = (
await session.exec(
select(Flow.id, Flow.folder_id).where(
Flow.id.in_(excluded_flows), # type: ignore[attr-defined]
Flow.user_id == current_user.id,
Flow.user_id == project_owner_id,
)
)
).all()
@ -483,7 +508,12 @@ async def update_project(
new_folder_id=my_collection_project.id,
)
update_statement_my_collection = (
update(Flow).where(Flow.id.in_(excluded_flows)).values(folder_id=my_collection_project.id) # type: ignore[attr-defined]
update(Flow)
.where(
Flow.id.in_(excluded_flows), # type: ignore[attr-defined]
Flow.user_id == project_owner_id,
)
.values(folder_id=my_collection_project.id)
)
await session.exec(update_statement_my_collection)
@ -492,7 +522,7 @@ async def update_project(
await session.exec(
select(Flow.id, Flow.folder_id).where(
Flow.id.in_(concat_project_components), # type: ignore[attr-defined]
Flow.user_id == current_user.id,
Flow.user_id == project_owner_id,
)
)
).all()
@ -502,7 +532,12 @@ async def update_project(
new_folder_id=existing_project.id,
)
update_statement_components = (
update(Flow).where(Flow.id.in_(concat_project_components)).values(folder_id=existing_project.id) # type: ignore[attr-defined]
update(Flow)
.where(
Flow.id.in_(concat_project_components), # type: ignore[attr-defined]
Flow.user_id == project_owner_id,
)
.values(folder_id=existing_project.id)
)
await session.exec(update_statement_components)

6
src/backend/base/langflow/api/v1/schemas/authz_shares.py
View File

@ -71,7 +71,11 @@ class ShareRead(BaseModel):
"""Read-only projection of an ``authz_share`` row."""
id: UUID
resource_type: str
# Use the same Literal as ``ShareCreate.resource_type`` so OpenAPI clients
# see a stable enum on both sides of the round trip. A bare ``str`` here
# would let a row inserted by a future code path with a typoed
# ``resource_type`` round-trip back unvalidated.
resource_type: ShareResourceType
resource_id: UUID
scope: ShareScopeLiteral
target_id: UUID | None

65
src/backend/base/langflow/helpers/flow.py
View File

@ -166,9 +166,30 @@ async def get_flow_by_id_or_name(
async def load_flow(
user_id: str, flow_id: str | None = None, flow_name: str | None = None, tweaks: dict | None = None
) -> Graph:
"""Load a flow as a Graph after authorizing the caller.
``user_id`` is the caller (the user executing the graph), and ``flow_id``
identifies the flow to load. This function is reachable from custom
components, sub-flow runners, and ``run_flow`` — anywhere a user-supplied
``flow_id`` might end up. We must authorize EXECUTE here so a malicious
component or webhook cannot pass an arbitrary id and pull another user's
flow definition (which contains the flow author's prompts, tools, and
embedded credentials).
OSS default (``AUTHZ_ENABLED=false`` or no enterprise plugin): the fetch
stays owner-scoped, so a non-owner ``flow_id`` returns ``None`` and the
function raises ``ValueError`` exactly as before.
Enterprise (``AUTHZ_ENABLED=true`` and ``supports_cross_user_fetch=True``):
the fetch widens to id-only and ``ensure_flow_permission(EXECUTE)`` decides
access via the plugin's policy.
"""
from lfx.graph.graph.base import Graph
from langflow.processing.process import process_tweaks
from langflow.services.authorization import FlowAction, ensure_flow_permission
from langflow.services.authorization.fetch import authorized_or_owner_scoped
from langflow.services.database.models.user.model import User
if not flow_id and not flow_name:
msg = "Flow ID or Flow Name is required"
@ -179,8 +200,50 @@ async def load_flow(
msg = f"Flow {flow_name} not found"
raise ValueError(msg)
uuid_user_id = UUID(user_id) if isinstance(user_id, str) else user_id
uuid_flow_id = UUID(flow_id) if isinstance(flow_id, str) else flow_id
async with session_scope() as session:
graph_data = flow.data if (flow := await session.get(Flow, flow_id)) else None
flow = await authorized_or_owner_scoped(
session,
Flow,
id_column=Flow.id,
resource_id=uuid_flow_id,
owner_column=Flow.user_id,
owner_id=uuid_user_id,
)
if flow is None:
msg = f"Flow {flow_id} not found"
raise ValueError(msg)
# Authorize EXECUTE. ``ensure_flow_permission`` short-circuits on owner
# override, so the OSS pass-through path is unchanged. With an
# enterprise plugin a deny raises ``HTTPException(403)`` which we
# translate back to ``ValueError`` so callers (component execution,
# ``run_flow``) keep the existing exception contract.
caller = await session.get(User, uuid_user_id)
if caller is None:
msg = "Session is invalid"
raise ValueError(msg)
try:
await ensure_flow_permission(
caller,
FlowAction.EXECUTE,
flow_id=flow.id,
flow_user_id=flow.user_id,
workspace_id=flow.workspace_id,
folder_id=flow.folder_id,
)
except HTTPException as exc:
from fastapi import status as http_status
if exc.status_code == http_status.HTTP_403_FORBIDDEN:
msg = f"Flow {flow_id} not found"
raise ValueError(msg) from exc
raise
graph_data = flow.data
if not graph_data:
msg = f"Flow {flow_id} not found"
raise ValueError(msg)

10
src/backend/base/langflow/main.py
View File

@ -551,6 +551,16 @@ def get_lifespan(*, fix_migration=False, version=None):
# Step 2: Cleaning Up Services
with shutdown_progress.step(2):
# Drain pending audit writes before services tear down so
# rows scheduled mid-request still land in the DB. We do
# this here (not in teardown_services) because the DB
# session factory must still be alive.
try:
from langflow.services.authorization.utils import drain_pending_audit_writes
await drain_pending_audit_writes(timeout=5.0)
except Exception as drain_exc: # noqa: BLE001 — never block shutdown on audit
await logger.awarning(f"drain_pending_audit_writes failed: {drain_exc}")
try:
await asyncio.wait_for(teardown_services(), timeout=30)
except asyncio.TimeoutError:

28
src/backend/base/langflow/services/authorization/fetch.py
View File

@ -87,20 +87,18 @@ async def authorized_or_owner_scoped(
def deny_to_404(exc: HTTPException, detail: str = "Not found") -> HTTPException:
"""Convert a 403 from ``ensure_*_permission`` into a 404 for UUID privacy.
Re-raises any non-403 exception untouched. Callers wrap their guard call:
For 403s, returns a fresh ``HTTPException(404, detail=detail)``.
.. code-block:: python
try:
await ensure_flow_permission(user, FlowAction.READ, flow_id=flow.id, ...)
except HTTPException as exc:
raise deny_to_404(exc) from exc
Rationale: a route that was reachable purely through a share grant must
not leak the resource's existence to callers who can no longer reach it
(e.g. after the share is revoked). 403 confirms the row exists; 404 does
not.
For non-403s, returns the original exception with its ``detail`` replaced
by the caller-supplied ``detail`` and any ``headers`` dropped. This
sanitisation is conservative on purpose: helper callers are guard sites,
so the exception about to propagate has been filtered through an authz
plugin whose detail string may quote resource UUIDs, owner ids, or
policy tuples. Letting that detail land in the client response would
leak information about resources the caller is not authorised to see.
Routes that need to expose a richer detail (e.g. a 409 with a real
conflict message) should catch and re-raise outside ``deny_to_404``.
"""
if exc.status_code != status.HTTP_403_FORBIDDEN:
return exc
return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=detail)
if exc.status_code == status.HTTP_403_FORBIDDEN:
return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=detail)
return HTTPException(status_code=exc.status_code, detail=detail)

112
src/backend/base/langflow/services/authorization/utils.py
View File

@ -117,6 +117,35 @@ def _split_obj(obj: str) -> tuple[str | None, UUID | None]:
return resource_type, None
# In-flight audit tasks. Keeping a strong reference prevents the event loop
# from garbage-collecting a pending task before it writes — and gives
# ``drain_pending_audit_writes`` (called on shutdown) something to await.
# A bare ``asyncio.create_task`` without this reference can be dropped by
# the GC mid-write and silently lose the audit row.
_pending_audit_tasks: set[asyncio.Task[None]] = set()
async def drain_pending_audit_writes(timeout: float = 5.0) -> None:
"""Wait for any in-flight audit-write tasks to complete.
Called on application shutdown so audit rows scheduled mid-request still
land in the DB before the event loop closes. Returns silently after
``timeout`` even if some tasks are still pending — audit must never block
shutdown indefinitely.
"""
pending = {task for task in _pending_audit_tasks if not task.done()}
if not pending:
return
done, still_pending = await asyncio.wait(pending, timeout=timeout)
if still_pending:
logger.warning("drain_pending_audit_writes timed out with %d pending tasks", len(still_pending))
# Surface task exceptions in logs but never raise — audit must not block shutdown.
for task in done:
exc = task.exception() if not task.cancelled() else None
if exc is not None:
logger.warning("Audit task raised during drain: %s", exc)
async def audit_decision(
*,
user_id: UUID | None,
@ -129,11 +158,19 @@ async def audit_decision(
Schedules an asyncio task and returns immediately. Failures inside the task
are logged but never propagate to the caller so audit writes can never
block a real API response.
block a real API response. The task reference is held in
``_pending_audit_tasks`` until completion so the event loop cannot GC it
mid-write; ``drain_pending_audit_writes`` awaits this set on shutdown so
pending rows are not lost.
"""
settings = get_settings_service()
auth_settings = settings.auth_settings
if not auth_settings.AUTHZ_ENABLED or not getattr(auth_settings, "AUTHZ_AUDIT_ENABLED", True):
# Audit is independent of enforcement: an operator can keep audit on
# while AUTHZ_ENABLED=false to observe traffic before flipping the
# flag. Previously this short-circuited on AUTHZ_ENABLED too, which
# meant share CRUD writes (which the OSS floor allows in default mode)
# produced no audit trail at all.
if not getattr(auth_settings, "AUTHZ_AUDIT_ENABLED", True):
return
resource_type, resource_id = _split_obj(obj)
@ -157,8 +194,9 @@ async def audit_decision(
except Exception: # noqa: BLE001 — audit must never raise into the request path
logger.exception("Failed to write AuthzAuditLog row")
# Bare create_task is the langflow convention (see main.py:135, main.py:173).
asyncio.create_task(_write()) # noqa: RUF006 — fire-and-forget audit task
task = asyncio.create_task(_write())
_pending_audit_tasks.add(task)
task.add_done_callback(_pending_audit_tasks.discard)
async def ensure_permission(
@ -181,21 +219,41 @@ async def ensure_permission(
# User-derived auth fields (e.g. is_superuser) must remain authoritative; caller
# context is merged first so it cannot overwrite them.
merged_context = {**(context or {}), **_auth_context(user)}
allowed = await authz.enforce(
user_id=user.id,
domain=domain,
obj=obj,
act=act,
context=merged_context,
)
# Fail-closed contract: if the plugin's ``enforce`` raises (Casbin DB
# down, policy parse error, etc.), we treat the request as denied and
# log an explicit error-audit row. Returning False from ``enforce`` is
# the plugin's documented way to deny; raising should not silently
# become an HTTP 500 that bypasses both the deny path and the audit log.
audit_action = f"{obj.split(':', 1)[0]}:{act}" if ":" in obj else act
audit_details = {"domain": domain}
for owner_key in _OWNER_CONTEXT_KEYS:
if owner_key in merged_context and merged_context[owner_key] is not None:
audit_details[owner_key] = str(merged_context[owner_key])
try:
allowed = await authz.enforce(
user_id=user.id,
domain=domain,
obj=obj,
act=act,
context=merged_context,
)
except Exception as exc:
logger.exception("Authorization plugin raised during enforce; failing closed")
await audit_decision(
user_id=user.id,
action=audit_action,
obj=obj,
result=_AUDIT_DENY,
details={**audit_details, "error": str(exc)},
)
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Insufficient permissions to {act} on {obj}",
) from exc
await audit_decision(
user_id=user.id,
action=f"{obj.split(':', 1)[0]}:{act}" if ":" in obj else act,
action=audit_action,
obj=obj,
result=_AUDIT_ALLOW if allowed else _AUDIT_DENY,
details=audit_details,
@ -265,15 +323,16 @@ async def _ensure_resource_permission(
# Owner override: a resource owner can always operate on their own resource.
if owner_id is not None and getattr(user, "id", None) == owner_id:
settings = get_settings_service()
if settings.auth_settings.AUTHZ_ENABLED:
await audit_decision(
user_id=user.id,
action=f"{resource_type}:{act_str}",
obj=obj,
result=_AUDIT_OWNER_OVERRIDE,
details={"domain": resolved_domain},
)
# ``audit_decision`` is gated on ``AUTHZ_AUDIT_ENABLED`` internally;
# we no longer suppress the call on ``AUTHZ_ENABLED=false`` so
# operators observing pre-enforcement traffic see owner overrides too.
await audit_decision(
user_id=user.id,
action=f"{resource_type}:{act_str}",
obj=obj,
result=_AUDIT_OWNER_OVERRIDE,
details={"domain": resolved_domain},
)
return
await ensure_permission(
@ -556,11 +615,16 @@ async def filter_visible_resources(
decisions[index] = True
if enforce_items:
# Use the already-extracted ``user_id`` instead of re-reading
# ``user.id`` here. The earlier owner-override partition used
# ``getattr(user, "id", None)``, so consistency means a non-User
# dependency (e.g. a thin context proxy) lands an explicit None
# rather than an AttributeError at this call site.
if domain_extractor is None:
# Single-domain fast path. Preserves the original single-call shape.
requests = [(f"{resource_type}:{extractor(item)}", act_str) for item in enforce_items]
results = await authz.batch_enforce(
user_id=user.id,
user_id=user_id,
domain=domain,
requests=requests,
context=_auth_context(user),
@ -580,7 +644,7 @@ async def filter_visible_resources(
for resolved_domain, bucket in buckets.items():
bucket_requests = [(f"{resource_type}:{extractor(item)}", act_str) for _, item in bucket]
bucket_results = await authz.batch_enforce(
user_id=user.id,
user_id=user_id,
domain=resolved_domain,
requests=bucket_requests,
context=auth_context,

36
src/backend/base/langflow/services/database/models/auth/authz.py
View File

@ -129,6 +129,15 @@ class AuthzRoleAssignment(SQLModel, table=True): # type: ignore[call-arg]
postgresql_where=text("domain_id IS NULL"),
sqlite_where=text("domain_id IS NULL"),
),
# Hot-path lookup: "all assignments for user X scoped to a domain"
# — every enterprise enforce() call. Composite avoids two non-covering
# single-column scans.
Index(
"ix_authz_role_assignment_user_domain",
"user_id",
"domain_type",
"domain_id",
),
)
id: UUIDstr = Field(default_factory=uuid4, primary_key=True)
@ -139,7 +148,14 @@ class AuthzRoleAssignment(SQLModel, table=True): # type: ignore[call-arg]
sa_column=Column(sa.Uuid(), ForeignKey("authz_role.id", ondelete="CASCADE"), nullable=False, index=True),
)
domain_type: str = Field(default="global", description="global, org, workspace")
domain_id: UUIDstr | None = Field(default=None, index=True)
# Explicit ``sa_column`` so SQLModel emits ``sa.Uuid()`` matching the
# migration's column type. Without this, SQLModel can fall back to
# ``AutoString``/``CHAR(32)`` on SQLite, which drifts from the migration
# and confuses downstream type introspection.
domain_id: UUIDstr | None = Field(
default=None,
sa_column=Column(sa.Uuid(), nullable=True, index=True),
)
assigned_at: datetime = Field(default_factory=_tz_aware_now, sa_column=_tz_column())
assigned_by: UUIDstr | None = Field(
default=None,
@ -201,11 +217,11 @@ class AuthzShare(SQLModel, table=True): # type: ignore[call-arg]
# partial unique indexes (which match on the lowercase form).
CheckConstraint(
"scope IN ('private', 'team', 'user', 'public')",
name="scope_enum",
name="ck_authz_share_scope_enum",
),
CheckConstraint(
"permission_level IN ('read', 'write', 'execute', 'admin')",
name="permission_enum",
name="ck_authz_share_permission_enum",
),
# Targeted (TEAM/USER) shares require a target_id; untargeted
# (PRIVATE/PUBLIC) shares forbid one. Matches the partial-unique-index
@ -214,7 +230,7 @@ class AuthzShare(SQLModel, table=True): # type: ignore[call-arg]
CheckConstraint(
"(scope IN ('team', 'user') AND target_id IS NOT NULL) "
"OR (scope IN ('private', 'public') AND target_id IS NULL)",
name="scope_target_consistency",
name="ck_authz_share_scope_target_consistency",
),
Index(
"uq_authz_share_targeted",
@ -263,7 +279,9 @@ class AuthzEditLock(SQLModel, table=True): # type: ignore[call-arg]
sa_column=Column(sa.Uuid(), ForeignKey("user.id", ondelete="CASCADE"), nullable=False),
)
acquired_at: datetime = Field(default_factory=_tz_aware_now, sa_column=_tz_column())
expires_at: datetime = Field(sa_column=_tz_column())
# ``index=True`` matches the migration's ``ix_authz_edit_lock_expires_at``
# so the expired-lock sweeper can do an index seek instead of a table scan.
expires_at: datetime = Field(sa_column=_tz_column(index=True))
# TODO: AuthzAuditLog is append-only and unbounded. At enterprise scale this
@ -279,6 +297,14 @@ class AuthzAuditLog(SQLModel, table=True): # type: ignore[call-arg]
__table_args__ = (
Index("ix_authz_audit_log_user_timestamp", "user_id", "timestamp"),
Index("ix_authz_audit_log_resource", "resource_type", "resource_id"),
# ``owner_override`` is the third value the framework writes (see
# ``_AUDIT_OWNER_OVERRIDE`` in services/authorization/utils.py); it
# must be in the CHECK set or owner-shortcut audit rows would
# silently fail the constraint.
CheckConstraint(
"result IN ('allow', 'deny', 'owner_override')",
name="ck_authz_audit_log_result_enum",
),
)
id: UUIDstr = Field(default_factory=uuid4, primary_key=True)

376
src/backend/tests/unit/api/v1/test_authz_share_routes.py Normal file
View File

@ -0,0 +1,376 @@
"""Route-level tests for the OSS share-administration floor.
These tests assert the documented OSS contract: only the resource owner or a
superuser may administer ``authz_share`` rows for a resource when the
authorization service is the pass-through default (no enterprise plugin). The
schema-level happy path is covered in ``test_authz_share_schemas.py``; what
these tests prevent is the floor being silently dropped on a refactor.
We exercise the handlers directly (not via the FastAPI app) using a fake async
session that records writes. This keeps the test fast and isolates the floor
logic from DB and Casbin concerns.
"""
from __future__ import annotations
from types import SimpleNamespace
from typing import Any
from uuid import UUID, uuid4
import pytest
from fastapi import HTTPException
from langflow.api.v1 import authz_shares as shares_module
from langflow.api.v1.schemas.authz_shares import ShareCreate, ShareUpdate
from langflow.services.database.models.auth import AuthzShare, SharePermissionLevel, ShareScope
class _FakeAsyncSession:
"""Minimal async-session stand-in: stores get() results and records writes."""
def __init__(self, get_by_type: dict[tuple[type, UUID], Any] | None = None) -> None:
self._get_by_type = get_by_type or {}
self.added: list[Any] = []
self.deleted: list[Any] = []
self.flushed = 0
self.rolled_back = 0
async def get(self, model: type, key: UUID) -> Any:
return self._get_by_type.get((model, key))
def add(self, obj: Any) -> None:
self.added.append(obj)
async def delete(self, obj: Any) -> None:
self.deleted.append(obj)
async def flush(self) -> None:
self.flushed += 1
async def refresh(self, obj: Any) -> None: # noqa: ARG002
return None
async def rollback(self) -> None:
self.rolled_back += 1
async def exec(self, _stmt: Any) -> list[Any]:
# list_shares is not exercised here; return empty by default.
return []
class _StubAuthz:
"""Pass-through authz service: allow everything, no cross-user fetch."""
def __init__(self, *, cross_user: bool = False, enabled: bool = False) -> None:
self._cross_user = cross_user
self._enabled = enabled
async def supports_cross_user_fetch(self) -> bool:
return self._cross_user
async def is_enabled(self) -> bool:
return self._enabled
async def enforce(self, **_kwargs) -> bool:
return True
async def batch_enforce(self, **kwargs) -> list[bool]:
return [True] * len(kwargs.get("requests", []))
async def invalidate_user(self, *_args, **_kwargs) -> None:
return None
async def invalidate_all(self, *_args, **_kwargs) -> None:
return None
@pytest.fixture
def patch_authz(monkeypatch):
"""Install a stub authz service into both the shares module and utils."""
from langflow.services.authorization import utils as authz_utils
def _apply(*, cross_user: bool = False, enabled: bool = False) -> _StubAuthz:
stub = _StubAuthz(cross_user=cross_user, enabled=enabled)
monkeypatch.setattr(shares_module, "get_authorization_service", lambda: stub)
monkeypatch.setattr(authz_utils, "get_authorization_service", lambda: stub)
# Disable audit + AUTHZ_ENABLED gating so audit calls inside ensure_*
# don't open real sessions.
settings = SimpleNamespace(
auth_settings=SimpleNamespace(AUTHZ_ENABLED=False, AUTHZ_AUDIT_ENABLED=False),
)
monkeypatch.setattr(authz_utils, "get_settings_service", lambda: settings)
return stub
return _apply
@pytest.fixture
def silence_audit(monkeypatch):
"""Replace audit_decision with a no-op so we don't spawn background tasks."""
async def _noop(**_kwargs):
return None
monkeypatch.setattr(shares_module, "audit_decision", _noop)
def _make_user(*, is_superuser: bool = False) -> SimpleNamespace:
return SimpleNamespace(id=uuid4(), is_superuser=is_superuser, username="u")
def _make_flow_owned_by(owner_id: UUID) -> Any:
from langflow.services.database.models.flow.model import Flow
return SimpleNamespace(_model=Flow, id=uuid4(), user_id=owner_id)
def _payload_for(resource_id: UUID) -> ShareCreate:
return ShareCreate(
resource_type="flow",
resource_id=resource_id,
scope=ShareScope.USER.value,
target_id=uuid4(),
permission_level=SharePermissionLevel.READ.value,
)
# --------------------------------------------------------------------------- #
# CREATE — OSS floor must block non-owner / non-superuser
# --------------------------------------------------------------------------- #
@pytest.mark.asyncio
async def test_create_share_blocks_non_owner_under_oss_passthrough(patch_authz, silence_audit): # noqa: ARG001
"""A non-owner cannot mint a share row for another user's flow under OSS."""
from langflow.services.database.models.flow.model import Flow
patch_authz(cross_user=False, enabled=False)
owner = _make_user()
attacker = _make_user()
flow = SimpleNamespace(id=uuid4(), user_id=owner.id)
session = _FakeAsyncSession({(Flow, flow.id): flow})
payload = _payload_for(flow.id)
with pytest.raises(HTTPException) as excinfo:
await shares_module.create_share(payload=payload, current_user=attacker, session=session)
assert excinfo.value.status_code == 403
assert "Only the resource owner" in excinfo.value.detail
# Floor fires before any DB write — no share row was added.
assert session.added == []
@pytest.mark.asyncio
async def test_create_share_allows_owner_under_oss_passthrough(patch_authz, silence_audit): # noqa: ARG001
"""The resource owner can mint a share row under OSS pass-through."""
from langflow.services.database.models.flow.model import Flow
patch_authz(cross_user=False, enabled=False)
owner = _make_user()
flow = SimpleNamespace(id=uuid4(), user_id=owner.id)
session = _FakeAsyncSession({(Flow, flow.id): flow})
payload = _payload_for(flow.id)
result = await shares_module.create_share(payload=payload, current_user=owner, session=session)
assert result.resource_id == flow.id
assert len(session.added) == 1
assert session.flushed == 1
@pytest.mark.asyncio
async def test_create_share_allows_superuser_under_oss_passthrough(patch_authz, silence_audit): # noqa: ARG001
"""A superuser can mint a share row for a resource they don't own."""
from langflow.services.database.models.flow.model import Flow
patch_authz(cross_user=False, enabled=False)
owner = _make_user()
admin = _make_user(is_superuser=True)
flow = SimpleNamespace(id=uuid4(), user_id=owner.id)
session = _FakeAsyncSession({(Flow, flow.id): flow})
payload = _payload_for(flow.id)
result = await shares_module.create_share(payload=payload, current_user=admin, session=session)
assert result.resource_id == flow.id
assert len(session.added) == 1
@pytest.mark.asyncio
async def test_create_share_returns_404_when_resource_missing(patch_authz, silence_audit): # noqa: ARG001
"""A missing resource yields 404 — not 403 — to preserve UUID privacy."""
patch_authz(cross_user=False, enabled=False)
attacker = _make_user()
session = _FakeAsyncSession({}) # no resource present
payload = _payload_for(uuid4())
with pytest.raises(HTTPException) as excinfo:
await shares_module.create_share(payload=payload, current_user=attacker, session=session)
assert excinfo.value.status_code == 404
# --------------------------------------------------------------------------- #
# PATCH — same floor
# --------------------------------------------------------------------------- #
@pytest.mark.asyncio
async def test_update_share_blocks_non_owner_under_oss_passthrough(patch_authz, silence_audit): # noqa: ARG001
"""A non-owner cannot PATCH a share on another user's resource under OSS."""
from langflow.services.database.models.flow.model import Flow
patch_authz(cross_user=False, enabled=False)
owner = _make_user()
attacker = _make_user()
flow = SimpleNamespace(id=uuid4(), user_id=owner.id)
share = AuthzShare(
id=uuid4(),
resource_type="flow",
resource_id=flow.id,
scope=ShareScope.USER.value,
target_id=uuid4(),
permission_level=SharePermissionLevel.READ.value,
created_by=attacker.id, # attacker is even the creator — floor still blocks
)
session = _FakeAsyncSession({(AuthzShare, share.id): share, (Flow, flow.id): flow})
update = ShareUpdate(permission_level=SharePermissionLevel.WRITE.value)
with pytest.raises(HTTPException) as excinfo:
await shares_module.update_share(
share_id=share.id,
payload=update,
current_user=attacker,
session=session,
)
assert excinfo.value.status_code == 403
# PATCH was rejected — no flush should have happened.
assert session.flushed == 0
@pytest.mark.asyncio
async def test_update_share_allows_owner_under_oss_passthrough(patch_authz, silence_audit): # noqa: ARG001
"""The resource owner can PATCH a share on their resource under OSS."""
from langflow.services.database.models.flow.model import Flow
patch_authz(cross_user=False, enabled=False)
owner = _make_user()
flow = SimpleNamespace(id=uuid4(), user_id=owner.id)
share = AuthzShare(
id=uuid4(),
resource_type="flow",
resource_id=flow.id,
scope=ShareScope.USER.value,
target_id=uuid4(),
permission_level=SharePermissionLevel.READ.value,
created_by=owner.id,
)
session = _FakeAsyncSession({(AuthzShare, share.id): share, (Flow, flow.id): flow})
update = ShareUpdate(permission_level=SharePermissionLevel.WRITE.value)
result = await shares_module.update_share(
share_id=share.id,
payload=update,
current_user=owner,
session=session,
)
assert result.permission_level == SharePermissionLevel.WRITE.value
assert session.flushed == 1
# --------------------------------------------------------------------------- #
# DELETE — same floor
# --------------------------------------------------------------------------- #
@pytest.mark.asyncio
async def test_delete_share_blocks_non_owner_under_oss_passthrough(patch_authz, silence_audit): # noqa: ARG001
"""A non-owner cannot DELETE a share on another user's resource under OSS."""
from langflow.services.database.models.flow.model import Flow
patch_authz(cross_user=False, enabled=False)
owner = _make_user()
attacker = _make_user()
flow = SimpleNamespace(id=uuid4(), user_id=owner.id)
share = AuthzShare(
id=uuid4(),
resource_type="flow",
resource_id=flow.id,
scope=ShareScope.USER.value,
target_id=uuid4(),
permission_level=SharePermissionLevel.READ.value,
created_by=attacker.id, # attacker created it but is not the resource owner
)
session = _FakeAsyncSession({(AuthzShare, share.id): share, (Flow, flow.id): flow})
with pytest.raises(HTTPException) as excinfo:
await shares_module.delete_share(share_id=share.id, current_user=attacker, session=session)
assert excinfo.value.status_code == 403
# The floor blocks before the DELETE.
assert session.deleted == []
@pytest.mark.asyncio
async def test_delete_share_allows_owner_under_oss_passthrough(patch_authz, silence_audit): # noqa: ARG001
"""The resource owner can DELETE a share on their resource under OSS."""
from langflow.services.database.models.flow.model import Flow
patch_authz(cross_user=False, enabled=False)
owner = _make_user()
flow = SimpleNamespace(id=uuid4(), user_id=owner.id)
share = AuthzShare(
id=uuid4(),
resource_type="flow",
resource_id=flow.id,
scope=ShareScope.USER.value,
target_id=uuid4(),
permission_level=SharePermissionLevel.READ.value,
created_by=owner.id,
)
session = _FakeAsyncSession({(AuthzShare, share.id): share, (Flow, flow.id): flow})
await shares_module.delete_share(share_id=share.id, current_user=owner, session=session)
assert len(session.deleted) == 1
# --------------------------------------------------------------------------- #
# Floor behavior when the enterprise plugin is active
# --------------------------------------------------------------------------- #
@pytest.mark.asyncio
async def test_floor_is_skipped_when_enterprise_plugin_active(patch_authz, silence_audit): # noqa: ARG001
"""OSS floor is skipped when the enterprise plugin is actively enforcing.
When supports_cross_user_fetch=True AND AUTHZ_ENABLED=true, the OSS floor
is skipped so a plugin-granted share:create role can administer shares on
another user's resource. ``ensure_share_permission`` (mocked to allow here)
becomes the authoritative check.
"""
from langflow.services.database.models.flow.model import Flow
patch_authz(cross_user=True, enabled=True)
owner = _make_user()
delegate = _make_user() # non-owner, but allowed by enterprise policy
flow = SimpleNamespace(id=uuid4(), user_id=owner.id)
session = _FakeAsyncSession({(Flow, flow.id): flow})
payload = _payload_for(flow.id)
result = await shares_module.create_share(payload=payload, current_user=delegate, session=session)
# The floor is skipped; the stub authz allows the operation; the row is
# written. If the floor still fired we'd see a 403 instead.
assert result.resource_id == flow.id
assert len(session.added) == 1

142
src/backend/tests/unit/services/authorization/test_route_guard_regressions.py Normal file
View File

@ -0,0 +1,142 @@
"""Regression tests for route guard fixes from the OSS authorization PR review.
Source-level checks asserting that:
* The paginated ``read_flows`` branch passes ``owner_extractor`` to
``filter_visible_resources`` (so an enterprise enforcer cannot hide the
caller's own flows in paginated mode).
* ``upsert_flow`` consults ``supports_cross_user_fetch`` + ``is_enabled``
before falling back to the hardcoded ownership check, and wraps
``ensure_flow_permission`` with ``deny_to_404``.
* ``delete_multiple_flows`` and ``download_multiple_file`` do not pre-scope by
``Flow.user_id == user.id`` unconditionally; they gate that filter on the
same capability check.
* ``load_flow`` authorizes EXECUTE before returning a Graph.
"""
from __future__ import annotations
import ast
from pathlib import Path
import pytest
_API_V1 = Path(__file__).resolve().parents[4] / "base" / "langflow" / "api" / "v1"
_FLOWS_FILE = _API_V1 / "flows.py"
_HELPERS_FLOW = Path(__file__).resolve().parents[4] / "base" / "langflow" / "helpers" / "flow.py"
def _parse_async_funcs(path: Path) -> dict[str, ast.AsyncFunctionDef]:
tree = ast.parse(path.read_text())
return {node.name: node for node in ast.walk(tree) if isinstance(node, ast.AsyncFunctionDef)}
def _calls(func: ast.AsyncFunctionDef, name: str) -> list[ast.Call]:
out: list[ast.Call] = []
for node in ast.walk(func):
if not isinstance(node, ast.Call):
continue
target = node.func
is_named = isinstance(target, ast.Name) and target.id == name
is_attr = isinstance(target, ast.Attribute) and target.attr == name
if is_named or is_attr:
out.append(node)
return out
def _has_keyword(call: ast.Call, name: str) -> bool:
return any(kw.arg == name for kw in call.keywords)
@pytest.fixture(scope="module")
def flows_routes() -> dict[str, ast.AsyncFunctionDef]:
return _parse_async_funcs(_FLOWS_FILE)
@pytest.fixture(scope="module")
def helpers_funcs() -> dict[str, ast.AsyncFunctionDef]:
return _parse_async_funcs(_HELPERS_FLOW)
def test_read_flows_paginated_branch_passes_owner_extractor(flows_routes):
"""The paginated filter_visible_resources call must pass owner_extractor.
Both branches of read_flows (``get_all=True`` and pagination) must give
the enforcer the same hint about which flows are owner-owned. Omitting it
on the paginated branch lets an enterprise plugin without an explicit
owner-allow policy hide the caller's own flows when paginating.
"""
func = flows_routes["read_flows"]
fvr_calls = _calls(func, "filter_visible_resources")
assert len(fvr_calls) >= 2, "read_flows should call filter_visible_resources on both branches"
for call in fvr_calls:
assert _has_keyword(call, "owner_extractor"), (
"Every filter_visible_resources call in read_flows must pass owner_extractor "
"so the enforcer short-circuits on owner-allow"
)
def test_upsert_flow_consults_cross_user_fetch_capability(flows_routes):
"""upsert_flow must not unconditionally raise 404 for non-owners.
With an enterprise plugin registered, a valid WRITE share grant must be
honored. The current shape gates the hardcoded ownership floor on
``supports_cross_user_fetch() and is_enabled()``.
"""
func = flows_routes["upsert_flow"]
src = ast.unparse(func)
assert "supports_cross_user_fetch" in src, "upsert_flow must consult cross-user-fetch capability"
assert "is_enabled" in src, "upsert_flow must consult is_enabled before the OSS floor"
def test_upsert_flow_wraps_ensure_in_deny_to_404(flows_routes):
"""upsert_flow must convert ensure_flow_permission 403 to 404 for UUID privacy."""
func = flows_routes["upsert_flow"]
src = ast.unparse(func)
assert "deny_to_404" in src, "upsert_flow must wrap ensure_flow_permission with deny_to_404"
def test_delete_multiple_flows_does_not_unconditionally_prescope(flows_routes):
"""Bulk delete must not unconditionally filter by Flow.user_id == user.id.
The pre-scope is now gated on whether the registered service supports
cross-user fetch — otherwise valid share DELETE grants are silently
dropped from the working set.
"""
func = flows_routes["delete_multiple_flows"]
src = ast.unparse(func)
assert "supports_cross_user_fetch" in src, (
"delete_multiple_flows must gate the owner pre-scope on cross-user-fetch capability"
)
def test_download_multiple_file_does_not_unconditionally_prescope(flows_routes):
"""Same as above for the bulk download endpoint."""
func = flows_routes["download_multiple_file"]
src = ast.unparse(func)
assert "supports_cross_user_fetch" in src, (
"download_multiple_file must gate the owner pre-scope on cross-user-fetch capability"
)
assert "deny_to_404" in src, (
"download_multiple_file must convert ensure_flow_permission 403 to 404 for UUID privacy"
)
def test_load_flow_calls_ensure_flow_permission(helpers_funcs):
"""load_flow (component-reachable graph loader) must authorize EXECUTE.
Reachable from custom components, sub-flow runners, and ``run_flow``.
Without this guard, a caller can pass an arbitrary flow_id and pull
another user's flow definition (prompts, tools, embedded credentials).
"""
func = helpers_funcs["load_flow"]
calls = _calls(func, "ensure_flow_permission")
assert calls, "load_flow must call ensure_flow_permission"
# Action must be EXECUTE — the function loads a Graph for execution.
saw_execute = False
for call in calls:
if len(call.args) >= 2:
arg = call.args[1]
if isinstance(arg, ast.Attribute) and arg.attr == "EXECUTE":
saw_execute = True
assert saw_execute, "load_flow must authorize FlowAction.EXECUTE"

128
src/backend/tests/unit/services/authorization/test_utils.py
View File

@ -344,9 +344,15 @@ async def test_owner_override_skips_enforce(monkeypatch, fake_user):
@pytest.mark.anyio
async def test_owner_override_when_disabled_skips_audit(monkeypatch, fake_user):
"""Owner override on a disabled-authz install does not even write audit (consistency with ensure_permission)."""
_install_settings(monkeypatch, authz_enabled=False)
async def test_owner_override_audits_even_when_authz_disabled(monkeypatch, fake_user):
"""Owner override on a disabled-authz install still writes an audit row.
Audit is now gated only on ``AUTHZ_AUDIT_ENABLED`` (the audit recorder
here intercepts at that level), so operators can observe traffic ahead of
flipping ``AUTHZ_ENABLED``. ``_install_settings`` defaults the audit flag
to False so we explicitly enable it here.
"""
_install_settings(monkeypatch, authz_enabled=False, audit_enabled=True)
_install_authz(monkeypatch, _StubAuthorizationService(allow=False))
audit_calls = _install_audit_recorder(monkeypatch)
@ -356,7 +362,8 @@ async def test_owner_override_when_disabled_skips_audit(monkeypatch, fake_user):
flow_id=uuid4(),
flow_user_id=fake_user.id,
)
assert audit_calls == []
assert len(audit_calls) == 1
assert audit_calls[0]["result"] == "owner_override"
@pytest.mark.anyio
@ -477,14 +484,39 @@ def test_split_obj_non_uuid_suffix_returns_none_id():
@pytest.mark.anyio
async def test_audit_decision_noop_when_authz_disabled(monkeypatch):
"""audit_decision exits immediately when AUTHZ_ENABLED=False."""
async def test_audit_decision_runs_when_authz_disabled_but_audit_on(monkeypatch):
"""Audit is independent of enforcement now.
Previously ``audit_decision`` short-circuited when ``AUTHZ_ENABLED=False``,
which meant share CRUD writes left no audit trail on default installs. The
new contract gates only on ``AUTHZ_AUDIT_ENABLED`` so operators can
observe traffic before flipping enforcement on.
"""
_install_settings(monkeypatch, authz_enabled=False, audit_enabled=True)
scheduled: list[object] = []
monkeypatch.setattr("asyncio.create_task", lambda coro: scheduled.append(coro) or coro)
class _FakeTask:
def add_done_callback(self, cb):
self._cb = cb
def done(self) -> bool:
return False
def _capture(coro):
coro.close()
task = _FakeTask()
scheduled.append(task)
return task
monkeypatch.setattr("asyncio.create_task", _capture)
await authz_utils.audit_decision(user_id=uuid4(), action="flow:read", obj="flow:x", result="allow")
assert scheduled == []
try:
assert len(scheduled) == 1
finally:
# Don't pollute the module-global pending set for downstream tests.
for task in scheduled:
authz_utils._pending_audit_tasks.discard(task)
@pytest.mark.anyio
@ -500,18 +532,92 @@ async def test_audit_decision_noop_when_audit_disabled(monkeypatch):
@pytest.mark.anyio
async def test_audit_decision_schedules_task_when_enabled(monkeypatch):
"""A scheduled coroutine is produced when both flags are on."""
"""A scheduled coroutine is produced when both flags are on.
The new implementation tracks the task in ``_pending_audit_tasks`` so the
event loop cannot GC it mid-write and so shutdown can drain it. Our mock
returns a lightweight stand-in that supports ``add_done_callback`` exactly
as a real ``asyncio.Task`` does.
"""
_install_settings(monkeypatch, authz_enabled=True, audit_enabled=True)
scheduled: list[object] = []
class _FakeTask:
def __init__(self, coro):
self._coro = coro
self._callbacks: list = []
def add_done_callback(self, cb):
self._callbacks.append(cb)
def done(self) -> bool:
return False
def _capture(coro):
scheduled.append(coro)
coro.close() # don't actually run — the function imports from DB modules.
task = _FakeTask(coro)
scheduled.append(task)
return task
monkeypatch.setattr("asyncio.create_task", _capture)
await authz_utils.audit_decision(user_id=uuid4(), action="flow:read", obj="flow:x", result="allow")
assert len(scheduled) == 1
try:
assert len(scheduled) == 1
# The implementation must hold a reference so the GC can't drop the task.
assert scheduled[0] in authz_utils._pending_audit_tasks
# And it must register a done-callback that removes itself from the set.
assert scheduled[0]._callbacks, "audit_decision must register a done-callback to clean up _pending_audit_tasks"
finally:
# The fake task is not a real ``asyncio.Task`` — clean up the
# module-global set so the next test starts from an empty state.
authz_utils._pending_audit_tasks.discard(scheduled[0])
@pytest.mark.anyio
async def test_drain_pending_audit_writes_awaits_tasks():
"""drain_pending_audit_writes waits for tracked tasks to finish."""
import asyncio
finished = asyncio.Event()
async def _slow() -> None:
await asyncio.sleep(0.01)
finished.set()
task = asyncio.create_task(_slow())
authz_utils._pending_audit_tasks.add(task)
task.add_done_callback(authz_utils._pending_audit_tasks.discard)
await authz_utils.drain_pending_audit_writes(timeout=1.0)
assert finished.is_set()
assert task not in authz_utils._pending_audit_tasks
@pytest.mark.anyio
async def test_ensure_permission_fails_closed_on_plugin_exception(monkeypatch, fake_user):
"""If the authz plugin raises, ensure_permission must deny (403), not bubble 500."""
_install_settings(monkeypatch, authz_enabled=True, audit_enabled=False)
class _BrokenPlugin:
async def enforce(self, **_kwargs):
msg = "casbin db down"
raise RuntimeError(msg)
async def batch_enforce(self, **_kwargs):
return []
monkeypatch.setattr(authz_utils, "get_authorization_service", lambda: _BrokenPlugin())
captured = _install_audit_recorder(monkeypatch)
with pytest.raises(HTTPException) as excinfo:
await authz_utils.ensure_permission(fake_user, domain="*", obj="flow:abc", act="read")
assert excinfo.value.status_code == 403, "Plugin exceptions must fail closed (deny), not 500"
# The deny path must still emit an audit row so the operator can see the failure.
assert captured, "Plugin exception must still produce an audit row"
assert captured[0]["result"] == "deny"
assert "error" in captured[0]["details"]
# ----------------------------------------------------------------------------- #

44
src/lfx/src/lfx/services/authorization/base.py
View File

@ -3,7 +3,8 @@
from __future__ import annotations
import abc
from typing import TYPE_CHECKING, Any, ClassVar
from typing import TYPE_CHECKING, Any, ClassVar, TypedDict
from uuid import UUID as _UUID
from lfx.services.base import Service
from lfx.services.schema import ServiceType
@ -13,6 +14,47 @@ if TYPE_CHECKING:
from uuid import UUID
class AuthzContext(TypedDict, total=False):
"""Documented shape of the ``context`` dict passed to authz plugins.
Plugins receive a ``dict[str, Any]`` for forward compatibility — new
callers may add fields without forcing a plugin recompile — but this
TypedDict documents the keys the framework promises to populate and the
keys plugins should be prepared to read.
Framework-populated keys
------------------------
is_superuser : bool
Always present. ``getattr(user, "is_superuser", False)``.
Resource-owner hints (one is populated per call, others omitted)
----------------------------------------------------------------
flow_user_id, deployment_user_id, project_user_id, knowledge_base_user_id,
variable_user_id, file_user_id, share_user_id : UUID | None
The id of the user who owns the resource being authorized. The
framework's ``ensure_*_permission`` helpers short-circuit on owner
match, so this key is the source of truth for the owner-override path.
Domain / scope hints
--------------------
workspace_id : UUID | None
The workspace this resource belongs to, when applicable.
folder_id : UUID | None
The folder/project this resource belongs to, when applicable.
"""
is_superuser: bool
flow_user_id: _UUID | None
deployment_user_id: _UUID | None
project_user_id: _UUID | None
knowledge_base_user_id: _UUID | None
variable_user_id: _UUID | None
file_user_id: _UUID | None
share_user_id: _UUID | None
workspace_id: _UUID | None
folder_id: _UUID | None
class BaseAuthorizationService(Service, abc.ABC):
"""Abstract base class for authorization (RBAC) services.

29
src/lfx/src/lfx/services/manager.py
View File

@ -390,17 +390,44 @@ class ServiceManager:
)
return
# Map of service_type → expected base class. Used to reject malformed
# entry-point plugins before they take over an authorization-critical
# service. Without this check an arbitrary class from an unrelated
# PyPI package that happened to register the same entry-point name
# would silently replace the OSS implementation.
expected_bases: dict[ServiceType, type] = {}
try:
from lfx.services.authorization.base import BaseAuthorizationService
expected_bases[ServiceType.AUTHORIZATION_SERVICE] = BaseAuthorizationService
except Exception as exc: # noqa: BLE001 — optional import, validation just skipped
logger.debug(f"BaseAuthorizationService unavailable; entry-point validation skipped: {exc}")
for ep in eps:
try:
service_class = ep.load()
# Entry point name should match ServiceType enum value
service_type = ServiceType(ep.name)
expected_base = expected_bases.get(service_type)
if expected_base is not None and not (
isinstance(service_class, type) and issubclass(service_class, expected_base)
):
logger.warning(
f"Entry point {ep.name} resolved to {service_class!r}, "
f"which is not a subclass of {expected_base.__name__}. "
f"Skipping registration to avoid silently replacing the "
f"built-in service with an incompatible plugin."
)
continue
self.register_service_class(service_type, service_class, override=False)
logger.debug(f"Loaded service from entry point: {ep.name}")
except (ValueError, AttributeError) as exc:
logger.warning(f"Failed to load entry point {ep.name}: {exc}")
except Exception as exc: # noqa: BLE001
logger.debug(f"Error loading entry point {ep.name}: {exc}")
# Authz plugin failures are operator-visible — silent
# degradation to the OSS pass-through is exactly the kind
# of behavior change we want noisy.
logger.warning(f"Error loading entry point {ep.name}: {exc}")
def _discover_from_config(self, config_dir: Path) -> None:
"""Discover services from config files (lfx.toml / pyproject.toml)."""

5
src/lfx/src/lfx/services/settings/auth.py
View File

@ -145,8 +145,9 @@ class AuthSettings(BaseSettings):
AUTHZ_AUDIT_ENABLED: bool = Field(
default=True,
description=(
"Write an AuthzAuditLog row for every authorization decision when AUTHZ_ENABLED=true. "
"Ignored when AUTHZ_ENABLED=false."
"Write an AuthzAuditLog row for every authorization decision and share-administration "
"action. Independent of AUTHZ_ENABLED — keep audit on while enforcement is off to "
"observe traffic before flipping the AUTHZ_ENABLED flag."
),
)