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

refactor(authz): split utils.py and address PR #13153 review

Browse Source 
Address review feedback on the OSS authorization foundations PR:

- B1/I1: Split services/authorization/utils.py (694 lines, mixed
  responsibilities) into focused modules:
    - audit.py    — batched audit pipeline (audit_decision, writer loop)
    - guards.py   — ensure_*_permission family (collapsed via _RESOURCE_SPECS
                    registry; nine 25-line clones become 7 thin wrappers)
    - listing.py  — filter_visible_resources
  utils.py remains as a back-compat re-export shim so existing call sites
  (api/v1/flows.py, deployments.py, etc.) keep working untouched.

- B2: Split tests/unit/services/authorization/test_utils.py (1004 lines)
  into test_audit.py, test_guards.py, test_filter_visible.py,
  test_domain_resolution.py. Shared stubs + monkeypatch helpers extracted
  into _common.py; pytest fixtures into conftest.py.

- I2: ensure_permission default 403 detail is now "Permission denied" —
  callers can opt into a richer message via the new `detail=` kwarg. The
  previous default echoed flow:<uuid> in the response, leaking existence
  to non-owners on any route that forgot to wrap in deny_to_404.

- I4: Document the OSS floor contract on _ensure_can_administer_share —
  the early-return is dead in OSS (supports_cross_user_fetch() is False)
  and the explicit 403 is the actual floor; the enterprise plugin gates
  via the downstream ensure_share_permission call.

- R2: Audit drop logging is now time-based (first drop + at most every
  10s during persistent saturation) instead of every 1000th drop. Low
  drop rates no longer go minutes without a log line.

- R5: Seed migration's non-Postgres/SQLite fallback wraps the INSERT in
  a SAVEPOINT and swallows IntegrityError. Two concurrent migration
  runners can no longer race past the SELECT-then-INSERT check.

- R6: IntegrityError tests in test_authz_models.py now verify the
  *specific* constraint fired — column names for unique partial indexes
  (SQLite doesn't surface the index name), constraint name substring
  for CHECK constraints. A generic NOT NULL regression would no longer
  pass these tests.

All 180 authz tests pass; ruff clean.
This commit is contained in:
Eric Hare
2026-05-26 11:13:57 -07:00
parent 3997fe905c
commit 4b0644f577
17 changed files with 2148 additions and 1716 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/backend/base/langflow/alembic/versions/7c8d9e0f1a2b_authz_foundations.py
View File

@ -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:

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

@ -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,

28
src/backend/base/langflow/cli/authz_dry_run.py
View File

@ -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()

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

@ -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",

277
src/backend/base/langflow/services/authorization/audit.py Normal file
View File

@ -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,
)

495
src/backend/base/langflow/services/authorization/guards.py Normal file
View File

@ -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,
)

92
src/backend/base/langflow/services/authorization/listing.py Normal file
View File

@ -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]

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

@ -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",
]

12
src/backend/tests/unit/api/v1/test_authz_share_routes.py
View File

@ -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

80
src/backend/tests/unit/services/authorization/_common.py Normal file
View File

@ -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

20
src/backend/tests/unit/services/authorization/conftest.py Normal file
View File

@ -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)

149
src/backend/tests/unit/services/authorization/test_audit.py Normal file
View File

@ -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"]

46
src/backend/tests/unit/services/authorization/test_domain_resolution.py Normal file
View File

@ -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:<uuid> 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

172
src/backend/tests/unit/services/authorization/test_filter_visible.py Normal file
View File

@ -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

634
src/backend/tests/unit/services/authorization/test_guards.py Normal file
View File

@ -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"

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

File diff suppressed because it is too large Load Diff

49
src/backend/tests/unit/test_authz_models.py
View File

@ -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()