From 2ac7568f0152f84f54e145e86040b116a1cb67f2 Mon Sep 17 00:00:00 2001 From: himavarshagoutham Date: Tue, 26 May 2026 10:37:26 -0400 Subject: [PATCH] docs(authz): shorten docstrings and drop enterprise/Casbin comments Use one-line module and helper docstrings across authorization code and route guards; keep table/type identifiers unchanged. --- .../7c8d9e0f1a2b_authz_foundations.py | 33 +-- .../8d3a1f9c2e0b_seed_authz_system_roles.py | 21 +- .../base/langflow/api/utils/flow_utils.py | 2 +- .../base/langflow/api/v1/authz_shares.py | 120 ++--------- src/backend/base/langflow/api/v1/chat.py | 4 +- .../base/langflow/api/v1/deployments.py | 2 +- src/backend/base/langflow/api/v1/endpoints.py | 4 +- src/backend/base/langflow/api/v1/flows.py | 60 +----- .../base/langflow/api/v1/flows_helpers.py | 2 +- .../base/langflow/api/v1/knowledge_bases.py | 8 +- .../api/v1/mappers/deployments/helpers.py | 2 +- src/backend/base/langflow/api/v1/monitor.py | 11 +- .../base/langflow/api/v1/openai_responses.py | 2 +- src/backend/base/langflow/api/v1/projects.py | 8 +- .../langflow/api/v1/schemas/authz_shares.py | 41 +--- src/backend/base/langflow/api/v2/files.py | 4 +- src/backend/base/langflow/api/v2/workflow.py | 2 +- .../base/langflow/cli/authz_dry_run.py | 10 +- src/backend/base/langflow/helpers/flow.py | 38 +--- src/backend/base/langflow/plugin_routes.py | 2 +- .../services/authorization/__init__.py | 2 +- .../services/authorization/actions.py | 35 +-- .../langflow/services/authorization/fetch.py | 68 +----- .../services/authorization/service.py | 19 +- .../langflow/services/authorization/utils.py | 204 +++--------------- .../services/database/models/auth/authz.py | 29 +-- .../database/models/deployment/crud.py | 2 +- .../alembic/test_seed_authz_system_roles.py | 2 +- .../unit/api/v1/test_authz_share_routes.py | 21 +- .../services/authorization/test_actions.py | 8 +- .../test_authorization_service.py | 2 +- .../authorization/test_capability_flag.py | 9 +- .../unit/services/authorization/test_fetch.py | 12 +- .../authorization/test_flow_route_guards.py | 18 +- .../test_route_guard_regressions.py | 20 +- .../unit/services/authorization/test_utils.py | 24 +-- .../tests/unit/test_authz_dry_run_cli.py | 2 +- .../src/lfx/services/authorization/base.py | 41 +--- .../src/lfx/services/authorization/service.py | 6 +- src/lfx/src/lfx/services/settings/auth.py | 4 +- 40 files changed, 168 insertions(+), 736 deletions(-) diff --git a/src/backend/base/langflow/alembic/versions/7c8d9e0f1a2b_authz_foundations.py b/src/backend/base/langflow/alembic/versions/7c8d9e0f1a2b_authz_foundations.py index ffd9f7a5cc..ce25d1396b 100644 --- a/src/backend/base/langflow/alembic/versions/7c8d9e0f1a2b_authz_foundations.py +++ b/src/backend/base/langflow/alembic/versions/7c8d9e0f1a2b_authz_foundations.py @@ -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')", diff --git a/src/backend/base/langflow/alembic/versions/8d3a1f9c2e0b_seed_authz_system_roles.py b/src/backend/base/langflow/alembic/versions/8d3a1f9c2e0b_seed_authz_system_roles.py index 2b195e5751..41f2e10b29 100644 --- a/src/backend/base/langflow/alembic/versions/8d3a1f9c2e0b_seed_authz_system_roles.py +++ b/src/backend/base/langflow/alembic/versions/8d3a1f9c2e0b_seed_authz_system_roles.py @@ -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 diff --git a/src/backend/base/langflow/api/utils/flow_utils.py b/src/backend/base/langflow/api/utils/flow_utils.py index 34384aab84..54500614fa 100644 --- a/src/backend/base/langflow/api/utils/flow_utils.py +++ b/src/backend/base/langflow/api/utils/flow_utils.py @@ -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) diff --git a/src/backend/base/langflow/api/v1/authz_shares.py b/src/backend/base/langflow/api/v1/authz_shares.py index df9f484553..2403e8194d 100644 --- a/src/backend/base/langflow/api/v1/authz_shares.py +++ b/src/backend/base/langflow/api/v1/authz_shares.py @@ -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: diff --git a/src/backend/base/langflow/api/v1/chat.py b/src/backend/base/langflow/api/v1/chat.py index 05dfe2968a..96bbaa9b5c 100644 --- a/src/backend/base/langflow/api/v1/chat.py +++ b/src/backend/base/langflow/api/v1/chat.py @@ -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, diff --git a/src/backend/base/langflow/api/v1/deployments.py b/src/backend/base/langflow/api/v1/deployments.py index 7561814733..0e7b9cd2f7 100644 --- a/src/backend/base/langflow/api/v1/deployments.py +++ b/src/backend/base/langflow/api/v1/deployments.py @@ -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 — diff --git a/src/backend/base/langflow/api/v1/endpoints.py b/src/backend/base/langflow/api/v1/endpoints.py index 9cf2e8dee4..26edc9bc3c 100644 --- a/src/backend/base/langflow/api/v1/endpoints.py +++ b/src/backend/base/langflow/api/v1/endpoints.py @@ -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() diff --git a/src/backend/base/langflow/api/v1/flows.py b/src/backend/base/langflow/api/v1/flows.py index b6af6af4ff..bf2bbeaac2 100644 --- a/src/backend/base/langflow/api/v1/flows.py +++ b/src/backend/base/langflow/api/v1/flows.py @@ -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, diff --git a/src/backend/base/langflow/api/v1/flows_helpers.py b/src/backend/base/langflow/api/v1/flows_helpers.py index a5b5c7e899..7df6d9a587 100644 --- a/src/backend/base/langflow/api/v1/flows_helpers.py +++ b/src/backend/base/langflow/api/v1/flows_helpers.py @@ -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. """ diff --git a/src/backend/base/langflow/api/v1/knowledge_bases.py b/src/backend/base/langflow/api/v1/knowledge_bases.py index ce96cf3ee4..c8837c050c 100644 --- a/src/backend/base/langflow/api/v1/knowledge_bases.py +++ b/src/backend/base/langflow/api/v1/knowledge_bases.py @@ -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: diff --git a/src/backend/base/langflow/api/v1/mappers/deployments/helpers.py b/src/backend/base/langflow/api/v1/mappers/deployments/helpers.py index f6d856fb1a..5191f1933f 100644 --- a/src/backend/base/langflow/api/v1/mappers/deployments/helpers.py +++ b/src/backend/base/langflow/api/v1/mappers/deployments/helpers.py @@ -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. diff --git a/src/backend/base/langflow/api/v1/monitor.py b/src/backend/base/langflow/api/v1/monitor.py index f3063500ab..1966490815 100644 --- a/src/backend/base/langflow/api/v1/monitor.py +++ b/src/backend/base/langflow/api/v1/monitor.py @@ -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: diff --git a/src/backend/base/langflow/api/v1/openai_responses.py b/src/backend/base/langflow/api/v1/openai_responses.py index b5b9e315e4..132512948d 100644 --- a/src/backend/base/langflow/api/v1/openai_responses.py +++ b/src/backend/base/langflow/api/v1/openai_responses.py @@ -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: diff --git a/src/backend/base/langflow/api/v1/projects.py b/src/backend/base/langflow/api/v1/projects.py index 9dd5981667..0084698237 100644 --- a/src/backend/base/langflow/api/v1/projects.py +++ b/src/backend/base/langflow/api/v1/projects.py @@ -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( diff --git a/src/backend/base/langflow/api/v1/schemas/authz_shares.py b/src/backend/base/langflow/api/v1/schemas/authz_shares.py index 9a51aabcd2..bdaaac3060 100644 --- a/src/backend/base/langflow/api/v1/schemas/authz_shares.py +++ b/src/backend/base/langflow/api/v1/schemas/authz_shares.py @@ -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 diff --git a/src/backend/base/langflow/api/v2/files.py b/src/backend/base/langflow/api/v2/files.py index 802bb8cdad..43a6bb07bd 100644 --- a/src/backend/base/langflow/api/v2/files.py +++ b/src/backend/base/langflow/api/v2/files.py @@ -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. diff --git a/src/backend/base/langflow/api/v2/workflow.py b/src/backend/base/langflow/api/v2/workflow.py index 1c795993ae..5393d8d807 100644 --- a/src/backend/base/langflow/api/v2/workflow.py +++ b/src/backend/base/langflow/api/v2/workflow.py @@ -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) diff --git a/src/backend/base/langflow/cli/authz_dry_run.py b/src/backend/base/langflow/cli/authz_dry_run.py index d079fdc0bd..cb07868350 100644 --- a/src/backend/base/langflow/cli/authz_dry_run.py +++ b/src/backend/base/langflow/cli/authz_dry_run.py @@ -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)) diff --git a/src/backend/base/langflow/helpers/flow.py b/src/backend/base/langflow/helpers/flow.py index e085d3a0ad..4a449161c9 100644 --- a/src/backend/base/langflow/helpers/flow.py +++ b/src/backend/base/langflow/helpers/flow.py @@ -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: diff --git a/src/backend/base/langflow/plugin_routes.py b/src/backend/base/langflow/plugin_routes.py index cb9253ce21..deeffb6396 100644 --- a/src/backend/base/langflow/plugin_routes.py +++ b/src/backend/base/langflow/plugin_routes.py @@ -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:: diff --git a/src/backend/base/langflow/services/authorization/__init__.py b/src/backend/base/langflow/services/authorization/__init__.py index 455fe6aed0..49ade762c2 100644 --- a/src/backend/base/langflow/services/authorization/__init__.py +++ b/src/backend/base/langflow/services/authorization/__init__.py @@ -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, diff --git a/src/backend/base/langflow/services/authorization/actions.py b/src/backend/base/langflow/services/authorization/actions.py index 5487480abf..1819602c5a 100644 --- a/src/backend/base/langflow/services/authorization/actions.py +++ b/src/backend/base/langflow/services/authorization/actions.py @@ -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" diff --git a/src/backend/base/langflow/services/authorization/fetch.py b/src/backend/base/langflow/services/authorization/fetch.py index 7a3f065a57..061f2b19f4 100644 --- a/src/backend/base/langflow/services/authorization/fetch.py +++ b/src/backend/base/langflow/services/authorization/fetch.py @@ -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) diff --git a/src/backend/base/langflow/services/authorization/service.py b/src/backend/base/langflow/services/authorization/service.py index fe579cca86..10d2adb463 100644 --- a/src/backend/base/langflow/services/authorization/service.py +++ b/src/backend/base/langflow/services/authorization/service.py @@ -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) diff --git a/src/backend/base/langflow/services/authorization/utils.py b/src/backend/base/langflow/services/authorization/utils.py index 518f1c9581..50d2ac97d1 100644 --- a/src/backend/base/langflow/services/authorization/utils.py +++ b/src/backend/base/langflow/services/authorization/utils.py @@ -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=)`` 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)) diff --git a/src/backend/base/langflow/services/database/models/auth/authz.py b/src/backend/base/langflow/services/database/models/auth/authz.py index 1e1294484c..06fbd0b112 100644 --- a/src/backend/base/langflow/services/database/models/auth/authz.py +++ b/src/backend/base/langflow/services/database/models/auth/authz.py @@ -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 diff --git a/src/backend/base/langflow/services/database/models/deployment/crud.py b/src/backend/base/langflow/services/database/models/deployment/crud.py index 22813990d1..35e5ca08d1 100644 --- a/src/backend/base/langflow/services/database/models/deployment/crud.py +++ b/src/backend/base/langflow/services/database/models/deployment/crud.py @@ -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. diff --git a/src/backend/tests/unit/alembic/test_seed_authz_system_roles.py b/src/backend/tests/unit/alembic/test_seed_authz_system_roles.py index 227b452961..bfa6c0dde7 100644 --- a/src/backend/tests/unit/alembic/test_seed_authz_system_roles.py +++ b/src/backend/tests/unit/alembic/test_seed_authz_system_roles.py @@ -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 diff --git a/src/backend/tests/unit/api/v1/test_authz_share_routes.py b/src/backend/tests/unit/api/v1/test_authz_share_routes.py index 68c3355df2..0d69538e0b 100644 --- a/src/backend/tests/unit/api/v1/test_authz_share_routes.py +++ b/src/backend/tests/unit/api/v1/test_authz_share_routes.py @@ -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) diff --git a/src/backend/tests/unit/services/authorization/test_actions.py b/src/backend/tests/unit/services/authorization/test_actions.py index a4c8e814ae..afbcb24523 100644 --- a/src/backend/tests/unit/services/authorization/test_actions.py +++ b/src/backend/tests/unit/services/authorization/test_actions.py @@ -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" diff --git a/src/backend/tests/unit/services/authorization/test_authorization_service.py b/src/backend/tests/unit/services/authorization/test_authorization_service.py index a35d2a23d0..eeb9a9ca10 100644 --- a/src/backend/tests/unit/services/authorization/test_authorization_service.py +++ b/src/backend/tests/unit/services/authorization/test_authorization_service.py @@ -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, diff --git a/src/backend/tests/unit/services/authorization/test_capability_flag.py b/src/backend/tests/unit/services/authorization/test_capability_flag.py index f7dac8ef4e..2e4fa15f48 100644 --- a/src/backend/tests/unit/services/authorization/test_capability_flag.py +++ b/src/backend/tests/unit/services/authorization/test_capability_flag.py @@ -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 diff --git a/src/backend/tests/unit/services/authorization/test_fetch.py b/src/backend/tests/unit/services/authorization/test_fetch.py index 44d76edc46..6df19112b2 100644 --- a/src/backend/tests/unit/services/authorization/test_fetch.py +++ b/src/backend/tests/unit/services/authorization/test_fetch.py @@ -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( diff --git a/src/backend/tests/unit/services/authorization/test_flow_route_guards.py b/src/backend/tests/unit/services/authorization/test_flow_route_guards.py index be0cb38980..6f1535565b 100644 --- a/src/backend/tests/unit/services/authorization/test_flow_route_guards.py +++ b/src/backend/tests/unit/services/authorization/test_flow_route_guards.py @@ -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 diff --git a/src/backend/tests/unit/services/authorization/test_route_guard_regressions.py b/src/backend/tests/unit/services/authorization/test_route_guard_regressions.py index fae3cc0727..68b243ba58 100644 --- a/src/backend/tests/unit/services/authorization/test_route_guard_regressions.py +++ b/src/backend/tests/unit/services/authorization/test_route_guard_regressions.py @@ -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()``. """ diff --git a/src/backend/tests/unit/services/authorization/test_utils.py b/src/backend/tests/unit/services/authorization/test_utils.py index 180f0202e4..7b9911400f 100644 --- a/src/backend/tests/unit/services/authorization/test_utils.py +++ b/src/backend/tests/unit/services/authorization/test_utils.py @@ -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) diff --git a/src/backend/tests/unit/test_authz_dry_run_cli.py b/src/backend/tests/unit/test_authz_dry_run_cli.py index d1c3713a9e..331155badb 100644 --- a/src/backend/tests/unit/test_authz_dry_run_cli.py +++ b/src/backend/tests/unit/test_authz_dry_run_cli.py @@ -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:``. diff --git a/src/lfx/src/lfx/services/authorization/base.py b/src/lfx/src/lfx/services/authorization/base.py index 1276eb369f..a6094359a3 100644 --- a/src/lfx/src/lfx/services/authorization/base.py +++ b/src/lfx/src/lfx/services/authorization/base.py @@ -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: diff --git a/src/lfx/src/lfx/services/authorization/service.py b/src/lfx/src/lfx/services/authorization/service.py index 6acd5b0886..302cdd490f 100644 --- a/src/lfx/src/lfx/services/authorization/service.py +++ b/src/lfx/src/lfx/services/authorization/service.py @@ -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).""" diff --git a/src/lfx/src/lfx/services/settings/auth.py b/src/lfx/src/lfx/services/settings/auth.py index 5f058a5d88..e713cfc600 100644 --- a/src/lfx/src/lfx/services/settings/auth.py +++ b/src/lfx/src/lfx/services/settings/auth.py @@ -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,