pengjunjie/langflow
1
0
Fork 1
mirror of https://github.com/langflow-ai/langflow.git synced 2026-07-28 03:26:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(authz): add DI dependencies and permission decorators (#13368)

Browse Source 
* feat(authz): add DI dependencies and permission decorators

Address PR #13153 feedback by introducing FastAPI authorized-flow
dependencies for route handlers and requires_flow_permission for
non-HTTP paths such as helpers.load_flow.

* feat(authz): add RequireFlowCreate dependency for POST /flows/

Move create_flow CREATE authorization into a FastAPI Depends so the
route signature declares protection like read/update/delete.

* [autofix.ci] apply automated fixes

* fix(authz): resolve ruff lint failures on route-pattern branch

Fix undefined User type, unused decorator assignment, FAST003 path params,
and related test lint issues so the Ruff style check passes in CI.

* [autofix.ci] apply automated fixes

* fix: couple small nits

* chore: drop unrelated component_index.json drift from PR

* [autofix.ci] apply automated fixes

* refactor(authz): drop OptionalAuthorizedReadFlow; note_translations returns 404

GET /flows/{id}/note_translations now uses AuthorizedReadFlow, so a missing
or inaccessible flow yields 404 instead of 200 {} — matching GET /flows/{id}
and the rest of the flow API. The owner-scoped uniform 404 is itself
UUID-privacy-preserving, so the special-cased empty body added no real
privacy over the existing posture. The sole frontend caller (NoteNode)
already treats an error as "no translations" and renders the original text.

Removes get_optional_authorized_flow_for_read and the OptionalAuthorizedReadFlow
alias, and updates the i18n + route-guard tests to assert the 404.

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Eric Hare <ericrhare@gmail.com>
Co-authored-by: Debojit Kaushik <Kaushik.debojit@gmail.com>
This commit is contained in:
Himavarsha
2026-05-28 16:47:12 -04:00
committed by GitHub
parent 5ca53ef994
commit ee9b6c2fca
9 changed files with 620 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
src/backend/base/langflow/api/v1/authz_route_dependencies.py Normal file
View File

@ -0,0 +1,85 @@
"""FastAPI dependencies that fetch a resource and enforce authorization in one step."""
from __future__ import annotations
from typing import Annotated
from uuid import UUID
from fastapi import Depends, HTTPException
from langflow.api.utils import CurrentActiveUser, DbSession
from langflow.api.v1.flows_helpers import _read_flow
from langflow.services.authorization import FlowAction, ensure_flow_permission
from langflow.services.authorization.fetch import deny_to_404
from langflow.services.database.models.flow.model import Flow, FlowCreate
async def _get_authorized_flow(
act: FlowAction,
*,
flow_id: UUID,
current_user: CurrentActiveUser,
session: DbSession,
) -> Flow:
"""Load a flow (share-aware when the plugin supports it) and enforce *act*."""
flow = await _read_flow(session, flow_id, current_user.id)
if flow is None:
raise HTTPException(status_code=404, detail="Flow not found")
try:
await ensure_flow_permission(
current_user,
act,
flow_id=flow_id,
flow_user_id=flow.user_id,
workspace_id=flow.workspace_id,
folder_id=flow.folder_id,
)
except HTTPException as exc:
raise deny_to_404(exc, detail="Flow not found") from exc
return flow
async def get_authorized_flow_for_read(
flow_id: UUID,
current_user: CurrentActiveUser,
session: DbSession,
) -> Flow:
"""Return a flow the caller may read (404 when denied or missing)."""
return await _get_authorized_flow(FlowAction.READ, flow_id=flow_id, current_user=current_user, session=session)
async def get_authorized_flow_for_write(
flow_id: UUID,
current_user: CurrentActiveUser,
session: DbSession,
) -> Flow:
"""Return a flow the caller may write (404 when denied or missing)."""
return await _get_authorized_flow(FlowAction.WRITE, flow_id=flow_id, current_user=current_user, session=session)
async def get_authorized_flow_for_delete(
flow_id: UUID,
current_user: CurrentActiveUser,
session: DbSession,
) -> Flow:
"""Return a flow the caller may delete (404 when denied or missing)."""
return await _get_authorized_flow(FlowAction.DELETE, flow_id=flow_id, current_user=current_user, session=session)
async def require_flow_create_permission(
current_user: CurrentActiveUser,
flow: FlowCreate,
) -> None:
"""Authorize CREATE at the destination workspace/folder before inserting a flow."""
await ensure_flow_permission(
current_user,
FlowAction.CREATE,
workspace_id=flow.workspace_id,
folder_id=flow.folder_id,
)
AuthorizedReadFlow = Annotated[Flow, Depends(get_authorized_flow_for_read)]
AuthorizedWriteFlow = Annotated[Flow, Depends(get_authorized_flow_for_write)]
AuthorizedDeleteFlow = Annotated[Flow, Depends(get_authorized_flow_for_delete)]
RequireFlowCreate = Annotated[None, Depends(require_flow_create_permission)]

