#!/usr/bin/env python3 # -*- coding: utf-8 -*- """grant_surface_sync.py —— 新表授权面收敛（部署链步） 设计口径（2026-09-28，项目Bot；派单件 yanxue-20260928-newtable-grant-surface-automation） ------------------------------------------------------------------------------ 问题：新表的表级授权不在任何自动步骤内。`db_migrate.py --apply` 只建表（create_all）， `treasury_migrate.py --apply` 的授权收敛步的 `NEW_TABLES` 硬编码 6 张 treasury 表 ⇒ 新建业务表「表在、写权缺」→ 端点 500 / 1142 → 人工派单 + 铸审批 ID + 手工 GRANT。 实测同形已 3 例：subject_balances(09-21) / product_inventory(09-22) / quality_rule_config(09-28)。 本件（方案 A 的机制实现）三条不变量： 1. 只增不减：只发 GRANT，**从不** REVOKE；白名单外已存在的权限只报「漂移」，由基线巡检+人工处置。 2. fail-closed：① 本次部署新建的表若不在白名单 write_tables（= 无 table_ops 裁定）→ **零写入**中止； ② 白名单不可解析 / 缺 table_ops 裁定 / table_ops 含表级 SELECT → 中止； ③ 缺「本次新建表」锚 → apply 模式中止（不猜、不静默跳过覆盖检查）。 3. 集合键只出自白名单事实源（write_tables / table_ops），不从现网反推、不从自声明文本取。 用法 ---- python3 grant_surface_sync.py --check [--new-tables a,b] # 只读计划（零写入） python3 grant_surface_sync.py --apply --new-tables a,b # 收敛（GRANT 白名单集合） python3 grant_surface_sync.py --apply --anchor-file # 收敛（锚 = db_migrate 指针件） python3 grant_surface_sync.py --audit # 只读双向差集巡检（方案 B 核心） 锚（Δ1 / Δ3 / R1） ------------------ * 「缺锚」只认三种：`--new-tables` 未给 / `--anchor-file` 缺件或坏件 / 锚件 `status != applied`； **`created_tables=[]` = 合法零集**（幂等发布不中止，仍按白名单收敛）。 * R1-a 同源自证：锚件须携带本次 journal 引用（`journal`）与 `finished_at`；收敛步校验该 journal 在册、`status=applied`、`finished_at` 与锚件一致、且是目录内**最新**的 `-applied` 件； 任一不符 ⇒ rc=2、零写入（挡住「指针写失败但仍读到上一轮旧锚」的静默路径）。 * R1-b 未收敛锚不回退（累积）：`db_migrate.write_anchor_pointer()` 写新指针时，若旧指针 `status=applied` 且无 `converged_at`，则 `created_tables` = 旧集合 ∪ 本次 created。 * R1-c 收敛回写：本步 `--apply` rc=0 时把 `converged_at`/语句数/`statements_md5` 原子回写锚件（幂等）。 连接（特权通道，与 db_migrate.py / treasury_migrate.py 同口径；Δ2） env CMA_MIGRATE_DB_USER/PASS/HOST/PORT 优先（专用迁移账号，TCP） 或 --host/--port/--user/--password；CLI 显式连接旗标优先于 env（沙箱夹具用） 否则本机 unix socket + root（/var/run/mysqld/mysqld.sock） [WARN] 应用运行账号 `cma_api` **不得**作迁移身份（无 GRANT 权限）—— 显式拒绝。 退出码 0 成功 / 1 回读不符(apply) / 2 用法或缺锚或锚不可用 / 3 新表缺裁定(fail-closed, 零写入) 4 白名单或连接不可用 / 5 越权漂移且 --drift-policy fail """ import argparse import hashlib import json import os import sys from datetime import datetime import pymysql DEFAULT_WHITELIST = os.getenv( "CMA_WRITE_WHITELIST", "/root/WB_WS/_reports/_cma_write_table_whitelist.json") DEFAULT_SOCKET = os.getenv("CMA_MIGRATE_DB_SOCKET", "/var/run/mysqld/mysqld.sock") DEFAULT_DB = os.getenv("CMA_MIGRATE_DB_NAME", "cma") APP_ACCOUNT = os.getenv("CMA_APP_ACCOUNT", "cma_api") HOSTS = ("localhost", "127.0.0.1") # 授权面域：本步只可能涉及这 4 种表级 DML 权限（读面由库级 SELECT ON .* 覆盖） PRIV_DOMAIN = ("INSERT", "UPDATE", "DELETE") # Δ2：应用运行账号禁止作迁移身份（与 db_migrate.py:FORBIDDEN_RUN_ACCOUNTS 同口径） FORBIDDEN_RUN_ACCOUNTS = {"cma_api"} ANCHOR_STATUS_APPLIED = "applied" def md5f(path): h = hashlib.md5() with open(path, "rb") as fh: for chunk in iter(lambda: fh.read(65536), b""): h.update(chunk) return h.hexdigest() def parse_privs(raw): """table_ops 值 -> 权限集合。值形如 'INSERT,UPDATE (...中文说明（含嵌套括号）)'。""" 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_whitelist(path): """读白名单 -> (write_tables:set, ops_fn, meta)。fail-closed：不可解析即退出。""" if not os.path.exists(path): raise SystemExit("[FAIL] 白名单不存在：%s" % path) try: with open(path, encoding="utf-8") as fh: doc = json.load(fh) except Exception as exc: raise SystemExit("[FAIL] 白名单不可解析 %s: %s" % (path, exc)) wt = doc.get("write_tables") or [] ops = doc.get("table_ops") or {} if not wt: raise SystemExit("[FAIL] 白名单 write_tables 为空（fail-closed，不猜）") default_raw = ops.get("_default", "INSERT,UPDATE,DELETE") return set(wt), ops, {"path": path, "md5": md5f(path), "default_ops": default_raw, "write_tables_count": len(wt)} def wanted_privs(table, ops, default_raw): raw = ops.get(table, default_raw) privs = parse_privs(raw) bad = sorted(p for p in privs if p not in PRIV_DOMAIN) if bad: raise SystemExit( "[FAIL] 白名单 %s 授权集含域外/冗余项 %s —— 读面由库级 SELECT 覆盖，" "表级 SELECT 正是要收敛的冗余；缺裁定请回报调度方另单" % (table, bad)) return tuple(p for p in PRIV_DOMAIN if p in privs) def resolve_identity(args): """Δ2 连接身份解析 -> (kind, kwargs, desc)。与 db_migrate.py::build_engine 同口径： env CMA_MIGRATE_DB_USER（专用迁移账号，TCP）优先；未设则本机 unix socket + root。 CLI 显式连接旗标（--host / 非默认 --socket）优先于 env —— 保留沙箱夹具既有用法。 `cma_api` 等应用运行账号一律拒绝（无 GRANT 权限 = 本机制不成立）。""" def _reject(user): if user in FORBIDDEN_RUN_ACCOUNTS: raise SystemExit( "[FAIL] 连接身份 %s 是**应用运行账号**（无 GRANT 权限）——\n" " 授权面收敛必须走特权迁移身份（CMA_MIGRATE_DB_USER 或本机 root socket）。" "已拒绝执行。" % user) cli_explicit = bool(args.host) or bool(args.socket and args.socket != DEFAULT_SOCKET) if cli_explicit: user = (args.user or "root").strip() _reject(user) if args.host: desc = "tcp(%s@%s:%s/%s)" % (user, args.host, args.port, args.db) return "tcp", {"host": args.host, "port": args.port, "user": user, "password": args.password, "database": args.db}, desc desc = "socket(%s@%s/%s)" % (user, args.socket, args.db) return "socket", {"unix_socket": args.socket, "user": user, "password": args.password, "database": args.db}, desc env_user = os.getenv("CMA_MIGRATE_DB_USER", "").strip() if env_user: _reject(env_user) host = os.getenv("CMA_MIGRATE_DB_HOST", "127.0.0.1") port = int(os.getenv("CMA_MIGRATE_DB_PORT", "3306")) desc = "tcp(%s@%s:%s/%s)" % (env_user, host, port, args.db) return "tcp", {"host": host, "port": port, "user": env_user, "password": os.getenv("CMA_MIGRATE_DB_PASS", ""), "database": args.db}, desc sock = args.socket or os.getenv("CMA_MIGRATE_DB_SOCKET") or DEFAULT_SOCKET user = (args.user or "root").strip() _reject(user) desc = "socket(%s@%s/%s)" % (user, sock, args.db) return "socket", {"unix_socket": sock, "user": user, "password": args.password, "database": args.db}, desc def connect(args): kind, kwargs, desc = resolve_identity(args) return pymysql.connect(autocommit=True, charset="utf8mb4", **kwargs), desc def load_anchor_pointer(path): """Δ3 + R1-a：读锚指针件（`db_migrate.py --apply` 原子写）-> (created_tables, meta)。 任一情形不符 ⇒ SystemExit（调用方转 rc=2、零写入）： ① 缺件 / 非 JSON / 结构非法；② status != applied； ③ R1-a 同源自证不符（缺 journal 引用 / journal 不在册 / 该 journal status != applied / finished_at 不一致 / 该 journal 不是目录内最新 -applied 件）。""" if not path or not os.path.exists(path): raise SystemExit("[FAIL] 锚件不存在：%s（apply 缺锚 ⇒ fail-closed，零写入）" % path) try: with open(path, encoding="utf-8") as fh: doc = json.load(fh) except Exception as exc: raise SystemExit("[FAIL] 锚件不可解析 %s: %s ⇒ 零写入" % (path, exc)) if not isinstance(doc, dict): raise SystemExit("[FAIL] 锚件结构非法（非 JSON 对象）：%s ⇒ 零写入" % path) status = str(doc.get("status") or "") if status != ANCHOR_STATUS_APPLIED: raise SystemExit("[FAIL] 锚件 status=%s != applied（本次 apply 未成功）⇒ 零写入" % (status or "")) jname = str(doc.get("journal") or "") if not jname: raise SystemExit("[FAIL] R1-a：锚件缺 journal 引用（无法自证出自本次 apply）⇒ 零写入") jdir = os.path.dirname(os.path.abspath(path)) jpath = os.path.join(jdir, os.path.basename(jname)) if not os.path.exists(jpath): raise SystemExit("[FAIL] R1-a：锚件引用的 journal 不在册 %s ⇒ 零写入" % jname) try: with open(jpath, encoding="utf-8") as fh: jdoc = json.load(fh) except Exception as exc: raise SystemExit("[FAIL] R1-a：journal 不可解析 %s: %s ⇒ 零写入" % (jname, exc)) if str(jdoc.get("status") or "") != ANCHOR_STATUS_APPLIED: raise SystemExit("[FAIL] R1-a：journal %s status=%s != applied ⇒ 零写入" % (jname, jdoc.get("status"))) if doc.get("finished_at") != jdoc.get("finished_at"): raise SystemExit("[FAIL] R1-a：锚件 finished_at=%s != journal %s 的 %s ⇒ 零写入" % (doc.get("finished_at"), jname, jdoc.get("finished_at"))) applied = sorted(f for f in os.listdir(jdir) if f.endswith("-applied.json")) if not applied or applied[-1] != os.path.basename(jname): raise SystemExit("[FAIL] R1-a：锚件引用的 journal %s 不是目录内最新 apply 记录" "（最新=%s）⇒ 锚非出自本次 apply，零写入" % (os.path.basename(jname), applied[-1] if applied else "<无>")) created = list(doc.get("created_tables") or []) meta = {"path": path, "journal": os.path.basename(jname), "status": status, "finished_at": doc.get("finished_at"), "created_tables": created, "carried_over": bool(doc.get("carried_over")), "carried_from": doc.get("carried_from"), "converged_at": doc.get("converged_at")} return created, meta def atomic_write_json(path, doc): """原子替换写 JSON（同目录 tmp + os.replace；模式/属组跟随原件的现状）。""" tmp = "%s.tmp.%d" % (path, os.getpid()) with open(tmp, "w", encoding="utf-8") as fh: json.dump(doc, fh, ensure_ascii=False, indent=2) fh.write("\n") try: if os.path.exists(path): st = os.stat(path) os.chmod(tmp, st.st_mode & 0o777) os.chown(tmp, st.st_uid, st.st_gid) else: os.chmod(tmp, 0o644) except Exception: pass os.replace(tmp, path) def write_back_converged(anchor_path, out): """R1-c：收敛步 apply rc=0 时把 converged_at（+ 语句数 / statements_md5）原子回写锚件（幂等）。 rc!=0 不调用。回写不改动 created_tables / status / journal —— 只加收敛态四键。""" try: with open(anchor_path, encoding="utf-8") as fh: doc = json.load(fh) except Exception as exc: return {"ok": False, "error": "%s: %s" % (type(exc).__name__, exc)} stmts = list(out.get("executed") or []) doc["converged_at"] = datetime.now().isoformat(timespec="seconds") doc["converged_statements"] = len(stmts) doc["converged_statements_md5"] = hashlib.md5("\n".join(stmts).encode("utf-8")).hexdigest() doc["converged_by"] = "grant_surface_sync.py" try: atomic_write_json(anchor_path, doc) except Exception as exc: return {"ok": False, "error": "%s: %s" % (type(exc).__name__, exc)} return {"ok": True, "path": anchor_path, "converged_at": doc["converged_at"], "statements": doc["converged_statements"], "statements_md5": doc["converged_statements_md5"]} def db_tables(cur, db): cur.execute("""SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA=%s AND TABLE_TYPE='BASE TABLE'""", (db,)) return {r[0] for r in cur.fetchall()} def table_privs(cur, db, tables): """{(host, table): {priv}} —— 只枚举 cma_api 的双 host（带引号字面量，LIKE 恒 0 行）。""" grantees = ["'%s'@'%s'" % (APP_ACCOUNT, h) for h in HOSTS] out = {} cur.execute("""SELECT GRANTEE, TABLE_NAME, PRIVILEGE_TYPE FROM information_schema.TABLE_PRIVILEGES WHERE TABLE_SCHEMA=%s AND GRANTEE IN (%s, %s)""", (db, grantees[0], grantees[1])) for grantee, table, priv in cur.fetchall(): _u, _s, host = str(grantee).rpartition("@") out.setdefault((host.strip().strip("'"), str(table)), set()).add(str(priv).upper()) return out def build_plan(dbn, tables, wt, ops, default_raw, cur_privs): """纯函数（只依赖白名单 + 现网只读快照）-> (statements, matrix, drift)。 statements : [GRANT ...] —— 逐 host/表白名单集合减去现持集合（只增不减，无 REVOKE） matrix : 逐 (host,表) 的 wanted/have/missing/extras drift : 白名单外已持权限（只报，不 REVOKE） """ stmts, matrix, drift = [], [], [] for host in HOSTS: for t in sorted(wt): if t not in tables: continue want = wanted_privs(t, ops, default_raw) have = cur_privs.get((host, t), set()) missing = tuple(p for p in want if p not in have) extras = tuple(sorted(have - set(want))) matrix.append({"host": host, "table": t, "wanted": list(want), "have": sorted(have), "missing": list(missing), "extras": list(extras)}) if extras: drift.append("DRIFT %s@%s: 白名单外已持 %s（本步不 REVOKE，交基线巡检/ops 处置）" % (t, host, list(extras))) if missing: stmts.append("GRANT %s ON `%s`.`%s` TO '%s'@'%s'" % (", ".join(missing), dbn, t, APP_ACCOUNT, host)) return stmts, matrix, drift def main(): ap = argparse.ArgumentParser() g = ap.add_mutually_exclusive_group() g.add_argument("--check", action="store_true", help="只读计划（默认，零写入）") g.add_argument("--apply", action="store_true", help="执行收敛（只 GRANT）") g.add_argument("--audit", action="store_true", help="只读双向差集巡检（方案 B）") ap.add_argument("--whitelist", default=DEFAULT_WHITELIST) ap.add_argument("--db", default=DEFAULT_DB) ap.add_argument("--new-tables", default=None, help="本次部署新建表（逗号分隔）；= create_all 的 created_tables") ap.add_argument("--anchor-file", default=None, help="锚指针件（db_migrate --apply 写的 journal/last-anchor.json）；" "与 --new-tables 二选一") ap.add_argument("--drift-policy", choices=("warn", "fail"), default="warn") ap.add_argument("--socket", default=DEFAULT_SOCKET) ap.add_argument("--host", default=None) ap.add_argument("--port", type=int, default=3306) ap.add_argument("--user", default="root") ap.add_argument("--password", default="") args = ap.parse_args() out = {"script": "grant_surface_sync.py", "mode": "apply" if args.apply else ("audit" if args.audit else "check"), "db": args.db, "hosts": list(HOSTS), "app_account": APP_ACCOUNT, "drift_policy": args.drift_policy} try: wt, ops, wl_meta = load_whitelist(args.whitelist) except SystemExit as exc: print(json.dumps(out, ensure_ascii=False, indent=2)) print(str(exc), file=sys.stderr) return 4 out["whitelist"] = wl_meta try: conn, conn_desc = connect(args) except SystemExit as exc: # Δ2：非法身份（应用运行账号）⇒ 用法级 fail-closed print(json.dumps(out, ensure_ascii=False, indent=2)) print(str(exc), file=sys.stderr) return 2 except Exception as exc: print(json.dumps(out, ensure_ascii=False, indent=2)) print("[FAIL] 连接失败: %s" % exc, file=sys.stderr) return 4 out["connection"] = conn_desc try: cur = conn.cursor() tables = db_tables(cur, args.db) cur_privs = table_privs(cur, args.db, tables) # ── 巡检模式（方案 B 核心）：只读双向差集，零写入 ── if args.audit: in_db_not_wl = sorted(t for t in tables if t not in wt) in_wl_not_db = sorted(t for t in wt if t not in tables) wl_no_grant = sorted("%s@%s" % (t, h) for t in sorted(wt) if t in tables for h in HOSTS if not cur_privs.get((h, t))) out.update({"audit": {"db_tables": len(tables), "whitelist_tables": len(wt), "db_tables_not_in_whitelist": in_db_not_wl, "whitelist_tables_absent_in_db": in_wl_not_db, "whitelist_tables_without_grant": wl_no_grant}, "note": "方向一（库内表⊄白名单）需代码证据判读：只读表合法无裁定，" "故只报不判；方向二（白名单⊄实授权）即待授权清单，判 GAP"}) # 判据只取「授权面缺口」：白名单表缺授权 / 白名单表缺表。 # 库内多出的表只报不判（只读表本就无须裁定，判它等于自造假缺陷）。 grant_gap = bool(in_wl_not_db or wl_no_grant) out["verdict"] = ("GAP" if grant_gap else ("NO_GRANT_GAP_DB_TABLES_NOT_WHITELISTED" if in_db_not_wl else "NO_GAP")) print(json.dumps(out, ensure_ascii=False, indent=2)) return 1 if grant_gap else 0 # ── 锚解析（Δ1/Δ3）：缺锚只认「未给/缺件坏件/status!=applied」；空集 = 合法零集 ── anchor_meta, new_tables = None, [] if args.new_tables is not None: # 显式给了（含空串 = 合法零集） new_tables = [t.strip() for t in args.new_tables.split(",") if t.strip()] anchor_source = "--new-tables" elif args.anchor_file: try: new_tables, anchor_meta = load_anchor_pointer(args.anchor_file) # R1-a except SystemExit as exc: out["anchor_source"] = "--anchor-file" out["verdict"] = "FAIL_CLOSED_ANCHOR_UNUSABLE" print(json.dumps(out, ensure_ascii=False, indent=2)) print(str(exc), file=sys.stderr) return 2 anchor_source = "--anchor-file:%s" % anchor_meta["journal"] else: anchor_source = "none" out["anchor_source"] = anchor_source # Δ4 coverage = {"anchor": (args.new_tables if args.new_tables is not None else (anchor_meta or {}).get("created_tables")), "anchor_source": anchor_source, "new_tables": new_tables, "anchor_meta": anchor_meta, "missing_caliber": []} if args.apply and anchor_source == "none": out["coverage"] = coverage print(json.dumps(out, ensure_ascii=False, indent=2)) print("[FAIL] apply 缺「本次部署新建表」锚（--new-tables / --anchor-file）" "—— fail-closed，零写入", file=sys.stderr) return 2 coverage["missing_caliber"] = sorted(t for t in new_tables if t not in wt) out["coverage"] = coverage stmts, matrix, drift = build_plan(args.db, tables, wt, ops, wl_meta["default_ops"], cur_privs) out["plan"] = {"statements": stmts, "statement_count": len(stmts), "matrix": matrix, "drift": drift, "tables_in_whitelist_present_in_db": sorted(t for t in wt if t in tables), "whitelist_tables_absent_in_db": sorted(t for t in wt if t not in tables)} if coverage["missing_caliber"]: out["verdict"] = "FAIL_CLOSED_NEW_TABLE_WITHOUT_CALIBER" print(json.dumps(out, ensure_ascii=False, indent=2)) print("[FAIL] 本次部署新建表缺白名单裁定 %s —— 零写入中止（部署链应据此非零退出）" % coverage["missing_caliber"], file=sys.stderr) return 3 if args.drift_policy == "fail" and drift: out["verdict"] = "DRIFT" print(json.dumps(out, ensure_ascii=False, indent=2)) print("[FAIL] 检测到白名单外已持权限（drift-policy=fail）：%s" % drift, file=sys.stderr) return 5 for d in drift: print("[warn] " + d, file=sys.stderr) if not args.apply: out["verdict"] = "PLAN" if stmts else "CONVERGED_ALREADY" out["statements_md5"] = hashlib.md5("\n".join(stmts).encode("utf-8")).hexdigest() print(json.dumps(out, ensure_ascii=False, indent=2)) return 0 # ── 执行（只 GRANT；不做任何 REVOKE/DML/DDL）── executed, skipped = [], [] for stmt in stmts: cur.execute(stmt) executed.append(stmt) out["executed"] = executed out["skipped_idempotent"] = skipped after = table_privs(cur, args.db, tables) readback, bad = [], [] for host in HOSTS: for t in sorted(t for t in wt if t in tables): want = set(wanted_privs(t, ops, wl_meta["default_ops"])) have = after.get((host, t), set()) readback.append({"host": host, "table": t, "wanted": sorted(want), "have": sorted(have), "ok": want <= have}) if not want <= have: bad.append("%s@%s have=%s want=%s" % (t, host, sorted(have), sorted(want))) out["readback"] = readback out["converged"] = not bad out["converged_mismatch"] = bad out["verdict"] = "APPLIED" if not bad else "READBACK_MISMATCH" # 幂等自证：二跑语句数应为 0（在同一连接、同一现网态上重算计划） stmts2, _m2, _d2 = build_plan(args.db, tables, wt, ops, wl_meta["default_ops"], table_privs(cur, args.db, tables)) out["idempotent_second_run_statement_count"] = len(stmts2) # R1-c：仅收敛成功（rc=0）时回写锚件；rc!=0 与 --check 都不写 if not bad and anchor_meta is not None: out["anchor_writeback"] = write_back_converged(anchor_meta["path"], out) print(json.dumps(out, ensure_ascii=False, indent=2)) if bad: print("[FAIL] 授权回读不符白名单：%s" % bad, file=sys.stderr) return 1 return 0 finally: conn.close() if __name__ == "__main__": sys.exit(main())