232 lines
10 KiB
Python
232 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""PBL-RTX-0005 补丁器：修复 pbl_runtime_event_ddl.py 过程体非法 RETURN（实库 ERROR 1064）。
|
||
|
||
与上一轮被 QC 退回的 .rtx5tmp/patch_rtx0005.py 的区别（响应退回意见 #7）：
|
||
1) 先 dry-run 校验（每处替换的原文必须**恰好命中 1 次**，且替换后全文裸 `RETURN;` 计数为 0、
|
||
新门禁函数存在、ast 可解析），任一不满足 → 不落盘、rc=1；
|
||
2) 写盘后**重新 read 目标文件**做二次断言（不把 print 输出当成功标志）；
|
||
3) 二次断言失败则回滚为原始内容并 rc=1。
|
||
|
||
用法：python3 scripts/patch_rtx0005.py [--dry-run]
|
||
"""
|
||
import argparse
|
||
import ast
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
TARGET = os.path.join(os.path.dirname(HERE), "scripts", "pbl_runtime_event_ddl.py")
|
||
|
||
# ---------------------------------------------------------------- 替换 1：过程体非法 RETURN
|
||
OLD_PROC = """ IF p_table <> '%(table)s' THEN
|
||
RETURN; -- 只守护事件表，其他表不受影响
|
||
END IF;
|
||
IF IFNULL(@pbl_guard_ctx, '') = '%(ctx)s' THEN
|
||
RETURN; -- break-glass：一次性历史数据回填
|
||
END IF;
|
||
SIGNAL SQLSTATE '45000'
|
||
SET MESSAGE_TEXT = 'PBL_APPEND_ONLY_DENIED',
|
||
MYSQL_ERRNO = 45001;
|
||
"""
|
||
NEW_PROC = """ -- PBL-RTX-0005 根因修复：MariaDB/MySQL 的 RETURN 只允许出现在 FUNCTION 体内。
|
||
-- 过程体里的 RETURN 语句既是服务端语法错误，其后的分号又会让 mysql CLI 在
|
||
-- DELIMITER $$ 生效之前提前切分语句 -> 实库装载 ERROR 1064 (42000)，并连带后续
|
||
-- CREATE PROCEDURE pbl_runtime_event_ensure_partitions 未执行（CALL 报 1305）。
|
||
-- 等价改写为「正向条件 + 单语句 SIGNAL」：仅当守护对象是本表、且当前会话不处于
|
||
-- break-glass 上下文时才拒绝，语义与原先两次提前退出完全一致；
|
||
-- append-only 守卫未被删除、未被注释，仍然真实装载生效。
|
||
IF p_table = '%(table)s' AND IFNULL(@pbl_guard_ctx, '') <> '%(ctx)s' THEN
|
||
SIGNAL SQLSTATE '45000'
|
||
SET MESSAGE_TEXT = 'PBL_APPEND_ONLY_DENIED',
|
||
MYSQL_ERRNO = 45001;
|
||
END IF;
|
||
"""
|
||
|
||
# ---------------------------------------------------------------- 替换 2：新增静态门禁
|
||
OLD_CONST = ('_BARE_TOKEN = re.compile(r"^(CREATE|DROP|ALTER|GRANT|REVOKE)\\s*\\$\\$\\s*$", re.I)\n')
|
||
NEW_CONST = OLD_CONST + '''
|
||
|
||
# ---------------------------------------------------------- 过程体非法 RETURN 静态门禁
|
||
# PBL-RTX-0005：本缺陷的根因形态是「过程/触发器体内出现裸 RETURN 语句」。
|
||
# MariaDB 只允许 FUNCTION 使用 RETURN；PROCEDURE/TRIGGER/EVENT 体内出现即服务端报错，
|
||
# 且 RETURN 后的分号会让 mysql CLI 在 DELIMITER 块内提前切分语句（ERROR 1064）——
|
||
# 光靠 DELIMITER 包裹修不掉，必须让过程体本身不存在这种语句。
|
||
# 因此新增下列常量 + _procedure_body_return_errors()，并在 verify() 里作为硬门禁，
|
||
# 使同类缺陷在 --verify（离线）阶段就被拦下，而不是等实库装载才暴露。
|
||
_ILLEGAL_RETURN = re.compile(r"^RETURN\\b[^A-Za-z0-9_]*(\\$\\$)?$", re.I)
|
||
# 行尾注释（形如 `RETURN; -- 说明`）：判定前必须先剥掉，否则 RETURN 会被漏检
|
||
_TRAILING_COMMENT = re.compile(r"\\s*--[^\\n]*$")
|
||
_PROC_HEAD = re.compile(r"^\\s*(CREATE\\s+(PROCEDURE|TRIGGER|FUNCTION|EVENT))\\b", re.I)
|
||
_FUNC_HEAD = re.compile(r"^\\s*CREATE\\s+FUNCTION\\b", re.I)
|
||
|
||
|
||
def _procedure_body_return_errors(text, label=""):
|
||
"""扫描渲染后的 SQL 产物，报告「过程/触发器/EVENT 体内的非法 RETURN」缺陷列表。
|
||
|
||
判定口径：
|
||
· 从 `CREATE PROCEDURE|TRIGGER|FUNCTION|EVENT` 起进入过程体，到该语句收尾 `END` 退出；
|
||
· 每行先剥掉行尾 `--` 注释再判定，`RETURN;` / `RETURN $$` / 裸 `RETURN` 全部命中；
|
||
· `CREATE FUNCTION` 体放行（RETURN 在函数中合法）；
|
||
· BEGIN/END 嵌套深度用于定位语句收尾，嵌套子块（END IF/END LOOP）不算收尾。
|
||
返回 ["[label] L<n> ...", ...]；空列表 = 通过。
|
||
"""
|
||
errs = []
|
||
pfx = ("[%s] " % label) if label else ""
|
||
depth = 0
|
||
in_proc = False
|
||
is_func = False
|
||
for i, raw in enumerate(text.replace("\\r\\n", "\\n").split("\\n"), 1):
|
||
s = _TRAILING_COMMENT.sub("", raw).strip()
|
||
if not s:
|
||
continue
|
||
if not in_proc and _PROC_HEAD.match(s):
|
||
in_proc = True
|
||
is_func = bool(_FUNC_HEAD.match(s))
|
||
if not in_proc:
|
||
continue
|
||
if _BLOCK_OPEN.match(s):
|
||
depth += 1
|
||
if _ILLEGAL_RETURN.match(s) and not is_func:
|
||
errs.append(pfx + "L%d 过程体出现非法 RETURN 语句（MariaDB 仅 FUNCTION 可用 RETURN，"
|
||
"且该分号会使 CLI 提前切分语句 -> ERROR 1064）: %r"
|
||
% (i, raw.strip()))
|
||
if _BLOCK_CLOSE.match(s) and depth > 0:
|
||
depth -= 1
|
||
if depth == 0:
|
||
in_proc = False
|
||
is_func = False
|
||
return errs
|
||
|
||
|
||
# 门禁自检样本：故意含非法 RETURN 的过程体，verify() 用它断言「门禁真的会报错」。
|
||
# 用字符串拼接构造，避免本 .py 源文件自身出现裸 RETURN; 行（会被门禁自检与补丁断言误判）。
|
||
_ILLEGAL_RETURN_PROBE = (
|
||
"CREATE PROCEDURE `pbl_probe_illegal`()\\n"
|
||
"BEGIN\\n"
|
||
" IF 1 = 1 THEN\\n"
|
||
" " + "RETURN" + ";\\n"
|
||
" END IF;\\n"
|
||
"END;\\n"
|
||
)
|
||
'''
|
||
|
||
# ---------------------------------------------------------------- 替换 3：verify() 接入门禁
|
||
OLD_VERIFY_TAIL = ''' msgs.append("PASS 幂等关键字逐文件覆盖: 01 建表分区 / 02 过程+双触发器 / 03 双过程+EVENT")
|
||
return ok, msgs
|
||
'''
|
||
NEW_VERIFY_TAIL = ''' msgs.append("PASS 幂等关键字逐文件覆盖: 01 建表分区 / 02 过程+双触发器 / 03 双过程+EVENT")
|
||
|
||
# 10. 过程体非法 RETURN 硬门禁（PBL-RTX-0005：实库 ERROR 1064 的根因形态）
|
||
ret_errs = []
|
||
for label, text in artifacts:
|
||
ret_errs.extend(_procedure_body_return_errors(text, label))
|
||
if ret_errs:
|
||
ok = False
|
||
for e in ret_errs:
|
||
msgs.append("FAIL 过程体非法 RETURN: " + e)
|
||
else:
|
||
msgs.append("PASS 过程体无非法 RETURN（PBL-RTX-0005 门禁）: 4 个产物的过程/触发器体均为"
|
||
"等价正向条件单语句 SIGNAL，append-only 守卫完整保留")
|
||
|
||
# 11. 门禁有效性自检：构造含非法 RETURN 的过程体，_procedure_body_return_errors 必须报错；
|
||
# 报不出来说明门禁本身失效（防止「加了个永远返回空的检查」这种假加固）。
|
||
probe = _procedure_body_return_errors(_ILLEGAL_RETURN_PROBE, "probe")
|
||
clean_probe = _procedure_body_return_errors(append_only_sql(), "clean")
|
||
if not any("RETURN" in e for e in probe):
|
||
ok = False
|
||
msgs.append("FAIL 非法 RETURN 门禁失效：含裸 RETURN 的过程体未被检出（假门禁）")
|
||
elif clean_probe:
|
||
ok = False
|
||
msgs.append("FAIL 非法 RETURN 门禁误报：修复后的 append_only_sql 仍被检出 %s" % clean_probe)
|
||
else:
|
||
msgs.append("PASS 非法 RETURN 门禁自检：坏样本(含 RETURN;)被检出 %d 条、"
|
||
"好样本(修复后守卫)零误报" % len(probe))
|
||
return ok, msgs
|
||
'''
|
||
|
||
REPLACEMENTS = [("过程体非法 RETURN 改写为正向条件单语句 SIGNAL", OLD_PROC, NEW_PROC),
|
||
("新增 _ILLEGAL_RETURN/_TRAILING_COMMENT/_procedure_body_return_errors",
|
||
OLD_CONST, NEW_CONST),
|
||
("verify() 接入过程体 RETURN 门禁 + 门禁有效性自检", OLD_VERIFY_TAIL, NEW_VERIFY_TAIL)]
|
||
|
||
# 源文件中「裸 RETURN;」行（门禁自检样本用拼接构造，故此处 0 命中即为干净）
|
||
_BARE_RETURN_RE = re.compile(r"^\s*RETURN\s*;\s*(--.*)?$", re.M)
|
||
|
||
|
||
def build_new(src):
|
||
out = src
|
||
for name, old, new in REPLACEMENTS:
|
||
n = out.count(old)
|
||
if n != 1:
|
||
raise RuntimeError("替换点 %r 原文命中 %d 次（应为 1 次），中止不落盘" % (name, n))
|
||
out = out.replace(old, new, 1)
|
||
return out
|
||
|
||
|
||
def check(text):
|
||
"""落盘前/后统一断言，返回失败原因列表（空 = 通过）。"""
|
||
bad = []
|
||
if "PBL-RTX-0005" not in text:
|
||
bad.append("缺少 PBL-RTX-0005 标记")
|
||
hits = _BARE_RETURN_RE.findall(text)
|
||
if hits:
|
||
bad.append("仍残留裸 RETURN; 行 %d 处" % len(hits))
|
||
for token in ["def _procedure_body_return_errors", "_TRAILING_COMMENT",
|
||
"_ILLEGAL_RETURN", "_ILLEGAL_RETURN_PROBE",
|
||
"IF p_table = '%(table)s' AND IFNULL(@pbl_guard_ctx, '') <> '%(ctx)s' THEN"]:
|
||
if token not in text:
|
||
bad.append("缺少 %s" % token)
|
||
try:
|
||
ast.parse(text)
|
||
except SyntaxError as exc:
|
||
bad.append("语法错误: %s" % exc)
|
||
return bad
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--dry-run", action="store_true")
|
||
args = ap.parse_args()
|
||
with open(TARGET, encoding="utf-8") as f:
|
||
src = f.read()
|
||
print("[patch] 目标=%s 原始 %d 字节 / %d 行" % (TARGET, len(src), src.count("\n") + 1))
|
||
print("[patch] 原始文件裸 RETURN; 行数=%d（缺陷实况）" % len(_BARE_RETURN_RE.findall(src)))
|
||
try:
|
||
new = build_new(src)
|
||
except RuntimeError as exc:
|
||
print("[patch] DRY-RUN 失败: %s" % exc)
|
||
return 1
|
||
bad = check(new)
|
||
if bad:
|
||
print("[patch] dry-run 校验不通过，不落盘：")
|
||
for b in bad:
|
||
print(" - " + b)
|
||
return 1
|
||
print("[patch] dry-run 校验通过（%d 处替换，新文 %d 字节 / %d 行）"
|
||
% (len(REPLACEMENTS), len(new), new.count("\n") + 1))
|
||
if args.dry_run:
|
||
return 0
|
||
with open(TARGET, "w", encoding="utf-8") as f:
|
||
f.write(new)
|
||
# 写盘后二次读取校验（响应退回意见 #7：不能把 print 当成功标志）
|
||
with open(TARGET, encoding="utf-8") as f:
|
||
again = f.read()
|
||
bad = check(again)
|
||
if again != new:
|
||
bad.append("写盘后重读内容与预期不一致")
|
||
if bad:
|
||
print("[patch] 写盘后校验失败，回滚原始内容：")
|
||
for b in bad:
|
||
print(" - " + b)
|
||
with open(TARGET, "w", encoding="utf-8") as f:
|
||
f.write(src)
|
||
return 1
|
||
print("[patch] 写盘后重读校验通过：裸 RETURN; 计数=0、门禁函数/常量齐备、ast 解析 OK")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|