mirror of
https://github.com/langflow-ai/langflow.git
synced 2026-07-24 08:57:31 +08:00
fix: remove dry run script
This commit is contained in:
Eric Hare
@ -121,8 +121,6 @@ The Casbin request shape is `(subject, domain, object, action)`:
|
||||
- object = `flow:{uuid}` / `deployment:{uuid}` / `project:{uuid}` / `flow:*` / etc.
|
||||
- action = `read` / `write` / `create` / `delete` / `execute` / `deploy`
|
||||
|
||||
Use `langflow authz dry-run` to simulate a built-in policy against the live audit table without enabling enforcement.
|
||||
|
||||
**Share-aware fetch (Phase 3):** route fetch helpers (`_read_flow`, `get_flow_by_id_or_endpoint_name`, `get_deployment`, project reads in `projects.py`, v2 file fetcher) branch on `BaseAuthorizationService.supports_cross_user_fetch()`. The OSS pass-through reports `False` so the existing owner-scoped queries are preserved — enabling `LANGFLOW_AUTHZ_ENABLED=true` without an enterprise plugin cannot widen visibility. Enterprise plugins set `SUPPORTS_CROSS_USER_FETCH=True` so resources load by id alone and `ensure_*_permission` decides access; route handlers can convert a plugin-deny `HTTPException(403)` to `HTTPException(404)` via `langflow.services.authorization.fetch.deny_to_404` to preserve UUID privacy.
|
||||
|
||||
**Share CRUD API (Phase 3):** `/api/v1/authz/shares` provides POST / GET / PATCH / DELETE on `authz_share` rows. The handler enforces an OSS floor (resource owner or superuser may administer shares for that resource) so the OSS pass-through cannot let a non-owner mint share rows. Each write fires `BaseAuthorizationService.invalidate_user` / `invalidate_all` so an enterprise enforcer can drop cached policy. Audit rows are written via `audit_decision` with `share:create` / `share:update` / `share:delete` actions.
|
||||
|
||||
@ -81,14 +81,6 @@ except ImportError:
|
||||
# LFX not available, skip adding the sub-app
|
||||
pass
|
||||
|
||||
# Add authz utilities (`langflow authz dry-run`, ...).
|
||||
try:
|
||||
from langflow.cli.authz_dry_run import authz_app
|
||||
|
||||
app.add_typer(authz_app, name="authz")
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
|
||||
class ProcessManager:
|
||||
"""Manages the lifecycle of the backend process."""
|
||||
|
||||
@ -542,12 +542,29 @@ async def get_flow_for_current_user(
|
||||
return await get_flow_by_id_or_endpoint_name(flow_id_or_name, current_user.id, widen_for_shares=True)
|
||||
|
||||
|
||||
class SseAuth:
|
||||
"""Helper to carry both authenticated user and flow for SSE subscription."""
|
||||
|
||||
def __init__(self, user: User | UserRead, flow: FlowRead):
|
||||
self.user = user
|
||||
self.flow = flow
|
||||
|
||||
|
||||
async def get_flow_for_sse_user(
|
||||
flow_id_or_name: str,
|
||||
user: Annotated[User | UserRead, Depends(get_current_user_for_sse)],
|
||||
) -> FlowRead:
|
||||
"""Auth-aware wrapper around ``get_flow_by_id_or_endpoint_name`` for SSE routes."""
|
||||
return await get_flow_by_id_or_endpoint_name(flow_id_or_name, user_id=user.id, widen_for_shares=True)
|
||||
) -> SseAuth:
|
||||
"""Auth-aware dependency for SSE routes.
|
||||
|
||||
Returns both the SSE user and the flow so the route can call
|
||||
``ensure_flow_permission`` *before* subscribing to the event stream.
|
||||
Widening to share-aware lookup is safe here only because the route
|
||||
immediately enforces ``flow:read``; without that enforcement, a non-owner
|
||||
with cross-user fetch enabled could subscribe to another user's webhook
|
||||
event stream and exfiltrate flow id/name plus event payloads.
|
||||
"""
|
||||
flow = await get_flow_by_id_or_endpoint_name(flow_id_or_name, user_id=user.id, widen_for_shares=True)
|
||||
return SseAuth(user=user, flow=flow)
|
||||
|
||||
|
||||
class WebhookAuth:
|
||||
@ -861,7 +878,7 @@ async def simplified_run_flow_session(
|
||||
|
||||
@router.get("/webhook-events/{flow_id_or_name}", include_in_schema=False)
|
||||
async def webhook_events_stream(
|
||||
flow: Annotated[FlowRead, Depends(get_flow_for_sse_user)],
|
||||
auth: Annotated[SseAuth, Depends(get_flow_for_sse_user)],
|
||||
request: Request,
|
||||
):
|
||||
"""Server-Sent Events (SSE) endpoint for real-time webhook build updates.
|
||||
@ -870,8 +887,21 @@ async def webhook_events_stream(
|
||||
of webhook execution progress, similar to clicking "Play" in the UI.
|
||||
|
||||
Authentication: Requires user to be logged in (via cookie) or provide API key.
|
||||
The user must own the flow to subscribe to its events.
|
||||
The user must own the flow OR have an authorization-plugin-granted
|
||||
``flow:read`` permission to subscribe to its events.
|
||||
"""
|
||||
flow = auth.flow
|
||||
# Enforce flow:read before subscribing — the SSE fetcher uses share-aware
|
||||
# lookup, so without this check a non-owner with cross-user fetch enabled
|
||||
# would receive another user's webhook event payloads.
|
||||
await ensure_flow_permission(
|
||||
auth.user,
|
||||
FlowAction.READ,
|
||||
flow_id=flow.id,
|
||||
flow_user_id=flow.user_id,
|
||||
workspace_id=getattr(flow, "workspace_id", None),
|
||||
folder_id=getattr(flow, "folder_id", None),
|
||||
)
|
||||
|
||||
async def event_generator() -> AsyncGenerator[str, None]:
|
||||
"""Generate SSE events from the webhook event manager."""
|
||||
|
||||
@ -1,444 +0,0 @@
|
||||
"""``langflow authz dry-run`` — simulate every flow guard against a stub policy.
|
||||
|
||||
The OSS ``LangflowAuthorizationService`` always allows, so a real install can't
|
||||
demonstrate enforcement without the authorization plugin. This subcommand
|
||||
replaces the live authorization service with a small in-memory stub for one
|
||||
invocation, walks every flow-CRUD guard site, and prints what *would* happen
|
||||
under the chosen policy — including the policy tuple, the audit row that
|
||||
would have been written, and the resulting HTTP outcome. Useful for:
|
||||
|
||||
* Validating that ``AUTHZ_ENABLED=true`` + a future policy will produce
|
||||
the expected behaviour before shipping it to production.
|
||||
* Showing operators what the audit log will look like.
|
||||
* Smoke-testing the guard wiring after a refactor.
|
||||
|
||||
The command never hits the network or writes to the real database; it patches
|
||||
the helpers in :mod:`langflow.services.authorization.utils` for the duration
|
||||
of the run and restores them on exit.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import sys
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import asdict, dataclass
|
||||
from enum import Enum
|
||||
from pathlib import Path # noqa: TC003 — typer resolves this annotation at runtime
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
import typer
|
||||
from fastapi import HTTPException
|
||||
from rich import box
|
||||
from rich.console import Console
|
||||
from rich.table import Table
|
||||
|
||||
authz_app = typer.Typer(
|
||||
name="authz",
|
||||
help="Authorization (RBAC) utilities.",
|
||||
no_args_is_help=True,
|
||||
)
|
||||
|
||||
_console = Console()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Stub policies
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
class StubPolicy(str, Enum):
|
||||
"""Built-in stand-ins for an authorization plugin policy."""
|
||||
|
||||
ALLOW_ALL = "allow-all"
|
||||
DENY_NON_OWNER = "deny-non-owner"
|
||||
DENY_WRITES = "deny-writes"
|
||||
OWNER_ONLY = "owner-only"
|
||||
|
||||
|
||||
_POLICY_DESCRIPTIONS = {
|
||||
StubPolicy.ALLOW_ALL: "Pass-through (matches the OSS default).",
|
||||
StubPolicy.DENY_NON_OWNER: "Owners and superusers allowed; everyone else denied.",
|
||||
StubPolicy.DENY_WRITES: "Read/execute allowed; write/create/delete denied.",
|
||||
StubPolicy.OWNER_ONLY: "Only the flow owner allowed (no superuser bypass).",
|
||||
}
|
||||
|
||||
|
||||
class _StubAuthorizationService:
|
||||
"""Configurable enforce stub used only during ``dry-run``."""
|
||||
|
||||
def __init__(self, policy: StubPolicy) -> None:
|
||||
self.policy = policy
|
||||
|
||||
async def enforce(
|
||||
self,
|
||||
*,
|
||||
user_id: UUID,
|
||||
domain: str, # noqa: ARG002
|
||||
obj: str, # noqa: ARG002
|
||||
act: str,
|
||||
context: dict[str, Any] | None = None,
|
||||
) -> bool:
|
||||
"""Decide allow/deny per the configured stub policy."""
|
||||
ctx = context or {}
|
||||
if self.policy is StubPolicy.ALLOW_ALL:
|
||||
return True
|
||||
if self.policy is StubPolicy.DENY_NON_OWNER:
|
||||
if ctx.get("is_superuser"):
|
||||
return True
|
||||
return ctx.get("flow_user_id") == user_id
|
||||
if self.policy is StubPolicy.DENY_WRITES:
|
||||
return act in {"read", "execute"}
|
||||
if self.policy is StubPolicy.OWNER_ONLY:
|
||||
return ctx.get("flow_user_id") == user_id
|
||||
return True
|
||||
|
||||
async def batch_enforce(
|
||||
self,
|
||||
*,
|
||||
user_id: UUID,
|
||||
domain: str,
|
||||
requests: list[tuple[str, str]],
|
||||
context: dict[str, Any] | None = None,
|
||||
) -> list[bool]:
|
||||
"""Apply :meth:`enforce` to each request, returning a parallel list."""
|
||||
return [
|
||||
await self.enforce(user_id=user_id, domain=domain, obj=obj, act=act, context=context)
|
||||
for obj, act in requests
|
||||
]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Scenario definitions
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _GuardSite:
|
||||
"""Static description of one ``ensure_flow_permission`` call site in flows.py."""
|
||||
|
||||
name: str
|
||||
route: str
|
||||
action: str # FlowAction value
|
||||
has_flow_id: bool # True ⇒ obj=flow:{id}, False ⇒ obj=flow:*
|
||||
has_owner: bool # True ⇒ flow_user_id is provided
|
||||
|
||||
|
||||
_FLOW_GUARDS: tuple[_GuardSite, ...] = (
|
||||
_GuardSite("create_flow", "POST /flows/", "create", has_flow_id=False, has_owner=False),
|
||||
_GuardSite("read_flow", "GET /flows/{id}", "read", has_flow_id=True, has_owner=True),
|
||||
_GuardSite("update_flow", "PATCH /flows/{id}", "write", has_flow_id=True, has_owner=True),
|
||||
_GuardSite("upsert_flow (existing)", "PUT /flows/{id}", "write", has_flow_id=True, has_owner=True),
|
||||
_GuardSite("upsert_flow (new)", "PUT /flows/{id}", "create", has_flow_id=False, has_owner=False),
|
||||
_GuardSite("delete_flow", "DELETE /flows/{id}", "delete", has_flow_id=True, has_owner=True),
|
||||
_GuardSite("create_flows (batch)", "POST /flows/batch/", "create", has_flow_id=False, has_owner=False),
|
||||
_GuardSite("upload_file", "POST /flows/upload/", "create", has_flow_id=False, has_owner=False),
|
||||
_GuardSite("delete_multiple_flows", "DELETE /flows/", "delete", has_flow_id=True, has_owner=True),
|
||||
_GuardSite("download_multiple_file", "POST /flows/download/", "read", has_flow_id=True, has_owner=True),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _Actor:
|
||||
"""Simulated caller of a flow route."""
|
||||
|
||||
name: str
|
||||
is_superuser: bool
|
||||
is_flow_owner: bool
|
||||
|
||||
|
||||
@dataclass
|
||||
class DryRunResult:
|
||||
"""One row in the dry-run report."""
|
||||
|
||||
scenario: str
|
||||
route: str
|
||||
actor: str
|
||||
action: str
|
||||
obj: str
|
||||
domain: str
|
||||
decision: str # allow / deny / owner_override
|
||||
http_status: int
|
||||
audit_action: str
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Runner
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
@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 audit as authz_audit
|
||||
from langflow.services.authorization import guards as authz_guards
|
||||
|
||||
settings = SimpleNamespace(
|
||||
auth_settings=SimpleNamespace(
|
||||
AUTHZ_ENABLED=True,
|
||||
AUTHZ_AUDIT_ENABLED=True,
|
||||
AUTHZ_SUPERUSER_BYPASS=True,
|
||||
)
|
||||
)
|
||||
|
||||
async def _record_audit(**kwargs: Any) -> None:
|
||||
audit_sink.append(kwargs)
|
||||
|
||||
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_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_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(
|
||||
guard: _GuardSite,
|
||||
actor: _Actor,
|
||||
*,
|
||||
owner_id: UUID,
|
||||
flow_id: UUID,
|
||||
workspace_id: UUID,
|
||||
folder_id: UUID,
|
||||
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 guards as authz_utils
|
||||
from langflow.services.authorization.actions import FlowAction
|
||||
|
||||
actor_id = owner_id if actor.is_flow_owner else uuid4()
|
||||
user_proxy = SimpleNamespace(id=actor_id, is_superuser=actor.is_superuser)
|
||||
|
||||
kwargs: dict[str, Any] = {
|
||||
"workspace_id": workspace_id,
|
||||
"folder_id": folder_id,
|
||||
}
|
||||
if guard.has_flow_id:
|
||||
kwargs["flow_id"] = flow_id
|
||||
if guard.has_owner:
|
||||
kwargs["flow_user_id"] = owner_id
|
||||
|
||||
audit_before = len(audit_sink)
|
||||
http_status = 200
|
||||
|
||||
try:
|
||||
await authz_utils.ensure_flow_permission(
|
||||
user_proxy, # type: ignore[arg-type]
|
||||
FlowAction(guard.action),
|
||||
**kwargs,
|
||||
)
|
||||
except HTTPException as exc:
|
||||
http_status = exc.status_code
|
||||
|
||||
rows_written = audit_sink[audit_before:]
|
||||
decision = rows_written[-1]["result"] if rows_written else "skipped"
|
||||
audit_action = rows_written[-1]["action"] if rows_written else ""
|
||||
obj = rows_written[-1]["obj"] if rows_written else (f"flow:{flow_id}" if guard.has_flow_id else "flow:*")
|
||||
# Reach for the audited domain first; only fall back when no audit row was written
|
||||
# (rare — AUTHZ_ENABLED is forced True in dry-run). The fallback mirrors what
|
||||
# `_resolve_flow_domain` would emit given both ids: project wins.
|
||||
domain = (
|
||||
rows_written[-1]["details"].get("domain", "")
|
||||
if rows_written and isinstance(rows_written[-1].get("details"), dict)
|
||||
else f"project:{folder_id}"
|
||||
)
|
||||
|
||||
return DryRunResult(
|
||||
scenario=guard.name,
|
||||
route=guard.route,
|
||||
actor=actor.name,
|
||||
action=guard.action,
|
||||
obj=obj,
|
||||
domain=domain,
|
||||
decision=decision,
|
||||
http_status=http_status,
|
||||
audit_action=audit_action,
|
||||
)
|
||||
|
||||
|
||||
async def _run_all(policy: StubPolicy) -> list[DryRunResult]:
|
||||
"""Run every guard x actor combo under ``policy`` and return result rows."""
|
||||
owner_id = uuid4()
|
||||
flow_id = uuid4()
|
||||
workspace_id = uuid4()
|
||||
folder_id = uuid4()
|
||||
|
||||
actors = (
|
||||
_Actor("alice (owner)", is_superuser=False, is_flow_owner=True),
|
||||
_Actor("bob (non-owner)", is_superuser=False, is_flow_owner=False),
|
||||
_Actor("carol (superuser)", is_superuser=True, is_flow_owner=False),
|
||||
)
|
||||
|
||||
stub = _StubAuthorizationService(policy)
|
||||
audit_sink: list[dict[str, Any]] = []
|
||||
|
||||
rows: list[DryRunResult] = []
|
||||
with _install_stubs(stub, audit_sink):
|
||||
for guard in _FLOW_GUARDS:
|
||||
for actor in actors:
|
||||
row = await _run_one(
|
||||
guard,
|
||||
actor,
|
||||
owner_id=owner_id,
|
||||
flow_id=flow_id,
|
||||
workspace_id=workspace_id,
|
||||
folder_id=folder_id,
|
||||
audit_sink=audit_sink,
|
||||
)
|
||||
rows.append(row)
|
||||
return rows
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Output formatters
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
_DECISION_STYLE = {
|
||||
"allow": "green",
|
||||
"deny": "red",
|
||||
"owner_override": "cyan",
|
||||
"skipped": "dim",
|
||||
}
|
||||
|
||||
|
||||
_UUID_PREFIX_LEN = 8
|
||||
|
||||
|
||||
def _short_uuid(value: str) -> str:
|
||||
"""Trim a UUID-bearing string to its first 8 hex characters for table readability."""
|
||||
if ":" in value:
|
||||
prefix, _, suffix = value.partition(":")
|
||||
if len(suffix) >= _UUID_PREFIX_LEN and "-" in suffix:
|
||||
return f"{prefix}:{suffix[:_UUID_PREFIX_LEN]}…"
|
||||
return value
|
||||
|
||||
|
||||
def _render_table(policy: StubPolicy, rows: list[DryRunResult]) -> Table:
|
||||
"""Build a Rich table summarising every dry-run result."""
|
||||
table = Table(
|
||||
title=f"authz dry-run · policy={policy.value} · {_POLICY_DESCRIPTIONS[policy]}",
|
||||
box=box.SIMPLE_HEAVY,
|
||||
show_lines=False,
|
||||
)
|
||||
table.add_column("Scenario", style="bold")
|
||||
table.add_column("Route")
|
||||
table.add_column("Actor")
|
||||
table.add_column("Action")
|
||||
table.add_column("Domain")
|
||||
table.add_column("Obj")
|
||||
table.add_column("Decision")
|
||||
table.add_column("HTTP", justify="right")
|
||||
for row in rows:
|
||||
style = _DECISION_STYLE.get(row.decision, "white")
|
||||
table.add_row(
|
||||
row.scenario,
|
||||
row.route,
|
||||
row.actor,
|
||||
row.action,
|
||||
_short_uuid(row.domain),
|
||||
_short_uuid(row.obj),
|
||||
f"[{style}]{row.decision}[/{style}]",
|
||||
str(row.http_status),
|
||||
)
|
||||
return table
|
||||
|
||||
|
||||
def _summary_counts(rows: list[DryRunResult]) -> dict[str, int]:
|
||||
"""Aggregate decisions across all rows for the footer line."""
|
||||
counts: dict[str, int] = {}
|
||||
for row in rows:
|
||||
counts[row.decision] = counts.get(row.decision, 0) + 1
|
||||
return counts
|
||||
|
||||
|
||||
def _emit_json(policy: StubPolicy, rows: list[DryRunResult], output: Path | None) -> None:
|
||||
"""Serialise the report as JSON to stdout (or a file)."""
|
||||
payload = {
|
||||
"policy": policy.value,
|
||||
"policy_description": _POLICY_DESCRIPTIONS[policy],
|
||||
"results": [asdict(r) for r in rows],
|
||||
"summary": _summary_counts(rows),
|
||||
}
|
||||
text = json.dumps(payload, indent=2, default=str)
|
||||
if output is not None:
|
||||
output.write_text(text)
|
||||
_console.print(f"[green]wrote[/green] {output} ({len(rows)} rows)")
|
||||
else:
|
||||
sys.stdout.write(text + "\n")
|
||||
|
||||
|
||||
def _emit_table(policy: StubPolicy, rows: list[DryRunResult], output: Path | None) -> None:
|
||||
"""Render the report as a Rich table to stdout (or a plain-text file)."""
|
||||
table = _render_table(policy, rows)
|
||||
summary = _summary_counts(rows)
|
||||
summary_line = " ".join(
|
||||
f"[{_DECISION_STYLE.get(k, 'white')}]{v} {k}[/{_DECISION_STYLE.get(k, 'white')}]"
|
||||
for k, v in sorted(summary.items())
|
||||
)
|
||||
if output is not None:
|
||||
with output.open("w") as fh:
|
||||
file_console = Console(file=fh, force_terminal=False, width=200)
|
||||
file_console.print(table)
|
||||
file_console.print(f"summary: {summary_line}")
|
||||
_console.print(f"[green]wrote[/green] {output} ({len(rows)} rows)")
|
||||
else:
|
||||
_console.print(table)
|
||||
_console.print(f"summary: {summary_line}")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Typer command
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
@authz_app.command(name="dry-run")
|
||||
def dry_run(
|
||||
policy: StubPolicy = typer.Option(
|
||||
StubPolicy.DENY_NON_OWNER,
|
||||
"--policy",
|
||||
"-p",
|
||||
help="Which stub policy to simulate.",
|
||||
case_sensitive=False,
|
||||
),
|
||||
output: Path | None = typer.Option(
|
||||
None,
|
||||
"--output",
|
||||
"-o",
|
||||
help="Write the report to this file instead of stdout.",
|
||||
),
|
||||
*,
|
||||
as_json: bool = typer.Option(
|
||||
False, # noqa: FBT003 — typer requires positional default value
|
||||
"--json",
|
||||
help="Emit structured JSON instead of the Rich table.",
|
||||
),
|
||||
) -> None:
|
||||
"""Simulate every flow guard under a stub policy and print what *would* happen.
|
||||
|
||||
The command runs entirely in-process. No HTTP request is made, no row is
|
||||
written to the real ``authz_audit_log`` table — the report describes the
|
||||
policy tuple and audit row that *would* be produced if an plugin
|
||||
plugin enforcing ``--policy`` were registered.
|
||||
"""
|
||||
rows = asyncio.run(_run_all(policy))
|
||||
if as_json:
|
||||
_emit_json(policy, rows, output)
|
||||
else:
|
||||
_emit_table(policy, rows, output)
|
||||
@ -1,243 +0,0 @@
|
||||
"""Tests for ``langflow authz dry-run`` (CLI scenario runner)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
from pathlib import Path # noqa: TC003 — used at runtime by the pytest tmp_path fixture annotation
|
||||
|
||||
import pytest
|
||||
from langflow.cli.authz_dry_run import (
|
||||
_FLOW_GUARDS,
|
||||
DryRunResult,
|
||||
StubPolicy,
|
||||
_run_all,
|
||||
_StubAuthorizationService,
|
||||
authz_app,
|
||||
)
|
||||
from typer.testing import CliRunner
|
||||
|
||||
_ACTOR_COUNT = 3 # alice, bob, carol — see _run_all
|
||||
_EXPECTED_ROWS = len(_FLOW_GUARDS) * _ACTOR_COUNT
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Stub policy
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _run(coro):
|
||||
"""Synchronous wrapper for tests that exercise the async stub directly."""
|
||||
return asyncio.run(coro)
|
||||
|
||||
|
||||
def test_stub_allow_all_returns_true_unconditionally():
|
||||
"""allow-all policy permits every request regardless of context."""
|
||||
stub = _StubAuthorizationService(StubPolicy.ALLOW_ALL)
|
||||
assert _run(stub.enforce(user_id="u", domain="*", obj="flow:x", act="delete")) is True
|
||||
|
||||
|
||||
def test_stub_deny_non_owner_blocks_non_owner_non_superuser():
|
||||
"""deny-non-owner: only owners and superusers pass."""
|
||||
stub = _StubAuthorizationService(StubPolicy.DENY_NON_OWNER)
|
||||
assert (
|
||||
_run(
|
||||
stub.enforce(
|
||||
user_id="alice",
|
||||
domain="*",
|
||||
obj="flow:x",
|
||||
act="read",
|
||||
context={"is_superuser": False, "flow_user_id": "alice"},
|
||||
)
|
||||
)
|
||||
is True
|
||||
)
|
||||
assert (
|
||||
_run(
|
||||
stub.enforce(
|
||||
user_id="bob",
|
||||
domain="*",
|
||||
obj="flow:x",
|
||||
act="read",
|
||||
context={"is_superuser": False, "flow_user_id": "alice"},
|
||||
)
|
||||
)
|
||||
is False
|
||||
)
|
||||
assert (
|
||||
_run(
|
||||
stub.enforce(
|
||||
user_id="carol",
|
||||
domain="*",
|
||||
obj="flow:x",
|
||||
act="delete",
|
||||
context={"is_superuser": True, "flow_user_id": "alice"},
|
||||
)
|
||||
)
|
||||
is True
|
||||
)
|
||||
|
||||
|
||||
def test_stub_deny_writes_allows_reads():
|
||||
"""deny-writes: read/execute pass, everything else denied."""
|
||||
stub = _StubAuthorizationService(StubPolicy.DENY_WRITES)
|
||||
assert _run(stub.enforce(user_id="u", domain="*", obj="flow:x", act="read")) is True
|
||||
assert _run(stub.enforce(user_id="u", domain="*", obj="flow:x", act="execute")) is True
|
||||
assert _run(stub.enforce(user_id="u", domain="*", obj="flow:x", act="write")) is False
|
||||
assert _run(stub.enforce(user_id="u", domain="*", obj="flow:x", act="delete")) is False
|
||||
|
||||
|
||||
def test_stub_owner_only_ignores_superuser():
|
||||
"""owner-only: only the owner passes, even a superuser is denied."""
|
||||
stub = _StubAuthorizationService(StubPolicy.OWNER_ONLY)
|
||||
assert (
|
||||
_run(
|
||||
stub.enforce(
|
||||
user_id="alice",
|
||||
domain="*",
|
||||
obj="flow:x",
|
||||
act="write",
|
||||
context={"is_superuser": True, "flow_user_id": "alice"},
|
||||
)
|
||||
)
|
||||
is True
|
||||
)
|
||||
assert (
|
||||
_run(
|
||||
stub.enforce(
|
||||
user_id="carol",
|
||||
domain="*",
|
||||
obj="flow:x",
|
||||
act="read",
|
||||
context={"is_superuser": True, "flow_user_id": "alice"},
|
||||
)
|
||||
)
|
||||
is False
|
||||
)
|
||||
|
||||
|
||||
def test_stub_batch_enforce_mirrors_enforce():
|
||||
"""batch_enforce applies enforce to each (obj, act) pair."""
|
||||
stub = _StubAuthorizationService(StubPolicy.DENY_WRITES)
|
||||
results = _run(
|
||||
stub.batch_enforce(
|
||||
user_id="u",
|
||||
domain="*",
|
||||
requests=[("flow:1", "read"), ("flow:2", "write")],
|
||||
)
|
||||
)
|
||||
assert results == [True, False]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Scenario runner
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_run_all_produces_one_row_per_guard_per_actor():
|
||||
"""Every (guard, actor) pair appears exactly once in the report."""
|
||||
rows = asyncio.run(_run_all(StubPolicy.ALLOW_ALL))
|
||||
assert len(rows) == _EXPECTED_ROWS
|
||||
|
||||
|
||||
def test_allow_all_policy_yields_no_denials():
|
||||
"""Under allow-all, no row decision should be 'deny'."""
|
||||
rows = asyncio.run(_run_all(StubPolicy.ALLOW_ALL))
|
||||
denied = [r for r in rows if r.decision == "deny"]
|
||||
assert denied == []
|
||||
# Every flow-bearing scenario reports flow:{uuid}, never flow:*.
|
||||
flow_obj_rows = [r for r in rows if r.scenario.startswith("read_flow")]
|
||||
assert all(":*" not in r.obj for r in flow_obj_rows)
|
||||
|
||||
|
||||
def test_deny_non_owner_denies_bob():
|
||||
"""Bob (non-owner, non-superuser) is denied wherever an owner is present."""
|
||||
rows = asyncio.run(_run_all(StubPolicy.DENY_NON_OWNER))
|
||||
bob_rows = [r for r in rows if r.actor.startswith("bob")]
|
||||
# Scenarios that carry an owner pointer → bob is denied.
|
||||
owner_bearing = [r for r in bob_rows if "flow_id" not in r.scenario or "(new)" not in r.scenario]
|
||||
denials = [r for r in owner_bearing if r.decision == "deny"]
|
||||
assert denials, "expected at least one deny for bob under deny-non-owner"
|
||||
|
||||
|
||||
def test_owner_override_records_decision_for_alice():
|
||||
"""Alice owns the flow → owner_override fires on every per-flow scenario."""
|
||||
rows = asyncio.run(_run_all(StubPolicy.OWNER_ONLY))
|
||||
alice_rows = [r for r in rows if r.actor.startswith("alice") and ":" in r.obj and r.obj != "flow:*"]
|
||||
assert alice_rows, "expected per-flow rows for alice"
|
||||
assert all(r.decision == "owner_override" for r in alice_rows)
|
||||
|
||||
|
||||
def test_deny_writes_blocks_writes_allows_reads():
|
||||
"""Under deny-writes, read/execute pass and write/create/delete deny."""
|
||||
rows = asyncio.run(_run_all(StubPolicy.DENY_WRITES))
|
||||
# Restrict to bob (no owner override, no superuser bypass).
|
||||
bob_rows = [r for r in rows if r.actor.startswith("bob")]
|
||||
for row in bob_rows:
|
||||
if row.action in {"read", "execute"}:
|
||||
assert row.decision == "allow", row
|
||||
else:
|
||||
assert row.decision == "deny", row
|
||||
|
||||
|
||||
def test_domain_is_project_prefixed():
|
||||
"""The recorded domain uses the project:{uuid} form for in-scope scenarios.
|
||||
|
||||
``_resolve_flow_domain`` prefers project over workspace because g2
|
||||
inheritance is directional — passing the more specific domain lets both
|
||||
workspace-scoped and project-scoped grants match. The dry-run CLI passes
|
||||
both ids on every scenario, so every audited row resolves to ``project:``.
|
||||
"""
|
||||
rows = asyncio.run(_run_all(StubPolicy.ALLOW_ALL))
|
||||
project_rows = [r for r in rows if r.domain.startswith("project:")]
|
||||
assert project_rows, "expected at least one row with a project domain"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Typer integration
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_cli_table_output_runs_clean():
|
||||
"""`langflow authz dry-run` renders without raising.
|
||||
|
||||
When ``authz_app`` has a single command, Typer collapses the subcommand
|
||||
namespace, so ``CliRunner`` invokes it without the ``dry-run`` segment.
|
||||
The fully-namespaced ``langflow authz dry-run`` path is exercised when the
|
||||
parent ``app`` is built — see ``test_cli.py`` for that integration.
|
||||
"""
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(authz_app, ["--policy", "allow-all"])
|
||||
assert result.exit_code == 0, result.output
|
||||
assert "allow-all" in result.output
|
||||
|
||||
|
||||
def test_cli_json_output_is_valid_json(tmp_path: Path):
|
||||
"""`--json --output FILE` writes a parseable JSON document."""
|
||||
runner = CliRunner()
|
||||
out = tmp_path / "report.json"
|
||||
result = runner.invoke(authz_app, ["--policy", "deny-writes", "--json", "--output", str(out)])
|
||||
assert result.exit_code == 0, result.output
|
||||
|
||||
payload = json.loads(out.read_text())
|
||||
assert payload["policy"] == "deny-writes"
|
||||
assert isinstance(payload["results"], list)
|
||||
assert len(payload["results"]) == _EXPECTED_ROWS
|
||||
# Every result entry has the expected keys.
|
||||
expected_keys = {f.name for f in DryRunResult.__dataclass_fields__.values()}
|
||||
assert expected_keys <= set(payload["results"][0].keys())
|
||||
|
||||
|
||||
def test_cli_rejects_unknown_policy():
|
||||
"""An unknown policy value exits non-zero with a typer error."""
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(authz_app, ["--policy", "nope"])
|
||||
assert result.exit_code != 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize("policy", list(StubPolicy))
|
||||
def test_cli_runs_for_every_built_in_policy(policy: StubPolicy):
|
||||
"""Every built-in policy produces a clean run."""
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(authz_app, ["--policy", policy.value])
|
||||
assert result.exit_code == 0, result.output
|
||||
@ -899,6 +899,53 @@ class TestWebhookEventsStreamAuth:
|
||||
assert response.status_code == 404
|
||||
assert f"Flow identifier {endpoint_name} not found" in response.json()["detail"]
|
||||
|
||||
async def test_sse_route_calls_ensure_flow_permission_for_shared_flow(self):
|
||||
"""Regression: webhook_events_stream must enforce flow:read before subscribing.
|
||||
|
||||
The SSE dependency widens to share-aware lookup when an authz plugin is
|
||||
active, so a non-owner *could* resolve another user's flow. The route
|
||||
guards against this by calling ``ensure_flow_permission`` before any
|
||||
event-bus subscription happens — without that, a non-owner with
|
||||
cross-user fetch enabled could exfiltrate webhook event payloads for
|
||||
another user's flow.
|
||||
"""
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
from uuid import uuid4
|
||||
|
||||
from fastapi import HTTPException
|
||||
from langflow.api.v1 import endpoints as endpoints_module
|
||||
|
||||
owner = SimpleNamespace(id=uuid4(), is_superuser=False)
|
||||
attacker = SimpleNamespace(id=uuid4(), is_superuser=False)
|
||||
flow = SimpleNamespace(
|
||||
id=uuid4(),
|
||||
name="shared-flow",
|
||||
user_id=owner.id,
|
||||
workspace_id=None,
|
||||
folder_id=None,
|
||||
)
|
||||
auth = endpoints_module.SseAuth(user=attacker, flow=flow)
|
||||
|
||||
async def _deny(*_args, **_kwargs):
|
||||
raise HTTPException(status_code=403, detail="Permission denied")
|
||||
|
||||
# Subscribe must NOT be reached when permission is denied — if the
|
||||
# route accidentally falls through to event_generator() and calls
|
||||
# subscribe(), this AsyncMock will fail the test.
|
||||
subscribe_mock = AsyncMock(side_effect=AssertionError("subscribe() must not be called when authz denies"))
|
||||
|
||||
with (
|
||||
patch.object(endpoints_module, "ensure_flow_permission", side_effect=_deny),
|
||||
patch.object(endpoints_module.webhook_event_manager, "subscribe", subscribe_mock),
|
||||
):
|
||||
request_stub = SimpleNamespace()
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await endpoints_module.webhook_events_stream(auth=auth, request=request_stub)
|
||||
|
||||
assert exc_info.value.status_code == 403
|
||||
subscribe_mock.assert_not_awaited()
|
||||
|
||||
async def test_webhook_run_cross_user_uuid_returns_404(self, client, added_webhook_test, user_two_api_key):
|
||||
"""Regression test: cross-user access to run webhook via UUID returns 404."""
|
||||
flow_id = added_webhook_test["id"]
|
||||
|
||||
Reference in New Issue
Block a user