casbin rbac sample enforcer

backend/api/UsersView.py

@@ -1,21 +1,27 @@
 from starlette.responses import JSONResponse, Response
 from common.paths import api_base_url
 from backend.managers.UsersManager import UsersManager
+from backend.managers.CasbinRoleManager import CasbinRoleManager
 from backend.pagination import parse_pagination_params
 from aiosqlite import IntegrityError
 from backend.schemas import UserSchema
 
 class UsersView:
     def __init__(self):
         self.um = UsersManager()
+        self.cb = CasbinRoleManager()
 
     async def get(self, id: str):
         user = await self.um.retrieve_user(id)
         if user is None:
             return JSONResponse(status_code=404, headers={"error": "User not found"})
         return JSONResponse(user.model_dump(), status_code=200)
 
-    async def post(self, body: dict):
+    async def post(self,token_info, body: dict):
+        enforcer = self.cb.get_enforcer()
+        if not enforcer.enforce(token_info["role"], 'user', 'POST'):
+            return JSONResponse({"message": "Permission denied"}, status_code=403)
+
         try:
             id = await self.um.create_user(body['name'], body['email'])
             return JSONResponse({"id": id}, status_code=201, headers={'Location': f'{api_base_url}/users/{id}'})

backend/rbac_model.conf

@@ -2,14 +2,13 @@
 r = role, res_id, act
 
 [policy_definition]
-p = role, res_id, act, eft
+p = role, res_id, act
 
 [policy_effect]
 e = some(where (p.eft == allow))
 
 [matchers]
-m = (r.res_id == p.res_id && r.act == p.act && r.role == p.role) || 
-    (r.res_id != p.res_id && p.res_id == "DEFAULT" && r.act == p.act && r.role == p.role)
+m = (r.res_id == p.res_id && r.act == p.act && r.role == p.role) || (p.res_id == "DEFAULT" && r.act == p.act && r.role == p.role)
 
 [role_definition]
 g = _, _