yumoqing/rbac
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

bugfix

Browse Source
This commit is contained in:
yumoqing 2024-11-10 09:22:56 +08:00
parent ce24dc4b90
commit 4730cecea1
1 changed files with 9 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
rbac/check_perm.py
View File

@ -1,5 +1,6 @@
from sqlor.dbpools import DBPools from sqlor.dbpools import DBPools
from appPublic.registerfunction import RegisterFunction from appPublic.registerfunction import RegisterFunction
from appPublic.log import debug, exception
from ahserver.auth_api import AuthAPI from ahserver.auth_api import AuthAPI
class CheckPerms: class CheckPerms:
@ -7,34 +8,38 @@ class CheckPerms:
self.users = [] self.users = []
self.userrole = [] self.userrole = []
self.roleperms = [] self.roleperms = []
self.permissions = [] self.permissions = {}
self._need_refresh = True self._need_refresh = True
self.rfexe = RegisterFunction().exe self.rfexe = RegisterFunction().exe
async def load(self): async def load(self):
db = DBPools() db = DBPools()
dbname = self.rfexe('get_module_dbname', 'rbac') dbname = await self.rfexe('get_module_dbname', 'rbac')
async with db.sqlorContext(dbname) as sor: async with db.sqlorContext(dbname) as sor:
self.users = await sor.R('users',{}) self.users = await sor.R('users',{})
self.userrole = await sor.R('userrole',{}) self.userrole = await sor.R('userrole',{})
self.roleperms = await sor.R('roleperms', {}) self.roleperms = await sor.R('rolepermission', {})
perms = await sor.R('permission', {}) perms = await sor.R('permission', {})
self.permissions = {p.path:p.id for p in perms} self.permissions = {p.path:p.id for p in perms}
async def checkperm(self, userid, path): async def checkperm(self, userid, path):
debug(f'{userid=}, {path=}, checkperm() called ..')
if self._need_refresh: if self._need_refresh:
await self.load() await self.load()
if not self.permissions.get(path): if not self.permissions.get(path):
debug(f'{path=} public access, checkperm() return true ..')
return True return True
roleids = [ rp.roleid for rp in self.roleperms ] roleids = [ rp.roleid for rp in self.roleperms ]
userroles = [ ur.roleid for ur in self.userrole if ur.userid == userid ] userroles = [ ur.roleid for ur in self.userrole if ur.userid == userid ]
for ur in userroles: for ur in userroles:
if ur in roleids: if ur in roleids:
debug(f'{user=} can access {path=} , checkperm() return true ..')
return True return True
debug(f'{userid=}, can not access {path=}, checkperm() return false ..')
return False return False
def objcheckperm(obj, userid, path): def objcheckperm(obj, userid, path):
cp = CheckPerm() cp = CheckPerms()
return cp.checkperm(userid, path) return cp.checkperm(userid, path)
AuthAPI.checkUserPermission = objcheckperm AuthAPI.checkUserPermission = objcheckperm