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)
This commit is contained in:
host committed 2026-11-04 21:03:32 +00:00
1 parent e6a1346420
commit 0b52a561df
4 files changed
+281 -7

No files matched your search

+11 -4
View File
@@ -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):