yumoqing
35258f3b9c
Passing 'admin' as literal roleid was wrong - rolepermission.roleid must reference role.id. Special roles (any/anonymous/logined) keep literal id (hardcoded in rbac userperm.py); other names resolve via role table with default orgtypeid='*', support 'orgtypeid.name' form.
76 lines
2.9 KiB
Python
76 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Register a single RBAC permission for pipeline modules.
|
||
Called by load_path.py or directly.
|
||
|
||
Usage:
|
||
python set_role_perm.py <role> <path>
|
||
python set_role_perm.py logined /pipeline-sdlc/workspace_edit.xterm
|
||
"""
|
||
import sys, os, asyncio
|
||
|
||
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
ROOT_DIR = os.path.dirname(SCRIPT_DIR) if os.path.basename(SCRIPT_DIR) == 'bin' else SCRIPT_DIR
|
||
sys.path.insert(0, os.path.join(ROOT_DIR, 'py3', 'lib', 'python3.10', 'site-packages'))
|
||
sys.path.insert(0, ROOT_DIR)
|
||
|
||
from sqlor.dbpools import DBPools
|
||
from appPublic.jsonConfig import getConfig
|
||
from appPublic.folderUtils import ProgramPath
|
||
from appPublic.uniqueID import getID
|
||
from ahserver.serverenv import ServerEnv
|
||
from ahserver.globalEnv import initEnv
|
||
|
||
|
||
async def main(role, path):
|
||
config = getConfig(ROOT_DIR,
|
||
NS={'workdir': ROOT_DIR, 'ProgramPath': ProgramPath()})
|
||
DBPools(config.databases)
|
||
initEnv()
|
||
env = ServerEnv()
|
||
env.get_module_dbname = lambda m: 'pipeline' if 'pipeline' in m else 'sage'
|
||
|
||
# pipeline RBAC schema: permission (id, path) + rolepermission (roleid, permid)
|
||
async with DBPools().sqlorContext('pipeline') as sor:
|
||
# role 解析：特殊角色(any/anonymous/logined)用字面 id（rbac userperm 硬编码）；
|
||
# 其余按 [orgtypeid.]name 查 role 表取真实 id（缺省 orgtypeid='*'）。
|
||
# 直接把 'admin' 当 roleid 写库是错的——roleid 必须是 role.id（2026-09-09）。
|
||
SPECIAL_ROLES = ('any', 'anonymous', 'logined')
|
||
if role in SPECIAL_ROLES:
|
||
role_id = role
|
||
else:
|
||
if '.' in role:
|
||
orgtypeid, role_name = role.split('.', 1)
|
||
else:
|
||
orgtypeid, role_name = '*', role
|
||
role_recs = await sor.R('role', {'orgtypeid': orgtypeid, 'name': role_name})
|
||
if not role_recs:
|
||
print(f'ERROR: role not found: {orgtypeid}.{role_name}')
|
||
sys.exit(1)
|
||
role_id = role_recs[0].id
|
||
|
||
recs = await sor.R('permission', {'path': path})
|
||
if not recs:
|
||
permid = getID()
|
||
await sor.C('permission', {'id': permid, 'path': path})
|
||
else:
|
||
permid = recs[0].id
|
||
|
||
rp = await sor.R('rolepermission', {'roleid': role_id, 'permid': permid})
|
||
if rp:
|
||
print(f'Permission exists: {role} {path}')
|
||
return
|
||
|
||
await sor.C('rolepermission', {'id': getID(), 'roleid': role_id, 'permid': permid})
|
||
print(f'Registered: {role}({role_id}) -> {path}')
|
||
|
||
|
||
if __name__ == '__main__':
|
||
if len(sys.argv) != 3:
|
||
print(f"Usage: {sys.argv[0]} <role> <path>")
|
||
print(f" roles: any, anonymous, logined, admin, owner.superuser ...")
|
||
print(f" path example: /pipeline-sdlc/workspace_edit.xterm")
|
||
sys.exit(1)
|
||
|
||
asyncio.run(main(sys.argv[1], sys.argv[2]))
|