From 1f24f1981fd7dbdbfa25e4955194eb0ed2002c56 Mon Sep 17 00:00:00 2001 From: Eric Hare Date: Tue, 26 May 2026 12:22:03 -0700 Subject: [PATCH] fix: remove dry run script --- AGENTS.md | 2 - src/backend/base/langflow/__main__.py | 8 - src/backend/base/langflow/api/v1/endpoints.py | 40 +- .../base/langflow/cli/authz_dry_run.py | 444 ------------------ .../tests/unit/test_authz_dry_run_cli.py | 243 ---------- src/backend/tests/unit/test_webhook.py | 47 ++ 6 files changed, 82 insertions(+), 702 deletions(-) delete mode 100644 src/backend/base/langflow/cli/authz_dry_run.py delete mode 100644 src/backend/tests/unit/test_authz_dry_run_cli.py diff --git a/AGENTS.md b/AGENTS.md index f3cc8be295..57d0d4c273 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -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. diff --git a/src/backend/base/langflow/__main__.py b/src/backend/base/langflow/__main__.py index 28fc2dd939..c5c6434343 100644 --- a/src/backend/base/langflow/__main__.py +++ b/src/backend/base/langflow/__main__.py @@ -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.""" diff --git a/src/backend/base/langflow/api/v1/endpoints.py b/src/backend/base/langflow/api/v1/endpoints.py index 217624aaa7..069d8b7df8 100644 --- a/src/backend/base/langflow/api/v1/endpoints.py +++ b/src/backend/base/langflow/api/v1/endpoints.py @@ -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.""" diff --git a/src/backend/base/langflow/cli/authz_dry_run.py b/src/backend/base/langflow/cli/authz_dry_run.py deleted file mode 100644 index 8057521d4e..0000000000 --- a/src/backend/base/langflow/cli/authz_dry_run.py +++ /dev/null @@ -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) diff --git a/src/backend/tests/unit/test_authz_dry_run_cli.py b/src/backend/tests/unit/test_authz_dry_run_cli.py deleted file mode 100644 index 331155badb..0000000000 --- a/src/backend/tests/unit/test_authz_dry_run_cli.py +++ /dev/null @@ -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 diff --git a/src/backend/tests/unit/test_webhook.py b/src/backend/tests/unit/test_webhook.py index 735e5ffd18..6c1be2f5b2 100644 --- a/src/backend/tests/unit/test_webhook.py +++ b/src/backend/tests/unit/test_webhook.py @@ -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"]