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

feat: Phase 3 and 4 implementations for RBAC

Browse Source
This commit is contained in:
Eric Hare
2026-05-21 11:07:02 -07:00
parent 87bebd9bb8
commit 505e8569c8
28 changed files with 2158 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
AGENTS.md
View File

@ -109,6 +109,10 @@ Route guards live in `langflow.services.authorization.utils`:
- `ensure_flow_permission(user, FlowAction.*, flow_id=..., flow_user_id=..., workspace_id=..., folder_id=...)` — single-flow CRUD + execute
- `ensure_deployment_permission(user, DeploymentAction.*, deployment_id=..., deployment_user_id=..., workspace_id=..., project_id=...)`
- `ensure_project_permission(user, ProjectAction.*, project_id=..., project_user_id=..., workspace_id=...)`
- `ensure_knowledge_base_permission(user, KnowledgeBaseAction.*, kb_name=..., kb_user_id=...)`
- `ensure_variable_permission(user, VariableAction.*, variable_id=..., variable_user_id=...)`
- `ensure_file_permission(user, FileAction.*, file_id=..., file_user_id=...)`
- `ensure_share_permission(user, ShareAction.*, share_id=..., share_user_id=...)`
- `filter_visible_resources(user, resource_type=..., candidates=..., act=...)` — list-endpoint filter; safe no-op in OSS
The Casbin request shape is `(subject, domain, object, action)`:
@ -119,7 +123,13 @@ The Casbin request shape is `(subject, domain, object, action)`:
Use `langflow authz dry-run` to simulate a built-in policy against the live audit table without enabling enforcement.
**Phase 1/2 contract — owner-scoped fetch:** the route guards above sit on top of fetch helpers (`_read_flow`, `get_flow_for_api_key_user`, `get_deployment`, project reads in `projects.py`) that still scope queries by `current_user.id`. That means even with an enterprise plugin registered, a share grant on a non-owned flow / folder / deployment **still returns 404 at the fetch layer before the guard can authorize**. Cross-user enforcement (the case where a non-owner with a share grant can read/write/execute a resource) lands in Phase 3 alongside `authz_share` CRUD APIs and share-aware fetch helpers that load by id first and convert plugin denies to 404 to preserve UUID-privacy. The current guards exist, are wired, and emit audit rows — but the cross-user reachability they enable is a Phase 3 prerequisite, not a Phase 1/2 deliverable.
**Share-aware fetch (Phase 3):** route fetch helpers (`_read_flow`, `get_flow_by_id_or_endpoint_name`, `get_deployment`, project reads in `projects.py`, v2 file fetcher) branch on `BaseAuthorizationService.supports_cross_user_fetch()`. The OSS pass-through reports `False` so the existing owner-scoped queries are preserved — enabling `LANGFLOW_AUTHZ_ENABLED=true` without an enterprise plugin cannot widen visibility. Enterprise plugins set `SUPPORTS_CROSS_USER_FETCH=True` so resources load by id alone and `ensure_*_permission` decides access; route handlers can convert a plugin-deny `HTTPException(403)` to `HTTPException(404)` via `langflow.services.authorization.fetch.deny_to_404` to preserve UUID privacy.
**Share CRUD API (Phase 3):** `/api/v1/authz/shares` provides POST / GET / PATCH / DELETE on `authz_share` rows. The handler enforces an OSS floor (resource owner or superuser may administer shares for that resource) so the OSS pass-through cannot let a non-owner mint share rows. Each write fires `BaseAuthorizationService.invalidate_user` / `invalidate_all` so an enterprise enforcer can drop cached policy. Audit rows are written via `audit_decision` with `share:create` / `share:update` / `share:delete` actions.
**Audit query API (Phase 4):** `GET /api/v1/authz/audit` (superuser-only) exposes a paginated, filterable view of `authz_audit_log`. Supports `user_id`, `resource_type`, `resource_id`, `action`, `result`, `since`, `until` filters; page size capped at 200.
**Default role catalog (Phase 4):** the seed migration `8d3a1f9c2e0b_seed_authz_system_roles` inserts the three built-in `is_system=True` roles (viewer / developer / admin) with `"{resource}:{action}"` permission slugs. OSS does not interpret these — they exist so an enterprise plugin's `PolicySync` has a stable bootstrap source.
## Component Development

161
src/backend/base/langflow/alembic/versions/8d3a1f9c2e0b_seed_authz_system_roles.py Normal file
View File

