diff --git a/src/backend/base/langflow/alembic/versions/7c8d9e0f1a2b_authz_foundations.py b/src/backend/base/langflow/alembic/versions/7c8d9e0f1a2b_authz_foundations.py index 897a4ee77c..e64d34df76 100644 --- a/src/backend/base/langflow/alembic/versions/7c8d9e0f1a2b_authz_foundations.py +++ b/src/backend/base/langflow/alembic/versions/7c8d9e0f1a2b_authz_foundations.py @@ -146,7 +146,19 @@ def _seed_system_roles(conn: sa.Connection) -> None: ).scalar() if already_present: continue - conn.execute(authz_role.insert().values(**values)) + # Why: SELECT-then-INSERT is racy on dialects without + # ``on_conflict_do_nothing``. Two concurrent migration runners can + # both observe ``already_present=False`` and both INSERT, losing + # the second one to the unique-constraint. Wrap in a SAVEPOINT and + # swallow IntegrityError so the migration stays idempotent under + # concurrent first-deploys. See PR #13153 review item R5. + from sqlalchemy.exc import IntegrityError + + try: + with conn.begin_nested(): + conn.execute(authz_role.insert().values(**values)) + except IntegrityError: + continue def upgrade() -> None: diff --git a/src/backend/base/langflow/api/v1/authz_shares.py b/src/backend/base/langflow/api/v1/authz_shares.py index 2403e8194d..8c6f9c058a 100644 --- a/src/backend/base/langflow/api/v1/authz_shares.py +++ b/src/backend/base/langflow/api/v1/authz_shares.py @@ -101,7 +101,13 @@ async def _ensure_can_administer_share( return authz = get_authorization_service() if await authz.supports_cross_user_fetch() and await authz.is_enabled(): - # Defer to ensure_share_permission when enforcement is active. + # Why: in OSS, ``supports_cross_user_fetch()`` is False (see + # LangflowAuthorizationService), so this early-return is dead and the + # explicit 403 below is the floor. In enterprise, the plugin is the + # authoritative gate via the ``ensure_share_permission()`` call that + # every share route makes immediately after this helper. Removing + # that downstream call — or weakening the plugin's cross-user fetch + # contract — REOPENS the ownership gap. Keep the two in lockstep. return raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, diff --git a/src/backend/base/langflow/cli/authz_dry_run.py b/src/backend/base/langflow/cli/authz_dry_run.py index cb07868350..8057521d4e 100644 --- a/src/backend/base/langflow/cli/authz_dry_run.py +++ b/src/backend/base/langflow/cli/authz_dry_run.py @@ -173,7 +173,8 @@ class DryRunResult: @contextmanager def _install_stubs(stub: _StubAuthorizationService, audit_sink: list[dict[str, Any]]): """Swap the live authz helpers for the dry-run stubs for the duration of the run.""" - from langflow.services.authorization import utils as authz_utils + from langflow.services.authorization import audit as authz_audit + from langflow.services.authorization import guards as authz_guards settings = SimpleNamespace( auth_settings=SimpleNamespace( @@ -186,20 +187,25 @@ def _install_stubs(stub: _StubAuthorizationService, audit_sink: list[dict[str, A async def _record_audit(**kwargs: Any) -> None: audit_sink.append(kwargs) - saved_settings = authz_utils.get_settings_service - saved_authz = authz_utils.get_authorization_service - saved_audit = authz_utils.audit_decision + saved = { + "guards_settings": authz_guards.get_settings_service, + "guards_authz": authz_guards.get_authorization_service, + "audit_settings": authz_audit.get_settings_service, + "audit_decision": authz_audit.audit_decision, + } - authz_utils.get_settings_service = lambda: settings # type: ignore[assignment] - authz_utils.get_authorization_service = lambda: stub # type: ignore[assignment] - authz_utils.audit_decision = _record_audit # type: ignore[assignment] + authz_guards.get_settings_service = lambda: settings # type: ignore[assignment] + authz_guards.get_authorization_service = lambda: stub # type: ignore[assignment] + authz_audit.get_settings_service = lambda: settings # type: ignore[assignment] + authz_audit.audit_decision = _record_audit # type: ignore[assignment] try: yield finally: - authz_utils.get_settings_service = saved_settings # type: ignore[assignment] - authz_utils.get_authorization_service = saved_authz # type: ignore[assignment] - authz_utils.audit_decision = saved_audit # type: ignore[assignment] + authz_guards.get_settings_service = saved["guards_settings"] # type: ignore[assignment] + authz_guards.get_authorization_service = saved["guards_authz"] # type: ignore[assignment] + authz_audit.get_settings_service = saved["audit_settings"] # type: ignore[assignment] + authz_audit.audit_decision = saved["audit_decision"] # type: ignore[assignment] async def _run_one( @@ -213,7 +219,7 @@ async def _run_one( audit_sink: list[dict[str, Any]], ) -> DryRunResult: """Invoke ``ensure_flow_permission`` for one (guard, actor) pair and record the outcome.""" - from langflow.services.authorization import utils as authz_utils + from langflow.services.authorization import guards as authz_utils from langflow.services.authorization.actions import FlowAction actor_id = owner_id if actor.is_flow_owner else uuid4() diff --git a/src/backend/base/langflow/services/authorization/__init__.py b/src/backend/base/langflow/services/authorization/__init__.py index 49ade762c2..6e0583bdc8 100644 --- a/src/backend/base/langflow/services/authorization/__init__.py +++ b/src/backend/base/langflow/services/authorization/__init__.py @@ -9,10 +9,12 @@ from langflow.services.authorization.actions import ( 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 ( +from langflow.services.authorization.audit import ( audit_decision, + drain_pending_audit_writes, +) +from langflow.services.authorization.fetch import authorized_or_owner_scoped, deny_to_404 +from langflow.services.authorization.guards import ( ensure_deployment_permission, ensure_file_permission, ensure_flow_permission, @@ -21,8 +23,9 @@ from langflow.services.authorization.utils import ( ensure_project_permission, ensure_share_permission, ensure_variable_permission, - filter_visible_resources, ) +from langflow.services.authorization.listing import filter_visible_resources +from langflow.services.authorization.service import LangflowAuthorizationService __all__ = [ "DeploymentAction", @@ -36,6 +39,7 @@ __all__ = [ "audit_decision", "authorized_or_owner_scoped", "deny_to_404", + "drain_pending_audit_writes", "ensure_deployment_permission", "ensure_file_permission", "ensure_flow_permission", diff --git a/src/backend/base/langflow/services/authorization/audit.py b/src/backend/base/langflow/services/authorization/audit.py new file mode 100644 index 0000000000..618a66e106 --- /dev/null +++ b/src/backend/base/langflow/services/authorization/audit.py @@ -0,0 +1,277 @@ +"""Batched audit pipeline for authorization decisions. + +An earlier revision did ``asyncio.create_task(_write())`` per authorization +decision, with each write opening its own ``session_scope()``. That works on +light traffic, but on a real workload (every authenticated request emits at +least one audit row) it turns into a connection-pool storm. + +This module routes every decision through a bounded queue drained by a +single long-lived writer task. The writer batches up to ``_AUDIT_BATCH_MAX`` +rows per ``session_scope()`` and commits them in one INSERT. +``audit_decision`` stays a non-blocking ``put_nowait``: if the queue is +saturated the row is dropped (with time-based warning) so the request path +is never gated on the audit DB. +""" + +from __future__ import annotations + +import asyncio +import time +from typing import Any +from uuid import UUID + +from lfx.log.logger import logger + +from langflow.services.deps import get_settings_service + +# Shared audit result vocabulary. +AUDIT_ALLOW = "allow" +AUDIT_DENY = "deny" +AUDIT_OWNER_OVERRIDE = "owner_override" + +_AUDIT_QUEUE_MAX = 10_000 +_AUDIT_BATCH_MAX = 100 + +# Minimum seconds between drop warnings while saturation persists. +_AUDIT_DROP_WARN_INTERVAL = 10.0 + + +def _split_obj(obj: str) -> tuple[str | None, UUID | None]: + """Parse an authz obj key like 'flow:abc' into (resource_type, resource_id). + + Wildcards (``flow:*``) and unparseable ids return None for ``resource_id`` + so audit rows are still written with the right ``resource_type``. + """ + if ":" not in obj: + return None, None + resource_type, _, suffix = obj.partition(":") + if not suffix or suffix == "*": + return resource_type, None + try: + return resource_type, UUID(suffix) + except (ValueError, TypeError): + return resource_type, None + + +class _AuditEntry: + """One pending audit row. + + A plain class (not a dataclass) so it can be instantiated cheaply from the + request path without dataclass overhead. The fields mirror + ``AuthzAuditLog`` columns plus the raw ``obj`` string — the writer splits + ``obj`` into ``(resource_type, resource_id)`` once per batch. + """ + + __slots__ = ("action", "details", "obj", "result", "user_id") + + def __init__( + self, + *, + user_id: UUID | None, + action: str, + obj: str, + result: str, + details: dict[str, Any] | None, + ) -> None: + self.user_id = user_id + self.action = action + self.obj = obj + self.result = result + self.details = details + + +# Module-level state. Bound to whichever event loop is running when the first +# ``audit_decision`` call happens. ``_audit_queue_loop`` lets us detect a fresh +# loop (e.g. between pytest test cases) and restart the writer in the new loop +# instead of writing to a queue tied to a dead loop. +_audit_queue: asyncio.Queue[_AuditEntry] | None = None +_audit_queue_loop: asyncio.AbstractEventLoop | None = None +_audit_writer_task: asyncio.Task[None] | None = None +_audit_dropped_count: int = 0 +_audit_last_drop_warn: float = 0.0 +# Kept as a vestigial public name for backward compatibility with downstream +# callers (and the existing drain helper). The new pipeline tracks the single +# writer task here so ``drain_pending_audit_writes`` can await it. +_pending_audit_tasks: set[asyncio.Task[None]] = set() + + +def _ensure_audit_writer_started() -> asyncio.Queue[_AuditEntry] | None: + """Lazily start the audit writer task in the current event loop. + + Returns the queue, or ``None`` if no event loop is running (audit is + skipped entirely in that case — there's no place to schedule the writer). + """ + global _audit_queue, _audit_queue_loop, _audit_writer_task # noqa: PLW0603 + + try: + loop = asyncio.get_running_loop() + except RuntimeError: + return None + + # A fresh event loop replaces the previous queue+writer. Without this, + # a subsequent ``audit_decision`` call (e.g. in a new pytest test) would + # ``put_nowait`` into a queue that no live task is consuming. + if _audit_queue_loop is not loop: + _audit_queue = asyncio.Queue(maxsize=_AUDIT_QUEUE_MAX) + _audit_queue_loop = loop + _audit_writer_task = None + _pending_audit_tasks.clear() + + if _audit_writer_task is None or _audit_writer_task.done(): + _audit_writer_task = loop.create_task(_audit_writer_loop()) + _pending_audit_tasks.add(_audit_writer_task) + _audit_writer_task.add_done_callback(_pending_audit_tasks.discard) + + return _audit_queue + + +async def _audit_writer_loop() -> None: + """Drain the audit queue and write batches to the DB. + + Loops until cancelled. Each iteration blocks on the first row, then greedily + pulls everything else already enqueued up to ``_AUDIT_BATCH_MAX`` and + commits them as a single batch insert. DB exceptions are logged and + swallowed — an audit-table outage must never crash the request path that + triggered the row. + """ + while True: + queue = _audit_queue + if queue is None: + return + try: + first = await queue.get() + except asyncio.CancelledError: + return + + batch: list[_AuditEntry] = [first] + try: + while len(batch) < _AUDIT_BATCH_MAX: + batch.append(queue.get_nowait()) + except asyncio.QueueEmpty: + pass + + try: + await _flush_audit_batch(batch) + except Exception: # noqa: BLE001 — never let the writer die quietly + logger.exception("Authz audit writer batch flush failed for %d row(s)", len(batch)) + finally: + for _ in batch: + queue.task_done() + + +async def _flush_audit_batch(batch: list[_AuditEntry]) -> None: + """Insert a batch of ``_AuditEntry`` rows in a single session.""" + if not batch: + return + # Imported lazily so the request path doesn't pull DB modules until the + # writer first runs (matches the lazy import in the old per-row path). + from langflow.services.database.models.auth import AuthzAuditLog + from langflow.services.deps import session_scope + + async with session_scope() as session: + for entry in batch: + resource_type, resource_id = _split_obj(entry.obj) + session.add( + AuthzAuditLog( + user_id=entry.user_id, + action=entry.action, + resource_type=resource_type, + resource_id=resource_id, + result=entry.result, + details=entry.details, + ) + ) + + +async def drain_pending_audit_writes(timeout: float = 5.0) -> None: + """Flush the audit queue and stop the writer (bounded by ``timeout``). + + Safe to call multiple times; safe to call when no audit traffic has run. + Splits the timeout between draining the queue and awaiting writer + cancellation so neither side can hang shutdown indefinitely. + """ + global _audit_writer_task # noqa: PLW0603 + + queue = _audit_queue + writer = _audit_writer_task + if queue is None or writer is None: + return + + drain_budget = max(0.1, timeout * 0.8) + cancel_budget = max(0.1, timeout - drain_budget) + + try: + await asyncio.wait_for(queue.join(), timeout=drain_budget) + except asyncio.TimeoutError: + logger.warning( + "drain_pending_audit_writes timed out after %.2fs with %d row(s) pending", + drain_budget, + queue.qsize(), + ) + + if not writer.done(): + writer.cancel() + from contextlib import suppress + + with suppress(asyncio.CancelledError): + await asyncio.wait_for(writer, timeout=cancel_budget) + + _pending_audit_tasks.discard(writer) + _audit_writer_task = None + + +async def audit_decision( + *, + user_id: UUID | None, + action: str, + obj: str, + result: str, + details: dict[str, Any] | None = None, +) -> None: + """Enqueue an AuthzAuditLog row for batched background insertion. + + Non-blocking from the caller's perspective. When the queue is saturated the + row is dropped and a time-bounded warning is emitted so a stuck pipeline is + operator-visible without flooding the log. Audit is fully bypassed when + ``AUTHZ_AUDIT_ENABLED=False`` (the default). + """ + global _audit_dropped_count, _audit_last_drop_warn # noqa: PLW0603 + + settings = get_settings_service() + auth_settings = settings.auth_settings + # Audit is independent of enforcement. ``AuthSettings.AUTHZ_AUDIT_ENABLED`` + # defaults to ``False`` (see lfx/services/settings/auth.py) because the + # background writer still consumes a DB connection; operators opt in. + if not getattr(auth_settings, "AUTHZ_AUDIT_ENABLED", False): + return + + queue = _ensure_audit_writer_started() + if queue is None: + # No running event loop — nothing to schedule against. The caller is + # likely outside an async context (e.g. a sync test); silently skip. + return + + entry = _AuditEntry( + user_id=user_id, + action=action, + obj=obj, + result=result, + details=details, + ) + try: + queue.put_nowait(entry) + except asyncio.QueueFull: + _audit_dropped_count += 1 + # Time-based: always log the first drop, then at most once per + # ``_AUDIT_DROP_WARN_INTERVAL`` while saturation persists. Cheaper to + # reason about for operators than the previous every-1000th heuristic + # (which could go minutes without a log line at low drop rates). + now = time.monotonic() + if _audit_dropped_count == 1 or (now - _audit_last_drop_warn) >= _AUDIT_DROP_WARN_INTERVAL: + _audit_last_drop_warn = now + logger.warning( + "AuthzAuditLog queue full (%d/%d); dropped %d row(s) total. DB writer is likely behind or stalled.", + queue.qsize(), + _AUDIT_QUEUE_MAX, + _audit_dropped_count, + ) diff --git a/src/backend/base/langflow/services/authorization/guards.py b/src/backend/base/langflow/services/authorization/guards.py new file mode 100644 index 0000000000..d57c4d25d1 --- /dev/null +++ b/src/backend/base/langflow/services/authorization/guards.py @@ -0,0 +1,495 @@ +"""Permission-enforcement guards for guarded API routes.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +from fastapi import HTTPException, status +from lfx.log.logger import logger + +from langflow.services.authorization import audit as _audit +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: + from uuid import UUID + + from langflow.services.database.models.user.model import User, UserRead + +# Action enums coerced to string values. +_ACTION_ENUMS = ( + FlowAction, + DeploymentAction, + ProjectAction, + KnowledgeBaseAction, + VariableAction, + FileAction, + ShareAction, +) + +# Resource-owner keys included in audit details (kept in one place so a new +# resource type does not need to hand-update ``ensure_permission``). +_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", +) + +# Default 403 detail. UUID-leaking detail strings are opt-in via ``detail=...`` +# on ``ensure_permission`` so the secure default is the easy path. See review +# item I2 on PR #13153. +_DEFAULT_DENY_DETAIL = "Permission denied" + + +def _auth_context(user: User | UserRead) -> dict[str, Any]: + """Build the base context dict passed to authorization enforce calls.""" + return {"is_superuser": getattr(user, "is_superuser", False)} + + +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, _ACTION_ENUMS): + return act.value + return act + + +def _resolve_casbin_domain(workspace_id: UUID | None, scope_id: UUID | None) -> str: + """Resolve policy domain: project scope, then workspace, else ``*``.""" + if scope_id is not None: + return f"project:{scope_id}" + if workspace_id is not None: + return f"workspace:{workspace_id}" + return "*" + + +# Backward-compatible alias. +_resolve_flow_domain = _resolve_casbin_domain + + +async def ensure_permission( + user: User | UserRead, + *, + domain: str, + obj: str, + act: str, + context: dict[str, Any] | None = None, + detail: str | None = None, +) -> None: + """Raise HTTP 403 when the user may not perform the action (audited). + + ``detail`` overrides the default non-disclosing 403 message. Callers that + have already verified the resource exists for the requester (e.g. routes + operating on user-owned data) may pass a richer string. The default is + intentionally generic so a missing ``deny_to_404`` wrap cannot leak the + resource UUID — see review item I2 on PR #13153. + """ + settings = get_settings_service() + if not settings.auth_settings.AUTHZ_ENABLED: + return + + authz = get_authorization_service() + # Caller context first; user auth fields cannot be overwritten. + merged_context = {**(context or {}), **_auth_context(user)} + # Fail closed when enforce() raises (deny + audit, not HTTP 500). + audit_action = f"{obj.split(':', 1)[0]}:{act}" if ":" in obj else act + audit_details: dict[str, Any] = {"domain": domain} + for owner_key in _OWNER_CONTEXT_KEYS: + if owner_key in merged_context and merged_context[owner_key] is not None: + audit_details[owner_key] = str(merged_context[owner_key]) + deny_detail = detail if detail is not None else _DEFAULT_DENY_DETAIL + try: + allowed = await authz.enforce( + user_id=user.id, + domain=domain, + obj=obj, + act=act, + context=merged_context, + ) + except Exception as exc: + logger.exception("Authorization plugin raised during enforce; failing closed") + await _audit.audit_decision( + user_id=user.id, + action=audit_action, + obj=obj, + result=_audit.AUDIT_DENY, + details={**audit_details, "error": str(exc)}, + ) + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=deny_detail, + ) from exc + + await _audit.audit_decision( + user_id=user.id, + action=audit_action, + obj=obj, + result=_audit.AUDIT_ALLOW if allowed else _audit.AUDIT_DENY, + details=audit_details, + ) + + if not allowed: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=deny_detail, + ) + + +async def _ensure_resource_permission( + user: User | UserRead, + *, + resource_type: str, + resource_id: UUID | str | None, + owner_id: UUID | None, + act_str: str, + resolved_domain: str, + extra_context: dict[str, Any], +) -> None: + """Build object key, apply owner override, else delegate to ensure_permission.""" + obj = f"{resource_type}:{resource_id}" if resource_id else f"{resource_type}:*" + + if owner_id is not None and getattr(user, "id", None) == owner_id: + await _audit.audit_decision( + user_id=user.id, + action=f"{resource_type}:{act_str}", + obj=obj, + result=_audit.AUDIT_OWNER_OVERRIDE, + details={"domain": resolved_domain}, + ) + return + + await ensure_permission( + user, + domain=resolved_domain, + obj=obj, + act=act_str, + context=extra_context, + ) + + +# --------------------------------------------------------------------------- # +# Resource registry — collapses the ensure_*_permission family into one path. +# See review item I1 on PR #13153. +# --------------------------------------------------------------------------- # + + +@dataclass(frozen=True) +class _ResourceSpec: + """Static metadata for one resource type's ``ensure_*_permission`` helper.""" + + resource_type: str # Object slug prefix, e.g. "flow" + owner_kw: str # Public kwarg carrying the owner id, e.g. "flow_user_id" + id_kw: str # Public kwarg carrying the resource id, e.g. "flow_id" + # Public kwarg the resolver consults for the *outer* domain bucket + # (workspace). ``None`` means the helper does not accept a workspace scope. + workspace_kw: str | None + # Public kwarg the resolver consults for the *inner* domain bucket + # (project/folder). ``None`` means the helper does not accept this scope. + scope_kw: str | None + # Extra kwargs forwarded verbatim into ``extra_context`` (no domain effect). + extra_context_kws: tuple[str, ...] = () + + +_RESOURCE_SPECS: dict[str, _ResourceSpec] = { + "flow": _ResourceSpec( + resource_type="flow", + owner_kw="flow_user_id", + id_kw="flow_id", + workspace_kw="workspace_id", + scope_kw="folder_id", + ), + "deployment": _ResourceSpec( + resource_type="deployment", + owner_kw="deployment_user_id", + id_kw="deployment_id", + workspace_kw="workspace_id", + scope_kw="project_id", + ), + "project": _ResourceSpec( + resource_type="project", + owner_kw="project_user_id", + id_kw="project_id", + workspace_kw="workspace_id", + scope_kw=None, + ), + "knowledge_base": _ResourceSpec( + resource_type="knowledge_base", + owner_kw="kb_user_id", + id_kw="kb_id", + workspace_kw="workspace_id", + scope_kw="project_id", + extra_context_kws=("kb_name",), + ), + "variable": _ResourceSpec( + resource_type="variable", + owner_kw="variable_user_id", + id_kw="variable_id", + workspace_kw="workspace_id", + scope_kw=None, + ), + "file": _ResourceSpec( + resource_type="file", + owner_kw="file_user_id", + id_kw="file_id", + workspace_kw="workspace_id", + scope_kw=None, + ), + "share": _ResourceSpec( + resource_type="share", + owner_kw="share_user_id", + id_kw="share_id", + workspace_kw=None, + scope_kw=None, + ), +} + + +async def _ensure_typed( + user: User | UserRead, + *, + spec_key: str, + act_str: str, + kwargs: dict[str, Any], + domain_override: str | None, +) -> None: + """Shared body for ``ensure_*_permission`` helpers. + + ``kwargs`` is the keyword-argument dict the caller passed to the public + helper. The registry tells us which key carries the id, the owner, and the + domain components; everything else flows into ``extra_context`` verbatim. + """ + spec = _RESOURCE_SPECS[spec_key] + + resource_id = kwargs.get(spec.id_kw) + owner_id = kwargs.get(spec.owner_kw) + workspace_id = kwargs.get(spec.workspace_kw) if spec.workspace_kw else None + scope_id = kwargs.get(spec.scope_kw) if spec.scope_kw else None + + if domain_override is not None: + resolved_domain = domain_override + elif spec.workspace_kw is None and spec.scope_kw is None: + # Shares have no domain bucket. + resolved_domain = "*" + else: + resolved_domain = _resolve_casbin_domain(workspace_id, scope_id) + + # ``extra_context`` mirrors the legacy clone shape so audit rows and + # plugin matchers continue to see the same key names. We forward every + # kwarg the helper declares except ``domain`` (already resolved). + extra_context: dict[str, Any] = {} + if spec.workspace_kw is not None: + extra_context[spec.workspace_kw] = workspace_id + if spec.scope_kw is not None: + extra_context[spec.scope_kw] = scope_id + extra_context[spec.owner_kw] = owner_id + if spec_key == "knowledge_base": + # KB also forwards the resource id under its public name for plugins + # that scope policy by kb_id directly (mirrors the legacy clone). + extra_context[spec.id_kw] = resource_id + for key in spec.extra_context_kws: + extra_context[key] = kwargs.get(key) + + await _ensure_resource_permission( + user, + resource_type=spec.resource_type, + resource_id=resource_id, + owner_id=owner_id, + act_str=act_str, + resolved_domain=resolved_domain, + extra_context=extra_context, + ) + + +# --------------------------------------------------------------------------- # +# Public ``ensure_*_permission`` helpers — thin wrappers over ``_ensure_typed``. +# Kept individually typed so call sites get IDE / mypy help on which kwargs +# each resource accepts. +# --------------------------------------------------------------------------- # + + +async def ensure_flow_permission( + user: User | UserRead, + act: FlowAction | str, + *, + flow_id: UUID | None = None, + flow_user_id: UUID | None = None, + workspace_id: UUID | None = None, + folder_id: UUID | None = None, + domain: str | None = None, +) -> None: + """Check flow permission (owner override, then plugin enforce).""" + await _ensure_typed( + user, + spec_key="flow", + act_str=_coerce_action(act), + kwargs={ + "flow_id": flow_id, + "flow_user_id": flow_user_id, + "workspace_id": workspace_id, + "folder_id": folder_id, + }, + domain_override=domain, + ) + + +async def ensure_deployment_permission( + user: User | UserRead, + act: DeploymentAction | str, + *, + deployment_id: UUID | None = None, + deployment_user_id: UUID | None = None, + workspace_id: UUID | None = None, + project_id: UUID | None = None, + domain: str | None = None, +) -> None: + """Check deployment permission (owner override, then plugin enforce).""" + await _ensure_typed( + user, + spec_key="deployment", + act_str=_coerce_action(act), + kwargs={ + "deployment_id": deployment_id, + "deployment_user_id": deployment_user_id, + "workspace_id": workspace_id, + "project_id": project_id, + }, + domain_override=domain, + ) + + +async def ensure_project_permission( + user: User | UserRead, + act: ProjectAction | str, + *, + project_id: UUID | None = None, + project_user_id: UUID | None = None, + workspace_id: UUID | None = None, + domain: str | None = None, +) -> None: + """Check project (folder) permission (owner override, then plugin enforce).""" + await _ensure_typed( + user, + spec_key="project", + act_str=_coerce_action(act), + kwargs={ + "project_id": project_id, + "project_user_id": project_user_id, + "workspace_id": workspace_id, + }, + domain_override=domain, + ) + + +async def ensure_knowledge_base_permission( + user: User | UserRead, + act: KnowledgeBaseAction | str, + *, + kb_id: UUID | None = None, + kb_user_id: UUID | None = None, + kb_name: str | None = None, + workspace_id: UUID | None = None, + project_id: UUID | None = None, + domain: str | None = None, +) -> None: + """Check knowledge-base permission (owner override, then plugin enforce).""" + await _ensure_typed( + user, + spec_key="knowledge_base", + act_str=_coerce_action(act), + kwargs={ + "kb_id": kb_id, + "kb_user_id": kb_user_id, + "kb_name": kb_name, + "workspace_id": workspace_id, + "project_id": project_id, + }, + domain_override=domain, + ) + + +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 permission (owner override, then plugin enforce).""" + await _ensure_typed( + user, + spec_key="variable", + act_str=_coerce_action(act), + kwargs={ + "variable_id": variable_id, + "variable_user_id": variable_user_id, + "workspace_id": workspace_id, + }, + domain_override=domain, + ) + + +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 permission (owner override, then plugin enforce).""" + await _ensure_typed( + user, + spec_key="file", + act_str=_coerce_action(act), + kwargs={ + "file_id": file_id, + "file_user_id": file_user_id, + "workspace_id": workspace_id, + }, + domain_override=domain, + ) + + +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 share-row permission (owner override, then plugin enforce).""" + await _ensure_typed( + user, + spec_key="share", + act_str=_coerce_action(act), + kwargs={ + "share_id": share_id, + "share_user_id": share_user_id, + }, + domain_override=domain, + ) diff --git a/src/backend/base/langflow/services/authorization/listing.py b/src/backend/base/langflow/services/authorization/listing.py new file mode 100644 index 0000000000..f9945a7c33 --- /dev/null +++ b/src/backend/base/langflow/services/authorization/listing.py @@ -0,0 +1,92 @@ +"""List-endpoint filter for authorization checks (batched enforce).""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, TypeVar + +from langflow.services.authorization.actions import FlowAction +from langflow.services.authorization.guards import _auth_context, _coerce_action +from langflow.services.deps import get_authorization_service, get_settings_service + +if TYPE_CHECKING: + from collections.abc import Callable + from uuid import UUID + + from langflow.services.database.models.user.model import User, UserRead + + +T = TypeVar("T") + + +def _default_resource_id_getter(item: Any) -> UUID: + """Default key extractor used by filter_visible_resources.""" + return item.id + + +async def filter_visible_resources( + user: User | UserRead, + *, + resource_type: str, + candidates: list[T], + key: Callable[[T], UUID] | None = None, + domain: str = "*", + domain_extractor: Callable[[T], str] | None = None, + owner_extractor: Callable[[T], UUID | None] | None = None, + act: FlowAction | str = FlowAction.READ, +) -> list[T]: + """Return candidates the user may read (no-op when AUTHZ_ENABLED is false).""" + settings = get_settings_service() + if not settings.auth_settings.AUTHZ_ENABLED or not candidates: + return candidates + + extractor = key if key is not None else _default_resource_id_getter + authz = get_authorization_service() + act_str = _coerce_action(act) + user_id = getattr(user, "id", None) + + # Owned rows skip batch_enforce (matches direct-read owner override). + owned_indices: set[int] = set() + enforce_indices: list[int] = [] + enforce_items: list[T] = [] + for index, item in enumerate(candidates): + if owner_extractor is not None and user_id is not None and owner_extractor(item) == user_id: + owned_indices.add(index) + else: + enforce_indices.append(index) + enforce_items.append(item) + + decisions: list[bool] = [False] * len(candidates) + for index in owned_indices: + decisions[index] = True + + if enforce_items: + if domain_extractor is None: + # Single-domain batch_enforce. + requests = [(f"{resource_type}:{extractor(item)}", act_str) for item in enforce_items] + results = await authz.batch_enforce( + user_id=user_id, + domain=domain, + requests=requests, + context=_auth_context(user), + ) + for original_index, allowed in zip(enforce_indices, results, strict=True): + decisions[original_index] = allowed + else: + # One batch_enforce per resolved domain. + buckets: dict[str, list[tuple[int, T]]] = {} + for original_index, item in zip(enforce_indices, enforce_items, strict=True): + buckets.setdefault(domain_extractor(item), []).append((original_index, item)) + + auth_context = _auth_context(user) + for resolved_domain, bucket in buckets.items(): + bucket_requests = [(f"{resource_type}:{extractor(item)}", act_str) for _, item in bucket] + bucket_results = await authz.batch_enforce( + user_id=user_id, + domain=resolved_domain, + requests=bucket_requests, + context=auth_context, + ) + for (original_index, _), allowed in zip(bucket, bucket_results, strict=True): + decisions[original_index] = allowed + + return [item for item, allowed in zip(candidates, decisions, strict=True) if allowed] diff --git a/src/backend/base/langflow/services/authorization/utils.py b/src/backend/base/langflow/services/authorization/utils.py index 4600a030e2..df5f95308c 100644 --- a/src/backend/base/langflow/services/authorization/utils.py +++ b/src/backend/base/langflow/services/authorization/utils.py @@ -1,694 +1,100 @@ -"""Authorization helpers for guarded API routes.""" +"""Backward-compatible re-exports for the split authorization helpers. + +The implementations now live in: + +* ``langflow.services.authorization.audit`` — batched audit pipeline +* ``langflow.services.authorization.guards`` — ``ensure_*_permission`` family +* ``langflow.services.authorization.listing`` — ``filter_visible_resources`` + +This module is preserved so existing imports +(``from langflow.services.authorization.utils import ensure_flow_permission``) +keep working. New code should import from the focused submodules above. +""" from __future__ import annotations -import asyncio -from typing import TYPE_CHECKING, Any, TypeVar -from uuid import UUID - from fastapi import HTTPException, status -from lfx.log.logger import logger -from langflow.services.authorization.actions import ( - DeploymentAction, - FileAction, - FlowAction, - KnowledgeBaseAction, - ProjectAction, - ShareAction, - VariableAction, +from langflow.services.authorization.audit import ( + _AUDIT_BATCH_MAX, + _AUDIT_QUEUE_MAX, + _audit_writer_loop, + _AuditEntry, + _ensure_audit_writer_started, + _flush_audit_batch, + _split_obj, + audit_decision, + drain_pending_audit_writes, +) +from langflow.services.authorization.audit import ( + AUDIT_ALLOW as _AUDIT_ALLOW, +) +from langflow.services.authorization.audit import ( + AUDIT_DENY as _AUDIT_DENY, +) +from langflow.services.authorization.audit import ( + AUDIT_OWNER_OVERRIDE as _AUDIT_OWNER_OVERRIDE, +) +from langflow.services.authorization.guards import ( + _ACTION_ENUMS, + _OWNER_CONTEXT_KEYS, + _auth_context, + _coerce_action, + _ensure_resource_permission, + _resolve_casbin_domain, + _resolve_flow_domain, + ensure_deployment_permission, + ensure_file_permission, + ensure_flow_permission, + ensure_knowledge_base_permission, + ensure_permission, + ensure_project_permission, + ensure_share_permission, + ensure_variable_permission, +) +from langflow.services.authorization.listing import ( + _default_resource_id_getter, + filter_visible_resources, ) from langflow.services.deps import get_authorization_service, get_settings_service -if TYPE_CHECKING: - from collections.abc import Callable - from langflow.services.auth.exceptions import InsufficientPermissionsError - from langflow.services.database.models.user.model import User, UserRead - - -T = TypeVar("T") - -# Shared audit result vocabulary. -_AUDIT_ALLOW = "allow" -_AUDIT_DENY = "deny" -_AUDIT_OWNER_OVERRIDE = "owner_override" - -# Resource-owner keys included in audit details. -_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 enums coerced to string values. -_ACTION_ENUMS = ( - FlowAction, - DeploymentAction, - ProjectAction, - KnowledgeBaseAction, - VariableAction, - FileAction, - ShareAction, -) - - -def _auth_context(user: User | UserRead) -> dict[str, Any]: - """Build the base context dict passed to authorization enforce calls.""" - return {"is_superuser": getattr(user, "is_superuser", False)} - - -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, _ACTION_ENUMS): - return act.value - return act - - -def _split_obj(obj: str) -> tuple[str | None, UUID | None]: - """Parse an authz obj key like 'flow:abc' into (resource_type, resource_id). - - Wildcards (``flow:*``) and unparseable ids return None for ``resource_id`` - so audit rows are still written with the right ``resource_type``. - """ - if ":" not in obj: - return None, None - resource_type, _, suffix = obj.partition(":") - if not suffix or suffix == "*": - return resource_type, None - try: - return resource_type, UUID(suffix) - except (ValueError, TypeError): - return resource_type, None - - -# ----------------------------------------------------------------------------- # -# Batched audit pipeline. -# -# An earlier revision did ``asyncio.create_task(_write())`` per authorization -# decision, with each write opening its own ``session_scope()``. That works on -# light traffic, but on a real workload (every authenticated request emits at -# least one audit row) it turns into a connection-pool storm: per request you -# pay for the request session + a fresh audit session, and the unbounded -# ``_pending_audit_tasks`` set grows until shutdown drain. -# -# This version routes every decision through a bounded queue drained by a -# single long-lived writer task. The writer batches up to ``_AUDIT_BATCH_MAX`` -# rows per ``session_scope()`` and commits them in one INSERT. ``audit_decision`` -# stays a non-blocking ``put_nowait``: if the queue is saturated the row is -# dropped (with a periodic warning) so the request path is never gated on the -# audit DB. -# ----------------------------------------------------------------------------- # - -_AUDIT_QUEUE_MAX = 10_000 -_AUDIT_BATCH_MAX = 100 - - -class _AuditEntry: - """One pending audit row. - - A plain class (not a dataclass) so it can be instantiated cheaply from the - request path without dataclass overhead. The fields mirror - ``AuthzAuditLog`` columns plus the raw ``obj`` string — the writer splits - ``obj`` into ``(resource_type, resource_id)`` once per batch. - """ - - __slots__ = ("action", "details", "obj", "result", "user_id") - - def __init__( - self, - *, - user_id: UUID | None, - action: str, - obj: str, - result: str, - details: dict[str, Any] | None, - ) -> None: - self.user_id = user_id - self.action = action - self.obj = obj - self.result = result - self.details = details - - -# Module-level state. Bound to whichever event loop is running when the first -# ``audit_decision`` call happens. ``_audit_queue_loop`` lets us detect a fresh -# loop (e.g. between pytest test cases) and restart the writer in the new loop -# instead of writing to a queue tied to a dead loop. -_audit_queue: asyncio.Queue[_AuditEntry] | None = None -_audit_queue_loop: asyncio.AbstractEventLoop | None = None -_audit_writer_task: asyncio.Task[None] | None = None -_audit_dropped_count: int = 0 -# Kept as a vestigial public name for backward compatibility with downstream -# callers (and the existing drain helper). The new pipeline tracks the single -# writer task here so ``drain_pending_audit_writes`` can await it. -_pending_audit_tasks: set[asyncio.Task[None]] = set() - - -def _ensure_audit_writer_started() -> asyncio.Queue[_AuditEntry] | None: - """Lazily start the audit writer task in the current event loop. - - Returns the queue, or ``None`` if no event loop is running (audit is - skipped entirely in that case — there's no place to schedule the writer). - """ - global _audit_queue, _audit_queue_loop, _audit_writer_task # noqa: PLW0603 - - try: - loop = asyncio.get_running_loop() - except RuntimeError: - return None - - # A fresh event loop replaces the previous queue+writer. Without this, - # a subsequent ``audit_decision`` call (e.g. in a new pytest test) would - # ``put_nowait`` into a queue that no live task is consuming. - if _audit_queue_loop is not loop: - _audit_queue = asyncio.Queue(maxsize=_AUDIT_QUEUE_MAX) - _audit_queue_loop = loop - _audit_writer_task = None - _pending_audit_tasks.clear() - - if _audit_writer_task is None or _audit_writer_task.done(): - _audit_writer_task = loop.create_task(_audit_writer_loop()) - _pending_audit_tasks.add(_audit_writer_task) - _audit_writer_task.add_done_callback(_pending_audit_tasks.discard) - - return _audit_queue - - -async def _audit_writer_loop() -> None: - """Drain the audit queue and write batches to the DB. - - Loops until cancelled. Each iteration blocks on the first row, then greedily - pulls everything else already enqueued up to ``_AUDIT_BATCH_MAX`` and - commits them as a single batch insert. DB exceptions are logged and - swallowed — an audit-table outage must never crash the request path that - triggered the row. - """ - while True: - queue = _audit_queue - if queue is None: - return - try: - first = await queue.get() - except asyncio.CancelledError: - return - - batch: list[_AuditEntry] = [first] - try: - while len(batch) < _AUDIT_BATCH_MAX: - batch.append(queue.get_nowait()) - except asyncio.QueueEmpty: - pass - - try: - await _flush_audit_batch(batch) - except Exception: # noqa: BLE001 — never let the writer die quietly - logger.exception("Authz audit writer batch flush failed for %d row(s)", len(batch)) - finally: - for _ in batch: - queue.task_done() - - -async def _flush_audit_batch(batch: list[_AuditEntry]) -> None: - """Insert a batch of ``_AuditEntry`` rows in a single session.""" - if not batch: - return - # Imported lazily so the request path doesn't pull DB modules until the - # writer first runs (matches the lazy import in the old per-row path). - from langflow.services.database.models.auth import AuthzAuditLog - from langflow.services.deps import session_scope - - async with session_scope() as session: - for entry in batch: - resource_type, resource_id = _split_obj(entry.obj) - session.add( - AuthzAuditLog( - user_id=entry.user_id, - action=entry.action, - resource_type=resource_type, - resource_id=resource_id, - result=entry.result, - details=entry.details, - ) - ) - - -async def drain_pending_audit_writes(timeout: float = 5.0) -> None: - """Flush the audit queue and stop the writer (bounded by ``timeout``). - - Safe to call multiple times; safe to call when no audit traffic has run. - Splits the timeout between draining the queue and awaiting writer - cancellation so neither side can hang shutdown indefinitely. - """ - global _audit_writer_task # noqa: PLW0603 - - queue = _audit_queue - writer = _audit_writer_task - if queue is None or writer is None: - return - - drain_budget = max(0.1, timeout * 0.8) - cancel_budget = max(0.1, timeout - drain_budget) - - try: - await asyncio.wait_for(queue.join(), timeout=drain_budget) - except asyncio.TimeoutError: - logger.warning( - "drain_pending_audit_writes timed out after %.2fs with %d row(s) pending", - drain_budget, - queue.qsize(), - ) - - if not writer.done(): - writer.cancel() - from contextlib import suppress - - with suppress(asyncio.CancelledError): - await asyncio.wait_for(writer, timeout=cancel_budget) - - _pending_audit_tasks.discard(writer) - _audit_writer_task = None - - -async def audit_decision( - *, - user_id: UUID | None, - action: str, - obj: str, - result: str, - details: dict[str, Any] | None = None, -) -> None: - """Enqueue an AuthzAuditLog row for batched background insertion. - - Non-blocking from the caller's perspective. When the queue is saturated the - row is dropped and a sample-rate warning is emitted so a stuck pipeline is - operator-visible. Audit is fully bypassed when - ``AUTHZ_AUDIT_ENABLED=False`` (the default). - """ - global _audit_dropped_count # noqa: PLW0603 - - settings = get_settings_service() - auth_settings = settings.auth_settings - # Audit is independent of enforcement. ``AuthSettings.AUTHZ_AUDIT_ENABLED`` - # defaults to ``False`` (see lfx/services/settings/auth.py) because the - # background writer still consumes a DB connection; operators opt in. - if not getattr(auth_settings, "AUTHZ_AUDIT_ENABLED", False): - return - - queue = _ensure_audit_writer_started() - if queue is None: - # No running event loop — nothing to schedule against. The caller is - # likely outside an async context (e.g. a sync test); silently skip. - return - - entry = _AuditEntry( - user_id=user_id, - action=action, - obj=obj, - result=result, - details=details, - ) - try: - queue.put_nowait(entry) - except asyncio.QueueFull: - _audit_dropped_count += 1 - # Sample-log so a saturated pipeline surfaces in ops dashboards without - # spamming the log on every drop. First drop + every 1000th drop. - if _audit_dropped_count == 1 or _audit_dropped_count % 1000 == 0: - logger.warning( - "AuthzAuditLog queue full (%d/%d); dropped %d row(s) total. DB writer is likely behind or stalled.", - queue.qsize(), - _AUDIT_QUEUE_MAX, - _audit_dropped_count, - ) - - -async def ensure_permission( - user: User | UserRead, - *, - domain: str, - obj: str, - act: str, - context: dict[str, Any] | None = None, -) -> None: - """Raise HTTP 403 when the user may not perform the action (audited).""" - settings = get_settings_service() - if not settings.auth_settings.AUTHZ_ENABLED: - return - - authz = get_authorization_service() - # Caller context first; user auth fields cannot be overwritten. - merged_context = {**(context or {}), **_auth_context(user)} - # Fail closed when enforce() raises (deny + audit, not HTTP 500). - audit_action = f"{obj.split(':', 1)[0]}:{act}" if ":" in obj else act - audit_details = {"domain": domain} - for owner_key in _OWNER_CONTEXT_KEYS: - if owner_key in merged_context and merged_context[owner_key] is not None: - audit_details[owner_key] = str(merged_context[owner_key]) - try: - allowed = await authz.enforce( - user_id=user.id, - domain=domain, - obj=obj, - act=act, - context=merged_context, - ) - except Exception as exc: - logger.exception("Authorization plugin raised during enforce; failing closed") - await audit_decision( - user_id=user.id, - action=audit_action, - obj=obj, - result=_AUDIT_DENY, - details={**audit_details, "error": str(exc)}, - ) - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail=f"Insufficient permissions to {act} on {obj}", - ) from exc - - await audit_decision( - user_id=user.id, - action=audit_action, - obj=obj, - result=_AUDIT_ALLOW if allowed else _AUDIT_DENY, - details=audit_details, - ) - - if not allowed: - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail=f"Insufficient permissions to {act} on {obj}", - ) - - -def _resolve_casbin_domain(workspace_id: UUID | None, scope_id: UUID | None) -> str: - """Resolve policy domain: project scope, then workspace, else ``*``.""" - if scope_id is not None: - return f"project:{scope_id}" - if workspace_id is not None: - return f"workspace:{workspace_id}" - return "*" - - -# Backward-compatible alias. -_resolve_flow_domain = _resolve_casbin_domain - - -async def _ensure_resource_permission( - user: User | UserRead, - *, - resource_type: str, - resource_id: UUID | str | None, - owner_id: UUID | None, - act_str: str, - resolved_domain: str, - extra_context: dict[str, Any], -) -> None: - """Build object key, apply owner override, else delegate to ensure_permission.""" - obj = f"{resource_type}:{resource_id}" if resource_id else f"{resource_type}:*" - - if owner_id is not None and getattr(user, "id", None) == owner_id: - await audit_decision( - user_id=user.id, - action=f"{resource_type}:{act_str}", - obj=obj, - result=_AUDIT_OWNER_OVERRIDE, - details={"domain": resolved_domain}, - ) - return - - await ensure_permission( - user, - domain=resolved_domain, - obj=obj, - act=act_str, - context=extra_context, - ) - - -async def ensure_flow_permission( - user: User | UserRead, - act: FlowAction | str, - *, - flow_id: UUID | None = None, - flow_user_id: UUID | None = None, - workspace_id: UUID | None = None, - folder_id: UUID | None = None, - domain: str | None = None, -) -> None: - """Check flow permission (owner override, then plugin enforce).""" - await _ensure_resource_permission( - user, - resource_type="flow", - resource_id=flow_id, - owner_id=flow_user_id, - act_str=_coerce_action(act), - resolved_domain=domain if domain is not None else _resolve_casbin_domain(workspace_id, folder_id), - extra_context={ - "flow_user_id": flow_user_id, - "workspace_id": workspace_id, - "folder_id": folder_id, - }, - ) - - -async def ensure_deployment_permission( - user: User | UserRead, - act: DeploymentAction | str, - *, - deployment_id: UUID | None = None, - deployment_user_id: UUID | None = None, - workspace_id: UUID | None = None, - project_id: UUID | None = None, - domain: str | None = None, -) -> None: - """Check deployment permission (owner override, then plugin enforce).""" - await _ensure_resource_permission( - user, - resource_type="deployment", - resource_id=deployment_id, - owner_id=deployment_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={ - "deployment_user_id": deployment_user_id, - "workspace_id": workspace_id, - "project_id": project_id, - }, - ) - - -async def ensure_project_permission( - user: User | UserRead, - act: ProjectAction | str, - *, - project_id: UUID | None = None, - project_user_id: UUID | None = None, - workspace_id: UUID | None = None, - domain: str | None = None, -) -> None: - """Check project (folder) permission (owner override, then plugin enforce).""" - await _ensure_resource_permission( - user, - resource_type="project", - resource_id=project_id, - owner_id=project_user_id, - act_str=_coerce_action(act), - resolved_domain=domain if domain is not None else _resolve_casbin_domain(workspace_id, None), - extra_context={ - "project_user_id": project_user_id, - "workspace_id": workspace_id, - }, - ) - - -async def ensure_knowledge_base_permission( - user: User | UserRead, - act: KnowledgeBaseAction | str, - *, - kb_id: UUID | None = None, - kb_user_id: UUID | None = None, - kb_name: str | None = None, - workspace_id: UUID | None = None, - project_id: UUID | None = None, - domain: str | None = None, -) -> None: - """Check knowledge-base permission (owner override, then plugin enforce).""" - await _ensure_resource_permission( - user, - resource_type="knowledge_base", - resource_id=kb_id, - 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_id": kb_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 permission (owner override, then plugin enforce).""" - 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 permission (owner override, then plugin enforce).""" - 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 share-row permission (owner override, then plugin enforce).""" - 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, - *, - resource_type: str, - candidates: list[T], - key: Callable[[T], UUID] | None = None, - domain: str = "*", - domain_extractor: Callable[[T], str] | None = None, - owner_extractor: Callable[[T], UUID | None] | None = None, - act: FlowAction | str = FlowAction.READ, -) -> list[T]: - """Return candidates the user may read (no-op when AUTHZ_ENABLED is false).""" - settings = get_settings_service() - if not settings.auth_settings.AUTHZ_ENABLED or not candidates: - return candidates - - extractor = key if key is not None else _default_resource_id_getter - authz = get_authorization_service() - act_str = _coerce_action(act) - user_id = getattr(user, "id", None) - - # Owned rows skip batch_enforce (matches direct-read owner override). - owned_indices: set[int] = set() - enforce_indices: list[int] = [] - enforce_items: list[T] = [] - for index, item in enumerate(candidates): - if owner_extractor is not None and user_id is not None and owner_extractor(item) == user_id: - owned_indices.add(index) - else: - enforce_indices.append(index) - enforce_items.append(item) - - decisions: list[bool] = [False] * len(candidates) - for index in owned_indices: - decisions[index] = True - - if enforce_items: - if domain_extractor is None: - # Single-domain batch_enforce. - requests = [(f"{resource_type}:{extractor(item)}", act_str) for item in enforce_items] - results = await authz.batch_enforce( - user_id=user_id, - domain=domain, - requests=requests, - context=_auth_context(user), - ) - for original_index, allowed in zip(enforce_indices, results, strict=True): - decisions[original_index] = allowed - else: - # One batch_enforce per resolved domain. - buckets: dict[str, list[tuple[int, T]]] = {} - for original_index, item in zip(enforce_indices, enforce_items, strict=True): - buckets.setdefault(domain_extractor(item), []).append((original_index, item)) - - auth_context = _auth_context(user) - for resolved_domain, bucket in buckets.items(): - bucket_requests = [(f"{resource_type}:{extractor(item)}", act_str) for _, item in bucket] - bucket_results = await authz.batch_enforce( - user_id=user_id, - domain=resolved_domain, - requests=bucket_requests, - context=auth_context, - ) - for (original_index, _), allowed in zip(bucket, bucket_results, strict=True): - decisions[original_index] = allowed - - return [item for item, allowed in zip(candidates, decisions, strict=True) if allowed] - - -def _default_resource_id_getter(item: Any) -> UUID: - """Default key extractor used by filter_visible_resources.""" - return item.id - - -def permission_denied_to_http(exc: InsufficientPermissionsError) -> HTTPException: +def permission_denied_to_http(exc): """Translate an InsufficientPermissionsError into a 403 HTTPException.""" return HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=exc.message) + + +__all__ = [ + "_ACTION_ENUMS", + "_AUDIT_ALLOW", + "_AUDIT_BATCH_MAX", + "_AUDIT_DENY", + "_AUDIT_OWNER_OVERRIDE", + "_AUDIT_QUEUE_MAX", + "_OWNER_CONTEXT_KEYS", + "_AuditEntry", + "_audit_writer_loop", + "_auth_context", + "_coerce_action", + "_default_resource_id_getter", + "_ensure_audit_writer_started", + "_ensure_resource_permission", + "_flush_audit_batch", + "_resolve_casbin_domain", + "_resolve_flow_domain", + "_split_obj", + "audit_decision", + "drain_pending_audit_writes", + "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", + "get_authorization_service", + "get_settings_service", + "permission_denied_to_http", +] diff --git a/src/backend/tests/unit/api/v1/test_authz_share_routes.py b/src/backend/tests/unit/api/v1/test_authz_share_routes.py index 0d69538e0b..05e7e394f2 100644 --- a/src/backend/tests/unit/api/v1/test_authz_share_routes.py +++ b/src/backend/tests/unit/api/v1/test_authz_share_routes.py @@ -74,19 +74,23 @@ class _StubAuthz: @pytest.fixture def patch_authz(monkeypatch): - """Install a stub authz service into both the shares module and utils.""" - from langflow.services.authorization import utils as authz_utils + """Install a stub authz service into the shares module and the split helper modules.""" + from langflow.services.authorization import audit as authz_audit + from langflow.services.authorization import guards as authz_guards + from langflow.services.authorization import listing as authz_listing def _apply(*, cross_user: bool = False, enabled: bool = False) -> _StubAuthz: stub = _StubAuthz(cross_user=cross_user, enabled=enabled) monkeypatch.setattr(shares_module, "get_authorization_service", lambda: stub) - monkeypatch.setattr(authz_utils, "get_authorization_service", lambda: stub) + for module in (authz_guards, authz_listing): + monkeypatch.setattr(module, "get_authorization_service", lambda: stub) # Disable audit + AUTHZ_ENABLED gating so audit calls inside ensure_* # don't open real sessions. settings = SimpleNamespace( auth_settings=SimpleNamespace(AUTHZ_ENABLED=False, AUTHZ_AUDIT_ENABLED=False), ) - monkeypatch.setattr(authz_utils, "get_settings_service", lambda: settings) + for module in (authz_audit, authz_guards, authz_listing): + monkeypatch.setattr(module, "get_settings_service", lambda s=settings: s) return stub return _apply diff --git a/src/backend/tests/unit/services/authorization/_common.py b/src/backend/tests/unit/services/authorization/_common.py new file mode 100644 index 0000000000..6953706119 --- /dev/null +++ b/src/backend/tests/unit/services/authorization/_common.py @@ -0,0 +1,80 @@ +"""Shared helpers for the split authorization-helper tests. + +After the ``utils.py`` refactor on PR #13153, the runtime helpers live in +three modules: + +* ``langflow.services.authorization.audit`` — batched audit pipeline +* ``langflow.services.authorization.guards`` — ``ensure_*_permission`` family +* ``langflow.services.authorization.listing`` — ``filter_visible_resources`` + +Each module imports ``get_settings_service`` / ``get_authorization_service`` +at import time, so a test that wants to stub the live services must patch the +attributes on *every* module that uses them. This file centralises the +patching shape so individual test files stay focused. + +Pytest fixtures (``fake_user``, ``fake_superuser``) live in +``conftest.py`` so they are discovered automatically without per-file imports. +""" + +from __future__ import annotations + +from types import SimpleNamespace + +from langflow.services.authorization import audit as authz_audit +from langflow.services.authorization import guards as authz_guards +from langflow.services.authorization import listing as authz_listing + + +class _StubAuthorizationService: + """Minimal stand-in for BaseAuthorizationService that records calls.""" + + def __init__(self, *, allow: bool = True, batch_results: list[bool] | None = None) -> None: + self.allow = allow + self.batch_results = batch_results + self.calls: list[dict] = [] + self.batch_calls: list[dict] = [] + + async def enforce(self, **kwargs) -> bool: + self.calls.append(kwargs) + return self.allow + + async def batch_enforce(self, **kwargs) -> list[bool]: + self.batch_calls.append(kwargs) + if self.batch_results is not None: + return self.batch_results + return [self.allow] * len(kwargs.get("requests", [])) + + +def install_settings(monkeypatch, *, authz_enabled: bool, audit_enabled: bool = False) -> None: + """Patch settings on every module that calls ``get_settings_service``.""" + settings = SimpleNamespace( + auth_settings=SimpleNamespace( + AUTHZ_ENABLED=authz_enabled, + AUTHZ_AUDIT_ENABLED=audit_enabled, + ), + ) + for module in (authz_audit, authz_guards, authz_listing): + monkeypatch.setattr(module, "get_settings_service", lambda s=settings: s) + + +def install_authz(monkeypatch, service: _StubAuthorizationService) -> None: + """Patch the authorization service factory on guards + listing.""" + for module in (authz_guards, authz_listing): + monkeypatch.setattr(module, "get_authorization_service", lambda s=service: s) + + +def install_audit_recorder(monkeypatch) -> list[dict]: + """Replace ``audit_decision`` with a recorder so tests can assert audit writes. + + ``ensure_permission`` in ``guards`` reaches for ``audit.audit_decision`` + via the module reference (``from ... import audit as _audit``), so patching + the function attribute on the ``audit`` module is sufficient — every caller + sees the recorder. + """ + calls: list[dict] = [] + + async def _recorder(**kwargs): + calls.append(kwargs) + + monkeypatch.setattr(authz_audit, "audit_decision", _recorder) + return calls diff --git a/src/backend/tests/unit/services/authorization/conftest.py b/src/backend/tests/unit/services/authorization/conftest.py new file mode 100644 index 0000000000..e08961af0f --- /dev/null +++ b/src/backend/tests/unit/services/authorization/conftest.py @@ -0,0 +1,20 @@ +"""Shared fixtures for the split authorization-helper tests.""" + +from __future__ import annotations + +from types import SimpleNamespace +from uuid import uuid4 + +import pytest + + +@pytest.fixture +def fake_user(): + """Build a non-superuser user object compatible with ensure_permission.""" + return SimpleNamespace(id=uuid4(), is_superuser=False) + + +@pytest.fixture +def fake_superuser(): + """Build a superuser user object compatible with ensure_permission.""" + return SimpleNamespace(id=uuid4(), is_superuser=True) diff --git a/src/backend/tests/unit/services/authorization/test_audit.py b/src/backend/tests/unit/services/authorization/test_audit.py new file mode 100644 index 0000000000..fc8aa1a249 --- /dev/null +++ b/src/backend/tests/unit/services/authorization/test_audit.py @@ -0,0 +1,149 @@ +"""Tests for the batched audit pipeline (``audit_decision`` + writer).""" + +from __future__ import annotations + +from uuid import uuid4 + +import pytest +from fastapi import HTTPException +from langflow.services.authorization import audit as authz_audit + +from ._common import ( + install_audit_recorder, + install_settings, +) + + +@pytest.fixture +def patched_audit_flush(monkeypatch): + """Replace ``_flush_audit_batch`` with a recorder so we exercise the writer without touching the DB.""" + flushed: list[list[object]] = [] + + async def _record(batch): + flushed.append(list(batch)) + + monkeypatch.setattr(authz_audit, "_flush_audit_batch", _record) + return flushed + + +async def _reset_audit_pipeline() -> None: + """Best-effort teardown so each test starts with a clean audit pipeline.""" + await authz_audit.drain_pending_audit_writes(timeout=0.5) + authz_audit._audit_queue = None + authz_audit._audit_queue_loop = None + authz_audit._audit_writer_task = None + authz_audit._audit_dropped_count = 0 + authz_audit._audit_last_drop_warn = 0.0 + authz_audit._pending_audit_tasks.clear() + + +@pytest.mark.anyio +async def test_audit_decision_runs_when_authz_disabled_but_audit_on(monkeypatch, patched_audit_flush): + """Audit is independent of enforcement. + + Previously ``audit_decision`` short-circuited when ``AUTHZ_ENABLED=False``, + which meant share CRUD writes left no audit trail on default installs. The + new contract gates only on ``AUTHZ_AUDIT_ENABLED`` so operators can + observe traffic before flipping enforcement on. + """ + await _reset_audit_pipeline() + install_settings(monkeypatch, authz_enabled=False, audit_enabled=True) + + await authz_audit.audit_decision(user_id=uuid4(), action="flow:read", obj="flow:x", result="allow") + await authz_audit.drain_pending_audit_writes(timeout=1.0) + + assert sum(len(b) for b in patched_audit_flush) == 1 + + +@pytest.mark.anyio +async def test_audit_decision_noop_when_audit_disabled(monkeypatch, patched_audit_flush): + """AUTHZ_AUDIT_ENABLED=False suppresses audit writes.""" + await _reset_audit_pipeline() + install_settings(monkeypatch, authz_enabled=True, audit_enabled=False) + + await authz_audit.audit_decision(user_id=uuid4(), action="flow:read", obj="flow:x", result="allow") + await authz_audit.drain_pending_audit_writes(timeout=0.5) + + assert patched_audit_flush == [] + + +@pytest.mark.anyio +async def test_audit_decision_enqueues_when_enabled(monkeypatch, patched_audit_flush): + """When both flags are on, ``audit_decision`` enqueues an entry the background writer flushes.""" + await _reset_audit_pipeline() + install_settings(monkeypatch, authz_enabled=True, audit_enabled=True) + + user_id = uuid4() + await authz_audit.audit_decision(user_id=user_id, action="flow:read", obj="flow:abc", result="allow") + # Drain forces the writer to flush before we inspect. + await authz_audit.drain_pending_audit_writes(timeout=1.0) + + assert len(patched_audit_flush) == 1 + batch = patched_audit_flush[0] + assert len(batch) == 1 + entry = batch[0] + assert entry.user_id == user_id + assert entry.action == "flow:read" + assert entry.obj == "flow:abc" + assert entry.result == "allow" + + +@pytest.mark.anyio +async def test_audit_decision_batches_multiple_entries(monkeypatch, patched_audit_flush): + """Multiple concurrent ``audit_decision`` calls coalesce into a single DB batch. + + This is the contract we want — the writer should pull every entry already + in the queue when it wakes up, so we make N decisions before yielding and + expect ONE batch of N rows, not N separate ``session_scope`` opens. + """ + await _reset_audit_pipeline() + install_settings(monkeypatch, authz_enabled=True, audit_enabled=True) + + for _ in range(5): + await authz_audit.audit_decision(user_id=uuid4(), action="flow:read", obj="flow:x", result="allow") + + await authz_audit.drain_pending_audit_writes(timeout=1.0) + + total_rows = sum(len(batch) for batch in patched_audit_flush) + assert total_rows == 5 + # All entries are emitted before the first await, so they should land in a single batch. + assert len(patched_audit_flush) == 1, ( + f"Expected 1 batch of 5 rows, got {len(patched_audit_flush)} batches " + f"with sizes {[len(b) for b in patched_audit_flush]}" + ) + + +@pytest.mark.anyio +async def test_drain_pending_audit_writes_is_safe_when_idle(): + """``drain_pending_audit_writes`` is a no-op when no audit traffic has run.""" + await _reset_audit_pipeline() + # Must not raise. + await authz_audit.drain_pending_audit_writes(timeout=0.1) + + +@pytest.mark.anyio +async def test_ensure_permission_fails_closed_on_plugin_exception(monkeypatch, fake_user): + """If the authz plugin raises, ``ensure_permission`` must deny (403), not bubble 500.""" + from langflow.services.authorization import guards as authz_guards + + install_settings(monkeypatch, authz_enabled=True, audit_enabled=False) + + class _BrokenPlugin: + async def enforce(self, **_kwargs): + msg = "policy store down" + raise RuntimeError(msg) + + async def batch_enforce(self, **_kwargs): + return [] + + monkeypatch.setattr(authz_guards, "get_authorization_service", lambda: _BrokenPlugin()) + captured = install_audit_recorder(monkeypatch) + + with pytest.raises(HTTPException) as excinfo: + await authz_guards.ensure_permission(fake_user, domain="*", obj="flow:abc", act="read") + + assert excinfo.value.status_code == 403, "Plugin exceptions must fail closed (deny), not 500" + # The deny path must still emit an audit row so the operator can see the failure. + assert captured, "Plugin exception must still produce an audit row" + assert captured[0]["result"] == "deny" + assert "error" in captured[0]["details"] diff --git a/src/backend/tests/unit/services/authorization/test_domain_resolution.py b/src/backend/tests/unit/services/authorization/test_domain_resolution.py new file mode 100644 index 0000000000..ea204ca042 --- /dev/null +++ b/src/backend/tests/unit/services/authorization/test_domain_resolution.py @@ -0,0 +1,46 @@ +"""Tests for ``_resolve_casbin_domain``, ``_split_obj``, and action coercion.""" + +from __future__ import annotations + +from uuid import uuid4 + +from langflow.services.authorization import audit as authz_audit +from langflow.services.authorization import guards as authz_guards + + +def test_resolve_casbin_domain_precedence(): + """Domain precedence: project > workspace > '*'.""" + ws, scope = uuid4(), uuid4() + assert authz_guards._resolve_casbin_domain(workspace_id=ws, scope_id=scope) == f"project:{scope}" + assert authz_guards._resolve_casbin_domain(workspace_id=ws, scope_id=None) == f"workspace:{ws}" + assert authz_guards._resolve_casbin_domain(workspace_id=None, scope_id=scope) == f"project:{scope}" + assert authz_guards._resolve_casbin_domain(workspace_id=None, scope_id=None) == "*" + # Backward-compatible alias still resolves to the same function. + assert authz_guards._resolve_flow_domain is authz_guards._resolve_casbin_domain + + +def test_split_obj_parses_uuid_suffix(): + """flow: splits into ('flow', UUID).""" + flow_id = uuid4() + resource_type, resource_id = authz_audit._split_obj(f"flow:{flow_id}") + assert resource_type == "flow" + assert resource_id == flow_id + + +def test_split_obj_wildcard_returns_none_id(): + """flow:* keeps resource_type but emits None for resource_id.""" + resource_type, resource_id = authz_audit._split_obj("flow:*") + assert resource_type == "flow" + assert resource_id is None + + +def test_split_obj_malformed_returns_nones(): + """A key without a colon returns (None, None).""" + assert authz_audit._split_obj("nothing") == (None, None) + + +def test_split_obj_non_uuid_suffix_returns_none_id(): + """A non-UUID suffix is treated as a wildcard for the resource_id field.""" + resource_type, resource_id = authz_audit._split_obj("flow:not-a-uuid") + assert resource_type == "flow" + assert resource_id is None diff --git a/src/backend/tests/unit/services/authorization/test_filter_visible.py b/src/backend/tests/unit/services/authorization/test_filter_visible.py new file mode 100644 index 0000000000..5d178221fb --- /dev/null +++ b/src/backend/tests/unit/services/authorization/test_filter_visible.py @@ -0,0 +1,172 @@ +"""Tests for ``filter_visible_resources`` (list-endpoint authorization filter).""" + +from __future__ import annotations + +from types import SimpleNamespace +from uuid import uuid4 + +import pytest +from langflow.services.authorization import guards as authz_guards +from langflow.services.authorization import listing as authz_listing +from langflow.services.authorization.actions import FlowAction + +from ._common import ( + _StubAuthorizationService, + install_authz, + install_settings, +) + + +@pytest.mark.anyio +async def test_filter_visible_resources_noop_when_disabled(monkeypatch, fake_user): + """No batch_enforce call when AUTHZ_ENABLED=False; returns input unchanged.""" + install_settings(monkeypatch, authz_enabled=False) + service = _StubAuthorizationService(allow=False) + install_authz(monkeypatch, service) + + candidates = [SimpleNamespace(id=uuid4()) for _ in range(3)] + result = await authz_listing.filter_visible_resources( + fake_user, + resource_type="flow", + candidates=candidates, + ) + assert result == candidates + assert service.batch_calls == [] + + +@pytest.mark.anyio +async def test_filter_visible_resources_empty_returns_empty(monkeypatch, fake_user): + """An empty candidates list is returned unchanged without contacting the service.""" + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=True) + install_authz(monkeypatch, service) + + result = await authz_listing.filter_visible_resources(fake_user, resource_type="flow", candidates=[]) + assert result == [] + assert service.batch_calls == [] + + +@pytest.mark.anyio +async def test_filter_visible_resources_filters_via_batch_enforce(monkeypatch, fake_user): + """When AUTHZ_ENABLED=True, batch_enforce results filter the candidate list.""" + install_settings(monkeypatch, authz_enabled=True) + candidates = [SimpleNamespace(id=uuid4()) for _ in range(3)] + service = _StubAuthorizationService(batch_results=[True, False, True]) + install_authz(monkeypatch, service) + + result = await authz_listing.filter_visible_resources( + fake_user, + resource_type="flow", + candidates=candidates, + ) + + assert result == [candidates[0], candidates[2]] + assert service.batch_calls[0]["requests"] == [ + (f"flow:{candidates[0].id}", "read"), + (f"flow:{candidates[1].id}", "read"), + (f"flow:{candidates[2].id}", "read"), + ] + + +@pytest.mark.anyio +async def test_filter_visible_resources_accepts_custom_key(monkeypatch, fake_user): + """A custom key extractor lets callers filter non-id-bearing items.""" + install_settings(monkeypatch, authz_enabled=True) + items = [{"resource_id": uuid4()}, {"resource_id": uuid4()}] + service = _StubAuthorizationService(batch_results=[False, True]) + install_authz(monkeypatch, service) + + result = await authz_listing.filter_visible_resources( + fake_user, + resource_type="project", + candidates=items, + key=lambda r: r["resource_id"], + act=FlowAction.WRITE, + ) + + assert result == [items[1]] + assert service.batch_calls[0]["requests"][0][1] == "write" + + +@pytest.mark.anyio +async def test_filter_visible_resources_groups_by_extracted_domain(monkeypatch, fake_user): + """With ``domain_extractor`` set, batch_enforce is called once per unique domain. + + Each call sees only the candidates that resolved to that domain, so the + authorization plugin evaluates each candidate against the right policy tuple + (the single-domain default would force every candidate through the same + wildcard domain, hiding project-scoped grants). + """ + install_settings(monkeypatch, authz_enabled=True) + workspace_a = uuid4() + workspace_b = uuid4() + + items = [ + SimpleNamespace(id=uuid4(), workspace_id=workspace_a, folder_id=None), + SimpleNamespace(id=uuid4(), workspace_id=workspace_b, folder_id=None), + SimpleNamespace(id=uuid4(), workspace_id=workspace_a, folder_id=None), + ] + + # Deny everything in workspace_b, allow everything in workspace_a. + class _DomainAwareStub: + def __init__(self) -> None: + self.batch_calls: list[dict] = [] + + async def batch_enforce(self, **kwargs) -> list[bool]: + self.batch_calls.append(kwargs) + allowed = kwargs["domain"] != f"workspace:{workspace_b}" + return [allowed] * len(kwargs["requests"]) + + service = _DomainAwareStub() + install_authz(monkeypatch, service) + + result = await authz_listing.filter_visible_resources( + fake_user, + resource_type="project", + candidates=items, + domain_extractor=lambda project: authz_guards._resolve_casbin_domain(project.workspace_id, None), + act=FlowAction.READ, + ) + + # Two calls — one per unique domain. + domains_called = {call["domain"] for call in service.batch_calls} + assert domains_called == {f"workspace:{workspace_a}", f"workspace:{workspace_b}"} + + # Output preserves the original order, with workspace_b's item dropped. + assert result == [items[0], items[2]] + + +@pytest.mark.anyio +async def test_filter_visible_resources_owner_override_skips_enforcer(monkeypatch, fake_user): + """Items owned by the caller are force-included without consulting the enforcer. + + Mirrors the owner-override short-circuit in ``_ensure_resource_permission`` + so list and direct-read agree under plugin enforcement. Without this, + a deny-all plugin would hide the caller's own rows from the listing + response while letting them read the same rows directly. + """ + install_settings(monkeypatch, authz_enabled=True) + other_user = uuid4() + + items = [ + SimpleNamespace(id=uuid4(), user_id=fake_user.id), # owned → must keep + SimpleNamespace(id=uuid4(), user_id=other_user), # not owned → enforcer decides + SimpleNamespace(id=uuid4(), user_id=fake_user.id), # owned → must keep + ] + # Deny-all stub so any item that reaches the enforcer would be dropped. + service = _StubAuthorizationService(allow=False) + install_authz(monkeypatch, service) + + result = await authz_listing.filter_visible_resources( + fake_user, + resource_type="flow", + candidates=items, + owner_extractor=lambda item: item.user_id, + act=FlowAction.READ, + ) + + # Owned items kept (positions 0 and 2); non-owned item dropped by deny. + assert result == [items[0], items[2]] + # Enforcer was consulted only for the non-owned item. + assert len(service.batch_calls) == 1 + assert len(service.batch_calls[0]["requests"]) == 1 diff --git a/src/backend/tests/unit/services/authorization/test_guards.py b/src/backend/tests/unit/services/authorization/test_guards.py new file mode 100644 index 0000000000..5f85f5a399 --- /dev/null +++ b/src/backend/tests/unit/services/authorization/test_guards.py @@ -0,0 +1,634 @@ +"""Tests for ``ensure_permission`` and the ``ensure_*_permission`` family.""" + +from __future__ import annotations + +from uuid import uuid4 + +import pytest +from fastapi import HTTPException +from langflow.services.authorization import guards as authz_guards +from langflow.services.authorization.actions import ( + DeploymentAction, + FileAction, + FlowAction, + KnowledgeBaseAction, + ProjectAction, + ShareAction, + VariableAction, +) + +from ._common import ( + _StubAuthorizationService, + install_audit_recorder, + install_authz, + install_settings, +) + +# ----------------------------------------------------------------------------- # +# ensure_permission +# ----------------------------------------------------------------------------- # + + +@pytest.mark.anyio +async def test_ensure_permission_noop_when_disabled(monkeypatch, fake_user): + """When AUTHZ_ENABLED=False, the helper returns without consulting the service.""" + install_settings(monkeypatch, authz_enabled=False) + service = _StubAuthorizationService(allow=False) + install_authz(monkeypatch, service) + install_audit_recorder(monkeypatch) + + await authz_guards.ensure_permission(fake_user, domain="*", obj="flow:abc", act="read") + assert service.calls == [] + + +@pytest.mark.anyio +async def test_ensure_permission_allows_when_enforce_returns_true(monkeypatch, fake_user): + """A True enforce result returns None and forwards the merged context.""" + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=True) + install_authz(monkeypatch, service) + install_audit_recorder(monkeypatch) + + await authz_guards.ensure_permission( + fake_user, + domain="*", + obj="flow:abc", + act="read", + context={"extra": "value"}, + ) + + assert len(service.calls) == 1 + call = service.calls[0] + assert call["user_id"] == fake_user.id + assert call["obj"] == "flow:abc" + assert call["act"] == "read" + assert call["context"] == {"is_superuser": False, "extra": "value"} + + +@pytest.mark.anyio +async def test_ensure_permission_caller_context_cannot_override_is_superuser(monkeypatch, fake_user): + """Caller-supplied context must not be able to overwrite the user-derived is_superuser flag. + + Regression for PR #13153 review: previously the merge was + ``{**_auth_context(user), **(context or {})}`` which let a caller forge + ``context={"is_superuser": True}`` for a non-superuser. The merged context + forwarded to ``enforce`` must always reflect the user's actual privilege. + """ + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=True) + install_authz(monkeypatch, service) + install_audit_recorder(monkeypatch) + + await authz_guards.ensure_permission( + fake_user, + domain="*", + obj="flow:abc", + act="read", + context={"is_superuser": True, "extra": "value"}, + ) + + assert len(service.calls) == 1 + forwarded_context = service.calls[0]["context"] + assert forwarded_context["is_superuser"] is False + assert forwarded_context["extra"] == "value" + + +@pytest.mark.anyio +async def test_ensure_permission_raises_403_with_non_disclosing_default(monkeypatch, fake_user): + """Default deny detail must NOT echo the resource UUID — see PR #13153 review item I2.""" + install_settings(monkeypatch, authz_enabled=True) + install_authz(monkeypatch, _StubAuthorizationService(allow=False)) + install_audit_recorder(monkeypatch) + + flow_id = uuid4() + with pytest.raises(HTTPException) as exc_info: + await authz_guards.ensure_permission(fake_user, domain="*", obj=f"flow:{flow_id}", act="write") + + assert exc_info.value.status_code == 403 + # The default message MUST NOT contain the resource UUID or the action verb — + # otherwise a caller that forgets to wrap in deny_to_404 leaks existence. + detail = exc_info.value.detail.lower() + assert str(flow_id) not in detail + assert "write" not in detail + assert "permission denied" in detail + + +@pytest.mark.anyio +async def test_ensure_permission_accepts_explicit_detail_override(monkeypatch, fake_user): + """Callers that have already verified resource existence may pass a richer message.""" + install_settings(monkeypatch, authz_enabled=True) + install_authz(monkeypatch, _StubAuthorizationService(allow=False)) + install_audit_recorder(monkeypatch) + + with pytest.raises(HTTPException) as exc_info: + await authz_guards.ensure_permission( + fake_user, + domain="*", + obj="flow:abc", + act="write", + detail="Cannot edit a published flow.", + ) + + assert exc_info.value.detail == "Cannot edit a published flow." + + +@pytest.mark.anyio +async def test_ensure_permission_writes_audit_on_allow(monkeypatch, fake_user): + """Allow path schedules an audit row with result='allow'.""" + install_settings(monkeypatch, authz_enabled=True) + install_authz(monkeypatch, _StubAuthorizationService(allow=True)) + audit_calls = install_audit_recorder(monkeypatch) + + await authz_guards.ensure_permission(fake_user, domain="project:1", obj="flow:abc", act="read") + + assert len(audit_calls) == 1 + assert audit_calls[0]["result"] == "allow" + assert audit_calls[0]["obj"] == "flow:abc" + assert audit_calls[0]["action"] == "flow:read" + + +@pytest.mark.anyio +async def test_ensure_permission_writes_audit_on_deny(monkeypatch, fake_user): + """Deny path schedules an audit row with result='deny' before the 403 raises.""" + install_settings(monkeypatch, authz_enabled=True) + install_authz(monkeypatch, _StubAuthorizationService(allow=False)) + audit_calls = install_audit_recorder(monkeypatch) + + with pytest.raises(HTTPException): + await authz_guards.ensure_permission(fake_user, domain="*", obj="flow:abc", act="delete") + + assert len(audit_calls) == 1 + assert audit_calls[0]["result"] == "deny" + + +# ----------------------------------------------------------------------------- # +# ensure_flow_permission — enum coercion, domain, owner override +# ----------------------------------------------------------------------------- # + + +@pytest.mark.anyio +async def test_ensure_flow_permission_accepts_enum(monkeypatch, fake_user): + """A FlowAction enum is coerced to its string value before enforce.""" + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=True) + install_authz(monkeypatch, service) + install_audit_recorder(monkeypatch) + + flow_id = uuid4() + await authz_guards.ensure_flow_permission(fake_user, FlowAction.READ, flow_id=flow_id) + + assert service.calls[0]["act"] == "read" + assert service.calls[0]["obj"] == f"flow:{flow_id}" + + +@pytest.mark.anyio +async def test_ensure_flow_permission_accepts_string_for_backcompat(monkeypatch, fake_user): + """Bare string actions still work — gradual migration is allowed.""" + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=True) + install_authz(monkeypatch, service) + install_audit_recorder(monkeypatch) + + await authz_guards.ensure_flow_permission(fake_user, "write", flow_id=uuid4()) + assert service.calls[0]["act"] == "write" + + +@pytest.mark.anyio +async def test_ensure_flow_permission_computes_workspace_domain(monkeypatch, fake_user): + """workspace_id is mapped to a workspace: domain string and forwarded in context.""" + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=True) + install_authz(monkeypatch, service) + install_audit_recorder(monkeypatch) + + workspace_id = uuid4() + await authz_guards.ensure_flow_permission( + fake_user, + FlowAction.READ, + flow_id=uuid4(), + flow_user_id=uuid4(), + workspace_id=workspace_id, + ) + + assert service.calls[0]["domain"] == f"workspace:{workspace_id}" + assert service.calls[0]["context"]["workspace_id"] == workspace_id + + +@pytest.mark.anyio +async def test_ensure_flow_permission_falls_back_to_project_domain(monkeypatch, fake_user): + """Without workspace_id, folder_id maps to a project: domain string.""" + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=True) + install_authz(monkeypatch, service) + install_audit_recorder(monkeypatch) + + folder_id = uuid4() + await authz_guards.ensure_flow_permission( + fake_user, + FlowAction.READ, + flow_id=uuid4(), + flow_user_id=uuid4(), + folder_id=folder_id, + ) + + assert service.calls[0]["domain"] == f"project:{folder_id}" + assert service.calls[0]["context"]["folder_id"] == folder_id + assert service.calls[0]["context"]["workspace_id"] is None + + +@pytest.mark.anyio +async def test_ensure_flow_permission_project_beats_workspace(monkeypatch, fake_user): + """Project domain wins when both workspace_id and folder_id are set.""" + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=True) + install_authz(monkeypatch, service) + install_audit_recorder(monkeypatch) + + workspace_id = uuid4() + folder_id = uuid4() + await authz_guards.ensure_flow_permission( + fake_user, + FlowAction.READ, + flow_id=uuid4(), + flow_user_id=uuid4(), + workspace_id=workspace_id, + folder_id=folder_id, + ) + + assert service.calls[0]["domain"] == f"project:{folder_id}" + # workspace_id is still passed in context so the plugin can use it for ABAC matchers. + assert service.calls[0]["context"]["folder_id"] == folder_id + assert service.calls[0]["context"]["workspace_id"] == workspace_id + + +@pytest.mark.anyio +async def test_ensure_flow_permission_wildcard_domain_when_neither_set(monkeypatch, fake_user): + """With neither workspace_id nor folder_id, domain falls back to '*'.""" + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=True) + install_authz(monkeypatch, service) + install_audit_recorder(monkeypatch) + + await authz_guards.ensure_flow_permission(fake_user, FlowAction.CREATE) + assert service.calls[0]["domain"] == "*" + + +@pytest.mark.anyio +async def test_owner_override_skips_enforce(monkeypatch, fake_user): + """A flow owner short-circuits the enforce call entirely.""" + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=False) # Would deny if asked. + install_authz(monkeypatch, service) + audit_calls = install_audit_recorder(monkeypatch) + + await authz_guards.ensure_flow_permission( + fake_user, + FlowAction.DELETE, + flow_id=uuid4(), + flow_user_id=fake_user.id, + workspace_id=uuid4(), + ) + + assert service.calls == [] + assert len(audit_calls) == 1 + assert audit_calls[0]["result"] == "owner_override" + + +@pytest.mark.anyio +async def test_owner_override_audits_even_when_authz_disabled(monkeypatch, fake_user): + """Owner override on a disabled-authz install still writes an audit row. + + Audit is now gated only on ``AUTHZ_AUDIT_ENABLED`` (the audit recorder + here intercepts at that level), so operators can observe traffic ahead of + flipping ``AUTHZ_ENABLED``. ``install_settings`` defaults the audit flag + to False so we explicitly enable it here. + """ + install_settings(monkeypatch, authz_enabled=False, audit_enabled=True) + install_authz(monkeypatch, _StubAuthorizationService(allow=False)) + audit_calls = install_audit_recorder(monkeypatch) + + await authz_guards.ensure_flow_permission( + fake_user, + FlowAction.WRITE, + flow_id=uuid4(), + flow_user_id=fake_user.id, + ) + assert len(audit_calls) == 1 + assert audit_calls[0]["result"] == "owner_override" + + +@pytest.mark.anyio +async def test_non_owner_falls_through_to_enforce(monkeypatch, fake_user): + """When flow_user_id differs from current user, enforce is called normally.""" + install_settings(monkeypatch, authz_enabled=True) + install_authz(monkeypatch, _StubAuthorizationService(allow=False)) + install_audit_recorder(monkeypatch) + + with pytest.raises(HTTPException): + await authz_guards.ensure_flow_permission( + fake_user, + FlowAction.WRITE, + flow_id=uuid4(), + flow_user_id=uuid4(), + ) + + +# ----------------------------------------------------------------------------- # +# ensure_project_permission +# ----------------------------------------------------------------------------- # + + +@pytest.mark.anyio +async def test_ensure_project_permission_accepts_enum(monkeypatch, fake_user): + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=True) + install_authz(monkeypatch, service) + install_audit_recorder(monkeypatch) + + project_id = uuid4() + await authz_guards.ensure_project_permission(fake_user, ProjectAction.READ, project_id=project_id) + assert service.calls[0]["act"] == "read" + assert service.calls[0]["obj"] == f"project:{project_id}" + + +@pytest.mark.anyio +async def test_ensure_project_permission_uses_workspace_domain(monkeypatch, fake_user): + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=True) + install_authz(monkeypatch, service) + install_audit_recorder(monkeypatch) + + workspace_id = uuid4() + await authz_guards.ensure_project_permission( + fake_user, ProjectAction.WRITE, project_id=uuid4(), project_user_id=uuid4(), workspace_id=workspace_id + ) + assert service.calls[0]["domain"] == f"workspace:{workspace_id}" + assert service.calls[0]["context"]["workspace_id"] == workspace_id + + +@pytest.mark.anyio +async def test_project_owner_override_skips_enforce(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_guards.ensure_project_permission( + fake_user, + ProjectAction.DELETE, + project_id=uuid4(), + project_user_id=fake_user.id, + workspace_id=uuid4(), + ) + + assert service.calls == [] + assert len(audit_calls) == 1 + assert audit_calls[0]["result"] == "owner_override" + + +@pytest.mark.anyio +async def test_ensure_project_permission_create_uses_wildcard_obj(monkeypatch, fake_user): + """ProjectAction.CREATE without a project_id targets ``project:*`` (workspace-scoped create).""" + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=True) + install_authz(monkeypatch, service) + install_audit_recorder(monkeypatch) + + await authz_guards.ensure_project_permission(fake_user, ProjectAction.CREATE) + assert service.calls[0]["obj"] == "project:*" + assert service.calls[0]["act"] == "create" + + +# ----------------------------------------------------------------------------- # +# ensure_deployment_permission +# ----------------------------------------------------------------------------- # + + +@pytest.mark.anyio +async def test_ensure_deployment_permission_uses_project_domain(monkeypatch, fake_user): + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=True) + install_authz(monkeypatch, service) + install_audit_recorder(monkeypatch) + + project_id = uuid4() + await authz_guards.ensure_deployment_permission( + fake_user, + DeploymentAction.READ, + deployment_id=uuid4(), + deployment_user_id=uuid4(), + project_id=project_id, + ) + + assert service.calls[0]["domain"] == f"project:{project_id}" + assert service.calls[0]["context"]["project_id"] == project_id + + +@pytest.mark.anyio +async def test_deployment_owner_override_skips_enforce(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_guards.ensure_deployment_permission( + fake_user, + DeploymentAction.DELETE, + deployment_id=uuid4(), + deployment_user_id=fake_user.id, + project_id=uuid4(), + ) + + assert service.calls == [] + 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_uses_kb_id_object_slug(monkeypatch, fake_user): + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=True) + install_authz(monkeypatch, service) + install_audit_recorder(monkeypatch) + + kb_id = uuid4() + await authz_guards.ensure_knowledge_base_permission( + fake_user, + KnowledgeBaseAction.READ, + kb_id=kb_id, + kb_name="my-kb", + kb_user_id=uuid4(), + ) + + assert service.calls[0]["obj"] == f"knowledge_base:{kb_id}" + assert service.calls[0]["act"] == "read" + # ``kb_name`` is forwarded in the context for debugging. + assert service.calls[0]["context"]["kb_name"] == "my-kb" + assert service.calls[0]["context"]["kb_id"] == kb_id + + +@pytest.mark.anyio +async def test_kb_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_guards.ensure_knowledge_base_permission( + fake_user, + KnowledgeBaseAction.DELETE, + kb_id=uuid4(), + 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): + 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_guards.ensure_knowledge_base_permission( + fake_user, + KnowledgeBaseAction.DELETE, + kb_id=uuid4(), + kb_name="someone-elses", + kb_user_id=uuid4(), + ) + assert exc.value.status_code == 403 + + +@pytest.mark.anyio +async def test_kb_permission_create_uses_wildcard_slug(monkeypatch, fake_user): + """Without a kb_id (create flow) the owner override fires; no enforce call.""" + install_settings(monkeypatch, authz_enabled=True) + service = _StubAuthorizationService(allow=True) + install_authz(monkeypatch, service) + install_audit_recorder(monkeypatch) + + await authz_guards.ensure_knowledge_base_permission( + fake_user, + KnowledgeBaseAction.CREATE, + kb_user_id=fake_user.id, + ) + + assert service.calls == [] + + +# ----------------------------------------------------------------------------- # +# 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_guards.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_guards.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_guards.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_guards.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_guards.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/backend/tests/unit/services/authorization/test_utils.py b/src/backend/tests/unit/services/authorization/test_utils.py deleted file mode 100644 index a958ba4b68..0000000000 --- a/src/backend/tests/unit/services/authorization/test_utils.py +++ /dev/null @@ -1,1004 +0,0 @@ -"""Tests for ensure_permission, ensure_flow_permission, audit_decision, filter_visible_resources.""" - -from __future__ import annotations - -from types import SimpleNamespace -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, - FileAction, - FlowAction, - KnowledgeBaseAction, - ProjectAction, - ShareAction, - VariableAction, -) - - -class _StubAuthorizationService: - """Minimal stand-in for BaseAuthorizationService that records calls.""" - - def __init__(self, *, allow: bool = True, batch_results: list[bool] | None = None) -> None: - self.allow = allow - self.batch_results = batch_results - self.calls: list[dict] = [] - self.batch_calls: list[dict] = [] - - async def enforce(self, **kwargs) -> bool: - self.calls.append(kwargs) - return self.allow - - async def batch_enforce(self, **kwargs) -> list[bool]: - self.batch_calls.append(kwargs) - if self.batch_results is not None: - return self.batch_results - return [self.allow] * len(kwargs.get("requests", [])) - - -@pytest.fixture -def fake_user(): - """Build a non-superuser user object compatible with ensure_permission.""" - return SimpleNamespace(id=uuid4(), is_superuser=False) - - -@pytest.fixture -def fake_superuser(): - """Build a superuser user object compatible with ensure_permission.""" - return SimpleNamespace(id=uuid4(), is_superuser=True) - - -def _install_settings(monkeypatch, *, authz_enabled: bool, audit_enabled: bool = False) -> None: - settings = SimpleNamespace( - auth_settings=SimpleNamespace( - AUTHZ_ENABLED=authz_enabled, - AUTHZ_AUDIT_ENABLED=audit_enabled, - ), - ) - monkeypatch.setattr(authz_utils, "get_settings_service", lambda: settings) - - -def _install_authz(monkeypatch, service: _StubAuthorizationService) -> None: - monkeypatch.setattr(authz_utils, "get_authorization_service", lambda: service) - - -def _install_audit_recorder(monkeypatch) -> list[dict]: - """Replace audit_decision with a recorder so tests can assert audit writes.""" - calls: list[dict] = [] - - async def _recorder(**kwargs): - calls.append(kwargs) - - monkeypatch.setattr(authz_utils, "audit_decision", _recorder) - return calls - - -# ----------------------------------------------------------------------------- # -# ensure_permission -# ----------------------------------------------------------------------------- # - - -@pytest.mark.anyio -async def test_ensure_permission_noop_when_disabled(monkeypatch, fake_user): - """When AUTHZ_ENABLED=False, the helper returns without consulting the service.""" - _install_settings(monkeypatch, authz_enabled=False) - service = _StubAuthorizationService(allow=False) - _install_authz(monkeypatch, service) - _install_audit_recorder(monkeypatch) - - await authz_utils.ensure_permission(fake_user, domain="*", obj="flow:abc", act="read") - assert service.calls == [] - - -@pytest.mark.anyio -async def test_ensure_permission_allows_when_enforce_returns_true(monkeypatch, fake_user): - """A True enforce result returns None and forwards the merged context.""" - _install_settings(monkeypatch, authz_enabled=True) - service = _StubAuthorizationService(allow=True) - _install_authz(monkeypatch, service) - _install_audit_recorder(monkeypatch) - - await authz_utils.ensure_permission( - fake_user, - domain="*", - obj="flow:abc", - act="read", - context={"extra": "value"}, - ) - - assert len(service.calls) == 1 - call = service.calls[0] - assert call["user_id"] == fake_user.id - assert call["obj"] == "flow:abc" - assert call["act"] == "read" - assert call["context"] == {"is_superuser": False, "extra": "value"} - - -@pytest.mark.anyio -async def test_ensure_permission_caller_context_cannot_override_is_superuser(monkeypatch, fake_user): - """Caller-supplied context must not be able to overwrite the user-derived is_superuser flag. - - Regression for PR #13153 review: previously the merge was - ``{**_auth_context(user), **(context or {})}`` which let a caller forge - ``context={"is_superuser": True}`` for a non-superuser. The merged context - forwarded to ``enforce`` must always reflect the user's actual privilege. - """ - _install_settings(monkeypatch, authz_enabled=True) - service = _StubAuthorizationService(allow=True) - _install_authz(monkeypatch, service) - _install_audit_recorder(monkeypatch) - - await authz_utils.ensure_permission( - fake_user, - domain="*", - obj="flow:abc", - act="read", - context={"is_superuser": True, "extra": "value"}, - ) - - assert len(service.calls) == 1 - forwarded_context = service.calls[0]["context"] - assert forwarded_context["is_superuser"] is False - assert forwarded_context["extra"] == "value" - - -@pytest.mark.anyio -async def test_ensure_permission_raises_403_when_denied(monkeypatch, fake_user): - """A False enforce result raises HTTP 403 with a descriptive message.""" - _install_settings(monkeypatch, authz_enabled=True) - _install_authz(monkeypatch, _StubAuthorizationService(allow=False)) - _install_audit_recorder(monkeypatch) - - with pytest.raises(HTTPException) as exc_info: - await authz_utils.ensure_permission(fake_user, domain="*", obj="flow:abc", act="write") - assert exc_info.value.status_code == 403 - assert "write" in exc_info.value.detail - - -@pytest.mark.anyio -async def test_ensure_permission_writes_audit_on_allow(monkeypatch, fake_user): - """Allow path schedules an audit row with result='allow'.""" - _install_settings(monkeypatch, authz_enabled=True) - _install_authz(monkeypatch, _StubAuthorizationService(allow=True)) - audit_calls = _install_audit_recorder(monkeypatch) - - await authz_utils.ensure_permission(fake_user, domain="project:1", obj="flow:abc", act="read") - - assert len(audit_calls) == 1 - assert audit_calls[0]["result"] == "allow" - assert audit_calls[0]["obj"] == "flow:abc" - assert audit_calls[0]["action"] == "flow:read" - - -@pytest.mark.anyio -async def test_ensure_permission_writes_audit_on_deny(monkeypatch, fake_user): - """Deny path schedules an audit row with result='deny' before the 403 raises.""" - _install_settings(monkeypatch, authz_enabled=True) - _install_authz(monkeypatch, _StubAuthorizationService(allow=False)) - audit_calls = _install_audit_recorder(monkeypatch) - - with pytest.raises(HTTPException): - await authz_utils.ensure_permission(fake_user, domain="*", obj="flow:abc", act="delete") - - assert len(audit_calls) == 1 - assert audit_calls[0]["result"] == "deny" - - -# ----------------------------------------------------------------------------- # -# ensure_flow_permission — enum coercion, domain, owner override -# ----------------------------------------------------------------------------- # - - -@pytest.mark.anyio -async def test_ensure_flow_permission_accepts_enum(monkeypatch, fake_user): - """A FlowAction enum is coerced to its string value before enforce.""" - _install_settings(monkeypatch, authz_enabled=True) - service = _StubAuthorizationService(allow=True) - _install_authz(monkeypatch, service) - _install_audit_recorder(monkeypatch) - - flow_id = uuid4() - await authz_utils.ensure_flow_permission(fake_user, FlowAction.READ, flow_id=flow_id) - - assert service.calls[0]["act"] == "read" - assert service.calls[0]["obj"] == f"flow:{flow_id}" - - -@pytest.mark.anyio -async def test_ensure_flow_permission_accepts_string_for_backcompat(monkeypatch, fake_user): - """Bare string actions still work — gradual migration is allowed.""" - _install_settings(monkeypatch, authz_enabled=True) - service = _StubAuthorizationService(allow=True) - _install_authz(monkeypatch, service) - _install_audit_recorder(monkeypatch) - - await authz_utils.ensure_flow_permission(fake_user, "write", flow_id=uuid4()) - assert service.calls[0]["act"] == "write" - - -@pytest.mark.anyio -async def test_ensure_flow_permission_computes_workspace_domain(monkeypatch, fake_user): - """workspace_id is mapped to a workspace: domain string and forwarded in context.""" - _install_settings(monkeypatch, authz_enabled=True) - service = _StubAuthorizationService(allow=True) - _install_authz(monkeypatch, service) - _install_audit_recorder(monkeypatch) - - workspace_id = uuid4() - await authz_utils.ensure_flow_permission( - fake_user, - FlowAction.READ, - flow_id=uuid4(), - flow_user_id=uuid4(), - workspace_id=workspace_id, - ) - - assert service.calls[0]["domain"] == f"workspace:{workspace_id}" - assert service.calls[0]["context"]["workspace_id"] == workspace_id - - -@pytest.mark.anyio -async def test_ensure_flow_permission_falls_back_to_project_domain(monkeypatch, fake_user): - """Without workspace_id, folder_id maps to a project: domain string.""" - _install_settings(monkeypatch, authz_enabled=True) - service = _StubAuthorizationService(allow=True) - _install_authz(monkeypatch, service) - _install_audit_recorder(monkeypatch) - - folder_id = uuid4() - await authz_utils.ensure_flow_permission( - fake_user, - FlowAction.READ, - flow_id=uuid4(), - flow_user_id=uuid4(), - folder_id=folder_id, - ) - - assert service.calls[0]["domain"] == f"project:{folder_id}" - assert service.calls[0]["context"]["folder_id"] == folder_id - assert service.calls[0]["context"]["workspace_id"] is None - - -@pytest.mark.anyio -async def test_ensure_flow_permission_project_beats_workspace(monkeypatch, fake_user): - """Project domain wins when both workspace_id and folder_id are set.""" - _install_settings(monkeypatch, authz_enabled=True) - service = _StubAuthorizationService(allow=True) - _install_authz(monkeypatch, service) - _install_audit_recorder(monkeypatch) - - workspace_id = uuid4() - folder_id = uuid4() - await authz_utils.ensure_flow_permission( - fake_user, - FlowAction.READ, - flow_id=uuid4(), - flow_user_id=uuid4(), - workspace_id=workspace_id, - folder_id=folder_id, - ) - - assert service.calls[0]["domain"] == f"project:{folder_id}" - # workspace_id is still passed in context so the plugin can use it for ABAC matchers. - assert service.calls[0]["context"]["folder_id"] == folder_id - assert service.calls[0]["context"]["workspace_id"] == workspace_id - - -@pytest.mark.anyio -async def test_ensure_flow_permission_wildcard_domain_when_neither_set(monkeypatch, fake_user): - """With neither workspace_id nor folder_id, domain falls back to '*'.""" - _install_settings(monkeypatch, authz_enabled=True) - service = _StubAuthorizationService(allow=True) - _install_authz(monkeypatch, service) - _install_audit_recorder(monkeypatch) - - await authz_utils.ensure_flow_permission(fake_user, FlowAction.CREATE) - assert service.calls[0]["domain"] == "*" - - -def test_resolve_casbin_domain_precedence(): - """Domain precedence: project > workspace > '*'.""" - ws, scope = uuid4(), uuid4() - assert authz_utils._resolve_casbin_domain(workspace_id=ws, scope_id=scope) == f"project:{scope}" - assert authz_utils._resolve_casbin_domain(workspace_id=ws, scope_id=None) == f"workspace:{ws}" - assert authz_utils._resolve_casbin_domain(workspace_id=None, scope_id=scope) == f"project:{scope}" - assert authz_utils._resolve_casbin_domain(workspace_id=None, scope_id=None) == "*" - # Backward-compatible alias still resolves to the same function. - assert authz_utils._resolve_flow_domain is authz_utils._resolve_casbin_domain - - -@pytest.mark.anyio -async def test_owner_override_skips_enforce(monkeypatch, fake_user): - """A flow owner short-circuits the enforce call entirely.""" - _install_settings(monkeypatch, authz_enabled=True) - service = _StubAuthorizationService(allow=False) # Would deny if asked. - _install_authz(monkeypatch, service) - audit_calls = _install_audit_recorder(monkeypatch) - - await authz_utils.ensure_flow_permission( - fake_user, - FlowAction.DELETE, - flow_id=uuid4(), - flow_user_id=fake_user.id, - workspace_id=uuid4(), - ) - - assert service.calls == [] - assert len(audit_calls) == 1 - assert audit_calls[0]["result"] == "owner_override" - - -@pytest.mark.anyio -async def test_owner_override_audits_even_when_authz_disabled(monkeypatch, fake_user): - """Owner override on a disabled-authz install still writes an audit row. - - Audit is now gated only on ``AUTHZ_AUDIT_ENABLED`` (the audit recorder - here intercepts at that level), so operators can observe traffic ahead of - flipping ``AUTHZ_ENABLED``. ``_install_settings`` defaults the audit flag - to False so we explicitly enable it here. - """ - _install_settings(monkeypatch, authz_enabled=False, audit_enabled=True) - _install_authz(monkeypatch, _StubAuthorizationService(allow=False)) - audit_calls = _install_audit_recorder(monkeypatch) - - await authz_utils.ensure_flow_permission( - fake_user, - FlowAction.WRITE, - flow_id=uuid4(), - flow_user_id=fake_user.id, - ) - assert len(audit_calls) == 1 - assert audit_calls[0]["result"] == "owner_override" - - -@pytest.mark.anyio -async def test_non_owner_falls_through_to_enforce(monkeypatch, fake_user): - """When flow_user_id differs from current user, enforce is called normally.""" - _install_settings(monkeypatch, authz_enabled=True) - _install_authz(monkeypatch, _StubAuthorizationService(allow=False)) - _install_audit_recorder(monkeypatch) - - with pytest.raises(HTTPException): - await authz_utils.ensure_flow_permission( - fake_user, - FlowAction.WRITE, - flow_id=uuid4(), - flow_user_id=uuid4(), - ) - - -# ----------------------------------------------------------------------------- # -# ensure_project_permission -# ----------------------------------------------------------------------------- # - - -@pytest.mark.anyio -async def test_ensure_project_permission_accepts_enum(monkeypatch, fake_user): - """A ProjectAction enum is coerced to its string value before enforce.""" - _install_settings(monkeypatch, authz_enabled=True) - service = _StubAuthorizationService(allow=True) - _install_authz(monkeypatch, service) - _install_audit_recorder(monkeypatch) - - project_id = uuid4() - await authz_utils.ensure_project_permission(fake_user, ProjectAction.READ, project_id=project_id) - assert service.calls[0]["act"] == "read" - assert service.calls[0]["obj"] == f"project:{project_id}" - - -@pytest.mark.anyio -async def test_ensure_project_permission_uses_workspace_domain(monkeypatch, fake_user): - """workspace_id is mapped to a workspace: domain string and forwarded in context.""" - _install_settings(monkeypatch, authz_enabled=True) - service = _StubAuthorizationService(allow=True) - _install_authz(monkeypatch, service) - _install_audit_recorder(monkeypatch) - - workspace_id = uuid4() - await authz_utils.ensure_project_permission( - fake_user, ProjectAction.WRITE, project_id=uuid4(), project_user_id=uuid4(), workspace_id=workspace_id - ) - assert service.calls[0]["domain"] == f"workspace:{workspace_id}" - assert service.calls[0]["context"]["workspace_id"] == workspace_id - - -@pytest.mark.anyio -async def test_project_owner_override_skips_enforce(monkeypatch, fake_user): - """A project owner short-circuits the enforce call entirely.""" - _install_settings(monkeypatch, authz_enabled=True) - service = _StubAuthorizationService(allow=False) # Would deny if asked. - _install_authz(monkeypatch, service) - audit_calls = _install_audit_recorder(monkeypatch) - - await authz_utils.ensure_project_permission( - fake_user, - ProjectAction.DELETE, - project_id=uuid4(), - project_user_id=fake_user.id, - workspace_id=uuid4(), - ) - - assert service.calls == [] - assert len(audit_calls) == 1 - assert audit_calls[0]["result"] == "owner_override" - - -@pytest.mark.anyio -async def test_ensure_project_permission_create_uses_wildcard_obj(monkeypatch, fake_user): - """ProjectAction.CREATE without a project_id targets ``project:*`` (workspace-scoped create).""" - _install_settings(monkeypatch, authz_enabled=True) - service = _StubAuthorizationService(allow=True) - _install_authz(monkeypatch, service) - _install_audit_recorder(monkeypatch) - - await authz_utils.ensure_project_permission(fake_user, ProjectAction.CREATE) - assert service.calls[0]["obj"] == "project:*" - assert service.calls[0]["act"] == "create" - - -# ----------------------------------------------------------------------------- # -# audit_decision -# ----------------------------------------------------------------------------- # - - -def test_split_obj_parses_uuid_suffix(): - """flow: splits into ('flow', UUID).""" - flow_id = uuid4() - resource_type, resource_id = authz_utils._split_obj(f"flow:{flow_id}") - assert resource_type == "flow" - assert resource_id == flow_id - - -def test_split_obj_wildcard_returns_none_id(): - """flow:* keeps resource_type but emits None for resource_id.""" - resource_type, resource_id = authz_utils._split_obj("flow:*") - assert resource_type == "flow" - assert resource_id is None - - -def test_split_obj_malformed_returns_nones(): - """A key without a colon returns (None, None).""" - assert authz_utils._split_obj("nothing") == (None, None) - - -def test_split_obj_non_uuid_suffix_returns_none_id(): - """A non-UUID suffix is treated as a wildcard for the resource_id field.""" - resource_type, resource_id = authz_utils._split_obj("flow:not-a-uuid") - assert resource_type == "flow" - assert resource_id is None - - -@pytest.fixture -def patched_audit_flush(monkeypatch): - """Replace _flush_audit_batch with a recorder so we exercise the writer without touching the DB.""" - flushed: list[list[object]] = [] - - async def _record(batch): - flushed.append(list(batch)) - - monkeypatch.setattr(authz_utils, "_flush_audit_batch", _record) - return flushed - - -async def _reset_audit_pipeline() -> None: - """Best-effort teardown so each test starts with a clean audit pipeline.""" - await authz_utils.drain_pending_audit_writes(timeout=0.5) - authz_utils._audit_queue = None - authz_utils._audit_queue_loop = None - authz_utils._audit_writer_task = None - authz_utils._audit_dropped_count = 0 - authz_utils._pending_audit_tasks.clear() - - -@pytest.mark.anyio -async def test_audit_decision_runs_when_authz_disabled_but_audit_on(monkeypatch, patched_audit_flush): - """Audit is independent of enforcement. - - Previously ``audit_decision`` short-circuited when ``AUTHZ_ENABLED=False``, - which meant share CRUD writes left no audit trail on default installs. The - new contract gates only on ``AUTHZ_AUDIT_ENABLED`` so operators can - observe traffic before flipping enforcement on. - """ - await _reset_audit_pipeline() - _install_settings(monkeypatch, authz_enabled=False, audit_enabled=True) - - await authz_utils.audit_decision(user_id=uuid4(), action="flow:read", obj="flow:x", result="allow") - await authz_utils.drain_pending_audit_writes(timeout=1.0) - - assert sum(len(b) for b in patched_audit_flush) == 1 - - -@pytest.mark.anyio -async def test_audit_decision_noop_when_audit_disabled(monkeypatch, patched_audit_flush): - """AUTHZ_AUDIT_ENABLED=False suppresses audit writes.""" - await _reset_audit_pipeline() - _install_settings(monkeypatch, authz_enabled=True, audit_enabled=False) - - await authz_utils.audit_decision(user_id=uuid4(), action="flow:read", obj="flow:x", result="allow") - await authz_utils.drain_pending_audit_writes(timeout=0.5) - - assert patched_audit_flush == [] - - -@pytest.mark.anyio -async def test_audit_decision_enqueues_when_enabled(monkeypatch, patched_audit_flush): - """When both flags are on, audit_decision enqueues an entry that the background writer flushes.""" - await _reset_audit_pipeline() - _install_settings(monkeypatch, authz_enabled=True, audit_enabled=True) - - user_id = uuid4() - await authz_utils.audit_decision(user_id=user_id, action="flow:read", obj="flow:abc", result="allow") - # Drain forces the writer to flush before we inspect. - await authz_utils.drain_pending_audit_writes(timeout=1.0) - - assert len(patched_audit_flush) == 1 - batch = patched_audit_flush[0] - assert len(batch) == 1 - entry = batch[0] - assert entry.user_id == user_id - assert entry.action == "flow:read" - assert entry.obj == "flow:abc" - assert entry.result == "allow" - - -@pytest.mark.anyio -async def test_audit_decision_batches_multiple_entries(monkeypatch, patched_audit_flush): - """Multiple concurrent audit_decision calls coalesce into a single DB batch. - - This is the contract we want — the writer should pull every entry already - in the queue when it wakes up, so we make N decisions before yielding and - expect ONE batch of N rows, not N separate session_scope opens. - """ - await _reset_audit_pipeline() - _install_settings(monkeypatch, authz_enabled=True, audit_enabled=True) - - for _ in range(5): - await authz_utils.audit_decision(user_id=uuid4(), action="flow:read", obj="flow:x", result="allow") - - await authz_utils.drain_pending_audit_writes(timeout=1.0) - - total_rows = sum(len(batch) for batch in patched_audit_flush) - assert total_rows == 5 - # All entries are emitted before the first await, so they should land in a single batch. - assert len(patched_audit_flush) == 1, ( - f"Expected 1 batch of 5 rows, got {len(patched_audit_flush)} batches " - f"with sizes {[len(b) for b in patched_audit_flush]}" - ) - - -@pytest.mark.anyio -async def test_drain_pending_audit_writes_is_safe_when_idle(): - """drain_pending_audit_writes is a no-op when no audit traffic has run.""" - await _reset_audit_pipeline() - # Must not raise. - await authz_utils.drain_pending_audit_writes(timeout=0.1) - - -@pytest.mark.anyio -async def test_ensure_permission_fails_closed_on_plugin_exception(monkeypatch, fake_user): - """If the authz plugin raises, ensure_permission must deny (403), not bubble 500.""" - _install_settings(monkeypatch, authz_enabled=True, audit_enabled=False) - - class _BrokenPlugin: - async def enforce(self, **_kwargs): - msg = "policy store down" - raise RuntimeError(msg) - - async def batch_enforce(self, **_kwargs): - return [] - - monkeypatch.setattr(authz_utils, "get_authorization_service", lambda: _BrokenPlugin()) - captured = _install_audit_recorder(monkeypatch) - - with pytest.raises(HTTPException) as excinfo: - await authz_utils.ensure_permission(fake_user, domain="*", obj="flow:abc", act="read") - - assert excinfo.value.status_code == 403, "Plugin exceptions must fail closed (deny), not 500" - # The deny path must still emit an audit row so the operator can see the failure. - assert captured, "Plugin exception must still produce an audit row" - assert captured[0]["result"] == "deny" - assert "error" in captured[0]["details"] - - -# ----------------------------------------------------------------------------- # -# filter_visible_resources -# ----------------------------------------------------------------------------- # - - -@pytest.mark.anyio -async def test_filter_visible_resources_noop_when_disabled(monkeypatch, fake_user): - """No batch_enforce call when AUTHZ_ENABLED=False; returns input unchanged.""" - _install_settings(monkeypatch, authz_enabled=False) - service = _StubAuthorizationService(allow=False) - _install_authz(monkeypatch, service) - - candidates = [SimpleNamespace(id=uuid4()) for _ in range(3)] - result = await authz_utils.filter_visible_resources( - fake_user, - resource_type="flow", - candidates=candidates, - ) - assert result == candidates - assert service.batch_calls == [] - - -@pytest.mark.anyio -async def test_filter_visible_resources_empty_returns_empty(monkeypatch, fake_user): - """An empty candidates list is returned unchanged without contacting the service.""" - _install_settings(monkeypatch, authz_enabled=True) - service = _StubAuthorizationService(allow=True) - _install_authz(monkeypatch, service) - - result = await authz_utils.filter_visible_resources(fake_user, resource_type="flow", candidates=[]) - assert result == [] - assert service.batch_calls == [] - - -@pytest.mark.anyio -async def test_filter_visible_resources_filters_via_batch_enforce(monkeypatch, fake_user): - """When AUTHZ_ENABLED=True, batch_enforce results filter the candidate list.""" - _install_settings(monkeypatch, authz_enabled=True) - candidates = [SimpleNamespace(id=uuid4()) for _ in range(3)] - service = _StubAuthorizationService(batch_results=[True, False, True]) - _install_authz(monkeypatch, service) - - result = await authz_utils.filter_visible_resources( - fake_user, - resource_type="flow", - candidates=candidates, - ) - - assert result == [candidates[0], candidates[2]] - assert service.batch_calls[0]["requests"] == [ - (f"flow:{candidates[0].id}", "read"), - (f"flow:{candidates[1].id}", "read"), - (f"flow:{candidates[2].id}", "read"), - ] - - -@pytest.mark.anyio -async def test_filter_visible_resources_accepts_custom_key(monkeypatch, fake_user): - """A custom key extractor lets callers filter non-id-bearing items.""" - _install_settings(monkeypatch, authz_enabled=True) - items = [{"resource_id": uuid4()}, {"resource_id": uuid4()}] - service = _StubAuthorizationService(batch_results=[False, True]) - _install_authz(monkeypatch, service) - - result = await authz_utils.filter_visible_resources( - fake_user, - resource_type="project", - candidates=items, - key=lambda r: r["resource_id"], - act=FlowAction.WRITE, - ) - - assert result == [items[1]] - assert service.batch_calls[0]["requests"][0][1] == "write" - - -@pytest.mark.anyio -async def test_filter_visible_resources_groups_by_extracted_domain(monkeypatch, fake_user): - """With ``domain_extractor`` set, batch_enforce is called once per unique domain. - - Each call sees only the candidates that resolved to that domain, so the - authorization plugin evaluates each candidate against the right policy tuple - (the single-domain default would force every candidate through the same - wildcard domain, hiding project-scoped grants). - """ - _install_settings(monkeypatch, authz_enabled=True) - workspace_a = uuid4() - workspace_b = uuid4() - - items = [ - SimpleNamespace(id=uuid4(), workspace_id=workspace_a, folder_id=None), - SimpleNamespace(id=uuid4(), workspace_id=workspace_b, folder_id=None), - SimpleNamespace(id=uuid4(), workspace_id=workspace_a, folder_id=None), - ] - - # Deny everything in workspace_b, allow everything in workspace_a. - class _DomainAwareStub: - def __init__(self) -> None: - self.batch_calls: list[dict] = [] - - async def batch_enforce(self, **kwargs) -> list[bool]: - self.batch_calls.append(kwargs) - allowed = kwargs["domain"] != f"workspace:{workspace_b}" - return [allowed] * len(kwargs["requests"]) - - service = _DomainAwareStub() - _install_authz(monkeypatch, service) - - result = await authz_utils.filter_visible_resources( - fake_user, - resource_type="project", - candidates=items, - domain_extractor=lambda project: authz_utils._resolve_casbin_domain(project.workspace_id, None), - act=FlowAction.READ, - ) - - # Two calls — one per unique domain. - domains_called = {call["domain"] for call in service.batch_calls} - assert domains_called == {f"workspace:{workspace_a}", f"workspace:{workspace_b}"} - - # Output preserves the original order, with workspace_b's item dropped. - assert result == [items[0], items[2]] - - -@pytest.mark.anyio -async def test_filter_visible_resources_owner_override_skips_enforcer(monkeypatch, fake_user): - """Items owned by the caller are force-included without consulting the enforcer. - - Mirrors the owner-override short-circuit in ``_ensure_resource_permission`` - so list and direct-read agree under plugin enforcement. Without this, - a deny-all plugin would hide the caller's own rows from the listing - response while letting them read the same rows directly. - """ - _install_settings(monkeypatch, authz_enabled=True) - other_user = uuid4() - - items = [ - SimpleNamespace(id=uuid4(), user_id=fake_user.id), # owned → must keep - SimpleNamespace(id=uuid4(), user_id=other_user), # not owned → enforcer decides - SimpleNamespace(id=uuid4(), user_id=fake_user.id), # owned → must keep - ] - # Deny-all stub so any item that reaches the enforcer would be dropped. - service = _StubAuthorizationService(allow=False) - _install_authz(monkeypatch, service) - - result = await authz_utils.filter_visible_resources( - fake_user, - resource_type="flow", - candidates=items, - owner_extractor=lambda item: item.user_id, - act=FlowAction.READ, - ) - - # Owned items kept (positions 0 and 2); non-owned item dropped by deny. - assert result == [items[0], items[2]] - # Enforcer was consulted only for the non-owned item. - assert len(service.batch_calls) == 1 - assert len(service.batch_calls[0]["requests"]) == 1 - - -# ----------------------------------------------------------------------------- # -# ensure_deployment_permission -# ----------------------------------------------------------------------------- # - - -@pytest.mark.anyio -async def test_ensure_deployment_permission_uses_project_domain(monkeypatch, fake_user): - """project_id maps to project: domain (same helper as flows' folder_id).""" - _install_settings(monkeypatch, authz_enabled=True) - service = _StubAuthorizationService(allow=True) - _install_authz(monkeypatch, service) - _install_audit_recorder(monkeypatch) - - project_id = uuid4() - await authz_utils.ensure_deployment_permission( - fake_user, - DeploymentAction.READ, - deployment_id=uuid4(), - deployment_user_id=uuid4(), - project_id=project_id, - ) - - assert service.calls[0]["domain"] == f"project:{project_id}" - assert service.calls[0]["context"]["project_id"] == project_id - - -@pytest.mark.anyio -async def test_deployment_owner_override_skips_enforce(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_deployment_permission( - fake_user, - DeploymentAction.DELETE, - deployment_id=uuid4(), - deployment_user_id=fake_user.id, - project_id=uuid4(), - ) - - assert service.calls == [] - 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_uses_kb_id_object_slug(monkeypatch, fake_user): - """KB shares store UUIDs; the policy object slug must use ``kb_id``.""" - _install_settings(monkeypatch, authz_enabled=True) - service = _StubAuthorizationService(allow=True) - _install_authz(monkeypatch, service) - _install_audit_recorder(monkeypatch) - - kb_id = uuid4() - await authz_utils.ensure_knowledge_base_permission( - fake_user, - KnowledgeBaseAction.READ, - kb_id=kb_id, - kb_name="my-kb", - kb_user_id=uuid4(), - ) - - assert service.calls[0]["obj"] == f"knowledge_base:{kb_id}" - assert service.calls[0]["act"] == "read" - # ``kb_name`` is forwarded in the context for debugging. - assert service.calls[0]["context"]["kb_name"] == "my-kb" - assert service.calls[0]["context"]["kb_id"] == kb_id - - -@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_id=uuid4(), - 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_id=uuid4(), - kb_name="someone-elses", - kb_user_id=uuid4(), - ) - assert exc.value.status_code == 403 - - -@pytest.mark.anyio -async def test_kb_permission_create_uses_wildcard_slug(monkeypatch, fake_user): - """Without a kb_id (create flow) the slug is ``knowledge_base:*``.""" - _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.CREATE, - kb_user_id=fake_user.id, - ) - - # Owner-override fires; no enforce call. The audit row gets the wildcard - # slug so the dry-run CLI / audit query view shows it consistently. - assert service.calls == [] - - -# --------------------------------------------------------------------------- # -# 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/backend/tests/unit/test_authz_models.py b/src/backend/tests/unit/test_authz_models.py index fc81156cf1..db55001f61 100644 --- a/src/backend/tests/unit/test_authz_models.py +++ b/src/backend/tests/unit/test_authz_models.py @@ -130,8 +130,20 @@ async def test_authz_role_assignment_blocks_duplicate_global(authz_async_session await authz_async_session.commit() authz_async_session.add(AuthzRoleAssignment(user_id=user.id, role_id=role.id, domain_type="global")) - with pytest.raises(IntegrityError): + with pytest.raises(IntegrityError) as excinfo: await authz_async_session.commit() + # Assert the *specific* constraint fires (the partial unique index over + # user_id+role_id+domain_type, WHERE domain_id IS NULL) — otherwise a + # generic NOT NULL or random regression could mask the missing partial + # index. SQLite reports column names rather than the constraint name; we + # inspect only the failure clause before ``[SQL: ...]`` (which echoes every + # column in the INSERT and would otherwise smuggle ``domain_id`` in). + failure_clause = str(excinfo.value).lower().split("[sql:")[0] + assert "unique constraint failed" in failure_clause + assert "domain_type" in failure_clause + # Unscoped index does not cover domain_id; its absence distinguishes it + # from the scoped index (which also lists ``domain_id``). + assert "domain_id" not in failure_clause await authz_async_session.rollback() @@ -163,8 +175,13 @@ async def test_authz_role_assignment_blocks_duplicate_non_global_with_null_domai await authz_async_session.commit() authz_async_session.add(AuthzRoleAssignment(user_id=user.id, role_id=role.id, domain_type="org")) - with pytest.raises(IntegrityError): + with pytest.raises(IntegrityError) as excinfo: await authz_async_session.commit() + failure_clause = str(excinfo.value).lower().split("[sql:")[0] + assert "unique constraint failed" in failure_clause + assert "domain_type" in failure_clause + # Widened (unscoped) partial index — domain_id is NOT part of the constraint key. + assert "domain_id" not in failure_clause await authz_async_session.rollback() @@ -194,8 +211,12 @@ async def test_authz_role_assignment_blocks_duplicate_scoped(authz_async_session authz_async_session.add( AuthzRoleAssignment(user_id=user.id, role_id=role.id, domain_type="workspace", domain_id=workspace_id) ) - with pytest.raises(IntegrityError): + with pytest.raises(IntegrityError) as excinfo: await authz_async_session.commit() + failure_clause = str(excinfo.value).lower().split("[sql:")[0] + assert "unique constraint failed" in failure_clause + # The scoped index covers domain_id — its presence distinguishes it from the unscoped one. + assert "domain_id" in failure_clause await authz_async_session.rollback() @@ -338,8 +359,12 @@ async def test_authz_share_blocks_duplicate_targeted(authz_async_session: AsyncS created_by=user.id, ) ) - with pytest.raises(IntegrityError): + with pytest.raises(IntegrityError) as excinfo: await authz_async_session.commit() + failure_clause = str(excinfo.value).lower().split("[sql:")[0] + assert "unique constraint failed" in failure_clause + # targeted index covers target_id; untargeted does not + assert "target_id" in failure_clause await authz_async_session.rollback() @@ -384,8 +409,13 @@ async def test_authz_share_blocks_duplicate_untargeted(authz_async_session: Asyn created_by=user.id, ) ) - with pytest.raises(IntegrityError): + with pytest.raises(IntegrityError) as excinfo: await authz_async_session.commit() + failure_clause = str(excinfo.value).lower().split("[sql:")[0] + assert "unique constraint failed" in failure_clause + # The untargeted index covers (resource_type, resource_id, scope) — no target_id column. + assert "target_id" not in failure_clause + assert "scope" in failure_clause await authz_async_session.rollback() @@ -457,8 +487,9 @@ async def test_authz_share_rejects_unknown_scope(authz_async_session: AsyncSessi created_by=user.id, ) ) - with pytest.raises(IntegrityError): + with pytest.raises(IntegrityError) as excinfo: await authz_async_session.commit() + assert "ck_authz_share_scope_enum" in str(excinfo.value).lower() await authz_async_session.rollback() @@ -487,8 +518,9 @@ async def test_authz_share_rejects_targeted_with_null_target(authz_async_session created_by=user.id, ) ) - with pytest.raises(IntegrityError): + with pytest.raises(IntegrityError) as excinfo: await authz_async_session.commit() + assert "ck_authz_share_scope_target_consistency" in str(excinfo.value).lower() await authz_async_session.rollback() @@ -519,8 +551,9 @@ async def test_authz_share_rejects_untargeted_with_target(authz_async_session: A created_by=user.id, ) ) - with pytest.raises(IntegrityError): + with pytest.raises(IntegrityError) as excinfo: await authz_async_session.commit() + assert "ck_authz_share_scope_target_consistency" in str(excinfo.value).lower() await authz_async_session.rollback()