#!/usr/bin/env python3 # -*- coding: utf-8 -*- """pbl_evidence RBAC 路径注册(硬门禁 6.6 / QC 退回意见 #1 修复)。 约定 ---- - 路径 = 模块自动路由 ``/pbl_evidence/api/<契约>.dspy``,不带端口、不带 /wss 前缀; - 角色 ``logined`` = 登录即可访问;本模块无 ``any``(免登录)资源——所有接口都要登录态, 证据写入是服务端权威行为,绝不允许匿名调用; - 由 apps/pbls/build.sh 的 RBAC 步骤调用 ``register()``; rbac CLI(set_role_perm.py)不在位时逐条打印 PENDING 清单并以非零码退出(不静默跳过)。 铁律(module-development-spec / QC 硬门禁) ------------------------------------------ 1. **禁止任何通配符**:每条路径逐条显式列出,``%`` / ``*`` 一个都不允许出现在 PATHS 里 (通配符会让未注册的新接口被误放行,等于 RBAC 形同虚设)。 2. **清单与磁盘一一对应**:wwwroot 下每个 .dspy / .ui 都必须在册;新增文件必须同步 追加条目。用 ``python3 scripts/load_path.py --check`` 做自愈核对: 磁盘有文件但未在册 → 打印 MISSING 并非零退出;在册但文件不存在 → 打印 STALE。 3. **双层回退**:本脚本是「模块层」。中央宿主 ``apps/pbls/scripts/load_path.py`` 另有一份等价清单(不同代码路径,可靠回退)——模块层 set_role_perm 静默失败时, 中央层仍能把权限写进去。两层都必须登记本模块全部路径。 用法 ---- python3 scripts/load_path.py # 注册(rbac CLI 不在位则打印 PENDING) python3 scripts/load_path.py --check # 只做清单与磁盘的一致性核对 python3 scripts/load_path.py --list # 打印在册清单 RBAC_SET_PERM=/path/set_role_perm.py python3 scripts/load_path.py """ import argparse import os import subprocess import sys MODULE = 'pbl_evidence' WWWROOT = os.path.normpath( os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'wwwroot')) # ── (path, role) 清单:逐条显式,禁通配符 ───────────────────────────────── # 覆盖 wwwroot 下全部 9 个 .dspy + 模块入口 index.ui。M5a 主能力 # pbl_evidence_collect_from_events.dspy 与统计口 pbl_evidence_stats.dspy 在此登记, # 漏登记 = 部署后点击一律 403,整条批量采集链路不可用(QC 退回意见 #1 的实测后果)。 PATHS = [ # 模块入口页(wwwroot/index.ui) ('/pbl_evidence/index.ui', 'logined'), # 产出物 pbl_artifact CRUD(5 条) ('/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_runtime_event 采集(M5a 主能力) ('/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'), ] STATIC_SUFFIXES = ('.dspy', '.ui', '.js', '.css', '.wss') def find_set_perm(): """定位 rbac 的 set_role_perm.py:环境变量 > 常见 Sage 根 > PATH。""" 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 on_disk_paths(): """扫描 wwwroot 真实文件,换算成路由路径,用于与 PATHS 一致性核对。""" got = set() if not os.path.isdir(WWWROOT): return got for base, dirs, files in os.walk(WWWROOT): dirs[:] = [d for d in dirs if d not in ('__pycache__', 'node_modules')] for fn in files: if not fn.endswith(STATIC_SUFFIXES): continue rel = os.path.relpath(os.path.join(base, fn), WWWROOT).replace(os.sep, '/') got.add('/%s/%s' % (MODULE, rel)) return got def check(): """清单 vs 磁盘一致性核对,返回 (missing, stale, wildcard) 三个列表。""" registered = set(p for p, _role in PATHS) on_disk = on_disk_paths() missing = sorted(on_disk - registered) stale = sorted(registered - on_disk) wildcard = sorted(p for p, _role in PATHS if '%' in p or '*' in p) return missing, stale, wildcard def register(verbose=True): """逐条调用 set_role_perm.py 注册;返回未成功(待注册)的路径列表。""" tool = find_set_perm() if not tool: # rbac CLI 不在位:不静默跳过,逐条打印待注册清单。部署机由 build.sh 在装有 # rbac 的 venv 中重跑;中央宿主 apps/pbls/scripts/load_path.py 为第二层回退。 if verbose: print('[%s] set_role_perm.py not found ' '(check RBAC_SET_PERM / sage root / PATH)' % MODULE) return [p for p, _role 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('[%s] rbac paths: total=%d ok=%d pending=%d' % (MODULE, 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='%s RBAC path registration' % MODULE) ap.add_argument('--check', action='store_true', help='verify PATHS against wwwroot files') ap.add_argument('--list', action='store_true', help='print registered paths and exit') args = ap.parse_args() if args.list: for path, role in PATHS: print('%s %s' % (role, path)) return 0 if args.check: missing, stale, wildcard = check() print('[%s] PATHS entries: %d' % (MODULE, len(PATHS))) for p in missing: print('MISSING (on disk, not registered): ' + p) for p in stale: print('STALE (registered, file absent): ' + p) for p in wildcard: print('WILDCARD FORBIDDEN: ' + p) ok = not missing and not stale and not wildcard print('[%s] check: %s' % (MODULE, 'PASS' if ok else 'FAIL')) return 0 if ok else 1 pending = register() return 0 if not pending else 1 if __name__ == '__main__': sys.exit(main())