From 328ffdb7c6776c7c288405d2e2a4cd9c759e8d3d Mon Sep 17 00:00:00 2001 From: host Date: Mon, 2 Nov 2026 03:33:23 +0000 Subject: [PATCH] code-to10 v6 A2: single controlled PG guard entry (code side only) Postgres audit_ledger append-only guard gains exactly one execution entry: cloud/ledger_pg_guard.apply_pg_guard, runnable as 'python3 -m cloud.ledger_pg_guard' (explicit ops command, not a migration script). It executes PG_GUARD_SQL verbatim from the constant -- no second copy of the SQL anywhere -- refuses non-PgConnection targets, does not commit itself (caller/CLI commits), and refuses with exit 2 on a non-postgres backend so a no-PG environment can never claim hardening. ensure_schema still skips PG triggers; POSTGRES_SCHEMA still builds only the table. Tests lock all of this with a recording fake connection (no PG execution is faked), a source scan asserting only ledger.py and ledger_pg_guard.py reference the constant, and CLI dry-run/refusal cases. Real-PG syntax and UPDATE/DELETE blocking verification is explicitly left to B5, not claimed here. (cherry picked from commit b1f6c9158d628186965a9e57576ff8c9500e3f73) --- cloud/ledger.py | 15 ++- cloud/ledger_pg_guard.py | 129 +++++++++++++++++++++++ cloud/tests/test_ledger_pg_guard.py | 136 +++++++++++++++++++++++++ docs/release-gate-and-ledger-anchor.md | 8 +- 4 files changed, 281 insertions(+), 7 deletions(-) create mode 100644 cloud/ledger_pg_guard.py create mode 100644 cloud/tests/test_ledger_pg_guard.py diff --git a/cloud/ledger.py b/cloud/ledger.py index 88cad3e..2426e7b 100644 --- a/cloud/ledger.py +++ b/cloud/ledger.py @@ -130,9 +130,14 @@ 不得重开。audit 触发的推倒不走此门。 PG 只追加护栏：SQLite 触发器由 ``ensure_schema`` 自动建； -PG 的函数 + 触发器 SQL 只以 ``PG_GUARD_SQL`` 常量提供， -不自动执行——本机无真实 PG，未在真实 PG 上验证触发器 -语法，生产 PG 须运维手动执行并验证后再认“已加固”。 +PG 触发器有且仅有一个受控执行入口—— +``cloud/ledger_pg_guard.py`` 的 ``apply_pg_guard``（运维命令 +``python3 -m cloud.ledger_pg_guard``），它逐条执行且仅执行 +本模块的 ``PG_GUARD_SQL`` 常量，不另写一份 SQL。其它任何路径 +（``ensure_schema``、POSTGRES_SCHEMA 建表、迁移脚本）都不执行 +PG 触发器：本机无真实 PG，触发器语法未在真实 PG 上验证， +真实 PG 上的语法与 UPDATE/DELETE 阻挡验证属于 B5 环境到齐后 +另做，本项只做代码侧入口，绝不冒充“已加固”。 PG 表本身走 ``cloud/db.py`` 的 POSTGRES_SCHEMA（与老库 初始化同款全量执行，新表自动补建）。 @@ -232,7 +237,9 @@ def ensure_schema(conn): 不 commit：调用方在自己的事务边界提交（AuthStore.__init__ 随后即 commit；独立调用方须自己 commit）。 - PG 适配连接跳过触发器建（方言不同，见 PG_GUARD_SQL）。 + PG 适配连接跳过触发器建（方言不同）：PG 触发器只能经唯一 + 受控入口 cloud/ledger_pg_guard.apply_pg_guard 执行，绝不在 + 这里自动建（见模块头 PG 段与 PG_GUARD_SQL）。 """ conn.execute(_LEDGER_TABLE_SQL) if not _is_pg_conn(conn): diff --git a/cloud/ledger_pg_guard.py b/cloud/ledger_pg_guard.py new file mode 100644 index 0000000..9846c13 --- /dev/null +++ b/cloud/ledger_pg_guard.py @@ -0,0 +1,129 @@ +"""Postgres 审计账本只追加护栏：唯一受控执行入口（A2，代码侧）。 + +本模块是 ``cloud/ledger.py`` 的 ``PG_GUARD_SQL`` 常量在 Postgres +上的唯一执行入口——二选一里选的是“显式运维命令”，不是迁移脚本： + +- 函数入口：``apply_pg_guard(conn)``，逐条执行且仅执行 + ``PG_GUARD_SQL``（从常量引用，不另写、不复制、不拼接 SQL）。 + 入口本身不 commit：调用方在自己的事务边界提交（与 + ``ledger.ensure_schema`` 的约定一致）。 +- 命令入口：``python3 -m cloud.ledger_pg_guard``（``main``）， + 连接参数只走 ``CLOUD_DB_*`` 环境变量（经 + ``cloud.db.PostgresDatabase``），不接受命令行密码参数。 + +受控含义（grep 可证，全仓库除本模块外无其它执行路径）： +- ``ledger.ensure_schema`` 在 PG 适配连接上跳过触发器建； + ``cloud/db.py`` 的 POSTGRES_SCHEMA 只建 audit_ledger 表， + 不建触发器；迁移脚本不执行本常量。 +- 入口先验连接类型：不是 ``PgConnection``（按类名判定，与 + ``ledger._is_pg_conn`` 同款）即拒绝执行，绝不把 PG 方言 SQL + 发往 SQLite 连接。 +- 无 PG 环境下入口不得 claim 已加固：``main`` 在 + ``CLOUD_DB_BACKEND`` 不是 postgres/postgresql 时直接以 + exit 2 拒绝（不连接、不执行、不打印“已应用”）；``--dry-run`` + 只打印将执行的常量语句，不连接、不执行。 + +诚实边界（A2 只做代码侧，真实 PG 验证待 B5）： +- 本机无真实 PG，``PG_GUARD_SQL`` 的语法与 UPDATE/DELETE 阻挡 + 均未在真实 PG 上验证；B5 环境到齐后须在真实 PG 上跑本命令 + 并实测 UPDATE/DELETE 被拒绝，之后才可认“已加固”。在此之前， + 任何“PG 已加固”的表述都是冒充。 +- 本模块不打印、不返回任何密码或密钥：输出只有语句计数与 + DDL 常量本身。 + +只用标准库 + cloud.db / cloud.ledger，不引第三方。 +""" + +import argparse +import json +import os +import sys + +try: + from cloud.db import PostgresDatabase + from cloud.ledger import PG_GUARD_SQL +except ImportError: # 直接以脚本方式从仓库根外运行时的回退。 + sys.path.insert(0, os.path.dirname(os.path.dirname( + os.path.abspath(__file__)))) + from cloud.db import PostgresDatabase + from cloud.ledger import PG_GUARD_SQL + + +def apply_pg_guard(conn): + """在给定 PG 适配连接上执行只追加护栏（唯一执行入口）。 + + conn 须是 PgConnection（按类名判定）；其它连接一律拒绝， + 绝不把 PG 方言发往 SQLite。按 PG_GUARD_SQL 原顺序逐条执行， + 返回执行过的语句元组（即常量本身，供调用方/测试比对）。 + 不 commit：调用方提交。 + """ + if type(conn).__name__ != "PgConnection": + raise TypeError( + "apply_pg_guard requires a PgConnection, got %s; " + "refusing to run Postgres guard SQL on a non-PG connection" + % type(conn).__name__) + for statement in PG_GUARD_SQL: + conn.execute(statement) + return tuple(PG_GUARD_SQL) + + +def main(argv=None): + """运维命令：``python3 -m cloud.ledger_pg_guard [--dry-run]``。 + + 返回进程退出码：0 已应用（或 dry-run 展示完毕），1 执行失败， + 2 用法/环境错误（非 PG 后端、非 PG 连接——一律不 claim 已应用）。 + """ + parser = argparse.ArgumentParser( + description="Apply the audit_ledger append-only guard " + "(PG_GUARD_SQL) to Postgres. Single controlled " + "entry point; see module docstring.") + parser.add_argument( + "--dry-run", action="store_true", + help="Print the statements that would run, without " + "connecting or executing anything.") + args = parser.parse_args(argv) + if args.dry_run: + print(json.dumps({"applied": False, "dry_run": True, + "statements": list(PG_GUARD_SQL)}, + ensure_ascii=False, indent=2)) + return 0 + backend = (os.environ.get("CLOUD_DB_BACKEND") or "").strip().lower() + if backend not in ("postgres", "postgresql"): + print("refused: CLOUD_DB_BACKEND is not postgres " + "(got %r); nothing applied, nothing claimed" + % backend, file=sys.stderr) + return 2 + try: + conn = PostgresDatabase().connect() + except Exception as exc: + print("failed to connect to Postgres: %s" % exc, + file=sys.stderr) + return 1 + try: + if type(conn).__name__ != "PgConnection": + print("refused: connection is not a PgConnection " + "(got %s); nothing applied" + % type(conn).__name__, file=sys.stderr) + return 2 + statements = apply_pg_guard(conn) + conn.commit() + except Exception as exc: + try: + conn.rollback() + except Exception: + pass + print("failed to apply PG guard: %s" % exc, file=sys.stderr) + return 1 + finally: + try: + conn.close() + except Exception: + pass + print(json.dumps({"applied": True, "dry_run": False, + "statements": list(statements)}, + ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/cloud/tests/test_ledger_pg_guard.py b/cloud/tests/test_ledger_pg_guard.py new file mode 100644 index 0000000..e55d03b --- /dev/null +++ b/cloud/tests/test_ledger_pg_guard.py @@ -0,0 +1,136 @@ +"""PG 审计账本护栏唯一执行入口（cloud/ledger_pg_guard，A2 代码侧）。 + +锁定的事（不得伪造 PG 执行——本机无 PG、无 DSN，一律用假连接）： +- 入口执行的 SQL 逐字等于 cloud/ledger.PG_GUARD_SQL 常量 + （假连接录下 execute 入参，与常量逐条比对；入口不另写 SQL）。 +- 除入口模块外，全仓库 cloud/*.py 再无其它引用 PG_GUARD_SQL + 的生产文件（grep 式扫描断言），即有且仅有一个执行入口。 +- 入口拒绝非 PG 连接；ensure_schema 在 PG 连接上仍跳过触发器建。 +- 运维命令在无 PG 环境下拒绝 claim：非 postgres 后端时 exit 2、 + 不执行；--dry-run 只展示、不连接、不执行。 +真实 PG 上的语法与 UPDATE/DELETE 阻挡验证待 B5，本文件不冒充。 +""" + +import json +import os +import subprocess +import sys +import unittest +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] + +try: + from cloud.ledger import PG_GUARD_SQL, ensure_schema + from cloud.ledger_pg_guard import apply_pg_guard +except ImportError: + sys.path.insert(0, str(REPO_ROOT)) + from cloud.ledger import PG_GUARD_SQL, ensure_schema + from cloud.ledger_pg_guard import apply_pg_guard + + +class _RecordingConn: + """只录 execute 入参的假连接；commit/rollback/close 计数。""" + + def __init__(self): + self.executed = [] + self.commits = 0 + self.rollbacks = 0 + self.closed = False + + def execute(self, sql, params=()): + self.executed.append(sql) + return self + + def commit(self): + self.commits += 1 + + def rollback(self): + self.rollbacks += 1 + + def close(self): + self.closed = True + + +class PgConnection(_RecordingConn): + """类名须为 PgConnection：入口与 ensure_schema 均按类名判定。""" + + +class NotPgConnection(_RecordingConn): + """非 PG 假连接：入口必须拒绝，零执行。""" + + +class ApplyPgGuardTestCase(unittest.TestCase): + def test_apply_executes_constant_verbatim(self): + conn = PgConnection() + result = apply_pg_guard(conn) + self.assertEqual(result, tuple(PG_GUARD_SQL)) + self.assertEqual(conn.executed, list(PG_GUARD_SQL)) + # 入口不 commit：调用方（CLI）在自己的事务边界提交。 + self.assertEqual(conn.commits, 0) + + def test_apply_refuses_non_pg_connection(self): + conn = NotPgConnection() + with self.assertRaises(TypeError): + apply_pg_guard(conn) + self.assertEqual(conn.executed, []) + + def test_ensure_schema_on_pg_still_skips_guard(self): + conn = PgConnection() + ensure_schema(conn) + for statement in PG_GUARD_SQL: + self.assertNotIn(statement, conn.executed) + self.assertTrue(any("audit_ledger" in sql + for sql in conn.executed)) + + +class SingleEntryTestCase(unittest.TestCase): + def test_only_entry_module_references_constant(self): + # grep 式锁定：cloud/ 下生产文件引用 PG_GUARD_SQL 的 + # 有且仅有定义处（ledger.py）与唯一入口（ledger_pg_guard.py）。 + cloud_dir = REPO_ROOT / "cloud" + referencing = sorted( + path.name for path in cloud_dir.glob("*.py") + if "PG_GUARD_SQL" in path.read_text(encoding="utf-8")) + self.assertEqual(referencing, + ["ledger.py", "ledger_pg_guard.py"]) + + def test_entry_module_has_no_second_copy_of_sql(self): + # 入口不得各写一份 SQL 漂移：本体只出现在 ledger.py， + # 入口模块里不出现函数名/触发器名字面量。 + entry = (REPO_ROOT / "cloud" / "ledger_pg_guard.py").read_text( + encoding="utf-8") + self.assertNotIn("audit_ledger_no_mutation", entry) + self.assertNotIn("CREATE TRIGGER audit_ledger_no_update", entry) + + +class CliTestCase(unittest.TestCase): + def _run_cli(self, args, env_extra): + env = dict(os.environ) + env.pop("CLOUD_DB_BACKEND", None) + env.update(env_extra) + proc = subprocess.run( + [sys.executable, "-m", "cloud.ledger_pg_guard", *args], + cwd=str(REPO_ROOT), env=env, + capture_output=True, text=True, timeout=60) + return proc + + def test_dry_run_prints_constant_without_connecting(self): + proc = self._run_cli(["--dry-run"], {}) + self.assertEqual(proc.returncode, 0, proc.stderr) + report = json.loads(proc.stdout) + self.assertFalse(report["applied"]) + self.assertTrue(report["dry_run"]) + self.assertEqual(report["statements"], list(PG_GUARD_SQL)) + + def test_non_pg_backend_refuses_without_claim(self): + for backend in ("", "sqlite"): + env = {} if backend == "" else {"CLOUD_DB_BACKEND": backend} + proc = self._run_cli([], env) + self.assertEqual(proc.returncode, 2, proc.stdout) + self.assertIn("refused", proc.stderr) + self.assertNotIn('"applied": true', proc.stdout) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/release-gate-and-ledger-anchor.md b/docs/release-gate-and-ledger-anchor.md index 745996c..85303b4 100644 --- a/docs/release-gate-and-ledger-anchor.md +++ b/docs/release-gate-and-ledger-anchor.md @@ -334,9 +334,11 @@ Postgres URL）。输出为单 JSON 对象，便于定时任务解析。 - 本机无真实 PG，PG 触发器（`cloud/ledger.py` 的 `PG_GUARD_SQL` 常量）**未在真实 PG 上验证**，沿用 #241 的标注，不在本轨改口称已验证。 -- PG 路径的操作流程（生产 PG 须运维手动执行并验证后才认）： - 1. 在生产 PG 上手动执行 `PG_GUARD_SQL`（函数 + 两触发器）， - 并实测 UPDATE/DELETE 被拒绝； +- PG 路径的操作流程（生产 PG 须经唯一入口执行并验证后才认）： + 1. 在生产 PG 上经唯一受控入口 + `python3 -m cloud.ledger_pg_guard` 执行护栏（该命令执行的 + SQL 即 `PG_GUARD_SQL` 常量本身，不另写一份；除该入口外 + 无其它执行路径），并实测 UPDATE/DELETE 被拒绝； 2. 经 `GET /admin/audit-ledger/verify`（全局管理员）确认 `ok=true`； 3. 经 `GET /admin/audit-ledger/export` 取导出，用本脚本