From 0f5e22fe51162439059f09e29e6be35fd8b3795c Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Tue, 22 Sep 2026 18:42:13 +0800 Subject: [PATCH] =?UTF-8?q?approve:=20=E7=BC=96=E5=86=99=E5=B9=B6=E6=89=A7?= =?UTF-8?q?=E8=A1=8C=E5=B9=82=E7=AD=89/C-3=20fail-closed=20=E6=B5=8B?= =?UTF-8?q?=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/selfcheck_m5a.py | 364 +++++++++++++++++++++++++++++++++++ tests/m5a_live_db_harness.py | 197 +++++++++++++++---- 2 files changed, 520 insertions(+), 41 deletions(-) diff --git a/scripts/selfcheck_m5a.py b/scripts/selfcheck_m5a.py index 672338a..317eeee 100644 --- a/scripts/selfcheck_m5a.py +++ b/scripts/selfcheck_m5a.py @@ -23,6 +23,16 @@ 6. init/data.json 存在且 json.load 通过（禁占位词） 7. C-3 fail-closed：resolve_event_table() 返回空时 collector 必须 raise CollectError 且消息含 'M11b' 8. README 引用的 scripts/*.py 路径全部在磁盘命中（含本脚本自身） +9. harness 静态门禁：main() 外层有 try/except 收敛未捕获异常；raw()/q_all()/q_exec() 不再出现 + `args or {}` / `dict(params or {})` 反模式；种子 INSERT 用具名参数 `%(event_id)s` 而非位置 `%s` +10. harness 零明文凭据：password=/user=/test123 等口令字面量 0 命中；db_kwargs() 只从 PBL_M5A_DB + 或 projects/pbls/env/test.json 取凭据，取不到即 raise（禁 fallback 到硬编码口令） +11. 运行证据日志：projects/pbls/docs/02-develop/ 下 m5a-live-db-harness.log / m5a-selfcheck.log / + m5a-pytest-offline.log 三份均存在且含 `# RC=0` 收尾标记；harness 日志额外含 + `LIVE DB HARNESS ALL PASS` 与 C-3.1~C-3.4、IDM.1~IDM.9 全 [PASS] 行； + .qc_tmp/ 下的崩溃 traceback 日志不计为证据 +12. harness DDL 与模型一致：DDL_PBL_EVIDENCE 解析出的列集合 ⊆ models/pbl_evidence.json fields， + 且 UNIQUE KEY uk_ev_dedup 列组合 == (tenant_id, source_event_id, evidence_type) 关于「四段式」口径：database-table-definition-spec 规定的四段是 summary/fields/indexes/codes（任务书写的 primary/columns/indexes/constraints 是其 @@ -43,6 +53,7 @@ PKG_DIR = os.path.join(ROOT, 'pbl_evidence') WWWROOT = os.path.join(ROOT, 'wwwroot') MODELS_DIR = os.path.join(ROOT, 'models') JSON_DIR = os.path.join(ROOT, 'json') +HARNESS_PATH = os.path.join(ROOT, 'tests', 'm5a_live_db_harness.py') # 期望的幂等键三元组（C-3 硬约束） UK_EXPECTED = ['tenant_id', 'source_event_id', 'evidence_type'] @@ -58,6 +69,36 @@ CONTRACT_APIS = ( MIN_ENV_REGISTRATIONS = 10 MIN_UI_API_REFS = 9 +# ── 第 9~12 项门禁用的常量 ──────────────────────────────────────────── +# 项目过程文档落点（相对机构工作空间，见 project-directory-spec 第五章） +DOCS_REL = os.path.join('projects', 'pbls', 'docs', '02-develop') +# 必须存在的三份运行证据日志（只认 docs/02-develop 顶层，.qc_tmp/ 下的一律不算） +EVIDENCE_LOGS = ( + 'm5a-live-db-harness.log', + 'm5a-selfcheck.log', + 'm5a-pytest-offline.log', +) +RC_MARKER = '# RC=0' +HARNESS_ALL_PASS_TOKEN = 'LIVE DB HARNESS ALL PASS' +# harness 必须逐条 [PASS] 的断言编号（C-3 fail-closed 4 条 + 幂等 9 条） +HARNESS_CHECK_TOKENS = ( + 'C-3.1', 'C-3.2', 'C-3.3', 'C-3.4', + 'IDM.1', 'IDM.2', 'IDM.3', 'IDM.4', 'IDM.5', 'IDM.6', 'IDM.7', 'IDM.8', 'IDM.9', +) +# 明文口令字面量黑名单（harness 源码里 0 命中才算过） +CREDENTIAL_PATTERNS = ( + r"password\s*=\s*['\"]", + r'user\s*=\s*[\'"]test[\'"]', + r"['\"]test123['\"]", + r"passwd\s*=\s*['\"]", +) +# 数据访问层「静默吞参」反模式（会把 None 变成 {} 掩盖调用方漏传参的缺陷） +PARAM_SWALLOW_PATTERNS = ('args or {}', 'params or {}', 'dict(params or {})', 'or {}') +# harness 里需要收敛为断言的 phase 调用（main 外层 try 必须包住它们） +HARNESS_PHASE_CALLS = ('install_stub', 'phase_fail_closed', 'phase_idempotent', 'phase_third_run') +# 崩溃 traceback 特征（出现即说明该日志是崩溃现场，不是通过证据） +TRACEBACK_TOKEN = 'Traceback (most recent call last)' + def _read(path): with open(path, 'r', encoding='utf-8') as fh: @@ -127,6 +168,33 @@ def _env_registrations(path, known_symbols=None): return out +def _funcs_by_name(src, path, names): + """返回 {name: 源码片段}，收集文件内所有指定名字的函数/方法定义（含类内方法）。""" + tree = ast.parse(src, filename=path) + want = set(names) + out = {} + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name in want: + seg = ast.get_source_segment(src, node) or '' + # 同名函数（如类方法 vs 顶层）都保留，拼接后一起判定 + out[node.name] = (out.get(node.name, '') + '\n' + seg) + return out + + +def _find_docs_dir(): + """从模块目录向上定位机构工作空间根，返回 docs/02-develop 绝对路径（可能不存在）。""" + cur = ROOT + for _ in range(6): + parent = os.path.dirname(cur) + if parent == cur: + break + cand = os.path.join(parent, DOCS_REL) + if os.path.isdir(cand): + return cand + cur = parent + return os.path.join(ROOT, os.pardir, os.pardir, DOCS_REL) + + # ── 校验项 ───────────────────────────────────────────────────────────── def check_three_place_sync(): """① 包内定义 ⊇ ② __init__.__all__ ⊇ ③ init.py env 注册（≥10）。""" @@ -476,6 +544,298 @@ def check_readme_reference(): % (len(refs), len(external), ', '.join(refs))) +# ── 第 9~12 项：harness 与运行证据门禁（QC #4 退回项）───────────────── +def check_harness_static(): + """⑨ harness 源码静态门禁：异常收敛 + 无吞参反模式 + 种子 INSERT 具名参数。 + + 三项都只读源码做 AST/正则判定，不执行 harness（本脚本不连库、不跑子进程）。 + """ + if not os.path.isfile(HARNESS_PATH): + return False, '缺少 %s' % _rel(HARNESS_PATH) + src = _read(HARNESS_PATH) + try: + tree = ast.parse(src, filename=HARNESS_PATH) + except Exception as exc: + return False, '%s 语法错误，无法静态校验: %s' % (_rel(HARNESS_PATH), exc) + + problems = [] + detail = [] + + # (a) main() 外层 try/except 收敛未捕获异常：崩溃必须变成 FAIL 断言而非 traceback 逃逸 + main_fn = None + for node in tree.body: + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == 'main': + main_fn = node + break + if main_fn is None: + problems.append('harness 未定义 main()') + else: + converged = None + for node in ast.walk(main_fn): + if not isinstance(node, ast.Try): + continue + body_src = ast.get_source_segment(src, node) or '' + if not any(c in body_src for c in HARNESS_PHASE_CALLS): + continue # 只认包住 phase 调用的「外层」try + for handler in node.handlers: + htype = getattr(handler.type, 'id', None) or getattr(handler.type, 'attr', None) + if htype not in ('Exception', 'BaseException'): + continue + hsrc = ast.get_source_segment(src, handler) or '' + # 收敛 = 记失败/给出非零退出，而不是原样 raise 或静默 pass + if re.search(r'\braise\b', hsrc) and 'check(' not in hsrc: + continue + if 'check(' in hsrc or 'return 1' in hsrc or 'sys.exit(1)' in hsrc \ + or 'FAILED' in hsrc or 'format_exc' in hsrc: + converged = handler + break + if converged is not None: + break + if converged is None: + problems.append('main() 缺少包住 phase 调用的外层 try/except Exception 收敛' + '（未捕获异常会以 traceback 崩溃逃逸，日志变 .qc_tmp 不算证据）') + else: + detail.append('main() 外层 try/except 收敛未捕获异常') + + # (b) raw()/q_all()/q_exec() 不再出现 `args or {}` / `dict(params or {})` 吞参反模式 + funcs = _funcs_by_name(src, HARNESS_PATH, ('raw', 'q_all', 'q_exec')) + missing_fn = sorted(set(('raw', 'q_all', 'q_exec')) - set(funcs)) + if missing_fn: + problems.append('harness 缺函数 %s' % ', '.join(missing_fn)) + else: + bad = [] + for name in sorted(funcs): + for pat in PARAM_SWALLOW_PATTERNS: + if pat in funcs[name]: + bad.append('%s() 含 %s' % (name, pat)) + if bad: + problems.append('数据访问层静默吞参反模式（None→{} 掩盖漏传参）: %s' % '; '.join(sorted(bad))) + else: + detail.append('raw()/q_all()/q_exec() 无 args or {} / dict(params or {}) 反模式') + + # (c) 种子 INSERT 用具名参数 %(event_id)s，而非位置 %s（位置参数与列序耦合，改列即错位） + seed_hits = list(re.finditer(r'INSERT\s+INTO\s+`?pbl_runtime_event`?', src, re.I)) + if not seed_hits: + problems.append('harness 中未找到 pbl_runtime_event 种子 INSERT 语句') + else: + named_ok = 0 + positional = [] + for m in seed_hits: + window = src[m.start():m.start() + 600] + flat = re.sub(r'\s+', ' ', window) + if '%(event_id)s' in window: + named_ok += 1 + if re.search(r'VALUES\s*\(\s*%s\s*(,|\))', flat): + positional.append(flat[:60]) + if named_ok == 0: + problems.append('种子 INSERT 未使用具名参数 %%(event_id)s（%d 处均为位置参数）' + % len(seed_hits)) + elif named_ok < len(seed_hits): + problems.append('种子 INSERT 具名/位置混用：具名 %d 处 / 共 %d 处' + % (named_ok, len(seed_hits))) + else: + detail.append('种子 INSERT %d 处均用具名参数 %%(event_id)s' % named_ok) + if positional: + problems.append('种子 INSERT 仍含位置参数 VALUES (%s,...)：%d 处' + % ('%s', len(positional))) + + if problems: + return False, '; '.join(problems) + return True, ' | '.join(detail) + + +def check_no_hardcoded_credentials(): + """⑩ harness 零明文凭据：口令字面量 0 命中；db_kwargs() 只从 env 变量/env/test.json 取。 + + 环境信息唯一事实源是 projects/pbls/env/test.json（project-directory-spec 第八章）， + 应用/模块仓库内不得内嵌部署凭据——写死 fallback 口令既泄密又造成多份矛盾环境。 + """ + if not os.path.isfile(HARNESS_PATH): + return False, '缺少 %s' % _rel(HARNESS_PATH) + src = _read(HARNESS_PATH) + problems = [] + detail = [] + + hits = [] + for lineno, line in enumerate(src.splitlines(), 1): + for pat in CREDENTIAL_PATTERNS: + if re.search(pat, line, re.I): + hits.append('%s@L%d' % (re.sub(r'\s+', '', pat), lineno)) + if hits: + problems.append('harness 源码命中明文口令字面量 %d 处: %s' + % (len(hits), ', '.join(sorted(set(hits))))) + else: + detail.append('password=/user=/test123 等明文凭据字面量 0 命中') + + try: + tree = ast.parse(src, filename=HARNESS_PATH) + except Exception as exc: + problems.append('harness 语法错误，无法解析 db_kwargs(): %s' % exc) + return False, '; '.join(problems) + + db_kwargs_src = '' + for node in ast.walk(tree): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == 'db_kwargs': + db_kwargs_src += (ast.get_source_segment(src, node) or '') + if not db_kwargs_src: + problems.append('harness 未定义 db_kwargs()（凭据来源无从校验）') + else: + has_env_src = ('PBL_M5A_DB' in db_kwargs_src) or ('test.json' in db_kwargs_src) + if not has_env_src: + problems.append("db_kwargs() 未从 PBL_M5A_DB 或 env/test.json 读取凭据" + "（唯一事实源缺失，只能靠硬编码）") + else: + srcs = [] + if 'PBL_M5A_DB' in db_kwargs_src: + srcs.append('PBL_M5A_DB') + if 'test.json' in db_kwargs_src: + srcs.append('env/test.json') + detail.append('db_kwargs() 凭据来源: %s' % ' / '.join(srcs)) + # 取不到必须 raise：静默回落到硬编码口令 = 用错库还自称 PASS + if 'raise' not in db_kwargs_src: + problems.append('db_kwargs() 取不到凭据时未 raise（会静默回落到硬编码口令连错库）') + else: + detail.append('db_kwargs() 缺凭据即 raise，无静默回落') + + if problems: + return False, '; '.join(problems) + return True, ' | '.join(detail) + + +def check_evidence_logs(): + """⑪ 运行证据日志：三份日志齐备且含 `# RC=0`；harness 日志含 ALL PASS 与全 [PASS] 行。 + + 只认 projects/pbls/docs/02-develop/ 顶层文件：.qc_tmp/ 下的是崩溃 traceback 现场， + 属于「跑挂了」的记录而非「跑过了」的证据，计入会把失败伪装成交付。 + """ + docs = _find_docs_dir() + if not os.path.isdir(docs): + return False, '缺少证据目录 %s（相对工作空间 projects/pbls/docs/02-develop）' % DOCS_REL.replace(os.sep, '/') + top = sorted(f for f in os.listdir(docs) + if os.path.isfile(os.path.join(docs, f))) + problems = [] + detail = [] + + # .qc_tmp/ 下的同名文件不算证据（显式统计出来，便于定位「日志放错地方」） + qc_dir = os.path.join(docs, '.qc_tmp') + qc_names = sorted(f for f in (os.listdir(qc_dir) if os.path.isdir(qc_dir) else [])) + qc_only = sorted(set(EVIDENCE_LOGS) & set(qc_names) - set(top)) + + contents = {} + for name in EVIDENCE_LOGS: + if name not in top: + where = '（仅存在于 .qc_tmp/，不计为证据）' if name in qc_only else '' + problems.append('缺少证据日志 %s%s' % (name, where)) + continue + text = _read(os.path.join(docs, name)) + contents[name] = text + if RC_MARKER not in text: + problems.append('%s 缺 `%s` 收尾标记（未记录退出码=无法证明 rc=0）' % (name, RC_MARKER)) + if TRACEBACK_TOKEN in text and name != 'm5a-live-db-harness.log': + problems.append('%s 含崩溃 traceback，非通过证据' % name) + if problems: + return False, '; '.join(problems) + + detail.append('三份证据日志齐备且均含 `%s`: %s' % (RC_MARKER, ', '.join(sorted(EVIDENCE_LOGS)))) + + hlog = contents['m5a-live-db-harness.log'] + if HARNESS_ALL_PASS_TOKEN not in hlog: + problems.append('harness 日志缺 `%s` 收尾行' % HARNESS_ALL_PASS_TOKEN) + if TRACEBACK_TOKEN in hlog: + problems.append('harness 日志含崩溃 traceback（未收敛的未捕获异常）') + missing_pass = sorted(t for t in HARNESS_CHECK_TOKENS + if not re.search(r'\[PASS\]\s*' + re.escape(t) + r'\b', hlog)) + if missing_pass: + problems.append('harness 日志缺 [PASS] 断言行: %s' % ', '.join(missing_pass)) + else: + detail.append('harness 日志含 %s 与 C-3.1~C-3.4 / IDM.1~IDM.9 共 %d 条 [PASS] 行' + % (HARNESS_ALL_PASS_TOKEN, len(HARNESS_CHECK_TOKENS))) + + if problems: + return False, '; '.join(problems) + return True, ' | '.join(detail) + + +def _parse_ddl_columns(ddl): + """从 CREATE TABLE 正文里取「列定义」名（排除 PRIMARY/UNIQUE/KEY/CONSTRAINT 等索引行）。""" + body = re.search(r'\((.*)\)\s*(?:ENGINE|COMMENT|=|$)', ddl, re.S) + seg = body.group(1) if body else ddl + cols = [] + for line in seg.splitlines(): + s = line.strip().rstrip(',') + m = re.match(r'`(\w+)`\s+[A-Za-z]', s) + if not m: + continue + if re.match(r'(PRIMARY|UNIQUE|KEY|CONSTRAINT|INDEX)\b', s, re.I): + continue + cols.append(m.group(1)) + return cols + + +def _parse_ddl_unique(ddl, uk_name): + """取 `UNIQUE KEY \\`uk_ev_dedup\\` (\\`a\\`, \\`b\\`)` 的列组合（保序）。""" + m = re.search(r'UNIQUE\s+KEY\s+`?%s`?\s*\(([^)]*)\)' % re.escape(uk_name), ddl, re.I) + if not m: + return None + return [c.strip().strip('`').strip() for c in m.group(1).split(',') if c.strip()] + + +def check_harness_matches_model(): + """⑫ harness 沙箱 DDL 与 models/pbl_evidence.json 对齐（列 ⊆ 模型列、uk 三元组一致）。 + + harness 自建表跑出来的幂等结论，只有与真源（models/*.json）同构才有效；否则「沙箱 + 里幂等、生产上漂移」会被日志证据掩盖。 + """ + problems = [] + if not os.path.isfile(HARNESS_PATH): + return False, '缺少 %s' % _rel(HARNESS_PATH) + src = _read(HARNESS_PATH) + try: + tree = ast.parse(src, filename=HARNESS_PATH) + except Exception as exc: + return False, 'harness 语法错误，无法解析 DDL_PBL_EVIDENCE: %s' % exc + + ddl = None + for node in tree.body: + if isinstance(node, ast.Assign): + for tgt in node.targets: + if isinstance(tgt, ast.Name) and tgt.id == 'DDL_PBL_EVIDENCE': + v = node.value + if isinstance(v, ast.Constant) and isinstance(v.value, str): + ddl = v.value + if ddl is None: + return False, 'harness 未找到模块级常量 DDL_PBL_EVIDENCE（字符串字面量）' + + mpath = os.path.join(MODELS_DIR, 'pbl_evidence.json') + if not os.path.isfile(mpath): + return False, '缺少 %s' % _rel(mpath) + model = json.loads(_read(mpath)) + model_cols = [fd.get('name') for fd in (model.get('fields') or []) if fd.get('name')] + uk_idx = [i for i in (model.get('indexes') or []) if i.get('name') == 'uk_ev_dedup'] + if len(uk_idx) != 1: + problems.append('models/pbl_evidence.json 中 uk_ev_dedup 索引数 %d != 1' % len(uk_idx)) + model_uk = list(uk_idx[0].get('idxfields') or []) if uk_idx else [] + + ddl_cols = _parse_ddl_columns(ddl) + ddl_uk = _parse_ddl_unique(ddl, 'uk_ev_dedup') + + if not ddl_cols: + problems.append('DDL_PBL_EVIDENCE 未解析出任何列定义') + unknown = sorted(set(ddl_cols) - set(model_cols)) + if unknown: + problems.append('DDL 列不在模型 fields 中: %s' % ', '.join(unknown)) + if sorted(set(ddl_uk or [])) != sorted(UK_EXPECTED): + problems.append('DDL uk_ev_dedup 列组合 %s != %s' % (ddl_uk, UK_EXPECTED)) + if model_uk and ddl_uk and list(ddl_uk) != list(model_uk): + problems.append('DDL uk_ev_dedup 列序 %s != 模型 idxfields %s' % (ddl_uk, model_uk)) + + if problems: + return False, '; '.join(problems) + return True, ('DDL_PBL_EVIDENCE %d 列 ⊆ models/pbl_evidence.json %d 列；' + 'uk_ev_dedup == %s（与模型 idxfields 逐列一致）' + % (len(ddl_cols), len(model_cols), tuple(ddl_uk))) + + CHECKS = [ ('三处同步：包内定义 ⊇ __init__.__all__ ⊇ init.py env 注册（≥10）', check_three_place_sync), ('RBAC：wwwroot/api/*.dspy ⊇ index.ui 声明路径（≥9）且逐条登记 load_path', check_rbac_dspy_vs_ui), @@ -485,6 +845,10 @@ CHECKS = [ ('init/data.json 存在且 json.load 通过（无占位词、parentid 不超长）', check_init_data), ('C-3 fail-closed：事件表缺失时抛 CollectError 且消息含 M11b', check_fail_closed_source), ('README 引用的 scripts/*.py 路径全部磁盘命中（含 selfcheck_m5a.py）', check_readme_reference), + ('harness 静态门禁：main() 外层 try/except 收敛 + 无吞参反模式 + 种子 INSERT 具名参数', check_harness_static), + ('harness 零明文凭据：无 password/test123 字面量，db_kwargs() 只读 PBL_M5A_DB/env test.json 且缺即 raise', check_no_hardcoded_credentials), + ('运行证据日志：docs/02-develop 三份日志含 # RC=0，harness 日志含 ALL PASS 与 C-3.x/IDM.x 全 [PASS]', check_evidence_logs), + ('harness DDL 与模型一致：DDL 列 ⊆ models fields 且 uk_ev_dedup == (tenant_id, source_event_id, evidence_type)', check_harness_matches_model), ] diff --git a/tests/m5a_live_db_harness.py b/tests/m5a_live_db_harness.py index e5e295f..9287f03 100644 --- a/tests/m5a_live_db_harness.py +++ b/tests/m5a_live_db_harness.py @@ -19,15 +19,23 @@ python3 tests/m5a_live_db_harness.py # 跑完自动清理沙箱库 python3 tests/m5a_live_db_harness.py --keep # 保留沙箱库供人工核对 - PBL_M5A_DB=host:port:user:pwd 覆盖默认 test/test123@127.0.0.1:3306 -退出码：0 = 全部断言通过（stdout 末行 ``LIVE DB HARNESS ALL PASS``）；1 = 失败。 +沙箱库凭据（**代码内零明文口令**，唯一事实源二选一，取不到即 fail-fast）:: + + 1) 环境变量 PBL_M5A_DB=host:port:user:pwd + 2) 环境文件 projects/pbls/env/test.json 的 db.sandbox 段（scope=sandbox_only） + + 该沙箱账号**只用于一次性沙箱 schema**（本脚本自建库、跑完 DROP DATABASE）， + 不接触业务库、不承载任何应用运行时连接。日志对口令做脱敏，绝不打印。 + +退出码：0 = 全部断言通过（stdout 末行 ``LIVE DB HARNESS ALL PASS``）；1 = 失败 +（含未捕获异常——统一收敛为显式 FAIL 项，不靠 traceback 截断证据）。 """ import argparse -import contextlib import importlib import io +import json import os import pathlib import re @@ -36,7 +44,9 @@ import traceback import types REPO_ROOT = pathlib.Path(__file__).resolve().parent.parent # modules/pbl_evidence +WORKSPACE_ROOT = REPO_ROOT.parent.parent # 机构工作空间 PKG_DIR = REPO_ROOT / 'pbl_evidence' +ENV_FILE_DEFAULT = str(WORKSPACE_ROOT / 'projects' / 'pbls' / 'env' / 'test.json') SANDBOX_SCHEMA = os.environ.get('PBL_M5A_SANDBOX', 'pbl_m5a_u7rb') _TENANT = 't1' @@ -83,18 +93,39 @@ CREATE TABLE `pbl_runtime_event` ( ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci """ +# 种子事件：具名字段 + 字典（pymysql 具名参数 %(field)s），避免位置元组与 +# 列数不一致时触发 "not enough arguments for format string" 类崩溃。 +SEED_FIELDS = ('event_id', 'tenant_id', 'event_type', 'payload_json', 'occurred_at', + 'learner_id', 'session_id', 'blueprint_id', 'artifact_id') + EVENTS_SEED = ( - (_EVENT_ID_A, 'artifact_submitted', - '{"title":"bridge-design","score":88}', '2026-09-22 10:00:00', 'stu_01', 'sess_01', 'bp_01', 'art_01'), - (_EVENT_ID_B, 'team_message_sent', - '{"channel":"team","chars":120}', '2026-09-22 10:05:00', 'stu_01', 'sess_01', 'bp_01', '0'), + {'event_id': _EVENT_ID_A, 'tenant_id': _TENANT, 'event_type': 'artifact_submitted', + 'payload_json': '{"title":"bridge-design","score":88}', + 'occurred_at': '2026-09-22 10:00:00', 'learner_id': 'stu_01', + 'session_id': 'sess_01', 'blueprint_id': 'bp_01', 'artifact_id': 'art_01'}, + {'event_id': _EVENT_ID_B, 'tenant_id': _TENANT, 'event_type': 'team_message_sent', + 'payload_json': '{"channel":"team","chars":120}', + 'occurred_at': '2026-09-22 10:05:00', 'learner_id': 'stu_01', + 'session_id': 'sess_01', 'blueprint_id': 'bp_01', 'artifact_id': '0'}, ) _failures = [] +_secret_values = set() # 运行期已解析出的口令，日志统一脱敏 +_CRED_SOURCE = ['(未解析)'] # 凭据来源标签（只记来源，不记口令） + + +def _mask(text): + """把任何可能混入日志/异常消息的口令替换成 ******。""" + out = str(text) + for secret in _secret_values: + if secret: + out = out.replace(secret, '******') + return out def check(name, ok, detail=''): """记录一条断言并原样打印（PASS/FAIL），失败不中断，最后统一 exit 1。""" + detail = _mask(detail) print('[%s] %s%s' % ('PASS' if ok else 'FAIL', name, (' -> ' + detail) if detail else '')) if not ok: @@ -167,9 +198,14 @@ class Sandbox: cur.execute(stmt) def raw(self, sql, args=None): - """执行并返回 list[dict]（原样回显用）。""" + """执行并返回 list[dict]（原样回显用）。 + + 注意：无参 SQL 必须传 None 而不是空 dict —— DictCursor 下传 ``{}`` + 会让 pymysql 走 ``query % {}`` 格式化，SQL 里任何字面 ``%``（LIKE/DATE_FORMAT） + 都会抛 ``TypeError: not enough arguments for format string``。 + """ with self.conn.cursor() as cur: - cur.execute(sql, args or {}) + cur.execute(sql, args if args else None) try: return list(cur.fetchall() or []) except Exception: @@ -182,7 +218,7 @@ class Sandbox: def q_all(self, sql, params=None): with self.conn.cursor() as cur: - cur.execute(self.to_pymysql(sql), dict(params or {})) + cur.execute(self.to_pymysql(sql), dict(params) if params else None) return [dict(r) for r in (cur.fetchall() or [])] def q_one(self, sql, params=None): @@ -191,7 +227,7 @@ class Sandbox: def q_exec(self, sql, params=None): with self.conn.cursor() as cur: - n = cur.execute(self.to_pymysql(sql), dict(params or {})) + n = cur.execute(self.to_pymysql(sql), dict(params) if params else None) return {'affected_rows': n} def table_exists(self, table): @@ -258,14 +294,14 @@ def phase_fail_closed(collector, box): tb_text = ''.join(traceback.format_exc()) print('抛出的异常类型: %s' % (type(err).__name__ if err else 'None（未抛错=静默空成功）')) - print('异常消息原文: %s' % (str(err) if err else '<无>')) + print('异常消息原文: %s' % (_mask(str(err)) if err else '<无>')) if tb_text: print('traceback 原文:') for line in tb_text.rstrip().splitlines(): - print(' | ' + line) + print(' | ' + _mask(line)) check('C-3.2 collect_evidence_from_events(tenant_id="t1") 抛 CollectError', isinstance(err, collector.CollectError), type(err).__name__ if err else 'no raise') - msg = str(err or '') + msg = _mask(str(err or '')) check("C-3.3 异常消息含 'M11b'（硬依赖声明，非空成功）", 'M11b' in msg, '消息=%s' % msg[:120]) check('C-3.4 未产生任何证据行（fail-closed 不写脏数据）', @@ -286,12 +322,15 @@ def phase_idempotent(collector, box): .format(r.get('Seq_in_index'), r.get('Column_name'), r.get('Non_unique'), r.get('Key_name'))) + # 具名参数插入（列名与占位符同源 SEED_FIELDS，列/值不可能错位） + seed_cols = ', '.join('`' + f + '`' for f in SEED_FIELDS) + seed_vals = ', '.join('%(' + f + ')s' for f in SEED_FIELDS) + seed_sql = ('INSERT INTO `pbl_runtime_event` (' + seed_cols + ') VALUES (' + + seed_vals + ')') + print('事件种子 INSERT（具名参数）: %s' % seed_sql) with box.conn.cursor() as cur: for ev in EVENTS_SEED: - cur.execute( - 'INSERT INTO `pbl_runtime_event` (event_id, tenant_id, event_type, ' - 'payload_json, occurred_at, learner_id, session_id, blueprint_id, ' - 'artifact_id) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)', ev) + cur.execute(seed_sql, dict(ev)) print('事件表写入 %d 行（pbl_runtime_event）' % len(EVENTS_SEED)) import asyncio @@ -338,11 +377,13 @@ def phase_idempotent(collector, box): cur.execute( 'INSERT INTO `pbl_evidence` (id, tenant_id, source_event_id, ' 'evidence_type, occurred_at, dedup_key) ' - 'VALUES (%s,%s,%s,%s,%s,%s)', - ('manual_dup_1', _TENANT, _EVENT_ID_A, 'artifact', - '2026-09-22 10:00:00', 'x' * 32)) + 'VALUES (%(id)s,%(tenant_id)s,%(source_event_id)s,' + '%(evidence_type)s,%(occurred_at)s,%(dedup_key)s)', + {'id': 'manual_dup_1', 'tenant_id': _TENANT, + 'source_event_id': _EVENT_ID_A, 'evidence_type': 'artifact', + 'occurred_at': '2026-09-22 10:00:00', 'dedup_key': 'x' * 32}) except Exception as exc: # noqa: BLE001 - dup_err = '%s: %s' % (type(exc).__name__, exc) + dup_err = '%s: %s' % (type(exc).__name__, _mask(exc)) print('直接重复 INSERT 的 DB 原始报错（uk_ev_dedup 真防重）: %s' % (dup_err or '<未报错>')) check('IDM.6 唯一键 uk_ev_dedup 在 DB 层拦截重复三元组（1062 Duplicate entry）', @@ -354,7 +395,7 @@ def phase_idempotent(collector, box): 'ON DUPLICATE KEY UPDATE' in src) check('IDM.8 dedup_key 与 uk 三元组同源（md5(tenant|event|type)）', _dedup_matches(rows2)) - return rows2 + return s2, rows2 def _dedup_matches(rows): @@ -390,36 +431,90 @@ def phase_third_run(collector, box, s2_stats): def db_kwargs(): - env = os.environ.get('PBL_M5A_DB') - if env: - host, port, user, pwd = env.split(':') - return dict(host=host, port=int(port), user=user, password=pwd) - # 本机测试库凭据来自 projects/pbls/env/test.json 的 db 段（唯一事实源，未新增秘密） - return dict(host='127.0.0.1', port=3306, user='test', password='test123', - charset='utf8mb4', connect_timeout=5) + """沙箱库连接参数。凭据唯一事实源（**代码内零明文口令**，取不到即 fail-fast）： + + 1. 环境变量 ``PBL_M5A_DB=host:port:user:pwd``（口令允许含冒号，按前 3 段切分） + 2. 环境文件 ``projects/pbls/env/test.json`` 的 ``db.sandbox`` 段 + （``scope=sandbox_only``，仅授权一次性沙箱 schema 的建/删） + + 两者都拿不到 → 抛 ``RuntimeError``（不回退任何默认账号、不猜凭据）。 + """ + raw_env = (os.environ.get('PBL_M5A_DB') or '').strip() + if raw_env: + parts = raw_env.split(':', 3) + if len(parts) != 4: + raise RuntimeError( + 'PBL_M5A_DB 格式必须为 host:port:user:pwd（4 段冒号分隔），' + '实际段数=%d' % len(parts)) + host, port, user, pwd = parts + if not host or not user: + raise RuntimeError('PBL_M5A_DB 的 host/user 段为空，拒绝连接') + try: + port_i = int(port) + except ValueError: + raise RuntimeError('PBL_M5A_DB 的 port 段不是整数: %r' % (port,)) + _secret_values.add(pwd) + _CRED_SOURCE[0] = '环境变量 PBL_M5A_DB' + return dict(host=host, port=port_i, user=user, password=pwd, + charset='utf8mb4', connect_timeout=5) + + env_file = (os.environ.get('PBL_M5A_ENV_FILE') or '').strip() or ENV_FILE_DEFAULT + if not os.path.isfile(env_file): + raise RuntimeError( + '拿不到沙箱库凭据：环境变量 PBL_M5A_DB 未设置，且环境文件不存在: ' + + env_file + + '。请设置 PBL_M5A_DB=host:port:user:pwd，或在该文件的 db.sandbox 段' + '（engine/host/port/user/password/charset/scope=sandbox_only）补齐沙箱账号。') + try: + with io.open(env_file, 'r', encoding='utf-8') as fh: + cfg = json.load(fh) + except ValueError as exc: + raise RuntimeError('环境文件不是合法 JSON: %s (%s)' % (env_file, _mask(exc))) + sandbox = ((cfg.get('db') or {}).get('sandbox')) or {} + if not sandbox: + raise RuntimeError( + '环境文件缺少 db.sandbox 段（沙箱库凭据唯一事实源）: ' + env_file + + '；或改用环境变量 PBL_M5A_DB=host:port:user:pwd') + scope = str(sandbox.get('scope') or '') + if scope and scope != 'sandbox_only': + raise RuntimeError( + 'db.sandbox.scope=' + scope + ' 非 sandbox_only，拒绝用非沙箱账号跑 harness') + missing = [k for k in ('host', 'port', 'user') if sandbox.get(k) in (None, '')] + if 'password' not in sandbox: + missing.append('password') + if missing: + raise RuntimeError('db.sandbox 缺少必填字段: ' + ', '.join(missing) + + '（文件 ' + env_file + '）') + pwd = str(sandbox['password']) + _secret_values.add(pwd) + _CRED_SOURCE[0] = '环境文件 ' + env_file + ' 的 db.sandbox 段' + return dict(host=str(sandbox['host']), port=int(sandbox['port']), + user=str(sandbox['user']), password=pwd, + charset=str(sandbox.get('charset') or 'utf8mb4'), + connect_timeout=5) -def main(): - ap = argparse.ArgumentParser() - ap.add_argument('--keep', action='store_true', help='跑完保留沙箱库') - args = ap.parse_args() - - print('M5a live-DB harness | sandbox schema = %s | target = modules/pbl_evidence' - % SANDBOX_SCHEMA) +def run(args): + """harness 主体（异常由 main() 统一收敛为显式 FAIL）。""" try: import pymysql # noqa: F401 except Exception as exc: # noqa: BLE001 - print('FAIL 环境缺少 pymysql，无法跑真实库幂等: %r' % (exc,)) + check('ENV.1 运行环境具备 pymysql', False, repr(exc)) return 1 + conn_kwargs = db_kwargs() + print('沙箱库凭据来源: %s | 目标 = %s@%s:%s（口令不打印，日志已脱敏）' + % (_CRED_SOURCE[0], conn_kwargs['user'], conn_kwargs['host'], + conn_kwargs['port'])) + collector = load_collector() - box = Sandbox(db_kwargs()) + box = Sandbox(conn_kwargs) box.reset_schema() box.open() try: install_stub(collector, box) phase_fail_closed(collector, box) - _rows = phase_idempotent(collector, box) + _s2, _rows = phase_idempotent(collector, box) # 重跑一次统计对比（放在 phase 2 之后，避免影响其内部断言顺序） import asyncio loop = asyncio.get_event_loop_policy().new_event_loop() @@ -436,9 +531,29 @@ def main(): print('沙箱库已清理: DROP DATABASE %s' % SANDBOX_SCHEMA) else: print('--keep：沙箱库 %s 保留' % SANDBOX_SCHEMA) + return 0 + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument('--keep', action='store_true', help='跑完保留沙箱库') + args = ap.parse_args() + + print('M5a live-DB harness | sandbox schema = %s | target = modules/pbl_evidence' + % SANDBOX_SCHEMA) + try: + run(args) + except SystemExit: + raise + except Exception as exc: # noqa: BLE001 + # 未捕获异常（凭据缺失/连不上库/被测代码抛错）统一收敛为显式 FAIL 项， + # 打印异常类型 + 脱敏消息，绝不靠 traceback 截断证据。 + check('HARNESS.1 harness 无未捕获异常（凭据可用 + 沙箱库可连 + 各阶段跑完）', + False, '%s: %s' % (type(exc).__name__, _mask(exc))) + print('未捕获异常汇总（脱敏）: %s' % _mask(traceback.format_exc())) if _failures: - print('LIVE DB HARNESS FAILED (%d): %s' % (len(_failures), ' | '.join(_failures))) + print('LIVE DB HARNESS FAILED (%d): %s' + % (len(_failures), ' | '.join(_mask(f) for f in _failures))) return 1 print('LIVE DB HARNESS ALL PASS') return 0