92
src/backend/base/langflow/api/v1/flows.py
View File

@ -24,6 +24,12 @@ from langflow.api.utils import (
validate_is_component,
)
from langflow.api.utils.zip_utils import extract_flows_from_zip
from langflow.api.v1.authz_route_dependencies import (
AuthorizedDeleteFlow,
AuthorizedReadFlow,
AuthorizedWriteFlow,
RequireFlowCreate,
)
from langflow.api.v1.flows_helpers import (
_build_flows_download_response,
_get_safe_flow_path,
@ -96,12 +102,10 @@ async def create_flow(
session: DbSession,
flow: FlowCreate,
current_user: CurrentActiveUser,
_create: RequireFlowCreate,
storage_service: Annotated[StorageService, Depends(get_storage_service)],
):
try:
await ensure_flow_permission(
current_user, FlowAction.CREATE, workspace_id=flow.workspace_id, folder_id=flow.folder_id
)
return await _new_flow(session=session, flow=flow, user_id=current_user.id, storage_service=storage_service)
except HTTPException:
raise
@ -209,35 +213,18 @@ async def read_flows(
@router.get("/{flow_id}", response_model=FlowRead, status_code=200)
async def read_flow(
*,
session: DbSession,
flow_id: UUID,
current_user: CurrentActiveUser,
flow_id: UUID, # noqa: ARG001
flow: AuthorizedReadFlow,
):
"""Read a flow."""
# Share-aware fetch; plugin deny → 404 (UUID privacy).
if user_flow := await _read_flow(session, flow_id, current_user.id):
try:
await ensure_flow_permission(
current_user,
FlowAction.READ,
flow_id=flow_id,
flow_user_id=user_flow.user_id,
workspace_id=user_flow.workspace_id,
folder_id=user_flow.folder_id,
)
except HTTPException as exc:
raise deny_to_404(exc, detail="Flow not found") from exc
# Convert to FlowRead while session is still active to avoid detached instance errors
return FlowRead.model_validate(user_flow, from_attributes=True)
raise HTTPException(status_code=404, detail="Flow not found")
return FlowRead.model_validate(flow, from_attributes=True)
@router.get("/{flow_id}/note_translations", status_code=200)
async def get_note_translations(
*,
session: DbSession,
flow_id: UUID,
current_user: CurrentActiveUser,
flow_id: UUID, # noqa: ARG001
flow: AuthorizedReadFlow,
request: Request,
) -> dict[str, str]:
"""Return translated note node descriptions for the current locale.
@ -245,23 +232,16 @@ async def get_note_translations(
Returns a mapping of node_id → translated markdown text. Only nodes
with a matching translation key are included; nodes without translations
are omitted so the caller can leave them unchanged.
A missing or inaccessible flow yields 404 (via ``AuthorizedReadFlow``),
consistent with ``GET /flows/{id}``; the sole frontend caller (NoteNode)
treats that as "no translations" and renders the original text.
"""
from langflow.utils.i18n import translate
# Owner-scoped fetch; empty dict preserves UUID privacy vs 404.
flow = await _read_flow(session=session, flow_id=flow_id, user_id=current_user.id)
if not flow or not flow.data:
if not flow.data:
return {}
await ensure_flow_permission(
current_user,
FlowAction.READ,
flow_id=flow_id,
flow_user_id=flow.user_id,
workspace_id=flow.workspace_id,
folder_id=flow.folder_id,
)
locale = getattr(request.state, "locale", "en")
nodes = flow.data.get("nodes", [])
result: dict[str, str] = {}
@ -295,28 +275,13 @@ async def update_flow(
*,
session: DbSession,
flow_id: UUID,
db_flow: AuthorizedWriteFlow,
flow: FlowUpdate,
current_user: CurrentActiveUser,
storage_service: Annotated[StorageService, Depends(get_storage_service)],
):
"""Update a flow."""
try:
db_flow = await _read_flow(session=session, flow_id=flow_id, user_id=current_user.id)
if not db_flow:
raise HTTPException(status_code=404, detail="Flow not found")
try:
await ensure_flow_permission(
current_user,
FlowAction.WRITE,
flow_id=flow_id,
flow_user_id=db_flow.user_id,
workspace_id=db_flow.workspace_id,
folder_id=db_flow.folder_id,
)
except HTTPException as exc:
raise deny_to_404(exc, detail="Flow not found") from exc
# Destination check: if the payload moves the flow into a new
# workspace/folder, the caller must also be authorized to write at the
# destination scope. ``_patch_flow`` applies payload values via
@ -533,28 +498,11 @@ async def upsert_flow(
async def delete_flow(
*,
session: DbSession,
flow_id: UUID,
flow_id: UUID, # noqa: ARG001
flow: AuthorizedDeleteFlow,
current_user: CurrentActiveUser,
):
"""Delete a flow."""
flow = await _read_flow(
session=session,
flow_id=flow_id,
user_id=current_user.id,
)
if not flow:
raise HTTPException(status_code=404, detail="Flow not found")
try:
await ensure_flow_permission(
current_user,
FlowAction.DELETE,
flow_id=flow_id,
flow_user_id=flow.user_id,
workspace_id=flow.workspace_id,
folder_id=flow.folder_id,
)
except HTTPException as exc:
raise deny_to_404(exc, detail="Flow not found") from exc
await retry_flow_operation_on_deployment_guard(
db=session,
user_id=current_user.id,

68
src/backend/base/langflow/helpers/flow.py
View File

@ -20,6 +20,8 @@ if TYPE_CHECKING:
from lfx.graph.schema import RunOutputs
from lfx.graph.vertex.base import Vertex
from langflow.services.database.models.user.model import User
from langflow.schema.data import Data
INPUT_TYPE_MAP = {
@ -163,14 +165,34 @@ async def get_flow_by_id_or_name(
raise ValueError(msg) from e
async def _build_graph_from_authorized_flow(
*,
caller: User, # noqa: ARG001
flow: Flow,
flow_id: str,
user_id: str,
tweaks: dict | None,
) -> Graph:
"""Build a Graph from an already-loaded flow row (permission enforced by decorator)."""
from lfx.graph.graph.base import Graph
from langflow.processing.process import process_tweaks
graph_data = flow.data
if not graph_data:
msg = f"Flow {flow_id} not found"
raise ValueError(msg)
if tweaks:
graph_data = process_tweaks(graph_data=graph_data, tweaks=tweaks)
return Graph.from_payload(graph_data, flow_id=flow_id, user_id=user_id)
async def load_flow(
user_id: str, flow_id: str | None = None, flow_name: str | None = None, tweaks: dict | None = None
) -> Graph:
"""Load a flow graph after authorizing EXECUTE for the caller."""
from lfx.graph.graph.base import Graph
from langflow.processing.process import process_tweaks
from langflow.services.authorization import FlowAction, ensure_flow_permission
from langflow.services.authorization import FlowAction
from langflow.services.authorization.decorators import requires_flow_permission
from langflow.services.authorization.fetch import authorized_or_owner_scoped
from langflow.services.database.models.user.model import User
@ -199,36 +221,26 @@ async def load_flow(
msg = f"Flow {flow_id} not found"
raise ValueError(msg)
# Map plugin deny (403) to ValueError for existing callers.
caller = await session.get(User, uuid_user_id)
if caller is None:
msg = "Session is invalid"
raise ValueError(msg)
try:
await ensure_flow_permission(
caller,
FlowAction.EXECUTE,
flow_id=flow.id,
flow_user_id=flow.user_id,
workspace_id=flow.workspace_id,
folder_id=flow.folder_id,
)
except HTTPException as exc:
from fastapi import status as http_status
if exc.status_code == http_status.HTTP_403_FORBIDDEN:
msg = f"Flow {flow_id} not found"
raise ValueError(msg) from exc
raise
build_graph = requires_flow_permission(
FlowAction.EXECUTE,
user_param="caller",
flow_param="flow",
forbidden_as_not_found=True,
not_found_template=f"Flow {flow_id} not found",
)(_build_graph_from_authorized_flow)
graph_data = flow.data
if not graph_data:
msg = f"Flow {flow_id} not found"
raise ValueError(msg)
if tweaks:
graph_data = process_tweaks(graph_data=graph_data, tweaks=tweaks)
return Graph.from_payload(graph_data, flow_id=flow_id, user_id=user_id)
return await build_graph(
caller=caller,
flow=flow,
flow_id=flow_id,
user_id=user_id,
tweaks=tweaks,
)
async def find_flow(flow_name: str, user_id: str) -> str | None:

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

@ -13,6 +13,7 @@ from langflow.services.authorization.audit import (
audit_decision,
drain_pending_audit_writes,
)
from langflow.services.authorization.decorators import requires_flow_permission, requires_resource_permission
from langflow.services.authorization.fetch import authorized_or_owner_scoped, deny_to_404
from langflow.services.authorization.guards import (
ensure_deployment_permission,
@ -49,4 +50,6 @@ __all__ = [
"ensure_share_permission",
"ensure_variable_permission",
"filter_visible_resources",
"requires_flow_permission",
"requires_resource_permission",
]

281
src/backend/base/langflow/services/authorization/decorators.py Normal file
View File

@ -0,0 +1,281 @@
"""Declarative permission enforcement for non-route async call paths."""
from __future__ import annotations
import functools
import inspect
from collections.abc import Awaitable, Callable
from typing import TYPE_CHECKING, Any, TypeVar
from fastapi import HTTPException, status
from langflow.services.authorization.actions import (
DeploymentAction,
FileAction,
FlowAction,
KnowledgeBaseAction,
ProjectAction,
ShareAction,
VariableAction,
)
from langflow.services.authorization.guards import (
_RESOURCE_SPECS,
ensure_deployment_permission,
ensure_file_permission,
ensure_flow_permission,
ensure_knowledge_base_permission,
ensure_project_permission,
ensure_share_permission,
ensure_variable_permission,
)
if TYPE_CHECKING:
from langflow.services.database.models.user.model import User, UserRead
_Action = (
DeploymentAction
| FlowAction
| ProjectAction
| KnowledgeBaseAction
| VariableAction
| FileAction
| ShareAction
| str
)
_ENSURE_BY_SPEC: dict[str, Callable[..., Awaitable[None]]] = {
"flow": ensure_flow_permission,
"deployment": ensure_deployment_permission,
"project": ensure_project_permission,
"knowledge_base": ensure_knowledge_base_permission,
"variable": ensure_variable_permission,
"file": ensure_file_permission,
"share": ensure_share_permission,
}
# Map registry specs to ORM attribute names used when ``resource_param`` is set.
_ORM_ATTR_BY_SPEC_KW: dict[str, dict[str, str]] = {
"flow": {
"flow_id": "id",
"flow_user_id": "user_id",
"workspace_id": "workspace_id",
"folder_id": "folder_id",
},
"deployment": {
"deployment_id": "id",
"deployment_user_id": "user_id",
"workspace_id": "workspace_id",
"project_id": "project_id",
},
"project": {
"project_id": "id",
"project_user_id": "user_id",
"workspace_id": "workspace_id",
},
"knowledge_base": {
"kb_id": "id",
"kb_user_id": "user_id",
"workspace_id": "workspace_id",
"project_id": "project_id",
"kb_name": "name",
},
"variable": {
"variable_id": "id",
"variable_user_id": "user_id",
"workspace_id": "workspace_id",
},
"file": {
"file_id": "id",
"file_user_id": "user_id",
"workspace_id": "workspace_id",
},
"share": {
"share_id": "id",
"share_user_id": "user_id",
},
}
_F = TypeVar("_F", bound=Callable[..., Awaitable[Any]])
def _action_value(act: _Action) -> str:
"""Return the string value for enum-like and raw-string actions."""
return str(getattr(act, "value", act))
def _requires_resource_id(act: _Action) -> bool:
"""Return True when the action targets an existing resource."""
return _action_value(act) != "create"
def _raise_missing_resource_id(func_name: str, spec_key: str, id_kw: str) -> None:
msg = f"@requires_resource_permission: '{func_name}' requires a non-null '{id_kw}' for {spec_key} permission"
raise ValueError(msg)
def _kwargs_from_resource(spec_key: str, resource: Any) -> dict[str, Any]:
"""Build ensure_* kwargs from a loaded ORM row."""
spec = _RESOURCE_SPECS[spec_key]
attr_map = _ORM_ATTR_BY_SPEC_KW[spec_key]
out: dict[str, Any] = {}
if spec.id_kw in attr_map:
out[spec.id_kw] = getattr(resource, attr_map[spec.id_kw], None)
if spec.owner_kw in attr_map:
out[spec.owner_kw] = getattr(resource, attr_map[spec.owner_kw], None)
if spec.workspace_kw and spec.workspace_kw in attr_map:
out[spec.workspace_kw] = getattr(resource, attr_map[spec.workspace_kw], None)
if spec.scope_kw and spec.scope_kw in attr_map:
out[spec.scope_kw] = getattr(resource, attr_map[spec.scope_kw], None)
for extra in spec.extra_context_kws:
if extra in attr_map:
out[extra] = getattr(resource, attr_map[extra], None)
return out
def requires_resource_permission(
spec_key: str,
act: _Action,
*,
user_param: str = "user",
resource_param: str | None = None,
forbidden_as_not_found: bool = False,
not_found_template: str = "Resource not found",
) -> Callable[[_F], _F]:
"""Enforce a resource permission before the decorated async function runs.
When ``resource_param`` is set, permission kwargs are taken from that ORM
object (e.g. a loaded ``Flow``). Otherwise the decorated function must
expose parameters matching the registry's public kwarg names
(``flow_id``, ``flow_user_id``, ...).
``forbidden_as_not_found`` maps HTTP 403 to ``ValueError`` for helpers
that historically surfaced denials as "not found" (e.g. ``load_flow``).
"""
if spec_key not in _RESOURCE_SPECS:
msg = f"Unknown resource spec: {spec_key}"
raise ValueError(msg)
ensure_fn = _ENSURE_BY_SPEC[spec_key]
def decorator(func: _F) -> _F:
sig = inspect.signature(func)
if user_param not in sig.parameters:
msg = (
f"@requires_resource_permission: '{func.__name__}' must have "
f"a '{user_param}' parameter (got {list(sig.parameters)})"
)
raise TypeError(msg)
if resource_param is not None and resource_param not in sig.parameters:
msg = (
f"@requires_resource_permission: '{func.__name__}' must have "
f"a '{resource_param}' parameter when resource_param is set"
)
raise TypeError(msg)
spec = _RESOURCE_SPECS[spec_key]
if resource_param is None and _requires_resource_id(act) and spec.id_kw not in sig.parameters:
msg = (
f"@requires_resource_permission: '{func.__name__}' must have "
f"a '{spec.id_kw}' parameter for {spec_key} permission"
)
raise TypeError(msg)
@functools.wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Any:
bound = sig.bind(*args, **kwargs)
bound.apply_defaults()
arguments = bound.arguments
user: User | UserRead = arguments[user_param]
if resource_param is not None:
perm_kwargs = _kwargs_from_resource(spec_key, arguments[resource_param])
else:
perm_kwargs = {}
for kw in (
spec.id_kw,
spec.owner_kw,
*(spec.extra_context_kws or ()),
):
if kw in arguments:
perm_kwargs[kw] = arguments[kw]
if spec.workspace_kw and spec.workspace_kw in arguments:
perm_kwargs[spec.workspace_kw] = arguments[spec.workspace_kw]
if spec.scope_kw and spec.scope_kw in arguments:
perm_kwargs[spec.scope_kw] = arguments[spec.scope_kw]
if _requires_resource_id(act) and perm_kwargs.get(spec.id_kw) is None:
_raise_missing_resource_id(func.__name__, spec_key, spec.id_kw)
try:
await ensure_fn(user, act, **perm_kwargs)
except HTTPException as exc:
if forbidden_as_not_found and exc.status_code == status.HTTP_403_FORBIDDEN:
raise ValueError(not_found_template) from exc
raise
return await func(*args, **kwargs)
return wrapper # type: ignore[return-value]
return decorator
def requires_flow_permission(
act: FlowAction | str,
*,
user_param: str = "user",
flow_id_param: str = "flow_id",
flow_user_id_param: str = "flow_user_id",
flow_param: str | None = None,
forbidden_as_not_found: bool = False,
not_found_template: str = "Flow not found",
) -> Callable[[_F], _F]:
"""Flow-specific wrapper over :func:`requires_resource_permission`."""
if flow_param is not None:
return requires_resource_permission(
"flow",
act,
user_param=user_param,
resource_param=flow_param,
forbidden_as_not_found=forbidden_as_not_found,
not_found_template=not_found_template,
)
def decorator(func: _F) -> _F:
sig = inspect.signature(func)
if user_param not in sig.parameters:
msg = (
f"@requires_flow_permission: '{func.__name__}' must have "
f"a '{user_param}' parameter (got {list(sig.parameters)})"
)
raise TypeError(msg)
if _requires_resource_id(act) and flow_id_param not in sig.parameters:
msg = (
f"@requires_flow_permission: '{func.__name__}' must have "
f"a '{flow_id_param}' parameter for flow permission"
)
raise TypeError(msg)
@functools.wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Any:
bound = sig.bind(*args, **kwargs)
bound.apply_defaults()
arguments = bound.arguments
perm_kwargs = {
"flow_id": arguments.get(flow_id_param),
"flow_user_id": arguments.get(flow_user_id_param),
"workspace_id": arguments.get("workspace_id"),
"folder_id": arguments.get("folder_id"),
}
if _requires_resource_id(act) and perm_kwargs["flow_id"] is None:
_raise_missing_resource_id(func.__name__, "flow", "flow_id")
try:
await ensure_flow_permission(arguments[user_param], act, **perm_kwargs)
except HTTPException as exc:
if forbidden_as_not_found and exc.status_code == status.HTTP_403_FORBIDDEN:
raise ValueError(not_found_template) from exc
raise
return await func(*args, **kwargs)
return wrapper # type: ignore[return-value]
return decorator

123
src/backend/tests/unit/services/authorization/test_decorators.py Normal file
View File

@ -0,0 +1,123 @@
"""Tests for authorization decorators and route dependencies."""
from __future__ import annotations
from uuid import uuid4
import pytest
from fastapi import HTTPException
from langflow.services.authorization.actions import FlowAction
from langflow.services.authorization.decorators import requires_flow_permission
from ._common import install_audit_recorder, install_authz, install_settings
@pytest.mark.anyio
async def test_requires_flow_permission_denies_before_body(monkeypatch, fake_user):
"""Decorator runs ensure_flow_permission before the wrapped function."""
install_settings(monkeypatch, authz_enabled=True)
from ._common import _StubAuthorizationService
service = _StubAuthorizationService(allow=False)
install_authz(monkeypatch, service)
install_audit_recorder(monkeypatch)
flow_id = uuid4()
ran = False
@requires_flow_permission(
FlowAction.READ, user_param="user", flow_id_param="flow_id", flow_user_id_param="owner_id"
)
async def handler(*, user, flow_id, owner_id): # noqa: ARG001
nonlocal ran
ran = True
with pytest.raises(HTTPException) as exc_info:
await handler(user=fake_user, flow_id=flow_id, owner_id=uuid4())
assert exc_info.value.status_code == 403
assert ran is False
@pytest.mark.anyio
async def test_requires_flow_permission_maps_forbidden_to_value_error(monkeypatch, fake_user):
"""forbidden_as_not_found preserves load_flow's ValueError contract."""
install_settings(monkeypatch, authz_enabled=True)
from ._common import _StubAuthorizationService
service = _StubAuthorizationService(allow=False)
install_authz(monkeypatch, service)
install_audit_recorder(monkeypatch)
class _Flow:
id = uuid4()
user_id = uuid4()
workspace_id = None
folder_id = None
data = {"nodes": []}
@requires_flow_permission(
FlowAction.EXECUTE,
user_param="user",
flow_param="flow",
forbidden_as_not_found=True,
not_found_template="Flow missing",
)
async def handler(*, user, flow): # noqa: ARG001
return "ok"
with pytest.raises(ValueError, match="Flow missing"):
await handler(user=fake_user, flow=_Flow())
def test_requires_flow_permission_raises_on_missing_user_param():
"""Misconfigured decorators fail at decoration time."""
with pytest.raises(TypeError, match="must have a 'actor' parameter"):
@requires_flow_permission(FlowAction.READ, user_param="actor")
async def bad(*, user):
pass
def test_requires_flow_permission_raises_on_missing_flow_id_param():
"""Existing-resource actions must not silently authorize flow:*."""
with pytest.raises(TypeError, match="must have a 'missing_id' parameter"):
@requires_flow_permission(FlowAction.READ, user_param="user", flow_id_param="missing_id")
async def bad(*, user, flow_id):
pass
@pytest.mark.anyio
async def test_requires_flow_permission_rejects_none_flow_id_before_body(fake_user):
"""A None resource id fails closed before ensure_flow_permission can see flow:*."""
ran = False
@requires_flow_permission(FlowAction.READ, user_param="user", flow_id_param="flow_id")
async def handler(*, user, flow_id): # noqa: ARG001
nonlocal ran
ran = True
with pytest.raises(ValueError, match="non-null 'flow_id'"):
await handler(user=fake_user, flow_id=None)
assert ran is False
@pytest.mark.anyio
async def test_requires_flow_permission_resource_param_rejects_missing_id_before_body(fake_user):
"""Loaded-resource mode also fails closed when the row lacks an id."""
ran = False
class _Flow:
id = None
user_id = uuid4()
workspace_id = None
folder_id = None
@requires_flow_permission(FlowAction.EXECUTE, user_param="user", flow_param="flow")
async def handler(*, user, flow): # noqa: ARG001
nonlocal ran
ran = True
with pytest.raises(ValueError, match="non-null 'flow_id'"):
await handler(user=fake_user, flow=_Flow())
assert ran is False

77
src/backend/tests/unit/services/authorization/test_flow_route_guards.py
View File

@ -40,6 +40,20 @@ def _ensure_flow_permission_calls(func: ast.AsyncFunctionDef) -> list[ast.Call]:
return calls
def _uses_authorized_flow_dependency(func: ast.AsyncFunctionDef, alias: str) -> bool:
"""Return True if *func* declares an authorized-flow dependency alias in its signature."""
for arg in func.args.args + func.args.kwonlyargs:
if arg.annotation is None:
continue
ann = ast.unparse(arg.annotation)
if alias in ann:
return True
for node in ast.walk(func):
if isinstance(node, ast.AnnAssign) and node.annotation is not None and alias in ast.unparse(node.annotation):
return True
return False
def _action_arg(call: ast.Call) -> str | None:
"""Extract the action argument (positional[1]) as the dotted name like 'FlowAction.READ'."""
if len(call.args) < 2:
@ -72,9 +86,23 @@ def test_single_guard_route_uses_enum_action(routes, func_name, expected_action)
"""Single-call guard routes call ensure_flow_permission once with the right FlowAction."""
func = routes[func_name]
calls = _ensure_flow_permission_calls(func)
assert calls, f"{func_name} has no ensure_flow_permission call"
actions = {_action_arg(c) for c in calls}
assert expected_action in actions, f"{func_name} actions={actions}, expected {expected_action}"
if calls:
actions = {_action_arg(c) for c in calls}
assert expected_action in actions, f"{func_name} actions={actions}, expected {expected_action}"
return
# Routes migrated to FastAPI dependencies declare protection in the signature.
alias_by_action = {
"FlowAction.CREATE": "RequireFlowCreate",
"FlowAction.READ": "AuthorizedReadFlow",
"FlowAction.WRITE": "AuthorizedWriteFlow",
"FlowAction.DELETE": "AuthorizedDeleteFlow",
}
alias = alias_by_action.get(expected_action)
assert alias is not None
assert _uses_authorized_flow_dependency(func, alias), (
f"{func_name} has no ensure_flow_permission call or {alias} dependency"
)
def test_upsert_flow_guards_both_create_and_write(routes):
@ -224,34 +252,27 @@ def test_read_public_flow_remains_unguarded(routes):
def test_get_note_translations_is_owner_scoped_and_guarded(routes):
"""`GET /flows/{flow_id}/note_translations` must scope the fetch by owner and authorize READ.
Regression for the cross-user data leak found in the PR #13153 review: the
original handler used ``session.get(Flow, flow_id)`` with no ownership
filter and no ensure_flow_permission, so any authenticated user could read
note text from any flow by guessing the UUID. Fix scopes the fetch through
``_read_flow(..., user_id=current_user.id)`` and adds an explicit
``ensure_flow_permission(FlowAction.READ, ...)`` so authorization plugins can
extend visibility via shares.
"""
"""`GET /flows/{flow_id}/note_translations` must authorize READ via dependency or inline guard."""
func = routes["get_note_translations"]
calls = _ensure_flow_permission_calls(func)
assert calls, "get_note_translations is missing its ensure_flow_permission guard"
actions = {_action_arg(c) for c in calls}
assert "FlowAction.READ" in actions, f"expected FlowAction.READ, got {actions}"
if calls:
actions = {_action_arg(c) for c in calls}
assert "FlowAction.READ" in actions, f"expected FlowAction.READ, got {actions}"
read_flow_calls = [
node
for node in ast.walk(func)
if isinstance(node, ast.Call)
and (
(isinstance(node.func, ast.Name) and node.func.id == "_read_flow")
or (isinstance(node.func, ast.Attribute) and node.func.attr == "_read_flow")
)
]
assert read_flow_calls, "get_note_translations must fetch the flow via _read_flow (owner-scoped)"
return
# The fetch must go through _read_flow (which scopes by current_user.id),
# not a bare session.get(Flow, ...) call.
read_flow_calls = [
node
for node in ast.walk(func)
if isinstance(node, ast.Call)
and (
(isinstance(node.func, ast.Name) and node.func.id == "_read_flow")
or (isinstance(node.func, ast.Attribute) and node.func.attr == "_read_flow")
)
]
assert read_flow_calls, "get_note_translations must fetch the flow via _read_flow (owner-scoped)"
assert any(_uses_authorized_flow_dependency(func, alias) for alias in ("AuthorizedReadFlow",)), (
"get_note_translations must declare an authorized read dependency or call ensure_flow_permission"
)
def test_read_flows_list_uses_filter_visible_resources(routes):

24
src/backend/tests/unit/services/authorization/test_route_guard_regressions.py
View File

@ -140,12 +140,20 @@ def test_load_flow_calls_ensure_flow_permission(helpers_funcs):
"""
func = helpers_funcs["load_flow"]
calls = _calls(func, "ensure_flow_permission")
assert calls, "load_flow must call ensure_flow_permission"
# Action must be EXECUTE — the function loads a Graph for execution.
if calls:
saw_execute = False
for call in calls:
if len(call.args) >= 2:
arg = call.args[1]
if isinstance(arg, ast.Attribute) and arg.attr == "EXECUTE":
saw_execute = True
assert saw_execute, "load_flow must authorize FlowAction.EXECUTE"
return
requires_calls = _calls(func, "requires_flow_permission")
assert requires_calls, "load_flow must use requires_flow_permission or ensure_flow_permission"
saw_execute = False
for call in calls:
if len(call.args) >= 2:
arg = call.args[1]
if isinstance(arg, ast.Attribute) and arg.attr == "EXECUTE":
saw_execute = True
assert saw_execute, "load_flow must authorize FlowAction.EXECUTE"
for call in requires_calls:
if call.args and isinstance(call.args[0], ast.Attribute) and call.args[0].attr == "EXECUTE":
saw_execute = True
assert saw_execute, "load_flow must authorize FlowAction.EXECUTE via decorator"

7
src/backend/tests/unit/test_i18n_note_translation.py
View File

@ -200,9 +200,8 @@ async def test_note_translations_returns_translated_text_for_baked_node(
assert "Bonjour" in translations.values()
async def test_note_translations_returns_404_equivalent_for_missing_flow(client: AsyncClient, logged_in_headers):
"""Non-existent flow returns empty dict (endpoint returns {} not 404)."""
async def test_note_translations_returns_404_for_missing_flow(client: AsyncClient, logged_in_headers):
"""A non-existent (or inaccessible) flow returns 404, consistent with GET /flows/{id}."""
fake_id = str(uuid.uuid4())
resp = await client.get(f"api/v1/flows/{fake_id}/note_translations", headers=logged_in_headers)
assert resp.status_code == status.HTTP_200_OK
assert resp.json() == {}
assert resp.status_code == status.HTTP_404_NOT_FOUND