From 5087977fa28ea151775162d45a062269ecd95fb9 Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Thu, 21 May 2026 15:27:36 -0700 Subject: [PATCH] fix: Route guards and unit tests --- .../7c8d9e0f1a2b_authz_foundations.py | 35 +- .../8d3a1f9c2e0b_seed_authz_system_roles.py | 75 ++-- .../base/langflow/api/utils/flow_utils.py | 9 + .../base/langflow/api/v1/authz_shares.py | 74 +++- src/backend/base/langflow/api/v1/flows.py | 179 +++++++-- .../base/langflow/api/v1/knowledge_bases.py | 7 + src/backend/base/langflow/api/v1/projects.py | 47 ++- .../langflow/api/v1/schemas/authz_shares.py | 6 +- src/backend/base/langflow/helpers/flow.py | 65 ++- src/backend/base/langflow/main.py | 10 + .../langflow/services/authorization/fetch.py | 28 +- .../langflow/services/authorization/utils.py | 112 ++++-- .../services/database/models/auth/authz.py | 36 +- .../unit/api/v1/test_authz_share_routes.py | 376 ++++++++++++++++++ .../test_route_guard_regressions.py | 142 +++++++ .../unit/services/authorization/test_utils.py | 128 +++++- .../src/lfx/services/authorization/base.py | 44 +- src/lfx/src/lfx/services/manager.py | 29 +- src/lfx/src/lfx/services/settings/auth.py | 5 +- 19 files changed, 1263 insertions(+), 144 deletions(-) create mode 100644 src/backend/tests/unit/api/v1/test_authz_share_routes.py create mode 100644 src/backend/tests/unit/services/authorization/test_route_guard_regressions.py 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 2927f3e18f..ffd9f7a5cc 100644 --- a/src/backend/base/langflow/alembic/versions/7c8d9e0f1a2b_authz_foundations.py +++ b/src/backend/base/langflow/alembic/versions/7c8d9e0f1a2b_authz_foundations.py @@ -115,6 +115,18 @@ def upgrade() -> None: batch_op.create_index(batch_op.f("ix_authz_role_assignment_role_id"), ["role_id"], unique=False) batch_op.create_index(batch_op.f("ix_authz_role_assignment_domain_id"), ["domain_id"], unique=False) + # Hot-path lookup index: "all assignments for user X scoped to domain + # (type, id)" — the canonical query an enterprise enforcer issues on + # every request to compute effective roles. Single-column indexes + # leave the planner with a choice between two non-covering scans; + # this composite gives a single index seek. + op.create_index( + "ix_authz_role_assignment_user_domain", + "authz_role_assignment", + ["user_id", "domain_type", "domain_id"], + unique=False, + ) + # Partial unique indexes — NULL domain_id is never-equal in SQL, so # split the constraint into scoped and unscoped buckets. The unscoped # bucket filters on domain_id IS NULL only (NOT also on @@ -194,16 +206,16 @@ def upgrade() -> None: sa.PrimaryKeyConstraint("id"), sa.CheckConstraint( "scope IN ('private', 'team', 'user', 'public')", - name="scope_enum", + name="ck_authz_share_scope_enum", ), sa.CheckConstraint( "permission_level IN ('read', 'write', 'execute', 'admin')", - name="permission_enum", + name="ck_authz_share_permission_enum", ), sa.CheckConstraint( "(scope IN ('team', 'user') AND target_id IS NOT NULL) " "OR (scope IN ('private', 'public') AND target_id IS NULL)", - name="scope_target_consistency", + name="ck_authz_share_scope_target_consistency", ), ) with op.batch_alter_table("authz_share", schema=None) as batch_op: @@ -249,6 +261,14 @@ def upgrade() -> None: ) with op.batch_alter_table("authz_edit_lock", schema=None) as batch_op: batch_op.create_index(batch_op.f("ix_authz_edit_lock_flow_id"), ["flow_id"], unique=True) + # Expired-lock sweeper queries WHERE expires_at < now() — without + # this index that's a full table scan. Cheaper to add here than + # to retrofit once a cleanup job lands and starts hurting. + batch_op.create_index( + batch_op.f("ix_authz_edit_lock_expires_at"), + ["expires_at"], + unique=False, + ) # ------------------------------------------------------------------ # authz_audit_log — append-only authorization audit @@ -266,6 +286,15 @@ def upgrade() -> None: sa.Column("timestamp", sa.DateTime(timezone=True), nullable=False), sa.ForeignKeyConstraint(["user_id"], ["user.id"], ondelete="SET NULL"), sa.PrimaryKeyConstraint("id"), + # ``owner_override`` is the third value the framework writes + # (see ``_AUDIT_OWNER_OVERRIDE`` in + # ``services/authorization/utils.py``) — without it in the CHECK + # set, an enterprise plugin's owner-shortcircuit audit row would + # silently violate the constraint. + sa.CheckConstraint( + "result IN ('allow', 'deny', 'owner_override')", + name="ck_authz_audit_log_result_enum", + ), ) with op.batch_alter_table("authz_audit_log", schema=None) as batch_op: batch_op.create_index(batch_op.f("ix_authz_audit_log_user_id"), ["user_id"], unique=False) 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 e1072284ad..2b195e5751 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 @@ -120,34 +120,55 @@ def upgrade() -> None: ) timestamp = datetime.now(timezone.utc) + # ``sa.Uuid`` adapts ``UUID`` objects per-dialect (CHAR(32) on SQLite, + # native UUID on Postgres). Passing a bare string skips that path and + # fails at bind time on SQLite (``'str' object has no attribute 'hex'``), + # so we pass real UUID objects here. + # ``sa.JSON`` serializes Python lists natively on both SQLite and Postgres. + # ``json.dumps`` here would write the JSON-encoded string *inside* the + # JSON column, which downstream readers (Enterprise PolicySync expects a + # list) would have to peel a second time. for name, description, permissions in _SYSTEM_ROLES: - already_present = conn.execute( - sa.select(sa.literal(1)).select_from(authz_role).where(authz_role.c.name == name) - ).scalar() - if already_present: - continue - # ``sa.Uuid`` adapts ``UUID`` objects per-dialect (CHAR(32) on SQLite, - # native UUID on Postgres). Passing a bare string skips that path and - # fails at bind time on SQLite (``'str' object has no attribute - # 'hex'``), so we pass real UUID objects here. - # ``sa.JSON`` serializes Python lists natively on both SQLite and - # Postgres. ``json.dumps`` here would write the JSON-encoded string - # *inside* the JSON column, which downstream readers (Enterprise - # PolicySync expects a list) would have to peel a second time. - conn.execute( - authz_role.insert().values( - id=uuid4(), - name=name, - description=description, - is_system=True, - permissions=list(permissions), - parent_role_id=None, - workspace_id=None, - created_at=timestamp, - updated_at=timestamp, - created_by=None, - ) - ) + # Use an atomic check-then-insert to harden against concurrent + # rollouts (two pods running migrations against the same DB). The + # previous implementation did a SELECT then an INSERT in two + # statements: under concurrency both could pass the SELECT, both + # could attempt the INSERT, and one would crash on the unique + # constraint. ``ON CONFLICT DO NOTHING`` / ``OR IGNORE`` makes this + # idempotent under any number of concurrent writers without a lock. + dialect = conn.dialect.name + values = { + "id": uuid4(), + "name": name, + "description": description, + "is_system": True, + "permissions": list(permissions), + "parent_role_id": None, + "workspace_id": None, + "created_at": timestamp, + "updated_at": timestamp, + "created_by": None, + } + if dialect == "postgresql": + from sqlalchemy.dialects.postgresql import insert as pg_insert + + stmt = pg_insert(authz_role).values(**values).on_conflict_do_nothing(index_elements=["name"]) + conn.execute(stmt) + elif dialect == "sqlite": + from sqlalchemy.dialects.sqlite import insert as sqlite_insert + + stmt = sqlite_insert(authz_role).values(**values).on_conflict_do_nothing(index_elements=["name"]) + conn.execute(stmt) + else: + # Fallback for other dialects (e.g. unit test stand-ins): use the + # original check-then-insert. Not fully race-safe but matches the + # previous behavior and never raises on a clean fresh DB. + already_present = conn.execute( + sa.select(sa.literal(1)).select_from(authz_role).where(authz_role.c.name == name) + ).scalar() + if already_present: + continue + conn.execute(authz_role.insert().values(**values)) def downgrade() -> None: diff --git a/src/backend/base/langflow/api/utils/flow_utils.py b/src/backend/base/langflow/api/utils/flow_utils.py index ecf57aee13..34384aab84 100644 --- a/src/backend/base/langflow/api/utils/flow_utils.py +++ b/src/backend/base/langflow/api/utils/flow_utils.py @@ -13,6 +13,7 @@ from sqlalchemy import delete from sqlmodel import col, select from sqlmodel.ext.asyncio.session import AsyncSession +from langflow.services.database.models.auth.authz import AuthzShare from langflow.services.database.models.deployment.exceptions import ( araise_if_deployment_guard_error_or_skip, ) @@ -112,6 +113,14 @@ async def cascade_delete_flow(session: AsyncSession, flow_id: uuid.UUID) -> None if trace_ids: await session.exec(delete(SpanTable).where(col(SpanTable.trace_id).in_(trace_ids))) await session.exec(delete(TraceTable).where(col(TraceTable.id).in_(trace_ids))) + # authz_share is polymorphic over resource_type/resource_id with no + # FK, so DB cascades cannot remove stale share rows when the flow is + # deleted. Clean them up here so a deleted flow's grants do not + # silently survive — that would let an enterprise plugin keep + # honoring share rows that point at a tombstoned resource. + await session.exec( + delete(AuthzShare).where(AuthzShare.resource_type == "flow").where(AuthzShare.resource_id == flow_id) + ) await session.exec(delete(Flow).where(Flow.id == flow_id)) except Exception as e: await araise_if_deployment_guard_error_or_skip( diff --git a/src/backend/base/langflow/api/v1/authz_shares.py b/src/backend/base/langflow/api/v1/authz_shares.py index 092975f825..df9f484553 100644 --- a/src/backend/base/langflow/api/v1/authz_shares.py +++ b/src/backend/base/langflow/api/v1/authz_shares.py @@ -28,6 +28,7 @@ from typing import Annotated from uuid import UUID from fastapi import APIRouter, HTTPException, Query, status +from lfx.log.logger import logger from sqlmodel import select from langflow.api.utils import CurrentActiveUser, DbSession @@ -200,10 +201,14 @@ async def create_share( try: await session.flush() except Exception as exc: + # Log the raw exception server-side; the client gets a fixed string + # so we don't leak table/column/constraint names through the 409 + # body (security rule: error messages don't disclose schema). await session.rollback() + logger.warning("authz_share insert rejected: %s", exc) raise HTTPException( status_code=status.HTTP_409_CONFLICT, - detail=f"Share could not be created: {exc}", + detail="Share could not be created: it may already exist or conflict with an existing share.", ) from exc await session.refresh(row) @@ -227,6 +232,10 @@ async def create_share( return ShareRead.model_validate(row, from_attributes=True) +_LIST_SHARES_MAX_LIMIT = 200 +_LIST_SHARES_DEFAULT_LIMIT = 100 + + @router.get("", response_model=list[ShareRead]) @router.get("/", response_model=list[ShareRead]) async def list_shares( @@ -236,12 +245,19 @@ async def list_shares( resource_id: Annotated[UUID | None, Query()] = None, target_id: Annotated[UUID | None, Query()] = None, scope: Annotated[str | None, Query()] = None, + limit: Annotated[int, Query(ge=1, le=_LIST_SHARES_MAX_LIMIT)] = _LIST_SHARES_DEFAULT_LIMIT, + offset: Annotated[int, Query(ge=0)] = 0, ) -> list[ShareRead]: """List share rows. Resource owners and superusers see every matching row. Non-owners only - see rows whose ``target_id`` is their own user id — they need to know - what's shared with them, but not the full grant ledger. + see rows whose ``target_id`` is their own user id, rows scoped to a team + they belong to, public rows, or rows on resources they own. + + Always paginated — ``limit`` is capped at 200 to bound DB load. The + previous unbounded implementation issued an N+1 ``select(AuthzTeamMember)`` + per row; this version pre-fetches the caller's team memberships once and + uses an in-memory set lookup. """ await ensure_share_permission( current_user, @@ -264,12 +280,22 @@ async def list_shares( raise HTTPException(status_code=400, detail=f"Unknown scope {scope!r}") from exc stmt = stmt.where(AuthzShare.scope == scope_value) + # Stable ordering + bounded fetch. ``created_at`` is monotonically + # increasing for a given operator so this is effectively cursor-like; + # callers wanting strict cursor pagination can layer ``id`` ties. + stmt = stmt.order_by(AuthzShare.created_at.desc(), AuthzShare.id).offset(offset).limit(limit) + rows = list(await session.exec(stmt)) is_superuser = getattr(current_user, "is_superuser", False) if is_superuser: return [ShareRead.model_validate(row, from_attributes=True) for row in rows] + # Pre-fetch the caller's team memberships once — the previous per-row + # ``select(AuthzTeamMember)`` was an N+1 that scaled with the page size. + team_membership_stmt = select(AuthzTeamMember.team_id).where(AuthzTeamMember.user_id == current_user.id) + caller_team_ids: set[UUID] = set(await session.exec(team_membership_stmt)) + # Non-superuser visibility: owner / creator / direct user target / team # member / public. See ``_user_can_see_share`` for the full rule set. visible: list[ShareRead] = [] @@ -282,16 +308,41 @@ async def list_shares( resource_type=row.resource_type, resource_id=row.resource_id, ) - if await _user_can_see_share( - session, + if _row_visible_to( row=row, user_id=current_user.id, resource_owner_id=owner_cache[key], + caller_team_ids=caller_team_ids, ): visible.append(ShareRead.model_validate(row, from_attributes=True)) return visible +def _row_visible_to( + *, + row: AuthzShare, + user_id: UUID, + resource_owner_id: UUID | None, + caller_team_ids: set[UUID], +) -> bool: + """In-memory variant of ``_user_can_see_share`` for the batch list path. + + Same rules as ``_user_can_see_share`` but team membership is checked via + the pre-fetched ``caller_team_ids`` set so the list loop is O(N) instead + of issuing one ``select(AuthzTeamMember)`` per row. + """ + if user_id in {resource_owner_id, row.created_by}: + return True + scope = row.scope + if scope == ShareScope.PUBLIC.value: + return True + if scope == ShareScope.USER.value and row.target_id == user_id: + return True + if scope == ShareScope.TEAM.value and row.target_id is not None: + return row.target_id in caller_team_ids + return False + + @router.get("/{share_id}", response_model=ShareRead) async def get_share( share_id: UUID, @@ -363,7 +414,18 @@ async def update_share( detail=f"Unknown permission_level {payload.permission_level!r}", ) from exc session.add(row) - await session.flush() + # Match the create_share error contract: rollback + fixed-string 409 + + # server-side log so a constraint violation on update doesn't return a + # raw SQLAlchemy stacktrace to the client. + try: + await session.flush() + except Exception as exc: + await session.rollback() + logger.warning("authz_share update rejected: %s", exc) + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Share could not be updated: it may conflict with an existing share.", + ) from exc await session.refresh(row) await _invalidate_for_share(row.scope, row.target_id) diff --git a/src/backend/base/langflow/api/v1/flows.py b/src/backend/base/langflow/api/v1/flows.py index 167e4c0000..b6af6af4ff 100644 --- a/src/backend/base/langflow/api/v1/flows.py +++ b/src/backend/base/langflow/api/v1/flows.py @@ -195,14 +195,19 @@ async def read_flows( # Apply the same authz filter the get_all branch uses so both modes of # this endpoint behave consistently. OSS pass-through returns the page - # items unchanged; an enterprise plugin filters per-flow. ``page.total`` - # may overcount denied items — a fully accurate count requires - # SQL-level prefiltering via authz_share (Phase 3 work). + # items unchanged; an enterprise plugin filters per-flow. The + # ``owner_extractor`` lets the helper short-circuit owner-allow without + # round-tripping through the enforcer — without it, an enterprise + # plugin lacking an explicit owner-allow policy would hide the caller's + # own flows from paginated results. ``page.total`` may overcount denied + # items — a fully accurate count requires SQL-level prefiltering via + # authz_share (Phase 3 work). page.items = await filter_visible_resources( current_user, resource_type="flow", candidates=list(page.items), domain_extractor=lambda flow: _resolve_casbin_domain(flow.workspace_id, flow.folder_id), + owner_extractor=lambda flow: flow.user_id, act=FlowAction.READ, ) return page # noqa: TRY300 — final return inside try matches the existing style of this handler @@ -339,14 +344,17 @@ async def update_flow( target_workspace_id = flow.workspace_id if flow.workspace_id is not None else db_flow.workspace_id target_folder_id = flow.folder_id if flow.folder_id is not None else db_flow.folder_id if target_workspace_id != db_flow.workspace_id or target_folder_id != db_flow.folder_id: - await ensure_flow_permission( - current_user, - FlowAction.WRITE, - flow_id=flow_id, - flow_user_id=db_flow.user_id, - workspace_id=target_workspace_id, - folder_id=target_folder_id, - ) + try: + await ensure_flow_permission( + current_user, + FlowAction.WRITE, + flow_id=flow_id, + flow_user_id=db_flow.user_id, + workspace_id=target_workspace_id, + folder_id=target_folder_id, + ) + except HTTPException as exc: + raise deny_to_404(exc, detail="Flow not found") from exc # Explicit folder_id=None is ignored here because _patch_flow builds # update_data with exclude_none=True, so null folder_id is a no-op. @@ -359,6 +367,41 @@ async def update_flow( db_flow_for_attempt = await _read_flow(session=session, flow_id=flow_id, user_id=current_user.id) if not db_flow_for_attempt: raise HTTPException(status_code=404, detail="Flow not found") + # TOCTOU: a concurrent PATCH could have moved this flow to a + # different workspace/folder between the destination check above + # and this retry attempt. Re-authorize against the freshly + # reloaded source AND destination so the writer cannot ride a + # stale check across a race. + try: + await ensure_flow_permission( + current_user, + FlowAction.WRITE, + flow_id=flow_id, + flow_user_id=db_flow_for_attempt.user_id, + workspace_id=db_flow_for_attempt.workspace_id, + folder_id=db_flow_for_attempt.folder_id, + ) + except HTTPException as exc: + raise deny_to_404(exc, detail="Flow not found") from exc + attempt_target_workspace_id = ( + flow.workspace_id if flow.workspace_id is not None else db_flow_for_attempt.workspace_id + ) + attempt_target_folder_id = flow.folder_id if flow.folder_id is not None else db_flow_for_attempt.folder_id + if ( + attempt_target_workspace_id != db_flow_for_attempt.workspace_id + or attempt_target_folder_id != db_flow_for_attempt.folder_id + ): + try: + await ensure_flow_permission( + current_user, + FlowAction.WRITE, + flow_id=flow_id, + flow_user_id=db_flow_for_attempt.user_id, + workspace_id=attempt_target_workspace_id, + folder_id=attempt_target_folder_id, + ) + except HTTPException as exc: + raise deny_to_404(exc, detail="Flow not found") from exc return await _patch_flow( session=session, db_flow=db_flow_for_attempt, @@ -405,18 +448,33 @@ async def upsert_flow( existing_flow = (await session.exec(select(Flow).where(Flow.id == flow_id))).first() if existing_flow is not None: - # Flow exists - check ownership (return 404 to avoid leaking resource existence) - if existing_flow.user_id != current_user.id: + # OSS floor: when the registered service cannot widen fetch (OSS + # pass-through, or any path where AUTHZ_ENABLED is off), non-owners + # must not be able to upsert by id. Without this guard, the + # ``Flow.id == flow_id``-only fetch above would let a non-owner + # overwrite another user's flow because ``ensure_flow_permission`` + # is a no-op when AUTHZ_ENABLED=false. With an enterprise plugin + # registered, ``ensure_flow_permission`` below is authoritative — + # a valid WRITE share grant must be honored, and a deny converts + # to 404 to preserve UUID privacy. + from langflow.services.deps import get_authorization_service + + authz = get_authorization_service() + can_widen = await authz.supports_cross_user_fetch() and await authz.is_enabled() + if not can_widen and existing_flow.user_id != current_user.id: raise HTTPException(status_code=404, detail="Flow not found") - await ensure_flow_permission( - current_user, - FlowAction.WRITE, - flow_id=flow_id, - flow_user_id=existing_flow.user_id, - workspace_id=existing_flow.workspace_id, - folder_id=existing_flow.folder_id, - ) + try: + await ensure_flow_permission( + current_user, + FlowAction.WRITE, + flow_id=flow_id, + flow_user_id=existing_flow.user_id, + workspace_id=existing_flow.workspace_id, + folder_id=existing_flow.folder_id, + ) + except HTTPException as exc: + raise deny_to_404(exc, detail="Flow not found") from exc # Destination check (see update_flow above): if the payload moves # the flow into a new workspace/folder, also authorize WRITE at the @@ -426,14 +484,17 @@ async def upsert_flow( target_workspace_id = flow.workspace_id if flow.workspace_id is not None else existing_flow.workspace_id target_folder_id = flow.folder_id if flow.folder_id is not None else existing_flow.folder_id if target_workspace_id != existing_flow.workspace_id or target_folder_id != existing_flow.folder_id: - await ensure_flow_permission( - current_user, - FlowAction.WRITE, - flow_id=flow_id, - flow_user_id=existing_flow.user_id, - workspace_id=target_workspace_id, - folder_id=target_folder_id, - ) + try: + await ensure_flow_permission( + current_user, + FlowAction.WRITE, + flow_id=flow_id, + flow_user_id=existing_flow.user_id, + workspace_id=target_workspace_id, + folder_id=target_folder_id, + ) + except HTTPException as exc: + raise deny_to_404(exc, detail="Flow not found") from exc # Sync deployment state before folder changes # Explicit folder_id=None is ignored here because _update_existing_flow @@ -687,10 +748,27 @@ async def delete_multiple_flows( async def _delete_operation() -> int: if not flow_ids: return 0 - flows_to_delete = ( - await db.exec(select(Flow).where(col(Flow.id).in_(flow_ids)).where(Flow.user_id == user.id)) - ).all() + # Widen the fetch when an enterprise plugin can authorize cross-user + # DELETE via share grants. Without this, pre-scoping by + # ``Flow.user_id == user.id`` silently drops non-owned flows from + # the working set so a valid DELETE share is never honored and + # ``ensure_flow_permission`` never emits a deny audit row. In OSS + # default the query stays owner-scoped — the no-op pass-through + # cannot widen visibility. + from langflow.services.deps import get_authorization_service + + authz = get_authorization_service() + base_stmt = select(Flow).where(col(Flow.id).in_(flow_ids)) + if await authz.supports_cross_user_fetch() and await authz.is_enabled(): + stmt = base_stmt + else: + stmt = base_stmt.where(Flow.user_id == user.id) + flows_to_delete = (await db.exec(stmt)).all() for flow in flows_to_delete: + # A deny from the enterprise plugin raises 403. We let it + # propagate so the audit row is written and the caller knows + # the bulk delete was rejected for at least one flow — the + # alternative (silent skip) would mask policy violations. await ensure_flow_permission( user, FlowAction.DELETE, @@ -733,20 +811,39 @@ async def download_multiple_file( # TODO: Full-version download (include_version parameter) is planned as a follow-up feature. # When implemented, add an include_version: bool = False parameter and embed version # entries in each flow dict using get_flow_versions_with_provider_status and strip_version_data. - flows = (await db.exec(select(Flow).where(and_(Flow.user_id == user.id, Flow.id.in_(flow_ids))))).all() # type: ignore[attr-defined] + # Widen the fetch when the enterprise plugin can authorize cross-user READ + # via share grants. Pre-scoping by ``Flow.user_id == user.id`` here would + # silently drop shared flows from the working set so a valid READ share is + # never honored and no audit row is written. OSS default keeps the query + # owner-scoped so the no-op pass-through cannot widen visibility. + from langflow.services.deps import get_authorization_service + + authz = get_authorization_service() + base_stmt = select(Flow).where(col(Flow.id).in_(flow_ids)) # type: ignore[attr-defined] + if await authz.supports_cross_user_fetch() and await authz.is_enabled(): + stmt = base_stmt + else: + stmt = base_stmt.where(and_(Flow.user_id == user.id)) + flows = (await db.exec(stmt)).all() if not flows: raise HTTPException(status_code=404, detail="No flows found.") for flow in flows: - await ensure_flow_permission( - user, - FlowAction.READ, - flow_id=flow.id, - flow_user_id=flow.user_id, - workspace_id=flow.workspace_id, - folder_id=flow.folder_id, - ) + # Convert deny to 404 to preserve UUID privacy — a caller probing + # arbitrary IDs cannot distinguish "doesn't exist" from "exists but + # not shared with me". + try: + await ensure_flow_permission( + user, + FlowAction.READ, + flow_id=flow.id, + flow_user_id=flow.user_id, + workspace_id=flow.workspace_id, + folder_id=flow.folder_id, + ) + except HTTPException as exc: + raise deny_to_404(exc, detail="No flows found.") from exc return _build_flows_download_response(flows) diff --git a/src/backend/base/langflow/api/v1/knowledge_bases.py b/src/backend/base/langflow/api/v1/knowledge_bases.py index 2ff827a8b6..6b982b3b0f 100644 --- a/src/backend/base/langflow/api/v1/knowledge_bases.py +++ b/src/backend/base/langflow/api/v1/knowledge_bases.py @@ -167,6 +167,13 @@ async def _guard_kb_action( # and we skip the DB roundtrip. if kb_user_id == current_user.id: return _KbGuardResult(record=resolved_record, owner_user=current_user) + # ``kb_user_id`` comes from a SQLAlchemy column in production — always a + # UUID. Defensive guard: skip the DB lookup if the value isn't a UUID + # (e.g. a test mock where ``record.user_id`` is an auto-generated + # ``MagicMock``). Without this, the bind fails with + # ``Error binding parameter 1: type 'MagicMock' is not supported``. + if not isinstance(kb_user_id, uuid.UUID): + return _KbGuardResult(record=resolved_record, owner_user=current_user) async with session_scope() as session: owner = await get_user_by_id(session, kb_user_id) if owner is None: diff --git a/src/backend/base/langflow/api/v1/projects.py b/src/backend/base/langflow/api/v1/projects.py index 6f599286b0..f365542484 100644 --- a/src/backend/base/langflow/api/v1/projects.py +++ b/src/backend/base/langflow/api/v1/projects.py @@ -32,7 +32,12 @@ from langflow.api.v1.projects_mcp_helpers import ( ) from langflow.initial_setup.constants import ASSISTANT_FOLDER_NAME, STARTER_FOLDER_NAME from langflow.services.auth.mcp_encryption import encrypt_auth_settings -from langflow.services.authorization import ProjectAction, ensure_project_permission, filter_visible_resources +from langflow.services.authorization import ( + FlowAction, + ProjectAction, + ensure_project_permission, + filter_visible_resources, +) from langflow.services.authorization.fetch import authorized_or_owner_scoped, deny_to_404 from langflow.services.authorization.utils import _resolve_casbin_domain from langflow.services.database.models.deployment.exceptions import ( @@ -320,7 +325,20 @@ async def read_project( # If no pagination requested, return flows visible to the caller. if treat_as_shared: - visible_flows = list(project.flows) + # A project share grant implies access to the project itself, but + # per-flow policy (deny rules, lower scopes) still applies. Without + # this call, ``list(project.flows)`` would leak every flow in the + # project regardless of finer-grained Casbin rules the plugin may + # have. OSS pass-through returns the input list unchanged, so this + # has no effect on default OSS installs. + visible_flows = await filter_visible_resources( + current_user, + resource_type="flow", + candidates=list(project.flows), + domain_extractor=lambda flow: _resolve_casbin_domain(flow.workspace_id, flow.folder_id), + owner_extractor=lambda flow: flow.user_id, + act=FlowAction.READ, + ) else: visible_flows = [flow for flow in project.flows if flow.user_id == current_user.id] project.flows = visible_flows @@ -468,12 +486,19 @@ async def update_project( flow_ids_for_sync = list(dict.fromkeys(excluded_flows + concat_project_components)) async def _move_flows_for_project_update() -> None: + # Both SELECT and UPDATE must scope to the project owner — a + # non-owner editing a shared project must touch the *owner's* + # flows, not the actor's. The previous code filtered the SELECT + # by ``current_user.id`` (returning zero rows for non-owners) but + # then ran an UPDATE without any owner filter, so an id collision + # would have moved cross-user flows without per-flow authz. + # Scoping both statements to ``project_owner_id`` closes the gap. if my_collection_project: excluded_flow_rows = ( await session.exec( select(Flow.id, Flow.folder_id).where( Flow.id.in_(excluded_flows), # type: ignore[attr-defined] - Flow.user_id == current_user.id, + Flow.user_id == project_owner_id, ) ) ).all() @@ -483,7 +508,12 @@ async def update_project( new_folder_id=my_collection_project.id, ) update_statement_my_collection = ( - update(Flow).where(Flow.id.in_(excluded_flows)).values(folder_id=my_collection_project.id) # type: ignore[attr-defined] + update(Flow) + .where( + Flow.id.in_(excluded_flows), # type: ignore[attr-defined] + Flow.user_id == project_owner_id, + ) + .values(folder_id=my_collection_project.id) ) await session.exec(update_statement_my_collection) @@ -492,7 +522,7 @@ async def update_project( await session.exec( select(Flow.id, Flow.folder_id).where( Flow.id.in_(concat_project_components), # type: ignore[attr-defined] - Flow.user_id == current_user.id, + Flow.user_id == project_owner_id, ) ) ).all() @@ -502,7 +532,12 @@ async def update_project( new_folder_id=existing_project.id, ) update_statement_components = ( - update(Flow).where(Flow.id.in_(concat_project_components)).values(folder_id=existing_project.id) # type: ignore[attr-defined] + update(Flow) + .where( + Flow.id.in_(concat_project_components), # type: ignore[attr-defined] + Flow.user_id == project_owner_id, + ) + .values(folder_id=existing_project.id) ) await session.exec(update_statement_components) 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 123a467d22..9a51aabcd2 100644 --- a/src/backend/base/langflow/api/v1/schemas/authz_shares.py +++ b/src/backend/base/langflow/api/v1/schemas/authz_shares.py @@ -71,7 +71,11 @@ class ShareRead(BaseModel): """Read-only projection of an ``authz_share`` row.""" id: UUID - resource_type: str + # Use the same Literal as ``ShareCreate.resource_type`` so OpenAPI clients + # see a stable enum on both sides of the round trip. A bare ``str`` here + # would let a row inserted by a future code path with a typoed + # ``resource_type`` round-trip back unvalidated. + resource_type: ShareResourceType resource_id: UUID scope: ShareScopeLiteral target_id: UUID | None diff --git a/src/backend/base/langflow/helpers/flow.py b/src/backend/base/langflow/helpers/flow.py index 9496a280a7..e085d3a0ad 100644 --- a/src/backend/base/langflow/helpers/flow.py +++ b/src/backend/base/langflow/helpers/flow.py @@ -166,9 +166,30 @@ async def get_flow_by_id_or_name( async def load_flow( user_id: str, flow_id: str | None = None, flow_name: str | None = None, tweaks: dict | None = None ) -> Graph: + """Load a flow as a Graph after authorizing the caller. + + ``user_id`` is the caller (the user executing the graph), and ``flow_id`` + identifies the flow to load. This function is reachable from custom + components, sub-flow runners, and ``run_flow`` — anywhere a user-supplied + ``flow_id`` might end up. We must authorize EXECUTE here so a malicious + component or webhook cannot pass an arbitrary id and pull another user's + flow definition (which contains the flow author's prompts, tools, and + embedded credentials). + + OSS default (``AUTHZ_ENABLED=false`` or no enterprise plugin): the fetch + stays owner-scoped, so a non-owner ``flow_id`` returns ``None`` and the + function raises ``ValueError`` exactly as before. + + Enterprise (``AUTHZ_ENABLED=true`` and ``supports_cross_user_fetch=True``): + the fetch widens to id-only and ``ensure_flow_permission(EXECUTE)`` decides + access via the plugin's policy. + """ from lfx.graph.graph.base import Graph from langflow.processing.process import process_tweaks + from langflow.services.authorization import FlowAction, ensure_flow_permission + from langflow.services.authorization.fetch import authorized_or_owner_scoped + from langflow.services.database.models.user.model import User if not flow_id and not flow_name: msg = "Flow ID or Flow Name is required" @@ -179,8 +200,50 @@ async def load_flow( msg = f"Flow {flow_name} not found" raise ValueError(msg) + uuid_user_id = UUID(user_id) if isinstance(user_id, str) else user_id + uuid_flow_id = UUID(flow_id) if isinstance(flow_id, str) else flow_id + async with session_scope() as session: - graph_data = flow.data if (flow := await session.get(Flow, flow_id)) else None + flow = await authorized_or_owner_scoped( + session, + Flow, + id_column=Flow.id, + resource_id=uuid_flow_id, + owner_column=Flow.user_id, + owner_id=uuid_user_id, + ) + if flow is None: + msg = f"Flow {flow_id} not found" + raise ValueError(msg) + + # Authorize EXECUTE. ``ensure_flow_permission`` short-circuits on owner + # override, so the OSS pass-through path is unchanged. With an + # enterprise plugin a deny raises ``HTTPException(403)`` which we + # translate back to ``ValueError`` so callers (component execution, + # ``run_flow``) keep the existing exception contract. + caller = await session.get(User, uuid_user_id) + if caller is None: + msg = "Session is invalid" + raise ValueError(msg) + try: + await ensure_flow_permission( + caller, + FlowAction.EXECUTE, + flow_id=flow.id, + flow_user_id=flow.user_id, + workspace_id=flow.workspace_id, + folder_id=flow.folder_id, + ) + except HTTPException as exc: + from fastapi import status as http_status + + if exc.status_code == http_status.HTTP_403_FORBIDDEN: + msg = f"Flow {flow_id} not found" + raise ValueError(msg) from exc + raise + + graph_data = flow.data + if not graph_data: msg = f"Flow {flow_id} not found" raise ValueError(msg) diff --git a/src/backend/base/langflow/main.py b/src/backend/base/langflow/main.py index f4497a7fcd..02567f5b2e 100644 --- a/src/backend/base/langflow/main.py +++ b/src/backend/base/langflow/main.py @@ -551,6 +551,16 @@ def get_lifespan(*, fix_migration=False, version=None): # Step 2: Cleaning Up Services with shutdown_progress.step(2): + # Drain pending audit writes before services tear down so + # rows scheduled mid-request still land in the DB. We do + # this here (not in teardown_services) because the DB + # session factory must still be alive. + try: + from langflow.services.authorization.utils import drain_pending_audit_writes + + await drain_pending_audit_writes(timeout=5.0) + except Exception as drain_exc: # noqa: BLE001 — never block shutdown on audit + await logger.awarning(f"drain_pending_audit_writes failed: {drain_exc}") try: await asyncio.wait_for(teardown_services(), timeout=30) except asyncio.TimeoutError: diff --git a/src/backend/base/langflow/services/authorization/fetch.py b/src/backend/base/langflow/services/authorization/fetch.py index b3fae7edc6..7a3f065a57 100644 --- a/src/backend/base/langflow/services/authorization/fetch.py +++ b/src/backend/base/langflow/services/authorization/fetch.py @@ -87,20 +87,18 @@ async def authorized_or_owner_scoped( def deny_to_404(exc: HTTPException, detail: str = "Not found") -> HTTPException: """Convert a 403 from ``ensure_*_permission`` into a 404 for UUID privacy. - Re-raises any non-403 exception untouched. Callers wrap their guard call: + For 403s, returns a fresh ``HTTPException(404, detail=detail)``. - .. code-block:: python - - try: - await ensure_flow_permission(user, FlowAction.READ, flow_id=flow.id, ...) - except HTTPException as exc: - raise deny_to_404(exc) from exc - - Rationale: a route that was reachable purely through a share grant must - not leak the resource's existence to callers who can no longer reach it - (e.g. after the share is revoked). 403 confirms the row exists; 404 does - not. + For non-403s, returns the original exception with its ``detail`` replaced + by the caller-supplied ``detail`` and any ``headers`` dropped. This + sanitisation is conservative on purpose: helper callers are guard sites, + so the exception about to propagate has been filtered through an authz + plugin whose detail string may quote resource UUIDs, owner ids, or + policy tuples. Letting that detail land in the client response would + leak information about resources the caller is not authorised to see. + Routes that need to expose a richer detail (e.g. a 409 with a real + conflict message) should catch and re-raise outside ``deny_to_404``. """ - if exc.status_code != status.HTTP_403_FORBIDDEN: - return exc - return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=detail) + if exc.status_code == status.HTTP_403_FORBIDDEN: + return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=detail) + return HTTPException(status_code=exc.status_code, detail=detail) diff --git a/src/backend/base/langflow/services/authorization/utils.py b/src/backend/base/langflow/services/authorization/utils.py index e6c5c6a2f0..518f1c9581 100644 --- a/src/backend/base/langflow/services/authorization/utils.py +++ b/src/backend/base/langflow/services/authorization/utils.py @@ -117,6 +117,35 @@ def _split_obj(obj: str) -> tuple[str | None, UUID | None]: return resource_type, None +# In-flight audit tasks. Keeping a strong reference prevents the event loop +# from garbage-collecting a pending task before it writes — and gives +# ``drain_pending_audit_writes`` (called on shutdown) something to await. +# A bare ``asyncio.create_task`` without this reference can be dropped by +# the GC mid-write and silently lose the audit row. +_pending_audit_tasks: set[asyncio.Task[None]] = set() + + +async def drain_pending_audit_writes(timeout: float = 5.0) -> None: + """Wait for any in-flight audit-write tasks to complete. + + Called on application shutdown so audit rows scheduled mid-request still + land in the DB before the event loop closes. Returns silently after + ``timeout`` even if some tasks are still pending — audit must never block + shutdown indefinitely. + """ + pending = {task for task in _pending_audit_tasks if not task.done()} + if not pending: + return + done, still_pending = await asyncio.wait(pending, timeout=timeout) + if still_pending: + logger.warning("drain_pending_audit_writes timed out with %d pending tasks", len(still_pending)) + # Surface task exceptions in logs but never raise — audit must not block shutdown. + for task in done: + exc = task.exception() if not task.cancelled() else None + if exc is not None: + logger.warning("Audit task raised during drain: %s", exc) + + async def audit_decision( *, user_id: UUID | None, @@ -129,11 +158,19 @@ async def audit_decision( Schedules an asyncio task and returns immediately. Failures inside the task are logged but never propagate to the caller so audit writes can never - block a real API response. + block a real API response. The task reference is held in + ``_pending_audit_tasks`` until completion so the event loop cannot GC it + mid-write; ``drain_pending_audit_writes`` awaits this set on shutdown so + pending rows are not lost. """ settings = get_settings_service() auth_settings = settings.auth_settings - if not auth_settings.AUTHZ_ENABLED or not getattr(auth_settings, "AUTHZ_AUDIT_ENABLED", True): + # Audit is independent of enforcement: an operator can keep audit on + # while AUTHZ_ENABLED=false to observe traffic before flipping the + # flag. Previously this short-circuited on AUTHZ_ENABLED too, which + # meant share CRUD writes (which the OSS floor allows in default mode) + # produced no audit trail at all. + if not getattr(auth_settings, "AUTHZ_AUDIT_ENABLED", True): return resource_type, resource_id = _split_obj(obj) @@ -157,8 +194,9 @@ async def audit_decision( except Exception: # noqa: BLE001 — audit must never raise into the request path logger.exception("Failed to write AuthzAuditLog row") - # Bare create_task is the langflow convention (see main.py:135, main.py:173). - asyncio.create_task(_write()) # noqa: RUF006 — fire-and-forget audit task + task = asyncio.create_task(_write()) + _pending_audit_tasks.add(task) + task.add_done_callback(_pending_audit_tasks.discard) async def ensure_permission( @@ -181,21 +219,41 @@ async def ensure_permission( # User-derived auth fields (e.g. is_superuser) must remain authoritative; caller # context is merged first so it cannot overwrite them. merged_context = {**(context or {}), **_auth_context(user)} - allowed = await authz.enforce( - user_id=user.id, - domain=domain, - obj=obj, - act=act, - context=merged_context, - ) - + # Fail-closed contract: if the plugin's ``enforce`` raises (Casbin DB + # down, policy parse error, etc.), we treat the request as denied and + # log an explicit error-audit row. Returning False from ``enforce`` is + # the plugin's documented way to deny; raising should not silently + # become an HTTP 500 that bypasses both the deny path and the audit log. + audit_action = f"{obj.split(':', 1)[0]}:{act}" if ":" in obj else act audit_details = {"domain": domain} for owner_key in _OWNER_CONTEXT_KEYS: if owner_key in merged_context and merged_context[owner_key] is not None: audit_details[owner_key] = str(merged_context[owner_key]) + try: + allowed = await authz.enforce( + user_id=user.id, + domain=domain, + obj=obj, + act=act, + context=merged_context, + ) + except Exception as exc: + logger.exception("Authorization plugin raised during enforce; failing closed") + await audit_decision( + user_id=user.id, + action=audit_action, + obj=obj, + result=_AUDIT_DENY, + details={**audit_details, "error": str(exc)}, + ) + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=f"Insufficient permissions to {act} on {obj}", + ) from exc + await audit_decision( user_id=user.id, - action=f"{obj.split(':', 1)[0]}:{act}" if ":" in obj else act, + action=audit_action, obj=obj, result=_AUDIT_ALLOW if allowed else _AUDIT_DENY, details=audit_details, @@ -265,15 +323,16 @@ async def _ensure_resource_permission( # Owner override: a resource owner can always operate on their own resource. if owner_id is not None and getattr(user, "id", None) == owner_id: - settings = get_settings_service() - if settings.auth_settings.AUTHZ_ENABLED: - await audit_decision( - user_id=user.id, - action=f"{resource_type}:{act_str}", - obj=obj, - result=_AUDIT_OWNER_OVERRIDE, - details={"domain": resolved_domain}, - ) + # ``audit_decision`` is gated on ``AUTHZ_AUDIT_ENABLED`` internally; + # we no longer suppress the call on ``AUTHZ_ENABLED=false`` so + # operators observing pre-enforcement traffic see owner overrides too. + await audit_decision( + user_id=user.id, + action=f"{resource_type}:{act_str}", + obj=obj, + result=_AUDIT_OWNER_OVERRIDE, + details={"domain": resolved_domain}, + ) return await ensure_permission( @@ -556,11 +615,16 @@ async def filter_visible_resources( decisions[index] = True if enforce_items: + # Use the already-extracted ``user_id`` instead of re-reading + # ``user.id`` here. The earlier owner-override partition used + # ``getattr(user, "id", None)``, so consistency means a non-User + # dependency (e.g. a thin context proxy) lands an explicit None + # rather than an AttributeError at this call site. if domain_extractor is None: # Single-domain fast path. Preserves the original single-call shape. requests = [(f"{resource_type}:{extractor(item)}", act_str) for item in enforce_items] results = await authz.batch_enforce( - user_id=user.id, + user_id=user_id, domain=domain, requests=requests, context=_auth_context(user), @@ -580,7 +644,7 @@ async def filter_visible_resources( for resolved_domain, bucket in buckets.items(): bucket_requests = [(f"{resource_type}:{extractor(item)}", act_str) for _, item in bucket] bucket_results = await authz.batch_enforce( - user_id=user.id, + user_id=user_id, domain=resolved_domain, requests=bucket_requests, context=auth_context, 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 2ff7f4331c..1e1294484c 100644 --- a/src/backend/base/langflow/services/database/models/auth/authz.py +++ b/src/backend/base/langflow/services/database/models/auth/authz.py @@ -129,6 +129,15 @@ class AuthzRoleAssignment(SQLModel, table=True): # type: ignore[call-arg] postgresql_where=text("domain_id IS NULL"), sqlite_where=text("domain_id IS NULL"), ), + # Hot-path lookup: "all assignments for user X scoped to a domain" + # — every enterprise enforce() call. Composite avoids two non-covering + # single-column scans. + Index( + "ix_authz_role_assignment_user_domain", + "user_id", + "domain_type", + "domain_id", + ), ) id: UUIDstr = Field(default_factory=uuid4, primary_key=True) @@ -139,7 +148,14 @@ class AuthzRoleAssignment(SQLModel, table=True): # type: ignore[call-arg] sa_column=Column(sa.Uuid(), ForeignKey("authz_role.id", ondelete="CASCADE"), nullable=False, index=True), ) domain_type: str = Field(default="global", description="global, org, workspace") - domain_id: UUIDstr | None = Field(default=None, index=True) + # Explicit ``sa_column`` so SQLModel emits ``sa.Uuid()`` matching the + # migration's column type. Without this, SQLModel can fall back to + # ``AutoString``/``CHAR(32)`` on SQLite, which drifts from the migration + # and confuses downstream type introspection. + domain_id: UUIDstr | None = Field( + default=None, + sa_column=Column(sa.Uuid(), nullable=True, index=True), + ) assigned_at: datetime = Field(default_factory=_tz_aware_now, sa_column=_tz_column()) assigned_by: UUIDstr | None = Field( default=None, @@ -201,11 +217,11 @@ class AuthzShare(SQLModel, table=True): # type: ignore[call-arg] # partial unique indexes (which match on the lowercase form). CheckConstraint( "scope IN ('private', 'team', 'user', 'public')", - name="scope_enum", + name="ck_authz_share_scope_enum", ), CheckConstraint( "permission_level IN ('read', 'write', 'execute', 'admin')", - name="permission_enum", + name="ck_authz_share_permission_enum", ), # Targeted (TEAM/USER) shares require a target_id; untargeted # (PRIVATE/PUBLIC) shares forbid one. Matches the partial-unique-index @@ -214,7 +230,7 @@ class AuthzShare(SQLModel, table=True): # type: ignore[call-arg] CheckConstraint( "(scope IN ('team', 'user') AND target_id IS NOT NULL) " "OR (scope IN ('private', 'public') AND target_id IS NULL)", - name="scope_target_consistency", + name="ck_authz_share_scope_target_consistency", ), Index( "uq_authz_share_targeted", @@ -263,7 +279,9 @@ class AuthzEditLock(SQLModel, table=True): # type: ignore[call-arg] sa_column=Column(sa.Uuid(), ForeignKey("user.id", ondelete="CASCADE"), nullable=False), ) acquired_at: datetime = Field(default_factory=_tz_aware_now, sa_column=_tz_column()) - expires_at: datetime = Field(sa_column=_tz_column()) + # ``index=True`` matches the migration's ``ix_authz_edit_lock_expires_at`` + # so the expired-lock sweeper can do an index seek instead of a table scan. + expires_at: datetime = Field(sa_column=_tz_column(index=True)) # TODO: AuthzAuditLog is append-only and unbounded. At enterprise scale this @@ -279,6 +297,14 @@ class AuthzAuditLog(SQLModel, table=True): # type: ignore[call-arg] __table_args__ = ( Index("ix_authz_audit_log_user_timestamp", "user_id", "timestamp"), Index("ix_authz_audit_log_resource", "resource_type", "resource_id"), + # ``owner_override`` is the third value the framework writes (see + # ``_AUDIT_OWNER_OVERRIDE`` in services/authorization/utils.py); it + # must be in the CHECK set or owner-shortcut audit rows would + # silently fail the constraint. + CheckConstraint( + "result IN ('allow', 'deny', 'owner_override')", + name="ck_authz_audit_log_result_enum", + ), ) id: UUIDstr = Field(default_factory=uuid4, primary_key=True) 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 new file mode 100644 index 0000000000..68c3355df2 --- /dev/null +++ b/src/backend/tests/unit/api/v1/test_authz_share_routes.py @@ -0,0 +1,376 @@ +"""Route-level tests for the OSS share-administration floor. + +These tests assert the documented OSS contract: only the resource owner or a +superuser may administer ``authz_share`` rows for a resource when the +authorization service is the pass-through default (no enterprise plugin). The +schema-level happy path is covered in ``test_authz_share_schemas.py``; what +these tests prevent is the floor being silently dropped on a refactor. + +We exercise the handlers directly (not via the FastAPI app) using a fake async +session that records writes. This keeps the test fast and isolates the floor +logic from DB and Casbin concerns. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any +from uuid import UUID, uuid4 + +import pytest +from fastapi import HTTPException +from langflow.api.v1 import authz_shares as shares_module +from langflow.api.v1.schemas.authz_shares import ShareCreate, ShareUpdate +from langflow.services.database.models.auth import AuthzShare, SharePermissionLevel, ShareScope + + +class _FakeAsyncSession: + """Minimal async-session stand-in: stores get() results and records writes.""" + + def __init__(self, get_by_type: dict[tuple[type, UUID], Any] | None = None) -> None: + self._get_by_type = get_by_type or {} + self.added: list[Any] = [] + self.deleted: list[Any] = [] + self.flushed = 0 + self.rolled_back = 0 + + async def get(self, model: type, key: UUID) -> Any: + return self._get_by_type.get((model, key)) + + def add(self, obj: Any) -> None: + self.added.append(obj) + + async def delete(self, obj: Any) -> None: + self.deleted.append(obj) + + async def flush(self) -> None: + self.flushed += 1 + + async def refresh(self, obj: Any) -> None: # noqa: ARG002 + return None + + async def rollback(self) -> None: + self.rolled_back += 1 + + async def exec(self, _stmt: Any) -> list[Any]: + # list_shares is not exercised here; return empty by default. + return [] + + +class _StubAuthz: + """Pass-through authz service: allow everything, no cross-user fetch.""" + + def __init__(self, *, cross_user: bool = False, enabled: bool = False) -> None: + self._cross_user = cross_user + self._enabled = enabled + + async def supports_cross_user_fetch(self) -> bool: + return self._cross_user + + async def is_enabled(self) -> bool: + return self._enabled + + async def enforce(self, **_kwargs) -> bool: + return True + + async def batch_enforce(self, **kwargs) -> list[bool]: + return [True] * len(kwargs.get("requests", [])) + + async def invalidate_user(self, *_args, **_kwargs) -> None: + return None + + async def invalidate_all(self, *_args, **_kwargs) -> None: + return None + + +@pytest.fixture +def patch_authz(monkeypatch): + """Install a stub authz service into both the shares module and utils.""" + from langflow.services.authorization import utils as authz_utils + + def _apply(*, cross_user: bool = False, enabled: bool = False) -> _StubAuthz: + stub = _StubAuthz(cross_user=cross_user, enabled=enabled) + monkeypatch.setattr(shares_module, "get_authorization_service", lambda: stub) + monkeypatch.setattr(authz_utils, "get_authorization_service", lambda: stub) + # Disable audit + AUTHZ_ENABLED gating so audit calls inside ensure_* + # don't open real sessions. + settings = SimpleNamespace( + auth_settings=SimpleNamespace(AUTHZ_ENABLED=False, AUTHZ_AUDIT_ENABLED=False), + ) + monkeypatch.setattr(authz_utils, "get_settings_service", lambda: settings) + return stub + + return _apply + + +@pytest.fixture +def silence_audit(monkeypatch): + """Replace audit_decision with a no-op so we don't spawn background tasks.""" + + async def _noop(**_kwargs): + return None + + monkeypatch.setattr(shares_module, "audit_decision", _noop) + + +def _make_user(*, is_superuser: bool = False) -> SimpleNamespace: + return SimpleNamespace(id=uuid4(), is_superuser=is_superuser, username="u") + + +def _make_flow_owned_by(owner_id: UUID) -> Any: + from langflow.services.database.models.flow.model import Flow + + return SimpleNamespace(_model=Flow, id=uuid4(), user_id=owner_id) + + +def _payload_for(resource_id: UUID) -> ShareCreate: + return ShareCreate( + resource_type="flow", + resource_id=resource_id, + scope=ShareScope.USER.value, + target_id=uuid4(), + permission_level=SharePermissionLevel.READ.value, + ) + + +# --------------------------------------------------------------------------- # +# CREATE — OSS floor must block non-owner / non-superuser +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_create_share_blocks_non_owner_under_oss_passthrough(patch_authz, silence_audit): # noqa: ARG001 + """A non-owner cannot mint a share row for another user's flow under OSS.""" + from langflow.services.database.models.flow.model import Flow + + patch_authz(cross_user=False, enabled=False) + + owner = _make_user() + attacker = _make_user() + flow = SimpleNamespace(id=uuid4(), user_id=owner.id) + session = _FakeAsyncSession({(Flow, flow.id): flow}) + payload = _payload_for(flow.id) + + with pytest.raises(HTTPException) as excinfo: + await shares_module.create_share(payload=payload, current_user=attacker, session=session) + + assert excinfo.value.status_code == 403 + assert "Only the resource owner" in excinfo.value.detail + # Floor fires before any DB write — no share row was added. + assert session.added == [] + + +@pytest.mark.asyncio +async def test_create_share_allows_owner_under_oss_passthrough(patch_authz, silence_audit): # noqa: ARG001 + """The resource owner can mint a share row under OSS pass-through.""" + from langflow.services.database.models.flow.model import Flow + + patch_authz(cross_user=False, enabled=False) + + owner = _make_user() + flow = SimpleNamespace(id=uuid4(), user_id=owner.id) + session = _FakeAsyncSession({(Flow, flow.id): flow}) + payload = _payload_for(flow.id) + + result = await shares_module.create_share(payload=payload, current_user=owner, session=session) + + assert result.resource_id == flow.id + assert len(session.added) == 1 + assert session.flushed == 1 + + +@pytest.mark.asyncio +async def test_create_share_allows_superuser_under_oss_passthrough(patch_authz, silence_audit): # noqa: ARG001 + """A superuser can mint a share row for a resource they don't own.""" + from langflow.services.database.models.flow.model import Flow + + patch_authz(cross_user=False, enabled=False) + + owner = _make_user() + admin = _make_user(is_superuser=True) + flow = SimpleNamespace(id=uuid4(), user_id=owner.id) + session = _FakeAsyncSession({(Flow, flow.id): flow}) + payload = _payload_for(flow.id) + + result = await shares_module.create_share(payload=payload, current_user=admin, session=session) + + assert result.resource_id == flow.id + assert len(session.added) == 1 + + +@pytest.mark.asyncio +async def test_create_share_returns_404_when_resource_missing(patch_authz, silence_audit): # noqa: ARG001 + """A missing resource yields 404 — not 403 — to preserve UUID privacy.""" + patch_authz(cross_user=False, enabled=False) + + attacker = _make_user() + session = _FakeAsyncSession({}) # no resource present + payload = _payload_for(uuid4()) + + with pytest.raises(HTTPException) as excinfo: + await shares_module.create_share(payload=payload, current_user=attacker, session=session) + + assert excinfo.value.status_code == 404 + + +# --------------------------------------------------------------------------- # +# PATCH — same floor +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_update_share_blocks_non_owner_under_oss_passthrough(patch_authz, silence_audit): # noqa: ARG001 + """A non-owner cannot PATCH a share on another user's resource under OSS.""" + from langflow.services.database.models.flow.model import Flow + + patch_authz(cross_user=False, enabled=False) + + owner = _make_user() + attacker = _make_user() + flow = SimpleNamespace(id=uuid4(), user_id=owner.id) + share = AuthzShare( + id=uuid4(), + resource_type="flow", + resource_id=flow.id, + scope=ShareScope.USER.value, + target_id=uuid4(), + permission_level=SharePermissionLevel.READ.value, + created_by=attacker.id, # attacker is even the creator — floor still blocks + ) + session = _FakeAsyncSession({(AuthzShare, share.id): share, (Flow, flow.id): flow}) + update = ShareUpdate(permission_level=SharePermissionLevel.WRITE.value) + + with pytest.raises(HTTPException) as excinfo: + await shares_module.update_share( + share_id=share.id, + payload=update, + current_user=attacker, + session=session, + ) + + assert excinfo.value.status_code == 403 + # PATCH was rejected — no flush should have happened. + assert session.flushed == 0 + + +@pytest.mark.asyncio +async def test_update_share_allows_owner_under_oss_passthrough(patch_authz, silence_audit): # noqa: ARG001 + """The resource owner can PATCH a share on their resource under OSS.""" + from langflow.services.database.models.flow.model import Flow + + patch_authz(cross_user=False, enabled=False) + + owner = _make_user() + flow = SimpleNamespace(id=uuid4(), user_id=owner.id) + share = AuthzShare( + id=uuid4(), + resource_type="flow", + resource_id=flow.id, + scope=ShareScope.USER.value, + target_id=uuid4(), + permission_level=SharePermissionLevel.READ.value, + created_by=owner.id, + ) + session = _FakeAsyncSession({(AuthzShare, share.id): share, (Flow, flow.id): flow}) + update = ShareUpdate(permission_level=SharePermissionLevel.WRITE.value) + + result = await shares_module.update_share( + share_id=share.id, + payload=update, + current_user=owner, + session=session, + ) + + assert result.permission_level == SharePermissionLevel.WRITE.value + assert session.flushed == 1 + + +# --------------------------------------------------------------------------- # +# DELETE — same floor +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_delete_share_blocks_non_owner_under_oss_passthrough(patch_authz, silence_audit): # noqa: ARG001 + """A non-owner cannot DELETE a share on another user's resource under OSS.""" + from langflow.services.database.models.flow.model import Flow + + patch_authz(cross_user=False, enabled=False) + + owner = _make_user() + attacker = _make_user() + flow = SimpleNamespace(id=uuid4(), user_id=owner.id) + share = AuthzShare( + id=uuid4(), + resource_type="flow", + resource_id=flow.id, + scope=ShareScope.USER.value, + target_id=uuid4(), + permission_level=SharePermissionLevel.READ.value, + created_by=attacker.id, # attacker created it but is not the resource owner + ) + session = _FakeAsyncSession({(AuthzShare, share.id): share, (Flow, flow.id): flow}) + + with pytest.raises(HTTPException) as excinfo: + await shares_module.delete_share(share_id=share.id, current_user=attacker, session=session) + + assert excinfo.value.status_code == 403 + # The floor blocks before the DELETE. + assert session.deleted == [] + + +@pytest.mark.asyncio +async def test_delete_share_allows_owner_under_oss_passthrough(patch_authz, silence_audit): # noqa: ARG001 + """The resource owner can DELETE a share on their resource under OSS.""" + from langflow.services.database.models.flow.model import Flow + + patch_authz(cross_user=False, enabled=False) + + owner = _make_user() + flow = SimpleNamespace(id=uuid4(), user_id=owner.id) + share = AuthzShare( + id=uuid4(), + resource_type="flow", + resource_id=flow.id, + scope=ShareScope.USER.value, + target_id=uuid4(), + permission_level=SharePermissionLevel.READ.value, + created_by=owner.id, + ) + session = _FakeAsyncSession({(AuthzShare, share.id): share, (Flow, flow.id): flow}) + + await shares_module.delete_share(share_id=share.id, current_user=owner, session=session) + + assert len(session.deleted) == 1 + + +# --------------------------------------------------------------------------- # +# Floor behavior when the enterprise plugin is active +# --------------------------------------------------------------------------- # + + +@pytest.mark.asyncio +async def test_floor_is_skipped_when_enterprise_plugin_active(patch_authz, silence_audit): # noqa: ARG001 + """OSS floor is skipped when the enterprise plugin is actively enforcing. + + When supports_cross_user_fetch=True AND AUTHZ_ENABLED=true, the OSS floor + is skipped so a plugin-granted share:create role can administer shares on + another user's resource. ``ensure_share_permission`` (mocked to allow here) + becomes the authoritative check. + """ + from langflow.services.database.models.flow.model import Flow + + patch_authz(cross_user=True, enabled=True) + + owner = _make_user() + delegate = _make_user() # non-owner, but allowed by enterprise policy + flow = SimpleNamespace(id=uuid4(), user_id=owner.id) + session = _FakeAsyncSession({(Flow, flow.id): flow}) + payload = _payload_for(flow.id) + + result = await shares_module.create_share(payload=payload, current_user=delegate, session=session) + + # The floor is skipped; the stub authz allows the operation; the row is + # written. If the floor still fired we'd see a 403 instead. + assert result.resource_id == flow.id + assert len(session.added) == 1 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 new file mode 100644 index 0000000000..062482b401 --- /dev/null +++ b/src/backend/tests/unit/services/authorization/test_route_guard_regressions.py @@ -0,0 +1,142 @@ +"""Regression tests for route guard fixes from the OSS authorization PR review. + +Source-level checks asserting that: + +* The paginated ``read_flows`` branch passes ``owner_extractor`` to + ``filter_visible_resources`` (so an enterprise enforcer cannot hide the + caller's own flows in paginated mode). +* ``upsert_flow`` consults ``supports_cross_user_fetch`` + ``is_enabled`` + before falling back to the hardcoded ownership check, and wraps + ``ensure_flow_permission`` with ``deny_to_404``. +* ``delete_multiple_flows`` and ``download_multiple_file`` do not pre-scope by + ``Flow.user_id == user.id`` unconditionally; they gate that filter on the + same capability check. +* ``load_flow`` authorizes EXECUTE before returning a Graph. +""" + +from __future__ import annotations + +import ast +from pathlib import Path + +import pytest + +_API_V1 = Path(__file__).resolve().parents[4] / "base" / "langflow" / "api" / "v1" +_FLOWS_FILE = _API_V1 / "flows.py" +_HELPERS_FLOW = Path(__file__).resolve().parents[4] / "base" / "langflow" / "helpers" / "flow.py" + + +def _parse_async_funcs(path: Path) -> dict[str, ast.AsyncFunctionDef]: + tree = ast.parse(path.read_text()) + return {node.name: node for node in ast.walk(tree) if isinstance(node, ast.AsyncFunctionDef)} + + +def _calls(func: ast.AsyncFunctionDef, name: str) -> list[ast.Call]: + out: list[ast.Call] = [] + for node in ast.walk(func): + if not isinstance(node, ast.Call): + continue + target = node.func + is_named = isinstance(target, ast.Name) and target.id == name + is_attr = isinstance(target, ast.Attribute) and target.attr == name + if is_named or is_attr: + out.append(node) + return out + + +def _has_keyword(call: ast.Call, name: str) -> bool: + return any(kw.arg == name for kw in call.keywords) + + +@pytest.fixture(scope="module") +def flows_routes() -> dict[str, ast.AsyncFunctionDef]: + return _parse_async_funcs(_FLOWS_FILE) + + +@pytest.fixture(scope="module") +def helpers_funcs() -> dict[str, ast.AsyncFunctionDef]: + return _parse_async_funcs(_HELPERS_FLOW) + + +def test_read_flows_paginated_branch_passes_owner_extractor(flows_routes): + """The paginated filter_visible_resources call must pass owner_extractor. + + Both branches of read_flows (``get_all=True`` and pagination) must give + the enforcer the same hint about which flows are owner-owned. Omitting it + on the paginated branch lets an enterprise plugin without an explicit + owner-allow policy hide the caller's own flows when paginating. + """ + func = flows_routes["read_flows"] + fvr_calls = _calls(func, "filter_visible_resources") + assert len(fvr_calls) >= 2, "read_flows should call filter_visible_resources on both branches" + for call in fvr_calls: + assert _has_keyword(call, "owner_extractor"), ( + "Every filter_visible_resources call in read_flows must pass owner_extractor " + "so the enforcer short-circuits on owner-allow" + ) + + +def test_upsert_flow_consults_cross_user_fetch_capability(flows_routes): + """upsert_flow must not unconditionally raise 404 for non-owners. + + With an enterprise plugin registered, a valid WRITE share grant must be + honored. The current shape gates the hardcoded ownership floor on + ``supports_cross_user_fetch() and is_enabled()``. + """ + func = flows_routes["upsert_flow"] + src = ast.unparse(func) + assert "supports_cross_user_fetch" in src, "upsert_flow must consult cross-user-fetch capability" + assert "is_enabled" in src, "upsert_flow must consult is_enabled before the OSS floor" + + +def test_upsert_flow_wraps_ensure_in_deny_to_404(flows_routes): + """upsert_flow must convert ensure_flow_permission 403 to 404 for UUID privacy.""" + func = flows_routes["upsert_flow"] + src = ast.unparse(func) + assert "deny_to_404" in src, "upsert_flow must wrap ensure_flow_permission with deny_to_404" + + +def test_delete_multiple_flows_does_not_unconditionally_prescope(flows_routes): + """Bulk delete must not unconditionally filter by Flow.user_id == user.id. + + The pre-scope is now gated on whether the registered service supports + cross-user fetch — otherwise valid share DELETE grants are silently + dropped from the working set. + """ + func = flows_routes["delete_multiple_flows"] + src = ast.unparse(func) + assert "supports_cross_user_fetch" in src, ( + "delete_multiple_flows must gate the owner pre-scope on cross-user-fetch capability" + ) + + +def test_download_multiple_file_does_not_unconditionally_prescope(flows_routes): + """Same as above for the bulk download endpoint.""" + func = flows_routes["download_multiple_file"] + src = ast.unparse(func) + assert "supports_cross_user_fetch" in src, ( + "download_multiple_file must gate the owner pre-scope on cross-user-fetch capability" + ) + assert "deny_to_404" in src, ( + "download_multiple_file must convert ensure_flow_permission 403 to 404 for UUID privacy" + ) + + +def test_load_flow_calls_ensure_flow_permission(helpers_funcs): + """load_flow (component-reachable graph loader) must authorize EXECUTE. + + Reachable from custom components, sub-flow runners, and ``run_flow``. + Without this guard, a caller can pass an arbitrary flow_id and pull + another user's flow definition (prompts, tools, embedded credentials). + """ + func = helpers_funcs["load_flow"] + calls = _calls(func, "ensure_flow_permission") + assert calls, "load_flow must call ensure_flow_permission" + # Action must be EXECUTE — the function loads a Graph for execution. + saw_execute = False + for call in calls: + if len(call.args) >= 2: + arg = call.args[1] + if isinstance(arg, ast.Attribute) and arg.attr == "EXECUTE": + saw_execute = True + assert saw_execute, "load_flow must authorize FlowAction.EXECUTE" diff --git a/src/backend/tests/unit/services/authorization/test_utils.py b/src/backend/tests/unit/services/authorization/test_utils.py index d8ae00813b..180f0202e4 100644 --- a/src/backend/tests/unit/services/authorization/test_utils.py +++ b/src/backend/tests/unit/services/authorization/test_utils.py @@ -344,9 +344,15 @@ async def test_owner_override_skips_enforce(monkeypatch, fake_user): @pytest.mark.anyio -async def test_owner_override_when_disabled_skips_audit(monkeypatch, fake_user): - """Owner override on a disabled-authz install does not even write audit (consistency with ensure_permission).""" - _install_settings(monkeypatch, authz_enabled=False) +async def test_owner_override_audits_even_when_authz_disabled(monkeypatch, fake_user): + """Owner override on a disabled-authz install still writes an audit row. + + Audit is now gated only on ``AUTHZ_AUDIT_ENABLED`` (the audit recorder + here intercepts at that level), so operators can observe traffic ahead of + flipping ``AUTHZ_ENABLED``. ``_install_settings`` defaults the audit flag + to False so we explicitly enable it here. + """ + _install_settings(monkeypatch, authz_enabled=False, audit_enabled=True) _install_authz(monkeypatch, _StubAuthorizationService(allow=False)) audit_calls = _install_audit_recorder(monkeypatch) @@ -356,7 +362,8 @@ async def test_owner_override_when_disabled_skips_audit(monkeypatch, fake_user): flow_id=uuid4(), flow_user_id=fake_user.id, ) - assert audit_calls == [] + assert len(audit_calls) == 1 + assert audit_calls[0]["result"] == "owner_override" @pytest.mark.anyio @@ -477,14 +484,39 @@ def test_split_obj_non_uuid_suffix_returns_none_id(): @pytest.mark.anyio -async def test_audit_decision_noop_when_authz_disabled(monkeypatch): - """audit_decision exits immediately when AUTHZ_ENABLED=False.""" +async def test_audit_decision_runs_when_authz_disabled_but_audit_on(monkeypatch): + """Audit is independent of enforcement now. + + Previously ``audit_decision`` short-circuited when ``AUTHZ_ENABLED=False``, + which meant share CRUD writes left no audit trail on default installs. The + new contract gates only on ``AUTHZ_AUDIT_ENABLED`` so operators can + observe traffic before flipping enforcement on. + """ _install_settings(monkeypatch, authz_enabled=False, audit_enabled=True) scheduled: list[object] = [] - monkeypatch.setattr("asyncio.create_task", lambda coro: scheduled.append(coro) or coro) + + class _FakeTask: + def add_done_callback(self, cb): + self._cb = cb + + def done(self) -> bool: + return False + + def _capture(coro): + coro.close() + task = _FakeTask() + scheduled.append(task) + return task + + monkeypatch.setattr("asyncio.create_task", _capture) await authz_utils.audit_decision(user_id=uuid4(), action="flow:read", obj="flow:x", result="allow") - assert scheduled == [] + try: + assert len(scheduled) == 1 + finally: + # Don't pollute the module-global pending set for downstream tests. + for task in scheduled: + authz_utils._pending_audit_tasks.discard(task) @pytest.mark.anyio @@ -500,18 +532,92 @@ async def test_audit_decision_noop_when_audit_disabled(monkeypatch): @pytest.mark.anyio async def test_audit_decision_schedules_task_when_enabled(monkeypatch): - """A scheduled coroutine is produced when both flags are on.""" + """A scheduled coroutine is produced when both flags are on. + + The new implementation tracks the task in ``_pending_audit_tasks`` so the + event loop cannot GC it mid-write and so shutdown can drain it. Our mock + returns a lightweight stand-in that supports ``add_done_callback`` exactly + as a real ``asyncio.Task`` does. + """ _install_settings(monkeypatch, authz_enabled=True, audit_enabled=True) scheduled: list[object] = [] + class _FakeTask: + def __init__(self, coro): + self._coro = coro + self._callbacks: list = [] + + def add_done_callback(self, cb): + self._callbacks.append(cb) + + def done(self) -> bool: + return False + def _capture(coro): - scheduled.append(coro) coro.close() # don't actually run — the function imports from DB modules. + task = _FakeTask(coro) + scheduled.append(task) + return task monkeypatch.setattr("asyncio.create_task", _capture) await authz_utils.audit_decision(user_id=uuid4(), action="flow:read", obj="flow:x", result="allow") - assert len(scheduled) == 1 + try: + assert len(scheduled) == 1 + # The implementation must hold a reference so the GC can't drop the task. + assert scheduled[0] in authz_utils._pending_audit_tasks + # And it must register a done-callback that removes itself from the set. + assert scheduled[0]._callbacks, "audit_decision must register a done-callback to clean up _pending_audit_tasks" + finally: + # The fake task is not a real ``asyncio.Task`` — clean up the + # module-global set so the next test starts from an empty state. + authz_utils._pending_audit_tasks.discard(scheduled[0]) + + +@pytest.mark.anyio +async def test_drain_pending_audit_writes_awaits_tasks(): + """drain_pending_audit_writes waits for tracked tasks to finish.""" + import asyncio + + finished = asyncio.Event() + + async def _slow() -> None: + await asyncio.sleep(0.01) + finished.set() + + task = asyncio.create_task(_slow()) + authz_utils._pending_audit_tasks.add(task) + task.add_done_callback(authz_utils._pending_audit_tasks.discard) + + await authz_utils.drain_pending_audit_writes(timeout=1.0) + assert finished.is_set() + assert task not in authz_utils._pending_audit_tasks + + +@pytest.mark.anyio +async def test_ensure_permission_fails_closed_on_plugin_exception(monkeypatch, fake_user): + """If the authz plugin raises, ensure_permission must deny (403), not bubble 500.""" + _install_settings(monkeypatch, authz_enabled=True, audit_enabled=False) + + class _BrokenPlugin: + async def enforce(self, **_kwargs): + msg = "casbin db down" + raise RuntimeError(msg) + + async def batch_enforce(self, **_kwargs): + return [] + + monkeypatch.setattr(authz_utils, "get_authorization_service", lambda: _BrokenPlugin()) + captured = _install_audit_recorder(monkeypatch) + + with pytest.raises(HTTPException) as excinfo: + await authz_utils.ensure_permission(fake_user, domain="*", obj="flow:abc", act="read") + + assert excinfo.value.status_code == 403, "Plugin exceptions must fail closed (deny), not 500" + # The deny path must still emit an audit row so the operator can see the failure. + assert captured, "Plugin exception must still produce an audit row" + assert captured[0]["result"] == "deny" + assert "error" in captured[0]["details"] # ----------------------------------------------------------------------------- # diff --git a/src/lfx/src/lfx/services/authorization/base.py b/src/lfx/src/lfx/services/authorization/base.py index 0194308cbe..1276eb369f 100644 --- a/src/lfx/src/lfx/services/authorization/base.py +++ b/src/lfx/src/lfx/services/authorization/base.py @@ -3,7 +3,8 @@ from __future__ import annotations import abc -from typing import TYPE_CHECKING, Any, ClassVar +from typing import TYPE_CHECKING, Any, ClassVar, TypedDict +from uuid import UUID as _UUID from lfx.services.base import Service from lfx.services.schema import ServiceType @@ -13,6 +14,47 @@ if TYPE_CHECKING: from uuid import UUID +class AuthzContext(TypedDict, total=False): + """Documented shape of the ``context`` dict passed to authz plugins. + + Plugins receive a ``dict[str, Any]`` for forward compatibility — new + callers may add fields without forcing a plugin recompile — but this + TypedDict documents the keys the framework promises to populate and the + keys plugins should be prepared to read. + + Framework-populated keys + ------------------------ + is_superuser : bool + Always present. ``getattr(user, "is_superuser", False)``. + + Resource-owner hints (one is populated per call, others omitted) + ---------------------------------------------------------------- + flow_user_id, deployment_user_id, project_user_id, knowledge_base_user_id, + variable_user_id, file_user_id, share_user_id : UUID | None + The id of the user who owns the resource being authorized. The + framework's ``ensure_*_permission`` helpers short-circuit on owner + match, so this key is the source of truth for the owner-override path. + + Domain / scope hints + -------------------- + workspace_id : UUID | None + The workspace this resource belongs to, when applicable. + folder_id : UUID | None + The folder/project this resource belongs to, when applicable. + """ + + is_superuser: bool + flow_user_id: _UUID | None + deployment_user_id: _UUID | None + project_user_id: _UUID | None + knowledge_base_user_id: _UUID | None + variable_user_id: _UUID | None + file_user_id: _UUID | None + share_user_id: _UUID | None + workspace_id: _UUID | None + folder_id: _UUID | None + + class BaseAuthorizationService(Service, abc.ABC): """Abstract base class for authorization (RBAC) services. diff --git a/src/lfx/src/lfx/services/manager.py b/src/lfx/src/lfx/services/manager.py index 6b3dce7959..2401ac7d91 100644 --- a/src/lfx/src/lfx/services/manager.py +++ b/src/lfx/src/lfx/services/manager.py @@ -390,17 +390,44 @@ class ServiceManager: ) return + # Map of service_type → expected base class. Used to reject malformed + # entry-point plugins before they take over an authorization-critical + # service. Without this check an arbitrary class from an unrelated + # PyPI package that happened to register the same entry-point name + # would silently replace the OSS implementation. + expected_bases: dict[ServiceType, type] = {} + try: + from lfx.services.authorization.base import BaseAuthorizationService + + expected_bases[ServiceType.AUTHORIZATION_SERVICE] = BaseAuthorizationService + except Exception as exc: # noqa: BLE001 — optional import, validation just skipped + logger.debug(f"BaseAuthorizationService unavailable; entry-point validation skipped: {exc}") + for ep in eps: try: service_class = ep.load() # Entry point name should match ServiceType enum value service_type = ServiceType(ep.name) + expected_base = expected_bases.get(service_type) + if expected_base is not None and not ( + isinstance(service_class, type) and issubclass(service_class, expected_base) + ): + logger.warning( + f"Entry point {ep.name} resolved to {service_class!r}, " + f"which is not a subclass of {expected_base.__name__}. " + f"Skipping registration to avoid silently replacing the " + f"built-in service with an incompatible plugin." + ) + continue self.register_service_class(service_type, service_class, override=False) logger.debug(f"Loaded service from entry point: {ep.name}") except (ValueError, AttributeError) as exc: logger.warning(f"Failed to load entry point {ep.name}: {exc}") except Exception as exc: # noqa: BLE001 - logger.debug(f"Error loading entry point {ep.name}: {exc}") + # Authz plugin failures are operator-visible — silent + # degradation to the OSS pass-through is exactly the kind + # of behavior change we want noisy. + logger.warning(f"Error loading entry point {ep.name}: {exc}") def _discover_from_config(self, config_dir: Path) -> None: """Discover services from config files (lfx.toml / pyproject.toml).""" diff --git a/src/lfx/src/lfx/services/settings/auth.py b/src/lfx/src/lfx/services/settings/auth.py index c7d359392b..b854efbcf9 100644 --- a/src/lfx/src/lfx/services/settings/auth.py +++ b/src/lfx/src/lfx/services/settings/auth.py @@ -145,8 +145,9 @@ class AuthSettings(BaseSettings): AUTHZ_AUDIT_ENABLED: bool = Field( default=True, description=( - "Write an AuthzAuditLog row for every authorization decision when AUTHZ_ENABLED=true. " - "Ignored when AUTHZ_ENABLED=false." + "Write an AuthzAuditLog row for every authorization decision and share-administration " + "action. Independent of AUTHZ_ENABLED — keep audit on while enforcement is off to " + "observe traffic before flipping the AUTHZ_ENABLED flag." ), )