host
328ffdb7c6
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)
137 lines
5.0 KiB
Python
137 lines
5.0 KiB
Python
"""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()
|