118 lines
4.7 KiB
Python
Executable File
118 lines
4.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""S3 触发真实性门禁探针：在沙箱库中对 BEFORE INSERT trigger 做 SHOW / CREATE(历史适配) / DROP。
|
||
|
||
背景：git 80156fd 时期的 harness 曾在沙箱侧建 `trg_pbl_evidence_id` BEFORE INSERT
|
||
trigger 来补 pbl_evidence.id（掩盖 collector INSERT 不带 id）。DISC.1 闭环后该适配已
|
||
从 harness 源码删除，id 由应用层 pbl_evidence.pk.gen_pk 生成。本脚本用于**取证**：
|
||
先按历史 DDL 复原该 trigger（证明"它存在过、且可被 drop"），再 DROP 之，随后重跑
|
||
harness 仍 ALL PASS，从而证明走的是真实链路而非补丁生效。
|
||
|
||
凭据唯一事实源 = <workspace>/projects/pbls/env/test.json 的 db.sandbox 段
|
||
（scope=sandbox_only），口令一律不打印；异常消息经脱敏后输出。
|
||
|
||
用法::
|
||
|
||
python3 tests/s3_trigger_probe.py show
|
||
python3 tests/s3_trigger_probe.py create
|
||
python3 tests/s3_trigger_probe.py drop
|
||
python3 tests/s3_trigger_probe.py show --schema pbl_m5a_u7rb
|
||
|
||
退出码：0 = 动作成功；非 0 = 失败（供 chain 脚本 set -e 门禁使用）。
|
||
"""
|
||
import argparse
|
||
import json
|
||
import pathlib
|
||
import sys
|
||
|
||
import pymysql
|
||
|
||
TESTS_DIR = pathlib.Path(__file__).resolve().parent
|
||
REPO_ROOT = TESTS_DIR.parent # modules/pbl_evidence
|
||
WORKSPACE_ROOT = REPO_ROOT.parent.parent # 机构工作空间
|
||
ENV_FILE = WORKSPACE_ROOT / "projects" / "pbls" / "env" / "test.json"
|
||
|
||
TRIGGER_NAME = "trg_pbl_evidence_id"
|
||
# 历史适配 trigger 的原始 DDL（逐字来自 git 80156fd 时期 harness）
|
||
HISTORIC_DDL = (
|
||
"CREATE TRIGGER `%s` BEFORE INSERT ON `pbl_evidence` FOR EACH ROW\n"
|
||
"BEGIN\n"
|
||
" IF NEW.`id` IS NULL OR NEW.`id` = '' THEN\n"
|
||
" SET NEW.`id` = REPLACE(UUID(), '-', '');\n"
|
||
" END IF;\n"
|
||
"END" % TRIGGER_NAME
|
||
)
|
||
|
||
|
||
def load_sandbox():
|
||
"""读沙箱凭据；缺段或 scope 非 sandbox_only 直接 fail-fast（不猜、不回退业务库）。"""
|
||
cfg = json.loads(ENV_FILE.read_text(encoding="utf-8"))
|
||
sb = cfg["db"]["sandbox"]
|
||
if sb.get("scope") != "sandbox_only":
|
||
raise SystemExit("refuse non-sandbox scope: %r" % sb.get("scope"))
|
||
return sb
|
||
|
||
|
||
def connect(sb, schema):
|
||
return pymysql.connect(
|
||
host=sb["host"], port=int(sb["port"]), user=sb["user"],
|
||
password=sb["password"], charset="utf8mb4", database=schema,
|
||
cursorclass=pymysql.cursors.DictCursor, autocommit=True)
|
||
|
||
|
||
def mask(sb, text):
|
||
out = str(text)
|
||
for secret in (sb.get("password"), sb.get("user")):
|
||
if secret:
|
||
out = out.replace(str(secret), "******")
|
||
return out
|
||
|
||
|
||
def main(argv=None):
|
||
ap = argparse.ArgumentParser(
|
||
description="S3 取证：沙箱库 pbl_evidence 上 BEFORE INSERT trigger 的 SHOW/CREATE/DROP")
|
||
ap.add_argument("cmd", choices=["show", "create", "drop"],
|
||
help="show=SHOW TRIGGERS 原样回显; create=按历史 DDL 复原适配 trigger; "
|
||
"drop=DROP TRIGGER IF EXISTS")
|
||
ap.add_argument("--schema", default=None,
|
||
help="沙箱 schema（缺省取 env/test.json 的 db.sandbox.sandbox_schema）")
|
||
ns = ap.parse_args(argv)
|
||
|
||
sb = load_sandbox()
|
||
schema = ns.schema or sb["sandbox_schema"]
|
||
conn = connect(sb, schema)
|
||
rc = 0
|
||
try:
|
||
with conn.cursor() as cur:
|
||
if ns.cmd == "show":
|
||
cur.execute("SHOW TRIGGERS")
|
||
rows = list(cur.fetchall() or [])
|
||
print("SHOW TRIGGERS (schema=%s) -> %d 条" % (schema, len(rows)))
|
||
for r in rows:
|
||
print(json.dumps({k: mask(sb, v) for k, v in r.items()},
|
||
ensure_ascii=False))
|
||
elif ns.cmd == "create":
|
||
cur.execute("DROP TRIGGER IF EXISTS `%s`" % TRIGGER_NAME)
|
||
cur.execute(HISTORIC_DDL)
|
||
print("SQL> " + HISTORIC_DDL.replace("\n", " "))
|
||
print("CREATE TRIGGER 执行成功（历史适配 trigger，逐字来自 git 80156fd 时期 harness）")
|
||
else:
|
||
print("SQL> DROP TRIGGER IF EXISTS `%s`" % TRIGGER_NAME)
|
||
cur.execute("DROP TRIGGER IF EXISTS `%s`" % TRIGGER_NAME)
|
||
cur.execute("SHOW TRIGGERS")
|
||
left = list(cur.fetchall() or [])
|
||
print("DROP TRIGGER 执行完成；剩余 trigger = %d 条" % len(left))
|
||
if left:
|
||
print("FAIL: DROP 后仍存在 trigger，无法证明无补丁链路")
|
||
rc = 1
|
||
except Exception as exc: # noqa: BLE001
|
||
print("ERROR: %s: %s" % (type(exc).__name__, mask(sb, exc)))
|
||
rc = 1
|
||
finally:
|
||
conn.close()
|
||
return rc
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|