diff --git a/rbac/check_perm.py b/rbac/check_perm.py index 54d887c..7690157 100644 --- a/rbac/check_perm.py +++ b/rbac/check_perm.py @@ -3,44 +3,35 @@ from appPublic.registerfunction import RegisterFunction from appPublic.log import debug, exception from ahserver.auth_api import AuthAPI -class CheckPerms: - def __init__(self): - self.users = [] - self.userrole = [] - self.roleperms = [] - self.permissions = {} - self._need_refresh = True - self.rfexe = RegisterFunction().exe - - async def load(self): - db = DBPools() - dbname = await self.rfexe('get_module_dbname', 'rbac') - async with db.sqlorContext(dbname) as sor: - self.users = await sor.R('users',{}) - self.userrole = await sor.R('userrole',{}) - self.roleperms = await sor.R('rolepermission', {}) - perms = await sor.R('permission', {}) - self.permissions = {p.path:p.id for p in perms} - - async def checkperm(self, userid, path): - debug(f'{userid=}, {path=}, checkperm() called ..') - if self._need_refresh: - await self.load() - if not self.permissions.get(path): - debug(f'{path=} public access, checkperm() return true ..') - return True - roleids = [ rp.roleid for rp in self.roleperms ] - userroles = [ ur.roleid for ur in self.userrole if ur.userid == userid ] - for ur in userroles: - if ur in roleids: - debug(f'{user=} can access {path=} , checkperm() return true ..') - return True - debug(f'{userid=}, can not access {path=}, checkperm() return false ..') - return False - def objcheckperm(obj, userid, path): - cp = CheckPerms() - return cp.checkperm(userid, path) + sql = """select distinct a.*, c.userid from +(select id, path from permission where path=s${path}$ and del_flg='0') a +right join + rolepermission b on a.id = b.permid +right join userrole c on b.roleid = c.roleid +where c.userid = ${userid}$ +and b.del_flg='0' +and c.del_flg='0'""" + + rf = RegisterFunction() + dbname = await rf.exe('get_module_dbname', 'rbac') + db = DBPools() + async with db.sqlorContext(dbname) as sor: + perms = await sor.R('permission', {'path':path}) + if len(perms) == 0: + debug(f'{path=} not found in permission, can access') + return True + + recs = await sor.sqlExe(sql, {'path':path, 'userid':userid}) + for r in recs: + id = r['id'] + if id is not None: + debug(f'{userid=} can access {path=}') + return True + debug(f'{userid=} has not permission to call {path=}') + return False + debug(f'error happened {userid}, {path}') + return False AuthAPI.checkUserPermission = objcheckperm