#!/usr/bin/env python3 # -*- coding: utf-8 -*- """M11b-3 / M11b-4 权限种子导入器（pbl_runtime_ext 广播/轮询端点，8 条）。 背景：anchor3 自检 `scripts/load_path.py` rc=1，8 条 PENDING 全部属 M11b-3 服务端广播与 M11b-4 3s 轮询端点。本脚本把这 8 条 permission / rolepermission 种子**幂等**导入应用库，使 load_path.py 的自检收敛到 rc=0。 硬规定（module-development-spec 6.6 / crud-definition-spec references/endpoint-registration.md / sqlor-database-module）： 1. **逐条显式登记，禁 `%` / `*` / `?` 通配符**——导入前静态校验，命中即 FAIL 不写半截。 2. **load_path.py 单独不够**：新 api/*.dspy 必须同时落 DB `permission` + `rolepermission`，否则线上 403。角色用 `logined`（登录即可访问）。 3. **不猜列名**：permission/rolepermission 的实际列集合运行时从 information_schema 读取，只写交集列；缺关键列直接 FAIL 而非静默跳过。 4. **幂等**：permission 按 path 查重（存在则 UPDATE 非键列），rolepermission 按 (roleid, permid) 查重；id 用 `md5` 派生的稳定 32 位串，重复执行零差异。 5. **库名不硬编码**：dbname 依次取 --dbname > $PBL_RUNTIME_EXT_DBNAME > 应用 conf/config.json 的 module_dbname 映射（真实值由应用 get_module_dbname 决定）。**三级来源均缺 → 直接 FAIL 退出，不猜库名、无 'pbl' 兜底** （与 resolve_dbname() 实现一致：解析不到即返回 (None, 'unresolved')）。 真实表结构（rbac，见 modules/scense/scripts/ddl.sql 实测 DDL）： permission(id, name, description, ptype, parentid, path, title, icon) rolepermission(id, roleid, permid) 用法： python3 scripts/m11b34_rbac_seed.py --check # 离线校验，不连库（rc=0 即 errors=0） python3 scripts/m11b34_rbac_seed.py --emit-sql x.sql # 出 SQL 交 build.sh 第 9 步 python3 scripts/m11b34_rbac_seed.py --apply # 连库幂等导入 """ import argparse import hashlib import json import os import re import sys HERE = os.path.dirname(os.path.abspath(__file__)) MODULE_DIR = os.path.dirname(HERE) MODULE = 'pbl_runtime_ext' SEED_JSON = os.path.join(MODULE_DIR, 'init', 'rbac_seed.json') WWWROOT_API = os.path.join(MODULE_DIR, 'wwwroot', 'api') LOAD_PATH_PY = os.path.join(HERE, 'load_path.py') WILDCARD_RE = re.compile(r'[%*?]') # rbac 真实列（information_schema 取交集后才使用，缺列 = FAIL，不静默降级） PERM_REQUIRED = ['id', 'path'] PERM_OPTIONAL = ['name', 'description', 'ptype', 'parentid', 'title', 'icon'] RP_REQUIRED = ['id', 'roleid', 'permid'] # permission.ptype：api 类权限（与 rbac 菜单类权限区分，见 appbase 菜单 ptype='menu'） PTYPE_API = 'api' def load_seed(): with open(SEED_JSON, 'r', encoding='utf-8') as f: return json.load(f) def stable_id(prefix, *parts): """32 位稳定 id（重复执行幂等，不用 UUID——UUID 每次不同会插重复行）。""" h = hashlib.md5(('|'.join([MODULE] + [str(p) for p in parts])).encode('utf-8')) return ('%s%s' % (prefix, h.hexdigest()))[:32] def load_path_registered(): """从 scripts/load_path.py 以 AST 解析 PATHS 的 (path, role) 集合（不 exec）。""" import ast with open(LOAD_PATH_PY, 'r', encoding='utf-8') as f: tree = ast.parse(f.read(), filename=LOAD_PATH_PY) pairs = set() for node in ast.walk(tree): if isinstance(node, ast.Assign): for tgt in node.targets: if isinstance(tgt, ast.Name) and tgt.id == 'PATHS': for item in ast.literal_eval(node.value): if isinstance(item, (list, tuple)) and len(item) == 2: pairs.add((item[0], item[1])) return pairs def static_validate(seed): """离线三方一致性（种子 ↔ 磁盘 .dspy ↔ load_path.PATHS）+ 通配符门禁。""" errors = [] perms = seed.get('permissions') or [] if not perms: return False, ['init/rbac_seed.json permissions 为空'] disk = set(f for f in os.listdir(WWWROOT_API) if f.endswith('.dspy')) \ if os.path.isdir(WWWROOT_API) else set() registered = load_path_registered() names, paths = set(), set() for p in perms: name, path, role = p.get('name') or '', p.get('path') or '', p.get('role') or '' if name in names: errors.append('重复权限名: %s' % name) names.add(name) paths.add(path) if WILDCARD_RE.search(path): errors.append('禁止通配符（硬政策）: %s' % path) if not path.startswith('/%s/api/' % MODULE): errors.append('路径前缀不符合模块自动路由 /%s/api/: %s' % (MODULE, path)) if os.path.basename(path) not in disk: errors.append('磁盘缺 .dspy 端点文件: %s' % os.path.basename(path)) if (path, role) not in registered: errors.append('scripts/load_path.py PATHS 未登记: %s (%s)' % (path, role)) if not p.get('traceability'): errors.append('缺 traceability（无法追溯到端点与角色）: %s' % name) if not p.get('milestone'): errors.append('缺 milestone 归属: %s' % name) for path, role in sorted(registered): if path not in paths: errors.append('load_path.py 有登记但种子清单缺失: %s (%s)' % (path, role)) for role, lst in (seed.get('role_grants') or {}).items(): for gname in lst: if gname not in names: errors.append('role_grants[%s] 引用不存在的权限名: %s' % (role, gname)) return (not errors), errors def app_conf_candidates(): """应用 conf/config.json 候选路径（库名/连接从应用配置读，禁止在本模块硬编码 DBNAME）。""" ws = os.path.dirname(os.path.dirname(MODULE_DIR)) # .../sdlc_general return [os.path.join(ws, 'apps', 'pbls', 'conf', 'config.json'), os.environ.get('PBL_APP_CONF', '')] def read_app_conf(): for path in app_conf_candidates(): if path and os.path.isfile(path): try: with open(path, 'r', encoding='utf-8') as f: return json.load(f), path except Exception: # noqa: BLE001 continue return None, None def resolve_dbname(args): """dbname 优先级：--dbname > $PBL_RUNTIME_EXT_DBNAME > 应用 module_dbname 映射。""" if args.dbname: return args.dbname, '--dbname' env_name = os.environ.get('PBL_RUNTIME_EXT_DBNAME') if env_name: return env_name, '$PBL_RUNTIME_EXT_DBNAME' conf, path = read_app_conf() if conf: m = (conf.get('module_dbname') or {}) name = m.get(MODULE) or m.get('default') if name: return name, 'conf:%s' % path return None, 'unresolved' def db_kwargs(dbname): """从应用 conf 的 databases[dbname].kwargs 取连接参数（不硬编码凭据）。""" conf, _path = read_app_conf() kw = {} if conf: entry = (conf.get('databases') or {}).get(dbname) or {} kw = dict(entry.get('kwargs') or {}) kw.setdefault('host', os.environ.get('PBL_DB_HOST', '127.0.0.1')) kw.setdefault('port', int(os.environ.get('PBL_DB_PORT', '3306'))) return kw def table_columns(cur, dbname, table): cur.execute("SELECT COLUMN_NAME FROM information_schema.COLUMNS " "WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s", (dbname, table)) return [r[0] for r in cur.fetchall()] def perm_values(p, cols, role_hint='logined'): """按真实列集合构造 permission 行（只写交集列）。""" pid = stable_id('rp_', p['path']) row = {} vals = { 'id': pid, 'path': p['path'], 'name': p['name'], 'title': p.get('title') or p['name'], 'ptype': PTYPE_API, 'parentid': 'pbl_runtime_ext', 'description': '%s | %s | role=%s | %s' % ( p.get('title') or p['name'], p.get('milestone') or '', p.get('role') or role_hint, p.get('traceability') or ''), 'icon': 'api', } for c in cols: row[c] = vals.get(c) return pid, row def build_rows(seed, perm_cols, rp_cols): """生成 (permission 行, rolepermission 行) 列表，全部 id 稳定 → 幂等。""" perm_rows, rp_rows, name_to_pid = [], [], {} for p in seed['permissions']: pid, row = perm_values(p, perm_cols) name_to_pid[p['name']] = pid perm_rows.append((pid, row)) for role, names in (seed.get('role_grants') or {}).items(): for n in names: pid = name_to_pid.get(n) if not pid: continue rrow = {'id': stable_id('rr_', role, pid)} if 'roleid' in rp_cols: rrow['roleid'] = role if 'permid' in rp_cols: rrow['permid'] = pid rp_rows.append(rrow) return perm_rows, rp_rows def upsert(cur, table, cols, rows, key_col, stats): """按 key_col 查重：存在 UPDATE，不存在 INSERT。rows 为 dict 列表。""" for r in rows: cur.execute('SELECT COUNT(1) FROM `%s` WHERE `%s` = %%s' % (table, key_col), (r[key_col],)) exists = bool(cur.fetchone()[0]) present = [c for c in cols if c in r] if exists: sets = ', '.join('`%s` = %%s' % c for c in present if c != key_col) if sets: cur.execute('UPDATE `%s` SET %s WHERE `%s` = %%s' % (table, sets, key_col), [r[c] for c in present if c != key_col] + [r[key_col]]) stats['%s_updated' % table] = stats.get('%s_updated' % table, 0) + 1 else: ph = ', '.join(['%s'] * len(present)) cur.execute('INSERT INTO `%s` (%s) VALUES (%s)' % (table, ', '.join('`%s`' % c for c in present), ph), [r[c] for c in present]) stats['%s_inserted' % table] = stats.get('%s_inserted' % table, 0) + 1 return stats def emit_sql(seed, perm_cols, rp_cols, path): perm_rows, rp_rows = build_rows(seed, perm_cols, rp_cols) lines = ['-- M11b-3/M11b-4 pbl_runtime_ext 广播/轮询端点权限种子（8 条，逐条显式，无通配符）', '-- 由 scripts/m11b34_rbac_seed.py 生成，幂等（id 为 path/role 的 md5 派生）', ''] for pid, r in perm_rows: present = [c for c in perm_cols if c in r] vals = ', '.join("'%s'" % (str(r[c]).replace("'", "''")) for c in present) lines.append('INSERT INTO `permission` (%s) VALUES (%s)' % (', '.join('`%s`' % c for c in present), vals) + ' ON DUPLICATE KEY UPDATE ' + ', '.join('`%s`=VALUES(`%s`)' % (c, c) for c in present if c != 'id') + ';') lines.append('') for r in rp_rows: present = [c for c in rp_cols if c in r] vals = ', '.join("'%s'" % r[c] for c in present) lines.append('INSERT INTO `rolepermission` (%s) VALUES (%s)' % (', '.join('`%s`' % c for c in present), vals) + ' ON DUPLICATE KEY UPDATE `permid`=VALUES(`permid`);') lines.append('') lines.append('-- 校验：应为 8 条 permission + 8 条 rolepermission') lines.append("SELECT COUNT(1) AS perm_cnt FROM `permission` " "WHERE `path` LIKE '/pbl\\_runtime\\_ext/api/%';") with open(path, 'w', encoding='utf-8') as f: f.write('\n'.join(lines) + '\n') return len(perm_rows), len(rp_rows) def get_conn(dbname): try: import pymysql except ImportError: return None kw = db_kwargs(dbname) kw['charset'] = 'utf8mb4' kw['autocommit'] = False kw['connect_timeout'] = 6 return pymysql.connect(**kw) def main(): ap = argparse.ArgumentParser() ap.add_argument('--check', action='store_true', help='只做离线校验，不连库') ap.add_argument('--apply', action='store_true', help='连库幂等导入') ap.add_argument('--emit-sql', default=None, help='输出幂等 SQL 文件（不连库）') ap.add_argument('--dbname', default=None) args = ap.parse_args() seed = load_seed() ok, errors = static_validate(seed) print('[%s] rbac_seed 静态校验: permissions=%d errors=%d' % (MODULE, len(seed['permissions']), len(errors))) for e in errors: print(' FAIL ' + e) if not ok: return 1 print(' OK 全部路径逐条显式登记（无 % / * 通配符）；' '种子 ↔ 磁盘 .dspy ↔ load_path.PATHS 三方一致') if args.check: print('[%s] --check 通过（离线，未连库）' % MODULE) return 0 dbname, src = resolve_dbname(args) if not dbname: print(' FAIL 无法解析 %s 的库名（--dbname / $PBL_RUNTIME_EXT_DBNAME / ' '应用 conf module_dbname 均缺）' % MODULE) return 1 print(' dbname=%s (来源 %s)' % (dbname, src)) if args.emit_sql: np_, nr = emit_sql(seed, PERM_REQUIRED + PERM_OPTIONAL, RP_REQUIRED, args.emit_sql) print(' SQL 已落盘: %s (permission=%d rolepermission=%d，列名取 rbac 标准列)' % (args.emit_sql, np_, nr)) return 0 try: import pymysql # noqa: F401 except ImportError: print(' SKIP 无 pymysql：改用 --emit-sql 出 SQL 交 build.sh 第 9 步导入') return 2 try: conn = get_conn(dbname) except Exception as e: # noqa: BLE001 print(' SKIP 连库失败（%s）：DB 不可达时用 --emit-sql 出 SQL，' '交 build.sh 第 9 步（import_rp）导入' % str(e)[:100]) return 2 cur = conn.cursor() try: perm_cols = table_columns(cur, dbname, 'permission') rp_cols = table_columns(cur, dbname, 'rolepermission') except Exception as e: # noqa: BLE001 print(' SKIP 无法读取 %s 库的 rbac 表结构（%s）：' '先执行 rbac 建表或用 --emit-sql 出 SQL' % (dbname, str(e)[:80])) conn.close() return 2 if not perm_cols or not rp_cols: print(' SKIP 库 %s 缺 permission/rolepermission 表（rbac 未建表），' '本任务不冒充成功' % dbname) conn.close() return 2 missing = [c for c in PERM_REQUIRED if c not in perm_cols] \ + [c for c in RP_REQUIRED if c not in rp_cols] if missing: print(' FAIL rbac 表缺关键列: %s（实际列 %s / %s）' % (missing, perm_cols, rp_cols)) conn.close() return 1 use_perm = [c for c in PERM_REQUIRED + PERM_OPTIONAL if c in perm_cols] use_rp = [c for c in RP_REQUIRED if c in rp_cols] perm_rows, rp_rows = build_rows(seed, use_perm, use_rp) if not args.apply: print(' DRY：加 --apply 才写库，或 --emit-sql 出 SQL') conn.close() return 0 stats = {} stats = upsert(cur, 'permission', use_perm, [r for _, r in perm_rows], 'path', stats) stats = upsert(cur, 'rolepermission', use_rp, rp_rows, 'id', stats) conn.commit() cur.execute("SELECT COUNT(1) FROM `permission` WHERE `path` LIKE %s", ('/%s/api/%%' % MODULE,)) stats['verify_permission_in_db'] = cur.fetchone()[0] cur.execute("SELECT COUNT(1) FROM `rolepermission` rp JOIN `permission` p " "ON rp.permid = p.id WHERE p.path LIKE %s", ('/%s/api/%%' % MODULE,)) stats['verify_rolepermission_in_db'] = cur.fetchone()[0] conn.close() print('[%s] 导入结果: %s' % (MODULE, json.dumps(stats, ensure_ascii=False))) good = (stats.get('verify_permission_in_db') == len(seed['permissions']) and stats.get('verify_rolepermission_in_db') == len(seed['permissions'])) print(' %s 库内回查：permission=%s rolepermission=%s 期望各 %d' % ('OK' if good else 'FAIL', stats.get('verify_permission_in_db'), stats.get('verify_rolepermission_in_db'), len(seed['permissions']))) return 0 if good else 1 if __name__ == '__main__': sys.exit(main())