diff --git a/app/api/endpoints/workflow.py b/app/api/endpoints/workflow.py index 5c3351a4..2592f558 100644 --- a/app/api/endpoints/workflow.py +++ b/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: """ 删除工作流 diff --git a/app/db/user_oper.py b/app/db/user_oper.py index c413a61a..6fe1b193 100644 --- a/app/db/user_oper.py +++ b/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: diff --git a/tests/test_workflow_authorization.py b/tests/test_workflow_authorization.py new file mode 100644 index 00000000..bcd5c623 --- /dev/null +++ b/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