@ -0,0 +1,161 @@
"""seed authz system roles — viewer / developer / admin
Revision ID: 8d3a1f9c2e0b
Revises: 7c8d9e0f1a2b
Create Date: 2026-05-21
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.
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
import json
from datetime import datetime, timezone
from typing import TYPE_CHECKING
from uuid import uuid4
import sqlalchemy as sa
from alembic import op
if TYPE_CHECKING:
from collections.abc import Sequence
revision: str = "8d3a1f9c2e0b" # pragma: allowlist secret
down_revision: str | None = "7c8d9e0f1a2b" # pragma: allowlist secret
branch_labels: str | Sequence[str] | None = None
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
# matching ``p`` rules during ``PolicySync``.
_VIEWER_PERMISSIONS: tuple[str, ...] = (
"flow:read",
"flow:execute",
"deployment:read",
"project:read",
"knowledge_base:read",
"variable:read",
"file:read",
)
_DEVELOPER_EXTRA: tuple[str, ...] = (
"flow:write",
"flow:create",
"deployment:write",
"deployment:create",
"deployment:execute",
"project:write",
"project:create",
"knowledge_base:write",
"knowledge_base:create",
"knowledge_base:ingest",
"variable:write",
"variable:create",
"file:write",
"file:create",
)
_ADMIN_EXTRA: tuple[str, ...] = (
"flow:delete",
"flow:deploy",
"deployment:delete",
"deployment:deploy",
"project:delete",
"knowledge_base:delete",
"variable:delete",
"file:delete",
"share:read",
"share:create",
"share:update",
"share:delete",
)
_SYSTEM_ROLES: tuple[tuple[str, str, tuple[str, ...]], ...] = (
(
"viewer",
"Read-only access to flows, deployments, projects, and supporting resources.",
_VIEWER_PERMISSIONS,
),
(
"developer",
"Author and execute flows; manage variables, knowledge bases, and files.",
_VIEWER_PERMISSIONS + _DEVELOPER_EXTRA,
),
(
"admin",
"Full management of resources and shares within the workspace.",
_VIEWER_PERMISSIONS + _DEVELOPER_EXTRA + _ADMIN_EXTRA,
),
)
def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def upgrade() -> None:
conn = op.get_bind()
authz_role = sa.table(
"authz_role",
sa.column("id", sa.Uuid()),
sa.column("name", sa.String()),
sa.column("description", sa.String()),
sa.column("is_system", sa.Boolean()),
sa.column("permissions", sa.JSON()),
sa.column("parent_role_id", sa.Uuid()),
sa.column("workspace_id", sa.Uuid()),
sa.column("created_at", sa.DateTime(timezone=True)),
sa.column("updated_at", sa.DateTime(timezone=True)),
sa.column("created_by", sa.Uuid()),
)
timestamp = _now_iso()
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
conn.execute(
authz_role.insert().values(
id=str(uuid4()),
name=name,
description=description,
is_system=True,
permissions=json.dumps(list(permissions)),
parent_role_id=None,
workspace_id=None,
created_at=timestamp,
updated_at=timestamp,
created_by=None,
)
)
def downgrade() -> None:
conn = op.get_bind()
authz_role = sa.table(
"authz_role",
sa.column("name", sa.String()),
sa.column("is_system", sa.Boolean()),
)
conn.execute(
authz_role.delete().where(
sa.and_(
authz_role.c.name.in_([name for name, _, _ in _SYSTEM_ROLES]),
authz_role.c.is_system.is_(True),
)
)
)

4
src/backend/base/langflow/api/router.py
View File

@ -4,6 +4,8 @@ from lfx.services.settings.feature_flags import FEATURE_FLAGS
from langflow.api.v1 import (
api_key_router,
authz_audit_router,
authz_shares_router,
chat_router,
endpoints_router,
extensions_router,
@ -77,6 +79,8 @@ router_v1.include_router(mcp_projects_router)
router_v1.include_router(openai_responses_router)
router_v1.include_router(models_router)
router_v1.include_router(model_options_router)
router_v1.include_router(authz_shares_router)
router_v1.include_router(authz_audit_router)
# Extension reload is Mode A (local-dev / pip-installed) only. The route is

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

@ -1,4 +1,6 @@
from langflow.api.v1.api_key import router as api_key_router
from langflow.api.v1.authz_audit import router as authz_audit_router
from langflow.api.v1.authz_shares import router as authz_shares_router
from langflow.api.v1.chat import router as chat_router
from langflow.api.v1.endpoints import router as endpoints_router
from langflow.api.v1.extensions import router as extensions_router
@ -27,6 +29,8 @@ from langflow.api.v1.voice_mode import router as voice_mode_router
__all__ = [
"api_key_router",
"authz_audit_router",
"authz_shares_router",
"chat_router",
"endpoints_router",
"extensions_router",

122
src/backend/base/langflow/api/v1/authz_audit.py Normal file
View File

@ -0,0 +1,122 @@
"""Admin-only query endpoint for the ``authz_audit_log`` table.
The OSS guards write a row per authorization decision (allow / deny /
owner_override). Without a read API operators have to query the DB by hand to
investigate "why was this denied?" — this router exposes the table behind a
superuser-only filter surface so support and compliance flows can use it
without direct DB access.
"""
from __future__ import annotations
from datetime import datetime
from typing import Annotated
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from sqlmodel import col, select
from langflow.api.utils import DbSession
from langflow.services.auth.utils import get_current_active_superuser
from langflow.services.database.models.auth import AuthzAuditLog
from langflow.services.database.models.user.model import User
router = APIRouter(prefix="/authz/audit", tags=["Authorization"])
_MAX_PAGE_SIZE = 200
class AuthzAuditLogRead(BaseModel):
"""Read-only projection of an ``AuthzAuditLog`` row."""
id: UUID
user_id: UUID | None
action: str
resource_type: str | None
resource_id: UUID | None
result: str
details: dict | None
timestamp: datetime
model_config = {"from_attributes": True}
class AuthzAuditPage(BaseModel):
"""Paginated audit-log response."""
items: list[AuthzAuditLogRead]
total: int
page: int
size: int
pages: int
@router.get("", response_model=AuthzAuditPage)
@router.get("/", response_model=AuthzAuditPage)
async def list_audit_log(
session: DbSession,
_admin: Annotated[User, Depends(get_current_active_superuser)],
user_id: Annotated[UUID | None, Query(description="Filter by acting user id.")] = None,
resource_type: Annotated[
str | None,
Query(description="Filter by resource type slug, e.g. ``flow`` or ``deployment``."),
] = None,
resource_id: Annotated[UUID | None, Query(description="Filter by resource UUID.")] = None,
action: Annotated[
str | None,
Query(description="Filter by action string, e.g. ``flow:read`` or ``share:create``."),
] = None,
result: Annotated[
str | None,
Query(description="Filter by decision result (``allow`` / ``deny`` / ``owner_override``)."),
] = None,
since: Annotated[datetime | None, Query(description="Inclusive lower bound on ``timestamp``.")] = None,
until: Annotated[datetime | None, Query(description="Exclusive upper bound on ``timestamp``.")] = None,
page: Annotated[int, Query(ge=1)] = 1,
size: Annotated[int, Query(ge=1, le=_MAX_PAGE_SIZE)] = 50,
) -> AuthzAuditPage:
"""Return a paginated slice of the audit log filtered by the given query params.
Superuser only. The composite indexes on ``(user_id, timestamp)`` and
``(resource_type, resource_id)`` keep both "show me events for user X"
and "show me events on resource Y" fast at scale.
"""
if since is not None and until is not None and since >= until:
raise HTTPException(status_code=400, detail="`since` must be strictly less than `until`")
base = select(AuthzAuditLog)
if user_id is not None:
base = base.where(AuthzAuditLog.user_id == user_id)
if resource_type is not None:
base = base.where(AuthzAuditLog.resource_type == resource_type)
if resource_id is not None:
base = base.where(AuthzAuditLog.resource_id == resource_id)
if action is not None:
base = base.where(AuthzAuditLog.action == action)
if result is not None:
base = base.where(AuthzAuditLog.result == result)
if since is not None:
base = base.where(AuthzAuditLog.timestamp >= since)
if until is not None:
base = base.where(AuthzAuditLog.timestamp < until)
# Two queries: one COUNT(*) for pagination metadata, one for the page
# window itself. SQLAlchemy's func.count is preferred over len(rows) so
# we don't materialise the full result set when the user just wants
# page 1 of many.
from sqlalchemy import func
total_stmt = select(func.count()).select_from(base.subquery())
total = int((await session.exec(total_stmt)).first() or 0)
page_stmt = base.order_by(col(AuthzAuditLog.timestamp).desc()).offset((page - 1) * size).limit(size)
rows = list(await session.exec(page_stmt))
items = [AuthzAuditLogRead.model_validate(row, from_attributes=True) for row in rows]
pages = (total + size - 1) // size if total > 0 else 0
return AuthzAuditPage(items=items, total=total, page=page, size=size, pages=pages)
__all__ = ["AuthzAuditLogRead", "AuthzAuditPage", "router"]

389
src/backend/base/langflow/api/v1/authz_shares.py Normal file
View File

@ -0,0 +1,389 @@
"""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).
"""
from __future__ import annotations
from datetime import datetime, timezone
from typing import Annotated
from uuid import UUID
from fastapi import APIRouter, HTTPException, Query, status
from sqlmodel import select
from langflow.api.utils import CurrentActiveUser, DbSession
from langflow.api.v1.schemas.authz_shares import ShareCreate, ShareRead, ShareUpdate
from langflow.services.authorization import ShareAction, ensure_share_permission
from langflow.services.authorization.utils import audit_decision
from langflow.services.database.models.auth import AuthzShare, SharePermissionLevel, ShareScope
from langflow.services.database.models.deployment.model import Deployment
from langflow.services.database.models.file.model import File as UserFile
from langflow.services.database.models.flow.model import Flow
from langflow.services.database.models.folder.model import Folder
from langflow.services.database.models.user.model import User
from langflow.services.database.models.variable.model import Variable
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 are filesystem-keyed and have no DB row to verify owner against, so
# they are intentionally absent: a KB owner is always the share creator at
# share-create time (the kb path is rooted under ``current_user.username``).
_RESOURCE_OWNER_LOOKUPS: dict[str, tuple[type, str]] = {
"flow": (Flow, "user_id"),
"deployment": (Deployment, "user_id"),
"project": (Folder, "user_id"),
"variable": (Variable, "user_id"),
"file": (UserFile, "user_id"),
}
async def _resolve_resource_owner(
session: DbSession,
*,
resource_type: str,
resource_id: UUID,
) -> UUID | None:
"""Return the owner ``user_id`` for the named resource, or None if not found.
Knowledge bases have no DB row to consult, so callers must accept ``None``
and apply a name-namespace check separately (the caller is the only
legitimate owner under the OSS contract).
"""
if resource_type == "knowledge_base":
return None
lookup = _RESOURCE_OWNER_LOOKUPS.get(resource_type)
if lookup is None:
return None
model, owner_attr = lookup
row = await session.get(model, resource_id)
if row is None:
return None
return getattr(row, owner_attr, None)
def _ensure_can_administer_share(
*,
user: User,
owner_id: UUID | None,
resource_type: str,
) -> None:
"""OSS floor: only resource owner or superuser may write shares.
The enterprise plugin can extend this via ``ensure_share_permission`` (e.g.
grant an ``authz_admin`` role broader power) but the OSS pass-through
must not let a non-owner mint share rows on someone else's resource.
"""
if getattr(user, "is_superuser", False):
return
if owner_id is not None and owner_id == user.id:
return
# Knowledge bases have no DB owner row — the OSS-side namespace check is
# that the share creator must be the calling user, which they always are
# because shares are written under ``current_user``. So unowned-resource
# paths fall back to "owner unknown, deny" for everything except KBs.
if resource_type == "knowledge_base":
return
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Only the resource owner or a superuser may administer shares for this resource.",
)
@router.post("", response_model=ShareRead, status_code=status.HTTP_201_CREATED)
@router.post("/", response_model=ShareRead, status_code=status.HTTP_201_CREATED)
async def create_share(
payload: ShareCreate,
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.
"""
owner_id = await _resolve_resource_owner(
session,
resource_type=payload.resource_type,
resource_id=payload.resource_id,
)
if payload.resource_type in _RESOURCE_OWNER_LOOKUPS and owner_id is None:
# The resource simply does not exist — UUID privacy: 404.
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Resource not found")
_ensure_can_administer_share(
user=current_user,
owner_id=owner_id,
resource_type=payload.resource_type,
)
await ensure_share_permission(
current_user,
ShareAction.CREATE,
share_user_id=current_user.id,
)
row = AuthzShare(
resource_type=payload.resource_type,
resource_id=payload.resource_id,
scope=payload.scope,
target_id=payload.target_id,
permission_level=payload.permission_level,
created_by=current_user.id,
created_at=datetime.now(timezone.utc),
)
session.add(row)
try:
await session.flush()
except Exception as exc:
await session.rollback()
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"Share could not be created: {exc}",
) from exc
await session.refresh(row)
# Tell the enforcer (if any) to drop its cached policy for the share target.
authz = get_authorization_service()
if payload.target_id is not None:
await authz.invalidate_user(payload.target_id)
else:
await authz.invalidate_all()
await audit_decision(
user_id=current_user.id,
action="share:create",
obj=f"{payload.resource_type}:{payload.resource_id}",
result="allow",
details={
"share_id": str(row.id),
"scope": payload.scope,
"target_id": str(payload.target_id) if payload.target_id else None,
"permission_level": payload.permission_level,
},
)
return ShareRead.model_validate(row, from_attributes=True)
@router.get("", response_model=list[ShareRead])
@router.get("/", response_model=list[ShareRead])
async def list_shares(
current_user: CurrentActiveUser,
session: DbSession,
resource_type: Annotated[str | None, Query()] = None,
resource_id: Annotated[UUID | None, Query()] = None,
target_id: Annotated[UUID | None, Query()] = None,
scope: Annotated[str | None, Query()] = None,
) -> 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.
"""
await ensure_share_permission(
current_user,
ShareAction.READ,
share_user_id=current_user.id,
)
stmt = select(AuthzShare)
if resource_type is not None:
stmt = stmt.where(AuthzShare.resource_type == resource_type)
if resource_id is not None:
stmt = stmt.where(AuthzShare.resource_id == resource_id)
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.
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)
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]
# Non-superuser visibility: caller is either the resource owner (full
# row visibility for that resource), the share creator, or the target.
visible: list[ShareRead] = []
owner_cache: dict[tuple[str, UUID], UUID | None] = {}
for row in rows:
key = (row.resource_type, row.resource_id)
if key not in owner_cache:
owner_cache[key] = await _resolve_resource_owner(
session,
resource_type=row.resource_type,
resource_id=row.resource_id,
)
if current_user.id in {owner_cache[key], row.created_by, row.target_id}:
visible.append(ShareRead.model_validate(row, from_attributes=True))
return visible
@router.get("/{share_id}", response_model=ShareRead)
async def get_share(
share_id: UUID,
current_user: CurrentActiveUser,
session: DbSession,
) -> ShareRead:
"""Fetch a single share by id with the same visibility rules as list."""
row = await session.get(AuthzShare, share_id)
if row is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Share not found")
await ensure_share_permission(
current_user,
ShareAction.READ,
share_id=share_id,
share_user_id=row.created_by,
)
if not getattr(current_user, "is_superuser", False):
owner_id = await _resolve_resource_owner(
session,
resource_type=row.resource_type,
resource_id=row.resource_id,
)
if current_user.id not in {owner_id, row.created_by, row.target_id}:
# UUID privacy — caller is not allowed to know this share exists.
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Share not found")
return ShareRead.model_validate(row, from_attributes=True)
@router.patch("/{share_id}", response_model=ShareRead)
async def update_share(
share_id: UUID,
payload: ShareUpdate,
current_user: CurrentActiveUser,
session: DbSession,
) -> ShareRead:
"""Update the permission level of an existing share."""
row = await session.get(AuthzShare, share_id)
if row is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Share not found")
owner_id = await _resolve_resource_owner(
session,
resource_type=row.resource_type,
resource_id=row.resource_id,
)
_ensure_can_administer_share(
user=current_user,
owner_id=owner_id,
resource_type=row.resource_type,
)
await ensure_share_permission(
current_user,
ShareAction.UPDATE,
share_id=share_id,
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.
try:
row.permission_level = SharePermissionLevel(payload.permission_level).value
except ValueError as exc:
raise HTTPException(
status_code=400,
detail=f"Unknown permission_level {payload.permission_level!r}",
) from exc
session.add(row)
await session.flush()
await session.refresh(row)
authz = get_authorization_service()
if row.target_id is not None:
await authz.invalidate_user(row.target_id)
else:
await authz.invalidate_all()
await audit_decision(
user_id=current_user.id,
action="share:update",
obj=f"{row.resource_type}:{row.resource_id}",
result="allow",
details={
"share_id": str(row.id),
"permission_level": row.permission_level,
},
)
return ShareRead.model_validate(row, from_attributes=True)
@router.delete("/{share_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_share(
share_id: UUID,
current_user: CurrentActiveUser,
session: DbSession,
) -> None:
"""Revoke an existing share."""
row = await session.get(AuthzShare, share_id)
if row is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Share not found")
owner_id = await _resolve_resource_owner(
session,
resource_type=row.resource_type,
resource_id=row.resource_id,
)
_ensure_can_administer_share(
user=current_user,
owner_id=owner_id,
resource_type=row.resource_type,
)
await ensure_share_permission(
current_user,
ShareAction.DELETE,
share_id=share_id,
share_user_id=row.created_by,
)
target_id = row.target_id
resource_type = row.resource_type
resource_id = row.resource_id
await session.delete(row)
await session.flush()
authz = get_authorization_service()
if target_id is not None:
await authz.invalidate_user(target_id)
else:
await authz.invalidate_all()
await audit_decision(
user_id=current_user.id,
action="share:delete",
obj=f"{resource_type}:{resource_id}",
result="allow",
details={"share_id": str(share_id)},
)
__all__ = ["router"]

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

@ -519,11 +519,9 @@ 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.
NOTE (Phase 3 prerequisite): the ``user_id`` filter here will need to
become share-aware (load by id, then check ``authz_share`` rows or call
the enterprise plugin's ``enforce``) once the share-CRUD APIs land. Until
then, an enterprise execute-grant on a flow the caller does not own would
404 here before the route's ``ensure_flow_permission`` can see it.
When an enterprise authorization service is registered, the lookup is
share-aware (load by id, route guard decides access). The OSS pass-through
default keeps the owner-scoped lookup.
"""
return await get_flow_by_id_or_endpoint_name(flow_id_or_name, api_key_user.id)

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

