#!/usr/bin/env python3 # -*- coding: utf-8 -*- """pbl_compiler 接线/RBAC 一致性机械审计（QC 硬门禁自证工具）。 背景：连续多轮 QC 退回的根因都是「写了但没接线 / 接了线但没登记 RBAC」—— 人肉声明不可信，故本脚本用机械比对把四张登记面钉死一致： ① pbl_compiler/init.py CONTRACTS 契约名集合 ② pbl_compiler/__init__.py __all__ 导出集合（须 ⊇ CONTRACTS） ③ wwwroot/api/*.dspy 实际端点文件集合（须 == CONTRACTS） ④ scripts/load_path.py PATHS 登记集合（须 == CONTRACTS） 并额外检查： · load_path.py 无通配符（规范硬规定：禁 % / *） · 每个 .dspy 显式 return、除 sqlor.filter 外无顶层 import、无 ServerEnv() 取请求态 · 实现层：契约名必须能在 api.py / rules_export_api.py 中找到同名 async def（防幽灵注册） · wwwroot/ 下若出现 CRUD 生成子目录，其 index.ui + get_/add_/update_/delete_*.dspy 必须在 PATHS 中逐条登记（防「生成目录未登记 → 403」复发） 退出码：0 全通过；1 存在 FAIL。交付前必须实跑并在交付摘要引用输出。 """ import ast import os import re import sys HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(HERE) # 模块仓库根 PKG = os.path.join(ROOT, 'pbl_compiler') WWW = os.path.join(ROOT, 'wwwroot') API = os.path.join(WWW, 'api') failures = [] notes = [] def fail(msg): failures.append(msg) def ok(msg): notes.append('PASS ' + msg) def parse(path): """ast 解析：返回 (tree, 字面量常量, 顶层 def 名集合, 顶层赋值目标名)。""" with open(path, 'r', encoding='utf-8') as fh: tree = ast.parse(fh.read(), filename=path) consts, defs, assigned = {}, set(), [] for node in tree.body: if isinstance(node, ast.Assign): for tgt in node.targets: if isinstance(tgt, ast.Name): assigned.append(tgt.id) try: consts[tgt.id] = ast.literal_eval(node.value) except Exception: # noqa: BLE001 - 含函数引用的 dict 非字面量 consts[tgt.id] = node.value # 保留 AST 供专用提取 elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): defs.add(node.name) return tree, consts, defs, assigned def dict_keys(node): """从 dict AST（或已 literal_eval 的 dict）取字符串键集合。 init.py 的 CONTRACTS 形如 {'name': fn_obj}，值是实现函数引用， ast.literal_eval 整体求值会失败——必须只提取键，不能因此判为空。 """ if isinstance(node, dict): return {k for k in node if isinstance(k, str)} if isinstance(node, ast.Dict): return {k.value for k in node.keys if isinstance(k, ast.Constant) and isinstance(k.value, str)} return set() # ── ① init.py CONTRACTS ──────────────────────────────────────────────────── init_py = os.path.join(PKG, 'init.py') CONTRACTS = set() impl_defs = set() if not os.path.isfile(init_py): fail('缺 pbl_compiler/init.py') else: _, consts, _, _ = parse(init_py) if 'CONTRACTS' not in consts: fail('init.py 无 CONTRACTS 常量（三处同步的事实源丢失）') else: CONTRACTS = dict_keys(consts['CONTRACTS']) if not CONTRACTS: fail('init.py CONTRACTS 为空 dict') else: ok('init.py CONTRACTS = %d 个契约' % len(CONTRACTS)) # 实现层：契约必须在 api.py 或 rules_export_api.py 中有同名 async def for impl in ('api.py', 'rules_export_api.py'): p = os.path.join(PKG, impl) if os.path.isfile(p): impl_defs |= parse(p)[2] ghost = sorted(CONTRACTS - impl_defs) if ghost: fail('CONTRACTS 注册了实现层不存在的函数（幽灵注册 → AttributeError）: %s' % ghost) else: ok('%d 个契约在 api.py/rules_export_api.py 均有同名实现' % len(CONTRACTS)) # ── ② __init__.py __all__ ───────────────────────────────────────────────── pkg_init = os.path.join(PKG, '__init__.py') if not os.path.isfile(pkg_init): fail('缺 pbl_compiler/__init__.py') else: _, consts, _, _ = parse(pkg_init) all_names = set(consts.get('__all__') or []) missing = sorted(CONTRACTS - all_names) if missing: fail('__init__.py __all__ 未导出 CONTRACTS 契约（漏 ② → .dspy NameError）: %s' % missing) else: ok('__init__.py __all__ 覆盖全部 %d 个契约' % len(CONTRACTS)) # ── ③ wwwroot/api/*.dspy ────────────────────────────────────────────────── dspy_files = sorted(f for f in os.listdir(API) if f.endswith('.dspy')) \ if os.path.isdir(API) else [] DSPY_NAMES = {f[:-len('.dspy')] for f in dspy_files} if DSPY_NAMES != CONTRACTS: fail('wwwroot/api/ 端点与 CONTRACTS 不一致：缺端点=%s 多余端点=%s' % (sorted(CONTRACTS - DSPY_NAMES), sorted(DSPY_NAMES - CONTRACTS))) else: ok('wwwroot/api/*.dspy = %d 个，与 CONTRACTS 一一对应' % len(DSPY_NAMES)) # ── ④ scripts/load_path.py PATHS ────────────────────────────────────────── lp_py = os.path.join(HERE, 'load_path.py') PATH_NAMES = set() entries = [] if not os.path.isfile(lp_py): fail('缺 scripts/load_path.py') else: _, consts, _, _ = parse(lp_py) raw = consts.get('PATHS') if isinstance(raw, ast.List): raw = [ast.literal_eval(e) for e in raw.elts] if not raw: fail('load_path.py PATHS 为空（RBAC 零登记 → 全部 403）') for item in raw or []: if not (isinstance(item, (list, tuple)) and len(item) == 2): fail('load_path.py PATHS 元素必须是 (path, role) 二元组: %r' % (item,)) continue entries.append((item[0], item[1])) PATH_NAMES = {p.rsplit('/', 1)[-1][:-len('.dspy')] for p, _ in entries if '/api/' in p} if PATH_NAMES != CONTRACTS: fail('load_path.py PATHS 与 CONTRACTS 不一致：未登记=%s 幽灵登记=%s' % (sorted(CONTRACTS - PATH_NAMES), sorted(PATH_NAMES - CONTRACTS))) else: ok('load_path.py PATHS 登记 %d 条，与 CONTRACTS 一一对应' % len(PATH_NAMES)) for p, _role in entries: if '%' in p or '*' in p: fail('load_path.py 禁用通配符: %s' % p) if not any('%' in p or '*' in p for p, _ in entries): ok('load_path.py 无通配符（%d 条全显式）' % len(entries)) n_api = len([1 for p, _ in entries if '/api/' in p]) ok('grep 口径自证：PATHS 中 /api/ 条目=%d，wwwroot/api/*.dspy=%d' % (n_api, len(dspy_files))) if n_api != len(dspy_files): fail('端点数与 RBAC 登记数不等：%d != %d' % (n_api, len(dspy_files))) # ── .dspy 规范审计（dspy-file-implementation-spec）───────────────────────── bad_import, no_return, env_req = [], [], [] for fname in dspy_files: text = open(os.path.join(API, fname), 'r', encoding='utf-8').read() for line in text.splitlines(): s = line.strip() if s.startswith('import ') or s.startswith('from '): if 'sqlor.filter' not in s: bad_import.append('%s: %s' % (fname, s)) if not re.search(r'(?m)^\s*return\b', text): no_return.append(fname) if 'ServerEnv()' in text: env_req.append(fname) if bad_import: fail('.dspy 存在非法顶层 import（应由 load_*() 导出）: %s' % bad_import) else: ok('%d 个 .dspy 无非法 import（仅允许 sqlor.filter）' % len(dspy_files)) if no_return: fail('.dspy 缺显式 return（隐式返回 None → 前端拿到 null）: %s' % no_return) else: ok('%d 个 .dspy 均有显式 return' % len(dspy_files)) if env_req: fail('.dspy 不得用 ServerEnv() 取请求态: %s' % env_req) else: ok('.dspy 无 ServerEnv() 取请求态') # ── CRUD 生成目录登记面（防「生成目录未登记 → 403」）─────────────────────── SKIP_DIRS = {'api', 'i18n', 'imgs', 'styles', 'scripts', 'bricks'} crud_dirs = [d for d in sorted(os.listdir(WWW)) if os.path.isdir(os.path.join(WWW, d)) and d not in SKIP_DIRS] \ if os.path.isdir(WWW) else [] lp_paths = {p for p, _ in entries} for alias in crud_dirs: adir = os.path.join(WWW, alias) for fn in sorted(os.listdir(adir)): if fn == 'index.ui' or re.match(r'^(get_|add_|update_|delete_)', fn): expect = '/pbl_compiler/%s/%s' % (alias, fn) if expect not in lp_paths: fail('CRUD 生成文件未登记 RBAC: %s' % expect) if crud_dirs: ok('发现 CRUD 生成目录 %d 个，已逐文件核对登记' % len(crud_dirs)) else: ok('wwwroot/ 无 CRUD 生成子目录（只读列表走手写契约 api/pbl_script_rule_list.dspy），' '无需登记 CRUD 6 条路径') # ── 输出 ─────────────────────────────────────────────────────────────────── print('=== pbl_compiler 接线/RBAC parity 审计 ===') for line in notes: print(' ' + line) if failures: print('--- FAIL (%d) ---' % len(failures)) for line in failures: print(' ' + line) print('RESULT: FAIL') sys.exit(1) print('RESULT: PASS (contracts=%d, dspy=%d, rbac_api_paths=%d)' % (len(CONTRACTS), len(dspy_files), len(PATH_NAMES))) sys.exit(0)