120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""RBAC 权限注册 — tenant 模块
|
|
给 tenant 路径补充 owner.* 和 *.admin 等管理角色的 rolepermission 关联
|
|
"""
|
|
import os, sys, asyncio
|
|
|
|
MOD = "tenant"
|
|
|
|
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
for candidate in [
|
|
os.path.join(SCRIPT_DIR, "..", ".."),
|
|
os.path.expanduser("~/repos/sage"),
|
|
os.path.expanduser("~/sage"),
|
|
"/d/apitest/sage",
|
|
]:
|
|
if os.path.isdir(os.path.join(candidate, "wwwroot")):
|
|
SAGE_ROOT = candidate
|
|
break
|
|
else:
|
|
print("ERROR: Sage root not found")
|
|
sys.exit(1)
|
|
|
|
sys.path.insert(0, SAGE_ROOT)
|
|
from sqlor.dbpools import DBPools
|
|
from appPublic.jsonConfig import getConfig
|
|
from appPublic.dictObject import DictObject
|
|
from appPublic.uniqueID import getID
|
|
|
|
# tenant 所有需要权限的路径
|
|
TENANT_PATHS = [
|
|
f"/{MOD}",
|
|
f"/{MOD}/index.ui",
|
|
f"/{MOD}/tenant_domain",
|
|
f"/{MOD}/tenant_domain/index.ui",
|
|
f"/{MOD}/tenant_domain/get_tenant_domain.dspy",
|
|
f"/{MOD}/tenant_domain/add_tenant_domain.dspy",
|
|
f"/{MOD}/tenant_domain/update_tenant_domain.dspy",
|
|
f"/{MOD}/tenant_domain/delete_tenant_domain.dspy",
|
|
f"/{MOD}/api/tenant_domain_list.dspy",
|
|
f"/{MOD}/api/tenant_domain_get.dspy",
|
|
f"/{MOD}/api/tenant_domain_create.dspy",
|
|
f"/{MOD}/api/tenant_domain_update.dspy",
|
|
f"/{MOD}/api/tenant_domain_delete.dspy",
|
|
f"/{MOD}/api/get_org_list.dspy",
|
|
f"/{MOD}/api/tenant_domain.json",
|
|
]
|
|
|
|
# 需要授权的角色: (orgtypeid, name) → 用于匹配 role 表
|
|
ROLES_NEEDED = [
|
|
("owner", "superuser"),
|
|
("owner", "developer"),
|
|
("owner", "admin"),
|
|
("reseller", "superuser"),
|
|
("reseller", "developer"),
|
|
("reseller", "admin"),
|
|
]
|
|
|
|
|
|
async def main():
|
|
config = getConfig('.')
|
|
DBPools(config.databases)
|
|
cnt = 0
|
|
async with DBPools().sqlorContext('sage') as sor:
|
|
# 1. 查出所有角色 ID
|
|
role_ids = {}
|
|
for orgtypeid, name in ROLES_NEEDED:
|
|
recs = await sor.sqlExe(
|
|
'SELECT id FROM role WHERE orgtypeid=${oid}$ AND name=${name}$',
|
|
{'oid': orgtypeid, 'name': name}
|
|
)
|
|
if recs:
|
|
role_ids[f'{orgtypeid}.{name}'] = recs[0].id
|
|
|
|
print(f"Found {len(role_ids)} roles: {list(role_ids.keys())}")
|
|
|
|
# 2. 查出 tenant 路径的 permission ID
|
|
perm_ids = {}
|
|
for path in TENANT_PATHS:
|
|
recs = await sor.sqlExe(
|
|
'SELECT id FROM permission WHERE path=${path}$',
|
|
{'path': path}
|
|
)
|
|
if recs:
|
|
perm_ids[path] = recs[0].id
|
|
else:
|
|
# 创建缺失的 permission
|
|
pid = getID()
|
|
await sor.sqlExe(
|
|
'INSERT INTO permission (id, path, name) VALUES (${id}$, ${path}$, ${name}$)',
|
|
{'id': pid, 'path': path, 'name': path}
|
|
)
|
|
perm_ids[path] = pid
|
|
print(f" Created permission: {path}")
|
|
|
|
print(f"Found {len(perm_ids)} permission paths")
|
|
|
|
# 3. 补充缺失的 rolepermission 关联
|
|
for role_key, role_id in role_ids.items():
|
|
for path, perm_id in perm_ids.items():
|
|
existing = await sor.sqlExe(
|
|
'SELECT id FROM rolepermission WHERE roleid=${rid}$ AND permid=${pid}$',
|
|
{'rid': role_id, 'pid': perm_id}
|
|
)
|
|
if not existing:
|
|
await sor.sqlExe(
|
|
'INSERT INTO rolepermission (id, roleid, permid) VALUES (${id}$, ${rid}$, ${pid}$)',
|
|
{'id': getID(), 'rid': role_id, 'pid': perm_id}
|
|
)
|
|
cnt += 1
|
|
|
|
if cnt > 0:
|
|
print(f"tenant: {cnt} new rolepermission links inserted")
|
|
print(" Clear RBAC cache needed: restart Sage or wait for TTL")
|
|
else:
|
|
print("tenant: all rolepermission links already exist")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(main())
|