From 505e8569c8a1bc66294262100a9d111aff5655bd Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Thu, 21 May 2026 11:07:02 -0700 Subject: [PATCH] feat: Phase 3 and 4 implementations for RBAC --- AGENTS.md | 12 +- .../8d3a1f9c2e0b_seed_authz_system_roles.py | 161 ++++++++ src/backend/base/langflow/api/router.py | 4 + src/backend/base/langflow/api/v1/__init__.py | 4 + .../base/langflow/api/v1/authz_audit.py | 122 ++++++ .../base/langflow/api/v1/authz_shares.py | 389 ++++++++++++++++++ src/backend/base/langflow/api/v1/endpoints.py | 8 +- .../base/langflow/api/v1/flows_helpers.py | 19 +- .../base/langflow/api/v1/knowledge_bases.py | 103 ++++- src/backend/base/langflow/api/v1/monitor.py | 101 ++++- src/backend/base/langflow/api/v1/projects.py | 82 ++-- .../langflow/api/v1/schemas/authz_shares.py | 82 ++++ src/backend/base/langflow/api/v1/variable.py | 28 ++ src/backend/base/langflow/api/v2/files.py | 75 +++- src/backend/base/langflow/helpers/flow.py | 17 +- .../services/authorization/__init__.py | 25 +- .../services/authorization/actions.py | 48 +++ .../langflow/services/authorization/fetch.py | 101 +++++ .../langflow/services/authorization/utils.py | 157 ++++++- .../database/models/deployment/crud.py | 21 +- .../alembic/test_seed_authz_system_roles.py | 64 +++ .../unit/api/v1/test_authz_audit_schemas.py | 87 ++++ .../unit/api/v1/test_authz_share_schemas.py | 108 +++++ .../services/authorization/test_actions.py | 42 +- .../authorization/test_capability_flag.py | 55 +++ .../unit/services/authorization/test_fetch.py | 120 ++++++ .../unit/services/authorization/test_utils.py | 179 +++++++- .../src/lfx/services/authorization/base.py | 14 +- 28 files changed, 2158 insertions(+), 70 deletions(-) create mode 100644 src/backend/base/langflow/alembic/versions/8d3a1f9c2e0b_seed_authz_system_roles.py create mode 100644 src/backend/base/langflow/api/v1/authz_audit.py create mode 100644 src/backend/base/langflow/api/v1/authz_shares.py create mode 100644 src/backend/base/langflow/api/v1/schemas/authz_shares.py create mode 100644 src/backend/base/langflow/services/authorization/fetch.py create mode 100644 src/backend/tests/unit/alembic/test_seed_authz_system_roles.py create mode 100644 src/backend/tests/unit/api/v1/test_authz_audit_schemas.py create mode 100644 src/backend/tests/unit/api/v1/test_authz_share_schemas.py create mode 100644 src/backend/tests/unit/services/authorization/test_capability_flag.py create mode 100644 src/backend/tests/unit/services/authorization/test_fetch.py diff --git a/AGENTS.md b/AGENTS.md index 8de604e45e..f3cc8be295 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -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 diff --git a/src/backend/base/langflow/alembic/versions/8d3a1f9c2e0b_seed_authz_system_roles.py b/src/backend/base/langflow/alembic/versions/8d3a1f9c2e0b_seed_authz_system_roles.py new file mode 100644 index 0000000000..23042b2eb3 --- /dev/null +++ b/src/backend/base/langflow/alembic/versions/8d3a1f9c2e0b_seed_authz_system_roles.py @@ -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), + ) + ) + ) diff --git a/src/backend/base/langflow/api/router.py b/src/backend/base/langflow/api/router.py index 886a478d9d..3aec88f70b 100644 --- a/src/backend/base/langflow/api/router.py +++ b/src/backend/base/langflow/api/router.py @@ -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 diff --git a/src/backend/base/langflow/api/v1/__init__.py b/src/backend/base/langflow/api/v1/__init__.py index 003323c72a..4b289bcca9 100644 --- a/src/backend/base/langflow/api/v1/__init__.py +++ b/src/backend/base/langflow/api/v1/__init__.py @@ -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", diff --git a/src/backend/base/langflow/api/v1/authz_audit.py b/src/backend/base/langflow/api/v1/authz_audit.py new file mode 100644 index 0000000000..abc77344b0 --- /dev/null +++ b/src/backend/base/langflow/api/v1/authz_audit.py @@ -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"] diff --git a/src/backend/base/langflow/api/v1/authz_shares.py b/src/backend/base/langflow/api/v1/authz_shares.py new file mode 100644 index 0000000000..9825b7ff4f --- /dev/null +++ b/src/backend/base/langflow/api/v1/authz_shares.py @@ -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"] diff --git a/src/backend/base/langflow/api/v1/endpoints.py b/src/backend/base/langflow/api/v1/endpoints.py index c008e46d7b..7a1c62dc1c 100644 --- a/src/backend/base/langflow/api/v1/endpoints.py +++ b/src/backend/base/langflow/api/v1/endpoints.py @@ -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) diff --git a/src/backend/base/langflow/api/v1/flows_helpers.py b/src/backend/base/langflow/api/v1/flows_helpers.py index 04dc0c8a31..833867fb37 100644 --- a/src/backend/base/langflow/api/v1/flows_helpers.py +++ b/src/backend/base/langflow/api/v1/flows_helpers.py @@ -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( diff --git a/src/backend/base/langflow/api/v1/knowledge_bases.py b/src/backend/base/langflow/api/v1/knowledge_bases.py index 3666c85316..7cdd9cb68a 100644 --- a/src/backend/base/langflow/api/v1/knowledge_bases.py +++ b/src/backend/base/langflow/api/v1/knowledge_bases.py @@ -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) diff --git a/src/backend/base/langflow/api/v1/monitor.py b/src/backend/base/langflow/api/v1/monitor.py index 5e1d2f886b..85e037ba93 100644 --- a/src/backend/base/langflow/api/v1/monitor.py +++ b/src/backend/base/langflow/api/v1/monitor.py @@ -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 diff --git a/src/backend/base/langflow/api/v1/projects.py b/src/backend/base/langflow/api/v1/projects.py index 120ddb6018..2c19e5cde5 100644 --- a/src/backend/base/langflow/api/v1/projects.py +++ b/src/backend/base/langflow/api/v1/projects.py @@ -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( diff --git a/src/backend/base/langflow/api/v1/schemas/authz_shares.py b/src/backend/base/langflow/api/v1/schemas/authz_shares.py new file mode 100644 index 0000000000..123a467d22 --- /dev/null +++ b/src/backend/base/langflow/api/v1/schemas/authz_shares.py @@ -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} diff --git a/src/backend/base/langflow/api/v1/variable.py b/src/backend/base/langflow/api/v1/variable.py index e56c016abf..ef79c0e5e3 100644 --- a/src/backend/base/langflow/api/v1/variable.py +++ b/src/backend/base/langflow/api/v1/variable.py @@ -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 diff --git a/src/backend/base/langflow/api/v2/files.py b/src/backend/base/langflow/api/v2/files.py index 6837502d82..526d4856ef 100644 --- a/src/backend/base/langflow/api/v2/files.py +++ b/src/backend/base/langflow/api/v2/files.py @@ -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) diff --git a/src/backend/base/langflow/helpers/flow.py b/src/backend/base/langflow/helpers/flow.py index a175c8e6a3..bf5655793b 100644 --- a/src/backend/base/langflow/helpers/flow.py +++ b/src/backend/base/langflow/helpers/flow.py @@ -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: diff --git a/src/backend/base/langflow/services/authorization/__init__.py b/src/backend/base/langflow/services/authorization/__init__.py index 3f731d2f4b..455fe6aed0 100644 --- a/src/backend/base/langflow/services/authorization/__init__.py +++ b/src/backend/base/langflow/services/authorization/__init__.py @@ -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", ] diff --git a/src/backend/base/langflow/services/authorization/actions.py b/src/backend/base/langflow/services/authorization/actions.py index b48ed5fb62..5487480abf 100644 --- a/src/backend/base/langflow/services/authorization/actions.py +++ b/src/backend/base/langflow/services/authorization/actions.py @@ -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" diff --git a/src/backend/base/langflow/services/authorization/fetch.py b/src/backend/base/langflow/services/authorization/fetch.py new file mode 100644 index 0000000000..4600d72ac0 --- /dev/null +++ b/src/backend/base/langflow/services/authorization/fetch.py @@ -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) diff --git a/src/backend/base/langflow/services/authorization/utils.py b/src/backend/base/langflow/services/authorization/utils.py index 85d025614b..7ea2afc913 100644 --- a/src/backend/base/langflow/services/authorization/utils.py +++ b/src/backend/base/langflow/services/authorization/utils.py @@ -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, *, diff --git a/src/backend/base/langflow/services/database/models/deployment/crud.py b/src/backend/base/langflow/services/database/models/deployment/crud.py index 57f885ef62..22813990d1 100644 --- a/src/backend/base/langflow/services/database/models/deployment/crud.py +++ b/src/backend/base/langflow/services/database/models/deployment/crud.py @@ -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() diff --git a/src/backend/tests/unit/alembic/test_seed_authz_system_roles.py b/src/backend/tests/unit/alembic/test_seed_authz_system_roles.py new file mode 100644 index 0000000000..227b452961 --- /dev/null +++ b/src/backend/tests/unit/alembic/test_seed_authz_system_roles.py @@ -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" diff --git a/src/backend/tests/unit/api/v1/test_authz_audit_schemas.py b/src/backend/tests/unit/api/v1/test_authz_audit_schemas.py new file mode 100644 index 0000000000..479e6ce78c --- /dev/null +++ b/src/backend/tests/unit/api/v1/test_authz_audit_schemas.py @@ -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), + ) diff --git a/src/backend/tests/unit/api/v1/test_authz_share_schemas.py b/src/backend/tests/unit/api/v1/test_authz_share_schemas.py new file mode 100644 index 0000000000..ef5545381d --- /dev/null +++ b/src/backend/tests/unit/api/v1/test_authz_share_schemas.py @@ -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] + ) diff --git a/src/backend/tests/unit/services/authorization/test_actions.py b/src/backend/tests/unit/services/authorization/test_actions.py index f033ab051a..a4c8e814ae 100644 --- a/src/backend/tests/unit/services/authorization/test_actions.py +++ b/src/backend/tests/unit/services/authorization/test_actions.py @@ -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) diff --git a/src/backend/tests/unit/services/authorization/test_capability_flag.py b/src/backend/tests/unit/services/authorization/test_capability_flag.py new file mode 100644 index 0000000000..f7dac8ef4e --- /dev/null +++ b/src/backend/tests/unit/services/authorization/test_capability_flag.py @@ -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 diff --git a/src/backend/tests/unit/services/authorization/test_fetch.py b/src/backend/tests/unit/services/authorization/test_fetch.py new file mode 100644 index 0000000000..bd6e30b4ee --- /dev/null +++ b/src/backend/tests/unit/services/authorization/test_fetch.py @@ -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 diff --git a/src/backend/tests/unit/services/authorization/test_utils.py b/src/backend/tests/unit/services/authorization/test_utils.py index 7cc9101663..b19b51aa95 100644 --- a/src/backend/tests/unit/services/authorization/test_utils.py +++ b/src/backend/tests/unit/services/authorization/test_utils.py @@ -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" diff --git a/src/lfx/src/lfx/services/authorization/base.py b/src/lfx/src/lfx/services/authorization/base.py index 01d17fbb49..0194308cbe 100644 --- a/src/lfx/src/lfx/services/authorization/base.py +++ b/src/lfx/src/lfx/services/authorization/base.py @@ -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."""