141 lines
6.7 KiB
Python
141 lines
6.7 KiB
Python
#!/usr/bin/env python3
|
||
"""Register all pipeline module RBAC permissions in one run."""
|
||
import sys, os, asyncio
|
||
|
||
# 应用根目录 = scripts/ 的上一级（不写死 ~/pipeline，各环境 HOME 不同）
|
||
ROOT = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
|
||
sys.path.insert(0, os.path.join(ROOT, "py3", "lib", "python3.10", "site-packages"))
|
||
sys.path.insert(0, ROOT)
|
||
|
||
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
|
||
|
||
PERMS = [
|
||
# pipeline-sdlc
|
||
("/pipeline-sdlc", "pipeline_sdlc", "logined"),
|
||
("/pipeline-sdlc/sd_dashboard/*", "pipeline_sdlc", "logined"),
|
||
("/pipeline-sdlc/sd_projects/*", "pipeline_sdlc", "logined"),
|
||
("/pipeline-sdlc/sd_project_list/*", "pipeline_sdlc", "logined"),
|
||
("/pipeline-sdlc/sd_iterations/*", "pipeline_sdlc", "logined"),
|
||
("/pipeline-sdlc/sd_iteration_list/*", "pipeline_sdlc", "logined"),
|
||
("/pipeline-sdlc/sd_test_plans/*", "pipeline_sdlc", "logined"),
|
||
("/pipeline-sdlc/sd_test_plan_list/*", "pipeline_sdlc", "logined"),
|
||
("/pipeline-sdlc/sd_test_cases/*", "pipeline_sdlc", "logined"),
|
||
("/pipeline-sdlc/sd_test_case_list/*", "pipeline_sdlc", "logined"),
|
||
("/pipeline-sdlc/sd_bugs/*", "pipeline_sdlc", "logined"),
|
||
("/pipeline-sdlc/sd_bug_list/*", "pipeline_sdlc", "logined"),
|
||
("/pipeline-sdlc/sd_deploy_envs/*", "pipeline_sdlc", "logined"),
|
||
("/pipeline-sdlc/sd_deploy_env_list/*", "pipeline_sdlc", "logined"),
|
||
("/pipeline-sdlc/sd_review/*", "pipeline_sdlc", "logined"),
|
||
("/pipeline-sdlc/api/*", "pipeline_sdlc", "logined"),
|
||
# pipeline_core
|
||
("/pipeline_core", "pipeline_core", "logined"),
|
||
("/pipeline_core/pipelines/*", "pipeline_core", "logined"),
|
||
("/pipeline_core/pipeline_steps/*", "pipeline_core", "logined"),
|
||
("/pipeline_core/pipeline_versions/*", "pipeline_core", "logined"),
|
||
("/pipeline_core/pipeline_editor/*", "pipeline_core", "logined"),
|
||
("/pipeline_core/api/*", "pipeline_core", "logined"),
|
||
# pipeline_ops
|
||
("/pipeline_ops", "pipeline_ops", "logined"),
|
||
("/pipeline_ops/api/*", "pipeline_ops", "logined"),
|
||
("/pipeline_ops/pipeline_pricing/*", "pipeline_ops", "logined"),
|
||
("/pipeline_ops/pipeline_capacity/*", "pipeline_ops", "logined"),
|
||
("/pipeline_ops/pipeline_usage_log/*", "pipeline_ops", "logined"),
|
||
# pipeline_dist
|
||
("/pipeline_dist", "pipeline_dist", "logined"),
|
||
("/pipeline_dist/*", "pipeline_dist", "logined"),
|
||
# showcase
|
||
("/showcase", "showcase", "logined"),
|
||
("/showcase/*", "showcase", "logined"),
|
||
# app root (any)：仅登录页资产/全局静态资源。
|
||
# 注意：禁止 /** any——会让匿名访问全部路径（含五条产线）。
|
||
# 产线 shell（sdlc/core/ops/dist index.ui）须 logined，产线对匿名不可见。
|
||
("/index.ui", "app", "any"),
|
||
("/todo_badge.js", "app", "any"),
|
||
("/index_tab.js", "app", "any"),
|
||
("/theme_toggle.js", "app", "any"),
|
||
("/pipeline-sdlc/index.ui", "pipeline_sdlc", "logined"),
|
||
("/pipeline_core/index.ui", "pipeline_core", "logined"),
|
||
("/pipeline_ops/index.ui", "pipeline_ops", "logined"),
|
||
("/pipeline_dist/index.ui", "pipeline_dist", "logined"),
|
||
# dapi 客户自助 API Key 套件（apikey_manage.ui 数据按 orgid+userid 自过滤；
|
||
# downapps/get_apikey/apply_apikey 已挂 customer.*，此处补齐页面+操作端点，
|
||
# 口径 logined 与既有 create_apikey.dspy/copy_apikey.dspy 一致）
|
||
("/dapi", "dapi", "logined"),
|
||
("/dapi/apikey_manage.ui", "dapi", "logined"),
|
||
("/dapi/downapps.dspy", "dapi", "logined"),
|
||
("/dapi/get_apikey.dspy", "dapi", "logined"),
|
||
("/dapi/apply_apikey.dspy", "dapi", "logined"),
|
||
("/dapi/create_apikey.dspy", "dapi", "logined"),
|
||
("/dapi/create_apikey.ui", "dapi", "logined"),
|
||
("/dapi/copy_apikey.dspy", "dapi", "logined"),
|
||
("/dapi/update_apikey.dspy", "dapi", "logined"),
|
||
("/dapi/update_apikey.ui", "dapi", "logined"),
|
||
("/dapi/do_update_apikey.dspy", "dapi", "logined"),
|
||
("/dapi/delete_apikey.dspy", "dapi", "logined"),
|
||
]
|
||
|
||
|
||
async def register():
|
||
config = getConfig(ROOT, NS={"workdir": ROOT, "ProgramPath": ProgramPath()})
|
||
DBPools(config.databases)
|
||
initEnv()
|
||
env = ServerEnv()
|
||
env.get_module_dbname = lambda m: "pipeline"
|
||
|
||
# 特殊角色（any/anonymous/logined）roleid 用字面值：role 表里全局 logined 行
|
||
# id='logined'（name=NULL），而按 name='logined' 查会错配到 downapp 机构的同名
|
||
# 角色行（缓存键为 downapp.logined，普通登录用户不命中→403）。
|
||
# RBAC 缓存键规则（rbac/userperm.py load_roleperms）：id in (any/anonymous/logined)
|
||
# 才映射全局键；其余按 orgtypeid.name。
|
||
SPECIAL_ROLES = {"any", "anonymous", "logined"}
|
||
|
||
async with DBPools().sqlorContext("pipeline") as sor:
|
||
total = 0
|
||
for path, module, role_name in PERMS:
|
||
sql = "SELECT id FROM permission WHERE path=" + repr(path) + " LIMIT 1"
|
||
existing = await sor.sqlExe(sql, {})
|
||
pid = None
|
||
if existing:
|
||
pid = existing[0].id
|
||
else:
|
||
pid = getID()
|
||
await sor.C("permission", {
|
||
"id": pid, "name": module + ":" + path,
|
||
"path": path, "permtype": module
|
||
})
|
||
total += 1
|
||
# 补 role 关联（permission 已存在但 rolepermission 缺失时也要补，修复 /** 等历史漏关联）
|
||
if role_name in SPECIAL_ROLES:
|
||
role_id = role_name
|
||
else:
|
||
role_sql = "SELECT id FROM role WHERE name=" + repr(role_name) + " LIMIT 1"
|
||
role_recs = await sor.sqlExe(role_sql, {})
|
||
role_id = role_recs[0].id if role_recs else None
|
||
if role_id:
|
||
rp_sql = ("SELECT id FROM rolepermission WHERE roleid=" + repr(role_id)
|
||
+ " AND permid=" + repr(pid) + " LIMIT 1")
|
||
rp_existing = await sor.sqlExe(rp_sql, {})
|
||
if not rp_existing:
|
||
await sor.C("rolepermission", {
|
||
"id": getID(),
|
||
"roleid": role_id,
|
||
"permid": pid
|
||
})
|
||
|
||
print(f"Registered {total} new permissions")
|
||
# Verify
|
||
async with DBPools().sqlorContext("pipeline") as sor:
|
||
cnt = await sor.sqlExe(
|
||
"SELECT COUNT(*) as n FROM permission WHERE path LIKE ${p1}$ OR path LIKE ${p2}$",
|
||
{"p1": "/pipeline%", "p2": "/showcase%"})
|
||
print(f"Total pipeline permissions: {cnt[0].n}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(register())
|