diff --git a/cloud/auth.py b/cloud/auth.py index 97a44ce..dd7a6e4 100644 --- a/cloud/auth.py +++ b/cloud/auth.py @@ -24,6 +24,8 @@ approve_merge -> maintainer，manage_project -> owner（仅 owner）。 import hashlib import secrets from cloud.db import connect as _db_connect +from cloud.ledger import append_entry as _ledger_append +from cloud.ledger import ensure_schema as _ledger_ensure import time import uuid @@ -68,6 +70,9 @@ class AuthStore: "detail TEXT, " "created_at REAL)" ) + # 防篡改账本（哈希链）：audit_log 本身不动，链写进 + # audit_ledger；建表失败即建库失败，不静默丢链。 + _ledger_ensure(self.conn) self.conn.commit() def create_user(self, username): @@ -349,13 +354,38 @@ class AuthStore: ) def audit(self, user_id, project_id, action, detail=""): - """只追加一条审计记录。""" - self.conn.execute( - "INSERT INTO audit_log(user_id, project_id, action, detail, created_at)" - " VALUES(?, ?, ?, ?, ?)", - (user_id, project_id, action, detail, time.time()), - ) - self.conn.commit() + """只追加一条审计记录，并同步写哈希链（同事务）。 + + audit_log 行与 audit_ledger 链条目同 created_at、同事务 + 提交：链写失败时整对回滚并抛错（fail-closed，绝不留 + “audit_log 有行、链上无行”的静默缺口）。并发抢链头 + 导致唯一冲突时在事务边界重试整对写入，最多 5 次； + 其它错误直接回滚抛错，不重试。 + """ + 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) + self.conn.commit() + return + except Exception as exc: + last_exc = exc + try: + self.conn.rollback() + except Exception: + pass + if not isinstance(exc, sqlite3.IntegrityError) \ + and type(exc).__name__ != "PostgresIntegrityError": + raise + raise last_exc def list_audit(self, project_id, limit=100): """按 created_at 倒序返回该项目的审计列表（dict）。""" diff --git a/cloud/cost.py b/cloud/cost.py index 5011e40..308be1b 100644 --- a/cloud/cost.py +++ b/cloud/cost.py @@ -135,6 +135,8 @@ def price_usd_micros(model, input_tokens, cached_input_tokens, return int(round(usd * 1000000)) +# 双列文案：逐字复用 PR #212（feat-cost-quota，5b30c61）的 +# BILLING_NOTE，本轨不拥有成本口径、不另造一套说法。 BILLING_NOTE = ( "价格表代理价（经 CLOUD_MODEL_PRICES_JSON / CLOUD_MODEL_PRICES_FILE" " 覆盖层），非中转实际账单；未定价 token 不计入美元、不等于免费。" @@ -168,6 +170,24 @@ def quota_cutoff_reason(quota_usd_micros, used_usd_micros, return None +def dual_column_view(report): + """双列视图：在已有 _report 结果上加定价列别名与账单说明。 + + 不改 _report（旧单测按其键集断言）与旧键：usd_micros 仍是 + 有价行之和，unpriced_actual_tokens 仍是未定价 actual token； + 本函数只加 priced_usd_micros（与 usd_micros 同值）与 + billing_note。未定价模型分列（by_model / unpriced_models） + 与按创建者口径是 PR #212 的范围，本轨不重复实现——#212 + 合并后本函数由其 dual_column_* 取代，端点改走 dual_column_* + （同名同值键，切换无语义差）；#212 未合并前，本视图是 + 任务/项目成本端点的双列最小实现。 + """ + view = dict(report) + view["priced_usd_micros"] = view["usd_micros"] + view["billing_note"] = BILLING_NOTE + return view + + def unpriced_token_cap(): """Monthly unpriced-token ceiling for users who have a quota row. diff --git a/cloud/db.py b/cloud/db.py index bc33792..15442b7 100644 --- a/cloud/db.py +++ b/cloud/db.py @@ -426,7 +426,8 @@ _OR_REPLACE_PK = { } _IDENTITY_TABLES = frozenset( - {"summaries", "kb_docs", "cost_records", "audit_log"}) + {"summaries", "kb_docs", "cost_records", "audit_log", + "audit_ledger"}) _OR_REPLACE_RE = _re.compile( r"^\s*INSERT\s+OR\s+REPLACE\s+INTO\s+(\w+)\s*\(([^)]+)\)\s*" @@ -804,6 +805,19 @@ POSTGRES_SCHEMA = ( "action TEXT, " "detail TEXT, " "created_at DOUBLE PRECISION)", + # 防篡改账本（哈希链，第 4 轨）：与 cloud/ledger.py 的 SQLite + # DDL 同表同列；prev/entry_hash UNIQUE 是并发抢链头的 + # fail-closed 基础，绝不能去掉。 + "CREATE TABLE IF NOT EXISTS audit_ledger(" + "id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, " + "user_id TEXT, " + "project_id TEXT, " + "task_id TEXT, " + "action TEXT, " + "detail TEXT, " + "created_at DOUBLE PRECISION, " + "prev_hash TEXT NOT NULL UNIQUE, " + "entry_hash TEXT NOT NULL UNIQUE)", # P0 三件：断点与步骤事实源（PG 新库一次建齐；老库走 # server/worker 的 ensure 幂等补表，不丢数据）。 # task_checkpoints：每轮断点快照（worker 镜像线程 + 终局 diff --git a/cloud/ledger.py b/cloud/ledger.py new file mode 100644 index 0000000..3afec4e --- /dev/null +++ b/cloud/ledger.py @@ -0,0 +1,1393 @@ +"""防篡改审计账本：哈希链（审计加固第 4 轨）。 + +方案取舍（二选一，本轨选 ①，②被否原因见下）： +- ① 哈希链（本模块）：每条记录存 ``prev_hash``（上一条的 + ``entry_hash``）与 ``entry_hash``（本条规范载荷的 SHA-256）。 + 改一条历史记录 → 重算哈希与存量不一致，校验失败；删一条 + 中间记录 → 下一条的 ``prev_hash`` 悬空，校验失败；删尾巴 → + 链自洽但头哈希变了，需与此前导出的头哈希比对才能发现 + （见 ``verify_chain(expected_head=...)`` / ``verify_export``）。 +- ② 只追加约束 + 外部导出（被否）：触发器只能挡“走正常 SQL + 的误写”，有写库权限的攻击者可先删触发器再改数据；导出若 + 没有逐条哈希绑定，改库后重新导出看起来与原导出无异—— + 检测能力依赖导出时点而非记录本身。本模块仍提供导出 + （``export_chain``），但导出只是头哈希的外部锚点分发手段， + 检测能力来自哈希链本身，不是“②独立落地”。 + +诚实边界： +- 链防的是“事后悄悄改历史被当成没发生”，不是防超级用户： + 能删触发器、能重建整条链并伪造头哈希的人，任何库内机制 + 都拦不住；可信锚点必须落在库外（导出的头哈希定期存到 + 库外，如运维笔记、另一系统、离线备份），校验时比对。 +- 尾巴删除单靠库内校验发现不了：没有“之后还有记录”的证据 + 时，截断后的链自洽。这是哈希链的固有局限，不假装解决； + 对策就是导出头哈希 + 定期校验（头哈希变了或期望头对不上 + 即告警）。 +- 并发：``prev_hash`` / ``entry_hash`` 均 UNIQUE。两路并发追加 + 读到同一头时，一路因唯一冲突失败——调用方必须在事务边界 + 重试整对写入（见 ``AuthStore.audit`` 的重试循环），绝不能 + 靠放宽约束让链分叉。 +- ``created_at`` 一律按 float 规范化后入哈希：PG 的 REAL 是 + float4 会丢精度，本表在两端都用 DOUBLE PRECISION，入库前 + 也先转 float，避免 int/float 字面量差异导致重算失配。 + +历史迁移（绝不悄悄重写历史）： +- 老 ``audit_log`` 行没有哈希，迁移时不给它们补哈希、不改 + 它们、不往 ``audit_log`` 加列。``migrate_legacy_audit_log`` + 只在链为空且老表有行时写一条 ``ledger_genesis`` 标记条目， + 其 detail 记老表行数与 id/时间边界，声明“标记之前的一律 + 是 pre-chain、未上链”。链已启动后迁移函数拒绝执行。 +- 链为空且老表也有 0 行时不写标记，首条真实记录直接以 + ``GENESIS_HASH`` 为 prev。 + +第 3 轨自动推回接线口（本轨只定义接口，第 3 轨尚未落地）： +- ``append_pushback_round``：每轮推回在账本里留痕，字段见 + 函数 docstring。推时写一条（round_status="pushed"），重审 + 后再写一条（同 round_no，round_status="re_reviewed"，带 + fix_diff 与 re_review_result）——账本只追加不修改，一轮的 + 两条记录即完整轨迹。 +- 注意：这是本轨提出的契约，第 3 轨落地时须按此字段名接线； + 若第 3 轨已定其它字段名，以第 3 轨为准、本模块加兼容层， + 摘要里会写清不一致处（当前：无已落地实现可比对）。 + +第 3 轨口径同步（失败计数 / 推倒重做 / 时间线，本轨提案）： +- 三种条目（action 区分，同一条链、同一时间线）： + ``audit_verdict``（一次审核或验收结论）、 + ``audit_pushback_round``（一轮自动推回）、 + ``audit_rebuild``（一次推倒重做判定 / 完成）。 +- 字段表（严格类型、枚举值见各函数；第 3 轨须用这些 + 字段名与枚举值，否则计数与时间线分组对不上）： + verdict: module_id, contract_version, verdict=pass|fail, + source=audit|acceptance, failures[]; + pushback: round_no（推回轮次，1-3；>3 应走推倒，时间线 + 会标异常）、trigger_items[], pushback_package{}, + fix_diff{summary,diff_sha256,files_changed}|None, + re_review_result{verdict}|None, round_status=pushed| + re_reviewed, module_id, contract_version, + failure_source=audit|acceptance|None, + failure_counts{audit_fails,acceptance_fails,total_fails}|None; + rebuild: module_id, contract_version, rebuild_no>=1, + old_impl_ref{commit|tag|archive_ref 至少其一}, + new_impl_ref 同形或 None, root_cause=implementation| + contract_design, status=decided|rebuilt, + prior_failure_total|None。 +- 计数规则（由账本派生，不采信调用方自报的数字）： + 按 (module_id, contract_version) 计；一次 pass 清零； + 合同版本变了即新 key、重新计；一次已完成的重做 + （status=rebuilt）同样重置计数基线（新实现从零计）； + total = 自基线以来的 audit fail + acceptance fail； + total > 3（即第 4 次仍没通过）触发推倒，不再发第 4 轮 + 推回。推倒只影响该 module_id 的条目，不动其它模块。 +- 根因分流：root_cause=implementation 才走代码推倒重做； + root_cause=contract_design 不自动重做代码（转 G2 重新冻结 + 并升级人工复核或用户），账本只记录该判定，不授权重做。 +- 时间线复原：``audit_timeline`` 按 id 升序返回三种条目 + 的解析后序列与汇总；账本只追加，失败轮次永远留在链上， + 最终 pass 只是在序列末尾多一条 pass，绝不抹掉前面。 + +三条收紧（本轨提案，第 3 轨落地时按此接线）： +- 第三次失败先判根因（``audit_third_failure_assessment``）： + determination=early_rebuild（结构性错误，提前推倒， + max_extra_rounds=0）或 continuation_round（接近完成只差 + 收尾，再给 1 轮 targeted 修复，max_extra_rounds=1，再 fail + 即推倒）；basis 必须附判定依据（建议键 audit_conclusion、 + failure_items、diff_scope、passed_items、 + continuation_counting——续轮计数方式），事后可抽查。 + assessment 正常在 failure_total=3 时写；时间线会把 + failure_total != 3 的 assessment 标为异常。 +- 推倒保留资产（rebuild 条目的 retained_assets dict）： + old_test_set_ref（旧测试集引用）、passed_acceptance_items + （已通过验收项清单）、old_new_impl_mapping（新旧实现 + 对应关系）、old_tests_status=pass|fail|pending（新实现 + 第一道门——旧测试全过情况；过不了即重做失败，调用方须 + 另写一条 verdict fail 计入失败计数，账本不自动补写）。 + 推倒只推倒实现代码：测试、已通过验收项、合同、设计、 + 审计结论全部保留（账本条目本身即保留证据）。 +- 删改测试单独审计（``audit_test_change``）：test_changes + {added,modified,deleted}、auditor（经谁审计，必填）、 + contract_refrozen（是否走合同重新冻结，严格 bool）、 + refreeze_ref（重新冻结的合同版本或决策引用，refrozen + 时必填）。硬门：modified/deleted 非空且 contract_refrozen + 为 False 时拒绝写入——删改测试不走重新冻结的记录本身 + 就是违规，不应进账本当成合规记录（违规尝试请走 + verdict fail 或违规上报另记）。 +- 审计的审计（``audit_meta_review``）：事后抽查发现判错时 + 记 original_entry_id（原结论的链 id）、original_verdict、 + corrected_verdict、misjudgment_found、findings、 + triggered_process（纠正后新触发的流程，如 re-review / + rebuild / none）、reviewer。计数语义：时间线保留原结论 + 与纠正两条记录（不抹）；派生计数用纠正后的有效结论 + （同一原条目的纠正以后一条为准），纠正不改变链上历史。 + G6 人工验收不能被绕过：meta-review 条目没有任何“代替 + G6”的字段与效力，G6 仍只能用户勾。 +- 验收推倒前用户通知与确认（``audit_rebuild_notice``）： + notification_content（通知内容，str 或 dict）、 + notified_party、confirmation_status=pending|confirmed| + rejected、confirmed_by（confirmed 时必填）。 + 硬门：trigger_source=acceptance 的 rebuild 写入时，账本 + 内必须已有一条同 (task_id, module_id, contract_version, + rebuild_no) 的 confirmed 通知，否则拒绝写入——未确认 + 不得重开。audit 触发的推倒不走此门。 + +PG 只追加护栏：SQLite 触发器由 ``ensure_schema`` 自动建； +PG 的函数 + 触发器 SQL 只以 ``PG_GUARD_SQL`` 常量提供， +不自动执行——本机无真实 PG，未在真实 PG 上验证触发器 +语法，生产 PG 须运维手动执行并验证后再认“已加固”。 +PG 表本身走 ``cloud/db.py`` 的 POSTGRES_SCHEMA（与老库 +初始化同款全量执行，新表自动补建）。 + +零第三方依赖，只用标准库。 +""" + +import hashlib +import json +import sqlite3 +import time + +GENESIS_HASH = hashlib.sha256(b"audit-ledger-genesis-v1").hexdigest() +LEDGER_GENESIS_ACTION = "ledger_genesis" +PUSHBACK_ACTION = "audit_pushback_round" +VERDICT_ACTION = "audit_verdict" +REBUILD_ACTION = "audit_rebuild" +ASSESSMENT_ACTION = "audit_third_failure_assessment" +TEST_CHANGE_ACTION = "audit_test_change" +META_REVIEW_ACTION = "audit_meta_review" +REBUILD_NOTICE_ACTION = "audit_rebuild_notice" +TIMELINE_ACTIONS = (VERDICT_ACTION, PUSHBACK_ACTION, REBUILD_ACTION, + ASSESSMENT_ACTION, TEST_CHANGE_ACTION, + META_REVIEW_ACTION, REBUILD_NOTICE_ACTION) + +_VERDICT_VALUES = ("pass", "fail") +_FAILURE_SOURCES = ("audit", "acceptance") +_ROOT_CAUSES = ("implementation", "contract_design") +_REBUILD_STATUSES = ("decided", "rebuilt") +_DETERMINATIONS = ("early_rebuild", "continuation_round") +_CONFIRM_STATUSES = ("pending", "confirmed", "rejected") +_OLD_TESTS_STATUSES = ("pass", "fail", "pending") +_IMPL_REF_KEYS = ("commit", "tag", "archive_ref") + +_LEDGER_TABLE_SQL = ( + "CREATE TABLE IF NOT EXISTS audit_ledger(" + "id INTEGER PRIMARY KEY AUTOINCREMENT, " + "user_id TEXT, " + "project_id TEXT, " + "task_id TEXT, " + "action TEXT, " + "detail TEXT, " + "created_at DOUBLE PRECISION, " + "prev_hash TEXT NOT NULL UNIQUE, " + "entry_hash TEXT NOT NULL UNIQUE)" +) + +# SQLite 只追加触发器（PG 不走这段，见模块 docstring）。 +_SQLITE_GUARD_SQL = ( + "CREATE TRIGGER IF NOT EXISTS audit_ledger_no_update " + "BEFORE UPDATE ON audit_ledger BEGIN " + "SELECT RAISE(ABORT, 'audit_ledger is append-only'); END", + "CREATE TRIGGER IF NOT EXISTS audit_ledger_no_delete " + "BEFORE DELETE ON audit_ledger BEGIN " + "SELECT RAISE(ABORT, 'audit_ledger is append-only'); END", +) + +# PG 只追加护栏：手动执行用，不自动执行（未在真实 PG 验证）。 +PG_GUARD_SQL = ( + "CREATE OR REPLACE FUNCTION audit_ledger_no_mutation() " + "RETURNS trigger AS $$ BEGIN " + "RAISE EXCEPTION 'audit_ledger is append-only'; END; " + "$$ LANGUAGE plpgsql", + "DROP TRIGGER IF EXISTS audit_ledger_no_update ON audit_ledger", + "CREATE TRIGGER audit_ledger_no_update BEFORE UPDATE ON audit_ledger " + "FOR EACH ROW EXECUTE FUNCTION audit_ledger_no_mutation()", + "DROP TRIGGER IF EXISTS audit_ledger_no_delete ON audit_ledger", + "CREATE TRIGGER audit_ledger_no_delete BEFORE DELETE ON audit_ledger " + "FOR EACH ROW EXECUTE FUNCTION audit_ledger_no_mutation()", +) + + +def _is_pg_conn(conn): + """PgConnection 适配连接（触发器方言不同，须跳过 SQLite 护栏）。""" + return type(conn).__name__ == "PgConnection" + + +def _is_integrity_error(exc): + if isinstance(exc, sqlite3.IntegrityError): + return True + return type(exc).__name__ in ("PostgresIntegrityError",) + + +def ensure_schema(conn): + """建 audit_ledger 表（幂等）+ SQLite 只追加触发器。 + + 不 commit：调用方在自己的事务边界提交（AuthStore.__init__ + 随后即 commit；独立调用方须自己 commit）。 + PG 适配连接跳过触发器建（方言不同，见 PG_GUARD_SQL）。 + """ + conn.execute(_LEDGER_TABLE_SQL) + if not _is_pg_conn(conn): + for stmt in _SQLITE_GUARD_SQL: + conn.execute(stmt) + + +def _canonical(prev_hash, user_id, project_id, task_id, action, detail, + created_at): + payload = { + "prev_hash": prev_hash, + "user_id": user_id or "", + "project_id": project_id or "", + "task_id": task_id or "", + "action": action or "", + "detail": detail or "", + "created_at": float(created_at), + } + return json.dumps(payload, sort_keys=True, separators=(",", ":"), + ensure_ascii=False) + + +def compute_entry_hash(prev_hash, user_id, project_id, task_id, action, + detail, created_at): + """本条 entry_hash：规范载荷的 SHA-256 hex。""" + return hashlib.sha256( + _canonical(prev_hash, user_id, project_id, task_id, action, + detail, created_at).encode("utf-8")).hexdigest() + + +def _head_hash(conn): + row = conn.execute( + "SELECT entry_hash FROM audit_ledger ORDER BY id DESC LIMIT 1" + ).fetchone() + return row[0] if row is not None else GENESIS_HASH + + +def _table_exists(conn, table): + try: + row = conn.execute( + "SELECT COUNT(*) FROM %s" % table).fetchone() + except Exception: + return False + return row is not None + + +def append_entry(conn, user_id, project_id, action, detail="", + task_id=None, created_at=None): + """在给定连接上追加一条链记录（单次尝试，不 commit、不重试）。 + + 调用方拥有事务：成功后由调用方 commit；唯一冲突（并发抢头） + 时调用方须 rollback 后重试整对写入（audit_log 行 + 本条）， + 绝不能只重试本条——回滚会连同调用方的其它写入一起丢掉， + 只重试本条会造成 audit_log 缺行、链多行。 + 返回新条目 dict（含 id/prev_hash/entry_hash）。 + """ + ensure_schema(conn) + if created_at is None: + created_at = time.time() + created_at = float(created_at) + prev_hash = _head_hash(conn) + entry_hash = compute_entry_hash( + prev_hash, user_id, project_id, task_id, action, detail, + created_at) + cur = conn.execute( + "INSERT INTO audit_ledger(user_id, project_id, task_id, action," + " detail, created_at, prev_hash, entry_hash)" + " VALUES(?, ?, ?, ?, ?, ?, ?, ?)", + (user_id, project_id, task_id, action, detail, created_at, + prev_hash, entry_hash), + ) + entry_id = cur.lastrowid + if entry_id is None: + # PG 适配在 _IDENTITY_TABLES 外的表上 lastrowid 为空时 + # 的兜底：entry_hash 唯一，可据此找回刚写的一行。 + row = conn.execute( + "SELECT id FROM audit_ledger WHERE entry_hash = ?", + (entry_hash,)).fetchone() + entry_id = row[0] if row is not None else None + return { + "id": entry_id, + "user_id": user_id, + "project_id": project_id, + "task_id": task_id, + "action": action, + "detail": detail, + "created_at": created_at, + "prev_hash": prev_hash, + "entry_hash": entry_hash, + } + + +def _rows_as_entries(rows): + entries = [] + for row in rows: + entries.append({ + "id": row[0], + "user_id": row[1], + "project_id": row[2], + "task_id": row[3], + "action": row[4], + "detail": row[5], + "created_at": row[6], + "prev_hash": row[7], + "entry_hash": row[8], + }) + return entries + + +def _check_entries(entries, expected_head=None): + """纯函数校验： linkage + 重算哈希 + 头比对。返回 failures。""" + failures = [] + prev = GENESIS_HASH + last_id = None + for entry in entries: + entry_id = entry["id"] + if last_id is not None and entry_id <= last_id: + failures.append("entries out of order at id %r" % (entry_id,)) + last_id = entry_id + if entry["prev_hash"] != prev: + failures.append( + "chain broken before id %r: prev_hash does not match" + " previous entry_hash (record deleted or reordered)" + % (entry_id,)) + # 断链后仍继续重算每条自身哈希，尽量多报问题。 + recomputed = compute_entry_hash( + entry["prev_hash"], entry["user_id"], + entry["project_id"], entry["task_id"], entry["action"], + entry["detail"], entry["created_at"]) + if recomputed != entry["entry_hash"]: + failures.append( + "entry_hash mismatch at id %r (record tampered)" + % (entry_id,)) + prev = entry["entry_hash"] + head = prev if entries else GENESIS_HASH + if expected_head is not None and head != expected_head: + failures.append( + "head mismatch: chain head %s != expected %s" + " (tail truncated or chain replaced)" + % (head, expected_head)) + return failures, head + + +def _legacy_summary(conn): + """老 audit_log 概况：行数（表不存在时为 None）。""" + if not _table_exists(conn, "audit_log"): + return None + row = conn.execute("SELECT COUNT(*) FROM audit_log").fetchone() + return int(row[0]) if row is not None else 0 + + +def verify_chain(conn, expected_head=None): + """校验库内整条链。返回 dict（ok 为 False 时 failures 逐条说明）。 + + 链为空且老 audit_log 有行 → ok=False（legacy_not_migrated）： + 必须先跑 migrate_legacy_audit_log 再认账本完整，否则老历史 + 会被误读成“链上无记录即无事发生”。 + """ + if not _table_exists(conn, "audit_ledger"): + return { + "ok": False, + "failures": ["audit_ledger table missing: ledger not" + " initialized"], + "count": 0, + "head_hash": GENESIS_HASH, + "genesis_marker": None, + "legacy_audit_rows": _legacy_summary(conn), + } + rows = conn.execute( + "SELECT id, user_id, project_id, task_id, action, detail," + " created_at, prev_hash, entry_hash FROM audit_ledger" + " ORDER BY id ASC").fetchall() + entries = _rows_as_entries(rows) + failures, head = _check_entries(entries, expected_head=expected_head) + genesis_marker = None + if entries and entries[0]["action"] == LEDGER_GENESIS_ACTION: + try: + parsed = json.loads(entries[0]["detail"] or "") + except ValueError: + parsed = None + if isinstance(parsed, dict) and parsed.get("pre_chain"): + genesis_marker = parsed + legacy = _legacy_summary(conn) + if not entries and legacy: + failures.append( + "legacy_not_migrated: %d audit_log rows predate the chain;" + " run migrate_legacy_audit_log before trusting this ledger" + % legacy) + return { + "ok": not failures, + "failures": failures, + "count": len(entries), + "head_hash": head, + "genesis_marker": genesis_marker, + "legacy_audit_rows": legacy, + } + + +def export_chain(conn, limit=None): + """导出整条链（头哈希即外部锚点）。limit 为 None 时全量。""" + if not _table_exists(conn, "audit_ledger"): + return {"version": 1, "entries": [], "count": 0, + "head_hash": GENESIS_HASH} + sql = ("SELECT id, user_id, project_id, task_id, action, detail," + " created_at, prev_hash, entry_hash FROM audit_ledger" + " ORDER BY id ASC") + params = () + if limit is not None: + sql += " LIMIT ?" + params = (int(limit),) + rows = conn.execute(sql, params).fetchall() + entries = _rows_as_entries(rows) + _failures, head = _check_entries(entries) + return {"version": 1, "entries": entries, "count": len(entries), + "head_hash": head} + + +def verify_export(data, expected_head=None): + """校验一次导出（不碰数据库）。entries 须含全部校验字段。""" + if not isinstance(data, dict) or not isinstance( + data.get("entries"), list): + return {"ok": False, + "failures": ["export shape invalid: entries missing"], + "count": 0, "head_hash": GENESIS_HASH} + entries = [] + failures = [] + for raw in data["entries"]: + try: + entries.append({ + "id": raw["id"], + "user_id": raw["user_id"], + "project_id": raw["project_id"], + "task_id": raw["task_id"], + "action": raw["action"], + "detail": raw["detail"], + "created_at": float(raw["created_at"]), + "prev_hash": raw["prev_hash"], + "entry_hash": raw["entry_hash"], + }) + except (KeyError, TypeError, ValueError): + failures.append("export entry shape invalid: %r" % (raw,)) + chain_failures, head = _check_entries(entries, + expected_head=expected_head) + failures.extend(chain_failures) + if data.get("head_hash") is not None \ + and data["head_hash"] != head: + failures.append( + "export head_hash %s != recomputed head %s" + % (data["head_hash"], head)) + return {"ok": not failures, "failures": failures, + "count": len(entries), "head_hash": head} + + +def migrate_legacy_audit_log(conn): + """老账本迁移（幂等、只执行一次、绝不改写历史）。 + + 链已有行 → 拒绝（reason=chain already started），不补、不插、 + 不改。链空且老表 0 行 → 无事可做，不写标记。链空且老表有行 + → 写一条 ledger_genesis 标记，detail 记老表边界；调用方 + commit 后，verify_chain 的 legacy_not_migrated 即消失。 + 本函数不 commit，调用方提交。 + """ + ensure_schema(conn) + existing = conn.execute( + "SELECT COUNT(*) FROM audit_ledger").fetchone() + if existing is not None and int(existing[0]) > 0: + return {"migrated": False, "reason": "chain already started", + "marker": None} + if not _table_exists(conn, "audit_log"): + return {"migrated": False, "reason": "no audit_log table", + "marker": None} + row = conn.execute( + "SELECT COUNT(*), MIN(id), MAX(id), MAX(created_at)" + " FROM audit_log").fetchone() + count = int(row[0]) if row is not None else 0 + if count == 0: + return {"migrated": False, "reason": "no legacy rows", + "marker": None} + detail = json.dumps({ + "pre_chain": True, + "legacy_audit_rows": count, + "legacy_min_id": row[1], + "legacy_max_id": row[2], + "legacy_max_created_at": row[3], + "note": "Rows before this marker live only in audit_log," + " unhashed and never rewritten; they are NOT covered" + " by the hash chain.", + }, sort_keys=True, ensure_ascii=False) + marker = append_entry( + conn, "system", "", LEDGER_GENESIS_ACTION, detail=detail, + created_at=(float(row[3]) if row[3] is not None else None)) + return {"migrated": True, "reason": None, "marker": marker} + + +def _require_non_empty_str(value, name): + if not isinstance(value, str) or not value: + raise ValueError("%s must be a non-empty string" % name) + return value + + +def _require_int_ge(value, name, minimum): + if isinstance(value, bool) or not isinstance(value, int) \ + or value < minimum: + raise ValueError("%s must be an int >= %d" % (name, minimum)) + return value + + +def _require_impl_ref(ref, name): + if not isinstance(ref, dict): + raise ValueError("%s must be a dict" % name) + if not any(isinstance(ref.get(k), str) and ref[k] + for k in _IMPL_REF_KEYS): + raise ValueError( + "%s must carry at least one of commit/tag/archive_ref" + % name) + return ref + + +def append_pushback_round(conn, task_id, project_id, round_no, + trigger_items, pushback_package, + fix_diff=None, re_review_result=None, + user_id="system", created_at=None, + module_id=None, contract_version=None, + failure_source=None, failure_counts=None): + """第 3 轨自动推回接线口：一轮推回写一条链记录。 + + 字段契约（本轨提案，第 3 轨落地时按此接线）： + - task_id: str（必填，轮次归属的任务）； + - round_no: int >= 1（第几轮推回；按规则推回只到 3 轮， + round_no > 3 的推回条目会被时间线标为异常——第 4 次 + 失败应走推倒，不应再推回；这里不硬拒是怕调用方按 + 任务全局编号，硬拒会把编号口径分歧变成写入失败）； + - trigger_items: 非空 list（触发推回的失败项，str 或 dict）； + - pushback_package: dict（推回包：逐条失败项、证据、复现、 + 修复验收标准、上一轮产物哈希；多余键原样保存并入哈希）； + - fix_diff: dict 或 None（修复摘要：建议键 summary、 + diff_sha256、files_changed；只存摘要与哈希，不存全文， + 全文在任务产物里、哈希绑定）； + - re_review_result: dict 或 None（重审结果：建议键 verdict + 取 pass/fail，CONDITIONAL 不算通过是第 3 轨的判定语义， + 账本只原样记录）； + - round_status: 推时 "pushed"，重审后 "re_reviewed"，由 + re_review_result 是否为空推导；一轮留两条记录即完整轨迹。 + - module_id / contract_version: str 或 None（推回归属的 + 模块与合同版本；计数口径要求这两项，缺了时间线仍能 + 列出条目，但 module_failure_counts 无法把它归入计数）； + - failure_source: audit|acceptance 或 None（触发本轮的 + 那次失败的来源）； + - failure_counts: dict 或 None（推回时点的计数快照，建议 + 键 audit_fails/acceptance_fails/total_fails；快照只供 + 交叉核对，计数以 module_failure_counts 的派生为准）。 + 校验失败抛 ValueError，绝不写半条记录。 + """ + _require_non_empty_str(task_id, "task_id") + _require_int_ge(round_no, "round_no", 1) + if not isinstance(trigger_items, list) or not trigger_items: + raise ValueError("trigger_items must be a non-empty list") + if not isinstance(pushback_package, dict): + raise ValueError("pushback_package must be a dict") + if fix_diff is not None and not isinstance(fix_diff, dict): + raise ValueError("fix_diff must be a dict or None") + if re_review_result is not None \ + and not isinstance(re_review_result, dict): + raise ValueError("re_review_result must be a dict or None") + if module_id is not None: + _require_non_empty_str(module_id, "module_id") + if contract_version is not None: + _require_non_empty_str(contract_version, "contract_version") + if failure_source is not None \ + and failure_source not in _FAILURE_SOURCES: + raise ValueError( + "failure_source must be audit|acceptance or None") + if failure_counts is not None \ + and not isinstance(failure_counts, dict): + raise ValueError("failure_counts must be a dict or None") + detail = json.dumps({ + "round_no": round_no, + "trigger_items": trigger_items, + "pushback_package": pushback_package, + "fix_diff": fix_diff, + "re_review_result": re_review_result, + "round_status": ("re_reviewed" if re_review_result is not None + else "pushed"), + "module_id": module_id, + "contract_version": contract_version, + "failure_source": failure_source, + "failure_counts": failure_counts, + }, sort_keys=True, ensure_ascii=False) + return append_entry(conn, user_id, project_id, PUSHBACK_ACTION, + detail=detail, task_id=task_id, + created_at=created_at) + + +def list_pushback_rounds(conn, task_id=None, project_id=None, + limit=100): + """按 id 升序返回推回轮次记录（detail 已解析，解析失败不炸）。""" + sql = ("SELECT id, user_id, project_id, task_id, action, detail," + " created_at, prev_hash, entry_hash FROM audit_ledger" + " WHERE action = ?") + params = [PUSHBACK_ACTION] + if task_id is not None: + sql += " AND task_id = ?" + params.append(task_id) + if project_id is not None: + sql += " AND project_id = ?" + params.append(project_id) + sql += " ORDER BY id ASC LIMIT ?" + params.append(int(limit)) + rows = conn.execute(sql, tuple(params)).fetchall() + out = [] + for entry in _rows_as_entries(rows): + parsed = None + parse_error = None + try: + parsed = json.loads(entry["detail"] or "") + except ValueError as exc: + parse_error = str(exc) + entry["pushback"] = parsed + entry["pushback_parse_error"] = parse_error + out.append(entry) + return out + + +def append_audit_verdict(conn, task_id, project_id, module_id, + contract_version, verdict, source, + failures=None, user_id="system", + created_at=None, extra=None): + """记一次审核 / 验收结论（失败计数的唯一事实源）。 + + verdict 取 pass|fail（CONDITIONAL 不算通过：调用方须把 + CONDITIONAL 映射成 fail 再记，账本不替调用方解释 verdict + 语义）；source 取 audit|acceptance（分来源计数用）。 + failures 为失败项 list（可空）；extra 为调用方附加 dict。 + 校验失败抛 ValueError，绝不写半条记录。 + """ + _require_non_empty_str(task_id, "task_id") + _require_non_empty_str(module_id, "module_id") + _require_non_empty_str(contract_version, "contract_version") + if verdict not in _VERDICT_VALUES: + raise ValueError("verdict must be pass|fail") + if source not in _FAILURE_SOURCES: + raise ValueError("source must be audit|acceptance") + if failures is not None and not isinstance(failures, list): + raise ValueError("failures must be a list or None") + if extra is not None and not isinstance(extra, dict): + raise ValueError("extra must be a dict or None") + detail = json.dumps({ + "module_id": module_id, + "contract_version": contract_version, + "verdict": verdict, + "source": source, + "failures": failures or [], + "extra": extra or {}, + }, sort_keys=True, ensure_ascii=False) + return append_entry(conn, user_id, project_id, VERDICT_ACTION, + detail=detail, task_id=task_id, + created_at=created_at) + + +def _has_confirmed_rebuild_notice(conn, task_id, module_id, + contract_version, rebuild_no): + """库内是否已有同任务/模块/合同版本/推倒号的 confirmed 通知。""" + rows = conn.execute( + "SELECT detail FROM audit_ledger WHERE action = ?" + " AND task_id = ? ORDER BY id ASC", + (REBUILD_NOTICE_ACTION, task_id)).fetchall() + for row in rows: + try: + parsed = json.loads(row[0] or "") + except ValueError: + continue + if not isinstance(parsed, dict): + continue + if parsed.get("module_id") == module_id \ + and parsed.get("contract_version") == contract_version \ + and parsed.get("rebuild_no") == rebuild_no \ + and parsed.get("confirmation_status") == "confirmed": + return True + return False + + +def append_rebuild_decision(conn, task_id, project_id, module_id, + contract_version, rebuild_no, + old_impl_ref, new_impl_ref=None, + root_cause="implementation", + status="decided", prior_failure_total=None, + user_id="system", created_at=None, + retained_assets=None, trigger_source=None, + assessment_entry_id=None): + """记一次推倒重做判定 / 完成（只影响该 module_id）。 + + - rebuild_no: int >= 1，按 (module_id, contract_version) + 计（合同版本变了重新计；跨版本总数由时间线汇总另计）。 + - old_impl_ref: dict，commit/tag/archive_ref 至少其一—— + 旧实现以提交/tag/归档保留在 git 历史与审计记录里， + 本条目只记引用，不删旧实现。 + - new_impl_ref: dict 或 None；status=rebuilt 时必填（同形 + 引用），status=decided 时可空（新实现尚未产生）。 + - root_cause: implementation（代码实现问题，走代码推倒） + 或 contract_design（合同/设计问题，不自动重做代码， + 转 G2 重新冻结并升级人工复核或用户；本条目只记录 + 该判定，不授权任何重做）。 + - prior_failure_total: 触发本次推倒时的累计失败数（快照， + 供交叉核对；触发判定以 module_failure_counts 为准）。 + - retained_assets: dict 或 None（推倒保留资产：建议键 + old_test_set_ref、passed_acceptance_items、 + old_new_impl_mapping、old_tests_status=pass|fail| + pending；新实现第一道门是旧测试全过，过不了调用方 + 须另写 verdict fail 计入计数，账本不自动补写）。 + - trigger_source: audit|acceptance 或 None（触发推倒的 + 失败来源）。acceptance 时硬门：库内必须已有同 + (task_id, module_id, contract_version, rebuild_no) 的 + confirmed 用户通知，否则拒绝写入（未确认不得重开）； + audit 触发不走此门。 + - assessment_entry_id: 第三次失败判定条目的链 id 或 None。 + 校验失败抛 ValueError，绝不写半条记录。 + """ + _require_non_empty_str(task_id, "task_id") + _require_non_empty_str(module_id, "module_id") + _require_non_empty_str(contract_version, "contract_version") + _require_int_ge(rebuild_no, "rebuild_no", 1) + _require_impl_ref(old_impl_ref, "old_impl_ref") + if root_cause not in _ROOT_CAUSES: + raise ValueError( + "root_cause must be implementation|contract_design") + if status not in _REBUILD_STATUSES: + raise ValueError("status must be decided|rebuilt") + if new_impl_ref is not None: + _require_impl_ref(new_impl_ref, "new_impl_ref") + if status == "rebuilt" and new_impl_ref is None: + raise ValueError( + "new_impl_ref is required when status is rebuilt") + if prior_failure_total is not None: + _require_int_ge(prior_failure_total, "prior_failure_total", 0) + if retained_assets is not None: + if not isinstance(retained_assets, dict): + raise ValueError("retained_assets must be a dict or None") + if "old_tests_status" in retained_assets \ + and retained_assets["old_tests_status"] \ + not in _OLD_TESTS_STATUSES: + raise ValueError( + "retained_assets.old_tests_status must be" + " pass|fail|pending") + if trigger_source is not None \ + and trigger_source not in _FAILURE_SOURCES: + raise ValueError( + "trigger_source must be audit|acceptance or None") + if assessment_entry_id is not None: + _require_int_ge(assessment_entry_id, "assessment_entry_id", 1) + if trigger_source == "acceptance" and not _has_confirmed_rebuild_notice( + conn, task_id, module_id, contract_version, rebuild_no): + raise ValueError( + "acceptance-triggered rebuild requires a confirmed user" + " notice (audit_rebuild_notice, confirmation_status=" + "confirmed) for the same task/module/contract/rebuild_no") + detail = json.dumps({ + "module_id": module_id, + "contract_version": contract_version, + "rebuild_no": rebuild_no, + "old_impl_ref": old_impl_ref, + "new_impl_ref": new_impl_ref, + "root_cause": root_cause, + "status": status, + "prior_failure_total": prior_failure_total, + "retained_assets": retained_assets, + "trigger_source": trigger_source, + "assessment_entry_id": assessment_entry_id, + }, sort_keys=True, ensure_ascii=False) + return append_entry(conn, user_id, project_id, REBUILD_ACTION, + detail=detail, task_id=task_id, + created_at=created_at) + + +def list_rebuild_decisions(conn, task_id=None, module_id=None, + limit=100): + """按 id 升序返回推倒重做记录（detail 已解析）。""" + sql = ("SELECT id, user_id, project_id, task_id, action, detail," + " created_at, prev_hash, entry_hash FROM audit_ledger" + " WHERE action = ?") + params = [REBUILD_ACTION] + if task_id is not None: + sql += " AND task_id = ?" + params.append(task_id) + # module 无独立列、在 detail 里，解析后过滤（见下）。 + sql += " ORDER BY id ASC LIMIT ?" + params.append(int(limit)) + rows = conn.execute(sql, tuple(params)).fetchall() + out = [] + for entry in _rows_as_entries(rows): + try: + parsed = json.loads(entry["detail"] or "") + except ValueError as exc: + parsed = None + entry["rebuild_parse_error"] = str(exc) + entry["rebuild"] = parsed + if module_id is not None: + if not isinstance(parsed, dict) \ + or parsed.get("module_id") != module_id: + continue + out.append(entry) + return out + + +def _parse_timeline_detail(entry): + try: + return json.loads(entry["detail"] or ""), None + except ValueError as exc: + return None, str(exc) + + +def _count_walk(parsed_entries, corrections=None): + """对已解析的 verdict/rebuild 序列派生计数。 + + 基线移动：pass 或已完成的重做（status=rebuilt）都把 + 计数清零；decided（未完成）不移动基线。返回计数 dict。 + corrections: {原 verdict 条目 id: 纠正后 verdict}（审计的 + 审计产生；同一原条目以后一条纠正为准）。计数用有效结论， + 但链上原结论与纠正记录都保留，不抹历史。 + """ + corrections = corrections or {} + counts = {"audit_fails": 0, "acceptance_fails": 0, + "total_fails": 0, "last_pass_entry_id": None, + "last_rebuild_entry_id": None, + "last_fail_entry_id": None, "entries_considered": 0, + "parse_errors": 0} + for entry, parsed in parsed_entries: + if entry["action"] == REBUILD_ACTION: + if isinstance(parsed, dict) \ + and parsed.get("status") == "rebuilt": + counts["audit_fails"] = 0 + counts["acceptance_fails"] = 0 + counts["total_fails"] = 0 + counts["last_rebuild_entry_id"] = entry["id"] + counts["entries_considered"] += 1 + continue + if entry["action"] != VERDICT_ACTION: + continue + counts["entries_considered"] += 1 + if not isinstance(parsed, dict): + counts["parse_errors"] += 1 + continue + verdict = corrections.get(entry["id"], parsed.get("verdict")) + if verdict == "pass": + counts["audit_fails"] = 0 + counts["acceptance_fails"] = 0 + counts["total_fails"] = 0 + counts["last_pass_entry_id"] = entry["id"] + elif verdict == "fail": + if parsed.get("source") == "audit": + counts["audit_fails"] += 1 + elif parsed.get("source") == "acceptance": + counts["acceptance_fails"] += 1 + counts["total_fails"] += 1 + counts["last_fail_entry_id"] = entry["id"] + counts["rebuild_triggered"] = counts["total_fails"] > 3 + return counts + + +def _corrections_from(parsed_meta_entries): + """从审计的审计条目提有效结论映射：{原 verdict id: 纠正结论}。 + + 同一原条目以后一条纠正为准；misjudgment_found 为 False + 或无 corrected_verdict 的条目不产生纠正。 + """ + corrections = {} + for entry, parsed in parsed_meta_entries: + if entry["action"] != META_REVIEW_ACTION: + continue + if not isinstance(parsed, dict): + continue + if not parsed.get("misjudgment_found"): + continue + original_id = parsed.get("original_entry_id") + corrected = parsed.get("corrected_verdict") + if isinstance(original_id, int) and corrected in _VERDICT_VALUES: + corrections[original_id] = corrected + return corrections + + +def module_failure_counts(conn, module_id, contract_version, + task_id=None): + """派生失败计数：按 (module_id, contract_version) 计。 + + 只读 verdict + rebuild + meta-review 条目（全局，不截断）； + 调用方自报的 failure_counts 快照不参与计算。计数用有效 + 结论（审计的审计纠正后为准），链上原结论仍保留。 + total > 3 即推倒触发。 + """ + _require_non_empty_str(module_id, "module_id") + _require_non_empty_str(contract_version, "contract_version") + sql = ("SELECT id, user_id, project_id, task_id, action, detail," + " created_at, prev_hash, entry_hash FROM audit_ledger" + " WHERE action IN (?, ?, ?) ORDER BY id ASC") + params = [VERDICT_ACTION, REBUILD_ACTION, META_REVIEW_ACTION] + if task_id is not None: + sql = ("SELECT id, user_id, project_id, task_id, action," + " detail, created_at, prev_hash, entry_hash" + " FROM audit_ledger WHERE action IN (?, ?, ?)" + " AND task_id = ? ORDER BY id ASC") + params = [VERDICT_ACTION, REBUILD_ACTION, META_REVIEW_ACTION, + task_id] + rows = conn.execute(sql, tuple(params)).fetchall() + parsed_entries = [] + meta_entries = [] + for entry in _rows_as_entries(rows): + parsed, _err = _parse_timeline_detail(entry) + if not isinstance(parsed, dict): + continue + if parsed.get("module_id") != module_id: + continue + if parsed.get("contract_version") != contract_version: + continue + if entry["action"] == META_REVIEW_ACTION: + meta_entries.append((entry, parsed)) + else: + parsed_entries.append((entry, parsed)) + counts = _count_walk(parsed_entries, + corrections=_corrections_from(meta_entries)) + counts["module_id"] = module_id + counts["contract_version"] = contract_version + return counts + + +def audit_timeline(conn, task_id=None, module_id=None, + project_id=None, limit=200): + """复原时间线：全部审计类条目按 id 升序。 + + 返回 {entries, summary}：entries 为解析后条目（解析失败 + 的条目保留 raw 并标 parse_error，绝不丢）；summary 含 + 窗口内（limit 截断内）的 verdict 统计、推回轮次、推倒数、 + 第三次失败判定、测试变更单审、审计的审计、重开通知数、 + 按合同版本的失败计数（用有效结论，含窗口内纠正）与 + 异常（如推回 round_no > 3、assessment 不在第 3 次）。 + 窗口计数只反映窗口内条目，全局精确计数须用 + module_failure_counts——summary 置 window_only=True 明示。 + 最终 pass 只是序列末尾的一条，不抹掉前面的失败轮次。 + """ + placeholders = ", ".join("?" for _ in TIMELINE_ACTIONS) + sql = ("SELECT id, user_id, project_id, task_id, action, detail," + " created_at, prev_hash, entry_hash FROM audit_ledger" + " WHERE action IN (%s)" % placeholders) + params = list(TIMELINE_ACTIONS) + if task_id is not None: + sql += " AND task_id = ?" + params.append(task_id) + if project_id is not None: + sql += " AND project_id = ?" + params.append(project_id) + sql += " ORDER BY id ASC LIMIT ?" + params.append(int(limit)) + rows = conn.execute(sql, tuple(params)).fetchall() + entries = [] + for entry in _rows_as_entries(rows): + parsed, err = _parse_timeline_detail(entry) + entry["parsed"] = parsed + entry["parse_error"] = err + entries.append(entry) + if module_id is not None: + entries = [e for e in entries + if isinstance(e["parsed"], dict) + and e["parsed"].get("module_id") == module_id] + verdict_pass = sum( + 1 for e in entries + if e["action"] == VERDICT_ACTION + and isinstance(e["parsed"], dict) + and e["parsed"].get("verdict") == "pass") + verdict_fail = sum( + 1 for e in entries + if e["action"] == VERDICT_ACTION + and isinstance(e["parsed"], dict) + and e["parsed"].get("verdict") == "fail") + pushback_rounds = [e for e in entries + if e["action"] == PUSHBACK_ACTION] + rebuilds = [e for e in entries if e["action"] == REBUILD_ACTION] + anomalies = [] + for e in pushback_rounds: + parsed = e["parsed"] + if isinstance(parsed, dict) \ + and isinstance(parsed.get("round_no"), int) \ + and parsed["round_no"] > 3: + anomalies.append( + "pushback round_no %r at id %r exceeds 3: the 4th" + " failure should have been a rebuild, not a pushback" + % (parsed["round_no"], e["id"])) + for e in entries: + parsed = e["parsed"] + if e["action"] == ASSESSMENT_ACTION \ + and isinstance(parsed, dict) \ + and parsed.get("failure_total") != 3: + anomalies.append( + "third-failure assessment at id %r has failure_total" + " %r, expected 3" % (e["id"], parsed.get( + "failure_total"))) + counts_by_contract = {} + versions = sorted({ + e["parsed"]["contract_version"] for e in entries + if isinstance(e["parsed"], dict) + and e["parsed"].get("contract_version")}) + for version in versions: + window = [(e, e["parsed"]) for e in entries + if e["action"] in (VERDICT_ACTION, REBUILD_ACTION) + and isinstance(e["parsed"], dict) + and e["parsed"].get("contract_version") == version] + meta = [(e, e["parsed"]) for e in entries + if e["action"] == META_REVIEW_ACTION + and isinstance(e["parsed"], dict) + and e["parsed"].get("contract_version") == version] + counts_by_contract[version] = _count_walk( + window, corrections=_corrections_from(meta)) + summary = { + "window_only": True, + "truncated": len(entries) >= int(limit), + "verdict_pass": verdict_pass, + "verdict_fail": verdict_fail, + "pushback_rounds": len(pushback_rounds), + "rebuild_decisions": len(rebuilds), + "rebuilds_completed": sum( + 1 for e in rebuilds + if isinstance(e["parsed"], dict) + and e["parsed"].get("status") == "rebuilt"), + "third_failure_assessments": sum( + 1 for e in entries if e["action"] == ASSESSMENT_ACTION), + "test_change_audits": sum( + 1 for e in entries if e["action"] == TEST_CHANGE_ACTION), + "meta_reviews": sum( + 1 for e in entries if e["action"] == META_REVIEW_ACTION), + "rebuild_notices": sum( + 1 for e in entries if e["action"] == REBUILD_NOTICE_ACTION), + "counts_by_contract": counts_by_contract, + "anomalies": anomalies, + } + return {"entries": entries, "summary": summary} + + +def append_third_failure_assessment(conn, task_id, project_id, + module_id, contract_version, + determination, basis, + failure_total=3, + max_extra_rounds=None, + user_id="system", created_at=None): + """记第三次失败的根因判定（阈值不死板的一环）。 + + - determination: early_rebuild（结构性错误，提前推倒） + 或 continuation_round（接近完成只差收尾，再给 1 轮 + targeted 修复，再 fail 即推倒）。 + - basis: 非空 dict，判定依据（建议键 audit_conclusion、 + failure_items、diff_scope、passed_items、 + continuation_counting；failure_items/passed_items 若 + 出现必须为 list）。依据缺失的判定不进账本。 + - failure_total: 判定时的累计失败数，正常为 3；非 3 时 + 仍可写，但时间线标异常（计数口径分歧须事后核查）。 + - max_extra_rounds: early_rebuild 时必须为 0、 + continuation_round 时必须为 1；None 时按判定取值。 + 校验失败抛 ValueError，绝不写半条记录。 + """ + _require_non_empty_str(task_id, "task_id") + _require_non_empty_str(module_id, "module_id") + _require_non_empty_str(contract_version, "contract_version") + if determination not in _DETERMINATIONS: + raise ValueError( + "determination must be early_rebuild|continuation_round") + if not isinstance(basis, dict) or not basis: + raise ValueError("basis must be a non-empty dict") + for key in ("failure_items", "passed_items"): + if key in basis and not isinstance(basis[key], list): + raise ValueError("basis.%s must be a list" % key) + _require_int_ge(failure_total, "failure_total", 1) + if max_extra_rounds is None: + max_extra_rounds = 0 if determination == "early_rebuild" else 1 + if determination == "early_rebuild" and max_extra_rounds != 0: + raise ValueError( + "early_rebuild requires max_extra_rounds == 0") + if determination == "continuation_round" \ + and max_extra_rounds != 1: + raise ValueError( + "continuation_round requires max_extra_rounds == 1") + detail = json.dumps({ + "module_id": module_id, + "contract_version": contract_version, + "determination": determination, + "basis": basis, + "failure_total": failure_total, + "max_extra_rounds": max_extra_rounds, + }, sort_keys=True, ensure_ascii=False) + return append_entry(conn, user_id, project_id, ASSESSMENT_ACTION, + detail=detail, task_id=task_id, + created_at=created_at) + + +def append_test_change_audit(conn, task_id, project_id, module_id, + contract_version, test_changes, auditor, + contract_refrozen, refreeze_ref=None, + reason="", user_id="system", + created_at=None): + """记一次删改测试的单独审计（推倒保留资产的一环）。 + + - test_changes: dict，键 added/modified/deleted，各为 + 测试标识 str 的 list；三者至少其一非空。 + - auditor: 经谁审计（必填，非空 str，如审计员用户 id）。 + - contract_refrozen: 严格 bool。modified/deleted 非空 + 且 contract_refrozen 为 False 时拒绝写入（删改测试 + 必须走合同重新冻结；纯新增不走此门）。 + - refreeze_ref: contract_refrozen 为 True 时必填（新冻结 + 的合同版本或决策引用）；为 False 时须为空。 + - reason: 改动原因说明（可空 str）。 + 校验失败抛 ValueError，绝不写半条记录。 + """ + _require_non_empty_str(task_id, "task_id") + _require_non_empty_str(module_id, "module_id") + _require_non_empty_str(contract_version, "contract_version") + _require_non_empty_str(auditor, "auditor") + if not isinstance(test_changes, dict): + raise ValueError("test_changes must be a dict") + for key in ("added", "modified", "deleted"): + value = test_changes.get(key, []) + if not isinstance(value, list) or not all( + isinstance(item, str) for item in value): + raise ValueError( + "test_changes.%s must be a list of strings" % key) + if not any(test_changes.get(key) for key in + ("added", "modified", "deleted")): + raise ValueError( + "test_changes must change at least one test") + if not isinstance(contract_refrozen, bool): + raise ValueError("contract_refrozen must be a bool") + needs_refreeze = bool(test_changes.get("modified") or + test_changes.get("deleted")) + if needs_refreeze and not contract_refrozen: + raise ValueError( + "modified/deleted tests require contract_refrozen=True") + if contract_refrozen: + _require_non_empty_str(refreeze_ref, "refreeze_ref") + elif refreeze_ref is not None: + raise ValueError( + "refreeze_ref must be None when contract_refrozen is False") + if not isinstance(reason, str): + raise ValueError("reason must be a string") + detail = json.dumps({ + "module_id": module_id, + "contract_version": contract_version, + "test_changes": { + "added": test_changes.get("added", []), + "modified": test_changes.get("modified", []), + "deleted": test_changes.get("deleted", []), + }, + "auditor": auditor, + "contract_refrozen": contract_refrozen, + "refreeze_ref": refreeze_ref, + "reason": reason, + }, sort_keys=True, ensure_ascii=False) + return append_entry(conn, user_id, project_id, TEST_CHANGE_ACTION, + detail=detail, task_id=task_id, + created_at=created_at) + + +def append_audit_of_audit(conn, task_id, project_id, module_id, + contract_version, original_entry_id, + original_verdict, corrected_verdict, + misjudgment_found, reviewer, findings=None, + triggered_process=None, user_id="system", + created_at=None): + """记一次审计的审计（事后抽查纠错，留痕不改历史）。 + + - original_entry_id: 原结论的链 id（int >= 1）。 + - original_verdict: 原结论 pass|fail。 + - corrected_verdict: 纠正后结论 pass|fail；misjudgment_found + 为 True 时必填（可与原结论相同，仅推理纠正时由 findings + 说明）；为 False 时须为 None（无判错即无纠正）。 + - misjudgment_found: 严格 bool（是否发现判错）。 + - reviewer: 抽查人（必填，非空 str）。 + - findings: 发现清单 list（可空）。 + - triggered_process: 纠正后新触发的流程，如 re-review / + rebuild / none（可空 str；G6 人工验收不在此列—— + 本条目没有任何代替 G6 的效力）。 + 校验失败抛 ValueError，绝不写半条记录。 + """ + _require_non_empty_str(task_id, "task_id") + _require_non_empty_str(module_id, "module_id") + _require_non_empty_str(contract_version, "contract_version") + _require_int_ge(original_entry_id, "original_entry_id", 1) + if original_verdict not in _VERDICT_VALUES: + raise ValueError("original_verdict must be pass|fail") + if not isinstance(misjudgment_found, bool): + raise ValueError("misjudgment_found must be a bool") + if misjudgment_found: + if corrected_verdict not in _VERDICT_VALUES: + raise ValueError( + "corrected_verdict must be pass|fail when" + " misjudgment_found is True") + elif corrected_verdict is not None: + raise ValueError( + "corrected_verdict must be None when misjudgment_found" + " is False") + _require_non_empty_str(reviewer, "reviewer") + if findings is not None and not isinstance(findings, list): + raise ValueError("findings must be a list or None") + if triggered_process is not None: + _require_non_empty_str(triggered_process, "triggered_process") + detail = json.dumps({ + "module_id": module_id, + "contract_version": contract_version, + "original_entry_id": original_entry_id, + "original_verdict": original_verdict, + "corrected_verdict": corrected_verdict, + "misjudgment_found": misjudgment_found, + "reviewer": reviewer, + "findings": findings or [], + "triggered_process": triggered_process, + }, sort_keys=True, ensure_ascii=False) + return append_entry(conn, user_id, project_id, META_REVIEW_ACTION, + detail=detail, task_id=task_id, + created_at=created_at) + + +def append_rebuild_notice(conn, task_id, project_id, module_id, + contract_version, rebuild_no, + notification_content, notified_party, + confirmation_status="pending", + confirmed_by=None, user_id="system", + created_at=None): + """记一次推倒重开前的用户通知与确认（未确认不得重开）。 + + - notification_content: 通知内容（非空 str 或非空 dict， + 如新方向说明；内容本身进链，不可事后改口）。 + - notified_party: 被通知方（必填，非空 str，如用户 id）。 + - confirmation_status: pending|confirmed|rejected；账本 + 只追加，pending → confirmed 是两条记录，不是改状态。 + - confirmed_by: confirmed 时必填（确认人）；其它状态 + 须为空。 + 校验失败抛 ValueError，绝不写半条记录。 + """ + _require_non_empty_str(task_id, "task_id") + _require_non_empty_str(module_id, "module_id") + _require_non_empty_str(contract_version, "contract_version") + _require_int_ge(rebuild_no, "rebuild_no", 1) + if isinstance(notification_content, str): + if not notification_content: + raise ValueError( + "notification_content must be non-empty") + elif isinstance(notification_content, dict): + if not notification_content: + raise ValueError( + "notification_content must be non-empty") + else: + raise ValueError( + "notification_content must be a string or dict") + _require_non_empty_str(notified_party, "notified_party") + if confirmation_status not in _CONFIRM_STATUSES: + raise ValueError( + "confirmation_status must be pending|confirmed|rejected") + if confirmation_status == "confirmed": + _require_non_empty_str(confirmed_by, "confirmed_by") + elif confirmed_by is not None: + raise ValueError( + "confirmed_by must be None unless confirmation_status" + " is confirmed") + detail = json.dumps({ + "module_id": module_id, + "contract_version": contract_version, + "rebuild_no": rebuild_no, + "notification_content": notification_content, + "notified_party": notified_party, + "confirmation_status": confirmation_status, + "confirmed_by": confirmed_by, + }, sort_keys=True, ensure_ascii=False) + return append_entry(conn, user_id, project_id, + REBUILD_NOTICE_ACTION, detail=detail, + task_id=task_id, created_at=created_at) + + +class LedgerStore: + """db_path 版账本存储（测试与脚本用；服务端走连接版函数）。 + + 只提供追加 / 校验 / 导出 / 迁移 / 推回记录，没有修改与 + 删除方法——接口层面即只追加。 + """ + + def __init__(self, db_path): + from cloud.db import connect as _db_connect + self.db_path = str(db_path) + self.conn = _db_connect(self.db_path) + ensure_schema(self.conn) + self.conn.commit() + + def close(self): + try: + self.conn.close() + except Exception: + pass + + def append(self, user_id, project_id, action, detail="", + task_id=None, created_at=None): + """追加并提交；并发唯一冲突时在事务边界重试整条写入。""" + last_exc = None + for _attempt in range(5): + try: + entry = append_entry( + self.conn, user_id, project_id, action, + detail=detail, task_id=task_id, + created_at=created_at) + self.conn.commit() + return entry + except Exception as exc: + if not _is_integrity_error(exc): + raise + last_exc = exc + try: + self.conn.rollback() + except Exception: + pass + raise last_exc + + def verify(self, expected_head=None): + return verify_chain(self.conn, expected_head=expected_head) + + def export(self, limit=None): + return export_chain(self.conn, limit=limit) + + def migrate_legacy(self): + result = migrate_legacy_audit_log(self.conn) + self.conn.commit() + return result + + def append_pushback_round(self, *args, **kwargs): + entry = append_pushback_round(self.conn, *args, **kwargs) + self.conn.commit() + return entry + + def list_pushback_rounds(self, *args, **kwargs): + return list_pushback_rounds(self.conn, *args, **kwargs) + + def append_audit_verdict(self, *args, **kwargs): + entry = append_audit_verdict(self.conn, *args, **kwargs) + self.conn.commit() + return entry + + def append_rebuild_decision(self, *args, **kwargs): + entry = append_rebuild_decision(self.conn, *args, **kwargs) + self.conn.commit() + return entry + + def list_rebuild_decisions(self, *args, **kwargs): + return list_rebuild_decisions(self.conn, *args, **kwargs) + + def module_failure_counts(self, *args, **kwargs): + return module_failure_counts(self.conn, *args, **kwargs) + + def audit_timeline(self, *args, **kwargs): + return audit_timeline(self.conn, *args, **kwargs) + + def append_third_failure_assessment(self, *args, **kwargs): + entry = append_third_failure_assessment( + self.conn, *args, **kwargs) + self.conn.commit() + return entry + + def append_test_change_audit(self, *args, **kwargs): + entry = append_test_change_audit(self.conn, *args, **kwargs) + self.conn.commit() + return entry + + def append_audit_of_audit(self, *args, **kwargs): + entry = append_audit_of_audit(self.conn, *args, **kwargs) + self.conn.commit() + return entry + + def append_rebuild_notice(self, *args, **kwargs): + entry = append_rebuild_notice(self.conn, *args, **kwargs) + self.conn.commit() + return entry diff --git a/cloud/migrations.py b/cloud/migrations.py index db24df5..c09a4a9 100644 --- a/cloud/migrations.py +++ b/cloud/migrations.py @@ -46,6 +46,9 @@ MIGRATION_TABLES = ( "project_members", "tasks", "audit_log", + # 防篡改账本（哈希链，第 4 轨）：与 audit_log 同款身份列， + # 迁库时整行原样迁（id 与哈希不动，链不断），序列见下。 + "audit_ledger", "summaries", "kb_docs", "cost_records", @@ -91,12 +94,13 @@ MIGRATION_TABLES = ( _SKIP_TABLES = frozenset({"kb_fts", "sqlite_sequence"}) # 身份列的表：POSTGRES_SCHEMA 里唯一用 -# ``BIGINT GENERATED BY DEFAULT AS IDENTITY`` 的 4 张表。 +# ``BIGINT GENERATED BY DEFAULT AS IDENTITY`` 的 5 张表。 # 迁移时显式插 id 不会推序列，迁完必须 setval 到 MAX(id)， # 否则后续省略 id 的 INSERT 会撞主键。新增身份列表必须 # 同步改这里，test_p2_migrations.py 会拿 schema 实测对齐。 SEQUENCE_TABLES = ( "audit_log", + "audit_ledger", "summaries", "kb_docs", "cost_records", diff --git a/cloud/server.py b/cloud/server.py index 13cb459..f9c24fc 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -63,9 +63,11 @@ from cloud.auth import ROLES as _MEMBER_ROLES from cloud.auth import AuthStore from cloud.coordinator import parse_parts as _coord_parse_parts from cloud.coordinator import validate_parts as _coord_validate_parts -from cloud.cost import CostStore, unpriced_token_cap +from cloud.cost import unpriced_token_cap from cloud.gates import merge_state_decision as _merge_state_decision from cloud.gates import resolve_inside as _gate_resolve_inside +from cloud.ledger import export_chain as _ledger_export +from cloud.ledger import verify_chain as _ledger_verify 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 @@ -3075,6 +3077,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._admin_quota_get(user_id, query) elif path == "/admin/quotas" and method == "POST": self._admin_quota_set(user_id) + elif path == "/admin/audit-ledger/verify" and method == "GET": + self._admin_ledger_verify(user_id) + elif path == "/admin/audit-ledger/export" and method == "GET": + self._admin_ledger_export(user_id, query) elif path == "/admin/push/dispatch" and method == "POST": self._push_dispatch_endpoint(user_id) elif path.startswith("/tasks/"): @@ -8997,6 +9003,54 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): target, quota, used, month, self._quota_unpriced_tokens(target))) + def _admin_ledger_verify(self, user_id): + """GET /admin/audit-ledger/verify：全局管理员只读校验账本。 + + 只读：成功路径不写审计（写审计会改链头，校验读数本身 + 就不再可复现）；被拒才记 admin_ledger_denied。校验失败 + 仍回 200，ok=false + failures 供运维判读，不是 HTTP 错。 + """ + if not self._is_global_admin(user_id): + self._audit_admin(user_id, "admin_ledger_denied", + "verify ledger") + self._send_json(403, {"error": "forbidden"}) + return + conn = self._db() + try: + result = _ledger_verify(conn) + finally: + conn.close() + self._send_json(200, result) + + def _admin_ledger_export(self, user_id, query): + """GET /admin/audit-ledger/export：全局管理员只读导出账本。 + + ?limit=N 可截断（运维取头锚点用）；limit 非法回 400。 + 导出含头哈希，运维应把头哈希存到库外作比对锚点。 + """ + if not self._is_global_admin(user_id): + self._audit_admin(user_id, "admin_ledger_denied", + "export ledger") + self._send_json(403, {"error": "forbidden"}) + return + limit_raw = (query.get("limit") or [None])[0] + limit = None + if limit_raw is not None: + try: + limit = int(limit_raw) + except (TypeError, ValueError): + limit = -1 + if limit < 0: + self._send_json( + 400, {"error": "limit must be a non-negative integer"}) + return + conn = self._db() + try: + result = _ledger_export(conn, limit=limit) + finally: + conn.close() + self._send_json(200, result) + def _admin_quota_set(self, user_id): if not self._is_global_admin(user_id): self._audit_admin(user_id, "admin_quota_denied", diff --git a/cloud/tests/test_tamper_ledger.py b/cloud/tests/test_tamper_ledger.py new file mode 100644 index 0000000..70ba395 --- /dev/null +++ b/cloud/tests/test_tamper_ledger.py @@ -0,0 +1,596 @@ +"""防篡改账本（哈希链）+ 成本双列视图单测。 + +不联网、不跑真实模型：账本走临时 SQLite 文件，篡改/删除 +用 SQL 直写模拟（先 DROP 触发器，模拟有 DDL 权限的攻击者； +触发器本身另有单测），服务端点走本地临时端口。 +""" + +import http.client +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import unittest +from pathlib import Path + +try: + from cloud.auth import AuthStore + from cloud.cost import CostStore, dual_column_view, price_usd_micros + from cloud.ledger import ( + GENESIS_HASH, LedgerStore, append_pushback_round, + audit_timeline, export_chain, list_pushback_rounds, + module_failure_counts, verify_chain, verify_export) + from cloud.server import create_server, init_db +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.cost import CostStore, dual_column_view, price_usd_micros + from cloud.ledger import ( + GENESIS_HASH, LedgerStore, append_pushback_round, + audit_timeline, export_chain, list_pushback_rounds, + module_failure_counts, verify_chain, verify_export) + from cloud.server import create_server, init_db + + +def _req(port, method, path, key=None): + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10) + hdrs = {} + if key is not None: + hdrs["X-API-Key"] = key + try: + conn.request(method, path, headers=hdrs) + resp = conn.getresponse() + data = resp.read() + finally: + conn.close() + try: + parsed = json.loads(data.decode("utf-8")) if data else None + except ValueError: + parsed = None + return resp.status, parsed + + +class TestLedgerChain(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.store = LedgerStore(os.path.join(self.tmp.name, "led.db")) + + def tearDown(self): + self.store.close() + self.tmp.cleanup() + + def test_append_and_verify_ok(self): + self.store.append("u1", "p1", "create_task", "t1") + self.store.append("u1", "p1", "approve_merge", "t1") + result = self.store.verify() + self.assertTrue(result["ok"]) + self.assertEqual(result["count"], 2) + self.assertEqual(result["failures"], []) + self.assertNotEqual(result["head_hash"], GENESIS_HASH) + + def test_append_only_interface(self): + # 接口层面即只追加：没有修改/删除方法（test_auth 同款断言）。 + self.assertFalse(hasattr(LedgerStore, "delete_entry")) + self.assertFalse(hasattr(LedgerStore, "update_entry")) + self.assertFalse(hasattr(LedgerStore, "delete")) + self.assertFalse(hasattr(LedgerStore, "update")) + + def test_sqlite_triggers_block_update_and_delete(self): + self.store.append("u1", "p1", "create_task", "t1") + with self.assertRaises(Exception): + self.store.conn.execute( + "UPDATE audit_ledger SET detail='x' WHERE id=1") + with self.assertRaises(Exception): + self.store.conn.execute( + "DELETE FROM audit_ledger WHERE id=1") + self.store.conn.rollback() + self.assertTrue(self.store.verify()["ok"]) + + def test_tamper_detail_fails_verify(self): + self.store.append("u1", "p1", "create_task", "t1") + self.store.append("u1", "p1", "approve_merge", "t1") + # 模拟有 DDL 权限的攻击者：先卸触发器再改历史。 + self.store.conn.execute( + "DROP TRIGGER IF EXISTS audit_ledger_no_update") + self.store.conn.execute( + "UPDATE audit_ledger SET detail='forged' WHERE id=1") + self.store.conn.commit() + result = self.store.verify() + self.assertFalse(result["ok"]) + self.assertTrue(any("tampered" in f for f in result["failures"])) + + def test_delete_middle_breaks_chain(self): + for i in range(3): + self.store.append("u1", "p1", "act-%d" % i, "d%d" % i) + self.store.conn.execute( + "DROP TRIGGER IF EXISTS audit_ledger_no_delete") + self.store.conn.execute("DELETE FROM audit_ledger WHERE id=2") + self.store.conn.commit() + result = self.store.verify() + self.assertFalse(result["ok"]) + self.assertTrue(any("broken" in f for f in result["failures"])) + + def test_delete_tail_detected_by_old_head(self): + for i in range(3): + self.store.append("u1", "p1", "act-%d" % i, "d%d" % i) + old_export = self.store.export() + old_head = old_export["head_hash"] + self.store.conn.execute( + "DROP TRIGGER IF EXISTS audit_ledger_no_delete") + self.store.conn.execute( + "DELETE FROM audit_ledger WHERE id=3") + self.store.conn.commit() + # 截断后的链自洽（固有局限，如实断言），但旧头对不上。 + self.assertTrue(self.store.verify()["ok"]) + self.assertFalse( + self.store.verify(expected_head=old_head)["ok"]) + self.assertFalse( + verify_export(self.store.export(), + expected_head=old_head)["ok"]) + + def test_export_round_trip_and_tampered_export(self): + self.store.append("u1", "p1", "create_task", "t1") + self.store.append("u1", "p1", "approve_merge", "t1") + data = self.store.export() + self.assertTrue(verify_export(data)["ok"]) + self.assertEqual(data["count"], 2) + data["entries"][0]["detail"] = "forged" + self.assertFalse(verify_export(data)["ok"]) + data2 = self.store.export() + data2["head_hash"] = "0" * 64 + self.assertFalse(verify_export(data2)["ok"]) + + def test_legacy_migration_marks_pre_chain_without_rewrite(self): + # 老 audit_log 行（直接 INSERT，模拟链上线前的历史）。 + conn = sqlite3.connect(os.path.join(self.tmp.name, "led.db")) + try: + 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)") + conn.execute( + "INSERT INTO audit_log(user_id, project_id, action," + " detail, created_at) VALUES(?,?,?,?,?)", + ("u1", "p1", "create_task", "t0", 1000.0)) + conn.execute( + "INSERT INTO audit_log(user_id, project_id, action," + " detail, created_at) VALUES(?,?,?,?,?)", + ("u1", "p1", "approve_merge", "t0", 1001.0)) + conn.commit() + finally: + conn.close() + # 链已空但老表有行：verify 必须 fail-closed，不认账。 + fresh = LedgerStore(os.path.join(self.tmp.name, "led.db")) + try: + result = fresh.verify() + self.assertFalse(result["ok"]) + self.assertTrue(any("legacy_not_migrated" in f + for f in result["failures"])) + migrated = fresh.migrate_legacy() + self.assertTrue(migrated["migrated"]) + self.assertEqual( + migrated["marker"]["action"], "ledger_genesis") + after = fresh.verify() + self.assertTrue(after["ok"]) + self.assertIsNotNone(after["genesis_marker"]) + self.assertEqual( + after["genesis_marker"]["legacy_audit_rows"], 2) + # 迁移后链已启动，再跑迁移必须拒绝，不补不改。 + self.assertFalse(fresh.migrate_legacy()["migrated"]) + # 老行原样未动：仍 2 行、无哈希列。 + raw = sqlite3.connect(os.path.join(self.tmp.name, "led.db")) + try: + rows = raw.execute( + "SELECT COUNT(*) FROM audit_log").fetchone() + cols = [c[1] for c in raw.execute( + "PRAGMA table_info(audit_log)").fetchall()] + finally: + raw.close() + self.assertEqual(rows[0], 2) + self.assertNotIn("entry_hash", cols) + finally: + fresh.close() + + def test_auth_audit_writes_chain_same_transaction(self): + db_path = os.path.join(self.tmp.name, "auth.db") + auth = AuthStore(db_path) + try: + owner_id, _key = auth.create_user("owner") + auth.create_project("p1", owner_id) + auth.audit(owner_id, "p1", "create_task", "task-1") + auth.audit(owner_id, "p1", "approve_merge", "task-1") + self.assertEqual(len(auth.list_audit("p1")), 2) + finally: + auth.conn.close() + led = LedgerStore(db_path) + try: + result = led.verify() + self.assertTrue(result["ok"]) + self.assertEqual(result["count"], 2) + actions = [e["action"] for e in + led.export()["entries"]] + self.assertEqual(actions, ["create_task", "approve_merge"]) + finally: + led.close() + + +class TestPushbackAndRebuild(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.store = LedgerStore(os.path.join(self.tmp.name, "led.db")) + + def tearDown(self): + self.store.close() + self.tmp.cleanup() + + def _verdict(self, verdict, source, task="task-1"): + return self.store.append_audit_verdict( + task, "p1", "mod-a", "v1", verdict, source, + failures=["f1"] if verdict == "fail" else []) + + def test_failure_counts_split_sources_and_reset_on_pass(self): + self._verdict("fail", "audit") + self._verdict("fail", "acceptance") + counts = self.store.module_failure_counts("mod-a", "v1") + self.assertEqual(counts["audit_fails"], 1) + self.assertEqual(counts["acceptance_fails"], 1) + self.assertEqual(counts["total_fails"], 2) + self.assertFalse(counts["rebuild_triggered"]) + self._verdict("pass", "audit") + counts = self.store.module_failure_counts("mod-a", "v1") + self.assertEqual(counts["total_fails"], 0) + + def test_contract_version_change_restarts_count(self): + self._verdict("fail", "audit") + self._verdict("fail", "audit") + self.assertEqual( + self.store.module_failure_counts("mod-a", "v1")["total_fails"], + 2) + self.assertEqual( + self.store.module_failure_counts("mod-a", "v2")["total_fails"], + 0) + + def test_rebuild_trigger_and_rebuild_resets_baseline(self): + for source in ("audit", "acceptance", "audit"): + self._verdict("fail", source) + counts = self.store.module_failure_counts("mod-a", "v1") + self.assertEqual(counts["total_fails"], 3) + self.assertFalse(counts["rebuild_triggered"]) + self._verdict("fail", "audit") # 第 4 次：推倒触发点 + counts = self.store.module_failure_counts("mod-a", "v1") + self.assertTrue(counts["rebuild_triggered"]) + self.store.append_rebuild_decision( + "task-1", "p1", "mod-a", "v1", 1, + {"commit": "abc123"}, root_cause="implementation", + status="decided", prior_failure_total=4) + with self.assertRaises(ValueError): + # rebuilt 必须带新实现引用。 + self.store.append_rebuild_decision( + "task-1", "p1", "mod-a", "v1", 1, + {"commit": "abc123"}, root_cause="implementation", + status="rebuilt") + self.store.append_rebuild_decision( + "task-1", "p1", "mod-a", "v1", 1, + {"commit": "abc123"}, new_impl_ref={"commit": "def456"}, + root_cause="implementation", status="rebuilt", + prior_failure_total=4) + counts = self.store.module_failure_counts("mod-a", "v1") + self.assertEqual(counts["total_fails"], 0) + # 重做后再超 3 次：同样触发（不再自动重开是升级语义， + # 账本只负责计数与触发信号，不替第 3 轨做升级决策）。 + for _ in range(4): + self._verdict("fail", "audit") + self.assertTrue( + self.store.module_failure_counts( + "mod-a", "v1")["rebuild_triggered"]) + + def test_contract_design_root_cause_recorded_not_rebuild(self): + entry = self.store.append_rebuild_decision( + "task-1", "p1", "mod-a", "v1", 1, + {"tag": "v1-old"}, root_cause="contract_design", + status="decided") + self.assertEqual(entry["action"], "audit_rebuild") + with self.assertRaises(ValueError): + self.store.append_rebuild_decision( + "task-1", "p1", "mod-a", "v1", 2, + {"commit": "x"}, root_cause="something_else") + + def test_timeline_reconstructs_full_path_without_erasure(self): + # fail → 推回 1-3 轮 → 第 4 次 fail → 推倒 → 重做 → 再审 pass + self._verdict("fail", "audit") + for round_no in (1, 2, 3): + self.store.append_pushback_round( + "task-1", "p1", round_no, ["f1"], + {"failures": ["f1"]}, + module_id="mod-a", contract_version="v1", + failure_source="audit") + self._verdict("fail", "audit") + # 此时累计：首 fail + 3 轮后各 1 fail = 4 次，已触发推倒。 + self.assertTrue( + self.store.module_failure_counts( + "mod-a", "v1")["rebuild_triggered"]) + self.store.append_rebuild_decision( + "task-1", "p1", "mod-a", "v1", 1, + {"commit": "old"}, new_impl_ref={"commit": "new"}, + root_cause="implementation", status="rebuilt", + prior_failure_total=4) + self._verdict("pass", "audit") + timeline = self.store.audit_timeline(task_id="task-1") + actions = [e["action"] for e in timeline["entries"]] + self.assertEqual(actions.count("audit_verdict"), 5) + self.assertEqual(actions.count("audit_pushback_round"), 3) + self.assertEqual(actions.count("audit_rebuild"), 1) + # 最终 pass 在末尾，前面 4 次 fail 一个不少。 + self.assertEqual(timeline["entries"][-1]["parsed"]["verdict"], + "pass") + self.assertEqual(timeline["summary"]["verdict_fail"], 4) + self.assertEqual(timeline["summary"]["verdict_pass"], 1) + self.assertEqual(timeline["summary"]["rebuilds_completed"], 1) + self.assertTrue(self.store.verify()["ok"]) + + def test_pushback_round_over_3_flagged_in_timeline(self): + self.store.append_pushback_round( + "task-1", "p1", 4, ["f1"], {"failures": ["f1"]}, + module_id="mod-a", contract_version="v1") + timeline = self.store.audit_timeline(task_id="task-1") + self.assertEqual(len(timeline["summary"]["anomalies"]), 1) + + def test_pushback_rounds_list_and_validation(self): + append_pushback_round( + self.store.conn, "task-1", "p1", 1, ["f1"], + {"failures": ["f1"], "evidence": "e"}) + self.store.conn.commit() + rounds = list_pushback_rounds(self.store.conn, task_id="task-1") + self.assertEqual(len(rounds), 1) + self.assertEqual(rounds[0]["pushback"]["round_no"], 1) + with self.assertRaises(ValueError): + append_pushback_round( + self.store.conn, "task-1", "p1", 1, [], + {"failures": []}) + + +class TestTightenedCaliber(unittest.TestCase): + """三条收紧：第三次失败判定、保留资产、测试单审、 + 审计的审计、重开前用户确认。""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.store = LedgerStore(os.path.join(self.tmp.name, "led.db")) + + def tearDown(self): + self.store.close() + self.tmp.cleanup() + + def _verdict(self, verdict, source="audit", task="task-1"): + return self.store.append_audit_verdict( + task, "p1", "mod-a", "v1", verdict, source, + failures=["f1"] if verdict == "fail" else []) + + def test_third_failure_assessment_kinds(self): + entry = self.store.append_third_failure_assessment( + "task-1", "p1", "mod-a", "v1", "early_rebuild", + {"audit_conclusion": "structural", + "failure_items": ["f1"], + "diff_scope": "mod-a/*", + "passed_items": ["a1"], + "continuation_counting": "extra rounds count as fails"}) + self.assertEqual(entry["action"], + "audit_third_failure_assessment") + cont = self.store.append_third_failure_assessment( + "task-1", "p1", "mod-a", "v1", "continuation_round", + {"audit_conclusion": "nearly done"}) + parsed = json.loads(cont["detail"]) + self.assertEqual(parsed["max_extra_rounds"], 1) + with self.assertRaises(ValueError): + self.store.append_third_failure_assessment( + "task-1", "p1", "mod-a", "v1", "early_rebuild", + {"audit_conclusion": "x"}, max_extra_rounds=1) + with self.assertRaises(ValueError): + self.store.append_third_failure_assessment( + "task-1", "p1", "mod-a", "v1", "continuation_round", + {}) + timeline = self.store.audit_timeline(task_id="task-1") + self.assertEqual(timeline["summary"]["third_failure_assessments"], + 2) + + def test_rebuild_retained_assets(self): + entry = self.store.append_rebuild_decision( + "task-1", "p1", "mod-a", "v1", 1, + {"commit": "old"}, new_impl_ref={"commit": "new"}, + root_cause="implementation", status="rebuilt", + retained_assets={ + "old_test_set_ref": "tests/mod-a@old", + "passed_acceptance_items": ["a1", "a2"], + "old_new_impl_mapping": {"old": "new"}, + "old_tests_status": "pass"}) + parsed = json.loads(entry["detail"]) + self.assertEqual(parsed["retained_assets"]["old_tests_status"], + "pass") + with self.assertRaises(ValueError): + self.store.append_rebuild_decision( + "task-1", "p1", "mod-a", "v1", 2, + {"commit": "old"}, retained_assets={ + "old_tests_status": "mostly"}) + + def test_test_change_audit_gate(self): + # 纯新增不走重新冻结门。 + self.store.append_test_change_audit( + "task-1", "p1", "mod-a", "v1", + {"added": ["test_new"]}, auditor="aud-1", + contract_refrozen=False) + # 删改不走重新冻结：拒绝写入。 + with self.assertRaises(ValueError): + self.store.append_test_change_audit( + "task-1", "p1", "mod-a", "v1", + {"deleted": ["test_old"]}, auditor="aud-1", + contract_refrozen=False) + # 删改 + 重新冻结 + 引用：通过。 + entry = self.store.append_test_change_audit( + "task-1", "p1", "mod-a", "v1", + {"modified": ["test_m"], "deleted": ["test_old"]}, + auditor="aud-1", contract_refrozen=True, + refreeze_ref="v2", reason="contract change") + self.assertEqual(entry["action"], "audit_test_change") + with self.assertRaises(ValueError): + # refrozen 时引用必填。 + self.store.append_test_change_audit( + "task-1", "p1", "mod-a", "v1", + {"modified": ["test_m"]}, auditor="aud-1", + contract_refrozen=True) + + def test_meta_review_correction_changes_effective_count(self): + first = self._verdict("fail") + self._verdict("fail") + third = self._verdict("fail") + self.assertEqual( + self.store.module_failure_counts( + "mod-a", "v1")["total_fails"], 3) + # 纠正最新一条 fail → pass：计数清零（pass 即基线）。 + self.store.append_audit_of_audit( + "task-1", "p1", "mod-a", "v1", third["id"], "fail", + "pass", True, reviewer="meta-1", + findings=["evidence misread"], triggered_process="none") + self.assertEqual( + self.store.module_failure_counts( + "mod-a", "v1")["total_fails"], 0) + # 但链上原 fail 与纠正都保留，时间线不抹。 + timeline = self.store.audit_timeline(task_id="task-1") + self.assertEqual(timeline["summary"]["verdict_fail"], 3) + self.assertEqual(timeline["summary"]["meta_reviews"], 1) + self.assertTrue(self.store.verify()["ok"]) + + def test_meta_review_first_fail_correction_semantics(self): + first = self._verdict("fail") + self._verdict("fail") + self._verdict("fail") + # 纠正第一条 fail → pass：pass 清零后，后两条 fail 重计为 2。 + self.store.append_audit_of_audit( + "task-1", "p1", "mod-a", "v1", first["id"], "fail", + "pass", True, reviewer="meta-1", + findings=["wrong item"], triggered_process="re-review") + self.assertEqual( + self.store.module_failure_counts( + "mod-a", "v1")["total_fails"], 2) + + def test_acceptance_rebuild_requires_confirmed_notice(self): + # 无通知：拒绝。 + with self.assertRaises(ValueError): + self.store.append_rebuild_decision( + "task-1", "p1", "mod-a", "v1", 1, + {"commit": "old"}, root_cause="implementation", + trigger_source="acceptance") + # 仅 pending 通知：仍拒绝。 + self.store.append_rebuild_notice( + "task-1", "p1", "mod-a", "v1", 1, + "new direction: rebuild mod-a", notified_party="user-1") + with self.assertRaises(ValueError): + self.store.append_rebuild_decision( + "task-1", "p1", "mod-a", "v1", 1, + {"commit": "old"}, root_cause="implementation", + trigger_source="acceptance") + # confirmed 后：通过；audit 触发本来就不走此门。 + self.store.append_rebuild_notice( + "task-1", "p1", "mod-a", "v1", 1, + "new direction: rebuild mod-a", notified_party="user-1", + confirmation_status="confirmed", confirmed_by="user-1") + entry = self.store.append_rebuild_decision( + "task-1", "p1", "mod-a", "v1", 1, + {"commit": "old"}, root_cause="implementation", + trigger_source="acceptance") + self.assertEqual(entry["action"], "audit_rebuild") + self.store.append_rebuild_decision( + "task-1", "p1", "mod-a", "v1", 2, + {"commit": "old"}, root_cause="implementation", + trigger_source="audit") + timeline = self.store.audit_timeline(task_id="task-1") + self.assertEqual(timeline["summary"]["rebuild_notices"], 2) + + +class TestCostDualColumn(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.store = CostStore(os.path.join(self.tmp.name, "cost.db")) + + def tearDown(self): + try: + self.store._conn.close() + except Exception: + pass + self.tmp.cleanup() + + def test_unpriced_never_priced_as_zero(self): + priced = price_usd_micros("gpt-6.1-sol", 1000000, 0, 0) + self.assertIsNotNone(priced) + self.store.record("t1", "p1", "gpt-6.1-sol", + input_tokens=1000000, output_tokens=0, + usd_micros=priced, source="actual") + # 无价模型：usd_micros=None，绝不写成 0。 + self.store.record("t1", "p1", "unpriced-model", + input_tokens=5000, output_tokens=0, + usd_micros=None, source="actual") + view = dual_column_view(self.store.report_by_task("t1")) + self.assertEqual(view["priced_usd_micros"], priced) + self.assertEqual(view["usd_micros"], priced) + self.assertEqual(view["unpriced_actual_tokens"], 5000) + self.assertIn("billing_note", view) + self.assertIn("不等于免费", view["billing_note"]) + + +class TestLedgerServerEndpoints(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "srv.db") + auth = AuthStore(self.db_path) + try: + self.admin_id, self.admin_key = auth.create_user("admin") + self.user_id, self.user_key = auth.create_user("plain") + auth.create_project("p1", self.admin_id) + finally: + auth.conn.close() + init_db(self.db_path) + self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS") + os.environ["CODEX_GLOBAL_ADMINS"] = self.admin_id + self.server = create_server(self.db_path, port=0) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + + def tearDown(self): + try: + self.server.shutdown() + self.server.server_close() + except Exception: + pass + if self._old_admins is None: + os.environ.pop("CODEX_GLOBAL_ADMINS", None) + else: + os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins + self.tmp.cleanup() + + def test_verify_and_export_admin_only(self): + status, body = _req(self.port, "GET", + "/admin/audit-ledger/verify", + key=self.admin_key) + self.assertEqual(status, 200) + self.assertTrue(body["ok"]) + status, body = _req(self.port, "GET", + "/admin/audit-ledger/export?limit=10", + key=self.admin_key) + self.assertEqual(status, 200) + self.assertEqual(body["version"], 1) + status, _body = _req(self.port, "GET", + "/admin/audit-ledger/verify", + key=self.user_key) + self.assertEqual(status, 403) + status, _body = _req(self.port, "GET", + "/admin/audit-ledger/export?limit=xx", + key=self.admin_key) + self.assertEqual(status, 400) + + +if __name__ == "__main__": + unittest.main()