Add interface to get all menus of a role (#183)

* Add interface to get all menus of a role

* deduplicate data classes

* update get role menu routing

* update casbin authentication

backend/app/api/v1/role.py

@@ -10,6 +10,7 @@ from backend.app.common.pagination import PageDepends, paging_data
 from backend.app.common.response.response_schema import response_base
 from backend.app.database.db_mysql import CurrentSession
 from backend.app.schemas.role import GetAllRole, CreateRole, UpdateRole, UpdateRoleMenu
+from backend.app.services.menu_service import MenuService
 from backend.app.services.role_service import RoleService
 from backend.app.utils.serializers import select_to_json, select_to_list
 
@@ -30,6 +31,12 @@ async def get_user_all_roles(pk: int):
     return await response_base.success(data=data)
 
 
+@router.get('/{pk}/menus', summary='获取角色所有菜单', dependencies=[DependsJwtAuth])
+async def get_role_all_menus(pk: int):
+    menu = await MenuService.get_role_menu_tree(pk=pk)
+    return await response_base.success(data=menu)
+
+
 @router.get('/{pk}', summary='获取角色详情', dependencies=[DependsJwtAuth])
 async def get_role(pk: int):
     role = await RoleService.get(pk=pk)

backend/app/common/casbin_rbac.py

@@ -4,6 +4,7 @@ import casbin
 import casbin_async_sqlalchemy_adapter
 from fastapi import Request, Depends
 
+from backend.app.common.enums import StatusType
 from backend.app.common.exception.errors import AuthorizationError, TokenError
 from backend.app.common.jwt import DependsJwtAuth
 from backend.app.core.conf import settings
@@ -43,6 +44,8 @@ class RBAC:
             return
         # 检测角色数据权限范围
         user_roles = request.user.roles
+        if not user_roles:
+            raise AuthorizationError(msg='用户未分配角色，授权失败')
         data_scope = any(role.data_scope == 1 for role in user_roles)
         if data_scope:
             return
@@ -50,16 +53,27 @@ class RBAC:
             # 菜单权限校验
             path_auth = request.url.path.replace(f'{settings.API_V1_STR}', '').replace('/', ':')
             menu_perms = []
+            forbid_menu_perms = []
             for role in user_roles:
-                menu_perms.extend([menu.perms for menu in role.menus])
+                for menu in role.menus:
+                    menu_perms.append(menu.perms) if menu.status == StatusType.enable else forbid_menu_perms.append(
+                        menu.perms
+                    )
             if path_auth in set(settings.MENU_EXCLUDE):
                 return
-            if path_auth not in set(menu_perms):
+            if path_auth in set([perm for perms_str in forbid_menu_perms for perm in perms_str.split(',')]):
+                raise AuthorizationError(msg='菜单已禁用，授权失败')
+            if path_auth not in set([perm for perms_str in menu_perms for perm in perms_str.split(',')]):
                 raise AuthorizationError
         else:
             # casbin 权限校验
             method = request.method
             path = request.url.path
+            forbid_menu_path = [
+                menu.path for role in user_roles for menu in role.menus if menu.status == StatusType.disable
+            ]
+            if path.split('/')[-1] in forbid_menu_path:
+                raise AuthorizationError(msg='菜单已禁用，授权失败')
             if (method, path) in settings.CASBIN_EXCLUDE:
                 return
             user_uuid = request.user.uuid

backend/app/crud/crud_menu.py

@@ -30,10 +30,7 @@ class CRUDMenu(CRUDBase[Menu, CreateMenu, UpdateMenu]):
 
     async def get_role_menus(self, db, superuser: bool, menu_ids: list[int]) -> list[Menu]:
         se = select(self.model).order_by(asc(self.model.sort))
-        where_list = [
-            self.model.menu_type.in_([0, 1]),
-            self.model.status == 1,
-        ]
+        where_list = [self.model.menu_type.in_([0, 1])]
         if not superuser:
             where_list.append(self.model.id.in_(menu_ids))
         se = se.where(and_(*where_list))

backend/app/schemas/role.py

@@ -28,10 +28,6 @@ class UpdateRoleMenu(SchemaBase):
     menus: list[int]
 
 
-class UpdateRoleMenu(SchemaBase):
-    menus: list[int]
-
-
 class GetAllRole(RoleBase):
     id: int
     created_time: datetime

backend/app/services/menu_service.py

@@ -3,6 +3,7 @@
 from fastapi import Request
 from backend.app.common.exception import errors
 from backend.app.crud.crud_menu import MenuDao
+from backend.app.crud.crud_role import RoleDao
 from backend.app.database.db_mysql import async_db_session
 from backend.app.schemas.menu import CreateMenu, UpdateMenu
 from backend.app.utils.build_tree import get_tree_data
@@ -24,6 +25,17 @@ class MenuService:
             menu_tree = await get_tree_data(menu_select)
             return menu_tree
 
+    @staticmethod
+    async def get_role_menu_tree(*, pk: int):
+        async with async_db_session() as db:
+            role = await RoleDao.get_with_relation(db, pk)
+            if not role:
+                raise errors.NotFoundError(msg='角色不存在')
+            menu_ids = [menu.id for menu in role.menus]
+            menu_select = await MenuDao.get_role_menus(db, False, menu_ids)
+            menu_tree = await get_tree_data(menu_select)
+            return menu_tree
+
     @staticmethod
     async def get_user_menu_tree(*, request: Request):
         async with async_db_session() as db: