fix(cma): DDL 与启动剥离 —— 关闭启动 create_all + 部署链特权迁移步 (cma-create-all-minpriv-rootfix-20260916)
ci/woodpecker/push/main Pipeline failed

根治 2026-09-16 发布中断(新增模型触发启动 create_all,cma_api 无 CREATE → 崩溃循环):

- backend/app/database.py: 新增 CMA_AUTO_CREATE 开关(**默认关**),生产启动不再执行 DDL;
  运行库账号因此无需 CREATE(最小权限可保持)
- backend/scripts/db_migrate.py: 特权迁移步(--check/--apply/--rollback/--status);
  幂等(checkfirst)+ journal 留痕 + 可回滚(dump 失败/表有数据即拒绝,fail-closed);
  显式拒绝用应用运行账号 cma_api 执行
- deploy.sh: 新增 [4b] 结构迁移(**重启前**执行;失败即中止部署,后端旧码继续服务)
- backend/scripts/tests/db-migrate-sim-e2e.sh: 17 项 scratch 复现/验收(负控制复现 CREATE command denied)
- docs/db-migration-runbook.md: 运行手册(含 cma_api CREATE 回收步骤)

验收:scratch E2E 17/17;pytest 1113 passed/1 xfailed;生产 --check 缺失表 0 / 列级差异 0;
--apply 幂等空跑 created_new=0(journal 留痕)
This commit is contained in:
Hermes CI Fix
2026-09-16 06:53:28 +08:00
parent d54ffaf531
commit 443a779ece
6 changed files with 709 additions and 3 deletions
+432
View File
@@ -0,0 +1,432 @@
#!/usr/bin/env python3
"""CMA 数据库结构迁移(特权 DDL 步)—— 与应用启动 create_all 剥离
任务:cma-create-all-minpriv-rootfix-20260916(P2)
背景(2026-09-16 01:08 发布中断事故)
--------------------------------------
应用启动时 `app/database.py::init_db()` 调 `Base.metadata.create_all()`(需要 CREATE 权限),
而运行库账号 `cma_api` 只有 SELECT + 表级 DML 白名单(**无 DDL**)。新代码引入新模型
(`tax_filing_log` / `treasury_*`)→ 启动即 `CREATE command denied` → 崩溃重启循环。
本脚本 = 部署链里的**独立、幂等、可回滚**的特权迁移步,在应用重启**之前**执行:
deploy.sh [4b] → backend/scripts/db_migrate.py --apply → systemctl restart cma-backend
于是:DDL 不再随应用 boot 发生;运行库账号可保持最小权限(无需 CREATE)。
用法
----
python3 scripts/db_migrate.py --check # 只读预检(默认动作,零写入)
python3 scripts/db_migrate.py --apply # 幂等建表(缺什么建什么)+ 写 journal
python3 scripts/db_migrate.py --status # 列出历史 journal
python3 scripts/db_migrate.py --rollback latest # 回滚最近一次 --apply 新建的表
python3 scripts/db_migrate.py --rollback <journal.json>
选项
----
--json 机器可读输出(部署链/CI 断言用)
--force 回滚时允许删除**有数据**的表(默认拒绝,exit 3)
--no-dump 回滚时不 mysqldump(默认先备份到 journal 目录,dump 失败即拒绝回滚)
--socket PATH unix socket 路径(默认 /var/run/mysqld/mysqld.sock)
--db-url URL 显式连接串(覆盖下方全部推导;不含口令时应配合环境变量)
--db-name NAME 目标库名(默认 env CMA_MIGRATE_DB_NAME → CMA_DB_NAME → cma)
连接凭据(按优先级;口令只从环境读,禁止硬编码/回显)
----------------------------------------------------
1) CMA_MIGRATE_DB_USER + CMA_MIGRATE_DB_PASS(**专用迁移账号**,TCP)——推荐用于非本机场景
2) 未设 CMA_MIGRATE_DB_USER → 本机 unix socket + root(socket 认证,无需口令)
⚠️ 应用运行账号(backend/.env 的 CMA_DB_USER,= cma_api)**不得**被本脚本使用:
迁移必须走在有 DDL 权限的账号上,本脚本会显式拒绝。
幂等契约
--------
首跑 --apply:created_new = 缺失表数(本次实测:0 —— 库已由应急 create_all 补齐)
二跑 --apply:created_new = 0,零副作用
已存在的表**不** ALTER、**不** DROP、**不**动数据行(列级差异只报告,见下)
退出码
------
0 成功 | 1 迁移失败 / 迁移后自检不一致 / 连接失败(fail-closed,部署链据此中止)
2 用法错误 | 3 回滚被拒(目标表有数据且未 --force)
"""
import argparse
import fcntl
import json
import os
import random
import socket
import string
import subprocess
import sys
import tempfile
from datetime import datetime
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from sqlalchemy import create_engine, inspect, text # noqa: E402
from sqlalchemy.engine.url import make_url # noqa: E402
HERE = os.path.dirname(os.path.abspath(__file__))
BACKEND_DIR = os.path.dirname(HERE)
DEFAULT_JOURNAL_DIR = os.path.join(HERE, "backups", "db-migrate-journal")
DEFAULT_SOCKET = "/var/run/mysqld/mysqld.sock"
DEFAULT_DB = "cma"
# 应用运行账号:绝不允许用于本迁移步(无 DDL 权限 = 事故根因)
FORBIDDEN_RUN_ACCOUNTS = {"cma_api"}
# ────────────────────────────────────────────────────────────────
# 连接(特权通道)
# ────────────────────────────────────────────────────────────────
def _resolve_db_name(args) -> str:
return (
args.db_name
or os.getenv("CMA_MIGRATE_DB_NAME")
or os.getenv("CMA_DB_NAME")
or DEFAULT_DB
)
def build_engine(args):
"""构造特权连接。返回 (engine, 描述串)。描述串不含口令。"""
db_name = _resolve_db_name(args)
if args.db_url:
url = make_url(args.db_url)
if url.database is None:
url = url.set(database=db_name)
return create_engine(url, pool_pre_ping=True), f"db-url({url.drivername}@{url.host or 'socket'}/{url.database})"
user = os.getenv("CMA_MIGRATE_DB_USER", "").strip()
if user:
if user in FORBIDDEN_RUN_ACCOUNTS:
raise SystemExit(
f"[FAIL] CMA_MIGRATE_DB_USER={user} 是**应用运行账号**(无 DDL 权限)——\n"
f" 迁移步必须使用有 DDL 权限的专用账号(或本机 root socket)。已拒绝执行。"
)
password = os.getenv("CMA_MIGRATE_DB_PASS", "")
host = os.getenv("CMA_MIGRATE_DB_HOST", "127.0.0.1")
port = int(os.getenv("CMA_MIGRATE_DB_PORT", "3306"))
url = (
f"mysql+pymysql://{user}:{password}@{host}:{port}/{db_name}"
f"?charset=utf8mb4"
)
return create_engine(url, pool_pre_ping=True), f"tcp({user}@{host}:{port}/{db_name})"
# 默认:本机 unix socket + root(socket 认证,无口令落盘)
sock = args.socket or os.getenv("CMA_MIGRATE_DB_SOCKET") or DEFAULT_SOCKET
if not os.path.exists(sock):
raise SystemExit(
f"[FAIL] 未设 CMA_MIGRATE_DB_USER,且 unix socket 不存在:{sock}\n"
f" 请提供专用迁移账号(CMA_MIGRATE_DB_USER/PASS)或用 --socket 指定路径。"
)
url = f"mysql+pymysql://root@localhost/{db_name}?unix_socket={sock}&charset=utf8mb4"
return create_engine(url, pool_pre_ping=True), f"socket(root@{sock}/{db_name})"
# ────────────────────────────────────────────────────────────────
# schema 比对(ORM 期望 vs 库内实况)
# ────────────────────────────────────────────────────────────────
def _import_models():
import app.models # noqa: F401 触发全部模型注册
from app.database import Base
return Base
def schema_diff(engine):
"""返回 (missing_tables, missing_columns{表:[列]})。只读。"""
Base = _import_models()
insp = inspect(engine)
existing = set(insp.get_table_names())
missing_tables, missing_columns = [], {}
for table in Base.metadata.sorted_tables:
if table.name not in existing:
missing_tables.append(table.name)
continue
have = {c["name"] for c in insp.get_columns(table.name)}
miss = [c.name for c in table.columns if c.name not in have]
if miss:
missing_columns[table.name] = miss
return missing_tables, missing_columns
# ────────────────────────────────────────────────────────────────
# journal(迁移留痕:谁/何时/建了什么 → 回滚依据)
# ────────────────────────────────────────────────────────────────
def journal_dir() -> str:
d = os.getenv("CMA_MIGRATE_JOURNAL_DIR") or DEFAULT_JOURNAL_DIR
os.makedirs(d, exist_ok=True)
return d
def _git_head() -> str:
try:
return subprocess.run(
["git", "-C", os.path.dirname(BACKEND_DIR), "rev-parse", "--short", "HEAD"],
capture_output=True, text=True, timeout=10,
).stdout.strip() or "unknown"
except Exception:
return "unknown"
def write_journal(payload: dict) -> str:
path = os.path.join(
journal_dir(), f"{datetime.now().strftime('%Y%m%dT%H%M%S')}-{payload.get('status', 'na')}.json"
)
with open(path, "w", encoding="utf-8") as fh:
json.dump(payload, fh, ensure_ascii=False, indent=2)
return path
def list_journals():
d = journal_dir()
return sorted(
(os.path.join(d, f) for f in os.listdir(d) if f.endswith(".json")),
reverse=True,
)
def resolve_journal(ref: str) -> str:
if ref in ("latest", ""):
items = [p for p in list_journals() if "-applied" in os.path.basename(p)]
if not items:
raise SystemExit("[FAIL] 没有可回滚的 journal(无 --apply 记录)")
return items[0]
if os.path.exists(ref):
return ref
cand = os.path.join(journal_dir(), os.path.basename(ref))
if os.path.exists(cand):
return cand
raise SystemExit(f"[FAIL] journal 不存在:{ref}")
# ────────────────────────────────────────────────────────────────
# 动作
# ────────────────────────────────────────────────────────────────
def cmd_check(args, engine, conn_desc) -> int:
missing_tables, missing_cols = schema_diff(engine)
out = {
"action": "check",
"connection": conn_desc,
"missing_tables": missing_tables,
"missing_table_count": len(missing_tables),
"missing_columns": missing_cols,
}
if args.json:
print(json.dumps(out, ensure_ascii=False))
else:
print(f"[check] 连接: {conn_desc}")
print(f"[check] 缺失表 {len(missing_tables)} 张: {missing_tables if missing_tables else '(无)'}")
if missing_cols:
print(f"[check] 列级差异 {len(missing_cols)} 张表: {missing_cols}")
print(" ⚠️ create_all 不会补列 → 需要独立的 ALTER 迁移(见 journal/回报)")
else:
print("[check] 列级差异:无")
# 列级差异属"存在但结构过期":部署链上以告警呈现,不阻断(阻断判据 = 缺表未补齐)
return 0
def cmd_apply(args, engine, conn_desc) -> int:
Base = _import_models()
before_missing, before_cols = schema_diff(engine)
started = datetime.now().isoformat(timespec="seconds")
payload = {
"action": "apply",
"status": "running",
"started_at": started,
"finished_at": None,
"connection": conn_desc,
"git_head": _git_head(),
"actor": f"{os.getenv('USER', 'unknown')}@{socket.gethostname()}",
"db_name": _resolve_db_name(args),
"missing_tables_before": before_missing,
"missing_columns_before": before_cols,
"created_tables": [],
"created_new": 0,
"error": None,
}
print(f"[apply] 连接: {conn_desc}")
print(f"[apply] 迁移前缺失表 {len(before_missing)} 张: {before_missing if before_missing else '(无)'}")
try:
if before_missing:
Base.metadata.create_all(bind=engine) # checkfirst=True → 已存在的表零动作
after_missing, after_cols = schema_diff(engine)
created = [t for t in before_missing if t not in after_missing]
payload.update(
status="applied",
finished_at=datetime.now().isoformat(timespec="seconds"),
created_tables=created,
created_new=len(created),
missing_tables_after=after_missing,
)
path = write_journal(payload)
except Exception as exc: # DDL 无事务:失败时记录已建对象,供 --rollback 恢复
try:
after_missing, _ = schema_diff(engine)
except Exception:
after_missing = before_missing
created = [t for t in before_missing if t not in after_missing]
payload.update(
status="failed",
finished_at=datetime.now().isoformat(timespec="seconds"),
created_tables=created,
created_new=len(created),
error=f"{type(exc).__name__}: {exc}",
)
path = write_journal(payload)
print(f"[FAIL] 迁移失败:{payload['error']}")
print(f" 已建对象 {created}(journal: {path})")
print(f" 回滚:python3 scripts/db_migrate.py --rollback {os.path.basename(path)}")
return 1
ok = not payload["missing_tables_after"]
print(f"[apply] 新建 {payload['created_new']} 张表: {created if created else '(无 → 幂等空跑)'}")
print(f"[apply] 迁移后缺失表 {len(payload['missing_tables_after'])} 张: "
f"{payload['missing_tables_after'] if payload['missing_tables_after'] else '(无)'}")
if after_cols:
print(f"[apply] ⚠️ 列级差异(create_all 不补列,需独立 ALTER): {after_cols}")
print(f"[apply] journal: {path}")
if not ok:
print("[FAIL] 迁移后自检不一致(仍有缺失表)→ 部署链应中止")
return 1
return 0
def _dump_tables(tables, db_name) -> str:
"""mysqldump 备份(凭据走 600 临时 defaults 文件,不进 argv/日志)。返回文件路径。"""
fd, cfg = tempfile.mkstemp(prefix="cma-migrate-dump-", suffix=".cnf")
os.close(fd)
os.chmod(cfg, 0o600)
try:
lines = ["[client]"]
if os.getenv("CMA_MIGRATE_DB_USER"):
lines += [
f"user={os.getenv('CMA_MIGRATE_DB_USER')}",
f"password={os.getenv('CMA_MIGRATE_DB_PASS', '')}",
f"host={os.getenv('CMA_MIGRATE_DB_HOST', '127.0.0.1')}",
f"port={os.getenv('CMA_MIGRATE_DB_PORT', '3306')}",
]
else:
lines += ["user=root", f"socket={os.getenv('CMA_MIGRATE_DB_SOCKET', DEFAULT_SOCKET)}"]
with open(cfg, "w", encoding="utf-8") as fh:
fh.write("\n".join(lines) + "\n")
out = os.path.join(
journal_dir(), f"dump-{datetime.now().strftime('%Y%m%dT%H%M%S')}.sql"
)
cmd = ["mysqldump", f"--defaults-extra-file={cfg}", "--no-tablespaces", "--skip-add-locks", db_name, *tables]
with open(out, "wb") as fh:
subprocess.run(cmd, stdout=fh, stderr=subprocess.PIPE, check=True)
return out
finally:
os.unlink(cfg)
def cmd_rollback(args, engine, conn_desc) -> int:
path = resolve_journal(args.rollback)
with open(path, encoding="utf-8") as fh:
payload = json.load(fh)
tables = payload.get("created_tables") or []
print(f"[rollback] journal: {path}")
print(f"[rollback] 该次迁移新建的表: {tables if tables else '(无)'}")
if not tables:
print("[rollback] 无新建对象 → 无需回滚(空操作)")
return 0
db_name = payload.get("db_name") or _resolve_db_name(args)
if not args.no_dump:
try:
dump_path = _dump_tables(tables, db_name)
print(f"[rollback] 已备份: {dump_path}")
except Exception as exc:
print(f"[FAIL] 备份失败 → 拒绝回滚(fail-closed):{type(exc).__name__}: {exc}")
return 1
with engine.connect() as conn:
rows = {}
for t in tables:
try:
rows[t] = conn.execute(text(f"SELECT COUNT(*) FROM `{t}`")).scalar()
except Exception:
rows[t] = 0
non_empty = {t: n for t, n in rows.items() if n}
if non_empty and not args.force:
print(f"[REFUSED] 以下表已有数据,未加 --force 拒绝删除:{non_empty}")
return 3
for t in tables:
conn.execute(text(f"DROP TABLE IF EXISTS `{t}`"))
conn.commit()
print(f"[rollback] 已删除 {len(tables)} 张表: {tables}")
payload["rolled_back_at"] = datetime.now().isoformat(timespec="seconds")
payload["rolled_back_by"] = f"{os.getenv('USER', 'unknown')}@{socket.gethostname()}"
payload["rollback_dump"] = None if args.no_dump else "见同目录 dump-*.sql"
base = os.path.basename(path)
archived = os.path.join(journal_dir(), base.replace("-applied.json", "-rolled_back.json"))
with open(archived, "w", encoding="utf-8") as fh:
json.dump(payload, fh, ensure_ascii=False, indent=2)
print(f"[rollback] 记录: {archived}")
return 0
def cmd_status(args) -> int:
items = list_journals()
if not items:
print("[status] 无 journal 记录")
return 0
for p in items[:20]:
try:
with open(p, encoding="utf-8") as fh:
j = json.load(fh)
print(
f"{os.path.basename(p)} status={j.get('status')} created_new={j.get('created_new')} "
f"tables={len(j.get('created_tables') or [])} git={j.get('git_head')} by={j.get('actor')}"
)
except Exception as exc:
print(f"{os.path.basename(p)} [unreadable] {exc}")
return 0
# ────────────────────────────────────────────────────────────────
def main() -> int:
ap = argparse.ArgumentParser(
description="CMA 特权数据库迁移步(与应用启动 create_all 剥离)",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
g = ap.add_mutually_exclusive_group()
g.add_argument("--check", action="store_true", help="只读预检(默认)")
g.add_argument("--apply", action="store_true", help="幂等建表 + journal")
g.add_argument("--rollback", metavar="JOURNAL", help="回滚某次 --apply 新建的表(latest 或文件名)")
g.add_argument("--status", action="store_true", help="列出 journal 历史")
ap.add_argument("--json", action="store_true", help="机器可读输出")
ap.add_argument("--force", action="store_true", help="回滚允许删有数据的表")
ap.add_argument("--no-dump", action="store_true", help="回滚前不 dump")
ap.add_argument("--socket", default=None, help=f"unix socket 路径(默认 {DEFAULT_SOCKET})")
ap.add_argument("--db-url", default=None, help="显式连接串(覆盖推导)")
ap.add_argument("--db-name", default=None, help="目标库名")
args = ap.parse_args()
if not (args.apply or args.rollback or args.status):
args.check = True # 默认只读预检
if args.status:
return cmd_status(args)
engine, conn_desc = build_engine(args)
# 迁移互斥:防并发会话/手工操作同时跑 DDL
lock_path = os.path.join(journal_dir(), ".db-migrate.lock")
lock_fh = open(lock_path, "a+")
try:
fcntl.flock(lock_fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
print("[FAIL] 另一个迁移正在运行(锁被占用)→ 本次中止")
return 1
if args.rollback:
return cmd_rollback(args, engine, conn_desc)
if args.apply:
return cmd_apply(args, engine, conn_desc)
return cmd_check(args, engine, conn_desc)
if __name__ == "__main__":
sys.exit(main())