863 lines
36 KiB
Python
863 lines
36 KiB
Python
#!/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 '<lambda>'
|
||
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[<str>] = create_blueprint （Subscript 赋值）
|
||
3) API_FUNCS = [<str>, ...] （清单）
|
||
4) app.register_module(MODULE_NAME, {..., <api 字典键>: {契约名: 实现}})
|
||
—— 本模块真实契约（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 != '<lambda>':
|
||
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 != '<lambda>':
|
||
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 != '<lambda>':
|
||
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 != '<lambda>':
|
||
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())
|