diff --git a/scripts/verify_gate.py b/scripts/verify_gate.py index fce7b66..3a6492b 100644 --- a/scripts/verify_gate.py +++ b/scripts/verify_gate.py @@ -1,10 +1,667 @@ -# 已用 write_file 落盘（17141 字符）。pbl_blueprint M1a 严格退出码门禁 G1~G6： -# G1 py_compile 全量编译（doraise=True, cfile=os.devnull 不污染工作区） -# G2 models/*.json 四段式(summary/fields/indexes/codes) + primary==["id"] + id str32 + tenant_id 首业务字段 + 金额 double(18,2) -# G3 json/*.json 根键 tblname+params + 禁自创 table/list + .dspy>=3处 + browserfields 非空 -# G4 禁硬编码库名(DBNAME=/DB_NAME=/dbname='…'，放行 get_module_dbname，跳过注释行) + sqlor 白名单{C,U,D,R,I,sqlExe}(同扫 .py 与 .json) -# G5 函数三处注册同步(init.py env 注册 -> 包内 def -> __init__.py 导出，含 NOISE 噪声过滤防假阳性) -# G6 scripts/load_path.py RBAC 路径显式枚举、无 %/* 通配符、每条以 / 开头 -# 退出码契约：任一段失败 -> 汇总编号打印 -> 尾行 'GATE RESULT: FAIL rc=1' -> sys.exit(1)；全通过 -> 'GATE RESULT: PASS rc=0' -> sys.exit(0) -# 路径解析基于 os.path.dirname(os.path.abspath(__file__)) 上溯，不依赖 cwd；失败不首错即停，一次跑完暴露全部问题。 -# 复跑：cd /d/pipeline/workspaces/0/sdlc_general && python3 modules/pbl_blueprint/scripts/verify_gate.py; echo "gate rc=$?" \ No newline at end of file +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""pbl_blueprint M1a 严格退出码门禁 G1~G6（真实可执行代码，非注释大纲）。 + +复跑方式（机构工作空间根）:: + + cd /d/pipeline/workspaces/0/sdlc_general + python3 modules/pbl_blueprint/scripts/verify_gate.py; echo "gate rc=$?" + +退出码契约 + - 任一门禁项失败：逐条打印失败原因 -> 汇总失败编号 -> 尾行 ``GATE RESULT: FAIL rc=1`` -> ``sys.exit(1)`` + - 全部通过：尾行 ``GATE RESULT: PASS rc=0`` -> ``sys.exit(0)`` + - 不首错即停：一次跑完 G1~G6，暴露全部问题（便于 QC 一轮看清） + +门禁项 + G1 py_compile 全量编译（doraise=True，cfile=os.devnull，不污染工作区） + 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}（.py 走 AST，.json 走正则） + G5 函数三处注册同步：init.py 的 env 注册 -> 包内 def 定义 -> __init__.py 导出； + 反向核对幽灵导出（__all__ 声明但包内不存在） + G6 scripts/load_path.py RBAC 路径显式枚举、无 %/*/? 通配符 + +路径解析基于 ``os.path.dirname(os.path.abspath(__file__))`` 上溯，不依赖 cwd。 +本脚本只读，不写任何文件、不改 git 状态。 +""" + +import ast +import json +import os +import py_compile +import re +import sys + +# -------------------------------------------------------------------------- +# 路径与作用域常量 +# -------------------------------------------------------------------------- +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") +LOAD_PATH_PY = os.path.join(HERE, "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"]) + +SQLOR_WHITELIST = frozenset(["C", "U", "D", "R", "I", "sqlExe"]) +SQLOR_IGNORE = frozenset(["Sqlor", "sqlor", "__init__", "__all__", "py"]) + +FOUR_SECTIONS = ("summary", "fields", "indexes", "codes") + +MONEY_TOKENS = ("amount", "price", "fee", "cost", "money", "budget", "balance", "payment") +NON_MONEY_SUFFIX = ( + "_type", "_id", "_code", "_flag", "_name", "_at", "_by", "_no", + "_key", "_status", "_unit", "_desc", "_remark", "_json", "_str", +) +NUMERIC_TYPE_RE = re.compile(r"^(double|decimal|numeric|float|number|real|bigint|int|integer)\b", re.I) +STR32_RE = re.compile(r"^(str\s*\(\s*32\s*\)|str32|varchar\s*\(\s*32\s*\)|char\s*\(\s*32\s*\)|string\s*\(\s*32\s*\))$", re.I) +MONEY182_RE = re.compile(r"^(double|decimal|numeric)\s*\(\s*18\s*,\s*2\s*\)$", re.I) +DBNAME_TARGET_RE = re.compile(r"^(DBNAME|DB_NAME|DATABASE_NAME|dbname|db_name|database_name)$") +IDENT_RE = re.compile(r"^[a-z][a-z0-9_]{2,}$") +SQLOR_TEXT_RE = re.compile(r"\b(?:sor|sqlor)\.([A-Za-z_]\w*)") +PATH_LITERAL_RE = re.compile(r"""['"](/[A-Za-z0-9_./\-]+)['"]""") + +# G5 噪声过滤：这些名字即使出现在 env 注册解析结果里也不作为「对外契约函数」核对， +# 避免把模块对象/配置项/框架钩子误判为缺 def、缺导出（防假阳性）。 +NOISE = frozenset([ + "get_module_dbname", "load_pbl_blueprint", "pbl_blueprint", "init", "main", "setup", + "install", "register", "env", "server", "app", "conf", "config", "settings", "options", + "params", "context", "ctx", "user", "session", "request", "response", "db", "dbname", + "sqlor", "sor", "i18n", "lang", "static", "wwwroot", "menu", "routes", "apis", "tables", + "models", "module", "name", "path", "root", "log", "logger", "audit", "errors", "crud", + "service", "api", "tenant", "subobject", "subobjects", "templates", "tables_def", + "pool", "conn", "cursor", "transaction", "lock", "cache", "hooks", "handlers", + "middleware", "blueprint_crud", "api_blueprint", "db_adapter", "sql", "json", +]) + +FAILURES = [] # [(gate, message)] +WARNS = [] # [(gate, message)] +STATS = {} + + +# -------------------------------------------------------------------------- +# 通用工具 +# -------------------------------------------------------------------------- +def rel(path): + """转成相对模块仓库根的可读路径（失败则原样返回）。""" + try: + return os.path.relpath(path, MODULE_ROOT) + except ValueError: + return path + + +def fail(gate, message): + FAILURES.append((gate, message)) + + +def warn(gate, message): + WARNS.append((gate, message)) + + +def iter_files(root, suffix): + """递归产出 root 下指定后缀的文件绝对路径，跳过 .git/__pycache__ 等目录。""" + if not os.path.isdir(root): + return + 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 fn.endswith(suffix): + yield os.path.join(dirpath, fn) + + +def read_text(path): + try: + with open(path, "r", encoding="utf-8") as f: + return f.read() + except (IOError, OSError, UnicodeDecodeError) as e: + fail("G0", "文件读取失败 %s: %s" % (rel(path), e)) + return None + + +def parse_ast(path): + """返回 (tree_or_None, source_text)。语法错误由 G1 负责报错，此处不重复失败。""" + src = read_text(path) + if src is None: + return None, "" + try: + return ast.parse(src, filename=path), src + except SyntaxError: + return None, src + + +def load_json(path): + src = read_text(path) + if src is None: + return None + try: + return json.loads(src) + except ValueError as e: + fail("G0", "JSON 解析失败 %s: %s" % (rel(path), e)) + return None + + +def const_str(node): + """兼容 py3.7 的 ast.Str 与 py3.8+ 的 ast.Constant 取字符串值。""" + if isinstance(node, ast.Constant) and isinstance(node.value, str): + return node.value + if hasattr(ast, "Str") and isinstance(node, ast.Str): # pragma: no cover + return node.s + return None + + +def subscript_key(node): + """兼容 py3.8 的 ast.Index 包装，取 Subscript 的键节点。""" + sl = node.slice + if hasattr(ast, "Index") and isinstance(sl, ast.Index): # pragma: no cover + return sl.value + return sl + + +def find_key_values(obj, target_lower): + """在任意嵌套的 dict/list 里递归收集键名（小写比对）等于 target_lower 的值。""" + found = [] + if isinstance(obj, dict): + for k, v in obj.items(): + if str(k).lower() == target_lower: + found.append(v) + found.extend(find_key_values(v, target_lower)) + elif isinstance(obj, list): + for it in obj: + found.extend(find_key_values(it, target_lower)) + return found + + +def is_nonempty(v): + if v is None: + return False + if isinstance(v, (list, dict, str, tuple, set)): + return len(v) > 0 + return True + + +# -------------------------------------------------------------------------- +# G1 py_compile 全量编译 +# -------------------------------------------------------------------------- +def gate_g1(): + pys = sorted(iter_files(MODULE_ROOT, ".py")) + if not pys: + fail("G1", "模块仓库内未找到任何 .py 文件：%s" % MODULE_ROOT) + STATS["G1"] = {"total": 0, "bad": 0} + return + bad = 0 + for p in pys: + try: + py_compile.compile(p, cfile=os.devnull, doraise=True) + except py_compile.PyCompileError as e: + bad += 1 + detail = str(e).strip().splitlines() + fail("G1", "编译失败 %s: %s" % (rel(p), detail[0] if detail else e)) + except SyntaxError as e: + bad += 1 + fail("G1", "语法错误 %s: line %s: %s" % (rel(p), e.lineno, e.msg)) + except Exception as e: # noqa: BLE001 —— 门禁要把任何编译异常都暴露出来 + bad += 1 + fail("G1", "编译异常 %s: %s: %s" % (rel(p), type(e).__name__, e)) + STATS["G1"] = {"total": len(pys), "bad": bad} + print(" G1 py_compile 全量编译 %d 个 .py，失败 %d 个" % (len(pys), bad)) + + +# -------------------------------------------------------------------------- +# G2 models/*.json 表定义四段式与硬约束 +# -------------------------------------------------------------------------- +def norm_fields(obj): + """把 fields 段归一化为 [(name, typestr, meta_dict)]，兼容 list[dict] / dict 两种写法。""" + fields = obj.get("fields") + out = [] + if isinstance(fields, list): + for it in fields: + if isinstance(it, dict): + nm = it.get("name") or it.get("field") or it.get("column") or it.get("colname") or it.get("key") + tp = it.get("type") or it.get("datatype") or it.get("dbtype") or it.get("coltype") or "" + if nm is None: + continue + out.append((str(nm), str(tp), it)) + elif isinstance(it, str): + out.append((it, "", {})) + elif isinstance(fields, dict): + for k, v in fields.items(): + if isinstance(v, dict): + tp = v.get("type") or v.get("datatype") or v.get("dbtype") or "" + out.append((str(k), str(tp), v)) + else: + out.append((str(k), "" if v is None else str(v), {})) + return out + + +def find_primary(obj): + """在顶层 / summary / indexes 三处发现主键声明，返回 list[str] 或 None。""" + keys = ("primary", "primarykey", "primary_key", "pk") + for k in keys: + if k in obj: + v = obj[k] + return [str(x)] if isinstance(v, str) else [str(x) for x in v] if isinstance(v, (list, tuple)) else None + summary = obj.get("summary") + if isinstance(summary, dict): + for k in keys: + if k in summary: + v = summary[k] + return [str(x)] if isinstance(v, str) else [str(x) for x in v] if isinstance(v, (list, tuple)) else None + idx = obj.get("indexes") + entries = idx if isinstance(idx, list) else (list(idx.values()) if isinstance(idx, dict) else []) + for e in entries: + if isinstance(e, dict) and (e.get("primary") is True or str(e.get("type", "")).lower() in ("primary", "pk")): + cols = e.get("fields") or e.get("columns") or e.get("cols") or e.get("name") + if isinstance(cols, str): + return [cols] + if isinstance(cols, (list, tuple)): + return [str(c) for c in cols] + return None + + +def gate_g2(): + files = sorted(iter_files(MODELS_DIR, ".json")) + if not files: + fail("G2", "models 目录不存在或无表定义 JSON：%s" % rel(MODELS_DIR)) + STATS["G2"] = {"tables": 0} + return + money_checked = 0 + for p in files: + obj = load_json(p) + if obj is None: + continue + r = rel(p) + if not isinstance(obj, dict): + fail("G2", "%s 根节点不是 JSON 对象" % r) + continue + # (1) 四段式 + missing = [s for s in FOUR_SECTIONS if s not in obj] + if missing: + fail("G2", "%s 缺四段式字段段：%s" % (r, "/".join(missing))) + fields = norm_fields(obj) + if not fields: + fail("G2", "%s fields 段为空或格式不可识别" % r) + continue + names = [n for n, _t, _m in fields] + types = dict((n, t) for n, t, _m in fields) + # (2) 主键 primary == ["id"] + pk = find_primary(obj) + if pk is None: + if "id" not in names: + fail("G2", "%s 未声明 primary 且 fields 中无 id 字段" % r) + else: + warn("G2", "%s 未显式声明 primary=[\"id\"]（已按 fields 含 id 放行，建议补声明）" % r) + elif pk != ["id"]: + fail("G2", "%s primary 必须为 [\"id\"]，实际 %s" % (r, pk)) + # (3) id 为 str(32) + if "id" in names: + idt = (types.get("id") or "").strip() + if not STR32_RE.match(idt): + fail("G2", "%s 主键 id 类型必须为 str(32)，实际 %r" % (r, idt)) + if names[0] != "id": + warn("G2", "%s id 未置于 fields 首位（实际首位 %s）" % (r, names[0])) + # (4) tenant_id 为首业务字段（fail-closed 多租户硬约束） + if "tenant_id" not in names: + fail("G2", "%s 缺 tenant_id 字段（所有读写租户强制打头）" % r) + else: + biz = names[1:] if names and names[0] == "id" else names + if not biz or biz[0] != "tenant_id": + fail("G2", "%s tenant_id 必须是首业务字段，实际首业务字段为 %s" % (r, biz[0] if biz else "无")) + # (5) 金额字段 double(18,2) + for n, t, _m in fields: + low = n.lower() + if not any(tok in low for tok in MONEY_TOKENS): + continue + if low.endswith(NON_MONEY_SUFFIX): + continue + t = (t or "").strip() + if not NUMERIC_TYPE_RE.match(t): + continue # 非数值型（如 fee_type 之类编码字段）不按金额约束 + money_checked += 1 + if not MONEY182_RE.match(t): + fail("G2", "%s 金额字段 %s 必须为 double(18,2)，实际 %r" % (r, n, t)) + STATS["G2"] = {"tables": len(files), "money_fields_checked": money_checked} + print(" G2 表定义 %d 个（四段式/primary/id str32/tenant_id 首业务字段/金额 double(18,2)），金额字段核对 %d 个" + % (len(files), money_checked)) + + +# -------------------------------------------------------------------------- +# G3 json/*.json CRUD 契约 +# -------------------------------------------------------------------------- +def gate_g3(): + files = sorted(iter_files(JSON_DIR, ".json")) + if not files: + fail("G3", "json 目录不存在或无 CRUD 定义：%s" % rel(JSON_DIR)) + STATS["G3"] = {"crud": 0} + return + forbid_root = ("table", "list") + dspy_total = 0 + for p in files: + raw = read_text(p) + obj = load_json(p) + if obj is None: + continue + r = rel(p) + if not isinstance(obj, dict): + fail("G3", "%s 根节点不是 JSON 对象" % r) + continue + # (1) 根键 tblname + params + if "tblname" not in obj: + fail("G3", "%s 缺根键 tblname" % r) + elif not is_nonempty(obj.get("tblname")): + fail("G3", "%s 根键 tblname 为空" % r) + if "params" not in obj: + fail("G3", "%s 缺根键 params" % r) + # (2) 禁自创 table/list 根键格式 + for bad_key in forbid_root: + if bad_key in obj: + fail("G3", "%s 含自创根键 %r（平台 CRUD 契约只认 tblname+params）" % (r, bad_key)) + # (3) .dspy 引用：含 editable 的必须 >= 3 处（增/改/删或列表/编辑/删除三入口） + n_dspy = (raw or "").count(".dspy") + dspy_total += n_dspy + has_editable = len(find_key_values(obj, "editable")) > 0 + if has_editable: + if n_dspy < 3: + fail("G3", "%s 含 editable 但 .dspy 引用仅 %d 处（要求 >= 3）" % (r, n_dspy)) + elif n_dspy < 1: + warn("G3", "%s 无 editable 且无 .dspy 引用（若为纯查询契约可忽略）" % r) + # (4) browserfields 非空 + bfs = find_key_values(obj, "browserfields") + if not bfs: + warn("G3", "%s 未声明 browserfields（列表/浏览型契约应声明）" % r) + elif not any(is_nonempty(v) for v in bfs): + fail("G3", "%s browserfields 已声明但为空" % r) + STATS["G3"] = {"crud": len(files), "dspy_refs": dspy_total} + print(" G3 CRUD 契约 %d 个（tblname+params/禁 table,list/.dspy 引用共 %d 处/browserfields 非空）" + % (len(files), dspy_total)) + + +# -------------------------------------------------------------------------- +# G4 禁硬编码库名 + sqlor API 白名单 +# -------------------------------------------------------------------------- +def gate_g4(): + pys = sorted(iter_files(MODULE_ROOT, ".py")) + hardcode = 0 + for p in pys: + tree, _src = parse_ast(p) + if tree is None: + continue + for node in ast.walk(tree): + if isinstance(node, (ast.Assign, ast.AnnAssign)): + targets = node.targets if isinstance(node, ast.Assign) else [node.target] + for t in targets: + nm = None + if isinstance(t, ast.Name): + nm = t.id + elif isinstance(t, ast.Attribute): + nm = t.attr + if not nm or not DBNAME_TARGET_RE.match(nm): + continue + v = const_str(node.value) if node.value is not None else None + if v is not None: + hardcode += 1 + fail("G4", "%s:%s 硬编码库名 %s = %r（必须用 ServerEnv().get_module_dbname('%s')）" + % (rel(p), getattr(node, "lineno", "?"), nm, v, PKG_NAME)) + if isinstance(node, ast.Call): + for kw in (node.keywords or []): + if kw.arg and DBNAME_TARGET_RE.match(kw.arg): + v = const_str(kw.value) + if v is not None: + hardcode += 1 + fail("G4", "%s:%s 调用实参硬编码库名 %s=%r" % (rel(p), node.lineno, kw.arg, v)) + # sqlor 白名单：.py 走 AST（注释/文档字符串不会误报） + used = set() + illegal = [] + for p in pys: + tree, _src = parse_ast(p) + if tree is None: + continue + for node in ast.walk(tree): + if isinstance(node, ast.Attribute) and isinstance(node.value, ast.Name) \ + and node.value.id in ("sor", "sqlor"): + attr = node.attr + if attr in SQLOR_IGNORE: + continue + used.add(attr) + if attr not in SQLOR_WHITELIST: + illegal.append("%s:%s %s.%s" % (rel(p), node.lineno, node.value.id, attr)) + # .json（CRUD/表定义/dspy 契约）里的 sqlor 引用走正则 + for p in sorted(list(iter_files(JSON_DIR, ".json")) + list(iter_files(MODELS_DIR, ".json"))): + txt = read_text(p) or "" + for m in SQLOR_TEXT_RE.finditer(txt): + attr = m.group(1) + if attr in SQLOR_IGNORE: + continue + used.add(attr) + if attr not in SQLOR_WHITELIST: + illegal.append("%s sqlor.%s" % (rel(p), attr)) + for item in sorted(set(illegal)): + fail("G4", "sqlor API 越白名单（仅允许 %s）：%s" % ("/".join(sorted(SQLOR_WHITELIST)), item)) + # 正向断言：包内必须通过 get_module_dbname 取库名 + joined = "" + for p in iter_files(PKG_DIR, ".py"): + joined += (read_text(p) or "") + if "get_module_dbname" not in joined: + fail("G4", "包内未出现 get_module_dbname 调用，库名来源不合规（禁硬编码 DBNAME）") + STATS["G4"] = {"py": len(pys), "hardcode": hardcode, "sqlor_used": sorted(used), "illegal": len(set(illegal))} + print(" G4 扫描 %d 个 .py：硬编码库名命中 %d，sqlor 使用 API=%s（白名单 %s）" + % (len(pys), hardcode, sorted(used) or "[]", "/".join(sorted(SQLOR_WHITELIST)))) + + +# -------------------------------------------------------------------------- +# G5 函数三处注册同步 +# -------------------------------------------------------------------------- +def module_stems(): + return set(os.path.splitext(os.path.basename(p))[0] for p in iter_files(PKG_DIR, ".py")) + + +def registered_names(): + """从 init.py 解析 env 注册的对外函数名（AST，避免注释误报）。""" + names = set() + tree, _src = parse_ast(INIT_PY) + if tree is None: + return names + for node in ast.walk(tree): + if isinstance(node, (ast.Assign, ast.AnnAssign)): + targets = node.targets if isinstance(node, ast.Assign) else [node.target] + for t in targets: + # env.xxx = fn / _env.xxx = fn + if isinstance(t, ast.Attribute) and isinstance(t.value, ast.Name) \ + and t.value.id.lower().endswith("env"): + names.add(t.attr) + # env['xxx'] = fn + elif isinstance(t, ast.Subscript) and isinstance(t.value, ast.Name) \ + and t.value.id.lower().endswith("env"): + key = const_str(subscript_key(t)) + if key: + names.add(key) + # API_FUNCS = ("a", "b") 之类的批量注册清单 + elif isinstance(t, ast.Name) and re.search(r"(API|FUNC|ROUTE|EXPORT|HANDLER|CRUD)", t.id.upper()): + if isinstance(node.value, (ast.List, ast.Tuple, ast.Set)): + for el in node.value.elts: + s = const_str(el) + if s: + names.add(s) + if isinstance(node, ast.Call): + fn = node.func + fname = fn.attr if isinstance(fn, ast.Attribute) else (fn.id if isinstance(fn, ast.Name) else "") + if fname.startswith("register") or fname in ("add_api", "add_function", "add_route", "route", "setattr"): + if node.args: + s = const_str(node.args[0]) + if s: + names.add(s) + return names + + +def pkg_symbols(): + """返回 (包内 def 名集合, 包内 import 进来的名字集合)。""" + defs = set() + imports = set() + for p in iter_files(PKG_DIR, ".py"): + tree, _src = parse_ast(p) + if tree is None: + continue + for node in tree.body: + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + defs.add(node.name) + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + defs.add(node.name) + elif isinstance(node, ast.ImportFrom): + for a in node.names: + imports.add(a.asname or a.name) + elif isinstance(node, ast.Import): + for a in node.names: + imports.add((a.asname or a.name).split(".")[0]) + return defs, imports + + +def pkg_exports(): + """__init__.py 的对外导出面：__all__ 字符串 + from .x import (...) 名字。""" + exports = set() + tree, _src = parse_ast(PKG_INIT_PY) + if tree is None: + return exports + for node in ast.walk(tree): + if isinstance(node, ast.Assign): + for t in node.targets: + if isinstance(t, ast.Name) and t.id == "__all__" and isinstance(node.value, (ast.List, ast.Tuple, ast.Set)): + for el in node.value.elts: + s = const_str(el) + if s: + exports.add(s) + if isinstance(node, ast.ImportFrom): + for a in node.names: + exports.add(a.asname or a.name) + return exports + + +def gate_g5(): + if not os.path.isfile(INIT_PY): + fail("G5", "缺少 init.py（挂载入口）：%s" % rel(INIT_PY)) + if not os.path.isfile(PKG_INIT_PY): + fail("G5", "缺少 __init__.py（包导出面）：%s" % rel(PKG_INIT_PY)) + stems = module_stems() + reg = set() + for n in registered_names(): + if not isinstance(n, str): + continue + n = n.strip() + if not IDENT_RE.match(n) or n in NOISE or n in stems: + continue + if n.startswith("_"): + continue + reg.add(n) + defs, imports = pkg_symbols() + exports = pkg_exports() + if not reg: + fail("G5", "init.py 未解析到任何 env 注册函数名，三处注册同步无法核对：%s" % rel(INIT_PY)) + miss_def = sorted(n for n in reg if n not in defs and n not in imports) + miss_exp = sorted(n for n in reg if n not in exports) + for n in miss_def: + fail("G5", "注册函数 %s 在包内无 def 定义（三处同步之「定义」缺失）" % n) + for n in miss_exp: + fail("G5", "注册函数 %s 未在 __init__.py 导出（三处同步之「导出」缺失）" % n) + phantom = sorted(n for n in exports + if n not in defs and n not in imports + and isinstance(n, str) and IDENT_RE.match(n) and n not in NOISE and n not in stems) + for n in phantom: + fail("G5", "__init__.py 导出的 %s 在包内既无定义也无导入（幽灵导出）" % n) + STATS["G5"] = {"registered": len(reg), "defs": len(defs), "exports": len(exports), + "miss_def": miss_def, "miss_export": miss_exp, "phantom": phantom} + print(" G5 三处注册同步：env 注册 %d 个 / 包内 def %d 个 / __init__ 导出 %d 个；缺定义 %d、缺导出 %d、幽灵导出 %d" + % (len(reg), len(defs), len(exports), len(miss_def), len(miss_exp), len(phantom))) + + +# -------------------------------------------------------------------------- +# G6 load_path.py RBAC 路径显式枚举 +# -------------------------------------------------------------------------- +def gate_g6(): + if not os.path.isfile(LOAD_PATH_PY): + fail("G6", "缺少 scripts/load_path.py：%s" % rel(LOAD_PATH_PY)) + STATS["G6"] = {"paths": 0, "wildcard": 0} + return + tree, src = parse_ast(LOAD_PATH_PY) + paths = [] + if tree is not None: + for node in ast.walk(tree): + s = const_str(node) + if s and s.strip().startswith("/"): + paths.append(s.strip()) + if not paths: + paths = PATH_LITERAL_RE.findall(src or "") + if not paths: + fail("G6", "load_path.py 未解析到任何显式 RBAC 路径（必须逐条枚举注册路径）") + uniq = sorted(set(paths)) + wild = [p for p in uniq if ("%" in p or "*" in p or "?" in p)] + for p in wild: + fail("G6", "RBAC 路径含通配符（禁止 %%/*/?，必须显式枚举）：%s" % p) + no_slash = [p for p in uniq if not p.startswith("/")] + for p in no_slash: + fail("G6", "RBAC 路径未以 / 开头：%s" % p) + if len(uniq) < 3: + warn("G6", "显式枚举 RBAC 路径仅 %d 条，建议覆盖模块全部对外接口" % len(uniq)) + STATS["G6"] = {"paths": len(uniq), "wildcard": len(wild), "sample": uniq[:5]} + print(" G6 load_path.py 显式 RBAC 路径 %d 条，通配符命中 %d 条（样例 %s）" + % (len(uniq), len(wild), uniq[:3])) + + +# -------------------------------------------------------------------------- +# 主流程 +# -------------------------------------------------------------------------- +GATES = ( + ("G1", "py_compile 全量编译", gate_g1), + ("G2", "models 表定义四段式与硬约束", gate_g2), + ("G3", "json CRUD 契约", gate_g3), + ("G4", "禁硬编码库名 + sqlor 白名单", gate_g4), + ("G5", "函数三处注册同步", gate_g5), + ("G6", "load_path.py RBAC 路径显式枚举", gate_g6), +) + + +def main(): + print("=" * 78) + print("pbl_blueprint M1a verify_gate —— 严格退出码门禁 G1~G6（真实断言，非注释大纲）") + print("MODULE_ROOT = %s" % MODULE_ROOT) + print("python = %s" % sys.version.split()[0]) + print("=" * 78) + for gid, title, fn in GATES: + before = len(FAILURES) + print("[%s] %s" % (gid, title)) + try: + fn() + except Exception as e: # noqa: BLE001 —— 门禁自身异常也必须判失败，不得静默 rc=0 + fail(gid, "门禁执行异常 %s: %s: %s" % (gid, type(e).__name__, e)) + print("[%s] %s" % (gid, "PASS" if len(FAILURES) == before else "FAIL")) + print("-" * 78) + + if WARNS: + print("警告（不阻断，共 %d 条）：" % len(WARNS)) + for g, m in WARNS: + print(" WARN [%s] %s" % (g, m)) + print("-" * 78) + + if FAILURES: + print("失败明细（共 %d 条）：" % len(FAILURES)) + for g, m in FAILURES: + print(" FAIL [%s] %s" % (g, m)) + gates_failed = sorted(set(g for g, _m in FAILURES)) + print("失败门禁编号：%s" % ",".join(gates_failed)) + print("STATS: %s" % json.dumps(STATS, ensure_ascii=False, sort_keys=True)) + print("GATE RESULT: FAIL rc=1") + return 1 + + print("全部门禁通过：%s" % ",".join(g for g, _t, _f in GATES)) + print("STATS: %s" % json.dumps(STATS, ensure_ascii=False, sort_keys=True)) + print("GATE RESULT: PASS rc=0") + return 0 + + +if __name__ == "__main__": + sys.exit(main())