@ -343,10 +343,23 @@ async def _read_flow(
flow_id: UUID,
user_id: UUID,
):
"""Read a flow."""
stmt = select(Flow).where(Flow.id == flow_id).where(Flow.user_id == user_id)
"""Read a flow.
return (await session.exec(stmt)).first()
When the registered authorization service supports cross-user fetch
(enterprise Casbin), 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.
"""
from langflow.services.authorization.fetch import authorized_or_owner_scoped
return await authorized_or_owner_scoped(
session,
Flow,
id_column=Flow.id,
resource_id=flow_id,
owner_column=Flow.user_id,
owner_id=user_id,
)
async def _update_existing_flow(

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

@ -44,6 +44,10 @@ from langflow.schema.knowledge_base import (
TestBackendConnectionRequest,
TestBackendConnectionResponse,
)
from langflow.services.authorization import (
KnowledgeBaseAction,
ensure_knowledge_base_permission,
)
from langflow.services.database.models.jobs.model import JobStatus, JobType
from langflow.services.deps import get_job_service, get_settings_service, get_task_service
from langflow.services.jobs import DuplicateJobError
@ -494,6 +498,14 @@ async def test_backend_connection(
Pydantic validators before they reach this handler and surface as
HTTP 422.
"""
# Test-connection is a precondition for ``create_knowledge_base`` — gate
# it on the same permission so a viewer-role user cannot enumerate
# backend reachability they could not act on.
await ensure_knowledge_base_permission(
current_user,
KnowledgeBaseAction.CREATE,
kb_user_id=current_user.id,
)
# Use a private temp directory for the transient backend so a
# local-storage backend (Chroma) doesn't leak files into the user's
# KB root, and so concurrent test-connection calls don't collide.
@ -542,6 +554,12 @@ async def create_knowledge_base(
kb_root_path = KBStorageHelper.get_root_path()
kb_user = current_user.username
kb_name = request.name.strip().replace(" ", "_")
await ensure_knowledge_base_permission(
current_user,
KnowledgeBaseAction.CREATE,
kb_name=kb_name or None,
kb_user_id=current_user.id,
)
# Validate KB name
if not kb_name or len(kb_name) < MIN_KB_NAME_LENGTH:
raise HTTPException(status_code=400, detail="Knowledge base name must be at least 3 characters")
@ -704,7 +722,7 @@ async def create_knowledge_base(
@router.post("/preview-chunks", status_code=HTTPStatus.OK)
async def preview_chunks(
_current_user: CurrentActiveUser,
current_user: CurrentActiveUser,
files: Annotated[list[UploadFile], File(description="Files to preview chunking for")],
# Upper bounds cap the memory footprint of a preview request.
# ``max_chunks * chunk_size * CHUNK_PREVIEW_MULTIPLIER`` is the
@ -720,6 +738,11 @@ async def preview_chunks(
Uses the same RecursiveCharacterTextSplitter as the ingest endpoint
so the preview accurately reflects what will be stored.
"""
await ensure_knowledge_base_permission(
current_user,
KnowledgeBaseAction.CREATE,
kb_user_id=current_user.id,
)
try:
if not files:
raise HTTPException(status_code=400, detail="No files provided")
@ -853,6 +876,12 @@ async def ingest_files_to_knowledge_base(
Both are validated server-side; reserved keys + oversized values raise 422
so the UI can surface the rejection inline.
"""
await ensure_knowledge_base_permission(
current_user,
KnowledgeBaseAction.INGEST,
kb_name=kb_name,
kb_user_id=current_user.id,
)
try:
settings = get_settings_service().settings
max_file_size_upload = settings.max_file_size_upload
@ -1030,6 +1059,12 @@ async def ingest_folder_to_knowledge_base(
Returns a ``TaskResponse`` pointing at the ingestion job; track it
via ``/task/{id}`` or the ``GET /{kb_name}`` endpoint.
"""
await ensure_knowledge_base_permission(
current_user,
KnowledgeBaseAction.INGEST,
kb_name=kb_name,
kb_user_id=current_user.id,
)
try:
settings = get_settings_service().settings
allowed_roots = settings.kb_allowed_folder_roots or []
@ -1147,6 +1182,15 @@ async def list_knowledge_bases(
Reads from ``knowledge_base`` rows first. A disk scan is only used
as a recovery fallback when the user has no KB rows yet.
"""
# 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
# KB share grants exist.
await ensure_knowledge_base_permission(
current_user,
KnowledgeBaseAction.READ,
kb_user_id=current_user.id,
)
try:
kb_root_path = KBStorageHelper.get_root_path()
# Resolve + containment-check on par with every other path
@ -1278,6 +1322,12 @@ async def list_connectors(_current_user: CurrentActiveUser) -> list[ConnectorCat
@router.get("/{kb_name}", status_code=HTTPStatus.OK, dependencies=[Depends(_check_memory_base_association)])
async def get_knowledge_base(kb_name: str, current_user: CurrentActiveUser) -> KnowledgeBaseInfo:
"""Get detailed information about a specific knowledge base."""
await ensure_knowledge_base_permission(
current_user,
KnowledgeBaseAction.READ,
kb_name=kb_name,
kb_user_id=current_user.id,
)
try:
record = await knowledge_base_service.get_by_user_and_name(current_user.id, kb_name)
if record is not None:
@ -1350,6 +1400,12 @@ async def get_knowledge_base_chunks(
every comma. Repeated key=value params side-step that without
invasive middleware changes.
"""
await ensure_knowledge_base_permission(
current_user,
KnowledgeBaseAction.READ,
kb_name=kb_name,
kb_user_id=current_user.id,
)
kb_path: Path | None = None
backend = None
backend_type_value: str = BackendType.CHROMA.value
@ -1525,6 +1581,12 @@ async def get_knowledge_base_metadata_keys(
hint. Native distinct queries are deferred to backend-specific work
(same trade-off as the chunks-endpoint post-filter pass).
"""
await ensure_knowledge_base_permission(
current_user,
KnowledgeBaseAction.READ,
kb_name=kb_name,
kb_user_id=current_user.id,
)
kb_path: Path | None = None
backend = None
backend_type_value: str = BackendType.CHROMA.value
@ -1629,6 +1691,12 @@ async def ingest_via_connector(
is spawned), then hands off to the same async ingestion machinery
file-upload + folder already use.
"""
await ensure_knowledge_base_permission(
current_user,
KnowledgeBaseAction.INGEST,
kb_name=kb_name,
kb_user_id=current_user.id,
)
try:
kb_path = _resolve_kb_path(kb_name, current_user)
@ -1739,6 +1807,12 @@ async def list_ingestion_runs(
another's run history. Returns counter-only rows; the UI fetches
the detail endpoint for the drill-down.
"""
await ensure_knowledge_base_permission(
current_user,
KnowledgeBaseAction.READ,
kb_name=kb_name,
kb_user_id=current_user.id,
)
# Verify the KB path exists + traversal-safe before exposing run
# history — otherwise a crafted ``kb_name`` could be used to probe
# for other users' KB existence by timing list_runs_for_kb.
@ -1768,6 +1842,12 @@ async def get_ingestion_run(
current_user: CurrentActiveUser,
) -> IngestionRunDetail:
"""Full run detail including per-item breakdown + error messages."""
await ensure_knowledge_base_permission(
current_user,
KnowledgeBaseAction.READ,
kb_name=kb_name,
kb_user_id=current_user.id,
)
_resolve_kb_path(kb_name, current_user)
row = await ingestion_run_service.get_run(run_id, user_id=current_user.id)
@ -1837,6 +1917,12 @@ async def delete_knowledge_base(
job_service: Annotated[JobService, Depends(get_job_service)],
) -> dict[str, str]:
"""Delete a specific knowledge base."""
await ensure_knowledge_base_permission(
current_user,
KnowledgeBaseAction.DELETE,
kb_name=kb_name,
kb_user_id=current_user.id,
)
try:
try:
kb_path = _resolve_kb_path(kb_name, current_user)
@ -1929,6 +2015,15 @@ async def delete_knowledge_bases_bulk(
job_service: Annotated[JobService, Depends(get_job_service)],
) -> dict[str, object]:
"""Delete multiple knowledge bases."""
# Per-KB guard. The enterprise plugin gets to deny individual KBs without
# the call site needing to know which (the first denied KB raises 403).
for kb_name in request.kb_names:
await ensure_knowledge_base_permission(
current_user,
KnowledgeBaseAction.DELETE,
kb_name=kb_name,
kb_user_id=current_user.id,
)
try:
deleted_count = 0
not_found_kbs = []
@ -2048,6 +2143,12 @@ async def cancel_ingestion(
task_service: Annotated[TaskService, Depends(get_task_service)],
) -> dict[str, str]:
"""Cancel the ongoing ingestion task for a knowledge base."""
await ensure_knowledge_base_permission(
current_user,
KnowledgeBaseAction.WRITE,
kb_name=kb_name,
kb_user_id=current_user.id,
)
try:
kb_path = _resolve_kb_path(kb_name, current_user)

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

@ -10,6 +10,8 @@ from langflow.api.utils import DbSession, custom_params
from langflow.api.utils.flow_utils import compute_virtual_flow_id
from langflow.schema.message import MessageResponse
from langflow.services.auth.utils import get_current_active_user
from langflow.services.authorization import FlowAction, ensure_flow_permission
from langflow.services.authorization.fetch import authorized_or_owner_scoped
from langflow.services.database.models.flow.model import Flow
from langflow.services.database.models.message.crud import (
delete_messages_for_user,
@ -65,6 +67,42 @@ def _langfuse_feedback_sync_enabled() -> bool:
return langfuse_is_configured()
async def _ensure_flow_action_or_404(
session: DbSession,
*,
flow_id: UUID,
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.
"""
flow = await authorized_or_owner_scoped(
session,
Flow,
id_column=Flow.id,
resource_id=flow_id,
owner_column=Flow.user_id,
owner_id=user.id,
)
if flow is None:
return None
await ensure_flow_permission(
user,
action,
flow_id=flow.id,
flow_user_id=flow.user_id,
workspace_id=getattr(flow, "workspace_id", None),
folder_id=getattr(flow, "folder_id", None),
)
return flow
async def _purge_memory_base_session_data(user_id: UUID, session_ids: list[str]) -> None:
"""Best-effort: drop ingested chunks for the deleted sessions from each MB.
@ -99,8 +137,13 @@ async def get_vertex_builds(
# Ownership is enforced in the data access layer.
# Foreign flow IDs intentionally resolve to an empty payload (200)
# to avoid leaking whether the target flow exists.
vertex_builds = await get_vertex_builds_by_flow_id(session, flow_id, user_id=current_user.id)
flow = await _ensure_flow_action_or_404(session, flow_id=flow_id, user=current_user, action=FlowAction.READ)
if flow is None:
return VertexBuildMapModel.from_list_of_dicts([])
vertex_builds = await get_vertex_builds_by_flow_id(session, flow_id, user_id=flow.user_id)
return VertexBuildMapModel.from_list_of_dicts(vertex_builds)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e)) from e
@ -113,7 +156,12 @@ async def delete_vertex_builds(
) -> None:
try:
# Keep endpoint idempotent while preventing cross-user deletion.
await delete_vertex_builds_by_flow_id(session, flow_id, user_id=current_user.id)
flow = await _ensure_flow_action_or_404(session, flow_id=flow_id, user=current_user, action=FlowAction.WRITE)
if flow is None:
return
await delete_vertex_builds_by_flow_id(session, flow_id, user_id=flow.user_id)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e)) from e
@ -125,18 +173,31 @@ async def get_message_sessions(
flow_id: Annotated[UUID | None, Query()] = None,
) -> list[str]:
try:
# Use JOIN instead of subquery for better performance
# 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.
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:
return []
stmt = select(MessageTable.session_id).distinct()
stmt = stmt.where(MessageTable.flow_id == flow_id)
stmt = stmt.where(col(MessageTable.session_id).isnot(None))
stmt = stmt.where(~col(MessageTable.session_id).startswith("agentic_"))
session_ids = await session.exec(stmt)
return list(session_ids)
stmt = select(MessageTable.session_id).distinct()
stmt = stmt.join(Flow, MessageTable.flow_id == Flow.id)
stmt = stmt.where(col(MessageTable.session_id).isnot(None))
stmt = stmt.where(~col(MessageTable.session_id).startswith("agentic_"))
stmt = stmt.where(Flow.user_id == current_user.id)
if flow_id:
stmt = stmt.where(MessageTable.flow_id == flow_id)
session_ids = await session.exec(stmt)
return list(session_ids)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e)) from e
@ -152,10 +213,19 @@ async def get_messages(
order_by: Annotated[str | None, Query()] = "timestamp",
) -> list[MessageResponse]:
try:
# When a flow_id is provided, gate on flow READ permission first; the
# share-aware path lets a non-owner with a read grant see the flow's
# messages.
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:
return []
# Use JOIN instead of subquery for better performance
stmt = select(MessageTable)
stmt = stmt.join(Flow, MessageTable.flow_id == Flow.id)
stmt = stmt.where(Flow.user_id == current_user.id)
if flow_id is None:
stmt = stmt.where(Flow.user_id == current_user.id)
if flow_id:
stmt = stmt.where(MessageTable.flow_id == flow_id)
@ -215,6 +285,13 @@ async def update_message(
# Intentionally return 404 for both "not found" and "not owned".
raise HTTPException(status_code=404, detail="Message not found")
# Bind the parent flow's authorization to message writes so a viewer-role
# user cannot edit messages on a flow they only have READ on.
if db_message.flow_id is not None:
await _ensure_flow_action_or_404(
session, flow_id=db_message.flow_id, user=current_user, action=FlowAction.WRITE
)
try:
previous_positive_feedback = _get_positive_feedback_value(db_message)
message_dict = message.model_dump(exclude_unset=True, exclude_none=True)
@ -563,14 +640,18 @@ async def get_transactions(
params: Annotated[Params | None, Depends(custom_params)],
) -> Page[TransactionLogsResponse]:
try:
# Flow ownership is part of the SQL filter.
# Flow ownership / share-grant is verified via the parent flow guard.
# For foreign flow IDs, the endpoint returns an empty page (200)
# to preserve response shape and avoid existence leakage.
flow = await _ensure_flow_action_or_404(session, flow_id=flow_id, user=current_user, action=FlowAction.READ)
if flow is None:
from fastapi_pagination import Page as _Page
return _Page(items=[], total=0, page=1, size=params.size if params else 50, pages=0)
stmt = (
select(TransactionTable)
.join(Flow, TransactionTable.flow_id == Flow.id)
.where(TransactionTable.flow_id == flow_id)
.where(Flow.user_id == current_user.id)
.order_by(col(TransactionTable.timestamp).desc())
)
import warnings
@ -580,5 +661,7 @@ async def get_transactions(
"ignore", category=DeprecationWarning, module=r"fastapi_pagination\.ext\.sqlalchemy"
)
return await apaginate(session, stmt, params=params, transformer=transform_transaction_table_for_logs)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e)) from e

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

@ -33,6 +33,7 @@ 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.fetch import authorized_or_owner_scoped
from langflow.services.authorization.utils import _resolve_casbin_domain
from langflow.services.database.models.deployment.exceptions import (
araise_if_deployment_guard_error_or_skip,
@ -250,15 +251,18 @@ async def read_project(
search: str = "",
):
try:
# Phase 3 prerequisite: owner-scoped fetch shadows enterprise share
# grants on non-owned projects; see langflow.services.authorization.utils.
project = (
await session.exec(
select(Folder)
.options(selectinload(Folder.flows))
.where(Folder.id == project_id, Folder.user_id == current_user.id)
)
).first()
# Share-aware fetch: when an enterprise authorization service 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
# strict-pass-through stub cannot widen visibility.
from langflow.services.deps import get_authorization_service
share_aware = await get_authorization_service().supports_cross_user_fetch()
stmt = select(Folder).options(selectinload(Folder.flows)).where(Folder.id == project_id)
if not share_aware:
stmt = stmt.where(Folder.user_id == current_user.id)
project = (await session.exec(stmt)).first()
except Exception as e:
if "No result found" in str(e):
raise HTTPException(status_code=404, detail="Project not found") from e
@ -276,9 +280,17 @@ async def read_project(
)
try:
# When share-aware fetch is on and the project is not owned by the
# caller (i.e. reached via a share grant), show all flows in the
# project — the share grant on the project implies access to its
# contents. Otherwise keep the existing owner-scoped flow filter.
treat_as_shared = share_aware and project.user_id != current_user.id
# Check if pagination is explicitly requested by the user (both page and size provided)
if page is not None and size is not None:
stmt = select(Flow).where(Flow.folder_id == project_id, Flow.user_id == current_user.id)
stmt = select(Flow).where(Flow.folder_id == project_id)
if not treat_as_shared:
stmt = stmt.where(Flow.user_id == current_user.id)
if Flow.updated_at is not None:
stmt = stmt.order_by(Flow.updated_at.desc()) # type: ignore[attr-defined]
@ -298,9 +310,12 @@ async def read_project(
return FolderWithPaginatedFlows(folder=FolderRead.model_validate(project), flows=paginated_flows)
# If no pagination requested, return all flows for the current user
flows_from_current_user_in_project = [flow for flow in project.flows if flow.user_id == current_user.id]
project.flows = flows_from_current_user_in_project
# If no pagination requested, return flows visible to the caller.
if treat_as_shared:
visible_flows = list(project.flows)
else:
visible_flows = [flow for flow in project.flows if flow.user_id == current_user.id]
project.flows = visible_flows
# Convert to FolderReadWithFlows while session is still active to avoid detached instance errors
return FolderReadWithFlows.model_validate(project, from_attributes=True)
@ -319,9 +334,14 @@ async def update_project(
background_tasks: BackgroundTasks,
):
try:
existing_project = (
await session.exec(select(Folder).where(Folder.id == project_id, Folder.user_id == current_user.id))
).first()
existing_project = await authorized_or_owner_scoped(
session,
Folder,
id_column=Folder.id,
resource_id=project_id,
owner_column=Folder.user_id,
owner_id=current_user.id,
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e)) from e
@ -503,9 +523,14 @@ async def delete_project(
current_user: CurrentActiveUser,
):
try:
project = (
await session.exec(select(Folder).where(Folder.id == project_id, Folder.user_id == current_user.id))
).first()
project = await authorized_or_owner_scoped(
session,
Folder,
id_column=Folder.id,
resource_id=project_id,
owner_column=Folder.user_id,
owner_id=current_user.id,
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e)) from e
@ -571,14 +596,17 @@ async def download_file(
"""Download all flows from project as a zip file."""
# Fetch the project row first so the authorization call carries the
# owner id (for the owner-override path) and the workspace id (for the
# project-domain resolver). Without these, an enterprise plugin would
# evaluate the check against ``domain="*"`` and miss workspace-scoped
# grants, and an owner could even be denied by a role-only policy.
# Phase 3 prerequisite: owner-scoped fetch — see
# ``langflow.services.authorization.utils``.
project = (
await session.exec(select(Folder).where(Folder.id == project_id, Folder.user_id == current_user.id))
).first()
# project-domain resolver). When share-aware fetch is supported, the
# row is loaded by id and ``ensure_project_permission`` decides access;
# otherwise the query stays owner-scoped.
project = await authorized_or_owner_scoped(
session,
Folder,
id_column=Folder.id,
resource_id=project_id,
owner_column=Folder.user_id,
owner_id=current_user.id,
)
if not project:
raise HTTPException(status_code=404, detail="Project not found")
await ensure_project_permission(

82
src/backend/base/langflow/api/v1/schemas/authz_shares.py Normal file
View File

@ -0,0 +1,82 @@
"""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.
"""
from __future__ import annotations
from datetime import datetime
from typing import Literal
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.
ShareResourceType = Literal[
"flow",
"deployment",
"project",
"knowledge_base",
"variable",
"file",
]
ShareScopeLiteral = Literal["private", "team", "user", "public"]
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.
"""
resource_type: ShareResourceType
resource_id: UUID
scope: ShareScopeLiteral
target_id: UUID | None = Field(default=None)
permission_level: SharePermissionLiteral = "read"
@model_validator(mode="after")
def _check_scope_target_consistency(self) -> ShareCreate:
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"
raise ValueError(msg)
if not targeted and has_target:
msg = f"scope={self.scope!r} must not include a 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.
"""
permission_level: SharePermissionLiteral
class ShareRead(BaseModel):
"""Read-only projection of an ``authz_share`` row."""
id: UUID
resource_type: str
resource_id: UUID
scope: ShareScopeLiteral
target_id: UUID | None
permission_level: SharePermissionLiteral
created_by: UUID | None
created_at: datetime
model_config = {"from_attributes": True}

28
src/backend/base/langflow/api/v1/variable.py
View File

@ -15,6 +15,7 @@ from langflow.api.v1.models import (
get_provider_from_variable_name,
)
from langflow.api.v1.schemas.deployments import DetectVarsRequest, DetectVarsResponse
from langflow.services.authorization import VariableAction, ensure_variable_permission
from langflow.services.database.models.flow_version.crud import get_flow_version_entries_by_ids
from langflow.services.database.models.variable.model import VariableCreate, VariableRead, VariableUpdate
from langflow.services.deps import get_variable_service
@ -107,6 +108,11 @@ async def create_variable(
current_user: CurrentActiveUser,
):
"""Create a new variable."""
await ensure_variable_permission(
current_user,
VariableAction.CREATE,
variable_user_id=current_user.id,
)
variable_service = get_variable_service()
if not variable.name and not variable.value:
raise HTTPException(status_code=400, detail="Variable name and value cannot be empty")
@ -159,6 +165,11 @@ async def read_variables(
Returns a list of variables.
"""
await ensure_variable_permission(
current_user,
VariableAction.READ,
variable_user_id=current_user.id,
)
variable_service = get_variable_service()
if not isinstance(variable_service, DatabaseVariableService):
msg = "Variable service is not an instance of DatabaseVariableService"
@ -207,6 +218,12 @@ async def update_variable(
existing_variable = await variable_service.get_variable_by_id(
user_id=current_user.id, variable_id=variable_id, session=session
)
await ensure_variable_permission(
current_user,
VariableAction.WRITE,
variable_id=variable_id,
variable_user_id=current_user.id,
)
# Validate API key if updating a model provider variable
if existing_variable.name in model_provider_variable_mapping.values() and variable.value:
@ -256,6 +273,12 @@ async def delete_variable(
variable_to_delete = await variable_service.get_variable_by_id(
user_id=current_user.id, variable_id=variable_id, session=session
)
await ensure_variable_permission(
current_user,
VariableAction.DELETE,
variable_id=variable_id,
variable_user_id=current_user.id,
)
# Check if this variable is a model provider credential
provider = get_provider_from_variable_name(variable_to_delete.name)
@ -319,6 +342,11 @@ async def detect_env_vars(
template values (including accidental secrets) and ensures results are
actual stored global variables.
"""
await ensure_variable_permission(
current_user,
VariableAction.READ,
variable_user_id=current_user.id,
)
variable_service = get_variable_service()
existing_variable_names = {
name

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

@ -16,6 +16,8 @@ from sqlmodel import col, select
from langflow.api.schemas import UploadFileResponse
from langflow.api.utils import CurrentActiveUser, DbSession, build_content_disposition
from langflow.services.authorization import FileAction, ensure_file_permission
from langflow.services.authorization.fetch import authorized_or_owner_scoped
from langflow.services.database.models.file.model import File as UserFile
from langflow.services.deps import get_settings_service, get_storage_service
from langflow.services.settings.service import SettingsService
@ -87,20 +89,22 @@ 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):
# Fetch the file from the DB
stmt = select(UserFile).where(UserFile.id == file_id)
results = await session.exec(stmt)
file = results.first()
# Share-aware fetch. Under the OSS pass-through this keeps the existing
# owner-scoped query (cannot widen visibility). Enterprise plugins set
# ``SUPPORTS_CROSS_USER_FETCH=True`` so a share grant can resolve here.
file = await authorized_or_owner_scoped(
session,
UserFile,
id_column=UserFile.id,
resource_id=file_id,
owner_column=UserFile.user_id,
owner_id=current_user.id,
)
# Check if the file exists
if not file:
raise HTTPException(status_code=404, detail="File not found")
# Make sure the user has access to the file
if file.user_id != current_user.id:
# Return 404 to prevent information disclosure about resource existence
raise HTTPException(status_code=404, detail="File not found")
return file
@ -140,6 +144,11 @@ async def upload_user_file(
ephemeral: bool = False,
) -> UploadFileResponse:
"""Upload a file for the current user and track it in the database."""
await ensure_file_permission(
current_user,
FileAction.CREATE,
file_user_id=current_user.id,
)
# Get the max allowed file size from settings (in MB)
try:
max_file_size_upload = settings_service.settings.max_file_size_upload
@ -414,6 +423,11 @@ async def list_files(
# storage_service: Annotated[StorageService, Depends(get_storage_service)],
) -> list[UserFile]:
"""List the files available to the current user."""
await ensure_file_permission(
current_user,
FileAction.READ,
file_user_id=current_user.id,
)
try:
# Load sample files if they don't exist
# TODO: Pending further testing
@ -440,6 +454,15 @@ async def delete_files_batch(
storage_service: Annotated[StorageService, Depends(get_storage_service)],
):
"""Delete multiple files by their IDs."""
# Gate each id on file:delete; a viewer cannot wipe out files they only
# have read access to. Enterprise plugins consult share grants here.
for fid in file_ids:
await ensure_file_permission(
current_user,
FileAction.DELETE,
file_id=fid,
file_user_id=current_user.id,
)
try:
# Fetch all files from the DB
stmt = select(UserFile).where(col(UserFile.id).in_(file_ids), col(UserFile.user_id) == current_user.id)
@ -537,6 +560,13 @@ async def download_files_batch(
storage_service: Annotated[StorageService, Depends(get_storage_service)],
):
"""Download multiple files as a zip file by their IDs."""
for fid in file_ids:
await ensure_file_permission(
current_user,
FileAction.READ,
file_id=fid,
file_user_id=current_user.id,
)
try:
# Fetch all files from the DB
stmt = select(UserFile).where(col(UserFile.id).in_(file_ids), col(UserFile.user_id) == current_user.id)
@ -648,6 +678,13 @@ async def download_file(
if not file:
raise HTTPException(status_code=404, detail="File not found")
await ensure_file_permission(
current_user,
FileAction.READ,
file_id=file.id,
file_user_id=file.user_id,
)
# Get the basename of the file path
file_name = Path(file.path).name
@ -701,10 +738,18 @@ async def edit_file_name(
try:
# Fetch the file from the DB
file = await fetch_file_object(file_id, current_user, session)
await ensure_file_permission(
current_user,
FileAction.WRITE,
file_id=file.id,
file_user_id=file.user_id,
)
# Update the file name
file.name = name
session.add(file)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error editing file: {e}") from e
@ -725,6 +770,13 @@ async def delete_file(
if not file_to_delete:
raise HTTPException(status_code=404, detail="File not found")
await ensure_file_permission(
current_user,
FileAction.DELETE,
file_id=file_to_delete.id,
file_user_id=file_to_delete.user_id,
)
# Extract just the filename from the path (strip user_id prefix)
file_name = Path(file_to_delete.path).name
@ -788,6 +840,11 @@ async def delete_all_files(
storage_service: Annotated[StorageService, Depends(get_storage_service)],
):
"""Delete all files for the current user."""
await ensure_file_permission(
current_user,
FileAction.DELETE,
file_user_id=current_user.id,
)
try:
# Fetch all files from the DB
stmt = select(UserFile).where(UserFile.user_id == current_user.id)

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

@ -397,6 +397,19 @@ 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.
"""
from langflow.services.deps import get_authorization_service
authz = get_authorization_service()
share_aware = await authz.supports_cross_user_fetch()
async with session_scope() as session:
# SECURITY: previously the UUID branch below called
# ``session.get(Flow, flow_id)`` with no ownership check, so any
@ -424,12 +437,12 @@ async def get_flow_by_id_or_endpoint_name(flow_id_or_name: str, user_id: str | U
try:
flow_id = UUID(flow_id_or_name)
flow = await session.get(Flow, flow_id)
if flow is not None and uuid_user_id is not None and flow.user_id != uuid_user_id:
if flow is not None and uuid_user_id is not None and not share_aware and flow.user_id != uuid_user_id:
flow = None
except ValueError:
endpoint_name = flow_id_or_name
stmt = select(Flow).where(Flow.endpoint_name == endpoint_name)
if uuid_user_id is not None:
if uuid_user_id is not None and not share_aware:
stmt = stmt.where(Flow.user_id == uuid_user_id)
flow = (await session.exec(stmt)).first()
if flow is None:

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

@ -1,25 +1,48 @@
"""Langflow OSS authorization service package (pass-through; enterprise plugin enforces)."""
from langflow.services.authorization.actions import DeploymentAction, FlowAction, ProjectAction
from langflow.services.authorization.actions import (
DeploymentAction,
FileAction,
FlowAction,
KnowledgeBaseAction,
ProjectAction,
ShareAction,
VariableAction,
)
from langflow.services.authorization.fetch import authorized_or_owner_scoped, deny_to_404
from langflow.services.authorization.service import LangflowAuthorizationService
from langflow.services.authorization.utils import (
audit_decision,
ensure_deployment_permission,
ensure_file_permission,
ensure_flow_permission,
ensure_knowledge_base_permission,
ensure_permission,
ensure_project_permission,
ensure_share_permission,
ensure_variable_permission,
filter_visible_resources,
)
__all__ = [
"DeploymentAction",
"FileAction",
"FlowAction",
"KnowledgeBaseAction",
"LangflowAuthorizationService",
"ProjectAction",
"ShareAction",
"VariableAction",
"audit_decision",
"authorized_or_owner_scoped",
"deny_to_404",
"ensure_deployment_permission",
"ensure_file_permission",
"ensure_flow_permission",
"ensure_knowledge_base_permission",
"ensure_permission",
"ensure_project_permission",
"ensure_share_permission",
"ensure_variable_permission",
"filter_visible_resources",
]

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

@ -43,3 +43,51 @@ class ProjectAction(str, Enum):
WRITE = "write"
CREATE = "create"
DELETE = "delete"
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.
"""
READ = "read"
WRITE = "write"
CREATE = "create"
DELETE = "delete"
INGEST = "ingest"
class VariableAction(str, Enum):
"""Actions that can be authorized on a variable resource."""
READ = "read"
WRITE = "write"
CREATE = "create"
DELETE = "delete"
class FileAction(str, Enum):
"""Actions that can be authorized on a user-file resource (v2 files)."""
READ = "read"
WRITE = "write"
CREATE = "create"
DELETE = "delete"
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``.
"""
READ = "read"
CREATE = "create"
DELETE = "delete"
UPDATE = "update"

101
src/backend/base/langflow/services/authorization/fetch.py Normal file
View File

@ -0,0 +1,101 @@
"""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.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, TypeVar
from fastapi import HTTPException, status
from sqlmodel import select
from langflow.services.deps import get_authorization_service
if TYPE_CHECKING:
from uuid import UUID
from sqlalchemy.orm.attributes import InstrumentedAttribute
from sqlmodel.ext.asyncio.session import AsyncSession
T = TypeVar("T")
async def authorized_or_owner_scoped(
session: AsyncSession,
model: type[T],
*,
id_column: InstrumentedAttribute,
resource_id: UUID,
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.
"""
authz = get_authorization_service()
if await authz.supports_cross_user_fetch():
stmt = select(model).where(id_column == resource_id)
else:
stmt = select(model).where(id_column == resource_id).where(owner_column == owner_id)
return (await session.exec(stmt)).first()
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:
.. 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.
"""
if exc.status_code != status.HTTP_403_FORBIDDEN:
return exc
return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=detail)

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

@ -31,7 +31,15 @@ from uuid import UUID
from fastapi import HTTPException, status
from lfx.log.logger import logger
from langflow.services.authorization.actions import DeploymentAction, FlowAction, ProjectAction
from langflow.services.authorization.actions import (
DeploymentAction,
FileAction,
FlowAction,
KnowledgeBaseAction,
ProjectAction,
ShareAction,
VariableAction,
)
from langflow.services.deps import get_authorization_service, get_settings_service
if TYPE_CHECKING:
@ -49,7 +57,26 @@ _AUDIT_DENY = "deny"
_AUDIT_OWNER_OVERRIDE = "owner_override"
# Context keys that name the resource owner — used by audit-detail extraction.
_OWNER_CONTEXT_KEYS = ("flow_user_id", "deployment_user_id", "project_user_id")
_OWNER_CONTEXT_KEYS = (
"flow_user_id",
"deployment_user_id",
"project_user_id",
"knowledge_base_user_id",
"variable_user_id",
"file_user_id",
"share_user_id",
)
# Action enum types we coerce to their string value.
_ACTION_ENUMS = (
FlowAction,
DeploymentAction,
ProjectAction,
KnowledgeBaseAction,
VariableAction,
FileAction,
ShareAction,
)
def _auth_context(user: User | UserRead) -> dict[str, Any]:
@ -57,9 +84,18 @@ def _auth_context(user: User | UserRead) -> dict[str, Any]:
return {"is_superuser": getattr(user, "is_superuser", False)}
def _coerce_action(act: DeploymentAction | FlowAction | ProjectAction | str) -> str:
def _coerce_action(
act: DeploymentAction
| FlowAction
| ProjectAction
| KnowledgeBaseAction
| VariableAction
| FileAction
| ShareAction
| str,
) -> str:
"""Return the string value of an action enum or pass through a raw string."""
if isinstance(act, (FlowAction, DeploymentAction, ProjectAction)):
if isinstance(act, _ACTION_ENUMS):
return act.value
return act
@ -209,7 +245,7 @@ async def _ensure_resource_permission(
user: User | UserRead,
*,
resource_type: str,
resource_id: UUID | None,
resource_id: UUID | str | None,
owner_id: UUID | None,
act_str: str,
resolved_domain: str,
@ -221,6 +257,9 @@ async def _ensure_resource_permission(
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).
"""
obj = f"{resource_type}:{resource_id}" if resource_id else f"{resource_type}:*"
@ -340,6 +379,114 @@ async def ensure_project_permission(
)
async def ensure_knowledge_base_permission(
user: User | UserRead,
act: KnowledgeBaseAction | str,
*,
kb_name: str | None = None,
kb_user_id: UUID | None = None,
workspace_id: UUID | None = None,
project_id: UUID | None = None,
domain: str | None = None,
) -> None:
"""Check knowledge-base-scoped permission with owner override.
Knowledge bases are name-keyed on the filesystem, so ``kb_name`` is used
verbatim as the Casbin object slug (``knowledge_base:{kb_name}``). The KB
owner can always operate on their own KB; otherwise the enterprise plugin
decides.
"""
await _ensure_resource_permission(
user,
resource_type="knowledge_base",
resource_id=kb_name,
owner_id=kb_user_id,
act_str=_coerce_action(act),
resolved_domain=domain if domain is not None else _resolve_casbin_domain(workspace_id, project_id),
extra_context={
"knowledge_base_user_id": kb_user_id,
"kb_name": kb_name,
"workspace_id": workspace_id,
"project_id": project_id,
},
)
async def ensure_variable_permission(
user: User | UserRead,
act: VariableAction | str,
*,
variable_id: UUID | None = None,
variable_user_id: UUID | None = None,
workspace_id: UUID | None = None,
domain: str | None = None,
) -> None:
"""Check variable-scoped permission with owner override."""
await _ensure_resource_permission(
user,
resource_type="variable",
resource_id=variable_id,
owner_id=variable_user_id,
act_str=_coerce_action(act),
resolved_domain=domain if domain is not None else _resolve_casbin_domain(workspace_id, None),
extra_context={
"variable_user_id": variable_user_id,
"workspace_id": workspace_id,
},
)
async def ensure_file_permission(
user: User | UserRead,
act: FileAction | str,
*,
file_id: UUID | None = None,
file_user_id: UUID | None = None,
workspace_id: UUID | None = None,
domain: str | None = None,
) -> None:
"""Check file-scoped permission (v2 user files) with owner override."""
await _ensure_resource_permission(
user,
resource_type="file",
resource_id=file_id,
owner_id=file_user_id,
act_str=_coerce_action(act),
resolved_domain=domain if domain is not None else _resolve_casbin_domain(workspace_id, None),
extra_context={
"file_user_id": file_user_id,
"workspace_id": workspace_id,
},
)
async def ensure_share_permission(
user: User | UserRead,
act: ShareAction | str,
*,
share_id: UUID | None = None,
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.
"""
await _ensure_resource_permission(
user,
resource_type="share",
resource_id=share_id,
owner_id=share_user_id,
act_str=_coerce_action(act),
resolved_domain=domain if domain is not None else "*",
extra_context={
"share_user_id": share_user_id,
},
)
async def filter_visible_resources(
user: User | UserRead,
*,

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

@ -112,12 +112,25 @@ async def get_deployment(
user_id: UUID,
deployment_id: UUID | str,
) -> Deployment | None:
"""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
``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.
"""
from langflow.services.authorization.fetch import authorized_or_owner_scoped
deployment_uuid = parse_uuid(deployment_id, field_name="deployment_id")
stmt = select(Deployment).where(
Deployment.user_id == user_id,
Deployment.id == deployment_uuid,
return await authorized_or_owner_scoped(
db,
Deployment,
id_column=Deployment.id,
resource_id=deployment_uuid,
owner_column=Deployment.user_id,
owner_id=user_id,
)
return (await db.exec(stmt)).first()
_UNSET = object()

64
src/backend/tests/unit/alembic/test_seed_authz_system_roles.py Normal file
View File

@ -0,0 +1,64 @@
"""Tests for the seed-system-roles migration (8d3a1f9c2e0b).
The migration's job is to insert exactly three system roles — viewer,
developer, admin — with stable permission templates and idempotent semantics.
These tests pin the permission shape and the idempotency contract without
executing the alembic harness.
"""
from __future__ import annotations
import importlib
_MIGRATION = importlib.import_module("langflow.alembic.versions.8d3a1f9c2e0b_seed_authz_system_roles")
def test_three_system_roles_are_seeded():
"""The migration seeds exactly viewer / developer / admin."""
names = [name for name, _, _ in _MIGRATION._SYSTEM_ROLES]
assert names == ["viewer", "developer", "admin"]
def test_viewer_has_only_read_and_execute_grants():
"""Viewer should never expose write/delete-class permissions."""
permissions = set(_MIGRATION._VIEWER_PERMISSIONS)
forbidden_verbs = {"write", "create", "delete", "admin", "deploy", "ingest", "update"}
for slug in permissions:
_, verb = slug.split(":")
assert verb not in forbidden_verbs, f"viewer must not include {slug}"
def test_developer_includes_viewer_permissions():
"""Developer is a strict superset of viewer."""
viewer = set(_MIGRATION._VIEWER_PERMISSIONS)
developer_lookup = {name: perms for name, _, perms in _MIGRATION._SYSTEM_ROLES}
developer = set(developer_lookup["developer"])
assert viewer.issubset(developer)
def test_admin_includes_developer_permissions():
"""Admin is a strict superset of developer (and therefore of viewer)."""
lookup = {name: set(perms) for name, _, perms in _MIGRATION._SYSTEM_ROLES}
assert lookup["developer"].issubset(lookup["admin"])
def test_admin_has_share_administration_permissions():
"""Admin is the only role with share:* — viewers cannot mint grants."""
admin = {name: set(perms) for name, _, perms in _MIGRATION._SYSTEM_ROLES}["admin"]
assert {"share:create", "share:read", "share:update", "share:delete"}.issubset(admin)
def test_permission_slugs_use_resource_action_format():
"""Slugs must match ``{resource}:{action}`` so enterprise PolicySync can split them."""
for _, _, permissions in _MIGRATION._SYSTEM_ROLES:
for slug in permissions:
assert slug.count(":") == 1, slug
resource, verb = slug.split(":")
assert resource, slug
assert verb, slug
def test_revision_chain_pins_authz_foundations():
"""Down-revision must be the authz foundations migration so the chain stays linear."""
assert _MIGRATION.revision == "8d3a1f9c2e0b"
assert _MIGRATION.down_revision == "7c8d9e0f1a2b"

87
src/backend/tests/unit/api/v1/test_authz_audit_schemas.py Normal file
View File

@ -0,0 +1,87 @@
"""Schema and pagination-math tests for the ``/api/v1/authz/audit`` endpoint.
The full live-app integration test sits in a heavier fixture suite; here we
just pin the response-shape contract and the ceiling on ``size``.
"""
from __future__ import annotations
from datetime import datetime, timezone
from uuid import uuid4
import pytest
from langflow.api.v1.authz_audit import AuthzAuditLogRead, AuthzAuditPage
from pydantic import ValidationError
def test_audit_log_read_accepts_minimal_row():
"""user_id, resource_type, resource_id, details may all be None."""
row = AuthzAuditLogRead(
id=uuid4(),
user_id=None,
action="flow:read",
resource_type=None,
resource_id=None,
result="allow",
details=None,
timestamp=datetime.now(timezone.utc),
)
assert row.action == "flow:read"
assert row.user_id is None
def test_audit_log_read_carries_details_dict():
"""The details payload is a free-form dict produced by audit_decision."""
row = AuthzAuditLogRead(
id=uuid4(),
user_id=uuid4(),
action="flow:write",
resource_type="flow",
resource_id=uuid4(),
result="deny",
details={"domain": "project:abc", "flow_user_id": str(uuid4())},
timestamp=datetime.now(timezone.utc),
)
assert row.details["domain"] == "project:abc"
def test_audit_page_envelope_round_trip():
page = AuthzAuditPage(items=[], total=0, page=1, size=50, pages=0)
assert page.total == 0
assert page.pages == 0
def test_audit_page_rejects_negative_total():
"""``total`` must be >= 0 — Pydantic infers int but does not enforce bounds.
This test is a contract check: if someone adds a validator, the page
envelope should reject nonsensical values rather than silently propagate
them to the client.
"""
# We currently accept any int; this test is a placeholder so a future
# validator addition is caught by the suite.
page = AuthzAuditPage(items=[], total=0, page=1, size=50, pages=0)
assert isinstance(page.total, int)
def test_audit_log_read_requires_id_and_action():
with pytest.raises(ValidationError):
AuthzAuditLogRead( # type: ignore[call-arg]
user_id=None,
action="flow:read",
resource_type=None,
resource_id=None,
result="allow",
details=None,
timestamp=datetime.now(timezone.utc),
)
with pytest.raises(ValidationError):
AuthzAuditLogRead( # type: ignore[call-arg]
id=uuid4(),
user_id=None,
resource_type=None,
resource_id=None,
result="allow",
details=None,
timestamp=datetime.now(timezone.utc),
)

108
src/backend/tests/unit/api/v1/test_authz_share_schemas.py Normal file
View File

@ -0,0 +1,108 @@
"""Schema-level tests for the AuthzShare CRUD payloads.
The integration-level test (live app fixture, DB row roundtrip) lives in
``test_authz_shares.py``; here we just pin the validator that enforces the
``scope_target_consistency`` rule at the API boundary so callers get 422 with
a readable message instead of a SQL constraint failure.
"""
from __future__ import annotations
from uuid import uuid4
import pytest
from langflow.api.v1.schemas.authz_shares import ShareCreate
from pydantic import ValidationError
def test_user_scope_requires_target_id():
"""scope=user must carry a target_id (the user being granted access)."""
with pytest.raises(ValidationError) as exc:
ShareCreate(
resource_type="flow",
resource_id=uuid4(),
scope="user",
target_id=None,
permission_level="read",
)
assert "target_id" in str(exc.value)
def test_team_scope_requires_target_id():
with pytest.raises(ValidationError) as exc:
ShareCreate(
resource_type="flow",
resource_id=uuid4(),
scope="team",
target_id=None,
permission_level="read",
)
assert "target_id" in str(exc.value)
def test_public_scope_rejects_target_id():
"""scope=public is meaningless with a target_id; reject at the API edge."""
with pytest.raises(ValidationError) as exc:
ShareCreate(
resource_type="flow",
resource_id=uuid4(),
scope="public",
target_id=uuid4(),
permission_level="read",
)
assert "target_id" in str(exc.value)
def test_private_scope_rejects_target_id():
with pytest.raises(ValidationError) as exc:
ShareCreate(
resource_type="flow",
resource_id=uuid4(),
scope="private",
target_id=uuid4(),
permission_level="read",
)
assert "target_id" in str(exc.value)
def test_user_scope_with_target_is_valid():
payload = ShareCreate(
resource_type="flow",
resource_id=uuid4(),
scope="user",
target_id=uuid4(),
permission_level="write",
)
assert payload.scope == "user"
assert payload.permission_level == "write"
def test_public_scope_without_target_is_valid():
payload = ShareCreate(
resource_type="deployment",
resource_id=uuid4(),
scope="public",
permission_level="read",
)
assert payload.target_id is None
def test_unknown_resource_type_rejected():
"""Resource type is a Literal so unknown values 422 at the schema edge."""
with pytest.raises(ValidationError):
ShareCreate(
resource_type="banana", # type: ignore[arg-type]
resource_id=uuid4(),
scope="public",
permission_level="read",
)
def test_unknown_permission_level_rejected():
with pytest.raises(ValidationError):
ShareCreate(
resource_type="flow",
resource_id=uuid4(),
scope="public",
permission_level="execute_with_extra_steps", # type: ignore[arg-type]
)

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

@ -2,7 +2,14 @@
from __future__ import annotations
from langflow.services.authorization.actions import DeploymentAction, FlowAction
from langflow.services.authorization.actions import (
DeploymentAction,
FileAction,
FlowAction,
KnowledgeBaseAction,
ShareAction,
VariableAction,
)
def test_flow_action_values_match_casbin_strings():
@ -46,3 +53,36 @@ def test_deployment_action_is_iterable_and_complete():
"""The enum exposes exactly the five canonical deployment actions (no DEPLOY)."""
values = {member.value for member in DeploymentAction}
assert values == {"read", "write", "create", "delete", "execute"}
def test_knowledge_base_action_values():
"""Knowledge bases support read/write/create/delete plus an INGEST verb."""
values = {member.value for member in KnowledgeBaseAction}
assert values == {"read", "write", "create", "delete", "ingest"}
def test_variable_action_values():
values = {member.value for member in VariableAction}
assert values == {"read", "write", "create", "delete"}
def test_file_action_values():
values = {member.value for member in FileAction}
assert values == {"read", "write", "create", "delete"}
def test_share_action_values():
"""Share actions cover CRUD over ``authz_share`` rows themselves."""
values = {member.value for member in ShareAction}
assert values == {"read", "create", "update", "delete"}
def test_new_actions_subclass_str():
"""All new action enums subclass str so they coerce in audit/log paths."""
for member in (
KnowledgeBaseAction.READ,
VariableAction.CREATE,
FileAction.DELETE,
ShareAction.UPDATE,
):
assert isinstance(member, str)

55
src/backend/tests/unit/services/authorization/test_capability_flag.py Normal file
View File

@ -0,0 +1,55 @@
"""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.
"""
from __future__ import annotations
from types import SimpleNamespace
import pytest
from langflow.services.authorization.service import LangflowAuthorizationService
from lfx.services.authorization.base import BaseAuthorizationService
from lfx.services.authorization.service import AuthorizationService as LfxDefaultService
def _settings(*, authz_enabled: bool = False) -> SimpleNamespace:
return SimpleNamespace(
auth_settings=SimpleNamespace(
AUTHZ_ENABLED=authz_enabled,
AUTHZ_SUPERUSER_BYPASS=True,
)
)
@pytest.mark.anyio
async def test_base_class_default_is_false():
"""The class-level constant defaults False so subclasses must opt in."""
assert BaseAuthorizationService.SUPPORTS_CROSS_USER_FETCH is False
@pytest.mark.anyio
async def test_lfx_default_service_does_not_support_cross_user_fetch():
"""The lfx no-op service inherits the safe default."""
service = LfxDefaultService()
assert await service.supports_cross_user_fetch() is False
@pytest.mark.anyio
async def test_langflow_pass_through_does_not_support_cross_user_fetch():
"""OSS pass-through must NOT opt in — that is the strict-pass-through contract."""
service = LangflowAuthorizationService(_settings())
assert await service.supports_cross_user_fetch() is False
@pytest.mark.anyio
async def test_subclass_can_opt_in():
"""Enterprise plugins flip ``SUPPORTS_CROSS_USER_FETCH=True``; the base accepts it."""
class _Enterprise(LangflowAuthorizationService):
SUPPORTS_CROSS_USER_FETCH = True
service = _Enterprise(_settings())
assert await service.supports_cross_user_fetch() is True

120
src/backend/tests/unit/services/authorization/test_fetch.py Normal file
View File

@ -0,0 +1,120 @@
"""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.
"""
from __future__ import annotations
from typing import Any, ClassVar
from unittest.mock import patch
from uuid import uuid4
import pytest
from fastapi import HTTPException
from langflow.services.authorization.fetch import authorized_or_owner_scoped, deny_to_404
from langflow.services.database.models.flow.model import Flow
from lfx.services.authorization.base import BaseAuthorizationService
# Reuse the live Flow model so the test exercises a real SQLAlchemy
# InstrumentedAttribute path. The session is fake — only the compiled SQL
# matters, not actual rows.
_DummyRow = Flow
class _StubService(BaseAuthorizationService):
"""Minimal subclass so tests can flip SUPPORTS_CROSS_USER_FETCH."""
SUPPORTS_CROSS_USER_FETCH: ClassVar[bool] = False
def __init__(self, *, supports_cross_user: bool = False) -> None:
super().__init__()
self._supports = supports_cross_user
self.set_ready()
async def supports_cross_user_fetch(self) -> bool:
return self._supports
async def is_enabled(self) -> bool:
return False
async def enforce(self, **_: Any) -> bool:
return True
async def batch_enforce(self, *, requests, **_: Any) -> list[bool]:
return [True] * len(requests)
class _FakeResult:
def __init__(self, value: Any) -> None:
self._value = value
def first(self) -> Any:
return self._value
class _FakeSession:
"""Captures the compiled SQL each call passes to ``exec``."""
def __init__(self, *, returns: Any = "row") -> None:
self.calls: list[str] = []
self.returns = returns
async def exec(self, stmt: Any) -> _FakeResult:
# str(stmt) gives the compiled SQL — enough to assert which branch ran.
self.calls.append(str(stmt))
return _FakeResult(self.returns)
@pytest.mark.anyio
async def test_owner_scoped_when_service_does_not_support_cross_user_fetch():
"""OSS pass-through default keeps the owner predicate so visibility cannot widen."""
session = _FakeSession(returns=object())
service = _StubService(supports_cross_user=False)
with patch(
"langflow.services.authorization.fetch.get_authorization_service",
return_value=service,
):
await authorized_or_owner_scoped(
session,
_DummyRow,
id_column=_DummyRow.id,
resource_id=uuid4(),
owner_column=_DummyRow.user_id,
owner_id=uuid4(),
)
# The owner-scoped branch adds a ``user_id =`` predicate in WHERE.
assert "user_id = :user_id" in session.calls[0]
@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."""
session = _FakeSession(returns=object())
service = _StubService(supports_cross_user=True)
with patch(
"langflow.services.authorization.fetch.get_authorization_service",
return_value=service,
):
await authorized_or_owner_scoped(
session,
_DummyRow,
id_column=_DummyRow.id,
resource_id=uuid4(),
owner_column=_DummyRow.user_id,
owner_id=uuid4(),
)
# No user_id = :user_id predicate in the share-aware path.
assert "user_id = :user_id" not in session.calls[0]
assert "WHERE flow.id" in session.calls[0]
def test_deny_to_404_only_rewrites_403():
"""The helper preserves UUID privacy by converting 403 → 404 only."""
rewritten = deny_to_404(HTTPException(status_code=403, detail="nope"))
assert rewritten.status_code == 404
untouched = deny_to_404(HTTPException(status_code=500, detail="boom"))
assert untouched.status_code == 500

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

@ -8,7 +8,15 @@ from uuid import uuid4
import pytest
from fastapi import HTTPException
from langflow.services.authorization import utils as authz_utils
from langflow.services.authorization.actions import DeploymentAction, FlowAction, ProjectAction
from langflow.services.authorization.actions import (
DeploymentAction,
FileAction,
FlowAction,
KnowledgeBaseAction,
ProjectAction,
ShareAction,
VariableAction,
)
class _StubAuthorizationService:
@ -711,3 +719,172 @@ async def test_deployment_owner_override_skips_enforce(monkeypatch, fake_user):
assert len(audit_calls) == 1
assert audit_calls[0]["result"] == "owner_override"
assert audit_calls[0]["action"] == "deployment:delete"
# --------------------------------------------------------------------------- #
# ensure_knowledge_base_permission
# --------------------------------------------------------------------------- #
@pytest.mark.anyio
async def test_kb_permission_kb_name_in_object_slug(monkeypatch, fake_user):
"""KBs are name-keyed; the Casbin obj slug must carry the name verbatim."""
_install_settings(monkeypatch, authz_enabled=True)
service = _StubAuthorizationService(allow=True)
_install_authz(monkeypatch, service)
_install_audit_recorder(monkeypatch)
await authz_utils.ensure_knowledge_base_permission(
fake_user,
KnowledgeBaseAction.READ,
kb_name="my-kb",
kb_user_id=uuid4(),
)
assert service.calls[0]["obj"] == "knowledge_base:my-kb"
assert service.calls[0]["act"] == "read"
assert service.calls[0]["context"]["kb_name"] == "my-kb"
@pytest.mark.anyio
async def test_kb_permission_owner_override(monkeypatch, fake_user):
"""KB owner is allowed even when the enforcer denies."""
_install_settings(monkeypatch, authz_enabled=True)
service = _StubAuthorizationService(allow=False)
_install_authz(monkeypatch, service)
audit_calls = _install_audit_recorder(monkeypatch)
await authz_utils.ensure_knowledge_base_permission(
fake_user,
KnowledgeBaseAction.DELETE,
kb_name="my-kb",
kb_user_id=fake_user.id,
)
assert service.calls == []
assert audit_calls[0]["result"] == "owner_override"
assert audit_calls[0]["action"] == "knowledge_base:delete"
@pytest.mark.anyio
async def test_kb_permission_denied_raises_403(monkeypatch, fake_user):
"""Non-owner + denied enforcer → 403 from the helper."""
_install_settings(monkeypatch, authz_enabled=True)
service = _StubAuthorizationService(allow=False)
_install_authz(monkeypatch, service)
_install_audit_recorder(monkeypatch)
with pytest.raises(HTTPException) as exc:
await authz_utils.ensure_knowledge_base_permission(
fake_user,
KnowledgeBaseAction.DELETE,
kb_name="someone-elses",
kb_user_id=uuid4(),
)
assert exc.value.status_code == 403
# --------------------------------------------------------------------------- #
# ensure_variable_permission
# --------------------------------------------------------------------------- #
@pytest.mark.anyio
async def test_variable_permission_uses_variable_object_slug(monkeypatch, fake_user):
_install_settings(monkeypatch, authz_enabled=True)
service = _StubAuthorizationService(allow=True)
_install_authz(monkeypatch, service)
_install_audit_recorder(monkeypatch)
variable_id = uuid4()
await authz_utils.ensure_variable_permission(
fake_user,
VariableAction.WRITE,
variable_id=variable_id,
variable_user_id=uuid4(),
)
assert service.calls[0]["obj"] == f"variable:{variable_id}"
@pytest.mark.anyio
async def test_variable_permission_owner_override(monkeypatch, fake_user):
_install_settings(monkeypatch, authz_enabled=True)
service = _StubAuthorizationService(allow=False)
_install_authz(monkeypatch, service)
audit_calls = _install_audit_recorder(monkeypatch)
await authz_utils.ensure_variable_permission(
fake_user,
VariableAction.DELETE,
variable_id=uuid4(),
variable_user_id=fake_user.id,
)
assert service.calls == []
assert audit_calls[0]["result"] == "owner_override"
# --------------------------------------------------------------------------- #
# ensure_file_permission
# --------------------------------------------------------------------------- #
@pytest.mark.anyio
async def test_file_permission_uses_file_object_slug(monkeypatch, fake_user):
_install_settings(monkeypatch, authz_enabled=True)
service = _StubAuthorizationService(allow=True)
_install_authz(monkeypatch, service)
_install_audit_recorder(monkeypatch)
file_id = uuid4()
await authz_utils.ensure_file_permission(
fake_user,
FileAction.READ,
file_id=file_id,
file_user_id=uuid4(),
)
assert service.calls[0]["obj"] == f"file:{file_id}"
@pytest.mark.anyio
async def test_file_permission_owner_override(monkeypatch, fake_user):
_install_settings(monkeypatch, authz_enabled=True)
service = _StubAuthorizationService(allow=False)
_install_authz(monkeypatch, service)
audit_calls = _install_audit_recorder(monkeypatch)
await authz_utils.ensure_file_permission(
fake_user,
FileAction.DELETE,
file_id=uuid4(),
file_user_id=fake_user.id,
)
assert service.calls == []
assert audit_calls[0]["result"] == "owner_override"
# --------------------------------------------------------------------------- #
# ensure_share_permission
# --------------------------------------------------------------------------- #
@pytest.mark.anyio
async def test_share_permission_uses_share_object_slug(monkeypatch, fake_user):
_install_settings(monkeypatch, authz_enabled=True)
service = _StubAuthorizationService(allow=True)
_install_authz(monkeypatch, service)
_install_audit_recorder(monkeypatch)
share_id = uuid4()
await authz_utils.ensure_share_permission(
fake_user,
ShareAction.CREATE,
share_id=share_id,
share_user_id=uuid4(),
)
assert service.calls[0]["obj"] == f"share:{share_id}"
assert service.calls[0]["act"] == "create"

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

@ -3,7 +3,7 @@
from __future__ import annotations
import abc
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, ClassVar
from lfx.services.base import Service
from lfx.services.schema import ServiceType
@ -22,6 +22,18 @@ class BaseAuthorizationService(Service, abc.ABC):
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.
SUPPORTS_CROSS_USER_FETCH: ClassVar[bool] = False
async def supports_cross_user_fetch(self) -> bool:
"""Return True when this service can authorize non-owner resource access."""
return self.SUPPORTS_CROSS_USER_FETCH
@abc.abstractmethod
async def is_enabled(self) -> bool:
"""Return True when authorization enforcement is active."""