#!/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_* / world_sync 模块 load_path.py 里的路径常量， 而不是抄一份清单。理由：清单抄两遍必然漂移（模块加了新接口，中央层忘了同步 = 又是 403）。 聚合后中央层与模块层永远一致；同时对 pbl_evidence 保留一份**硬编码兜底清单**， 即使模块脚本 import 失败也一定登记进去。 聚合形态兼容（M11b-2c-T3 QC 退回意见 #2 修复：消除「整模块静默 skip」） -------------------------------------------------------------------- 模块层允许下列任一写法，中央层逐条**归一**为 ``(path, role)``： ① 扁平二元组 ``PATHS = [('/x/a.dspy', 'logined'), ...]`` ② 扁平纯字符串 ``PATHS = ['/x/a.dspy', ...]`` → role 取模块 ``DEFAULT_ROLE``（缺省 logined） ③ 三元组及以上 ``PATHS = [('/x/a.dspy','logined','备注'), ...]`` → 取前二 ④ 分层常量 ``PATHS_ANY`` / ``PATHS_LOGINED`` / ``PATHS_OWNER_SUPERUSER`` / ``PATHS_OWNER_OPERATOR``（world_sync 采用此写法） ⑤ 汇总别名 ``ALL_PATHS``，或 ``API_PATHS`` + ``UI_PATHS`` （pbl_validation / pbl_domain_ext 采用此写法，此前被静默跳过 = 同类缺口） ⑥ 显式零路径 ``NO_WEB_PATHS = True``（纯库函数模块，如 pbl_common / pbl_appcodes） —— 必须**显式声明**，不允许以「常量留空」表达「无路径」。 **单条形态非法**（既不是 str 也不是 (list,tuple)）→ 只跳过该条并记 WARN，不再丢弃整模块。 一致性门禁（QC 退回意见 #2 第 ② 点） --------------------------------- ``--check`` 除通配符审计外，新增**聚合完整性门禁**：MODULES 中凡存在 ``scripts/load_path.py`` 的模块，若 ①import 失败 ②一条路径都没聚合到且没有 ``NO_WEB_PATHS`` 显式声明 ③存在形态非法的路径条目 → 计入 FAIL 清单并 **rc=1**， 不再出现「聚合失败但 check 仍 PASS」。 用法 ---- python3 scripts/load_path.py # 聚合全部模块路径并注册 python3 scripts/load_path.py --check # 一致性核对 + 通配符审计（不连库） python3 scripts/load_path.py --list # 打印最终在册清单 WS_DIR=/path/to/workspace python3 scripts/load_path.py # 指定机构工作空间 铁律：路径逐条显式，禁止通配符（``%`` / ``*``）。注意区分「RBAC 通配符」与 「Python 格式化占位符」：模块源码里 ``"/api/x/%s.dspy" % a`` 的 ``%s`` 是**构造期** 占位符，展开后的路径字符串不含 ``%``；本脚本审计的是**展开后**的运行时字符串。 ═══════════════════════════════════════════════════════════════════════════════ 里程碑 RBAC 声明台账（中央层，与模块层双写等价） ═══════════════════════════════════════════════════════════════════════════════ M11b-2（T3，2026-09-22）： 结论 = 无新路径登记（中央层同样零新增条目）。 理由 = M11b-2 里程碑（world_sync 列级差异比对 + pbl_runtime_ext 单事务事件/状态写入） 只交付 **Python 库函数** 与 models/*.json 表定义，**不新增任何 .dspy / .ui / .js / .css 可访问路径**，不经 HTTP 路由，故 RBAC 零登记项。 双写 = 等价声明同时写在 modules/world_sync/scripts/load_path.py 文件头台账、 modules/pbl_runtime_ext/scripts/load_path.py（如存在）、以及说明文件 modules/world_sync/wwwroot/api/README-N-A-M11b2.md。 附带修复（QC #2）= 见上「聚合形态兼容」+「一致性门禁」两节：pbl_blueprint（扁平 纯字符串 PATHS，139 条）与 pbl_validation / pbl_domain_ext（API_PATHS / ALL_PATHS）此前在中央层被静默 skip，本次全部聚合成功。 ═══════════════════════════════════════════════════════════════════════════════ """ 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_* 全量 + 宿主挂载的 world_sync） 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', 'world_sync', ] # 模块未显式给出 role 时的缺省授权层（可用模块常量 DEFAULT_ROLE 覆盖） DEFAULT_ROLE = 'logined' # 分层常量 → role 映射 TIER_ATTRS = ( ('PATHS_ANY', 'any'), ('PATHS_LOGINED', 'logined'), ('PATHS_OWNER_SUPERUSER', 'owner.superuser'), ('PATHS_OWNER_OPERATOR', 'owner.operator'), ) FORBIDDEN_WILDCARDS = ('%', '*') # ── 里程碑声明台账（中央层，与模块层双写；--check / --list 会原样回显）──────── MILESTONE_DECLARATIONS = [ { 'milestone': 'M11b-2', 'new_paths': 0, 'declaration': ( 'M11b-2 本里程碑不新增 .dspy/.ui/.js/.css 路径，仅提供库函数' '（world_sync / pbl_runtime_ext 的 *.py 与 models/*.json），' '故无 RBAC 登记项；禁通配符（%/*），' '如有路径须按 any/logined/owner.* 分层逐条登记（中央层与模块层双写）。' ), }, ] # ── 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'), ] # 授权严格度排序：数值越大越严格。冲突时保留更严格者，避免合并顺序放宽权限。 ROLE_STRICTNESS = {'any': 0, 'logined': 1, 'owner.operator': 2, 'owner.superuser': 3} def _stricter(a, b): return a if ROLE_STRICTNESS.get(a, 1) >= ROLE_STRICTNESS.get(b, 1) else b def _normalize(entry, mod, origin, bad): """把一条路径声明归一为 (path, role)；形态非法则记入 bad 并返回 None。 只跳过**这一条**，绝不因单条异形而丢弃整个模块（QC 退回意见 #2 第 ① 点）。 """ role_default = DEFAULT_ROLE if isinstance(entry, (tuple, list)): if len(entry) >= 2: # 二元组直取；三元组取前二 path, role = str(entry[0]), str(entry[1]) elif len(entry) == 1: # 单元素容器按纯字符串处理 path, role = str(entry[0]), role_default else: bad.append('%s: %s 含空条目 %r（已跳过该条）' % (mod, origin, entry)) return None elif isinstance(entry, str): # 纯字符串 → 默认 role path, role = entry, role_default else: bad.append('%s: %s 含非法条目类型 %s（已跳过该条）' % (mod, origin, type(entry).__name__)) return None path = path.strip() if not path.startswith('/'): bad.append('%s: %s 条目 %r 不是绝对路径（已跳过该条）' % (mod, origin, path)) return None return (path, role) def _load_module_paths(mod): """从 modules/{mod}/scripts/load_path.py 读路径常量（不执行其注册逻辑）。 返回 ``(paths, note, failed)``： * ``failed=True`` 表示该模块**聚合失败**（import 异常 / 零路径且无显式 ``NO_WEB_PATHS`` 声明 / 存在被跳过的非法条目），``--check`` 据此判 FAIL。 * 脚本不存在时返回 ``(None, 'no scripts/load_path.py', False)``——不算失败 （该模块尚无声明入口，由 MODULES 维护者决定何时补）。 """ 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', False try: spec = importlib.util.spec_from_file_location('_lp_' + mod, script) m = importlib.util.module_from_spec(spec) spec.loader.exec_module(m) except Exception as exc: # noqa: BLE001 return [], 'import failed: %s: %s' % (type(exc).__name__, exc), True default_role = getattr(m, 'DEFAULT_ROLE', DEFAULT_ROLE) if isinstance(default_role, str) and default_role: globals()['DEFAULT_ROLE'] = default_role # 本模块归一期间生效 else: default_role = DEFAULT_ROLE out, bad = [], [] try: paths = getattr(m, 'PATHS', None) if paths: for e in paths: got = _normalize(e, mod, 'PATHS', bad) if got: out.append(got) for attr, role in TIER_ATTRS: tier = getattr(m, attr, None) if not tier: continue for e in tier: got = _normalize(e, mod, attr, bad) if got: out.append((got[0], role)) if not out: # 写法 ⑤ 汇总别名 # ALL_PATHS 优先（它通常就是 API_PATHS+UI_PATHS 的并集，避免重复计数） alias_attrs = (('ALL_PATHS',) if getattr(m, 'ALL_PATHS', None) else ('API_PATHS', 'UI_PATHS')) for attr in alias_attrs: alias = getattr(m, attr, None) if not alias: continue for e in alias: got = _normalize(e, mod, attr, bad) if got: out.append(got) finally: globals()['DEFAULT_ROLE'] = DEFAULT_ROLE # 去重：模块同时提供扁平 PATHS 与分层常量时（如 pbl_common）会重复采集， # 同一 path 保留更严格的 role，避免计数虚高。 seen, deduped = {}, [] for p, r in out: if p in seen: deduped[seen[p]] = (p, _stricter(deduped[seen[p]][1], r)) continue seen[p] = len(deduped) deduped.append((p, r)) out = deduped if not out: if bad: return [], 'all entries malformed -> ' + '; '.join(bad), True if getattr(m, 'NO_WEB_PATHS', False): # 写法 ⑥ 显式零路径 return [], 'declares NO_WEB_PATHS=True (zero web paths, explicit)', False if getattr(m, 'MILESTONE_DECLARATIONS', None): return [], 'declares milestone台账 only (zero web paths)', False return [], ('PATHS / PATHS_* / ALL_PATHS empty and NO_WEB_PATHS not declared' ' -> aggregation FAILED'), True note = 'aggregated %d paths' % len(out) if bad: note += ' (WARN %d malformed entries skipped: %s)' % (len(bad), '; '.join(bad)) return out, note, bool(bad) def collect(): """聚合全部模块路径 + 兜底清单。 返回 ``(paths, notes, failures)``： * ``paths`` 去重排序后的 ``(path, role)`` 清单； * ``notes`` 逐模块回显行； * ``failures``聚合失败模块清单（``--check`` 非空即 rc=1）。 """ merged, notes, failures, conflicts = {}, [], [], [] for mod in MODULES: # 脚本存在性判断已下沉到 _load_module_paths()（got is None 即代表脚本缺失）， # 此处不再重复 stat（M11b-2c-T3 QC 退回意见 #14：清理不可达死代码）。 got, err, failed = _load_module_paths(mod) if mod == 'pbl_evidence': # 兜底清单始终并入（union），保证模块脚本坏掉也不漏登记 for p, r in PBL_EVIDENCE_FALLBACK: merged[p] = r if got is None: notes.append('%s: no scripts/load_path.py (fallback list applied)' % mod) continue if failed: failures.append('%s: %s (fallback list applied)' % (mod, err)) notes.append('%s: %s (fallback list applied)' % (mod, err)) elif not got: notes.append('%s: %s (fallback list applied)' % (mod, err)) else: notes.append('%s: aggregated %d paths from module script + fallback' % (mod, len(got))) else: if got is None: notes.append('%s: %s (skipped)' % (mod, err)) continue if not got and not failed: # 合法的空：模块显式声明 NO_WEB_PATHS / 仅里程碑台账 notes.append('%s: %s' % (mod, err)) continue if failed and not got: failures.append('%s: %s' % (mod, err)) notes.append('%s: %s (FAILED)' % (mod, err)) continue for p, r in got: if p in merged and merged[p] != r: keep = _stricter(merged[p], r) conflicts.append('%s: %s role %s vs %s -> keep %s (stricter)' % (mod, p, merged[p], r, keep)) merged[p] = keep else: merged[p] = r tag = ' (WARN partial)' if failed else '' notes.append('%s: aggregated %d paths%s' % (mod, len(got), tag)) if failed: failures.append('%s: %s' % (mod, err)) # 注：此处无需再判 got is None —— 上方各分支在 got is None 时均已 continue # （M11b-2c-T3 QC 退回意见 #14 删除的不可达分支）。 for c in conflicts: notes.append('[conflict] ' + c) return sorted((p, merged[p]) for p in merged), notes, failures 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 print_milestone_declarations(): print('[apps/pbls] 里程碑 RBAC 声明台账（中央层，与模块层双写等价）') for d in MILESTONE_DECLARATIONS: print(' - %s: 新增路径 %d 条 -> %s' % (d['milestone'], d['new_paths'], d['declaration'])) def main(): ap = argparse.ArgumentParser(description='pbls central RBAC registration (fallback layer)') ap.add_argument('--check', action='store_true', help='aggregate + wildcard audit + consistency gate (no DB)') ap.add_argument('--list', action='store_true', help='print aggregated paths and exit') args = ap.parse_args() print_milestone_declarations() paths, notes, failures = collect() for note in notes: print('[collect] ' + note) bad = [p for p, _r in paths if any(w in p for w in FORBIDDEN_WILDCARDS)] 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 and not failures) else 1 if args.check: # 门禁 1：通配符（零容忍）；门禁 2：聚合完整性（有脚本却聚合失败 = FAIL） print('[apps/pbls] consistency: module_scripts=%d aggregated=%d ' 'zero_path_declarations=%d failed_modules=%d' % (len([n for n in notes if 'no scripts/load_path.py' not in n]), len([n for n in notes if 'aggregated' in n]), len([n for n in notes if 'NO_WEB_PATHS' in n or '台账 only' in n]), len(failures))) for f in failures: print('AGGREGATION FAILED: ' + f) ok = (not bad) and (not failures) print('[apps/pbls] check: total=%d wildcard=%d aggregation_failures=%d -> %s' % (len(paths), len(bad), len(failures), 'PASS' if ok else 'FAIL')) return 0 if ok else 1 pending = register(paths) return 0 if (not pending and not bad and not failures) else 1 if __name__ == '__main__': sys.exit(main())