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

docs(authz): shorten docstrings and drop enterprise/Casbin comments

Browse Source 
Use one-line module and helper docstrings across authorization code and
route guards; keep table/type identifiers unchanged.
This commit is contained in:
himavarshagoutham
2026-05-26 10:37:26 -04:00
parent 460392acd3
commit 2ac7568f01
40 changed files with 168 additions and 736 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,29 +1,8 @@
"""authz foundations — casbin_rule, authz_* tables, workspace_id columns
"""Authz foundations: policy rules, authz_* tables, workspace_id columns.
Revision ID: 7c8d9e0f1a2b
Revises: mb01b2c3d4e5
Create Date: 2026-05-20
Phase: EXPAND
Single consolidated migration for the OSS authorization layer (PR #13153).
Emits the final schema directly — partial unique indexes that handle NULL
columns, ``DateTime(timezone=True)`` everywhere, CHECK constraints on
``authz_share`` enum-like columns, and the ``workspace_id`` column on flow /
folder / deployment.
This replaces what was previously a chain of six smaller migrations that were
collapsed before any of them shipped:
* ``f7a8b9c0d1e2`` (initial authz tables)
* ``c8e5f4b2a9d7`` (workspace_id columns)
* ``d9e8f7a6b5c4`` (partial unique indexes on authz_role_assignment)
* ``e4f5a6b7c8d9`` (partial unique indexes on authz_share)
* ``f0a1b2c3d4e5`` (timestamps tz-aware, ptype index, FK fix, composite index, widened partial index)
* ``b2c3d4e5f6a1`` (CHECK constraints on authz_share)
Partial unique indexes use ``postgresql_where`` / ``sqlite_where`` — the
project targets PostgreSQL and SQLite only.
"""
from collections.abc import Sequence
@ -45,9 +24,7 @@ _WORKSPACE_TABLES = ("flow", "folder", "deployment")
def upgrade() -> None:
conn = op.get_bind()
# ------------------------------------------------------------------
# casbin_rule — Casbin policy storage (SQLAlchemy adapter compatible)
# ------------------------------------------------------------------
# policy rules table
if not migration.table_exists("casbin_rule", conn):
op.create_table(
"casbin_rule",
@ -61,7 +38,7 @@ def upgrade() -> None:
sa.Column("v5", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=True),
sa.PrimaryKeyConstraint("id"),
)
# Casbin's loader filters by ``ptype`` on every load_policy() and
# The policy loader's loader filters by ``ptype`` on every load_policy() and
# AddPolicy() — required for non-trivial policy volumes.
op.create_index("ix_casbin_rule_ptype", "casbin_rule", ["ptype"])
@ -116,7 +93,7 @@ def upgrade() -> None:
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
# (type, id)" — the canonical query an authorization plugin 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.
@ -289,7 +266,7 @@ def upgrade() -> None:
# ``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
# set, an authorization plugin's owner-shortcircuit audit row would
# silently violate the constraint.
sa.CheckConstraint(
"result IN ('allow', 'deny', 'owner_override')",

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

@ -1,23 +1,8 @@
"""seed authz system roles — viewer / developer / admin
"""Seed built-in authz roles (viewer, developer, admin).
Revision ID: 8d3a1f9c2e0b
Revises: 7c8d9e0f1a2b
Create Date: 2026-05-21
Phase: EXPAND
Phase 4 of the OSS RBAC rollout. Inserts the three built-in roles referenced
by the design document so an enterprise plugin has a stable bootstrap set
without having to ship its own seed migration. EXPAND because the change is
additive — new rows in an existing table, no schema or data semantics change.
OSS does not interpret these JSON permission lists — they are pure metadata.
The enterprise Casbin plugin reads them during ``PolicySync`` to compile
matching ``p`` rules in ``casbin_rule``.
Idempotent: existing rows with the same ``name`` are left untouched (each
insert is gated by a ``WHERE NOT EXISTS`` subquery against ``name``). Safe to
re-run.
"""
from __future__ import annotations
@ -39,7 +24,7 @@ depends_on: str | Sequence[str] | None = None
# Permission templates use ``"{resource}:{action}"`` slugs that map directly
# to Casbin object/action pairs. Enterprise plugins read these to seed
# to policy object/action pairs. Authorization plugins read these to seed
# matching ``p`` rules during ``PolicySync``.
_VIEWER_PERMISSIONS: tuple[str, ...] = (
"flow:read",
@ -126,7 +111,7 @@ def upgrade() -> None:
# 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
# JSON column, which downstream readers (PolicySync expects a
# list) would have to peel a second time.
for name, description, permissions in _SYSTEM_ROLES:
# Use an atomic check-then-insert to harden against concurrent

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

@ -116,7 +116,7 @@ async def cascade_delete_flow(session: AsyncSession, flow_id: uuid.UUID) -> None
# 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
# silently survive — that would let an authorization 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)

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

@ -1,25 +1,4 @@
"""CRUD API for ``authz_share`` rows.
This router is the OSS-side admin surface for resource-level shares. It does
not interpret share rows for enforcement — that is the Enterprise plugin's
job — but it writes the canonical rows so plugins have something to compile
into Casbin policy via ``PolicySync``.
Authorization model
-------------------
* Creating, listing, updating, or deleting a share on a resource is gated by
``share:{action}`` (via :func:`ensure_share_permission`) and falls through
to the OSS pass-through (allow-all) unless an enterprise plugin is
registered.
* The route handler also enforces an OSS-side floor: only the resource owner
or a superuser may write shares for that resource. This prevents the OSS
pass-through default from silently letting a viewer-role user hand out
grants on someone else's resource.
* Non-owners listing shares only see rows whose ``target_id`` matches their
own user id (so users can see what's been shared *with* them without
seeing the full grant ledger).
"""
"""CRUD API for authz_share rows (enforcement is delegated to authorization plugins)."""
from __future__ import annotations
@ -48,10 +27,7 @@ from langflow.services.deps import get_authorization_service
router = APIRouter(prefix="/authz/shares", tags=["Authorization"])
# Map resource type slug → (SQLModel, FK-to-user column attribute). Knowledge
# bases use ``KnowledgeBaseRecord`` (UUID primary key) so the share row's
# UUID-typed ``resource_id`` aligns with the Casbin object key
# (``knowledge_base:{kb_id}``) emitted by ``ensure_knowledge_base_permission``.
# resource_type slug → (model, owner column).
_RESOURCE_OWNER_LOOKUPS: dict[str, tuple[type, str]] = {
"flow": (Flow, "user_id"),
"deployment": (Deployment, "user_id"),
@ -86,16 +62,7 @@ async def _user_can_see_share(
user_id: UUID,
resource_owner_id: UUID | None,
) -> bool:
"""Return True when ``user_id`` is permitted to see ``row`` in list/get.
Visibility tiers (any one is sufficient):
* resource owner — full visibility on shares of their own resource;
* share creator — needs to manage their own grants;
* user-scoped target — needs to know what's been shared with them;
* team-scoped target — caller is a member of ``row.target_id``;
* public scope — anyone in the system can see the row exists.
"""
"""Return True when the user may see this share row."""
if user_id in {resource_owner_id, row.created_by}:
return True
scope = row.scope
@ -114,15 +81,7 @@ async def _user_can_see_share(
async def _invalidate_for_share(scope: str, target_id: UUID | None) -> None:
"""Drop cached policy after a share write, scoped to the share's audience.
The plugin-side ``invalidate_user(uuid)`` API only knows about user ids,
so ``target_id`` is *only* a valid argument when ``scope == "user"``.
For team / private / public shares (and for unknown scopes), fall back
to ``invalidate_all`` — otherwise we'd hand the enforcer a team uuid as
if it were a user uuid and the actual team members' cache would stay
stale.
"""
"""Invalidate cached policy after a share write (user scope vs invalidate_all)."""
authz = get_authorization_service()
if scope == ShareScope.USER.value and target_id is not None:
await authz.invalidate_user(target_id)
@ -135,24 +94,14 @@ async def _ensure_can_administer_share(
user: User,
owner_id: UUID | None,
) -> None:
"""OSS floor: only resource owner or superuser may write shares.
When an enterprise plugin is actively enforcing (cross-user fetch
supported AND ``LANGFLOW_AUTHZ_ENABLED=true``) this floor is skipped so
a workspace/admin role with ``share:create`` can administer shares for
resources it doesn't own; ``ensure_share_permission`` is the
authoritative check in that mode. Under the OSS pass-through (allow-all
enforce) the floor stays in place so a viewer cannot mint share rows for
someone else's resource.
"""
"""Require resource owner or superuser unless cross-user enforcement is active."""
if getattr(user, "is_superuser", False):
return
if owner_id is not None and owner_id == user.id:
return
authz = get_authorization_service()
if await authz.supports_cross_user_fetch() and await authz.is_enabled():
# Enterprise plugin is active — let ``ensure_share_permission``
# decide (its decision will fire downstream of this helper).
# Defer to ensure_share_permission when enforcement is active.
return
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
@ -167,19 +116,14 @@ async def create_share(
current_user: CurrentActiveUser,
session: DbSession,
) -> ShareRead:
"""Create an ``authz_share`` row granting access to a resource.
The caller must be the resource owner or a superuser. The plugin-level
``share:create`` guard fires after the OSS owner check, so an enterprise
plugin can additionally deny owners with insufficient role.
"""
"""Create an authz_share row for a resource."""
owner_id = await _resolve_resource_owner(
session,
resource_type=payload.resource_type,
resource_id=payload.resource_id,
)
if owner_id is None:
# The resource simply does not exist — UUID privacy: 404.
# UUID privacy: missing resource → 404.
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Resource not found")
await _ensure_can_administer_share(user=current_user, owner_id=owner_id)
await ensure_share_permission(
@ -201,9 +145,7 @@ 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).
# Log server-side; return a fixed 409 message (no schema leakage).
await session.rollback()
logger.warning("authz_share insert rejected: %s", exc)
raise HTTPException(
@ -212,9 +154,7 @@ async def create_share(
) from exc
await session.refresh(row)
# Tell the enforcer (if any) to drop its cached policy. Use the share's
# scope to pick the right invalidation — a team's target_id is a team
# uuid, not a user uuid, so ``invalidate_user`` would not reach members.
# Invalidate policy cache for the share audience.
await _invalidate_for_share(payload.scope, payload.target_id)
await audit_decision(
@ -248,17 +188,7 @@ async def list_shares(
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, 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.
"""
"""List share rows visible to the caller (paginated, max 200)."""
await ensure_share_permission(
current_user,
ShareAction.READ,
@ -273,16 +203,14 @@ async def list_shares(
if target_id is not None:
stmt = stmt.where(AuthzShare.target_id == target_id)
if scope is not None:
# Validate scope literal against the enum so unknown values 422 early.
# Reject unknown scope values early (422).
try:
scope_value = ShareScope(scope).value
except ValueError as exc:
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.
# Stable ordering with offset/limit pagination.
stmt = stmt.order_by(AuthzShare.created_at.desc(), AuthzShare.id).offset(offset).limit(limit)
rows = list(await session.exec(stmt))
@ -291,13 +219,11 @@ async def list_shares(
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.
# Pre-fetch team memberships (avoid N+1 per row).
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.
# Filter rows by visibility rules for non-superusers.
visible: list[ShareRead] = []
owner_cache: dict[tuple[str, UUID], UUID | None] = {}
for row in rows:
@ -325,12 +251,7 @@ def _row_visible_to(
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.
"""
"""Batch list visibility check using pre-fetched team ids."""
if user_id in {resource_owner_id, row.created_by}:
return True
scope = row.scope
@ -373,7 +294,7 @@ async def get_share(
user_id=current_user.id,
resource_owner_id=owner_id,
):
# UUID privacy — caller is not allowed to know this share exists.
# UUID privacy: forbidden share → 404.
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Share not found")
return ShareRead.model_validate(row, from_attributes=True)
@ -404,8 +325,7 @@ async def update_share(
share_user_id=row.created_by,
)
# Validate against the enum to keep the DB CHECK constraint happy and
# surface unknown values as 422 instead of a constraint error.
# Validate permission_level (422 before DB CHECK).
try:
row.permission_level = SharePermissionLevel(payload.permission_level).value
except ValueError as exc:
@ -414,9 +334,7 @@ async def update_share(
detail=f"Unknown permission_level {payload.permission_level!r}",
) from exc
session.add(row)
# 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.
# Rollback + fixed 409 on constraint failure (same as create_share).
try:
await session.flush()
except Exception as exc:

4
src/backend/base/langflow/api/v1/chat.py
View File

@ -232,7 +232,7 @@ async def build_flow(
# since build is a valid operation on shared flows.
#
# Phase 3 prerequisite: the owner-OR-public filter below short-circuits an
# enterprise execute-grant on a non-owned private flow. Cross-user build
# plugin execute-grant on a non-owned private flow. Cross-user build
# support requires share-aware loading (load by id, then check via the
# plugin, then convert deny to 404 for UUID-privacy) which lands with the
# `authz_share` CRUD work. See `langflow.services.authorization.utils`.
@ -251,7 +251,7 @@ async def build_flow(
)
raise HTTPException(status_code=404, detail=f"Flow with id {flow_id} not found")
# Authorize the execute action — runs the enterprise plugin if registered,
# Authorize the execute action — runs the authorization plugin if registered,
# no-op in OSS pass-through. Audited regardless.
await ensure_flow_permission(
current_user,

2
src/backend/base/langflow/api/v1/deployments.py
View File

@ -774,7 +774,7 @@ async def list_deployments(
)
# Per-deployment authorization filter. Mirrors GET /flows/ and GET /projects/:
# the coarse READ check above gates whether the caller can list deployments
# at all; this call drops individual rows the enterprise plugin denies. OSS
# at all; this call drops individual rows the authorization plugin denies. OSS
# pass-through returns the input unchanged. ``rows_with_counts`` is
# ``list[tuple[Deployment, int, list[...]]]`` so the key/domain extractors
# operate on the first element. ``total`` may overcount denied items —

4
src/backend/base/langflow/api/v1/endpoints.py
View File

@ -525,7 +525,7 @@ async def get_flow_for_api_key_user(
authenticated user from ``api_key_security`` and passes it to the helper,
so cross-user access fails closed with 404 at the helper layer.
When an enterprise authorization service is registered, the lookup is
When an authorization plugin is registered, the lookup is
share-aware (load by id, route guard decides access). The OSS pass-through
default keeps the owner-scoped lookup.
"""
@ -603,7 +603,7 @@ async def _run_flow_internal(
# Authorization happens upstream: every caller of _run_flow_internal must
# have called ensure_flow_permission(FlowAction.EXECUTE, ...) before
# invoking us. The owner-only check that used to live here would reject any
# enterprise execute-grant on a shared flow, defeating the plugin's
# plugin execute-grant on a shared flow, defeating the plugin's
# purpose. Adding a defense-in-depth check here would re-introduce the
# same regression.
telemetry_service = get_telemetry_service()

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

@ -160,12 +160,7 @@ async def read_flows(
flows = [flow for flow in flows if flow.is_component]
if remove_example_flows and starter_folder_id:
flows = [flow for flow in flows if flow.folder_id != starter_folder_id]
# When AUTHZ_ENABLED=true, drop any flows the user can't read. OSS
# default is pass-through (returns the input list unchanged); the
# enterprise plugin uses batch_enforce to apply share/role checks.
# Per-flow ``domain_extractor`` groups requests so project-scoped
# grants are evaluated against the right Casbin tuple — flows in
# different workspaces/projects can't share a single domain.
# Filter list rows when AUTHZ_ENABLED (per-flow domain_extractor).
flows = await filter_visible_resources(
current_user,
resource_type="flow",
@ -193,15 +188,7 @@ async def read_flows(
)
page = await apaginate(session, stmt, params=params)
# 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. 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).
# Same authz filter as get_all (page.total may overcount denied rows).
page.items = await filter_visible_resources(
current_user,
resource_type="flow",
@ -227,9 +214,7 @@ async def read_flow(
current_user: CurrentActiveUser,
):
"""Read a flow."""
# ``_read_flow`` is share-aware when an enterprise plugin is registered;
# otherwise it stays owner-scoped. A plugin deny becomes 404 here so we
# don't disclose existence of a flow the caller can't reach.
# Share-aware fetch; plugin deny → 404 (UUID privacy).
if user_flow := await _read_flow(session, flow_id, current_user.id):
try:
await ensure_flow_permission(
@ -263,11 +248,7 @@ async def get_note_translations(
"""
from langflow.utils.i18n import translate
# Scope the fetch to flows the caller already owns; an enterprise plugin
# extends visibility through ensure_flow_permission below. Returning ``{}``
# rather than 404 here matches the helper's "no notes" path so that
# missing-flow vs not-owned vs not-authorized are indistinguishable to
# an attacker probing UUIDs.
# Owner-scoped fetch; empty dict preserves UUID privacy vs 404.
flow = await _read_flow(session=session, flow_id=flow_id, user_id=current_user.id)
if not flow or not flow.data:
return {}
@ -448,15 +429,7 @@ async def upsert_flow(
existing_flow = (await session.exec(select(Flow).where(Flow.id == flow_id))).first()
if existing_flow is not None:
# 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.
# Block non-owner upsert when cross-user fetch is off (UUID privacy).
from langflow.services.deps import get_authorization_service
authz = get_authorization_service()
@ -748,13 +721,7 @@ async def delete_multiple_flows(
async def _delete_operation() -> int:
if not flow_ids:
return 0
# 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.
# Widen fetch when cross-user DELETE is supported; else owner-scoped.
from langflow.services.deps import get_authorization_service
authz = get_authorization_service()
@ -765,10 +732,7 @@ async def delete_multiple_flows(
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.
# Propagate plugin deny (403) so bulk delete fails audibly.
await ensure_flow_permission(
user,
FlowAction.DELETE,
@ -811,11 +775,7 @@ 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.
# 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.
# Widen fetch when cross-user READ is supported; else owner-scoped.
from langflow.services.deps import get_authorization_service
authz = get_authorization_service()
@ -830,9 +790,7 @@ async def download_multiple_file(
raise HTTPException(status_code=404, detail="No flows found.")
for flow in flows:
# 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".
# Plugin deny → 404 (UUID privacy).
try:
await ensure_flow_permission(
user,

2
src/backend/base/langflow/api/v1/flows_helpers.py
View File

@ -346,7 +346,7 @@ async def _read_flow(
"""Read a flow.
When the registered authorization service supports cross-user fetch
(enterprise Casbin), the row is loaded by id alone and the caller's
(authorization plugin), the row is loaded by id alone and the caller's
``ensure_flow_permission`` decides access. Otherwise the query stays
owner-scoped so the OSS pass-through default cannot widen visibility.
"""

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

@ -101,7 +101,7 @@ async def _guard_kb_action(
"""Guard a KB-scoped action and return the effective KB owner context.
Looks up ``KnowledgeBaseRecord(user_id=current_user.id, name=kb_name)`` so
the Casbin object key (``knowledge_base:{record.id}``) lines up with the
the policy object key (``knowledge_base:{record.id}``) lines up with the
UUID-typed ``authz_share.resource_id`` column. Legacy disk-only KBs (no
``KnowledgeBaseRecord`` row) fall back to ``kb_id=None`` so the enforcer
sees ``knowledge_base:*`` and the owner-override path still applies.
@ -272,7 +272,7 @@ def _check_memory_base_association(kb_name: str, current_user: CurrentActiveUser
Owner-scoped early gate — runs as ``Depends(...)`` before the route
body and only sees the actor. For shared KBs reached through an
enterprise share grant the actor has no same-named local KB so this
shared grant the actor has no same-named local KB so this
dep returns early; the route body then re-runs the check against the
resolved owner via :func:`_assert_kb_not_memory_base` so a shared
Memory-Base-managed KB still gets blocked.
@ -1308,7 +1308,7 @@ async def list_knowledge_bases(
"""
# List-level guard: a viewer-role user may still see the KBs they own,
# but a role with ``knowledge_base:read`` revoked entirely is rejected
# here. Per-row filtering is the enterprise plugin's responsibility once
# here. Per-row filtering is the authorization plugin's responsibility once
# KB share grants exist.
await _guard_kb_action(current_user=current_user, action=KnowledgeBaseAction.READ, kb_name=None)
try:
@ -2118,7 +2118,7 @@ async def delete_knowledge_bases_bulk(
) -> dict[str, object]:
"""Delete multiple knowledge bases."""
# Per-KB guard. Resolve each guard upfront so the loop body can use the
# owner context (path + DB row owner) when an enterprise plugin authorizes
# owner context (path + DB row owner) when an authorization plugin authorizes
# a non-owner via a share grant.
kb_guards: dict[str, _KbGuardResult] = {}
for kb_name in request.kb_names:

2
src/backend/base/langflow/api/v1/mappers/deployments/helpers.py
View File

@ -426,7 +426,7 @@ async def get_deployment_row_or_404(
user_id: UUID,
db: DbSession,
) -> Deployment:
# ``get_deployment_db`` is share-aware when an enterprise plugin is
# ``get_deployment_db`` is share-aware when an authorization plugin is
# registered and ``LANGFLOW_AUTHZ_ENABLED`` is on; otherwise it stays
# owner-scoped. Routes still call ``ensure_deployment_permission`` after
# this to authorize the actor against the resolved deployment.

11
src/backend/base/langflow/api/v1/monitor.py
View File

@ -97,14 +97,7 @@ async def _ensure_flow_action_or_404(
user: User,
action: FlowAction,
) -> Flow | None:
"""Load a flow (share-aware) and run ``ensure_flow_permission`` on it.
Returns the flow row, or ``None`` if no such flow exists (the caller
decides whether to 404 or to return an empty payload). Raises 403 if the
enterprise plugin denies; under the OSS pass-through default the
owner-scoped lookup ensures only the owner's flow is found at all so this
is effectively a no-op when no enterprise plugin is registered.
"""
"""Load a flow (share-aware), enforce permission, return None if missing."""
flow = await authorized_or_owner_scoped(
session,
Flow,
@ -199,7 +192,7 @@ async def get_message_sessions(
# When a flow_id is provided, gate on flow READ permission so a viewer
# without flow access cannot enumerate sessions. The bulk path
# (flow_id is None) keeps the user-scoped JOIN — share-aware listing
# across all visible flows is an enterprise-side optimisation.
# across all visible flows is an plugin optimisation.
if flow_id is not None:
flow = await _ensure_flow_action_or_404(session, flow_id=flow_id, user=current_user, action=FlowAction.READ)
if flow is None:

2
src/backend/base/langflow/api/v1/openai_responses.py
View File

@ -632,7 +632,7 @@ async def create_response(
return OpenAIErrorResponse(error=error_response["error"])
# Get flow using the model field (which contains flow_id). The lookup
# becomes share-aware when an enterprise plugin is registered, so we
# becomes share-aware when an authorization plugin is registered, so we
# also enforce ``flow:execute`` explicitly — otherwise an API key with
# cross-user fetch enabled would skip policy here.
try:

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

@ -222,9 +222,9 @@ async def read_projects(
).all()
projects = [project for project in projects if project.name != STARTER_FOLDER_NAME]
# When AUTHZ_ENABLED=true, drop projects the user can't read. OSS
# default is pass-through; the enterprise plugin honors role + share grants.
# default is pass-through; the authorization plugin honors role + share grants.
# ``domain_extractor`` groups requests by workspace so each batch is
# evaluated against the right Casbin tuple. Projects are the resource
# evaluated against the right policy tuple. Projects are the resource
# itself, so the domain falls back to workspace (or ``*``).
projects = await filter_visible_resources(
current_user,
@ -256,7 +256,7 @@ async def read_project(
search: str = "",
):
try:
# Share-aware fetch: when an enterprise authorization service is
# Share-aware fetch: when an authorization plugin is
# registered (``SUPPORTS_CROSS_USER_FETCH=True``) the project is
# loaded by id alone and ``ensure_project_permission`` below decides
# access. The OSS pass-through keeps the owner-scoped query so the
@ -347,7 +347,7 @@ async def read_project(
# 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
# project regardless of finer-grained policy engine 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(

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

@ -1,9 +1,4 @@
"""Pydantic schemas for the ``/api/v1/authz/shares`` router.
Mirrors the ``AuthzShare`` SQLModel but flattens the enum values to literals so
clients see a clear allow-list in the OpenAPI schema rather than the raw
SQLAlchemy enum class.
"""
"""Pydantic schemas for /api/v1/authz/shares."""
from __future__ import annotations
@ -13,9 +8,7 @@ from uuid import UUID
from pydantic import BaseModel, Field, model_validator
# Centralised slugs that match the Casbin object prefixes used elsewhere in
# this module. Keep this list aligned with the action-enum modules; adding a
# new shareable resource type requires touching both places.
# Shareable resource slugs (keep aligned with authorization action modules).
ShareResourceType = Literal[
"flow",
"deployment",
@ -30,13 +23,7 @@ SharePermissionLiteral = Literal["read", "write", "execute", "admin"]
class ShareCreate(BaseModel):
"""Payload for creating an ``authz_share`` row.
Targeted scopes (``team``, ``user``) require a ``target_id``; untargeted
scopes (``private``, ``public``) forbid one. The DB CHECK constraint
``scope_target_consistency`` enforces the same shape — the validator below
is just so the API returns 422 instead of leaking the DB error.
"""
"""Payload for creating an authz_share row."""
resource_type: ShareResourceType
resource_id: UUID
@ -46,35 +33,27 @@ class ShareCreate(BaseModel):
@model_validator(mode="after")
def _check_scope_target_consistency(self) -> ShareCreate:
"""Require target_id for user/team scopes; forbid it for private/public."""
targeted = self.scope in ("team", "user")
has_target = self.target_id is not None
if targeted and not has_target:
msg = f"scope={self.scope!r} requires target_id"
if targeted and self.target_id is None:
msg = f"scope {self.scope!r} requires target_id"
raise ValueError(msg)
if not targeted and has_target:
msg = f"scope={self.scope!r} must not include a target_id"
if not targeted and self.target_id is not None:
msg = f"scope {self.scope!r} must not set target_id"
raise ValueError(msg)
return self
class ShareUpdate(BaseModel):
"""Payload for updating an ``authz_share`` row.
Only ``permission_level`` is editable. Changing target_id or scope means
you wanted a different share — revoke and recreate.
"""
"""Payload for updating an authz_share permission level."""
permission_level: SharePermissionLiteral
class ShareRead(BaseModel):
"""Read-only projection of an ``authz_share`` row."""
"""Serialized authz_share row returned by the API."""
id: UUID
# 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

4
src/backend/base/langflow/api/v2/files.py
View File

@ -90,7 +90,7 @@ async def byte_stream_generator(file_input, chunk_size: int = 8192) -> AsyncGene
async def fetch_file_object(file_id: uuid.UUID, current_user: CurrentActiveUser, session: DbSession):
# Share-aware fetch. Under the OSS pass-through this keeps the existing
# owner-scoped query (cannot widen visibility). Enterprise plugins set
# owner-scoped query (cannot widen visibility). Authorization plugins set
# ``SUPPORTS_CROSS_USER_FETCH=True`` so a share grant can resolve here.
file = await authorized_or_owner_scoped(
session,
@ -455,7 +455,7 @@ async def delete_files_batch(
):
"""Delete multiple files by their IDs."""
try:
# Share-aware fetch: when an enterprise plugin supports cross-user
# Share-aware fetch: when an authorization plugin supports cross-user
# access, the SELECT loads by id alone so each row's true owner
# surfaces and per-row ``ensure_file_permission`` can decide. The OSS
# pass-through keeps the owner-scoped query.

2
src/backend/base/langflow/api/v2/workflow.py
View File

@ -150,7 +150,7 @@ async def execute_workflow(
try:
# Validate flow exists and user has permission. The lookup becomes
# share-aware when an enterprise plugin is registered, so we must
# share-aware when an authorization plugin is registered, so we must
# also enforce ``flow:execute`` explicitly — otherwise an API key
# with cross-user fetch enabled would bypass policy here.
flow = await get_flow_by_id_or_endpoint_name(workflow_request.flow_id, api_key_user.id)

10
src/backend/base/langflow/cli/authz_dry_run.py
View File

@ -1,13 +1,13 @@
"""``langflow authz dry-run`` — simulate every flow guard against a stub policy.
The OSS ``LangflowAuthorizationService`` always allows, so a real install can't
demonstrate enforcement without the enterprise Casbin plugin. This subcommand
demonstrate enforcement without the authorization plugin. This subcommand
replaces the live authorization service with a small in-memory stub for one
invocation, walks every flow-CRUD guard site, and prints what *would* happen
under the chosen policy — including the Casbin tuple, the audit row that
under the chosen policy — including the policy tuple, the audit row that
would have been written, and the resulting HTTP outcome. Useful for:
* Validating that ``AUTHZ_ENABLED=true`` + a future Casbin policy will produce
* Validating that ``AUTHZ_ENABLED=true`` + a future policy will produce
the expected behaviour before shipping it to production.
* Showing operators what the audit log will look like.
* Smoke-testing the guard wiring after a refactor.
@ -51,7 +51,7 @@ _console = Console()
class StubPolicy(str, Enum):
"""Built-in stand-ins for an enterprise Casbin policy."""
"""Built-in stand-ins for an authorization plugin policy."""
ALLOW_ALL = "allow-all"
DENY_NON_OWNER = "deny-non-owner"
@ -428,7 +428,7 @@ def dry_run(
The command runs entirely in-process. No HTTP request is made, no row is
written to the real ``authz_audit_log`` table — the report describes the
Casbin tuple and audit row that *would* be produced if an enterprise
policy tuple and audit row that *would* be produced if an plugin
plugin enforcing ``--policy`` were registered.
"""
rows = asyncio.run(_run_all(policy))

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

@ -166,24 +166,7 @@ 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.
"""
"""Load a flow graph after authorizing EXECUTE for the caller."""
from lfx.graph.graph.base import Graph
from langflow.processing.process import process_tweaks
@ -216,11 +199,7 @@ async def load_flow(
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.
# Map plugin deny (403) to ValueError for existing callers.
caller = await session.get(User, uuid_user_id)
if caller is None:
msg = "Session is invalid"
@ -460,20 +439,11 @@ def get_arg_names(inputs: list[Vertex]) -> list[dict[str, str]]:
async def get_flow_by_id_or_endpoint_name(flow_id_or_name: str, user_id: str | UUID | None = None) -> FlowRead:
"""Resolve a flow by UUID or ``endpoint_name``.
When the registered authorization service supports cross-user fetch
(enterprise Casbin), the owner-equality check is skipped and the route's
``ensure_flow_permission`` decides access. The OSS pass-through default
keeps the owner-scoped lookup so enabling ``LANGFLOW_AUTHZ_ENABLED`` alone
cannot widen visibility.
"""
"""Resolve a flow by UUID or endpoint_name (share-aware when enforcement is on)."""
from langflow.services.deps import get_authorization_service
authz = get_authorization_service()
# Cross-user fetch is only safe when enforcement is also active — otherwise
# the route guards are no-ops and widening the lookup would silently
# expose foreign flows without any policy check.
# Widen lookup only when cross-user fetch and AUTHZ_ENABLED are both on.
share_aware = await authz.supports_cross_user_fetch() and await authz.is_enabled()
async with session_scope() as session:

2
src/backend/base/langflow/plugin_routes.py
View File

@ -99,7 +99,7 @@ class _PluginAppWrapper:
def load_plugin_routes(app: FastAPI) -> None:
"""Discover and register additional routers from enterprise plugins.
"""Discover and register additional routers from authorization plugins.
Plugins register themselves via the ``langflow.plugins`` entry-point group.
Each entry point must expose a callable with the signature::

2
src/backend/base/langflow/services/authorization/__init__.py
View File

@ -1,4 +1,4 @@
"""Langflow OSS authorization service package (pass-through; enterprise plugin enforces)."""
"""OSS authorization service package (pass-through default; plugins enforce)."""
from langflow.services.authorization.actions import (
DeploymentAction,

35
src/backend/base/langflow/services/authorization/actions.py
View File

@ -6,12 +6,7 @@ from enum import Enum
class FlowAction(str, Enum):
"""Actions that can be authorized on a flow resource.
Values are the Casbin ``act`` strings consumed by the enterprise policy
engine. Subclassing ``str`` lets these be passed wherever a bare string is
accepted, so callers can migrate incrementally.
"""
"""Actions authorized on a flow resource."""
READ = "read"
WRITE = "write"
@ -22,7 +17,7 @@ class FlowAction(str, Enum):
class DeploymentAction(str, Enum):
"""Actions that can be authorized on a deployment resource."""
"""Actions authorized on a deployment resource."""
READ = "read"
WRITE = "write"
@ -32,12 +27,7 @@ class DeploymentAction(str, Enum):
class ProjectAction(str, Enum):
"""Actions that can be authorized on a project (folder) resource.
Projects are the OSS-side persistent name for the folder model; the
``project:`` Casbin object prefix matches the API surface and the
``project:{folder_id}`` domain used by ``_resolve_flow_domain``.
"""
"""Actions authorized on a project (folder) resource."""
READ = "read"
WRITE = "write"
@ -46,13 +36,7 @@ class ProjectAction(str, Enum):
class KnowledgeBaseAction(str, Enum):
"""Actions that can be authorized on a knowledge base resource.
Knowledge bases are name-keyed (``knowledge_base:{kb_name}``) rather than
UUID-keyed, but the action vocabulary mirrors other resources. ``ingest``
is a distinct verb because ingesting documents has a different cost and
permission posture than ordinary writes.
"""
"""Actions authorized on a knowledge base resource."""
READ = "read"
WRITE = "write"
@ -62,7 +46,7 @@ class KnowledgeBaseAction(str, Enum):
class VariableAction(str, Enum):
"""Actions that can be authorized on a variable resource."""
"""Actions authorized on a variable resource."""
READ = "read"
WRITE = "write"
@ -71,7 +55,7 @@ class VariableAction(str, Enum):
class FileAction(str, Enum):
"""Actions that can be authorized on a user-file resource (v2 files)."""
"""Actions authorized on a user-file resource (v2 files)."""
READ = "read"
WRITE = "write"
@ -80,12 +64,7 @@ class FileAction(str, Enum):
class ShareAction(str, Enum):
"""Actions that can be authorized on an authz_share row itself.
Shares are themselves authorizable: creating a share grants someone access
to a resource you own, so the action vocabulary lives in a dedicated enum
so audit rows can distinguish ``share:create`` from ``flow:create``.
"""
"""Actions authorized on an authz_share row."""
READ = "read"
CREATE = "create"

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

@ -1,26 +1,4 @@
"""Share-aware fetch helpers for guarded resource routes.
Phase 3 contract
----------------
Before Phase 3, route fetch helpers (``_read_flow``, ``get_flow_for_api_key_user``,
``get_deployment_db``, project reads in ``projects.py``) filtered every query by
``current_user.id``. That meant an enterprise plugin with a valid share grant
on a non-owned flow still saw 404 at the fetch layer before the route guard
could authorize the request.
These helpers fix that by branching on
:meth:`BaseAuthorizationService.supports_cross_user_fetch`:
* When the registered service reports ``True`` (enterprise Casbin), the row
is loaded by id alone. The route then invokes ``ensure_*_permission`` and
converts a plugin deny to **404** via :func:`deny_to_404` so non-shareholders
cannot probe UUIDs.
* When the registered service reports ``False`` (OSS pass-through default),
the helper keeps the existing owner-scoped query. That preserves the
strict-pass-through guarantee: enabling ``LANGFLOW_AUTHZ_ENABLED=true``
without an enterprise plugin must not silently widen cross-user visibility.
"""
"""Share-aware fetch helpers for authorization-guarded routes."""
from __future__ import annotations
@ -49,34 +27,9 @@ async def authorized_or_owner_scoped(
owner_column: InstrumentedAttribute,
owner_id: UUID,
) -> T | None:
"""Load a row by id when share-aware fetch is supported, else scope by owner.
Parameters
----------
session : AsyncSession
Active SQLAlchemy async session.
model : type[T]
SQLModel class to select from.
id_column : InstrumentedAttribute
Primary-key column attribute (e.g. ``Flow.id``).
resource_id : UUID
Primary-key value to look up.
owner_column : InstrumentedAttribute
Owner FK column attribute (e.g. ``Flow.user_id``).
owner_id : UUID
Caller's user id, used when the service does not support cross-user fetch.
Returns:
-------
The row, or ``None`` if no row matched. Routes should still raise 404 on
``None`` exactly as they did before.
"""
"""Load by id when cross-user fetch is supported; otherwise scope by owner."""
authz = get_authorization_service()
# Cross-user fetch requires BOTH the plugin capability AND enforcement
# to be enabled. If ``AUTHZ_ENABLED=false`` the route guards are no-ops
# (see ``ensure_permission``), so widening the fetch would let an
# enterprise plugin's capability silently expose other users' resources
# without any policy check. Gate on both.
# Require both plugin capability and AUTHZ_ENABLED before widening the query.
if await authz.supports_cross_user_fetch() and await authz.is_enabled():
stmt = select(model).where(id_column == resource_id)
else:
@ -85,20 +38,7 @@ 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.
For 403s, returns a fresh ``HTTPException(404, detail=detail)``.
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``.
"""
"""Map 403 from permission checks to 404; sanitize other error details."""
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)

19
src/backend/base/langflow/services/authorization/service.py
View File

@ -1,4 +1,4 @@
"""Langflow authorization service (OSS default — allows all; enterprise plugin enforces RBAC)."""
"""Langflow authorization service (OSS pass-through; plugins enforce RBAC)."""
from __future__ import annotations
@ -18,15 +18,10 @@ if TYPE_CHECKING:
class LangflowAuthorizationService(BaseAuthorizationService):
"""OSS authorization service (pass-through).
``ensure_*`` helpers still run when ``AUTHZ_ENABLED`` is True so routes stay wired,
but this implementation always allows. Register an enterprise
``authorization_service`` (e.g. Casbin) via ``lfx.toml`` for real enforcement.
"""
"""OSS pass-through authorization service (always allows)."""
def __init__(self, settings_service: SettingsService) -> None:
"""Initialize the service with a reference to the live settings service."""
"""Store the settings service reference."""
super().__init__()
self.settings_service = settings_service
self.set_ready()
@ -38,11 +33,11 @@ class LangflowAuthorizationService(BaseAuthorizationService):
return ServiceType.AUTHORIZATION_SERVICE.value
def _authz_settings(self) -> AuthSettings:
"""Return the live AuthSettings snapshot from the settings service."""
"""Return the live AuthSettings snapshot."""
return self.settings_service.auth_settings
async def is_enabled(self) -> bool:
"""Return True when AUTHZ_ENABLED is set in AuthSettings."""
"""Return True when AUTHZ_ENABLED is set."""
return self._authz_settings().AUTHZ_ENABLED
async def enforce(
@ -54,7 +49,7 @@ class LangflowAuthorizationService(BaseAuthorizationService):
act: str, # noqa: ARG002
context: dict[str, Any] | None = None, # noqa: ARG002
) -> bool:
"""Always allow — enterprise plugin overrides this for real enforcement."""
"""Allow every request in the OSS default."""
return True
async def batch_enforce(
@ -65,5 +60,5 @@ class LangflowAuthorizationService(BaseAuthorizationService):
requests: Sequence[tuple[str, str]],
context: dict[str, Any] | None = None, # noqa: ARG002
) -> list[bool]:
"""Return a True list matching the request count."""
"""Return True for each request."""
return [True] * len(requests)

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

@ -1,26 +1,4 @@
"""Authorization helpers for API routes.
Phase 1 contract (this PR)
---------------------------
Route guards (``ensure_flow_permission``, ``ensure_deployment_permission``,
``ensure_project_permission``) sit on top of fetch helpers that still scope
queries by ``current_user.id`` — see ``_read_flow`` in ``flows_helpers.py``,
``get_flow_for_api_key_user`` in ``endpoints.py``, ``get_deployment`` in
``deployment.crud``, and the owner-scoped folder reads in ``projects.py``.
That means an enterprise plugin which would otherwise grant a non-owner
shared / read / write / execute access to a flow, folder, or deployment
**still returns 404 at the fetch layer before** ``ensure_*_permission`` can
authorize the request. Cross-user enforcement therefore lands in Phase 3
alongside ``authz_share`` CRUD APIs and the share-aware fetch helpers that
load by id first and convert plugin denies to 404 to preserve UUID-privacy.
In Phase 1 the OSS pass-through allows all and the owner-scoped fetch is the
only effective gate; in Phase 2 (this PR) the guards exist, are wired, and
emit audit rows, but cross-user access is not yet reachable. See PR #13153
description and the Phase 3 prerequisite captured in the planning notes.
"""
"""Authorization helpers for guarded API routes."""
from __future__ import annotations
@ -51,12 +29,12 @@ if TYPE_CHECKING:
T = TypeVar("T")
# Audit result strings — kept here so callers and tests share the vocabulary.
# Shared audit result vocabulary.
_AUDIT_ALLOW = "allow"
_AUDIT_DENY = "deny"
_AUDIT_OWNER_OVERRIDE = "owner_override"
# Context keys that name the resource owner — used by audit-detail extraction.
# Resource-owner keys included in audit details.
_OWNER_CONTEXT_KEYS = (
"flow_user_id",
"deployment_user_id",
@ -67,7 +45,7 @@ _OWNER_CONTEXT_KEYS = (
"share_user_id",
)
# Action enum types we coerce to their string value.
# Action enums coerced to string values.
_ACTION_ENUMS = (
FlowAction,
DeploymentAction,
@ -117,29 +95,19 @@ 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.
# Strong refs for in-flight audit tasks (awaited on shutdown).
_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.
"""
"""Await in-flight audit writes during shutdown (bounded by timeout)."""
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.
# Log task failures; never block shutdown.
for task in done:
exc = task.exception() if not task.cancelled() else None
if exc is not None:
@ -154,22 +122,10 @@ async def audit_decision(
result: str,
details: dict[str, Any] | None = None,
) -> None:
"""Write an AuthzAuditLog row, fire-and-forget.
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. 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.
"""
"""Schedule an AuthzAuditLog write (fire-and-forget)."""
settings = get_settings_service()
auth_settings = settings.auth_settings
# 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.
# Audit can run while AUTHZ_ENABLED is false.
if not getattr(auth_settings, "AUTHZ_AUDIT_ENABLED", True):
return
@ -207,23 +163,15 @@ async def ensure_permission(
act: str,
context: dict[str, Any] | None = None,
) -> None:
"""Raise HTTP 403 if the user is not allowed to perform the action.
Writes an audit row on both allow and deny paths.
"""
"""Raise HTTP 403 when the user may not perform the action (audited)."""
settings = get_settings_service()
if not settings.auth_settings.AUTHZ_ENABLED:
return
authz = get_authorization_service()
# User-derived auth fields (e.g. is_superuser) must remain authoritative; caller
# context is merged first so it cannot overwrite them.
# Caller context first; user auth fields cannot be overwritten.
merged_context = {**(context or {}), **_auth_context(user)}
# 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.
# Fail closed when enforce() raises (deny + audit, not HTTP 500).
audit_action = f"{obj.split(':', 1)[0]}:{act}" if ":" in obj else act
audit_details = {"domain": domain}
for owner_key in _OWNER_CONTEXT_KEYS:
@ -267,27 +215,7 @@ async def ensure_permission(
def _resolve_casbin_domain(workspace_id: UUID | None, scope_id: UUID | None) -> str:
"""Pick the most specific Casbin domain for a resource check.
``scope_id`` is the folder/project id for flows and deployments, or ``None``
when the resource itself is the project (the project helper passes ``None``
so the domain falls back to workspace).
Precedence (inner → outer):
1. ``project:{scope_id}`` when set,
2. ``workspace:{workspace_id}`` when only a workspace is set,
3. ``"*"`` when neither is set.
Enterprise Casbin policies link the two via ``g2`` (e.g.
``g2, project:xyz, workspace:abc``) so that when the enforcer checks
against the **project** domain, both project-scoped and workspace-scoped
role grants match (workspace grants flow down to projects via g2).
Passing the workspace domain when a project is known would make
project-scoped grants invisible because g2 is directional — children
inherit from parents, not the other way round. ``workspace_id`` is also
forwarded in the enforce context for plugins that prefer ABAC-style
matchers.
"""
"""Resolve policy domain: project scope, then workspace, else ``*``."""
if scope_id is not None:
return f"project:{scope_id}"
if workspace_id is not None:
@ -295,7 +223,7 @@ def _resolve_casbin_domain(workspace_id: UUID | None, scope_id: UUID | None) ->
return "*"
# Backward-compatible alias for callers/tests that imported the old name.
# Backward-compatible alias.
_resolve_flow_domain = _resolve_casbin_domain
@ -309,23 +237,10 @@ async def _ensure_resource_permission(
resolved_domain: str,
extra_context: dict[str, Any],
) -> None:
"""Shared core for the per-resource ``ensure_*_permission`` helpers.
Builds the canonical ``{resource_type}:{id}|*`` object key, short-circuits
on owner override (audited as ``owner_override``), and otherwise delegates
to ``ensure_permission``. ``extra_context`` is forwarded verbatim — callers
own the key names so each resource type's audit row stays self-describing.
``resource_id`` accepts either a UUID (flows, deployments, projects, files,
variables) or a string slug (knowledge bases are name-keyed).
"""
"""Build object key, apply owner override, else delegate to ensure_permission."""
obj = f"{resource_type}:{resource_id}" if resource_id else f"{resource_type}:*"
# 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:
# ``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}",
@ -354,15 +269,7 @@ async def ensure_flow_permission(
folder_id: UUID | None = None,
domain: str | None = None,
) -> None:
"""Check flow-scoped permission with workspace/project domain + owner override.
The flow owner is always allowed (audited as ``owner_override``). Otherwise
delegates through :func:`ensure_permission` with the canonical Casbin tuple
``(user, domain, obj=flow:{id}|flow:*, act=<FlowAction>)`` where the domain
follows :func:`_resolve_casbin_domain`. Both ``workspace_id`` and
``folder_id`` are forwarded in the context dict so the enterprise plugin
can use whichever fits its policy model.
"""
"""Check flow permission (owner override, then plugin enforce)."""
await _ensure_resource_permission(
user,
resource_type="flow",
@ -388,11 +295,7 @@ async def ensure_deployment_permission(
project_id: UUID | None = None,
domain: str | None = None,
) -> None:
"""Check deployment-scoped permission with workspace/project domain + owner override.
Deployments use ``project_id`` (folder row) for the Casbin project domain, same as
flows use ``folder_id``. The deployment owner may always access their deployment.
"""
"""Check deployment permission (owner override, then plugin enforce)."""
await _ensure_resource_permission(
user,
resource_type="deployment",
@ -417,13 +320,7 @@ async def ensure_project_permission(
workspace_id: UUID | None = None,
domain: str | None = None,
) -> None:
"""Check project-scoped permission with workspace domain + owner override.
Projects are the OSS persistent name for folders. The Casbin object is
``project:{project_id}`` (or ``project:*`` for list/create). The domain
resolves to ``workspace:{workspace_id}`` when set, otherwise ``*`` — the
project itself is the resource, not the scope.
"""
"""Check project (folder) permission (owner override, then plugin enforce)."""
await _ensure_resource_permission(
user,
resource_type="project",
@ -449,17 +346,7 @@ async def ensure_knowledge_base_permission(
project_id: UUID | None = None,
domain: str | None = None,
) -> None:
"""Check knowledge-base-scoped permission with owner override.
Knowledge bases are tracked by ``KnowledgeBaseRecord`` (UUID primary key,
``(user_id, name)`` unique). The Casbin object slug is
``knowledge_base:{kb_id}`` so it matches the ``authz_share.resource_id``
column (UUID) without an extra translation. ``kb_name`` is forwarded in
the audit context for human-readable debugging.
For create-style checks (no KB row yet), pass ``kb_id=None``; the slug
becomes ``knowledge_base:*``.
"""
"""Check knowledge-base permission (owner override, then plugin enforce)."""
await _ensure_resource_permission(
user,
resource_type="knowledge_base",
@ -486,7 +373,7 @@ async def ensure_variable_permission(
workspace_id: UUID | None = None,
domain: str | None = None,
) -> None:
"""Check variable-scoped permission with owner override."""
"""Check variable permission (owner override, then plugin enforce)."""
await _ensure_resource_permission(
user,
resource_type="variable",
@ -510,7 +397,7 @@ async def ensure_file_permission(
workspace_id: UUID | None = None,
domain: str | None = None,
) -> None:
"""Check file-scoped permission (v2 user files) with owner override."""
"""Check file permission (owner override, then plugin enforce)."""
await _ensure_resource_permission(
user,
resource_type="file",
@ -533,12 +420,7 @@ async def ensure_share_permission(
share_user_id: UUID | None = None,
domain: str | None = None,
) -> None:
"""Check authz_share-scoped permission with owner override.
A share row is "owned" by the user who created it (``created_by``). The
resource owner is therefore always allowed to administer their own
shares; enterprise plugins decide everything else.
"""
"""Check share-row permission (owner override, then plugin enforce)."""
await _ensure_resource_permission(
user,
resource_type="share",
@ -563,32 +445,7 @@ async def filter_visible_resources(
owner_extractor: Callable[[T], UUID | None] | None = None,
act: FlowAction | str = FlowAction.READ,
) -> list[T]:
"""Return the subset of `candidates` that the user is allowed to read.
No-op when ``AUTHZ_ENABLED=false`` — returns the input list unchanged.
Plumbing for list endpoints; the OSS pass-through always returns the full
list, so calling this is safe to add ahead of enterprise plugin rollout.
``domain_extractor`` lets callers compute a per-candidate domain string
(typically via :func:`_resolve_casbin_domain`). Without it, every request
in the batch evaluates against the same ``domain``, which makes
project-scoped policy grants invisible when candidates live in different
workspaces or projects. With it, candidates are grouped by their resolved
domain and the enforcer is called once per group, so each request is
evaluated against the right Casbin tuple.
``owner_extractor`` returns the owner UUID for each candidate. Items
owned by the calling user are force-included **without** consulting the
enforcer, mirroring the owner-override short-circuit in
:func:`_ensure_resource_permission`. Without this, an enterprise plugin
that lacks an explicit owner policy would hide a caller's own rows from
a list view even though the same caller can read each row directly via
the single-resource guard — list and direct read would disagree, which
is the symptom this parameter is here to prevent.
The OSS pass-through ignores ``domain`` entirely and returns
``[True] * len(requests)`` for either path.
"""
"""Return candidates the user may read (no-op when AUTHZ_ENABLED is false)."""
settings = get_settings_service()
if not settings.auth_settings.AUTHZ_ENABLED or not candidates:
return candidates
@ -598,8 +455,7 @@ async def filter_visible_resources(
act_str = _coerce_action(act)
user_id = getattr(user, "id", None)
# Owner-override partition. Items the caller owns skip the enforcer
# entirely (matches the direct-read owner short-circuit).
# Owned rows skip batch_enforce (matches direct-read owner override).
owned_indices: set[int] = set()
enforce_indices: list[int] = []
enforce_items: list[T] = []
@ -615,13 +471,8 @@ 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.
# Single-domain batch_enforce.
requests = [(f"{resource_type}:{extractor(item)}", act_str) for item in enforce_items]
results = await authz.batch_enforce(
user_id=user_id,
@ -632,10 +483,7 @@ async def filter_visible_resources(
for original_index, allowed in zip(enforce_indices, results, strict=True):
decisions[original_index] = allowed
else:
# Group candidates by their resolved domain so each batch_enforce
# call evaluates against a single Casbin tuple. Preserves input
# order on output by carrying the original index through the
# per-domain bucket.
# One batch_enforce per resolved domain.
buckets: dict[str, list[tuple[int, T]]] = {}
for original_index, item in zip(enforce_indices, enforce_items, strict=True):
buckets.setdefault(domain_extractor(item), []).append((original_index, item))

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

@ -1,17 +1,4 @@
"""Authorization (RBAC) plugin tables.
Tables are owned by Langflow OSS (Alembic migrations). Enterprise plugins populate
policy data and may use the Casbin SQLAlchemy adapter against ``casbin_rule``.
Partial unique indexes use ``postgresql_where`` / ``sqlite_where``. The project
targets PostgreSQL and SQLite only; on any other dialect those kwargs are
silently ignored and the indexes would lose their uniqueness guarantee — see
``pyproject.toml`` for the supported driver set.
Timestamps standardize on ``DateTime(timezone=True)`` to match the rest of the
project (``deployment/model.py``, ``api_key/model.py``, etc.) and avoid TZ-naive
columns on Postgres that strip ``tzinfo`` on write.
"""
"""Authorization (RBAC) tables (Alembic-owned; plugins populate policy data)."""
from datetime import datetime, timezone
from enum import Enum
@ -53,11 +40,7 @@ def _tz_column(*, nullable: bool = False, index: bool = False) -> Column:
class CasbinRule(SQLModel, table=True): # type: ignore[call-arg]
"""Casbin policy storage (SQLAlchemy adapter compatible).
Casbin's loader filters by ``ptype`` on every ``load_policy()`` /
``AddPolicy()``; the index is required for non-trivial policy volumes.
"""
"""Policy rule storage (ptype-indexed for loader queries)."""
__tablename__ = "casbin_rule"
__table_args__ = (Index("ix_casbin_rule_ptype", "ptype"),)
@ -73,7 +56,7 @@ class CasbinRule(SQLModel, table=True): # type: ignore[call-arg]
class AuthzRole(SQLModel, table=True): # type: ignore[call-arg]
"""Role metadata for admin UI; enforceable policies live in ``casbin_rule``."""
"""Role metadata for admin UI."""
__tablename__ = "authz_role"
@ -130,7 +113,7 @@ class AuthzRoleAssignment(SQLModel, table=True): # type: ignore[call-arg]
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
# Composite index for enforce() lookups.
# single-column scans.
Index(
"ix_authz_role_assignment_user_domain",
@ -212,7 +195,7 @@ class AuthzShare(SQLModel, table=True): # type: ignore[call-arg]
# a bitmap-AND of two single-column indexes.
Index("ix_authz_share_resource", "resource_type", "resource_id"),
# Bound scope and permission_level to known enum values at the DB
# level. Without this, an enterprise plugin (or a manual INSERT) could
# Without this, a plugin (or manual INSERT) could
# write a typo like ``scope='PRIVATE'`` that silently bypasses the
# partial unique indexes (which match on the lowercase form).
CheckConstraint(
@ -284,7 +267,7 @@ class AuthzEditLock(SQLModel, table=True): # type: ignore[call-arg]
expires_at: datetime = Field(sa_column=_tz_column(index=True))
# TODO: AuthzAuditLog is append-only and unbounded. At enterprise scale this
# TODO: AuthzAuditLog is append-only and unbounded. At large scale this
# table will outgrow practical query windows. Decide between (a) Postgres native
# partitioning by ``timestamp`` (monthly/quarterly), (b) an out-of-band
# archival/TTL job, or (c) SIEM export per design note §5.3 Phase 5. The

2
src/backend/base/langflow/services/database/models/deployment/crud.py
View File

@ -115,7 +115,7 @@ async def get_deployment(
"""Load a deployment by id, with share-aware fetch when supported.
When the registered authorization service supports cross-user fetch
(enterprise Casbin), the deployment is loaded by id alone and the route's
(authorization plugin), the deployment is loaded by id alone and the route's
``ensure_deployment_permission`` decides whether the caller may see it.
Otherwise the query stays owner-scoped so the OSS pass-through default
cannot widen visibility.

2
src/backend/tests/unit/alembic/test_seed_authz_system_roles.py
View File

@ -49,7 +49,7 @@ def test_admin_has_share_administration_permissions():
def test_permission_slugs_use_resource_action_format():
"""Slugs must match ``{resource}:{action}`` so enterprise PolicySync can split them."""
"""Slugs must match ``{resource}:{action}`` so PolicySync can split them."""
for _, _, permissions in _MIGRATION._SYSTEM_ROLES:
for slug in permissions:
assert slug.count(":") == 1, slug

21
src/backend/tests/unit/api/v1/test_authz_share_routes.py
View File

@ -1,15 +1,4 @@
"""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.
"""
"""Route-level tests for the OSS share-administration owner floor."""
from __future__ import annotations
@ -345,13 +334,13 @@ async def test_delete_share_allows_owner_under_oss_passthrough(patch_authz, sile
# --------------------------------------------------------------------------- #
# Floor behavior when the enterprise plugin is active
# Floor behavior when the authorization 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.
async def test_floor_is_skipped_when_plugin_active(patch_authz, silence_audit): # noqa: ARG001
"""OSS floor is skipped when the authorization 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
@ -363,7 +352,7 @@ async def test_floor_is_skipped_when_enterprise_plugin_active(patch_authz, silen
patch_authz(cross_user=True, enabled=True)
owner = _make_user()
delegate = _make_user() # non-owner, but allowed by enterprise policy
delegate = _make_user() # non-owner, but allowed by plugin policy
flow = SimpleNamespace(id=uuid4(), user_id=owner.id)
session = _FakeAsyncSession({(Flow, flow.id): flow})
payload = _payload_for(flow.id)

8
src/backend/tests/unit/services/authorization/test_actions.py
View File

@ -12,8 +12,8 @@ from langflow.services.authorization.actions import (
)
def test_flow_action_values_match_casbin_strings():
"""Casbin policies use lowercase action strings; the enum values must match."""
def test_flow_action_values_match_policy_strings():
"""Policy action strings match FlowAction enum values."""
assert FlowAction.READ.value == "read"
assert FlowAction.WRITE.value == "write"
assert FlowAction.CREATE.value == "create"
@ -34,8 +34,8 @@ def test_flow_action_is_iterable_and_complete():
assert values == {"read", "write", "create", "delete", "execute", "deploy"}
def test_deployment_action_values_match_casbin_strings():
"""Casbin policies use lowercase action strings; the enum values must match."""
def test_deployment_action_values_match_policy_strings():
"""Policy action strings match DeploymentAction enum values."""
assert DeploymentAction.READ.value == "read"
assert DeploymentAction.WRITE.value == "write"
assert DeploymentAction.CREATE.value == "create"

2
src/backend/tests/unit/services/authorization/test_authorization_service.py
View File

@ -31,7 +31,7 @@ async def test_enforce_allows_all_when_disabled(authz_service):
@pytest.mark.anyio
async def test_enforce_allows_non_superuser_when_enabled():
"""OSS stub does not deny; enterprise Casbin replaces this service for enforcement."""
"""OSS stub does not deny; authorization plugin replaces this service for enforcement."""
settings = SimpleNamespace(
auth_settings=SimpleNamespace(
AUTHZ_ENABLED=True,

9
src/backend/tests/unit/services/authorization/test_capability_flag.py
View File

@ -1,9 +1,4 @@
"""Tests for the BaseAuthorizationService cross-user-fetch capability flag.
The capability is the single switch that decides whether share-aware fetch
helpers load resources by id alone. The OSS pass-through must never opt in;
enterprise plugins may.
"""
"""Tests for the cross-user-fetch capability flag."""
from __future__ import annotations
@ -46,7 +41,7 @@ async def test_langflow_pass_through_does_not_support_cross_user_fetch():
@pytest.mark.anyio
async def test_subclass_can_opt_in():
"""Enterprise plugins flip ``SUPPORTS_CROSS_USER_FETCH=True``; the base accepts it."""
"""Authorization plugins flip ``SUPPORTS_CROSS_USER_FETCH=True``; the base accepts it."""
class _Enterprise(LangflowAuthorizationService):
SUPPORTS_CROSS_USER_FETCH = True

12
src/backend/tests/unit/services/authorization/test_fetch.py
View File

@ -1,10 +1,4 @@
"""Tests for the share-aware fetch helpers in ``services/authorization/fetch``.
The OSS pass-through must NEVER cause cross-user fetch to succeed — only an
enterprise service that opts in via ``SUPPORTS_CROSS_USER_FETCH=True`` is
allowed to load resources by id alone. These tests pin both branches so the
strict-pass-through contract from the design note can't regress.
"""
"""Tests for share-aware fetch helpers (strict pass-through contract)."""
from __future__ import annotations
@ -29,7 +23,7 @@ class _StubService(BaseAuthorizationService):
Cross-user fetch in the helper requires *both* the plugin capability
*and* ``is_enabled()`` to be true, so tests opt into both together via
``supports_cross_user`` to mirror an enterprise plugin with
``supports_cross_user`` to mirror an authorization plugin with
``AUTHZ_ENABLED=true``.
"""
@ -97,7 +91,7 @@ async def test_owner_scoped_when_service_does_not_support_cross_user_fetch():
@pytest.mark.anyio
async def test_id_only_when_service_supports_cross_user_fetch():
"""Enterprise plugin loads by id alone; route guard then decides access."""
"""Authorization plugin loads by id alone; route guard then decides access."""
session = _FakeSession(returns=object())
service = _StubService(supports_cross_user=True)
with patch(

18
src/backend/tests/unit/services/authorization/test_flow_route_guards.py
View File

@ -1,10 +1,4 @@
"""Regression tests asserting every flow CRUD route is wired through ensure_flow_permission.
These are intentionally source-level checks rather than full FastAPI integration tests:
the helper-level behavior is covered exhaustively in ``test_utils.py``; what these
tests prevent is the route being silently dropped or reverting to a bare action
string. A future PR will add a full app fixture + stub enterprise plugin.
"""
"""Source-level tests that flow routes call ensure_flow_permission."""
from __future__ import annotations
@ -192,7 +186,7 @@ def test_update_flow_authorizes_destination_on_move(routes):
Without this, a caller could write to a flow in scope A and move it into
scope B even when they lack permission to write at B (e.g., share-based
access in Phase 3 or workspace-scoped roles in enterprise).
access in Phase 3 or workspace-scoped roles in plugin).
"""
func = routes["update_flow"]
assert _has_destination_check(
@ -237,7 +231,7 @@ def test_get_note_translations_is_owner_scoped_and_guarded(routes):
filter and no ensure_flow_permission, so any authenticated user could read
note text from any flow by guessing the UUID. Fix scopes the fetch through
``_read_flow(..., user_id=current_user.id)`` and adds an explicit
``ensure_flow_permission(FlowAction.READ, ...)`` so enterprise plugins can
``ensure_flow_permission(FlowAction.READ, ...)`` so authorization plugins can
extend visibility via shares.
"""
func = routes["get_note_translations"]
@ -264,7 +258,7 @@ def test_read_flows_list_uses_filter_visible_resources(routes):
"""GET /flows/ applies filter_visible_resources on BOTH the get_all and paginated paths.
The list helper drops items the user can't read. In OSS pass-through it
returns the input unchanged; the enterprise plugin uses batch_enforce to
returns the input unchanged; the authorization plugin uses batch_enforce to
honor role + share grants. Per-item ensure_flow_permission is intentionally
NOT used here — filtering is the right primitive for list endpoints.
@ -325,8 +319,8 @@ def test_check_flow_user_permission_is_gone():
"""The standalone owner-only ``check_flow_user_permission`` helper has been removed.
It duplicated the work that ``ensure_flow_permission`` now does at the
route boundary, AND it would have rejected legitimate enterprise execute
grants on shared flows (a real bug under Casbin enforcement). The new
route boundary, AND it would have rejected legitimate plugin execute
grants on shared flows (a real bug under policy engine enforcement). The new
contract is "every ``_run_flow_internal`` caller authorizes EXECUTE first";
leaving a downstream owner-only check would silently re-introduce the
regression on any future route that wires up the internal helper without

20
src/backend/tests/unit/services/authorization/test_route_guard_regressions.py
View File

@ -1,18 +1,4 @@
"""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.
"""
"""Source-level regression tests for authorization route guard wiring."""
from __future__ import annotations
@ -69,7 +55,7 @@ def test_read_flows_paginated_branch_passes_owner_extractor(flows_routes):
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
on the paginated branch lets an authorization plugin without an explicit
owner-allow policy hide the caller's own flows when paginating.
"""
func = flows_routes["read_flows"]
@ -85,7 +71,7 @@ def test_read_flows_paginated_branch_passes_owner_extractor(flows_routes):
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
With an authorization 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()``.
"""

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

@ -264,13 +264,7 @@ async def test_ensure_flow_permission_falls_back_to_project_domain(monkeypatch,
@pytest.mark.anyio
async def test_ensure_flow_permission_project_beats_workspace(monkeypatch, fake_user):
"""When both workspace_id and folder_id are set, the project domain wins.
Casbin g2 is directional (children inherit from parents). Passing the
project domain lets workspace-scoped grants flow down via g2 *and* keeps
project-scoped grants visible. Passing workspace would make project
grants invisible — that was the original (buggy) behavior.
"""
"""Project domain wins when both workspace_id and folder_id are set."""
_install_settings(monkeypatch, authz_enabled=True)
service = _StubAuthorizationService(allow=True)
_install_authz(monkeypatch, service)
@ -306,13 +300,7 @@ async def test_ensure_flow_permission_wildcard_domain_when_neither_set(monkeypat
def test_resolve_casbin_domain_precedence():
"""Unit test the precedence rule directly (project > workspace > '*').
Project is more specific than workspace; passing it lets workspace grants
flow down via Casbin g2 inheritance while keeping project-scoped grants
visible to the enforcer. Helper renamed from ``_resolve_flow_domain`` to
``_resolve_casbin_domain`` because it serves flows AND deployments.
"""
"""Domain precedence: project > workspace > '*'."""
ws, scope = uuid4(), uuid4()
assert authz_utils._resolve_casbin_domain(workspace_id=ws, scope_id=scope) == f"project:{scope}"
assert authz_utils._resolve_casbin_domain(workspace_id=ws, scope_id=None) == f"workspace:{ws}"
@ -601,7 +589,7 @@ async def test_ensure_permission_fails_closed_on_plugin_exception(monkeypatch, f
class _BrokenPlugin:
async def enforce(self, **_kwargs):
msg = "casbin db down"
msg = "policy store down"
raise RuntimeError(msg)
async def batch_enforce(self, **_kwargs):
@ -701,7 +689,7 @@ async def test_filter_visible_resources_groups_by_extracted_domain(monkeypatch,
"""With ``domain_extractor`` set, batch_enforce is called once per unique domain.
Each call sees only the candidates that resolved to that domain, so the
enterprise plugin evaluates each candidate against the right Casbin tuple
authorization plugin evaluates each candidate against the right policy tuple
(the single-domain default would force every candidate through the same
wildcard domain, hiding project-scoped grants).
"""
@ -749,7 +737,7 @@ async def test_filter_visible_resources_owner_override_skips_enforcer(monkeypatc
"""Items owned by the caller are force-included without consulting the enforcer.
Mirrors the owner-override short-circuit in ``_ensure_resource_permission``
so list and direct-read agree under enterprise enforcement. Without this,
so list and direct-read agree under plugin enforcement. Without this,
a deny-all plugin would hide the caller's own rows from the listing
response while letting them read the same rows directly.
"""
@ -834,7 +822,7 @@ async def test_deployment_owner_override_skips_enforce(monkeypatch, fake_user):
@pytest.mark.anyio
async def test_kb_permission_uses_kb_id_object_slug(monkeypatch, fake_user):
"""KB shares store UUIDs; the Casbin obj slug must use ``kb_id``."""
"""KB shares store UUIDs; the policy object slug must use ``kb_id``."""
_install_settings(monkeypatch, authz_enabled=True)
service = _StubAuthorizationService(allow=True)
_install_authz(monkeypatch, service)

2
src/backend/tests/unit/test_authz_dry_run_cli.py
View File

@ -183,7 +183,7 @@ def test_deny_writes_blocks_writes_allows_reads():
def test_domain_is_project_prefixed():
"""The recorded domain uses the project:{uuid} form for in-scope scenarios.
``_resolve_flow_domain`` prefers project over workspace because Casbin g2
``_resolve_flow_domain`` prefers project over workspace because g2
inheritance is directional — passing the more specific domain lets both
workspace-scoped and project-scoped grants match. The dry-run CLI passes
both ids on every scenario, so every audited row resolves to ``project:``.

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

@ -15,33 +15,7 @@ if TYPE_CHECKING:
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.
"""
"""Documented keys for the enforce/batch_enforce context dict."""
is_superuser: bool
flow_user_id: _UUID | None
@ -56,20 +30,11 @@ class AuthzContext(TypedDict, total=False):
class BaseAuthorizationService(Service, abc.ABC):
"""Abstract base class for authorization (RBAC) services.
Authentication establishes identity; authorization decides what that identity may do.
Enterprise plugins provide Casbin-backed implementations; OSS ships a no-op default.
"""
"""Abstract base class for authorization (RBAC) services."""
name = ServiceType.AUTHORIZATION_SERVICE.value
# Capability flag. Implementations that can authorize non-owner access (share
# grants, domain roles) set this to True so share-aware fetch helpers load
# resources by id and rely on enforce() to gate access. The OSS pass-through
# leaves this False so fetch helpers keep their owner-scoped queries — that
# way enabling AUTHZ_ENABLED without an enterprise plugin does not silently
# widen visibility.
# True when the service can authorize non-owner access (share-aware fetch).
SUPPORTS_CROSS_USER_FETCH: ClassVar[bool] = False
async def supports_cross_user_fetch(self) -> bool:

6
src/lfx/src/lfx/services/authorization/service.py
View File

@ -16,11 +16,7 @@ from lfx.services.schema import ServiceType
@register_service(ServiceType.AUTHORIZATION_SERVICE)
class AuthorizationService(BaseAuthorizationService):
"""Default LFX authorization service that permits all actions.
Langflow registers its own implementation at startup. Enterprise plugins may
override via the ``lfx.services`` entry point with ``authorization_service``.
"""
"""Default LFX authorization service that permits all actions."""
def __init__(self) -> None:
"""Mark the no-op service as ready immediately (no external resources)."""

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

@ -133,10 +133,10 @@ class AuthSettings(BaseSettings):
)
"""Path to YAML configuration file for SSO settings. Contains provider-specific configuration."""
# Authorization (RBAC) feature flags — enforcement implemented by enterprise Casbin plugin
# Authorization (RBAC) feature flags — enforcement via authorization_service plugin
AUTHZ_ENABLED: bool = Field(
default=False,
description="Enable authorization enforcement. Requires an authorization_service plugin (e.g. enterprise).",
description="Enable authorization enforcement. Requires an authorization_service plugin.",
)
AUTHZ_SUPERUSER_BYPASS: bool = Field(
default=True,