fix(security): require manage access for workflows (#6052)

app/api/endpoints/workflow.py

@@ -10,11 +10,14 @@ from app import schemas
 from app.chain.workflow import WorkflowChain
 from app.core.config import global_vars
 from app.core.plugin import PluginManager
-from app.core.security import verify_token
 from app.workflow import WorkFlowManager
 from app.db import get_async_db, get_db
-from app.db.models import Workflow
+from app.db.models import Workflow, User
 from app.db.systemconfig_oper import SystemConfigOper
+from app.db.user_oper import (
+    get_current_active_manage_user,
+    get_current_active_manage_user_async,
+)
 from app.db.workflow_oper import WorkflowOper
 from app.helper.server import MoviePilotServerHelper
 from app.scheduler import Scheduler
@@ -30,7 +33,7 @@ WORKFLOW_TRIGGER_MANUAL = "manual"
 @router.get("/", summary="所有工作流", response_model=List[schemas.Workflow])
 async def list_workflows(
     db: AsyncSession = Depends(get_async_db),
-    _: schemas.TokenPayload = Depends(verify_token),
+    _: User = Depends(get_current_active_manage_user_async),
 ) -> Any:
     """
     获取工作流列表
@@ -42,7 +45,7 @@ async def list_workflows(
 async def create_workflow(
     workflow: schemas.Workflow,
     db: AsyncSession = Depends(get_async_db),
-    _: schemas.TokenPayload = Depends(verify_token),
+    _: User = Depends(get_current_active_manage_user_async),
 ) -> Any:
     """
     创建工作流
@@ -62,7 +65,7 @@ async def create_workflow(
 
 @router.get("/plugin/actions", summary="查询插件动作", response_model=List[dict])
 def list_plugin_actions(
-    plugin_id: str = None, _: schemas.TokenPayload = Depends(verify_token)
+    plugin_id: str = None, _: User = Depends(get_current_active_manage_user)
 ) -> Any:
     """
     获取所有动作
@@ -71,7 +74,7 @@ def list_plugin_actions(
 
 
 @router.get("/actions", summary="所有动作", response_model=List[dict])
-async def list_actions(_: schemas.TokenPayload = Depends(verify_token)) -> Any:
+async def list_actions(_: User = Depends(get_current_active_manage_user_async)) -> Any:
     """
     获取所有动作
     """
@@ -79,7 +82,7 @@ async def list_actions(_: schemas.TokenPayload = Depends(verify_token)) -> Any:
 
 
 @router.get("/event_types", summary="获取所有事件类型", response_model=List[dict])
-async def get_event_types(_: schemas.TokenPayload = Depends(verify_token)) -> Any:
+async def get_event_types(_: User = Depends(get_current_active_manage_user_async)) -> Any:
     """
     获取所有事件类型
     """
@@ -94,7 +97,7 @@ async def get_event_types(_: schemas.TokenPayload = Depends(verify_token)) -> An
 
 @router.post("/share", summary="分享工作流", response_model=schemas.Response)
 async def workflow_share(
-    workflow: schemas.WorkflowShare, _: schemas.TokenPayload = Depends(verify_token)
+    workflow: schemas.WorkflowShare, _: User = Depends(get_current_active_manage_user_async)
 ) -> Any:
     """
     分享工作流
@@ -115,7 +118,7 @@ async def workflow_share(
 
 @router.delete("/share/{share_id}", summary="删除分享", response_model=schemas.Response)
 async def workflow_share_delete(
-    share_id: int, _: schemas.TokenPayload = Depends(verify_token)
+    share_id: int, _: User = Depends(get_current_active_manage_user_async)
 ) -> Any:
     """
     删除分享
@@ -128,7 +131,7 @@ async def workflow_share_delete(
 async def workflow_fork(
     workflow: schemas.WorkflowShare,
     db: AsyncSession = Depends(get_async_db),
-    _: schemas.User = Depends(verify_token),
+    _: User = Depends(get_current_active_manage_user_async),
 ) -> Any:
     """
     复用工作流
@@ -194,7 +197,7 @@ async def workflow_shares(
     name: Optional[str] = None,
     page: Optional[int] = 1,
     count: Optional[int] = 30,
-    _: schemas.TokenPayload = Depends(verify_token),
+    _: User = Depends(get_current_active_manage_user_async),
 ) -> Any:
     """
     查询分享的工作流
@@ -208,7 +211,7 @@ async def workflow_shares(
 def run_workflow(
     workflow_id: int,
     from_begin: Optional[bool] = True,
-    _: schemas.TokenPayload = Depends(verify_token),
+    _: User = Depends(get_current_active_manage_user),
 ) -> Any:
     """
     执行工作流
@@ -225,7 +228,7 @@ def run_workflow(
 def start_workflow(
     workflow_id: int,
     db: Session = Depends(get_db),
-    _: schemas.TokenPayload = Depends(verify_token),
+    _: User = Depends(get_current_active_manage_user),
 ) -> Any:
     """
     启用工作流
@@ -259,7 +262,7 @@ def start_workflow(
 def pause_workflow(
     workflow_id: int,
     db: Session = Depends(get_db),
-    _: schemas.TokenPayload = Depends(verify_token),
+    _: User = Depends(get_current_active_manage_user),
 ) -> Any:
     """
     停用工作流
@@ -287,7 +290,7 @@ def pause_workflow(
 async def reset_workflow(
     workflow_id: int,
     db: AsyncSession = Depends(get_async_db),
-    _: schemas.TokenPayload = Depends(verify_token),
+    _: User = Depends(get_current_active_manage_user_async),
 ) -> Any:
     """
     重置工作流
@@ -308,7 +311,7 @@ async def reset_workflow(
 async def get_workflow(
     workflow_id: int,
     db: AsyncSession = Depends(get_async_db),
-    _: schemas.TokenPayload = Depends(verify_token),
+    _: User = Depends(get_current_active_manage_user_async),
 ) -> Any:
     """
     获取工作流详情
@@ -320,7 +323,7 @@ async def get_workflow(
 def update_workflow(
     workflow: schemas.Workflow,
     db: Session = Depends(get_db),
-    _: schemas.TokenPayload = Depends(verify_token),
+    _: User = Depends(get_current_active_manage_user),
 ) -> Any:
     """
     更新工作流
@@ -350,7 +353,7 @@ def update_workflow(
 def delete_workflow(
     workflow_id: int,
     db: Session = Depends(get_db),
-    _: schemas.TokenPayload = Depends(verify_token),
+    _: User = Depends(get_current_active_manage_user),
 ) -> Any:
     """
     删除工作流

app/db/user_oper.py

@@ -58,6 +58,36 @@ async def get_current_active_user_async(
     return current_user
 
 
+def _ensure_manage_user(current_user: User) -> User:
+    """
+    校验用户具备全局管理权限。
+    """
+    permissions = current_user.permissions or {}
+    if not current_user.is_superuser and not bool(permissions.get("manage")):
+        raise HTTPException(
+            status_code=400, detail="用户权限不足"
+        )
+    return current_user
+
+
+def get_current_active_manage_user(
+        current_user: User = Depends(get_current_active_user),
+) -> User:
+    """
+    获取当前拥有管理权限的激活用户。
+    """
+    return _ensure_manage_user(current_user)
+
+
+async def get_current_active_manage_user_async(
+        current_user: User = Depends(get_current_active_user_async),
+) -> User:
+    """
+    异步获取当前拥有管理权限的激活用户。
+    """
+    return _ensure_manage_user(current_user)
+
+
 def get_current_active_superuser(
         current_user: User = Depends(get_current_user),
 ) -> User:

tests/test_workflow_authorization.py

@@ -0,0 +1,91 @@
+import asyncio
+import inspect
+from types import SimpleNamespace
+
+import pytest
+from fastapi import HTTPException
+from fastapi.routing import APIRoute
+
+from app.api.endpoints import workflow as workflow_endpoint
+from app.core.security import verify_token
+from app.db.user_oper import (
+    get_current_active_manage_user,
+    get_current_active_manage_user_async,
+    get_current_active_user,
+    get_current_active_user_async,
+)
+
+
+def _declared_dependencies(func):
+    """读取接口函数签名中直接声明的 FastAPI 依赖函数。"""
+    dependencies = []
+    for parameter in inspect.signature(func).parameters.values():
+        default = parameter.default
+        dependency = getattr(default, "dependency", None)
+        if dependency:
+            dependencies.append(dependency)
+    return dependencies
+
+
+def _workflow_routes():
+    """返回 Workflow API 当前注册的所有路由。"""
+    return [
+        route
+        for route in workflow_endpoint.router.routes
+        if isinstance(route, APIRoute)
+    ]
+
+
+@pytest.mark.parametrize(
+    "user",
+    [
+        SimpleNamespace(is_superuser=True, permissions={}),
+        SimpleNamespace(is_superuser=False, permissions={"manage": True}),
+    ],
+)
+def test_workflow_manage_dependency_allows_superuser_or_manage_user(user):
+    """Workflow 管理边界允许超级管理员或拥有 manage 权限的用户。"""
+    assert get_current_active_manage_user(current_user=user) is user
+    assert asyncio.run(get_current_active_manage_user_async(current_user=user)) is user
+
+
+@pytest.mark.parametrize("permissions", [None, {}, {"manage": False}])
+def test_workflow_manage_dependency_rejects_regular_user(permissions):
+    """Workflow 管理边界拒绝不具备 manage 权限的普通用户。"""
+    user = SimpleNamespace(is_superuser=False, permissions=permissions)
+
+    with pytest.raises(HTTPException) as sync_exc_info:
+        get_current_active_manage_user(current_user=user)
+    assert sync_exc_info.value.status_code == 400
+    assert sync_exc_info.value.detail == "用户权限不足"
+
+    with pytest.raises(HTTPException) as async_exc_info:
+        asyncio.run(get_current_active_manage_user_async(current_user=user))
+    assert async_exc_info.value.status_code == 400
+    assert async_exc_info.value.detail == "用户权限不足"
+
+
+def test_workflow_manage_dependencies_reuse_active_user_resolution():
+    """Workflow 管理依赖复用激活用户解析，保留未激活用户拒绝策略。"""
+    assert _declared_dependencies(get_current_active_manage_user) == [
+        get_current_active_user
+    ]
+    assert _declared_dependencies(get_current_active_manage_user_async) == [
+        get_current_active_user_async
+    ]
+
+
+def test_workflow_routes_require_manage_dependency_not_bare_verify_token():
+    """Workflow 路由必须使用管理权限依赖，不能直接裸用 verify_token。"""
+    routes = _workflow_routes()
+    assert routes
+
+    for route in routes:
+        dependencies = _declared_dependencies(route.endpoint)
+        assert verify_token not in dependencies, route.path
+        if inspect.iscoroutinefunction(route.endpoint):
+            assert get_current_active_manage_user_async in dependencies, route.path
+            assert get_current_active_manage_user not in dependencies, route.path
+        else:
+            assert get_current_active_manage_user in dependencies, route.path
+            assert get_current_active_manage_user_async not in dependencies, route.path