#!/usr/bin/env python3 # -*- coding: utf-8 -*- """M9-收口C 权限种子导入器（pbl_scense_ext 覆盖层 26 条路径，消 pending=26）。 背景：M9-收口B 一致性核验实测 `scripts/load_path.py`（register 分支）rc=1、 `rbac paths: total=26 ok=0 pending=26` —— 声明侧合规（--audit rc=0）但权限表零 落库，线上覆盖层 403。本脚本把 init/rbac_seed.json 的 26 条 permission / rolepermission 种子**幂等**导入应用库，并作为 scripts/rbac_set_perm.py 的底层 实现（load_path.register() 的 RBAC_SET_PERM 工具）。 实现模式对齐同项目已收口的 modules/pbl_runtime_ext/scripts/m11b34_rbac_seed.py （commit e141be2 幂等种子导入器 + b54ff00 traceability grep 实测）。 硬规定（module-development-spec 硬门禁 6.6 / rbac-permission-initialization -pattern / sqlor-database-module）： 1. **逐条显式登记，禁 `%` / `*` / `?` 通配符**——离线校验命中即 FAIL，不写半截。 2. **permission.path 唯一语义**：一个 path 一行 permission；多角色靠多条 rolepermission 指向同一 permid。rolepermission.id 显式生成（rbac 表无自增）。 3. **不猜列名**：permission/rolepermission/role 的实际列集合运行时从 information_schema 读取，只写交集列；缺关键列直接 FAIL 而非静默跳过。 4. **幂等**：permission 按 path 查重（存在则 UPDATE 非键列），rolepermission 按 (roleid, permid) 查重；id 用 md5(path)/md5(role|path) 派生的稳定 32 位串， 重复执行零差异、行数不增。 5. **logined / any 语义不混用**：种子里每条 permission 的 role 必须与其所在 role_grants 分桶一致，交叉即 FAIL。 6. **库名/凭据不硬编码**：dbname 依次取 --dbname > $PBL_SCENSE_EXT_DBNAME > 应用 conf/config.json 的 module_dbname 映射；连接参数取 databases[dbname].kwargs，可被 $PBL_DB_HOST/PORT/USER/PASSWORD 覆盖 （沙箱库验收用）。解析不到库名 → 直接 FAIL，不猜。 7. **超管授权运行时解析**（QC #12）：seed.superadmin_grant.role_names 是**角色 名称**候选，导入时查 role 表（列交集，按 role/name/rolelabel 命中）取真实 32 位 id 再写 rolepermission；解析不到打印 WARN 并在退出码中保持 0（伪角色 logined/any 已按 rbac userperm.py 语义覆盖全部登录/匿名主体），绝不静默。 用法： python3 scripts/m9c_rbac_seed.py --check # 离线校验，不连库（rc=0 即 errors=0） python3 scripts/m9c_rbac_seed.py --emit-sql x.sql # 出幂等 SQL（交 build.sh 第 9 步 import_rp） python3 scripts/m9c_rbac_seed.py --apply # 连库幂等导入 + 库内回查 python3 scripts/m9c_rbac_seed.py --report # 连库只读回查 26 条落库行数 python3 scripts/m9c_rbac_seed.py --upsert-one logined /pbl_scense_ext/xxx.dspy """ import argparse import hashlib import importlib.util import json import os import re import sys HERE = os.path.dirname(os.path.abspath(__file__)) MODULE_DIR = os.path.dirname(HERE) MODULE = 'pbl_scense_ext' APP_NAME = 'pbls' SEED_JSON = os.path.join(MODULE_DIR, 'init', 'rbac_seed.json') WWWROOT = os.path.join(MODULE_DIR, 'wwwroot') 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'] # role 表里可能承载「角色名称」的列（按实测 DDL：id/orgtypeid/role/rolelabel/name） ROLE_NAME_COLS = ['role', 'name', 'rolelabel'] # permission.ptype 按资源类别区分（api 端点 / page 页面壳 / static 静态资源） PTYPE_BY_KIND = {'api': 'api', 'page': 'page', 'static': 'static'} # rbac 伪角色（不是 role 表实体行，userperm.py 按主体类型直接判定；这里作为 # rolepermission.roleid 的授权键，与 pbl_runtime_ext 已收口做法一致） PSEUDO_ROLES = ('logined', 'any') 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_pairs(): """导入 scripts/load_path.py 读 PATHS（含循环 append 的 12 条页面壳）。 用 importlib 而非 AST：load_path.PATHS 由字面量 + for 循环 append 构成， ast.literal_eval 只能拿到字面量部分（会漏 13 条），三方一致性校验会假红。 load_path.py 有 __main__ 守卫，import 无副作用。 """ spec = importlib.util.spec_from_file_location('_pbl_scense_ext_load_path', LOAD_PATH_PY) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return [(p, r) for p, r in mod.PATHS], mod def path_to_fs(path): """URL 路径 → wwwroot 下期望文件/目录（目录型条目返回目录）。""" marker = '/%s' % MODULE if not path.startswith(marker): return None, None rel = path[len(marker):].lstrip('/') fs = os.path.join(WWWROOT, *rel.split('/')) if rel else WWWROOT return rel, fs def static_validate(seed): """离线三方一致性（种子 ↔ load_path.PATHS ↔ 磁盘文件）+ 通配符/语义门禁。""" errors, warns = [], [] perms = seed.get('permissions') or [] if not perms: return False, ['init/rbac_seed.json permissions 为空'], [] pairs, _mod = load_path_pairs() registered = set(pairs) names, paths = [], 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.append(name) if path in paths: errors.append('重复权限路径（permission.path 必须唯一）: %s' % path) paths.add(path) if WILDCARD_RE.search(path): errors.append('禁止通配符（硬政策）: %s' % path) if not path.startswith('/%s' % MODULE): errors.append('路径前缀不符合模块自动路由 /%s: %s' % (MODULE, path)) if role not in PSEUDO_ROLES: errors.append('role 必须是伪角色 %s 之一: %s -> %s' % (PSEUDO_ROLES, name, role)) rel, fs = path_to_fs(path) if fs and not (os.path.exists(fs) or os.path.isdir(fs)): errors.append('磁盘缺资源文件: %s (期望 %s)' % (path, os.path.relpath(fs, MODULE_DIR))) for key in ('traceability', 'milestone', 'title', 'kind', 'endpoint'): if not p.get(key): errors.append('缺 %s 字段: %s' % (key, name)) # 双向集合一致：种子 26 条 == load_path.PATHS 26 条（同 path 同 role） seed_pairs = set((p['path'], p['role']) for p in perms) for path, role in sorted(registered - seed_pairs): errors.append('load_path.py 有登记但种子缺失: %s (%s)' % (path, role)) for path, role in sorted(seed_pairs - registered): errors.append('种子有但 load_path.py PATHS 未登记: %s (%s)' % (path, role)) grants = seed.get('role_grants') or {} granted = set() for role, lst in grants.items(): for gname in lst: granted.add(gname) hit = [p for p in perms if p.get('name') == gname] if not hit: errors.append('role_grants[%s] 引用不存在的权限名: %s' % (role, gname)) elif hit[0].get('role') != role: # 退回意见 #12：logined / any 语义不得混用 errors.append('role_grants 分桶与 permission.role 不一致（语义混用）: %s 在 [%s] 桶但 role=%s' % (gname, role, hit[0].get('role'))) for p in perms: if p['name'] not in granted: warns.append('权限未进入任何 role_grants 分桶: %s' % p['name']) # 超管段语义（退回意见 #12）：role_ids 不得塞角色名；要么留真实 id，要么空 sup = seed.get('superadmin_grant') or {} for rid in sup.get('role_ids') or []: if not re.match(r'^[0-9a-f]{32}$', str(rid)): errors.append('superadmin_grant.role_ids 含非 32 位主键值（角色名应放 role_names）: %s' % rid) if not sup.get('role_names'): warns.append('superadmin_grant.role_names 为空：超管授权只能靠 role 表解析结果，未配置候选名') return (not errors), errors, warns # ── 库名 / 连接（禁止硬编码 DBNAME 与凭据）────────────────────────────────── def app_conf_candidates(): ws = os.path.dirname(os.path.dirname(MODULE_DIR)) return [os.path.join(ws, 'apps', APP_NAME, '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_SCENSE_EXT_DBNAME > 应用 module_dbname 映射。""" if getattr(args, 'dbname', None): return args.dbname, '--dbname' env_name = os.environ.get('PBL_SCENSE_EXT_DBNAME') if env_name: return env_name, '$PBL_SCENSE_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，允许 $PBL_DB_* 覆盖。""" conf, _path = read_app_conf() kw = {} entry = ((conf or {}).get('databases') or {}).get(dbname) or {} kw = dict(entry.get('kwargs') or {}) kw.pop('dbname', None) kw.pop('db', None) kw.pop('charset', None) if entry.get('password') and 'password' not in kw: kw['password'] = entry.get('password') kw.setdefault('host', os.environ.get('PBL_DB_HOST', '127.0.0.1')) kw.setdefault('port', int(os.environ.get('PBL_DB_PORT', '3306'))) for env_key, key in (('PBL_DB_USER', 'user'), ('PBL_DB_PASSWORD', 'password'), ('PBL_DB_SCHEMA', 'database')): if os.environ.get(env_key): kw[key] = os.environ[env_key] # conf 里没有该 dbname 的条目时（如沙箱库 --dbname），database 直接取解析出的库名 kw.setdefault('database', dbname) kw['charset'] = 'utf8mb4' kw['autocommit'] = False kw['connect_timeout'] = 6 return kw def get_conn(dbname): try: import pymysql except ImportError: return None return pymysql.connect(**db_kwargs(dbname)) 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()] # ── 行构造（id 稳定 → 幂等）──────────────────────────────────────────────── def perm_row(p, cols): pid = stable_id('rp_', p['path']) vals = { 'id': pid, 'path': p['path'], 'name': p['name'], 'title': p.get('title') or p['name'], 'ptype': PTYPE_BY_KIND.get(p.get('kind'), 'api'), 'parentid': MODULE, 'description': '%s | %s | role=%s | %s' % ( p.get('title') or p['name'], p.get('milestone') or '', p.get('role') or '', (p.get('traceability') or '')[:120]), 'icon': p.get('kind') or 'api', } return pid, {c: vals.get(c) for c in cols} def rp_row(role, pid, path, cols): rrow = {'id': stable_id('rr_', role, path)} if 'roleid' in cols: rrow['roleid'] = role if 'permid' in cols: rrow['permid'] = pid return rrow def build_rows(seed, perm_cols, rp_cols, superadmin_ids=()): """permission 每 path 一行；rolepermission = 伪角色分桶 + 解析到的超管实体角色。""" perm_rows, rp_rows, name_to = [], [], {} for p in seed['permissions']: pid, row = perm_row(p, perm_cols) name_to[p['name']] = (pid, p['path']) perm_rows.append((pid, row)) for role, lst in (seed.get('role_grants') or {}).items(): for n in lst: hit = name_to.get(n) if hit: rp_rows.append(rp_row(role, hit[0], hit[1], rp_cols)) for srole in superadmin_ids: for n, (pid, path) in name_to.items(): rp_rows.append(rp_row(srole, pid, path, rp_cols)) return perm_rows, rp_rows def upsert(cur, table, cols, rows, key_col, stats, key_val=None): """按 key_col 查重：存在 UPDATE 非键列，不存在 INSERT。rows 为 dict 列表。""" for r in rows: kv = key_val(r) if key_val else r[key_col] cur.execute('SELECT id FROM `%s` WHERE `%s` = %%s' % (table, key_col), (kv,)) hit = cur.fetchone() present = [c for c in cols if c in r] if hit: sets = [c for c in present if c != 'id'] if sets: cur.execute('UPDATE `%s` SET %s WHERE id = %%s' % (table, ', '.join('`%s` = %%s' % c for c in sets)), [r[c] for c in sets] + [hit[0]]) 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 resolve_superadmin(cur, dbname, seed, stats): """退回意见 #12：运行时按 role_names 查 role 表取真实 32 位 id，回填后授权。""" sup = seed.get('superadmin_grant') or {} names = [n for n in (sup.get('role_names') or []) if n] configured_ids = [r for r in (sup.get('role_ids') or []) if re.match(r'^[0-9a-f]{32}$', str(r))] if not sup.get('grant_all'): print(' INFO superadmin_grant.grant_all=false，跳过超管授权') return configured_ids cols = table_columns(cur, dbname, 'role') if not cols: stats['superadmin'] = 'WARN: role 表不存在，未授权超管' print(' WARN role 表不存在（rbac 未建表），超管授权跳过') return configured_ids name_cols = [c for c in ROLE_NAME_COLS if c in cols] found = list(configured_ids) for nc in name_cols: for n in names: cur.execute('SELECT id FROM `role` WHERE `%s` = %%s' % nc, (n,)) for row in cur.fetchall(): rid = str(row[0]) if len(rid) <= 32 and rid not in found: found.append(rid) print(' OK 超管角色解析命中: role.%s=%s -> id=%s' % (nc, n, rid)) if not found: stats['superadmin'] = 'WARN: 未解析到超管实体角色（候选名 %s）' % names print(' WARN 未解析到超管实体角色（候选 %s，扫描列 %s）；' 'logined/any 伪角色已覆盖全部登录/匿名主体，不 FAIL' % (names, name_cols)) else: stats['superadmin_resolved'] = len(found) return found def emit_sql(seed, perm_cols, rp_cols, out_path, superadmin_ids=()): perm_rows, rp_rows = build_rows(seed, perm_cols, rp_cols, superadmin_ids) lines = ['-- M9-收口C pbl_scense_ext 覆盖层权限种子（26 条，逐条显式，无 % / * 通配符）', '-- 由 scripts/m9c_rbac_seed.py 生成；id 为 path/role 的 md5 派生 → 幂等（可重复执行）', '-- 用法：交 apps/pbls/build.sh 第 9 步（import_rp）或直接 mysql < 本文件', ''] 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 += ['', '-- 校验：permission 应为 26 行，rolepermission 应为 26 行（logined 22 + any 4）', "SELECT COUNT(1) AS perm_cnt FROM `permission` WHERE `parentid` = '%s';" % MODULE] with open(out_path, 'w', encoding='utf-8') as f: f.write('\n'.join(lines) + '\n') return len(perm_rows), len(rp_rows) def verify_db(cur, dbname, seed, stats): paths = [p['path'] for p in seed['permissions']] ph = ', '.join(['%s'] * len(paths)) cur.execute('SELECT COUNT(1) FROM `permission` WHERE `path` IN (%s)' % ph, paths) 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 IN (%s)' % ph, paths) stats['verify_rolepermission_in_db'] = cur.fetchone()[0] cur.execute('SELECT COUNT(DISTINCT rp.roleid) FROM `rolepermission` rp JOIN ' '`permission` p ON rp.permid = p.id WHERE p.path IN (%s)' % ph, paths) stats['verify_distinct_roles'] = cur.fetchone()[0] return stats def report_db(cur, dbname, seed): """逐条打印 26 条路径在 permission / rolepermission 的实测行数（验收证据）。""" print(' %-58s %-8s %6s %6s' % ('path', 'role', 'perm', 'rolem')) total_p = total_r = 0 for p in seed['permissions']: cur.execute('SELECT id FROM `permission` WHERE `path` = %s', (p['path'],)) rows = cur.fetchall() pid = rows[0][0] if rows else None if pid: cur.execute('SELECT COUNT(1) FROM `rolepermission` WHERE `permid` = %s', (pid,)) rc = cur.fetchone()[0] else: rc = 0 total_p += len(rows) total_r += rc print(' %-58s %-8s %6d %6d' % (p['path'], p['role'], len(rows), rc)) print(' TOTAL permission=%d rolepermission=%d（期望 26 / >=26）' % (total_p, total_r)) return total_p, total_r def apply_seed(args, single=None): """连库幂等导入。single=(role, path) 时只 upsert 一条（rbac_set_perm.py 用）。""" seed = load_seed() ok, errors, warns = static_validate(seed) for e in errors: print(' FAIL ' + e) if not ok: return 1 dbname, src = resolve_dbname(args) if not dbname: print(' FAIL 无法解析 %s 的库名（--dbname / $PBL_SCENSE_EXT_DBNAME / ' '应用 conf module_dbname 均缺）' % MODULE) return 1 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)[:120]) return 2 if conn is None: print(' SKIP 无 pymysql：改用 --emit-sql 出 SQL 交 build.sh 第 9 步导入') 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）' % (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] if single: role, path = single hit = [p for p in seed['permissions'] if p['path'] == path] if not hit: print(' FAIL 种子清单里没有这条路径: %s' % path) conn.close() return 1 if hit[0]['role'] != role: print(' FAIL 角色与种子声明不一致: %s 应为 %s（传入 %s）' % (path, hit[0]['role'], role)) conn.close() return 1 pid, row = perm_row(hit[0], use_perm) stats = {} stats = upsert(cur, 'permission', use_perm, [row], 'path', stats) stats = upsert(cur, 'rolepermission', use_rp, [rp_row(role, pid, path, use_rp)], 'id', stats) conn.commit() print('[%s] upsert-one ok: role=%s path=%s %s' % (MODULE, role, path, json.dumps(stats, ensure_ascii=False))) conn.close() return 0 stats = {} sup_ids = resolve_superadmin(cur, dbname, seed, stats) perm_rows, rp_rows = build_rows(seed, use_perm, use_rp, sup_ids) if not (args.apply or args.report): print(' DRY：加 --apply 才写库，或 --emit-sql 出 SQL') conn.close() return 0 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() verify_db(cur, dbname, seed, stats) print('[%s] 导入结果 dbname=%s(%s): %s' % (MODULE, dbname, src, json.dumps(stats, ensure_ascii=False))) if args.report: report_db(cur, dbname, seed) expect_p = len(seed['permissions']) good = stats.get('verify_permission_in_db') == expect_p \ and stats.get('verify_rolepermission_in_db', 0) >= expect_p print(' %s 库内回查：permission=%s rolepermission=%s（期望 %d / >=%d）' % ('OK' if good else 'FAIL', stats.get('verify_permission_in_db'), stats.get('verify_rolepermission_in_db'), expect_p, expect_p)) conn.close() return 0 if good else 1 def main(): ap = argparse.ArgumentParser() ap.add_argument('--check', action='store_true', help='只做离线校验，不连库') ap.add_argument('--apply', action='store_true', help='连库幂等导入') ap.add_argument('--report', action='store_true', help='连库幂等导入 + 逐条落库行数回查') ap.add_argument('--emit-sql', default=None, help='输出幂等 SQL 文件（不连库）') ap.add_argument('--upsert-one', nargs=2, metavar=('ROLE', 'PATH'), help='单条幂等 upsert（scripts/rbac_set_perm.py 的内部入口）') ap.add_argument('--dbname', default=None) args = ap.parse_args() seed = load_seed() ok, errors, warns = static_validate(seed) pairs, _mod = load_path_pairs() print('[%s] rbac_seed 静态校验: seed=%d load_path.PATHS=%d errors=%d warns=%d' % (MODULE, len(seed['permissions']), len(pairs), len(errors), len(warns))) for e in errors: print(' FAIL ' + e) for w in warns: print(' WARN ' + w) if not ok: return 1 print(' OK 26 条路径逐条显式登记（无 %% / * 通配符）；' '种子 ↔ load_path.PATHS ↔ 磁盘资源 三方一致；logined/any 分桶无混用' % ()) if args.check: print('[%s] --check 通过（离线，未连库）' % MODULE) return 0 if args.upsert_one: return apply_seed(args, single=tuple(args.upsert_one)) 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 return apply_seed(args) if __name__ == '__main__': sys.exit(main())