#!/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"), # showcase ("/showcase", "showcase", "logined"), ("/showcase/*", "showcase", "logined"), # app root (any)：仅登录页资产/全局静态资源。 # 注意：禁止 /** any——会让匿名访问全部路径（含五条产线）。 # 产线 shell（sdlc/core/ops index.ui）须 logined，产线对匿名不可见。 ("/index.ui", "app", "any"), ("/todo_badge.js", "app", "any"), ("/theme_toggle.js", "app", "any"), # 手机端 H5 外壳（2026-09-07）：「我的」整页（底部 TabBar 第4个 Tab） ("/mine.ui", "app", "logined"), # 产线工作界面外壳（手机端「产线」Tab：ChipBar 已购产线 + AgentIO 全屏） ("/pipeline_core/mobile_agent.ui", "pipeline_core", "logined"), # 已购产线列表（ChipBar 数据源，订阅过滤在服务端） ("/pipeline_core/api/my_pipelines.dspy", "pipeline_core", "logined"), ("/pipeline-sdlc/index.ui", "pipeline_sdlc", "logined"), ("/pipeline_core/index.ui", "pipeline_core", "logined"), ("/pipeline_ops/index.ui", "pipeline_ops", "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())