feat(authz): add MANAGE action gating for sensitive PATCH/PUT fields

Introduces a dedicated MANAGE action on FlowAction, DeploymentAction,
and ProjectAction so route handlers can require a strictly higher
privilege than WRITE when a payload touches administrative fields:

- New services/authorization/sensitive_fields.py defines the stable
  field sets:
    * SENSITIVE_FLOW_FIELDS = {locked, access_type, endpoint_name,
      webhook, mcp_enabled}
    * SENSITIVE_PROJECT_FIELDS = {auth_settings, parent_id}
    * SENSITIVE_DEPLOYMENT_FIELDS = frozenset() (enum exists for
      future use; no sensitive deployment field today)
- Route handlers compute required_action via the corresponding
  requires_*_manage(payload.model_fields_set) predicate and pass the
  result to ensure_*_permission. Applied to update_flow, upsert_flow,
  and update_project.
- OSS pass-through still allows everything; the enterprise plugin
  decides who has MANAGE. Lock-bypass is intentionally not an OSS env
  var - the enterprise policy is the single source of truth.

Tests: 25 unit tests for the action enums + 13 tests for the field
predicates. Updates test_flow_route_guards.py with a dynamic-action
resolver so AST assertions still recognize the new
required_action = MANAGE if ... else WRITE gating pattern.

Co-Authored-By: phact <estevezsebastian@gmail.com>
Co-Authored-By: Lucas Oliveira <62335616+lucaseduoli@users.noreply.github.com>
Based-On: langflow-ai/langflow#13280

src/backend/base/langflow/api/v1/flows.py

@@ -39,7 +39,12 @@ from langflow.api.v1.mappers.deployments.sync import retry_flow_operation_on_dep
 from langflow.api.v1.schemas import FlowListCreate
 from langflow.initial_setup.constants import STARTER_FOLDER_NAME
 from langflow.services.auth.utils import get_current_active_user
-from langflow.services.authorization import FlowAction, ensure_flow_permission, filter_visible_resources
+from langflow.services.authorization import (
+    FlowAction,
+    ensure_flow_permission,
+    filter_visible_resources,
+    requires_flow_manage,
+)
 from langflow.services.authorization.fetch import deny_to_404
 from langflow.services.authorization.utils import _resolve_casbin_domain
 from langflow.services.cache.service import ThreadingInMemoryCache
