#!/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 下全部 12 个 .dspy + 模块入口 index.ui（与磁盘一一对应， # `python3 scripts/load_path.py --check` 做自愈核对）。M5a 主能力 # pbl_evidence_collect_from_events.dspy 与统计口 pbl_evidence_stats.dspy 在此登记， # CRUD 适配三端点 pbl_artifact_options / pbl_evidence_update / pbl_evidence_delete # （json/pbl_evidence.json 的 editable 三 URL + alters.artifact_id.dataurl 所指）也必须登记， # 漏登记 = 部署后点击一律 403，整条批量采集/编辑/删除链路不可用（QC 退回意见 #1/#3 的实测后果）。 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'), # CRUD 框架适配端点（json/pbl_evidence.json editable 三 URL + artifact_id 下拉） ('/pbl_evidence/api/pbl_artifact_options.dspy', 'logined'), ('/pbl_evidence/api/pbl_evidence_update.dspy', 'logined'), ('/pbl_evidence/api/pbl_evidence_delete.dspy', 'logined'), # ── CRUD 生成物见下方 GENERATED_PATHS（xls2ui 产物，build 时才落盘）── ] # xls2ui 从 json/pbl_artifact.json / json/pbl_evidence.json 生成的 CRUD 目录 # （alias = tblname = pbl_artifact / pbl_evidence）。每个 alias 6 条：目录 + # index.ui + get_/add_/update_/delete_ 端点，逐条显式、禁通配符。这些文件是 # **构建产物**（.gitignore 排除、不入库），所以只出现在部署环境的 wwwroot 下； # 注册权限时必须一并登记，否则 CRUD 列表页/新增/编辑/删除一律 403。 CRUD_ALIASES = ('pbl_artifact', 'pbl_evidence') GENERATED_PATHS = [] for _alias in CRUD_ALIASES: GENERATED_PATHS.append(('/pbl_evidence/%s' % _alias, 'logined')) GENERATED_PATHS.append(('/pbl_evidence/%s/index.ui' % _alias, 'logined')) for _op in ('get', 'add', 'update', 'delete'): GENERATED_PATHS.append( ('/pbl_evidence/%s/%s_%s.dspy' % (_alias, _op, _alias), 'logined')) # 注册用的全量清单 = 手写端点 + CRUD 生成物（顺序即登记顺序） PATHS = PATHS + GENERATED_PATHS # ── permission / rolepermission 幂等 INSERT（供部署说明与手工兜底使用）────── # 与 PATHS 一一对应：rbac 三表链路 permission.permcode = 路径本身， # rolepermission 关联 role.rolecode='logined'。列名以 models 真实字段为准 # （permission: permcode/name；role: rolecode；rolepermission: roleid/permid）， # 用 INSERT ... SELECT 反查 id，避免硬编码主键。 def permission_inserts(): """返回与 PATHS 等价的幂等 SQL（角色固定 logined）。""" out = [] for path, role in PATHS: out.append( "INSERT INTO `permission` (`id`, `permcode`, `name`) " "SELECT REPLACE(UUID(),'-',''), '%(p)s', '%(p)s' " "WHERE NOT EXISTS (SELECT 1 FROM `permission` WHERE `permcode`='%(p)s');" % {'p': path}) out.append( "INSERT INTO `rolepermission` (`id`, `roleid`, `permid`) " "SELECT REPLACE(UUID(),'-',''), r.`id`, p.`id` FROM `role` r, `permission` p " "WHERE r.`rolecode`='%(rl)s' AND p.`permcode`='%(p)s' " "AND NOT EXISTS (SELECT 1 FROM `rolepermission` rp " "WHERE rp.`roleid`=r.`id` AND rp.`permid`=p.`id`);" % {'rl': role, 'p': path}) return out def print_permission_sql(): """--sql 子命令：打印 INSERT 清单（部署时可直接管道进 mysql）。""" for stmt in permission_inserts(): print(stmt) 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(include_generated=False): """清单 vs 磁盘一致性核对，返回 (missing, stale, wildcard) 三个列表。 CRUD 生成物（GENERATED_PATHS）是 build.sh 里 xls2ui 的产物，开发仓库中被 .gitignore 排除、不落盘，因此默认不计入 stale（否则开发环境永远误报）； 部署后跑 `--check --generated` 才把它们纳入核对。手写端点与磁盘必须一一对应。 """ handwritten = set(p for p, _r in PATHS) - set(p for p, _r in GENERATED_PATHS) registered = set(p for p, _role in PATHS) if include_generated else handwritten 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('--generated', action='store_true', help='with --check: also reconcile xls2ui-generated CRUD paths ' '(run this on the deployment host, after build.sh)') ap.add_argument('--list', action='store_true', help='print registered paths and exit') ap.add_argument('--sql', action='store_true', help='print idempotent permission/rolepermission INSERTs and exit') args = ap.parse_args() if args.sql: print_permission_sql() return 0 if args.list: for path, role in PATHS: print('%s %s' % (role, path)) return 0 if args.check: missing, stale, wildcard = check(include_generated=args.generated) 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())