yumoqing
b2bb281559
解决独立应用(如CMS)的RBAC数据库不匹配问题: - set_role_perm.py 原来硬编码 'sage' 数据库 - 独立应用 get_module_dbname 返回自己的数据库(如 ocai_cms) - 导致写入 sage 但读取 ocai_cms，权限检查失败 - 现在通过 SAGE_RBAC_DB 环境变量指定，默认仍为 sage
75 lines
1.7 KiB
Python
75 lines
1.7 KiB
Python
import os
|
|
import sys
|
|
import asyncio
|
|
from sqlor.dbpools import DBPools
|
|
from appPublic.jsonConfig import getConfig
|
|
from appPublic.dictObject import DictObject
|
|
from appPublic.uniqueID import getID
|
|
|
|
async def delete_anonymous_perm(sor, permid):
|
|
await sor.D('rolepermission', {
|
|
'roleid': 'anonymous',
|
|
'permid': permid
|
|
})
|
|
|
|
async def add_roleperm(sor, roleid, permid):
|
|
ns = {
|
|
'roleid': roleid,
|
|
'permid': permid
|
|
}
|
|
recs = await sor.R('rolepermission', ns.copy())
|
|
if recs:
|
|
print(f'{roleid}, {permid} 已经存在')
|
|
return
|
|
ns['id'] = getID()
|
|
await sor.C('rolepermission', ns)
|
|
print(f'{roleid}, {permid} perm add')
|
|
return
|
|
|
|
async def main():
|
|
config = getConfig('.')
|
|
db = DBPools(config.databases)
|
|
if len(sys.argv) < 3:
|
|
print(f'{sys.argv[0]} role path')
|
|
sys.exit(1)
|
|
|
|
role = sys.argv[1]
|
|
path = sys.argv[2]
|
|
dbname = os.environ.get('SAGE_RBAC_DB', 'sage')
|
|
async with db.sqlorContext(dbname) as sor:
|
|
perms = None
|
|
if '%' in path:
|
|
perms = await sor.sqlExe("select * from permission where path like ${path}$", {'path': path})
|
|
else:
|
|
perms = await sor.R('permission', {'path': path})
|
|
if len(perms) < 1:
|
|
perms = [ DictObject(**{
|
|
'id': getID(),
|
|
'path': path
|
|
}) ]
|
|
await sor.C('permission', perms[0].copy())
|
|
|
|
if role in ['anonymous', 'any', 'logined']:
|
|
for p in perms:
|
|
await add_roleperm(sor, role, p.id)
|
|
return
|
|
orgtypeid, name = role.split('.')
|
|
ns = {
|
|
'orgtypeid': orgtypeid,
|
|
'name': name
|
|
}
|
|
roles = await sor.R('role', ns.copy())
|
|
if not roles:
|
|
ns['id'] = getID()
|
|
await sor.C('role', ns.copy())
|
|
else:
|
|
ns['id'] = roles[0].id
|
|
for p in perms:
|
|
await add_roleperm(sor, ns['id'], p.id)
|
|
await delete_anonymous_perm(sor, p.id)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.get_event_loop().run_until_complete(main())
|
|
|