@@ -324,10 +329,16 @@ async def update_flow(
         if not db_flow:
             raise HTTPException(status_code=404, detail="Flow not found")
 
+        # Sensitive administrative fields (locked, access_type, endpoint_name,
+        # webhook, mcp_enabled) require FlowAction.MANAGE instead of WRITE.
+        # The OSS pass-through still allows everything; the enterprise plugin
+        # decides who has MANAGE.
+        required_action = FlowAction.MANAGE if requires_flow_manage(flow.model_fields_set) else FlowAction.WRITE
+
         try:
             await ensure_flow_permission(
                 current_user,
-                FlowAction.WRITE,
+                required_action,
                 flow_id=flow_id,
                 flow_user_id=db_flow.user_id,
                 workspace_id=db_flow.workspace_id,
@@ -347,7 +358,7 @@ async def update_flow(
             try:
                 await ensure_flow_permission(
                     current_user,
-                    FlowAction.WRITE,
+                    required_action,
                     flow_id=flow_id,
                     flow_user_id=db_flow.user_id,
                     workspace_id=target_workspace_id,
@@ -375,7 +386,7 @@ async def update_flow(
             try:
                 await ensure_flow_permission(
                     current_user,
-                    FlowAction.WRITE,
+                    required_action,
                     flow_id=flow_id,
                     flow_user_id=db_flow_for_attempt.user_id,
                     workspace_id=db_flow_for_attempt.workspace_id,
@@ -394,7 +405,7 @@ async def update_flow(
                 try:
                     await ensure_flow_permission(
                         current_user,
-                        FlowAction.WRITE,
+                        required_action,
                         flow_id=flow_id,
                         flow_user_id=db_flow_for_attempt.user_id,
                         workspace_id=attempt_target_workspace_id,
@@ -464,10 +475,17 @@ async def upsert_flow(
             if not can_widen and existing_flow.user_id != current_user.id:
                 raise HTTPException(status_code=404, detail="Flow not found")
 
+            # Upgrade WRITE → MANAGE when the incoming payload touches any
+            # administrative field (locked, access_type, endpoint_name,
+            # webhook, mcp_enabled). model_fields_set reflects only the
+            # fields the client actually sent so default values do not
+            # accidentally escalate the required action.
+            required_action = FlowAction.MANAGE if requires_flow_manage(flow.model_fields_set) else FlowAction.WRITE
+
             try:
                 await ensure_flow_permission(
                     current_user,
-                    FlowAction.WRITE,
+                    required_action,
                     flow_id=flow_id,
                     flow_user_id=existing_flow.user_id,
                     workspace_id=existing_flow.workspace_id,
@@ -487,7 +505,7 @@ async def upsert_flow(
                 try:
                     await ensure_flow_permission(
                         current_user,
-                        FlowAction.WRITE,
+                        required_action,
                         flow_id=flow_id,
                         flow_user_id=existing_flow.user_id,
                         workspace_id=target_workspace_id,

src/backend/base/langflow/api/v1/projects.py

@@ -37,6 +37,7 @@ from langflow.services.authorization import (
     ProjectAction,
     ensure_project_permission,
     filter_visible_resources,
+    requires_project_manage,
 )
 from langflow.services.authorization.fetch import authorized_or_owner_scoped, deny_to_404
 from langflow.services.authorization.utils import _resolve_casbin_domain
@@ -393,10 +394,14 @@ async def update_project(
     if not existing_project:
         raise HTTPException(status_code=404, detail="Project not found")
 
+    # Reparenting (parent_id) and governance toggles (auth_settings) require
+    # ProjectAction.MANAGE — a strictly higher privilege than WRITE.
+    project_action = ProjectAction.MANAGE if requires_project_manage(project.model_fields_set) else ProjectAction.WRITE
+
     try:
         await ensure_project_permission(
             current_user,
-            ProjectAction.WRITE,
+            project_action,
             project_id=project_id,
             project_user_id=existing_project.user_id,
             workspace_id=existing_project.workspace_id,

src/backend/base/langflow/services/authorization/__init__.py

@@ -10,6 +10,14 @@ from langflow.services.authorization.actions import (
     VariableAction,
 )
 from langflow.services.authorization.fetch import authorized_or_owner_scoped, deny_to_404
+from langflow.services.authorization.sensitive_fields import (
+    SENSITIVE_DEPLOYMENT_FIELDS,
+    SENSITIVE_FLOW_FIELDS,
+    SENSITIVE_PROJECT_FIELDS,
+    requires_deployment_manage,
+    requires_flow_manage,
+    requires_project_manage,
+)
 from langflow.services.authorization.service import LangflowAuthorizationService
 from langflow.services.authorization.utils import (
     audit_decision,
@@ -25,6 +33,9 @@ from langflow.services.authorization.utils import (
 )
 
 __all__ = [
+    "SENSITIVE_DEPLOYMENT_FIELDS",
+    "SENSITIVE_FLOW_FIELDS",
+    "SENSITIVE_PROJECT_FIELDS",
     "DeploymentAction",
     "FileAction",
     "FlowAction",
@@ -45,4 +56,7 @@ __all__ = [
     "ensure_share_permission",
     "ensure_variable_permission",
     "filter_visible_resources",
+    "requires_deployment_manage",
+    "requires_flow_manage",
+    "requires_project_manage",
 ]

src/backend/base/langflow/services/authorization/actions.py

@@ -11,6 +11,10 @@ class FlowAction(str, Enum):
     Values are the Casbin ``act`` strings consumed by the enterprise policy
     engine. Subclassing ``str`` lets these be passed wherever a bare string is
     accepted, so callers can migrate incrementally.
+
+    ``MANAGE`` is a strictly higher privilege than ``WRITE``: route handlers
+    gate it on PATCH/PUT payloads that touch sensitive administrative fields
+    (see :data:`langflow.services.authorization.sensitive_fields.SENSITIVE_FLOW_FIELDS`).
     """
 
     READ = "read"
@@ -19,16 +23,23 @@ class FlowAction(str, Enum):
     DELETE = "delete"
     EXECUTE = "execute"
     DEPLOY = "deploy"
+    MANAGE = "manage"
 
 
 class DeploymentAction(str, Enum):
-    """Actions that can be authorized on a deployment resource."""
+    """Actions that can be authorized on a deployment resource.
+
+    ``MANAGE`` mirrors :attr:`FlowAction.MANAGE`. No deployment field maps to
+    it today; the action exists so the enterprise plugin has a stable verb
+    to grant when sensitive deployment fields land.
+    """
 
     READ = "read"
     WRITE = "write"
     CREATE = "create"
     DELETE = "delete"
     EXECUTE = "execute"
+    MANAGE = "manage"
 
 
 class ProjectAction(str, Enum):
@@ -37,12 +48,17 @@ class ProjectAction(str, Enum):
     Projects are the OSS-side persistent name for the folder model; the
     ``project:`` Casbin object prefix matches the API surface and the
     ``project:{folder_id}`` domain used by ``_resolve_flow_domain``.
+
+    ``MANAGE`` mirrors :attr:`FlowAction.MANAGE`. Gated on PATCH payloads
+    that touch governance / hierarchy fields — see
+    :data:`langflow.services.authorization.sensitive_fields.SENSITIVE_PROJECT_FIELDS`.
     """
 
     READ = "read"
     WRITE = "write"
     CREATE = "create"
     DELETE = "delete"
+    MANAGE = "manage"
 
 
 class KnowledgeBaseAction(str, Enum):

src/backend/base/langflow/services/authorization/sensitive_fields.py

@@ -0,0 +1,67 @@
+"""Canonical sets of administrative resource fields gated on the ``MANAGE`` action.
+
+Route PATCH/PUT handlers consult these sets to decide whether a payload
+requires :attr:`FlowAction.MANAGE` (resp. ``DeploymentAction.MANAGE`` /
+``ProjectAction.MANAGE``) instead of the regular ``WRITE`` action. Keep these
+sets small and stable — the enterprise plugin's policy contract depends on
+operators reasoning about a fixed set of higher-privilege fields, not a
+moving target.
+
+Why a single ``MANAGE`` per resource (instead of per-field actions)?
+- One verb keeps the Casbin policy surface compact.
+- The enterprise plugin can still attach fine-grained policies by reading
+  ``AuthzContext`` (e.g. the resource id / owner) when MANAGE is requested.
+- New sensitive fields land here without forcing a new enum value or
+  invalidating existing policies.
+"""
+
+from __future__ import annotations
+
+from typing import Final
+
+# Flow administrative fields:
+#  - locked: edit-lock toggle; bypassing the lock is a higher-privilege action.
+#  - access_type: PRIVATE/PUBLIC visibility toggle; widens external exposure.
+#  - endpoint_name: public/webhook URL slug; rebinding it can hijack callers.
+#  - webhook: enables the webhook endpoint surface.
+#  - mcp_enabled: exposes the flow over the MCP server.
+SENSITIVE_FLOW_FIELDS: Final[frozenset[str]] = frozenset(
+    {
+        "locked",
+        "access_type",
+        "endpoint_name",
+        "webhook",
+        "mcp_enabled",
+    }
+)
+
+# Deployment administrative fields. Empty today: PATCH only accepts
+# name/description/provider_data (provider-opaque). MANAGE is still
+# introduced on the action enum so policies can grant it once sensitive
+# deployment fields exist.
+SENSITIVE_DEPLOYMENT_FIELDS: Final[frozenset[str]] = frozenset()
+
+# Project (folder) administrative fields:
+#  - auth_settings: governance/policy JSON dict.
+#  - parent_id: reparenting changes folder hierarchy / domain scope.
+SENSITIVE_PROJECT_FIELDS: Final[frozenset[str]] = frozenset(
+    {
+        "auth_settings",
+        "parent_id",
+    }
+)
+
+
+def requires_flow_manage(payload_field_set: set[str] | frozenset[str]) -> bool:
+    """Return True when a flow PATCH/PUT payload touches a MANAGE-gated field."""
+    return bool(SENSITIVE_FLOW_FIELDS & payload_field_set)
+
+
+def requires_deployment_manage(payload_field_set: set[str] | frozenset[str]) -> bool:
+    """Return True when a deployment PATCH/PUT payload touches a MANAGE-gated field."""
+    return bool(SENSITIVE_DEPLOYMENT_FIELDS & payload_field_set)
+
+
+def requires_project_manage(payload_field_set: set[str] | frozenset[str]) -> bool:
+    """Return True when a project PATCH/PUT payload touches a MANAGE-gated field."""
+    return bool(SENSITIVE_PROJECT_FIELDS & payload_field_set)

src/backend/tests/unit/services/authorization/test_actions.py

@@ -7,6 +7,7 @@ from langflow.services.authorization.actions import (
     FileAction,
     FlowAction,
     KnowledgeBaseAction,
+    ProjectAction,
     ShareAction,
     VariableAction,
 )
@@ -20,18 +21,20 @@ def test_flow_action_values_match_casbin_strings():
     assert FlowAction.DELETE.value == "delete"
     assert FlowAction.EXECUTE.value == "execute"
     assert FlowAction.DEPLOY.value == "deploy"
+    assert FlowAction.MANAGE.value == "manage"
 
 
 def test_flow_action_subclasses_str():
     """Subclassing str lets the enum be passed wherever a string is accepted."""
     assert isinstance(FlowAction.READ, str)
     assert FlowAction.WRITE == "write"
+    assert FlowAction.MANAGE == "manage"
 
 
 def test_flow_action_is_iterable_and_complete():
-    """The enum exposes exactly the six canonical actions."""
+    """The enum exposes the seven canonical actions including MANAGE."""
     values = {member.value for member in FlowAction}
-    assert values == {"read", "write", "create", "delete", "execute", "deploy"}
+    assert values == {"read", "write", "create", "delete", "execute", "deploy", "manage"}
 
 
 def test_deployment_action_values_match_casbin_strings():
@@ -41,6 +44,7 @@ def test_deployment_action_values_match_casbin_strings():
     assert DeploymentAction.CREATE.value == "create"
     assert DeploymentAction.DELETE.value == "delete"
     assert DeploymentAction.EXECUTE.value == "execute"
+    assert DeploymentAction.MANAGE.value == "manage"
 
 
 def test_deployment_action_subclasses_str():
@@ -50,9 +54,16 @@ def test_deployment_action_subclasses_str():
 
 
 def test_deployment_action_is_iterable_and_complete():
-    """The enum exposes exactly the five canonical deployment actions (no DEPLOY)."""
+    """The enum exposes the six canonical deployment actions (no DEPLOY)."""
     values = {member.value for member in DeploymentAction}
-    assert values == {"read", "write", "create", "delete", "execute"}
+    assert values == {"read", "write", "create", "delete", "execute", "manage"}
+
+
+def test_project_action_includes_manage():
+    """Projects expose MANAGE for reparenting / auth_settings changes."""
+    values = {member.value for member in ProjectAction}
+    assert values == {"read", "write", "create", "delete", "manage"}
+    assert ProjectAction.MANAGE == "manage"
 
 
 def test_knowledge_base_action_values():

src/backend/tests/unit/services/authorization/test_flow_route_guards.py

@@ -47,7 +47,12 @@ def _ensure_flow_permission_calls(func: ast.AsyncFunctionDef) -> list[ast.Call]:
 
 
 def _action_arg(call: ast.Call) -> str | None:
-    """Extract the action argument (positional[1]) as the dotted name like 'FlowAction.READ'."""
+    """Extract the action argument (positional[1]) as the dotted name like 'FlowAction.READ'.
+
+    Also recognizes :class:`ast.Name` references (e.g. ``required_action``)
+    used by the dynamic MANAGE/WRITE gating pattern and returns them in the
+    ``var:NAME`` form so resolution can find the underlying enum branches.
+    """
     if len(call.args) < 2:
         return None
     arg = call.args[1]
@@ -55,9 +60,51 @@ def _action_arg(call: ast.Call) -> str | None:
         return f"{arg.value.id}.{arg.attr}"
     if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
         return arg.value
+    if isinstance(arg, ast.Name):
+        return f"var:{arg.id}"
     return None
 
 
+def _resolve_var_action_branches(func: ast.AsyncFunctionDef, var_name: str) -> set[str]:
+    """Find ``var_name = X if cond else Y`` assignments and return both enum branches.
+
+    Used to expand the dynamic ``required_action`` / ``project_action`` pattern
+    introduced for MANAGE-gated PATCH/PUT handlers so the regression tests can
+    still assert that the underlying WRITE/MANAGE enum values are reachable.
+    """
+    actions: set[str] = set()
+    for node in ast.walk(func):
+        if not isinstance(node, ast.Assign):
+            continue
+        if not any(isinstance(target, ast.Name) and target.id == var_name for target in node.targets):
+            continue
+        value = node.value
+        candidates: list[ast.expr] = [value.body, value.orelse] if isinstance(value, ast.IfExp) else [value]
+        for candidate in candidates:
+            if isinstance(candidate, ast.Attribute) and isinstance(candidate.value, ast.Name):
+                actions.add(f"{candidate.value.id}.{candidate.attr}")
+    return actions
+
+
+def _resolved_actions(func: ast.AsyncFunctionDef, calls: list[ast.Call]) -> set[str]:
+    """Return the set of enum action names reachable from *calls* inside *func*.
+
+    Resolves ``var:NAME`` placeholders via :func:`_resolve_var_action_branches`
+    so dynamic gating patterns (e.g. ``required_action = MANAGE if ... else WRITE``)
+    appear in the result set as both enum branches.
+    """
+    resolved: set[str] = set()
+    for call in calls:
+        action = _action_arg(call)
+        if action is None:
+            continue
+        if action.startswith("var:"):
+            resolved.update(_resolve_var_action_branches(func, action[len("var:") :]))
+        else:
+            resolved.add(action)
+    return resolved
+
+
 @pytest.fixture(scope="module")
 def routes() -> dict[str, ast.AsyncFunctionDef]:
     return _parse_flow_routes()
@@ -75,33 +122,49 @@ def routes() -> dict[str, ast.AsyncFunctionDef]:
     ],
 )
 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."""
+    """Single-call guard routes call ensure_flow_permission once with the right FlowAction.
+
+    Resolves dynamic ``required_action = MANAGE if ... else WRITE`` patterns
+    introduced for MANAGE-gated PATCH/PUT handlers so the WRITE branch still
+    shows up in the resolved set.
+    """
     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}
+    actions = _resolved_actions(func, calls)
     assert expected_action in actions, f"{func_name} actions={actions}, expected {expected_action}"
 
 
+def test_update_flow_gates_sensitive_fields_on_manage(routes):
+    """PATCH /flows/{flow_id} must reach FlowAction.MANAGE for the sensitive-field branch."""
+    func = routes["update_flow"]
+    actions = _resolved_actions(func, _ensure_flow_permission_calls(func))
+    assert "FlowAction.MANAGE" in actions, (
+        f"update_flow must gate sensitive fields on MANAGE; resolved actions={actions}"
+    )
+
+
 def test_upsert_flow_guards_both_create_and_write(routes):
-    """upsert_flow has two guard branches — WRITE for existing flows, CREATE for new ones."""
+    """upsert_flow has two guard branches — WRITE/MANAGE for existing flows, CREATE for new ones."""
     func = routes["upsert_flow"]
-    actions = {_action_arg(c) for c in _ensure_flow_permission_calls(func)}
+    actions = _resolved_actions(func, _ensure_flow_permission_calls(func))
     assert "FlowAction.WRITE" in actions
     assert "FlowAction.CREATE" in actions
+    # Upsert reuses the same MANAGE-gating pattern as PATCH for the update branch.
+    assert "FlowAction.MANAGE" in actions, f"upsert_flow must reach MANAGE for sensitive fields; got {actions}"
 
 
 def test_delete_multiple_flows_guards_each_flow(routes):
     """Bulk delete iterates and calls ensure_flow_permission per loaded flow."""
     func = routes["delete_multiple_flows"]
-    actions = {_action_arg(c) for c in _ensure_flow_permission_calls(func)}
+    actions = _resolved_actions(func, _ensure_flow_permission_calls(func))
     assert actions == {"FlowAction.DELETE"}
 
 
 def test_download_multiple_file_guards_each_flow(routes):
     """Bulk download iterates and calls ensure_flow_permission per loaded flow."""
     func = routes["download_multiple_file"]
-    actions = {_action_arg(c) for c in _ensure_flow_permission_calls(func)}
+    actions = _resolved_actions(func, _ensure_flow_permission_calls(func))
     assert actions == {"FlowAction.READ"}
 
 
@@ -176,9 +239,16 @@ def _has_destination_check(
     target_workspace_kw: str,
     target_folder_kw: str,
 ) -> bool:
-    """Return True if *func* has a WRITE ensure_flow_permission call with the destination kwargs."""
+    """Return True if *func* has a WRITE/MANAGE ensure_flow_permission call with the destination kwargs.
+
+    Accepts either the literal ``FlowAction.WRITE`` / ``FlowAction.MANAGE`` or a
+    variable reference (``required_action``) used by the dynamic gating pattern.
+    """
     for call in _ensure_flow_permission_calls(func):
-        if _action_arg(call) != "FlowAction.WRITE":
+        action = _action_arg(call)
+        if action not in {"FlowAction.WRITE", "FlowAction.MANAGE"} and not (
+            action is not None and action.startswith("var:")
+        ):
             continue
         ws = _kwarg_source(call, "workspace_id")
         fld = _kwarg_source(call, "folder_id")
@@ -213,13 +283,20 @@ def test_upsert_flow_update_branch_authorizes_destination_on_move(routes):
 
 
 def test_no_bare_string_actions_remain(routes):
-    """No flow-route guard should use a bare string action — the enum is now canonical."""
+    """No flow-route guard should use a bare string action — the enum is now canonical.
+
+    Variable references (``var:NAME``) for the dynamic MANAGE/WRITE gating
+    pattern are accepted because each branch resolves to a ``FlowAction.*``
+    enum value (verified by :func:`test_update_flow_gates_sensitive_fields_on_manage`
+    and the per-route assertions above).
+    """
     offenders: list[str] = []
     for name, func in routes.items():
         for call in _ensure_flow_permission_calls(func):
             arg = _action_arg(call)
-            if arg is not None and not arg.startswith("FlowAction."):
-                offenders.append(f"{name}:{arg}")
+            if arg is None or arg.startswith(("FlowAction.", "var:")):
+                continue
+            offenders.append(f"{name}:{arg}")
     assert offenders == [], f"bare-string actions found: {offenders}"
 
 
@@ -315,9 +392,10 @@ def test_execute_surfaces_guard_with_flow_execute(module_path, func_name):
     """build/run/webhook handlers must call ensure_flow_permission(FlowAction.EXECUTE)."""
     funcs = _parse_async_funcs(module_path)
     assert func_name in funcs, f"{func_name} missing from {module_path.name}"
-    calls = _ensure_flow_permission_calls(funcs[func_name])
+    func = funcs[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}
+    actions = _resolved_actions(func, calls)
     assert "FlowAction.EXECUTE" in actions, f"{func_name} actions={actions}, expected FlowAction.EXECUTE"
 
 
@@ -371,10 +449,15 @@ def _ensure_project_permission_calls(func: ast.AsyncFunctionDef) -> list[ast.Cal
     ],
 )
 def test_project_routes_guarded(func_name, expected_action):
-    """Each project CRUD handler calls ensure_project_permission with the right ProjectAction."""
+    """Each project CRUD handler calls ensure_project_permission with the right ProjectAction.
+
+    Dynamic ``project_action = MANAGE if ... else WRITE`` is resolved so the
+    WRITE branch still satisfies the existing assertions.
+    """
     funcs = _parse_async_funcs(_PROJECTS_FILE)
     assert func_name in funcs, f"{func_name} missing from projects.py"
-    calls = _ensure_project_permission_calls(funcs[func_name])
+    func = funcs[func_name]
+    calls = _ensure_project_permission_calls(func)
     assert calls, f"{func_name} has no ensure_project_permission call"
-    actions = {_action_arg(c) for c in calls}
+    actions = _resolved_actions(func, calls)
     assert expected_action in actions, f"{func_name} actions={actions}, expected {expected_action}"

src/backend/tests/unit/services/authorization/test_sensitive_fields.py

@@ -0,0 +1,71 @@
+"""Tests for the sensitive-field sets that gate MANAGE-level actions."""
+
+from __future__ import annotations
+
+import pytest
+from langflow.services.authorization import (
+    SENSITIVE_DEPLOYMENT_FIELDS,
+    SENSITIVE_FLOW_FIELDS,
+    SENSITIVE_PROJECT_FIELDS,
+    requires_deployment_manage,
+    requires_flow_manage,
+    requires_project_manage,
+)
+
+
+def test_flow_sensitive_fields_locked_in():
+    """The contract for flow MANAGE is exactly these five administrative fields."""
+    assert (
+        frozenset(
+            {"locked", "access_type", "endpoint_name", "webhook", "mcp_enabled"},
+        )
+        == SENSITIVE_FLOW_FIELDS
+    )
+
+
+def test_project_sensitive_fields_locked_in():
+    """Projects gate MANAGE on governance (auth_settings) and reparenting (parent_id)."""
+    assert frozenset({"auth_settings", "parent_id"}) == SENSITIVE_PROJECT_FIELDS
+
+
+def test_deployment_sensitive_fields_empty_today():
+    """No deployment PATCH field is administrative yet; the set is intentionally empty."""
+    assert frozenset() == SENSITIVE_DEPLOYMENT_FIELDS
+
+
+@pytest.mark.parametrize(
+    ("payload_field_set", "expected"),
+    [
+        (set(), False),
+        ({"name"}, False),
+        ({"name", "description"}, False),
+        ({"locked"}, True),
+        ({"name", "access_type"}, True),
+        ({"endpoint_name"}, True),
+        ({"webhook", "name"}, True),
+        ({"mcp_enabled"}, True),
+    ],
+)
+def test_requires_flow_manage(payload_field_set, expected):
+    assert requires_flow_manage(payload_field_set) is expected
+
+
+@pytest.mark.parametrize(
+    ("payload_field_set", "expected"),
+    [
+        (set(), False),
+        ({"name"}, False),
+        ({"description"}, False),
+        ({"parent_id"}, True),
+        ({"auth_settings"}, True),
+        ({"name", "parent_id"}, True),
+    ],
+)
+def test_requires_project_manage(payload_field_set, expected):
+    assert requires_project_manage(payload_field_set) is expected
+
+
+def test_requires_deployment_manage_always_false_today():
+    """Until sensitive deployment fields exist, the predicate is a constant False."""
+    assert requires_deployment_manage(set()) is False
+    assert requires_deployment_manage({"name", "description", "provider_data"}) is False