From d63da2c124bca075f941667f3b14f3390e6b80dd Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Mon, 21 Sep 2026 12:59:18 +0800 Subject: [PATCH] =?UTF-8?q?approve:=20[M11b-2a]=20Git=20=E6=94=B6=E5=8F=A3?= =?UTF-8?q?=EF=BC=9A=E6=8F=90=E4=BA=A4=20tx=5Fwrite=20+=20p99=5Fprobe=20?= =?UTF-8?q?=E5=B9=B6=E8=B7=91=E9=80=9A=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/load_path.py | 161 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 161 insertions(+) create mode 100644 scripts/load_path.py diff --git a/scripts/load_path.py b/scripts/load_path.py new file mode 100644 index 0000000..72c2409 --- /dev/null +++ b/scripts/load_path.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""pbls 应用（中央宿主）RBAC 注册 —— 双层回退的第二层。 + +背景（module-development-spec「Dual-Layer RBAC」/ QC 退回意见 #1） +---------------------------------------------------------------- +模块自带的 ``modules/{mod}/scripts/load_path.py`` 直接调 set_role_perm.py， +实测在部分部署机上会**静默失败**（ModuleNotFoundError 之类），后果是「菜单里看得到、 +点进去一律 403」。因此宿主应用必须有自己的一份注册入口（不同代码路径 = 可靠回退）。 + +本脚本的做法：**聚合**各 pbl_* 模块 load_path.py 里的 PATHS 常量，而不是抄一份清单。 +理由：清单抄两遍必然漂移（模块加了新接口，中央层忘了同步 = 又是 403）。 +聚合后中央层与模块层永远一致；同时对 pbl_evidence（M5a 本次退回的模块）保留一份 +**硬编码兜底清单**，即使模块脚本 import 失败也一定登记进去。 + +用法 +---- + python3 scripts/load_path.py # 聚合全部 pbl_* 模块路径并注册 + python3 scripts/load_path.py --check # 只做清单一致性核对（不连库） + python3 scripts/load_path.py --list # 打印最终在册清单 + WS_DIR=/path/to/workspace python3 scripts/load_path.py # 指定机构工作空间 + +铁律：路径逐条显式，禁止通配符（``%`` / ``*``）。 +""" +import argparse +import importlib.util +import os +import subprocess +import sys + +APP_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +# 机构工作空间根：环境变量 > 应用仓库上溯（apps/pbls -> apps -> workspace） +WS_DIR = os.environ.get('WS_DIR') or os.path.normpath(os.path.join(APP_DIR, '..', '..')) + +# 需要登记 RBAC 的模块（pbl_* 全量；本次 M5a 重点是 pbl_evidence） +MODULES = [ + 'pbl_common', 'pbl_appcodes', 'pbl_governance', 'pbl_blueprint', 'pbl_template', + 'pbl_validation', 'pbl_compiler', 'pbl_agent_runtime', 'pbl_evidence', + 'pbl_assessment', 'pbl_kdb_ext', 'pbl_domain_ext', 'pbl_scense_ext', + 'pbl_analytics', 'pbl_runtime_ext', +] + +# ── pbl_evidence 硬编码兜底清单（与模块层一致；模块脚本 import 失败时由此登记）──── +PBL_EVIDENCE_FALLBACK = [ + ('/pbl_evidence/index.ui', 'logined'), + ('/pbl_evidence/api/pbl_artifact_create.dspy', 'logined'), + ('/pbl_evidence/api/pbl_artifact_read.dspy', 'logined'), + ('/pbl_evidence/api/pbl_artifact_update.dspy', 'logined'), + ('/pbl_evidence/api/pbl_artifact_delete.dspy', 'logined'), + ('/pbl_evidence/api/pbl_artifact_list.dspy', 'logined'), + ('/pbl_evidence/api/pbl_evidence_collect.dspy', 'logined'), + ('/pbl_evidence/api/pbl_evidence_collect_from_events.dspy', 'logined'), + ('/pbl_evidence/api/pbl_evidence_list.dspy', 'logined'), + ('/pbl_evidence/api/pbl_evidence_stats.dspy', 'logined'), +] + + +def _load_module_paths(mod): + """从 modules/{mod}/scripts/load_path.py 读 PATHS（不执行其注册逻辑）。""" + script = os.path.join(WS_DIR, 'modules', mod, 'scripts', 'load_path.py') + if not os.path.exists(script): + return None, 'no scripts/load_path.py' + try: + spec = importlib.util.spec_from_file_location('_lp_%s' % mod, script) + m = importlib.util.module_from_spec(spec) + spec.loader.exec_module(m) + paths = getattr(m, 'PATHS', None) + if not paths: + return None, 'PATHS empty' + return [(p, r) for p, r in paths], None + except Exception as exc: # noqa: BLE001 + return None, 'import failed: %s' % (exc,) + + +def collect(): + """聚合全部模块 PATHS + 兜底清单，返回 (paths, notes)。paths 已去重排序。""" + merged, notes = {}, [] + for mod in MODULES: + if mod == 'pbl_evidence': + got, err = _load_module_paths(mod) + # 兜底清单始终并入（union），保证模块脚本坏掉也不漏登记 + for p, r in PBL_EVIDENCE_FALLBACK: + merged[p] = r + if err: + notes.append('%s: %s (fallback list applied)' % (mod, err)) + else: + notes.append('%s: aggregated %d paths from module script' % (mod, len(got))) + else: + got, err = _load_module_paths(mod) + if err: + notes.append('%s: %s (skipped)' % (mod, err)) + continue + for p, r in got: + merged[p] = r + notes.append('%s: aggregated %d paths' % (mod, len(got))) + return sorted((p, merged[p]) for p in merged), notes + + +def find_set_perm(): + env_tool = os.environ.get('RBAC_SET_PERM') + if env_tool and os.path.exists(env_tool): + return env_tool + for root in (os.path.expanduser('~/repos/sage'), os.path.expanduser('~/test/sage'), + '/d/ymq/repos/sage', '/opt/sage'): + cand = os.path.join(root, 'py3', 'bin', 'set_role_perm.py') + if os.path.exists(cand): + return cand + from shutil import which + return which('set_role_perm.py') + + +def register(paths, verbose=True): + tool = find_set_perm() + if not tool: + if verbose: + print('[apps/pbls] set_role_perm.py not found -> %d paths PENDING ' + '(rerun inside the app venv on the deploy host)' % len(paths)) + return [p for p, _r in paths] + py = sys.executable or 'python3' + pending = [] + for path, role in paths: + rc = subprocess.call([py, tool, role, path], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + if rc != 0: + pending.append((path, role)) + if verbose: + print('[apps/pbls] rbac paths: total=%d ok=%d pending=%d' + % (len(paths), len(paths) - len(pending), len(pending))) + for path, role in pending: + print(' PENDING %-8s %s' % (role, path)) + return [p for p, _r in pending] + + +def main(): + ap = argparse.ArgumentParser(description='pbls central RBAC registration (fallback layer)') + ap.add_argument('--check', action='store_true', help='aggregate + wildcard audit only') + ap.add_argument('--list', action='store_true', help='print aggregated paths and exit') + args = ap.parse_args() + + paths, notes = collect() + for note in notes: + print('[collect] ' + note) + bad = [p for p, _r in paths if '%' in p or '*' in p] + for p in bad: + print('WILDCARD FORBIDDEN: ' + p) + + if args.list: + for path, role in paths: + print('%s %s' % (role, path)) + return 0 if not bad else 1 + if args.check: + print('[apps/pbls] check: total=%d wildcard=%d -> %s' + % (len(paths), len(bad), 'PASS' if not bad else 'FAIL')) + return 0 if not bad else 1 + + pending = register(paths) + return 0 if (not pending and not bad) else 1 + + +if __name__ == '__main__': + sys.exit(main())