Add casbine-related interfaces (#107)

* Add casbin-related interfaces

* format

Author: wu-clan

backend/app/api/v1/casbin.py

@@ -1,7 +1,93 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-from fastapi import APIRouter
+from typing import Annotated
+
+from fastapi import APIRouter, Query
+
+from backend.app.common.casbin_rbac import DependsRBAC
+from backend.app.common.jwt import DependsJwtAuth
+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.casbin_rule import (
+    CreatePolicy,
+    UpdatePolicy,
+    DeletePolicy,
+    CreateUserRole,
+    DeleteUserRole,
+    GetAllPolicy,
+)
+from backend.app.services.casbin_service import CasbinService
 
 router = APIRouter()
 
-# TODO: 添加 casbin 相关接口
+
+@router.get('', summary='（模糊条件）分页获取所有 casbin 规则', dependencies=[DependsJwtAuth, PageDepends])
+async def get_all_casbin(
+    db: CurrentSession,
+    ptype: Annotated[str | None, Query()] = None,
+    sub: Annotated[str | None, Query()] = None,
+):
+    casbin_select = await CasbinService.get_casbin_list(ptype=ptype, sub=sub)
+    page_data = await paging_data(db, casbin_select, GetAllPolicy)
+    return await response_base.success(data=page_data)
+
+
+@router.get('/policies', summary='获取所有 P 规则', dependencies=[DependsJwtAuth])
+async def get_all_policies():
+    policies = await CasbinService.get_policy_list()
+    return await response_base.success(data=policies)
+
+
+@router.post('/policy', summary='添加基于角色(主)/用户(次)的访问权限', dependencies=[DependsRBAC])
+async def create_policy(p: CreatePolicy):
+    """
+    p 规则:
+
+    - 推荐添加基于角色的访问权限, 需配合添加 g 规则才能真正拥有访问权限，适合配置全局接口访问策略<br>
+    **格式**: 角色 role + 访问路径 path + 访问方法 method
+
+    - 如果添加基于用户的访问权限, 不需配合添加 g 规则就能真正拥有权限，适合配置指定用户接口访问策略<br>
+    **格式**: 用户 uuid + 访问路径 path + 访问方法 method
+    """
+    data = await CasbinService.create_policy(p=p)
+    return await response_base.success(data=data)
+
+
+@router.put('/policy', summary='更新基于角色(主)/用户(次)的访问权限', dependencies=[DependsRBAC])
+async def update_policy(old: UpdatePolicy, new: UpdatePolicy):
+    data = await CasbinService.update_policy(old=old, new=new)
+    return await response_base.success(data=data)
+
+
+@router.delete('/policy', summary='删除基于角色(主)/用户的访问权限', dependencies=[DependsRBAC])
+async def delete_policy(p: DeletePolicy):
+    data = await CasbinService.delete_policy(p=p)
+    return await response_base.success(data=data)
+
+
+@router.get('/groups', summary='获取所有 g 规则', dependencies=[DependsJwtAuth])
+async def get_all_groups():
+    data = await CasbinService.get_group_list()
+    return await response_base.success(data=data)
+
+
+@router.post('/group', summary='添加基于用户组的访问权限', dependencies=[DependsRBAC])
+async def create_group(g: CreateUserRole):
+    """
+    g 规则 (**依赖 p 规则**):
+
+    - 如果在 p 规则中添加了基于角色的访问权限, 则还需要在 g 规则中添加基于用户组的访问权限, 才能真正拥有访问权限<br>
+    **格式**: 用户 uuid + 角色 role
+
+    - 如果在p策略中添加了基于用户的访问权限, 则不添加相应的 g 规则能直接拥有访问权限<br>
+    但是拥有的不是用户角色的所有权限, 而只是单一的对应的 p 规则所添加的访问权限
+    """
+    data = await CasbinService.create_group(g=g)
+    return await response_base.success(data=data)
+
+
+@router.delete('/group', summary='删除基于用户组的访问权限', dependencies=[DependsRBAC])
+async def delete_group(g: DeleteUserRole):
+    data = await CasbinService.delete_group(g=g)
+    return await response_base.success(data=data)

backend/app/api/v1/config.py

@@ -58,5 +58,13 @@ async def get_all_route(request: Request):
     data = []
     for route in request.app.routes:
         if isinstance(route, APIRoute):
-            data.append({'path': route.path, 'name': route.name, 'summary': route.summary, 'methods': route.methods})
+            data.append(
+                {
+                    'path': route.path,
+                    'name': route.name,
+                    'summary': route.summary,
+                    'methods': route.methods,
+                    'dependencies': route.dependencies,
+                }
+            )
     return await response_base.success(data={'route_list': data})

backend/app/common/casbin_rbac.py

@@ -14,7 +14,7 @@
 
 class RBAC:
     @staticmethod
-    async def get_casbin_enforcer() -> casbin.Enforcer:
+    def enforcer() -> casbin.Enforcer:
         """
         获取 casbin 执行器
 
@@ -28,35 +28,34 @@ async def get_casbin_enforcer() -> casbin.Enforcer:
 
     async def rbac_verify(self, request: Request, _: str = DependsJwtAuth) -> None:
         """
-        权限校验，超级用户跳过校验，默认拥有所有权限
+        权限校验
 
         :param request:
         :param _:
         :return:
         """
-        user_uuid = request.user.user_uuid
-        user_roles = request.user.roles
-        role_data_scope = [role.data_scope for role in user_roles]
         super_user = request.user.is_superuser
-        path = request.url.path
-        method = request.method
-
         if super_user:
             return
 
-        for ce in settings.CASBIN_EXCLUDE:
-            if ce['method'] == method and ce['path'] == path:
-                return
+        method = request.method
+        path = request.url.path
+        if (method, path) in settings.CASBIN_EXCLUDE:
+            return
 
-        if 1 in set(role_data_scope):
+        user_roles = request.user.roles
+        data_scope = [role.data_scope for role in user_roles if role.data_scope == 1]
+        if data_scope:
             return
 
         # TODO: 通过 redis 做鉴权查询优化，减少数据库查询
-        enforcer = await self.get_casbin_enforcer()
+        user_uuid = request.user.user_uuid
+        enforcer = self.enforcer()
         if not enforcer.enforce(user_uuid, path, method):
             raise AuthorizationError
 
 
-rbac = RBAC()
+RBAC = RBAC()
+RbacEnforcer = RBAC.enforcer()
 # RBAC 依赖注入
-DependsRBAC = Depends(rbac.rbac_verify)
+DependsRBAC = Depends(RBAC.rbac_verify)

backend/app/core/conf.py

@@ -98,12 +98,12 @@ def validator_api_url(cls, values):
 
     # Casbin
     CASBIN_RBAC_MODEL_NAME: str = 'rbac_model.conf'
-    CASBIN_EXCLUDE: list[dict[str, str]] = [
-        {'method': 'POST', 'path': '/v1/auth/swagger_login'},
-        {'method': 'POST', 'path': '/v1/auth/login'},
-        {'method': 'POST', 'path': '/v1/auth/register'},
-        {'method': 'POST', 'path': '/v1/auth/password/reset'},
-    ]
+    CASBIN_EXCLUDE: set[tuple[str, str]] = {
+        ('POST', '/v1/auth/swagger_login'),
+        ('POST', '/v1/auth/login'),
+        ('POST', '/v1/auth/register'),
+        ('POST', '/v1/auth/password/reset'),
+    }
 
     # Opera log
     OPERA_LOG_EXCLUDE: list[str] = [

backend/app/crud/crud_casbin.py

@@ -1,13 +1,23 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
+from sqlalchemy import Select, select, and_
+
 from backend.app.crud.base import CRUDBase
 from backend.app.models import CasbinRule
 from backend.app.schemas.casbin_rule import CreatePolicy, UpdatePolicy
 
 
 class CRUDCasbin(CRUDBase[CasbinRule, CreatePolicy, UpdatePolicy]):
-    # TODO: 添加 casbin 相关数据库操作
-    pass
+    async def get_all_policy(self, ptype: str, sub: str) -> Select:
+        se = select(self.model).order_by(self.model.id)
+        where_list = []
+        if ptype:
+            where_list.append(self.model.ptype == ptype)
+        if sub:
+            where_list.append(self.model.v0.like(f'%{sub}%'))
+        if where_list:
+            se = se.where(and_(*where_list))
+        return se
 
 
 CasbinDao: CRUDCasbin = CRUDCasbin(CasbinRule)

backend/app/schemas/api.py

@@ -9,7 +9,7 @@
 
 class ApiBase(BaseModel):
     name: str
-    method: str = Field(..., description='请求方法')
+    method: str = Field(default=MethodType.GET, description='请求方法')
     path: str = Field(..., description='api路径')
     remark: str | None = None
 

backend/app/schemas/casbin_rule.py

@@ -5,12 +5,9 @@
 from backend.app.common.enums import MethodType
 
 
-class RBACBase(BaseModel):
+class CreatePolicy(BaseModel):
     sub: str = Field(..., description='用户uuid / 角色')
-
-
-class CreatePolicy(RBACBase):
-    path: str = Field(..., description='api路径')
+    path: str = Field(..., description='api 路径')
     method: str = Field(default=MethodType.GET, description='请求方法')
 
     @validator('method')
@@ -32,15 +29,19 @@ class DeletePolicy(CreatePolicy):
 
 
 class CreateUserRole(BaseModel):
-    uuid: str = Field(..., description='用户uuid')
+    uuid: str = Field(..., description='用户 uuid')
     role: str = Field(..., description='角色')
 
 
+class DeleteUserRole(CreateUserRole):
+    pass
+
+
 class GetAllPolicy(BaseModel):
     id: int
-    ptype: str
-    v0: str
-    v1: str
+    ptype: str = Field(..., description='规则类型, p 或 g')
+    v0: str = Field(..., description='用户 uuid / 角色')
+    v1: str = Field(..., description='api 路径 / 角色')
     v2: str | None = None
     v3: str | None = None
     v4: str | None = None

backend/app/services/casbin_service.py

@@ -1,7 +1,62 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
+from sqlalchemy import Select
+
+from backend.app.common.casbin_rbac import RbacEnforcer
+from backend.app.common.exception import errors
+from backend.app.crud.crud_casbin import CasbinDao
+from backend.app.schemas.casbin_rule import CreatePolicy, UpdatePolicy, DeletePolicy, CreateUserRole, DeleteUserRole
 
 
 class CasbinService:
-    # TODO: 添加 casbin 相关服务
-    pass
+    @staticmethod
+    async def get_casbin_list(*, ptype: str, sub: str) -> Select:
+        return await CasbinDao.get_all_policy(ptype, sub)
+
+    @staticmethod
+    async def get_policy_list():
+        data = await RbacEnforcer.get_policy()
+        return data
+
+    @staticmethod
+    async def create_policy(*, p: CreatePolicy):
+        data = await RbacEnforcer.add_policy(p.sub, p.path, p.method)
+        if not data:
+            raise errors.ForbiddenError(msg='权限已存在')
+        return data
+
+    @staticmethod
+    async def update_policy(*, old: UpdatePolicy, new: UpdatePolicy):
+        _p = await RbacEnforcer.has_named_policy('p', old.sub, old.path, old.method)
+        if not _p:
+            raise errors.NotFoundError(msg='权限不存在')
+        data = await RbacEnforcer.update_policy([old.sub, old.path, old.method], [new.sub, new.path, new.method])
+        return data
+
+    @staticmethod
+    async def delete_policy(*, p: DeletePolicy):
+        _p = await RbacEnforcer.has_named_policy('p', p.sub, p.path, p.method)
+        if not _p:
+            raise errors.NotFoundError(msg='权限不存在')
+        data = await RbacEnforcer.remove_policy(p.sub, p.path, p.method)
+        return data
+
+    @staticmethod
+    async def get_group_list():
+        data = await RbacEnforcer.get_grouping_policy()
+        return data
+
+    @staticmethod
+    async def create_group(*, g: CreateUserRole):
+        data = await RbacEnforcer.add_grouping_policy(g.uuid, g.role)
+        if not data:
+            raise errors.ForbiddenError(msg='权限已存在')
+        return data
+
+    @staticmethod
+    async def delete_group(*, g: DeleteUserRole):
+        _g = await RbacEnforcer.has_named_grouping_policy('g', g.uuid, g.role)
+        if not _g:
+            raise errors.NotFoundError(msg='权限不存在')
+        data = await RbacEnforcer.remove_grouping_policy(g.uuid, g.role)
+        return data