From b8493f8d046927a7a7ab17953de7cd9acd0d4a1f Mon Sep 17 00:00:00 2001 From: wecom-fullstack Date: Mon, 28 Sep 2026 17:01:39 +0800 Subject: [PATCH] =?UTF-8?q?feat(db):=20=E6=96=B0=E8=A1=A8=E6=8E=88?= =?UTF-8?q?=E6=9D=83=E9=9D=A2=E6=94=B6=E6=95=9B=E6=9C=BA=E5=88=B6=E8=90=BD?= =?UTF-8?q?=E5=9C=B0=EF=BC=88=E7=99=BD=E5=90=8D=E5=8D=95=E9=A9=B1=E5=8A=A8?= =?UTF-8?q?=20=C2=B7=20=E5=A2=9E=E9=87=8F=E9=94=9A=E5=AE=9A=20=C2=B7=20fai?= =?UTF-8?q?l-closed=20=C2=B7=20=E5=8F=AA=E5=A2=9E=E4=B8=8D=E5=87=8F?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 依据：A 侧（wecom-yanxue）裁定 编号 2「准予但需调整」+ R1（裁定件 yanxue-20260928-newtable-grant-surface-impl-plan-ruling.md，md5 272d1a8b4db0db0e9f35f5e62695ae50） 三条开口 (a) 空锚=合法零集 / (b) --anchor-file 载体等价 / (c) drill WARN 不阻断（两条独立分支） 改动面（3 件，全部开发工作树）： - 新增 backend/scripts/grant_surface_sync.py：项目侧原型逐字复制 + Δ1/Δ2/Δ3/Δ4 + R1 Δ1 锚语义（未给=缺锚 rc=2 / 空集=合法零集）、Δ2 连接身份与 db_migrate 同口径并拒绝 cma_api、 Δ3 --anchor-file（读 db-migrate-journal 指针件）、Δ4 anchor_source 入 JSON R1-a 同源自证（journal 在册 + status=applied + finished_at 一致 + 目录内最新 -applied） R1-b 未收敛锚累积由 db_migrate.write_anchor_pointer 实现、R1-c 收敛回写 converged_at - db_migrate.py 加性 +2 处：新增 write_anchor_pointer()（原子写 journal/last-anchor.json）； cmd_apply 成功/失败两路径各调一次；零变更面未动（列级差异只告警 / create_all 幂等 / --check 不写指针 / --rollback 与 resolve_journal 语义 / JSON 键集） - deploy.sh 加性 1 步 [4b-2/5] 授权面收敛（插在「✓ 结构迁移完成」与「# 5. 重启后端服务」之间）： apply 跑 --apply --anchor-file ... --drift-policy warn，rc!=0 ⇒ 打印缺裁定清单 + exit 1 （后端不重启、旧码继续服务）；drill 跑 --check，读数原样打印、WARN 不阻断 证据（全部零生产面）： - py_compile x3 通过；bash -n deploy.sh 通过（PYTHONPYCACHEPREFIX 隔离，仓内零新增 pyc） - 纯函数用例 11/11（零连库）：语句均为 GRANT / 无 REVOKE / 语句表 ⊆ write_tables / 域外权限 SystemExit - 沙箱 S1–S9 回归 9/9 一致（SUMMARY.txt 与项目侧基线逐字节相同 md5 1a697e002a2cba8ae9b5ad9cc71b022d） - 锚/R1 新例 18/18：缺锚/缺件/status=failed/R1-a 四负例均 rc=2 零写入；空锚 rc=0 executed=[] - R1 验收链：发布1 建表 X(∉白名单) ⇒ rc=3 零写入 + 片段 exit 1 ⇒ 不修白名单重跑 ⇒ 仍 rc=3 ⇒ 补白名单 ⇒ rc=0 + X 获授权 + converged_at 落定 ⇒ 三次重跑 executed=[]（幂等） - 既有 backend/tests：1509 passed / 0 failed，改前（HEAD 导出树）与改后失败集零差异 边界：本地 commit，零 push / 零部署 / 零重启 / 零连生产库 / 零 DDL-DML-GRANT-REVOKE； 上线（push + 部署 ⇒ 授权面生效）另单另铸 approval ID 且必附 R2 预演 diff（本单不为任何动作铸 ID） --- backend/scripts/db_migrate.py | 66 ++++ backend/scripts/grant_surface_sync.py | 465 ++++++++++++++++++++++++++ deploy.sh | 40 +++ 3 files changed, 571 insertions(+) create mode 100644 backend/scripts/grant_surface_sync.py diff --git a/backend/scripts/db_migrate.py b/backend/scripts/db_migrate.py index 5cd1c890..20f190f6 100644 --- a/backend/scripts/db_migrate.py +++ b/backend/scripts/db_migrate.py @@ -177,6 +177,63 @@ def write_journal(payload: dict) -> str: json.dump(payload, fh, ensure_ascii=False, indent=2) return path +def write_anchor_pointer(payload: dict, journal_path: str) -> str: + """原子写 journal/last-anchor.json —— 授权面收敛步（grant_surface_sync.py）消费的锚（R1） + + 键：journal（本次 journal 文件名，供 R1-a 同源自证）/ status / created_tables / created_new / + anchor_csv / finished_at / git_head / db_name / actor / carried_over / carried_from / + written_at / converged_at + R1-b：已存在指针且其 status=applied 而未收敛（无 converged_at）时，本次 created_tables = + 旧集合 ∪ 本次 created，并记 carried_over / carried_from；无旧指针或旧指针已收敛时 = 本次 + 加性写：不改 journal、不改任何既有键语义；失败由调用方按告警处理（收敛步会据此 rc=2 中止） + """ + d = journal_dir() + path = os.path.join(d, "last-anchor.json") + prev = None + try: + with open(path, encoding="utf-8") as fh: + prev = json.load(fh) + except Exception: + prev = None + created = [t for t in (payload.get("created_tables") or [])] + carried_over, carried_from = False, None + if (isinstance(prev, dict) and prev.get("status") == "applied" + and not prev.get("converged_at") and (prev.get("created_tables") or [])): + old = list(prev.get("created_tables") or []) + created = sorted(set(old) | set(created)) + carried_from = os.path.basename(str(prev.get("journal") or "")) or None + carried_over = True + doc = { + "pointer": "last-anchor.json", + "journal": os.path.basename(journal_path), + "status": payload.get("status"), + "created_tables": created, + "created_new": payload.get("created_new"), + "anchor_csv": ",".join(created), + "carried_over": carried_over, + "carried_from": carried_from, + "finished_at": payload.get("finished_at"), + "git_head": payload.get("git_head"), + "db_name": payload.get("db_name"), + "actor": payload.get("actor"), + "written_at": datetime.now().isoformat(timespec="seconds"), + "converged_at": None, + } + 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: + ref = path if os.path.exists(path) else journal_path + if os.path.exists(ref): + st = os.stat(ref) + os.chmod(tmp, st.st_mode & 0o777) + os.chown(tmp, st.st_uid, st.st_gid) + except Exception: + pass + os.replace(tmp, path) + return path + def list_journals(): d = journal_dir() @@ -282,6 +339,11 @@ def cmd_apply(args, engine, conn_desc) -> int: structural=structural, ) path = write_journal(payload) + try: + write_anchor_pointer(payload, path) + except Exception as exc: # 加性：指针写失败不阻断迁移，由收敛步 fail-closed 兜住 + print(f"[apply] [warn] 锚指针写入失败（收敛步会据此 rc=2 中止部署）：" + f"{type(exc).__name__}: {exc}") except Exception as exc: # DDL 无事务：失败时记录已建对象，供 --rollback 恢复 try: after_missing, _ = schema_diff(engine) @@ -296,6 +358,10 @@ def cmd_apply(args, engine, conn_desc) -> int: error=f"{type(exc).__name__}: {exc}", ) path = write_journal(payload) + try: + write_anchor_pointer(payload, path) + except Exception as exc: # 加性：失败路径同样留锚（status=failed），收敛步只消费 applied + print(f"[apply] [warn] 锚指针写入失败（失败路径）：{type(exc).__name__}: {exc}") print(f"[FAIL] 迁移失败：{payload['error']}") print(f" 已建对象 {created}（journal: {path}）") print(f" 回滚：python3 scripts/db_migrate.py --rollback {os.path.basename(path)}") diff --git a/backend/scripts/grant_surface_sync.py b/backend/scripts/grant_surface_sync.py new file mode 100644 index 00000000..b9201e5b --- /dev/null +++ b/backend/scripts/grant_surface_sync.py @@ -0,0 +1,465 @@ +#!/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()) diff --git a/deploy.sh b/deploy.sh index 2f5ea7e0..29267d98 100755 --- a/deploy.sh +++ b/deploy.sh @@ -213,6 +213,8 @@ echo " ✓ 依赖检查完成" # cma-create-all-minpriv-rootfix-20260916：DDL 已从应用启动剥离（CMA_AUTO_CREATE 默认关）， # 新模型必须由**有 DDL 权限的账号**在重启前建表；此处失败即中止部署（后端保持旧码运行）， # 避免 2026-09-16 那种「启动 create_all 被 cma_api 拒 → 崩溃循环」。 +# [4b-2] 授权面收敛：与本节同属「重启前特权步」，锚 = 本次 create_all 的 created_tables +# （经 journal/last-anchor.json 转交）；本步只发 GRANT、从不 REVOKE、缺裁定即中止 echo "[4b/5] 数据库结构迁移（特权账号，重启前）..." cd "$DEPLOY_SRC/backend" MIGRATE_ARG="--apply" @@ -226,6 +228,44 @@ fi cd "$DEPLOY_SRC" echo " ✓ 结构迁移完成（幂等；journal: $DEPLOY_SRC/backend/scripts/backups/db-migrate-journal/）" +# 4b-2. 授权面收敛（白名单驱动 · 增量锚定 · fail-closed · 只增不减） +# 背景：新表建表（[4b] create_all）不在任何自动授权步内 ⇒ 表在、写权缺 ⇒ 端点 500 / 1142 +# 锚 = 本次 db_migrate --apply 写出的 journal/last-anchor.json（R1-a 同源自证：锚件引用本次 journal） +# 本步只发 GRANT（从不 REVOKE）；缺白名单裁定 ⇒ 非零退出、后端不重启、旧码继续服务 +# 回滚：去掉本步（[4b-2/5]）即可 —— 本步不建对象、不改表结构、不 REVOKE，不执行即无残留 +echo "[4b-2/5] 授权面收敛（白名单驱动 · 增量锚定 · fail-closed）..." +cd "$DEPLOY_SRC/backend" +if [ "$MODE" = "drill" ]; then + # drill：只读预检，读数原样打印，WARN 不阻断（与 [4b] 的「只读预检」口径一致） + SYNC_RC=0 + ./venv/bin/python3 scripts/grant_surface_sync.py --check \ + --anchor-file scripts/backups/db-migrate-journal/last-anchor.json || SYNC_RC=$? + if [ "$SYNC_RC" -ne 0 ]; then + echo " [warn] [4b-2] drill 读数非零 rc=$SYNC_RC（缺裁定/缺锚清单见上方 coverage.missing_caliber）" + echo " [warn] 生产 apply 时会据此中止（exit 1，后端不重启）；drill 不阻断" + fi + SYNC_PLAN_RC=0 + ./venv/bin/python3 scripts/grant_surface_sync.py --check || SYNC_PLAN_RC=$? + if [ "$SYNC_PLAN_RC" -ne 0 ]; then + echo " [warn] [4b-2] drill 全域只读预演 rc=$SYNC_PLAN_RC（见上方 plan/matrix/drift）" + else + echo " [info] [4b-2] drill 全域只读预演已完成（语句计划与 statements_md5 见上方 plan）" + fi +else + SYNC_RC=0 + ./venv/bin/python3 scripts/grant_surface_sync.py --apply \ + --anchor-file scripts/backups/db-migrate-journal/last-anchor.json --drift-policy warn \ + || SYNC_RC=$? + if [ "$SYNC_RC" -ne 0 ]; then + echo "❌ [4b-2] 授权面收敛失败 rc=$SYNC_RC → 中止部署（后端未重启，旧码继续服务）" + echo " 缺裁定清单见上方 coverage.missing_caliber（未登记白名单的新建表须先走裁定另单）" + echo " 回滚: 去掉本步（[4b-2/5]）即可；本步从不 REVOKE/DROP，不执行即无残留" + exit 1 + fi +fi +cd "$DEPLOY_SRC" +echo " ✓ 授权面收敛完成（只增不减；指针: $DEPLOY_SRC/backend/scripts/backups/db-migrate-journal/last-anchor.json）" + # 5. 重启后端服务（运行树 = 部署树：幂等改写 unit，改前备份，附一行回滚） echo "[5/5] 重启后端服务..." if [ "$MODE" = "drill" ]; then