From df2be301b3a80480577a58da153fde1cbfb6f565 Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Fri, 18 Sep 2026 06:04:48 +0800 Subject: [PATCH] fix(cma): treasury grant step follows whitelist min set and converges - spec read from _cma_write_table_whitelist.json table_ops (single source of truth) - revoke non-whitelist table privs before grant (idempotent; 1141 handled as no-op) - add zero-DB --grant-plan mode; post-apply convergence assertion via information_schema - task: yanxue-20260918-treasury-grant-idempotent-fix --- backend/scripts/treasury_migrate.py | 244 ++++++++++++++++++++++++++-- 1 file changed, 226 insertions(+), 18 deletions(-) diff --git a/backend/scripts/treasury_migrate.py b/backend/scripts/treasury_migrate.py index 06c1837f..7559422d 100644 --- a/backend/scripts/treasury_migrate.py +++ b/backend/scripts/treasury_migrate.py @@ -8,8 +8,9 @@ G1 修正项 ：P0-1 up() 语句过滤器 / P0-2 计数口径 / P1-3 down() 部 用法: python3 treasury_migrate.py --parse-only # 零 DB：只解析 DDL，打印命中语句与期望计数（P0-1 复算证据） + python3 treasury_migrate.py --grant-plan # 零 DB：打印授权收敛语句计划（授权面幂等自证证据） python3 treasury_migrate.py --check # 只读预检：缺表/缺列/缺索引/种子行数（零写入，默认动作） - python3 treasury_migrate.py --apply # 建表 + 加列 + 索引 + 种子（幂等） + python3 treasury_migrate.py --apply # 建表 + 加列 + 索引 + 种子 + 授权收敛（幂等） python3 treasury_migrate.py --down # 回滚（先 dump 后删；有数据时要求 --yes） 幂等契约（P0-2 计数口径 = 新增对象数，不是执行语句数）: @@ -17,12 +18,28 @@ G1 修正项 ：P0-1 up() 语句过滤器 / P0-2 计数口径 / P1-3 down() 部 二跑 --apply 期望：created_new=0 / added_cols=0 / added_indexes=0 / 种子行数不变 指纹行：import_logs(batch=treasury-ddl-)，按 (import_type,batch) 先查后写， 不依赖表上是否有唯一键（P2-7：entity_id=1 为约定——迁移指纹属实体无关元数据） -边界：只动 6 张 treasury_* 新表 + cash_plans 加列/加索引；不碰 cash.py / 既有端点契约 / 其他模块表结构。 + +授权面契约（2026-09-18 F-1 根因侧：部署管道不再扩面，而是「按白名单最小面 + 收敛」）: + 旧形态（本文件 2026-09-18 前）= 每次 --apply 无条件 `GRANT SELECT,INSERT,UPDATE,DELETE` + 于 6 表 × 2 host = 12 条 → 白名单外的权限（表级 SELECT / balances·payments 的 DELETE / + hits 的 UPDATE,DELETE）每次部署回涨，基线巡检恒 BASELINE_MISMATCH。 + 新形态 = 授权集合取自白名单文件 table_ops（授权面唯一事实源），并对白名单外权限发 REVOKE： + · 语句计划 grant_plan() 是**纯函数**（只依赖 spec，不读现网）⇒ 连跑两次逐字相同； + · 重复执行结果一致（幂等）：白名单内 GRANT 幂等，白名单外 REVOKE 在收敛态报 1141 + （no such grant）→ 视为幂等空操作记账，不报错； + · 执行后回读 information_schema.TABLE_PRIVILEGES 断言逐表 == 白名单，不符则非零退出。 + 白名单文件路径：env `CMA_WRITE_WHITELIST` 优先，默认 /root/WB_WS/_reports/_cma_write_table_whitelist.json； + 文件不可用时退回 TREASURY_GRANT_SPEC_FALLBACK（须与白名单一致，否则 stderr 告警）；文件在位但 + 不合规（缺表裁定 / 不可解析 / 空集 / 含表级 SELECT）→ fail-closed，不猜不降级。 + +边界：只动 6 张 treasury_* 新表 + cash_plans 加列/加索引 + cma_api 授权收敛； + 不碰 cash.py / 既有端点契约 / 其他模块表结构 / 司库业务读写路径。 """ import argparse import hashlib import json import os +import re import subprocess import sys import tempfile @@ -60,6 +77,21 @@ CASH_PLAN_ALTER = """ALTER TABLE cash_plans DEFAULT_SOCKET = "/var/run/mysqld/mysqld.sock" FORBIDDEN_RUN_ACCOUNTS = {"cma_api"} GRANT_APP_ACCOUNT = "cma_api" +GRANT_HOSTS = ("localhost", "127.0.0.1") +# 授权面域：本迁移步只可能授出这 4 种（旧形态的全集）；收敛 = 该域内白名单外的项一律 REVOKE +GRANT_PRIV_DOMAIN = ("SELECT", "INSERT", "UPDATE", "DELETE") +# 白名单文件 = 授权面唯一事实源（后续白名单变更只改该文件一处） +WHITELIST_PATH = os.getenv( + "CMA_WRITE_WHITELIST", "/root/WB_WS/_reports/_cma_write_table_whitelist.json") +# 兜底常量：与白名单 table_ops 的 6 表裁定逐表一致（文件不可用时的退路；不一致时 stderr 告警） +TREASURY_GRANT_SPEC_FALLBACK = { + "treasury_accounts": ("INSERT", "UPDATE", "DELETE"), + "treasury_control_rules": ("INSERT", "UPDATE", "DELETE"), + "treasury_limits": ("INSERT", "UPDATE", "DELETE"), + "treasury_balances": ("INSERT", "UPDATE"), + "treasury_payments": ("INSERT", "UPDATE"), + "treasury_control_hits": ("INSERT",), +} def get_engine(): @@ -87,19 +119,161 @@ def get_engine(): pool_pre_ping=True) -def grant_app_account(conn): - """授权应用运行账号 cma_api 对 6 张 treasury 新表的 DML（幂等 GRANT，双 host）。 - 必须由特权连接执行；否则应用端点读不到/写不进新表（1142/1044）。""" - from app import database as db - dbn = (os.getenv("CMA_MIGRATE_DB_NAME") or getattr(db, "DB_NAME", None) or "cma") - granted = [] - for host in ("localhost", "127.0.0.1"): - for t in NEW_TABLES: - conn.execute(text( - f"GRANT SELECT, INSERT, UPDATE, DELETE ON `{dbn}`.`{t}` " - f"TO '{GRANT_APP_ACCOUNT}'@{host}")) - granted.append(f"{t}@{host}") - return granted +def _resolve_dbn(): + """库名：env CMA_MIGRATE_DB_NAME 优先，否则应用配置（app.database.DB_NAME），最后 'cma'。""" + name = (os.getenv("CMA_MIGRATE_DB_NAME") or "").strip() + if name: + return name + try: + from app import database as db + return getattr(db, "DB_NAME", None) or "cma" + except Exception: + return "cma" + + +def _parse_priv_text(raw): + """白名单 table_ops 值 → 权限集合。值形如 'INSERT,UPDATE,DELETE (中文说明（含嵌套括号）)'。""" + head = str(raw) + for sep in ("(", "（"): + i = head.find(sep) + if i >= 0: + head = head[:i] + return {p.strip().upper() for p in head.split(",") if p.strip()} + + +def load_grant_spec(): + """读白名单 table_ops 的 6 张 treasury 表裁定 → (spec, source, note)。 + + spec : {table: (priv, ...)}，权限按 GRANT_PRIV_DOMAIN 顺序归一（确定性） + source : 'whitelist' | 'fallback-constant' + note : 常量兜底与白名单不一致时的告警文本（否则 None） + fail-closed：文件在位但不可解析 / 缺表裁定 / 空集 / 含表级 SELECT → SystemExit，不猜不降级。 + """ + fallback = {t: tuple(TREASURY_GRANT_SPEC_FALLBACK[t]) for t in NEW_TABLES} + if not os.path.exists(WHITELIST_PATH): + return fallback, "fallback-constant", None + try: + with open(WHITELIST_PATH, encoding="utf-8") as fh: + ops = (json.load(fh) or {}).get("table_ops") or {} + except Exception as exc: + raise SystemExit(f"[FAIL] 白名单不可解析 {WHITELIST_PATH}: {exc}") + spec = {} + for t in NEW_TABLES: + raw = ops.get(t) + if raw is None: + raise SystemExit(f"[FAIL] 白名单 table_ops 缺 {t} 裁定（{WHITELIST_PATH}）") + privs = tuple(p for p in GRANT_PRIV_DOMAIN if p in _parse_priv_text(raw)) + if not privs: + raise SystemExit(f"[FAIL] 白名单 {t} 授权集为空（{raw!r}）") + if "SELECT" in privs: + raise SystemExit( + f"[FAIL] 白名单 {t} 含表级 SELECT —— 读面由库级 SELECT ON cma.* 覆盖，" + "表级 SELECT 正是本单要收敛的冗余；如需改基线口径请回报调度方另单") + spec[t] = privs + note = None + if spec != fallback: + note = (f"兜底常量与白名单不一致（以白名单为准）: fallback={fallback} whitelist={spec}") + return spec, "whitelist", note + + +def grant_plan(spec, dbn): + """授权收敛语句计划（纯函数：只依赖 spec/dbn，不读现网）⇒ 连跑两次逐字相同。 + + 每 (host, 表)：先 REVOKE 白名单外权限（部署成为「收敛」通道），再 GRANT 白名单集合。 + 收敛态下 REVOKE 会报 1141（no such grant）→ 执行侧记为幂等空操作，非错误。 + """ + stmts = [] + for host in GRANT_HOSTS: + for t in sorted(NEW_TABLES): + wanted = tuple(p for p in GRANT_PRIV_DOMAIN if p in spec[t]) + extras = tuple(p for p in GRANT_PRIV_DOMAIN if p not in wanted) + if extras: + stmts.append( + f"REVOKE {', '.join(extras)} ON `{dbn}`.`{t}` FROM '{GRANT_APP_ACCOUNT}'@{host}") + stmts.append( + f"GRANT {', '.join(wanted)} ON `{dbn}`.`{t}` TO '{GRANT_APP_ACCOUNT}'@{host}") + return stmts + + +def _is_nonexistent_grant(exc) -> bool: + """MySQL 1141/1147（no such grant）：REVOKE 未持有权限 —— 收敛态的正常返回。""" + orig = getattr(exc, "orig", exc) + args = getattr(orig, "args", ()) or () + if args and args[0] in (1141, 1147): + return True + return "no such grant" in str(exc).lower() + + +_GRANT_STMT_RE = re.compile( + r"^(?PREVOKE|GRANT)\s+(?P.+?)\s+ON\s+`(?P[^`]+)`\.`(?P[^`]+)`" + r"\s+(?:FROM|TO)\s+'(?P[^']+)'@(?P\S+)$") + + +def _stmt_entry(stmt: str) -> dict: + """授权语句 → {verb, privs, db, table, user, host}（解析失败即抛，不静默）。""" + m = _GRANT_STMT_RE.match(stmt.strip()) + if not m: + raise RuntimeError(f"授权语句形态非预期，无法解析: {stmt!r}") + d = m.groupdict() + d["privs"] = ", ".join(p.strip() for p in d["privs"].split(",")) + return d + + +def _current_table_privs(conn, dbn): + """只读回查 information_schema.TABLE_PRIVILEGES：{(host, table): {priv}}（仅 cma_api）。""" + rows = conn.execute(text( + "SELECT GRANTEE, TABLE_NAME, PRIVILEGE_TYPE FROM information_schema.TABLE_PRIVILEGES " + "WHERE TABLE_SCHEMA = :d"), {"d": dbn}).fetchall() + out = {} + for grantee, table, priv in rows: + if table not in NEW_TABLES: + continue + user, _sep, host = str(grantee).rpartition("@") + if user.strip().strip("'") != GRANT_APP_ACCOUNT: + continue + out.setdefault((host.strip().strip("'"), table), set()).add(str(priv).upper()) + return out + + +def verify_grant_sync(conn, spec, dbn): + """收敛断言：逐 (host, 表) 现网表级权限集合 == 白名单集合；返回差异清单（空 = 已收敛）。""" + current = _current_table_privs(conn, dbn) + bad = [] + for host in GRANT_HOSTS: + for t in sorted(NEW_TABLES): + have = current.get((host, t), set()) + want = set(spec[t]) + if have != want: + bad.append(f"{t}@{host}: 现持={sorted(have)} 期望={sorted(want)}") + return bad + + +def apply_grant_sync(conn, spec, source): + """执行授权收敛（按白名单最小面 + 幂等）：REVOKE 白名单外 + GRANT 白名单，回读断言。""" + dbn = _resolve_dbn() + granted, revoked, absent = [], [], [] + for stmt in grant_plan(spec, dbn): + try: + conn.execute(text(stmt)) + except Exception as exc: + if stmt.startswith("REVOKE") and _is_nonexistent_grant(exc): + absent.append(_stmt_entry(stmt)) + continue + raise + (granted if stmt.startswith("GRANT") else revoked).append(_stmt_entry(stmt)) + conn.commit() + mismatch = verify_grant_sync(conn, spec, dbn) + return { + "spec_source": source, + "spec": {t: ",".join(spec[t]) for t in sorted(spec)}, + "granted": [f"{e['table']}@{e['host']}" for e in granted], + "revoked": [f"{e['table']}@{e['host']} ({e['privs']}) [REVOKE]" for e in revoked], + "revoke_absent": [f"{e['table']}@{e['host']} ({e['privs']}) [REVOKE-ABSENT]" + for e in absent], + "statements": len(granted) + len(revoked) + len(absent), + "converged": not mismatch, + "converged_mismatch": mismatch, + } # ───────────────────────────────────────────────────────────── @@ -279,8 +453,13 @@ def up(conn): if CASH_PLAN_NEW_INDEX not in before_idx: conn.execute(text(p["index"])) - # ③b 授权应用账号 cma_api 对本 6 表 DML（幂等 GRANT，须特权连接） - granted = grant_app_account(conn) + # ③b 授权收敛（按白名单最小面 + 幂等，须特权连接）：REVOKE 白名单外 + GRANT 白名单 + # 旧形态为无条件 12 条全量 GRANT（含表级 SELECT 等白名单外权限）→ 每次部署扩面， + # 基线巡检恒 BASELINE_MISMATCH（2026-09-18 F-1）。现改为收敛：重复执行结果不变。 + spec, spec_source, spec_note = load_grant_spec() + if spec_note: + print("[warn] " + spec_note, file=sys.stderr) + grant_sync = apply_grant_sync(conn, spec, spec_source) # ④ 种子（ON DUPLICATE KEY UPDATE，数据级幂等） for stmt in p["seeds"]: @@ -315,10 +494,14 @@ def up(conn): "limits_rows": counts["limits_rows"], "import_logs_row": action, "batch": batch, - "granted": granted, + "granted": grant_sync["granted"], + "grant_sync": grant_sync, "missing_tables": post["missing_tables"], "cash_plans_rows": post["cash_plans_rows"], }, ensure_ascii=False)) + if not grant_sync["converged"]: + print(f"❌ 授权收敛未达成 {grant_sync['converged_mismatch']}（非零退出）", file=sys.stderr) + return 6 return 0 @@ -380,6 +563,28 @@ def down(conn): # 入口 # ───────────────────────────────────────────────────────────── +def grant_plan_only(): + """零 DB：打印授权收敛语句计划（幂等自证证据：连跑两次逐字相同）。""" + spec, source, note = load_grant_spec() + dbn = _resolve_dbn() + stmts = grant_plan(spec, dbn) + out = { + "mode": "grant-plan", + "whitelist_path": WHITELIST_PATH, + "whitelist_present": os.path.exists(WHITELIST_PATH), + "spec_source": source, + "spec_note": note, + "spec": {t: ",".join(spec[t]) for t in sorted(spec)}, + "hosts": list(GRANT_HOSTS), + "db_name": dbn, + "statement_count": len(stmts), + "statements": stmts, + "statements_md5": hashlib.md5("\n".join(stmts).encode("utf-8")).hexdigest(), + } + print(json.dumps(out, ensure_ascii=False, indent=2)) + return 0 + + def parse_only(): """零 DB 的 DDL 解析复算（P0-1 证据）：命中 6 条 CREATE + 表名清单 + 种子语句。""" p = parse_ddl() @@ -407,6 +612,7 @@ def main(): ap = argparse.ArgumentParser() g = ap.add_mutually_exclusive_group() g.add_argument("--parse-only", action="store_true", help="零 DB：只解析 DDL（P0-1 复算证据）") + g.add_argument("--grant-plan", action="store_true", help="零 DB：打印授权收敛语句计划") g.add_argument("--check", action="store_true", help="只读预检（默认）") g.add_argument("--apply", action="store_true") g.add_argument("--down", action="store_true") @@ -415,6 +621,8 @@ def main(): if args.parse_only: return parse_only() + if args.grant_plan: + return grant_plan_only() engine = get_engine() with engine.connect() as conn: