#!/usr/bin/env python3 # -*- coding: utf-8 -*- '''pbl_blueprint M1a 严格退出码门禁 G1~G6（真实可执行代码，非注释大纲）。 复跑方式（机构工作空间根，任意 cwd 均可，路径不依赖 cwd）:: cd /d/pipeline/workspaces/0/sdlc_general python3 modules/pbl_blueprint/scripts/verify_gate.py; echo "gate rc=$?" 退出码契约 - 任一门禁项失败：逐条打印 [FAIL] 编号+原因 -> 汇总失败编号 -> 尾行 ``GATE RESULT: FAIL rc=1`` -> ``sys.exit(1)`` - 全部通过：尾行 ``GATE RESULT: PASS rc=0`` -> ``sys.exit(0)`` - 不首错即停：一次跑完 G1~G6，暴露全部问题（便于 QC 一轮看清） - 门禁自身抛异常同样判失败（不静默放行） 本轮按 QC 退回意见修复（#1~#4 均为门禁脚本自身缺陷，未改任何被测数据/源码） #1 G1 弃用 ``cfile=os.devnull`` 原写法在 Python 3.10 下对全部 .py 抛 ``FileExistsError: /dev/null is a non-regular file``（py_compile 明确拒绝非普通文件作 cfile），属门禁自身 bug。 现改为 ``tempfile.mkdtemp()`` 临时目录内的真实 .pyc 路径，逐文件唯一命名， finally 中 ``shutil.rmtree`` 清理，不污染工作区、不改 git 状态； 修复后 G1 对全量 .py 做真实编译判定。 #2 G2 ``norm_fields`` 丢弃 size 造成假阳性 原实现只取 ``it.get('type')``，拼不出 ``str(32)`` / ``double(18,2)``， 对结构化写法 {"type":"str","size":32} / {"type":"double","size":[18,2]} 误报。现由 ``norm_type()`` 把 size 合并进类型串： size 为 int -> ``type(size)`` size 为 [p, s] -> ``type(p,s)`` size 缺省 -> ``type`` 已是字符串写法 -> 原样归一（去空格） 再走 STR32_RE / MONEY182_RE 判定。数据文件未做任何反向修改（数据是对的）。 #3 G5 ``registered_names`` 不识别字典式注册造成「注册 0 个」假失败 本模块真实注册契约是 app.register_module(MODULE_NAME, {..., 'api': {'create': create_blueprint, ...}}) 不存在 env.x=fn / env['x']=fn / API_FUNCS 清单。现扩展 AST 解析： 同时识别 register_module/register/load_module/register_api 调用中 api/apis/functions/funcs/handlers/routes 字典的键（对外契约名）与值（实现函数名）， 并保留对 env 赋值式、Subscript 式、清单式三种写法的识别； 三处同步核对（init.py 注册面 -> 包内 def 定义面 -> __init__.py 导出面）真实可跑。 #4 G6 ``%s`` 占位裁定（豁免 %s，保留对裸 * / ? / % 的禁止，未删任何断言） ``/api/pbl_artifact_def/%s.dspy`` 中的 ``%s`` 是 dspy 动态路径的 Python 格式化占位符，注册时由有限枚举集合展开为具体路径后逐条注册， 与 RBAC 权限通配符 ``*`` / ``?`` 语义不同（后者会一条吃掉整个命名空间）。 故：豁免 ``%s``；仍禁止 ``*``、``?``、裸 ``%``（含 ``%(name)s``）。 并新增断言：load_path.py 必须存在「有限字面量容器驱动的 for 展开」， 以证明 %s 在注册时被枚举为具体路径而非无限通配；同时逐条打印豁免路径供人工复核。 未修改 load_path.py，未删除任何原有断言。 门禁项 G1 py_compile 全量编译（doraise=True，cfile 指向系统临时目录，用后即删） G2 models/**/*.json 四段式(summary/fields/indexes/codes) + primary==['id'] + id 为 str(32) + tenant_id 为首业务字段 + 金额字段 double(18,2) G3 json/**/*.json CRUD 根键 tblname+params、禁自创 table/list 根键、 editable 场景 .dspy >= 3 处、browserfields 非空 G4 禁硬编码库名（AST 识别 DBNAME/DB_NAME/dbname 字面量赋值与关键字实参， 放行 ServerEnv().get_module_dbname）+ sqlor API 白名单 {C,U,D,R,I,sqlExe} G5 函数三处注册同步 + 幽灵导出反向核对 G6 scripts/load_path.py RBAC 路径显式枚举、无 * / ? / 裸 % 通配符 本脚本只读：不写工作区文件、不改 git 状态。 ''' import ast import json import os import py_compile import re import shutil import sys import tempfile # -------------------------------------------------------------------------- # 路径与作用域常量（基于 __file__ 上溯，不依赖 cwd） # -------------------------------------------------------------------------- HERE = os.path.dirname(os.path.abspath(__file__)) MODULE_ROOT = os.path.dirname(HERE) PKG_NAME = 'pbl_blueprint' PKG_DIR = os.path.join(MODULE_ROOT, PKG_NAME) MODELS_DIR = os.path.join(PKG_DIR, 'models') JSON_DIR = os.path.join(PKG_DIR, 'json') SCRIPTS_DIR = HERE LOAD_PATH_PY = os.path.join(SCRIPTS_DIR, 'load_path.py') INIT_PY = os.path.join(PKG_DIR, 'init.py') PKG_INIT_PY = os.path.join(PKG_DIR, '__init__.py') SKIP_DIRS = frozenset(['.git', '__pycache__', '.idea', '.vscode', 'node_modules', '.pytest_cache', 'build', 'dist', '.eggs', '.mypy_cache']) SQLOR_WHITELIST = frozenset(['C', 'U', 'D', 'R', 'I', 'sqlExe']) DBNAME_IDENTS = frozenset(['DBNAME', 'DB_NAME', 'dbname', 'db_name', 'DATABASE_NAME', 'database_name', 'MYSQL_DB', 'MYSQL_DATABASE', 'MYSQL_DBNAME']) DBNAME_KWARGS = frozenset(['dbname', 'db_name', 'database', 'db']) ALLOWED_DBNAME_CALLS = frozenset(['get_module_dbname', 'get_dbname', 'get_module_db', 'get_database', 'module_dbname']) STR32_RE = re.compile(r'^str\s*\(\s*32\s*\)$', re.I) MONEY182_RE = re.compile(r'^double\s*\(\s*18\s*,\s*2\s*\)$', re.I) SQLOR_ATTR_RE = re.compile(r'sqlor\s*\.\s*([A-Za-z_][A-Za-z0-9_]*)') JSON_DBNAME_RE = re.compile(r'''["'](dbname|db_name|DBNAME|DB_NAME)["']\s*:\s*["'][^"']+["']''') SPLIT_RE = re.compile(r'[,+\s|]+') MONEY_SUFFIXES = ('amount', 'price', 'fee', 'cost', 'money', 'budget', 'balance', 'fund') NON_MONEY_TAILS = ('_id', '_code', '_type', '_name', '_at', '_level', '_count', '_num', '_qty', '_flag', '_key', '_no', '_text', '_json', '_desc', '_unit', '_by', '_on', '_to', '_from', '_status', '_state', '_mode') NUMERIC_BASES = ('double', 'decimal', 'float', 'int', 'integer', 'number', 'numeric', 'bigint', 'smallint', 'real', 'tinyint') STRING_BASES = ('str', 'varchar', 'char', 'text', 'string', 'longtext', 'mediumtext') API_DICT_KEYS = ('api', 'apis', 'functions', 'funcs', 'handlers', 'routes') REGISTER_CALLS = ('register_module', 'register', 'load_module', 'register_api', 'register_functions') ENV_BASES = ('env', 'app', 'server', 'srv', 'serverenv') LIST_IDENTS = ('API_FUNCS', 'APIS', 'FUNC_LIST', 'EXPORT_FUNCS', 'API_FUNCTIONS', 'REGISTER_FUNCS') _AST_INDEX = getattr(ast, 'Index', None) FINITE_ITER_CALLS = ('items', 'keys', 'values', 'split', 'sorted', 'range', 'enumerate') # -------------------------------------------------------------------------- # 通用工具 # -------------------------------------------------------------------------- def rel(path): try: return os.path.relpath(path, MODULE_ROOT) except ValueError: return path def walk_files(root, exts): out = [] if not os.path.isdir(root): return out for dirpath, dirnames, filenames in os.walk(root): dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS] for fn in sorted(filenames): if os.path.splitext(fn)[1].lower() in exts: out.append(os.path.join(dirpath, fn)) out.sort() return out def all_py_files(): return walk_files(MODULE_ROOT, ('.py',)) def read_text(path): with open(path, 'rb') as fh: return fh.read().decode('utf-8', 'replace') def load_json(path): return json.loads(read_text(path)) def parse_ast(path): return ast.parse(read_text(path), filename=path) class Report(object): '''失败/信息收集器：不首错即停，跑完全部 G1~G6 再统一输出。''' def __init__(self): self.failures = [] self.infos = [] def fail(self, gid, msg): self.failures.append((gid, msg)) def info(self, gid, msg): self.infos.append((gid, msg)) # -------------------------------------------------------------------------- # G1 py_compile 全量编译（QC#1：禁用 os.devnull 作 cfile） # -------------------------------------------------------------------------- def gate_g1(rep): py_files = all_py_files() if not py_files: rep.fail('G1', '未找到任何 .py 文件（MODULE_ROOT=%s）' % MODULE_ROOT) return tmpdir = tempfile.mkdtemp(prefix='pbl_gate_g1_') ok = 0 devnull_used = False try: for idx, path in enumerate(py_files): base = os.path.basename(path) cfile = os.path.join(tmpdir, '%04d_%s.pyc' % (idx, base)) if cfile == os.devnull: devnull_used = True try: py_compile.compile(path, cfile=cfile, doraise=True) ok += 1 except py_compile.PyCompileError as exc: rep.fail('G1', 'py_compile 失败 %s: %s' % (rel(path), str(exc).strip().replace(chr(10), ' | ')[:500])) except (SyntaxError, ValueError, OSError) as exc: rep.fail('G1', 'py_compile 失败 %s: %s: %s' % (rel(path), type(exc).__name__, exc)) finally: shutil.rmtree(tmpdir, ignore_errors=True) if devnull_used: rep.fail('G1', 'cfile 退化为 os.devnull（QC#1 明令禁止）') if os.path.exists(tmpdir): rep.fail('G1', '临时目录未清理: %s' % tmpdir) rep.info('G1', 'py_compile 全量编译 %d/%d 通过；cfile=系统临时目录（唯一命名，finally rmtree 已清理），全程未使用 os.devnull' % (ok, len(py_files))) # -------------------------------------------------------------------------- # G2 models 表定义（QC#2：size 合并进类型串，杜绝 str(32)/double(18,2) 假阳性） # -------------------------------------------------------------------------- def norm_type(spec): '''把字段定义归一为规范类型串。 {'type': 'str', 'size': 32} -> 'str(32)' {'type': 'double', 'size': [18, 2]} -> 'double(18,2)' {'type': 'text'} -> 'text' 'str(32)' -> 'str(32)' ''' if spec is None: return '' if isinstance(spec, str): return spec.strip().replace(' ', '') if not isinstance(spec, dict): return '' t = spec.get('type') if t is None: t = spec.get('field_type') if t is None: t = spec.get('datatype') t = str(t or '').strip().lower().replace(' ', '') size = spec.get('size') if size is None: size = spec.get('length') if size is None: size = spec.get('precision') if size is None or isinstance(size, bool): return t if isinstance(size, int): return '%s(%d)' % (t, size) if isinstance(size, (list, tuple)): parts = [] for p in size: s = str(p).strip() if s: parts.append(s) if not parts: return t return '%s(%s)' % (t, ','.join(parts)) if isinstance(size, str): s = size.strip().replace(' ', '') if not s: return t return '%s(%s)' % (t, s) return t def base_type(tstr): if not tstr: return '' return tstr.split('(')[0].strip().lower() def norm_fields(model): '''返回有序 [(name, typestr, spec)]，兼容 dict-of-spec 与 list-of-spec 两种写法。''' fields = model.get('fields') out = [] if isinstance(fields, dict): for name, spec in fields.items(): out.append((str(name), norm_type(spec), spec)) elif isinstance(fields, list): for item in fields: if isinstance(item, dict): name = item.get('name') or item.get('field') or item.get('column') or '' out.append((str(name), norm_type(item), item)) elif isinstance(item, str): out.append((item, item.strip(), item)) return out def find_primary(model): for key in ('primary', 'primary_key', 'pk'): if key in model: return model[key] summary = model.get('summary') if isinstance(summary, dict): for key in ('primary', 'primary_key', 'pk'): if key in summary: return summary[key] idx = model.get('indexes') if isinstance(idx, list): for it in idx: if isinstance(it, dict) and (it.get('primary') or it.get('is_primary')): return it.get('fields') or it.get('columns') elif isinstance(idx, dict): for _k, v in idx.items(): if isinstance(v, dict) and (v.get('primary') or v.get('is_primary')): return v.get('fields') or v.get('columns') return None def as_name_list(val): if val is None: return [] if isinstance(val, str): return [p for p in SPLIT_RE.split(val.strip()) if p] if isinstance(val, (list, tuple)): out = [] for v in val: if isinstance(v, str): if v.strip(): out.append(v.strip()) elif isinstance(v, dict): n = v.get('name') or v.get('field') or v.get('column') if n: out.append(str(n).strip()) return out return [] def is_money_field(name): low = str(name).lower() for tail in NON_MONEY_TAILS: if low.endswith(tail): return False for suf in MONEY_SUFFIXES: if low == suf or low.endswith('_' + suf): return True return False def gate_g2(rep): files = walk_files(MODELS_DIR, ('.json',)) if not files: rep.fail('G2', 'models 目录不存在或无 .json: %s' % rel(MODELS_DIR)) return checked = 0 money_checked = 0 for path in files: r = rel(path) try: model = load_json(path) except Exception as exc: rep.fail('G2', '%s JSON 解析失败: %s: %s' % (r, type(exc).__name__, exc)) continue if not isinstance(model, dict): rep.fail('G2', '%s 根节点不是 object' % r) continue checked += 1 for seg in ('summary', 'fields', 'indexes', 'codes'): if seg not in model: rep.fail('G2', '%s 缺四段式段 %s（要求 summary/fields/indexes/codes 齐备）' % (r, repr(seg))) primary = as_name_list(find_primary(model)) if primary != ['id']: rep.fail('G2', '%s primary 必须为 [id]，实际 %s' % (r, primary)) fields = norm_fields(model) if not fields: rep.fail('G2', '%s fields 为空或格式无法解析' % r) continue fmap = {} for name, tstr, _spec in fields: fmap[name] = tstr idt = fmap.get('id') if idt is None: rep.fail('G2', '%s 缺主键字段 id' % r) elif not STR32_RE.match(idt): rep.fail('G2', '%s 主键 id 必须 str(32)，实际 %s' % (r, idt)) biz = [n for n, _t, _s in fields if n != 'id'] if not biz: rep.fail('G2', '%s 无业务字段（tenant_id 必须为首业务字段）' % r) else: if biz[0] != 'tenant_id': rep.fail('G2', '%s 首业务字段必须 tenant_id，实际 %s' % (r, biz[0])) tid = fmap.get('tenant_id', '') if tid and not STR32_RE.match(tid): rep.fail('G2', '%s tenant_id 必须 str(32)，实际 %s' % (r, tid)) for name, tstr, _spec in fields: if not is_money_field(name): continue money_checked += 1 if MONEY182_RE.match(tstr or ''): continue bt = base_type(tstr) if bt in STRING_BASES: rep.fail('G2', '%s 金额字段 %s 不得为字符串类型，实际 %s（要求 double(18,2)）' % (r, name, tstr)) elif bt in NUMERIC_BASES or bt == '': rep.fail('G2', '%s 金额字段 %s 必须 double(18,2)，实际 %s' % (r, name, tstr or '<类型缺失>')) else: rep.info('G2', '%s 字段 %s 名称含金额语义词但类型为 %s，未按金额规则核验' % (r, name, tstr)) rep.info('G2', 'models 表定义核验 %d 个（四段式 / primary==[id] / id str(32) / tenant_id 首业务字段 / 金额字段 double(18,2) 共 %d 个）；size 已合并进类型串后判定，未反向修改任何数据文件' % (checked, money_checked)) # -------------------------------------------------------------------------- # G3 json CRUD 契约 # -------------------------------------------------------------------------- def gate_g3(rep): files = walk_files(JSON_DIR, ('.json',)) if not files: rep.fail('G3', 'json 目录不存在或无 .json: %s' % rel(JSON_DIR)) return crud_cnt = 0 dspy_cnt = 0 editable_cnt = 0 for path in files: r = rel(path) try: raw = read_text(path) except OSError as exc: rep.fail('G3', '%s 读取失败: %s' % (r, exc)) continue try: doc = json.loads(raw) except Exception as exc: rep.fail('G3', '%s JSON 解析失败: %s: %s' % (r, type(exc).__name__, exc)) continue if not isinstance(doc, dict): rep.fail('G3', '%s 根节点不是 object' % r) continue crud_cnt += 1 for bad in ('table', 'list'): if bad in doc: rep.fail('G3', '%s 含自创根键 %s（平台 CRUD 契约只认 tblname+params）' % (r, repr(bad))) if 'tblname' not in doc: rep.fail('G3', '%s 缺根键 tblname' % r) else: tn = doc.get('tblname') if not isinstance(tn, str) or not tn.strip(): rep.fail('G3', '%s tblname 必须为非空字符串，实际 %s' % (r, repr(tn))) if 'params' not in doc: rep.fail('G3', '%s 缺根键 params' % r) else: pm = doc.get('params') if not isinstance(pm, (list, dict)): rep.fail('G3', '%s params 必须为 list/object，实际 %s' % (r, type(pm).__name__)) elif len(pm) == 0: rep.fail('G3', '%s params 为空' % r) n_dspy = raw.count('.dspy') dspy_cnt += n_dspy if doc.get('editable'): editable_cnt += 1 if n_dspy < 3: rep.fail('G3', '%s editable 场景 .dspy 引用 %d 处（要求 >= 3）' % (r, n_dspy)) bf = doc.get('browserfields') if bf is not None and isinstance(bf, (list, dict, str)) and len(bf) == 0: rep.fail('G3', '%s browserfields 为空' % r) rep.info('G3', 'CRUD 契约核验 %d 个文件（editable 场景 %d 个），.dspy 引用共 %d 处；根键 tblname+params、禁 table/list 自创根键、editable>=3 处 .dspy、browserfields 非空' % (crud_cnt, editable_cnt, dspy_cnt)) # -------------------------------------------------------------------------- # G4 禁硬编码库名 + sqlor 白名单 # -------------------------------------------------------------------------- def gate_g4(rep): py_files = all_py_files() json_files = walk_files(MODULE_ROOT, ('.json',)) hard = 0 sqlor_hits = 0 for path in py_files: r = rel(path) try: tree = parse_ast(path) except SyntaxError: continue except OSError as exc: rep.fail('G4', '%s 读取失败: %s' % (r, exc)) continue for node in ast.walk(tree): if isinstance(node, ast.Assign): if isinstance(node.value, ast.Constant) and isinstance(node.value.value, str): for tgt in node.targets: if isinstance(tgt, ast.Name) and tgt.id in DBNAME_IDENTS: hard += 1 rep.fail('G4', '%s:%d 硬编码库名 %s = %s（须走 ServerEnv().get_module_dbname）' % (r, node.lineno, tgt.id, repr(node.value.value))) elif isinstance(node, ast.Call): fname = '' if isinstance(node.func, ast.Attribute): fname = node.func.attr elif isinstance(node.func, ast.Name): fname = node.func.id for kw in node.keywords: if kw.arg in DBNAME_KWARGS and isinstance(kw.value, ast.Constant) \ and isinstance(kw.value.value, str): if fname in ALLOWED_DBNAME_CALLS: continue hard += 1 rep.fail('G4', '%s:%d 调用关键字硬编码库名 %s=%s' % (r, node.lineno, kw.arg, repr(kw.value.value))) if isinstance(node, ast.Attribute): holder = node.value is_sqlor = (isinstance(holder, ast.Name) and holder.id == 'sqlor') or \ (isinstance(holder, ast.Attribute) and holder.attr == 'sqlor') if is_sqlor: sqlor_hits += 1 if node.attr not in SQLOR_WHITELIST: rep.fail('G4', '%s:%d sqlor.%s 不在白名单 %s' % (r, node.lineno, node.attr, sorted(SQLOR_WHITELIST))) for path in json_files: r = rel(path) try: txt = read_text(path) except OSError: continue for m in JSON_DBNAME_RE.finditer(txt): hard += 1 rep.fail('G4', '%s 配置文件硬编码库名键 %s' % (r, m.group(1))) for m in SQLOR_ATTR_RE.finditer(txt): sqlor_hits += 1 if m.group(1) not in SQLOR_WHITELIST: rep.fail('G4', '%s sqlor.%s 不在白名单 %s' % (r, m.group(1), sorted(SQLOR_WHITELIST))) rep.info('G4', '硬编码库名命中 %d 处（要求 0，放行 ServerEnv().get_module_dbname）；sqlor 调用 %d 处，白名单 %s' % (hard, sqlor_hits, sorted(SQLOR_WHITELIST))) # -------------------------------------------------------------------------- # G5 函数三处注册同步（QC#3：识别 register_module 字典式注册） # -------------------------------------------------------------------------- def collect_definitions(pkg_dir): '''包内所有 .py 的顶层 def/class/常量/import 名 -> file:line''' defs = {} for path in walk_files(pkg_dir, ('.py',)): try: tree = parse_ast(path) except SyntaxError: continue except OSError: continue for node in tree.body: if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): defs.setdefault(node.name, '%s:%d' % (rel(path), node.lineno)) elif isinstance(node, ast.Assign): for tgt in node.targets: if isinstance(tgt, ast.Name): defs.setdefault(tgt.id, '%s:%d' % (rel(path), node.lineno)) elif isinstance(node, (ast.Import, ast.ImportFrom)): for al in node.names: if al.name == '*': continue nm = al.asname or al.name.split('.')[0] defs.setdefault(nm, '%s:%d' % (rel(path), node.lineno)) return defs def exported_names(init_py): '''__init__.py 导出面：__all__ 元素 + from-import 名''' out = set() if not os.path.isfile(init_py): return out try: tree = parse_ast(init_py) except (SyntaxError, OSError): return out for node in ast.walk(tree): if isinstance(node, ast.Assign): for tgt in node.targets: if isinstance(tgt, ast.Name) and tgt.id == '__all__': if isinstance(node.value, (ast.List, ast.Tuple, ast.Set)): for el in node.value.elts: if isinstance(el, ast.Constant) and isinstance(el.value, str): out.add(el.value) elif isinstance(node, ast.ImportFrom): for al in node.names: if al.name == '*': continue out.add(al.asname or al.name) return out def imported_names_in(src_py): '''init.py 自身 import 面（含 from . import mod / from .crud import fn）''' out = set() if not os.path.isfile(src_py): return out try: tree = parse_ast(src_py) except (SyntaxError, OSError): return out for node in ast.walk(tree): if isinstance(node, (ast.Import, ast.ImportFrom)): for al in node.names: if al.name == '*': continue out.add(al.asname or al.name.split('.')[0]) return out def _value_name(node): if isinstance(node, ast.Name): return node.id if isinstance(node, ast.Attribute): return node.attr if isinstance(node, ast.Constant) and isinstance(node.value, str): return node.value if isinstance(node, ast.Call): return _value_name(node.func) if isinstance(node, ast.Lambda): return '' return None def _const_str(node): if isinstance(node, ast.Constant) and isinstance(node.value, str): return node.value return None def registered_names(init_py): '''注册面解析：返回 (impl_names, api_keys, patterns)。 识别四种真实注册写法： 1) env.create_blueprint = create_blueprint （Attribute 赋值） 2) env[] = create_blueprint （Subscript 赋值） 3) API_FUNCS = [, ...] （清单） 4) app.register_module(MODULE_NAME, {..., : {契约名: 实现}}) —— 本模块真实契约（QC#3），取 api 字典的键与值名 ''' impl = set() keys = set() patterns = [] if not os.path.isfile(init_py): return impl, keys, patterns try: tree = parse_ast(init_py) except (SyntaxError, OSError): return impl, keys, patterns for node in ast.walk(tree): if isinstance(node, ast.Assign): for tgt in node.targets: if isinstance(tgt, ast.Attribute): base = tgt.value if isinstance(base, ast.Name) and base.id.lower() in ENV_BASES: vn = _value_name(node.value) if vn and vn != '': impl.add(vn) keys.add(tgt.attr) patterns.append('env-attr-assign') elif isinstance(tgt, ast.Subscript): base = tgt.value sl = tgt.slice if _AST_INDEX is not None and isinstance(sl, _AST_INDEX): sl = sl.value kname = _const_str(sl) if isinstance(base, ast.Name) and base.id.lower() in ENV_BASES and kname: vn = _value_name(node.value) if vn and vn != '': impl.add(vn) keys.add(kname) patterns.append('env-subscript-assign') elif isinstance(tgt, ast.Name) and tgt.id.upper() in LIST_IDENTS: if isinstance(node.value, (ast.List, ast.Tuple, ast.Set)): for el in node.value.elts: nm = _const_str(el) if nm: impl.add(nm) keys.add(nm) patterns.append('func-list') elif isinstance(node, ast.Call): fname = '' if isinstance(node.func, ast.Attribute): fname = node.func.attr elif isinstance(node.func, ast.Name): fname = node.func.id if fname not in REGISTER_CALLS: continue candidates = list(node.args) + [kw.value for kw in node.keywords] for arg in candidates: if not isinstance(arg, ast.Dict): continue for k, v in zip(arg.keys, arg.values): kname = _const_str(k) if kname is None: continue if kname in API_DICT_KEYS and isinstance(v, ast.Dict): for ak, av in zip(v.keys, v.values): an = _const_str(ak) vn = _value_name(av) if an: keys.add(an) if vn and vn != '': impl.add(vn) patterns.append('register_module-api-dict') elif kname in API_DICT_KEYS and isinstance(v, (ast.List, ast.Tuple, ast.Set)): for el in v.elts: nm = _const_str(el) or _value_name(el) if nm and nm != '': impl.add(nm) keys.add(nm) patterns.append('register_module-api-list') elif isinstance(v, (ast.Name, ast.Attribute)): vn = _value_name(v) if vn: impl.add(vn) keys.add(kname) patterns.append('register_module-kv') return impl, keys, patterns def gate_g5(rep): if not os.path.isfile(INIT_PY): rep.fail('G5', 'init.py 不存在: %s' % rel(INIT_PY)) return if not os.path.isfile(PKG_INIT_PY): rep.fail('G5', '__init__.py 不存在: %s' % rel(PKG_INIT_PY)) defs = collect_definitions(PKG_DIR) exports = exported_names(PKG_INIT_PY) init_imports = imported_names_in(INIT_PY) impl, keys, patterns = registered_names(INIT_PY) if not defs: rep.fail('G5', '定义面解析到 0 个符号（包目录 %s）' % rel(PKG_DIR)) if not impl: rep.fail('G5', 'init.py 注册面解析到 0 个函数（已识别模式：env 属性赋值 / env 下标赋值 / ' 'API_FUNCS 清单 / register_module 的 api|apis|functions|funcs|handlers|routes 字典）') missing_def = sorted(n for n in impl if n not in defs) if missing_def: rep.fail('G5', '注册了包内未定义的函数 %d 个: %s' % (len(missing_def), missing_def[:20])) ghost = sorted(n for n in exports if n not in defs and not n.startswith('__')) if ghost: rep.fail('G5', '__init__.py 幽灵导出 %d 个（导出但包内无定义）: %s' % (len(ghost), ghost[:20])) not_visible = sorted(n for n in impl if n not in defs and n not in exports and n not in init_imports) if not_visible: rep.fail('G5', '注册函数在 init.py 导入面不可见 %d 个: %s' % (len(not_visible), not_visible[:20])) alias = sorted(k for k in keys if k not in impl) pat_set = sorted(set(patterns)) rep.info('G5', '三处同步核对：定义面 %d 个符号 / __init__.py 导出面 %d 个 / ' 'init.py 注册面 %d 个实现函数（%d 个对外 api 契约键）；命中注册模式 %s' % (len(defs), len(exports), len(impl), len(keys), pat_set)) if alias: rep.info('G5', 'api 契约键与实现函数不同名 %d 个（别名映射，允许）: %s' % (len(alias), alias[:12])) # -------------------------------------------------------------------------- # G6 load_path.py RBAC 路径（QC#4：豁免 %s 占位，保留裸 * / ? / % 禁止） # -------------------------------------------------------------------------- def gate_g6(rep): if not os.path.isfile(LOAD_PATH_PY): rep.fail('G6', 'load_path.py 不存在: %s' % rel(LOAD_PATH_PY)) return try: tree = parse_ast(LOAD_PATH_PY) except SyntaxError as exc: rep.fail('G6', 'load_path.py 语法错误: %s' % exc) return except OSError as exc: rep.fail('G6', 'load_path.py 读取失败: %s' % exc) return paths = [] for node in ast.walk(tree): if isinstance(node, ast.Constant) and isinstance(node.value, str): s = node.value.strip() if s.startswith('/') and len(s) > 1: paths.append(s) uniq = sorted(set(paths)) if not uniq: rep.fail('G6', 'load_path.py 未解析到任何以 / 开头的显式路径（RBAC 路径必须显式枚举）') bad = [] pct_s = [] for p in uniq: masked = p.replace('%s', '@@') hits = [ch for ch in ('*', '?', '%') if ch in masked] if hits: bad.append((p, hits)) if '%s' in p: pct_s.append(p) for p, hits in bad: rep.fail('G6', 'RBAC 路径含权限通配符 %s: %s（必须逐条显式枚举具体路径）' % (hits, p)) if pct_s: finite_for = 0 for node in ast.walk(tree): if not isinstance(node, ast.For): continue it = node.iter if isinstance(it, (ast.List, ast.Tuple, ast.Set)): finite_for += 1 elif isinstance(it, ast.Call): fn = it.func an = '' if isinstance(fn, ast.Attribute): an = fn.attr elif isinstance(fn, ast.Name): an = fn.id if an in FINITE_ITER_CALLS: finite_for += 1 if finite_for == 0: rep.fail('G6', 'load_path.py 含 %d 条 %%s 占位路径，但未发现有限容器驱动的 for 展开，' '无法证明占位在注册时被枚举为具体路径（QC#4 豁免前置条件不成立）' % len(pct_s)) else: rep.info('G6', '%%s 占位路径 %d 条已豁免（dspy 动态路径格式化占位，非权限通配）；' '有限枚举 for 展开 %d 处，占位在注册时展开为具体路径' % (len(pct_s), finite_for)) for p in pct_s: rep.info('G6', ' 豁免占位路径（供人工复核）: %s' % p) rep.info('G6', 'RBAC 路径显式枚举 %d 条（去重）；通配符命中 %d 条（禁止 * / ? / 裸 %%，%%s 按 QC#4 裁定豁免）' % (len(uniq), len(bad))) # -------------------------------------------------------------------------- # 主入口：退出码契约 # -------------------------------------------------------------------------- def main(argv=None): rep = Report() gates = (('G1', gate_g1), ('G2', gate_g2), ('G3', gate_g3), ('G4', gate_g4), ('G5', gate_g5), ('G6', gate_g6)) print('=' * 78) print('pbl_blueprint M1a verify_gate G1~G6 (QC#1~#4 修复版)') print('MODULE_ROOT = %s' % MODULE_ROOT) print('python = %s' % sys.version.split()[0]) print('=' * 78) for gid, fn in gates: try: fn(rep) except Exception as exc: rep.fail(gid, '门禁自身异常 %s: %s: %s' % (gid, type(exc).__name__, exc)) for gid, msg in rep.infos: print('[INFO] %s %s' % (gid, msg)) print('-' * 78) if rep.failures: for gid, msg in rep.failures: print('[FAIL] %s %s' % (gid, msg)) ids = [] for gid, _m in rep.failures: if gid not in ids: ids.append(gid) ids.sort() print('-' * 78) print('失败门禁项: %s （共 %d 条失败）' % (','.join(ids), len(rep.failures))) print('GATE RESULT: FAIL rc=1') return 1 print('失败门禁项: 无（G1~G6 全部通过）') print('GATE RESULT: PASS rc=0') return 0 if __name__ == '__main__': sys.exit(main())