From 8006a6fcba5ef6eef0a588172100591a83f55875 Mon Sep 17 00:00:00 2001 From: host Date: Sat, 31 Oct 2026 04:53:57 +0000 Subject: [PATCH] audit: route automation audit rows through hash-chain pair (gap 4, option A) Bypass closed: automations._audit_row (direct audit_log INSERT, best-effort swallow) is deleted. run_due_automation now writes via cloud.auth.audit_on_conn, the same audit_log + audit_ledger pair AuthStore.audit() writes, sharing created_at. The audit pair runs before record_automation_run's commit, so task row, run row and audit pair commit or roll back together; chain-head IntegrityError retries the whole block up to 5 times, other errors roll back and raise (fail-closed). The claim stays committed, so a failed run is retried next interval, never recorded as a failed run. AuthStore.audit() now delegates to audit_on_conn; its pair, retry count and fail-closed semantics are unchanged. --- cloud/auth.py | 73 ++++-- cloud/automations.py | 175 ++++++------- cloud/tests/test_audit_automations_chain.py | 260 ++++++++++++++++++++ cloud/tests/test_automation_run_due.py | 80 ++++-- 4 files changed, 467 insertions(+), 121 deletions(-) create mode 100644 cloud/tests/test_audit_automations_chain.py diff --git a/cloud/auth.py b/cloud/auth.py index dd7a6e4..039659e 100644 --- a/cloud/auth.py +++ b/cloud/auth.py @@ -40,6 +40,55 @@ _ACTION_MIN_ROLE = { "manage_project": "owner", } +# audit_log 建表 SQL（AuthStore.__init__ 与 audit_on_conn 共用同一 +# 份，绝不能各写一份后漂移）。PG 适配层把其中的 +# INTEGER PRIMARY KEY AUTOINCREMENT 翻译成 identity 列，见 +# cloud/db.py；SQLite 下原样执行。 +_AUDIT_LOG_TABLE_SQL = ( + "CREATE TABLE IF NOT EXISTS audit_log(" + "id INTEGER PRIMARY KEY AUTOINCREMENT, " + "user_id TEXT, " + "project_id TEXT, " + "action TEXT, " + "detail TEXT, " + "created_at REAL)" +) + + +def audit_on_conn(conn, user_id, project_id, action, detail="", + created_at=None): + """在调用方给定的连接上写一对审计记录（单次尝试）。 + + 与 AuthStore.audit() 是同一对写：audit_log 行 + audit_ledger + 链条目同 created_at（调用方不给时取当前时间），先建两张表 + （幂等），再写两行。本函数不 commit、不 rollback、不重试： + 事务边界属于调用方——调用方须在自己的事务里 commit，失败 + 时 rollback 整段业务写、链头唯一冲突时 rollback 后重试整段 + （只重试本对写会丢掉调用方同事务里已写但未提交的业务行； + 见 cloud/ledger.py append_entry 的警告）。AuthStore.audit() + 即本函数 + “自有连接上 commit、唯一冲突重试 5 次”的包装。 + + 存在理由：automations.run_due_automation 拿到的是调用方连接 + （任务 INSERT 与产出记录都在该连接的事务里），不能像服务端 + handler 那样另开 AuthStore（自有连接、自有事务）去写审计—— + 那样审计与任务不在同一事务，审计失败时任务已落盘，所谓 + “整对回滚”只保审计对、不保业务对。本函数让自动化执行器 + 的审计行与任务行、产出记录行同事务提交、同事务回滚。 + """ + if created_at is None: + created_at = time.time() + created_at = float(created_at) + conn.execute(_AUDIT_LOG_TABLE_SQL) + _ledger_ensure(conn) + conn.execute( + "INSERT INTO audit_log(user_id, project_id, action," + " detail, created_at)" + " VALUES(?, ?, ?, ?, ?)", + (user_id, project_id, action, detail, created_at), + ) + _ledger_append(conn, user_id, project_id, action, + detail=detail, created_at=created_at) + class AuthStore: """基于 SQLite 的用户 / 项目成员 / 审计存储。""" @@ -61,15 +110,7 @@ class AuthStore: "role TEXT, " "PRIMARY KEY(project_id, user_id))" ) - self.conn.execute( - "CREATE TABLE IF NOT EXISTS audit_log(" - "id INTEGER PRIMARY KEY AUTOINCREMENT, " - "user_id TEXT, " - "project_id TEXT, " - "action TEXT, " - "detail TEXT, " - "created_at REAL)" - ) + self.conn.execute(_AUDIT_LOG_TABLE_SQL) # 防篡改账本（哈希链）：audit_log 本身不动，链写进 # audit_ledger；建表失败即建库失败，不静默丢链。 _ledger_ensure(self.conn) @@ -361,19 +402,17 @@ class AuthStore: “audit_log 有行、链上无行”的静默缺口）。并发抢链头 导致唯一冲突时在事务边界重试整对写入，最多 5 次； 其它错误直接回滚抛错，不重试。 + + 实现即 audit_on_conn() + 本连接上的 commit/重试包装， + 对写本身与 #241 完全一致（同一函数、同一 created_at + 语义），不是第二套审计写路径。 """ now = time.time() last_exc = None for _attempt in range(5): try: - self.conn.execute( - "INSERT INTO audit_log(user_id, project_id, action," - " detail, created_at)" - " VALUES(?, ?, ?, ?, ?)", - (user_id, project_id, action, detail, now), - ) - _ledger_append(self.conn, user_id, project_id, action, - detail=detail, created_at=now) + audit_on_conn(self.conn, user_id, project_id, action, + detail=detail, created_at=now) self.conn.commit() return except Exception as exc: diff --git a/cloud/automations.py b/cloud/automations.py index 291e582..1c0d00f 100644 --- a/cloud/automations.py +++ b/cloud/automations.py @@ -38,13 +38,20 @@ id 记在 automation_runs.task_id 里，派发与任务绑定，不用 为自动化另起一套通知系统)。 -本模块只依赖标准库与同为标准库叶子的 cloud.quota，不 -import 其他 cloud 模块，避免循环 import；表迁移写法仿 -peer_messages(CREATE TABLE IF NOT EXISTS + PRAGMA 查列补齐， -SQLite 与 Postgres 适配层同款)。 +本模块的 cloud 依赖只有两处叶子引用，避免循环 import： +配额判定走同为标准库叶子的 cloud.quota；审计行上链 +（方案 A，见 run_due_automation）只走 cloud.auth.audit_on_conn—— +与 AuthStore.audit() 同一对写（audit_log 行 + audit_ledger +链条目，同 created_at），不再有 audit_log 直写旁路。 +auth 只依赖 db / ledger 两个叶模块，三者都不 import +automations / server / worker，不构成循环 import；其余 SQL +（配额口径、表迁移）仍就地实现，不另 import 业务模块。 +表迁移写法仿 peer_messages(CREATE TABLE IF NOT +EXISTS + PRAGMA 查列补齐，SQLite 与 Postgres 适配层同款）。 """ import os +import sqlite3 import time import uuid from datetime import datetime, timezone @@ -53,6 +60,7 @@ from cloud.quota import decide as _quota_decide from cloud.quota import default_quota_usd_micros as _quota_default from cloud.quota import monthly_usage as _quota_monthly_usage from cloud.quota import quota_for_owner as _quota_for_owner +from cloud.auth import audit_on_conn as _audit_on_conn MIN_INTERVAL_SECONDS = 60 MAX_INTERVAL_SECONDS = 30 * 86400 @@ -595,51 +603,19 @@ def _automation_quota_exceeded(conn, owner): return None -def _audit_row(conn, project_id, action, detail, now): - """best-effort 写审计行：失败不影响自动化运行本身。 +def _is_integrity_error(exc): + """链头唯一冲突判定：与 cloud/ledger.py _is_integrity_error + 同规则（sqlite3.IntegrityError，或 PG 适配层映射后的 + PostgresIntegrityError，按名判定，驱动类不跨模块 import）。 - 旧实现显式插 TEXT uuid 当 id：SQLite 下能跑，但生产 PG 的 - audit_log.id 是 BIGINT GENERATED BY DEFAULT AS IDENTITY， - 插 TEXT 直接报错、审计行全丢（best-effort 吞掉后无声无息）。 - 修法：INSERT 一律不带 id，让两端自增/identity 生成； - 表不存在时先按 PG 口径建、再按 SQLite 口径建、逐个重试， - 哪边建表成功就用哪边——automations 按设计不 import 其它 - cloud 模块，所以用“先插、失败再按口径重试”做方言探测， - 不靠 import 判断后端。 + 只有这类错误值得在事务边界重试整段；其它错误重试无用， + 必须立即回滚抛错，不能被重试掩盖成“锁/冲突问题”。 + 旧 _audit_row 已删除：本模块不再有任何 audit_log 直写， + 审计只走 _audit_on_conn（见模块文档）。 """ - params = ("system", project_id, action, detail, now) - try: - conn.execute( - "INSERT INTO audit_log(user_id, project_id, action," - " detail, created_at) VALUES(?, ?, ?, ?, ?)", params) - conn.commit() - return - except Exception: - try: - conn.rollback() - except Exception: - pass - for ddl in ( - "CREATE TABLE IF NOT EXISTS audit_log(" - "id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, " - "user_id TEXT, project_id TEXT, action TEXT, detail TEXT, " - "created_at DOUBLE PRECISION)", - "CREATE TABLE IF NOT EXISTS audit_log(" - "id INTEGER PRIMARY KEY AUTOINCREMENT, " - "user_id TEXT, project_id TEXT, action TEXT, detail TEXT, " - "created_at REAL)"): - try: - conn.execute(ddl) - conn.execute( - "INSERT INTO audit_log(user_id, project_id, action," - " detail, created_at) VALUES(?, ?, ?, ?, ?)", params) - conn.commit() - return - except Exception: - try: - conn.rollback() - except Exception: - pass + if isinstance(exc, sqlite3.IntegrityError): + return True + return type(exc).__name__ == "PostgresIntegrityError" def run_due_automation(conn, automation_id, now=None): @@ -651,9 +627,22 @@ def run_due_automation(conn, automation_id, now=None): 失败重试循环）并记审计，返回该产出；配额内才 INSERT 云 端任务（queued，executor/model 留 NULL 走 worker 默认， 与服务端建任务缺省行为一致），产出记 created 并记审计。 - 任务 INSERT 与产出记录由 record_automation_run 的 commit - 一并落盘。调用方（未来的定时执行器）只调本函数，绝不 + 调用方（未来的定时执行器）只调本函数，绝不 自己拼 claim + INSERT，否则配额门可被绕过。 + + 审计失败语义（方案 A，旧 best-effort _audit_row 已删除）： + 审计对写（audit_log 行 + audit_ledger 链条目，同 created_at + = now，走 cloud.auth.audit_on_conn）先于产出记录的 commit + 执行，任务行、产出记录行、审计对行同事务提交：审计失败 + 时三者一起回滚，异常抛给执行器——不记“任务已建、审计缺 + 失”的半截状态，也不把审计失败记成 failed 产出、不进失败 + 重试循环。抢单已单独提交（last_run_at 已写），按 claim 的 + 文档，这次执行等下个间隔再跑，与“抢单后崩掉等下个间隔 + 补跑”一致。链头唯一冲突时整段回滚重试，最多 5 次（与 + AuthStore.audit 同数）；其它错误立即回滚抛错，不重试。 + 审计行 user_id 仍为 "system"，detail 格式不变 + （created："<自动化id> -> <任务id>"；skipped： + "<自动化id>: <配额门文本>"）。 """ if now is None: now = time.time() @@ -670,46 +659,62 @@ def run_due_automation(conn, automation_id, now=None): prompt = row_get(auto, "prompt") owner = row_get(auto, "created_by") gate = _automation_quota_exceeded(conn, owner) - if gate is not None: - run = record_automation_run( - conn, automation_id, project_id, task_id=None, - status="skipped_quota", detail=gate, now=now) - _audit_row(conn, project_id, "automation_run_quota_skipped", - "%s: %s" % (automation_id, gate), now) - return run task_id = uuid.uuid4().hex - # 任务列补齐（与 ensure_automations_tables 的补列惯例同款）： - # 裸库/老库的 tasks 表可能没有 executor 等列，INSERT 前补齐， - # 否则建任务直接报错；PG 适配层会翻译 PRAGMA 查询。 - existing_tasks = _existing_columns(conn, "tasks") - for col, ddl in ( - ("executor", "TEXT"), - ("model", "TEXT"), - ("effort", "TEXT"), - ("environment_id", "TEXT"), - ("parent_task_id", "TEXT"), - ("coordinator_part", "TEXT")): - if col not in existing_tasks: + last_exc = None + for _attempt in range(5): + try: + if gate is not None: + _audit_on_conn( + conn, "system", project_id, + "automation_run_quota_skipped", + "%s: %s" % (automation_id, gate), created_at=now) + return record_automation_run( + conn, automation_id, project_id, task_id=None, + status="skipped_quota", detail=gate, now=now) + # 任务列补齐（与 ensure_automations_tables 的补列惯例 + # 同款）：裸库/老库的 tasks 表可能没有 executor 等列， + # INSERT 前补齐，否则建任务直接报错；PG 适配层会翻译 + # PRAGMA 查询。放在重试循环内：回滚会撤销未提交的 + # 补列，重试时须重新查列补齐，不能只补一次。 + existing_tasks = _existing_columns(conn, "tasks") + for col, ddl in ( + ("executor", "TEXT"), + ("model", "TEXT"), + ("effort", "TEXT"), + ("environment_id", "TEXT"), + ("parent_task_id", "TEXT"), + ("coordinator_part", "TEXT")): + if col not in existing_tasks: + try: + conn.execute( + "ALTER TABLE tasks ADD COLUMN %s %s" + % (col, ddl)) + except Exception: + pass + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " idempotency_key, depends_on, priority, created_by," + " created_at, updated_at, last_heartbeat, executor," + " model, effort, environment_id, parent_task_id," + " coordinator_part)" + " VALUES(?, ?, ?, 'queued', NULL, '[]', 0, ?, ?, ?," + " NULL, NULL, NULL, NULL, NULL, NULL, NULL)", + (task_id, project_id, prompt, owner, now, now)) + _audit_on_conn( + conn, "system", project_id, "automation_run_created", + "%s -> %s" % (automation_id, task_id), created_at=now) + return record_automation_run( + conn, automation_id, project_id, task_id=task_id, + status="created", detail="", now=now) + except Exception as exc: + last_exc = exc try: - conn.execute( - "ALTER TABLE tasks ADD COLUMN %s %s" % (col, ddl)) + conn.rollback() except Exception: pass - conn.execute( - "INSERT INTO tasks(id, project_id, prompt, status," - " idempotency_key, depends_on, priority, created_by," - " created_at, updated_at, last_heartbeat, executor," - " model, effort, environment_id, parent_task_id," - " coordinator_part)" - " VALUES(?, ?, ?, 'queued', NULL, '[]', 0, ?, ?, ?," - " NULL, NULL, NULL, NULL, NULL, NULL, NULL)", - (task_id, project_id, prompt, owner, now, now)) - run = record_automation_run( - conn, automation_id, project_id, task_id=task_id, - status="created", detail="", now=now) - _audit_row(conn, project_id, "automation_run_created", - "%s -> %s" % (automation_id, task_id), now) - return run + if not _is_integrity_error(exc): + raise + raise last_exc def list_automation_runs(conn, automation_id, limit=TRIAGE_LIMIT): diff --git a/cloud/tests/test_audit_automations_chain.py b/cloud/tests/test_audit_automations_chain.py new file mode 100644 index 0000000..73ec26e --- /dev/null +++ b/cloud/tests/test_audit_automations_chain.py @@ -0,0 +1,260 @@ +"""自动化执行器审计上链（缺口 4，方案 A）单测。 + +钉住的不是“写了审计行”，而是“审计行与链条目成对、与业务行 +同事务”： + +- created / skipped 两路：audit_log 行与 audit_ledger 行同 + created_at、同 user_id/action/detail，verify_chain ok。 +- 链头唯一冲突：整段（任务 + 产出 + 审计对）回滚重试一次后 + 成功，最终各恰好一行，不出现任务已建、审计缺失的半截态。 +- 非唯一冲突的审计失败：fail-closed 抛错，任务 + 产出 + 审计 + 对全部回滚（0 行），只有已单独提交的抢单 last_run_at 留下， + 下个间隔可补跑——不是吞错、不是 failed 产出。 +- audit_on_conn 与 AuthStore.audit() 是同一对写：前者写的行 + 能作为后者的链前驱，prev 衔接、verify_chain ok。 +- 同间隔内第二次跑不写任何新行。 +""" + +import os +import re +import sqlite3 +import tempfile +import time +import unittest +from pathlib import Path + +from cloud.auth import AuthStore, audit_on_conn +from cloud.automations import ( + create_automation, + ensure_automations_tables, + run_due_automation, +) +from cloud.ledger import GENESIS_HASH, verify_chain +from cloud.server import init_db + + +def _count(conn, sql, params=()): + return conn.execute(sql, params).fetchone()[0] + + +class _ProxyConn: + """把除指定拦截外的调用原样委托给真实连接的代理。 + + run_due_automation 只用 conn.execute / commit / rollback， + 代理这三者即足以模拟链写失败；其余属性（如 row_factory） + 不用代理——调用方不读它们。 + """ + + def __init__(self, real): + self._real = real + + def execute(self, sql, params=()): + return self._real.execute(sql, params) + + def commit(self): + return self._real.commit() + + def rollback(self): + return self._real.rollback() + + +class _FlakyLedgerConn(_ProxyConn): + """首个 audit_ledger INSERT 报唯一冲突，之后放行。""" + + def __init__(self, real): + super().__init__(real) + self.flaked = False + + def execute(self, sql, params=()): + if (not self.flaked + and sql.strip().upper().startswith( + "INSERT INTO AUDIT_LEDGER")): + self.flaked = True + raise sqlite3.IntegrityError( + "UNIQUE constraint failed: audit_ledger.entry_hash") + return super().execute(sql, params) + + +class _BrokenLedgerConn(_ProxyConn): + """全部 audit_ledger INSERT 报非唯一冲突错误。""" + + def execute(self, sql, params=()): + if sql.strip().upper().startswith("INSERT INTO AUDIT_LEDGER"): + raise RuntimeError("ledger storage boom") + return super().execute(sql, params) + + +class AutomationChainTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + init_db(self.db_path) + self.conn = sqlite3.connect(self.db_path) + self.conn.row_factory = sqlite3.Row + ensure_automations_tables(self.conn) + + def tearDown(self): + self.conn.close() + self.tmp.cleanup() + + def _auto(self, owner="owner1"): + return create_automation(self.conn, "proj", "daily", "do thing", + owner, interval_seconds=60) + + def _set_quota(self, user_id, micros): + self.conn.execute( + "CREATE TABLE IF NOT EXISTS user_quotas(" + "user_id TEXT PRIMARY KEY, quota_usd_micros BIGINT, " + " updated_at DOUBLE PRECISION)") + self.conn.execute( + "INSERT OR REPLACE INTO user_quotas(user_id," + " quota_usd_micros, updated_at) VALUES(?, ?, ?)", + (user_id, micros, time.time())) + self.conn.commit() + + def _pair_counts(self): + # 缺表计 0：SQLite 的 DDL 可事务回滚，首跑审计失败回滚时 + # 会连 audit_on_conn 刚建的 audit_log / audit_ledger 表一起 + # 撤销——下次跑由 CREATE IF NOT EXISTS 重建，不是丢数据。 + out = [] + for table in ("tasks", "automation_runs", "audit_log", + "audit_ledger"): + try: + out.append(_count( + self.conn, "SELECT COUNT(*) FROM %s" % table)) + except sqlite3.OperationalError: + out.append(0) + return tuple(out) + + def _check_pair(self, action, detail): + audit = self.conn.execute( + "SELECT user_id, project_id, action, detail, created_at" + " FROM audit_log WHERE action = ?", (action,)).fetchone() + self.assertIsNotNone(audit) + self.assertEqual(audit[0], "system") + self.assertEqual(audit[1], "proj") + self.assertEqual(audit[3], detail) + entry = self.conn.execute( + "SELECT user_id, project_id, task_id, action, detail," + " created_at, prev_hash, entry_hash FROM audit_ledger" + " WHERE action = ?", (action,)).fetchone() + self.assertIsNotNone(entry) + self.assertEqual(entry[0], "system") + self.assertEqual(entry[1], "proj") + self.assertEqual(entry[3], action) + self.assertEqual(entry[4], detail) + self.assertEqual(entry[5], audit[4]) + result = verify_chain(self.conn) + self.assertTrue(result["ok"], result) + return audit, entry + + def test_created_run_writes_chained_pair(self): + auto = self._auto() + run = run_due_automation(self.conn, auto["id"]) + self.assertIsNotNone(run) + self.assertEqual(run["status"], "created") + self.assertEqual(self._pair_counts(), (1, 1, 1, 1)) + self._check_pair("automation_run_created", + "%s -> %s" % (auto["id"], run["task_id"])) + + def test_skipped_run_writes_chained_pair(self): + auto = self._auto() + self._set_quota("owner1", 0) + run = run_due_automation(self.conn, auto["id"]) + self.assertIsNotNone(run) + self.assertEqual(run["status"], "skipped_quota") + self.assertIsNone(run["task_id"]) + self.assertEqual(self._pair_counts(), (0, 1, 1, 1)) + audit, _entry = self._check_pair( + "automation_run_quota_skipped", + "%s: %s" % (auto["id"], run["detail"])) + self.assertIn("quota", audit[3]) + + def test_chain_head_race_retries_whole_block_once(self): + auto = self._auto() + proxy = _FlakyLedgerConn(self.conn) + run = run_due_automation(proxy, auto["id"]) + self.assertTrue(proxy.flaked) + self.assertIsNotNone(run) + self.assertEqual(run["status"], "created") + # 重试后仍恰好各一行：回滚掉了第一次的任务 + 审计行， + # 没有“任务两行、审计一行”或反过来的半截态。 + self.assertEqual(self._pair_counts(), (1, 1, 1, 1)) + self._check_pair("automation_run_created", + "%s -> %s" % (auto["id"], run["task_id"])) + # 抢单在重试循环外、已单独提交，不因回滚丢失。 + claimed = self.conn.execute( + "SELECT last_run_at FROM automations WHERE id = ?", + (auto["id"],)).fetchone()[0] + self.assertIsNotNone(claimed) + + def test_non_integrity_audit_failure_rolls_back_all(self): + auto = self._auto() + proxy = _BrokenLedgerConn(self.conn) + with self.assertRaises(RuntimeError): + run_due_automation(proxy, auto["id"]) + # 任务、产出、审计对全部回滚：不是吞错后留半截。 + self.assertEqual(self._pair_counts(), (0, 0, 0, 0)) + # 抢单已提交：下个间隔可补跑，这次不记 failed 产出。 + claimed = self.conn.execute( + "SELECT last_run_at FROM automations WHERE id = ?", + (auto["id"],)).fetchone()[0] + self.assertIsNotNone(claimed) + + def test_second_run_same_interval_writes_nothing_more(self): + auto = self._auto() + run = run_due_automation(self.conn, auto["id"]) + self.assertIsNotNone(run) + self.assertIsNone(run_due_automation(self.conn, auto["id"])) + self.assertEqual(self._pair_counts(), (1, 1, 1, 1)) + + def test_audit_on_conn_chains_with_auth_store_audit(self): + conn = sqlite3.connect(self.db_path) + try: + audit_on_conn(conn, "u1", "p1", "manual_action", "d", + created_at=123.0) + conn.commit() + finally: + conn.close() + store = AuthStore(self.db_path) + try: + store.audit("u2", "p1", "second_action", "e") + result = verify_chain(store.conn) + self.assertTrue(result["ok"], result) + entries = store.conn.execute( + "SELECT action, prev_hash FROM audit_ledger" + " ORDER BY id").fetchall() + self.assertEqual(len(entries), 2) + self.assertEqual(entries[0][0], "manual_action") + self.assertEqual(entries[0][1], GENESIS_HASH) + first_hash = store.conn.execute( + "SELECT entry_hash FROM audit_ledger ORDER BY id" + " LIMIT 1").fetchone()[0] + self.assertEqual(entries[1][1], first_hash) + finally: + store.conn.close() + + +class NoAuditBypassTestCase(unittest.TestCase): + """生产代码里 audit_log 的 INSERT 只允许出现在 auth.py。 + + 这是缺口 4 的“无旁路”自证：将来任何模块再起 audit_log + 直写（哪怕只是 best-effort），本测试即红，须先走 + audit_on_conn / AuthStore.audit 才能合入。测试固件里的 + 直写（test_tamper_ledger 的篡改模拟、test_chat_history 的 + 旧行模拟）不在 cloud/*.py 内，不在本断言范围。 + """ + + def test_only_auth_writes_audit_log(self): + cloud_dir = Path(__file__).resolve().parents[1] + offenders = [] + for path in sorted(cloud_dir.glob("*.py")): + text = path.read_text(encoding="utf-8") + if re.search(r"INSERT\s+INTO\s+audit_log", text, + re.IGNORECASE): + offenders.append(path.name) + self.assertEqual(offenders, ["auth.py"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_automation_run_due.py b/cloud/tests/test_automation_run_due.py index 77150a3..3b5cfe3 100644 --- a/cloud/tests/test_automation_run_due.py +++ b/cloud/tests/test_automation_run_due.py @@ -5,13 +5,14 @@ import tempfile import time import unittest +from cloud.auth import audit_on_conn from cloud.automations import ( - _audit_row, create_automation, ensure_automations_tables, list_automation_runs, run_due_automation, ) +from cloud.ledger import GENESIS_HASH, compute_entry_hash from cloud.server import init_db @@ -49,25 +50,48 @@ def _set_quota(db_path, user_id, micros): conn.close() -class AuditRowPgIdentityTestCase(unittest.TestCase): +class AuditPairPgIdentityTestCase(unittest.TestCase): """模拟 PG：id 是 BIGINT identity，显式插 TEXT uuid 会报错。 - 假连接只接受不带 id 的 INSERT；旧实现（带 TEXT id）在此 - 假连接上应失败、新实现应成功且审计行落盘。 + 假连接只接受不带 id 的 INSERT；旧 _audit_row（带 TEXT id） + 在此假连接上应失败。接替它的 audit_on_conn 须同时做到： + audit_log 行不带 id、链条目与该行同 created_at、空链时 + prev 为 GENESIS、entry_hash 按规范载荷重算一致；且本函数 + 不 commit（事务属于调用方，见 run_due_automation 文档）。 + + 假连接类名须为 PgConnection：ledger.ensure_schema 靠类名 + 判定 PG 连接以跳过 SQLite 触发器，与真实 PG 适配层同款。 """ - class _PgLikeConn: + class _FakeCursor: + def __init__(self, row=None): + self._row = row + self.lastrowid = 1 + + def fetchone(self): + return self._row + + class PgConnection: def __init__(self): - self.rows = [] + self.audit_rows = [] + self.ledger_rows = [] self.committed = False def execute(self, sql, params=()): - if sql.strip().upper().startswith("INSERT INTO AUDIT_LOG(ID"): + upper = sql.strip().upper() + if upper.startswith("INSERT INTO AUDIT_LOG(ID"): raise ValueError("column \"id\" is of type bigint") - if sql.strip().upper().startswith("INSERT INTO AUDIT_LOG"): - self.rows.append(params) - return None - raise ValueError("unexpected sql in fake PG conn") + if upper.startswith("INSERT INTO AUDIT_LOG"): + self.audit_rows.append(params) + return AuditPairPgIdentityTestCase._FakeCursor() + if upper.startswith("SELECT ENTRY_HASH FROM AUDIT_LEDGER"): + return AuditPairPgIdentityTestCase._FakeCursor(None) + if upper.startswith("INSERT INTO AUDIT_LEDGER"): + self.ledger_rows.append(params) + return AuditPairPgIdentityTestCase._FakeCursor() + if upper.startswith("CREATE TABLE"): + return AuditPairPgIdentityTestCase._FakeCursor() + raise ValueError("unexpected sql in fake PG conn: %r" % sql) def commit(self): self.committed = True @@ -75,15 +99,33 @@ class AuditRowPgIdentityTestCase(unittest.TestCase): def rollback(self): pass - def test_insert_without_id_succeeds_on_pg_like_conn(self): - conn = self._PgLikeConn() - _audit_row(conn, "p1", "automation_run_created", - "auto1 -> task1", 123.0) - self.assertTrue(conn.committed) - self.assertEqual(len(conn.rows), 1) + def test_pair_without_id_succeeds_on_pg_like_conn(self): + conn = self.PgConnection() + audit_on_conn(conn, "system", "p1", "automation_run_created", + "auto1 -> task1", created_at=123.0) + # 调用方拥有事务：本函数自己不 commit。 + self.assertFalse(conn.committed) + self.assertEqual(len(conn.audit_rows), 1) # 参数里没有 uuid：第一位是 system，第二位是 project_id。 - self.assertEqual(conn.rows[0][0], "system") - self.assertEqual(conn.rows[0][1], "p1") + self.assertEqual(conn.audit_rows[0][0], "system") + self.assertEqual(conn.audit_rows[0][1], "p1") + self.assertEqual(conn.audit_rows[0][4], 123.0) + self.assertEqual(len(conn.ledger_rows), 1) + ledger = conn.ledger_rows[0] + # audit_ledger 列序：user_id, project_id, task_id, action, + # detail, created_at, prev_hash, entry_hash。 + self.assertEqual(ledger[0], "system") + self.assertEqual(ledger[1], "p1") + self.assertIsNone(ledger[2]) + self.assertEqual(ledger[3], "automation_run_created") + self.assertEqual(ledger[4], "auto1 -> task1") + self.assertEqual(ledger[5], 123.0) + self.assertEqual(ledger[6], GENESIS_HASH) + self.assertEqual( + ledger[7], + compute_entry_hash(GENESIS_HASH, "system", "p1", None, + "automation_run_created", + "auto1 -> task1", 123.0)) class RunDueAutomationTestCase(unittest.TestCase):