diff --git a/cloud/attempt_hook.py b/cloud/attempt_hook.py index a19cbc8..b30511c 100644 --- a/cloud/attempt_hook.py +++ b/cloud/attempt_hook.py @@ -210,6 +210,14 @@ POLICY (each rule is enforced in code below, not just prose): translate to ``%s`` through ``cloud.db``, DDL uses only Postgres-valid types, no autoincrement/identity column), exactly like gap 3's PG evidence, and labelled as such. + Integration v2 out-of-box closure (see + ``cloud/ledger_onboarding.py``): the DDL checklist, the + explicit enable path and the read-only + ``verify_ledger_ready`` / ``ledger_setup_report`` probe + close the "hook off by default + manual PG DDL" gap + WITHOUT changing the default (still OFF) or auto-running + DDL on production Postgres -- the manual PG step remains + an operator action, now checkable instead of assumed. This module imports ``cloud.attempt_ledger`` only, opens no database connection itself (callers pass a connection in), and diff --git a/cloud/evidence_binding.py b/cloud/evidence_binding.py new file mode 100644 index 0000000..e45c8cc --- /dev/null +++ b/cloud/evidence_binding.py @@ -0,0 +1,155 @@ +"""发布证据的新鲜度与单次使用绑定（集成 v2，代码层）。 + +首轮缺口（全自动 9.5 扣 0.5）：发布门“重算”保证的是“不用 +旧结论”，不是“证据一定新鲜”——证据体里没有时间戳/nonce +绑定，调用方把合并时同一份证据体在发布时原样重交，同样 +能通过重算。本模块补这一层，但诚实边界先写清： + +- 本绑定防的是**原样重放**（同一 nonce 二次提交）与**陈旧 + 证据**（时间戳超出窗口），不防**重盖戳**：没有签名的证据 + 体，调用方完全可以给旧内容盖一个新时间戳/新 nonce 再交， + 本模块无法识破。防篡改/防重盖戳需要签名证据或第 5 项的 + 只读求证（向 GitHub 实时核对），那是另一层控制，不在本 + 模块冒充。 +- 绑定是**调用方自愿出示**的：请求体里同时带 evidence_ts + 与 evidence_nonce 才进入绑定校验；两者都不带时走旧路径 + （与 #259 契约兼容，响应里标注 evidence_binding=absent， + 绝不谎称已验证新鲜）。只带其一是格式错误（400），不是 + “半绑定通过”。把绑定改成强制，是契约决定，须协调员/ + 用户在合轨时定，不在本轨悄悄改 #259。 +- 单次使用语义：nonce 在**首次通过校验的提交**即被消费， + 无论该次发布最终是 200 还是被其它门（未 merged、推回 + 阻塞）挡下——被挡的调用方下次必须取新证据重盖戳。这 + 是故意的：每次发布尝试都要求当次新鲜证据，而不是“一份 + 证据可反复试到放行”。 +- nonce 存储与审计链无关：evidence_nonces 表只记 nonce、 + 任务 id 与时间戳，不写 audit_log、不移动 #241 链头；409 + 不写审计的规则不受影响。表在写路径（首次消费）才建， + 读路径（校验已存在）不建表——查不到表时视为无消费记 + 录，不是错误。 + +零第三方依赖，只用标准库。时间一律用 Unix 秒（time.time）， +与 full_auto.audit_record 的 ts 同口径。 +""" + +import os +import time +import uuid + +# 新鲜度窗口：证据时间戳距提交时刻不得超过 1 小时；未来 +# 容忍 5 分钟时钟偏差，超出即视为异常（不是“更新鲜所以更好”）。 +MAX_AGE_SECONDS = 3600 +CLOCK_SKEW_SECONDS = 300 + +NONCE_TABLE_DDL = ( + "CREATE TABLE IF NOT EXISTS evidence_nonces(" + "nonce TEXT NOT NULL PRIMARY KEY, " + "task_id TEXT NOT NULL, " + "evidence_ts DOUBLE PRECISION NOT NULL, " + "consumed_at DOUBLE PRECISION NOT NULL)" +) + + +def bind_evidence(evidence, now=None, nonce=None): + """给一份证据体盖戳：返回原证据的浅拷贝 + 时间戳/nonce。 + + 调用方（协调员侧）在**取到证据的那一刻**调用本函数， + 不要在合并时盖好、发布时原样重交——那样时间戳本身 + 就是旧的，会被窗口校验挡下（这正是本绑定的意义）。 + nonce 缺省为 uuid4 hex；调用方自带 nonce 时须保证 + 全局唯一（重复 nonce 即重放）。 + """ + if not isinstance(evidence, dict): + raise ValueError("evidence must be a dict") + stamped = dict(evidence) + stamped["evidence_ts"] = time.time() if now is None else now + stamped["evidence_nonce"] = nonce or uuid.uuid4().hex + return stamped + + +def validate_bound_evidence(evidence, now=None): + """纯函数校验绑定证据的形状与新鲜度（不查 nonce 库）。 + + 返回 (ok, reason)：ok 为 False 时 reason 为 + evidence_binding_missing / evidence_binding_incomplete / + evidence_ts_not_a_number / evidence_nonce_empty / + evidence_too_old / evidence_from_future。nonce 重放由 + EvidenceNonceStore.consume 判定，不在本函数（本函数 + 无状态，同一份证据调两次结果相同，那不是重放判定）。 + """ + if not isinstance(evidence, dict): + return False, "evidence_binding_missing" + has_ts = evidence.get("evidence_ts") is not None + has_nonce = evidence.get("evidence_nonce") is not None + if not has_ts and not has_nonce: + return False, "evidence_binding_missing" + if not (has_ts and has_nonce): + return False, "evidence_binding_incomplete" + ts = evidence.get("evidence_ts") + if isinstance(ts, bool) or not isinstance(ts, (int, float)): + return False, "evidence_ts_not_a_number" + nonce = evidence.get("evidence_nonce") + if not isinstance(nonce, str) or not nonce.strip(): + return False, "evidence_nonce_empty" + moment = time.time() if now is None else now + if ts > moment + CLOCK_SKEW_SECONDS: + return False, "evidence_from_future" + if moment - ts > MAX_AGE_SECONDS: + return False, "evidence_too_old" + return True, None + + +def binding_presence(evidence): + """绑定出示状态：bound（成对出示）/ partial（只带其一）/ + absent（两者皆无）。服务端据此决定走绑定校验、报 400 + 还是走旧路径（并标注 absent，绝不谎称已验证）。""" + if not isinstance(evidence, dict): + return "absent" + has_ts = evidence.get("evidence_ts") is not None + has_nonce = evidence.get("evidence_nonce") is not None + if has_ts and has_nonce: + return "bound" + if has_ts or has_nonce: + return "partial" + return "absent" + + +class EvidenceNonceStore: + """nonce 单次使用存储（DB 连接由调用方传入并负责提交）。 + + consume 成功返回 True（首次见此 nonce）；nonce 已存在 + 返回 False（重放）。表不存在时 consume 先建表（写路径 + 才建）；has() 在表不存在时返回 False，不建表。 + """ + + def __init__(self, conn): + self.conn = conn + + def ensure_table(self): + self.conn.execute(NONCE_TABLE_DDL) + + def has(self, nonce): + try: + row = self.conn.execute( + "SELECT nonce FROM evidence_nonces WHERE nonce = ?", + (nonce,)).fetchone() + except Exception: + return False + return row is not None + + def consume(self, task_id, nonce, evidence_ts, consumed_at=None): + """消费一个 nonce：首次 True；已存在 False（不覆盖原行）。 + + 调用方须在 consume 后提交（本函数不 commit：服务端 + 在自己的连接生命周期里提交，与其余写操作保持一致）。 + """ + self.ensure_table() + if self.has(nonce): + return False + self.conn.execute( + "INSERT INTO evidence_nonces(nonce, task_id, evidence_ts," + " consumed_at) VALUES(?, ?, ?, ?)", + (nonce, task_id, evidence_ts, + time.time() if consumed_at is None else consumed_at), + ) + return True diff --git a/cloud/gate_verification.py b/cloud/gate_verification.py new file mode 100644 index 0000000..30a4779 --- /dev/null +++ b/cloud/gate_verification.py @@ -0,0 +1,140 @@ +"""合并/发布门禁证据的只读核对器（集成 v2，代码层）。 + +首轮缺口（门禁与审计 9.0 扣 0.5）：硬门禁与发布门禁强制 +的是“调用方提交的证据齐全且全绿”，不是“服务端已向 +GitHub 求证”——调用方谎报 CI 全绿 / 保护状态 / 协调员 +批准时，门禁会放行；docstring 称覆盖控制是“审计与 +PR/Checks 的只读核对”，但在合并/发布路径上没有这样 +一段“调 GitHub Checks API 核对证据”的代码（#210 的 +实时保护检查只属于 auto-config execute 端点，不属于 +merge/release 端点）。 + +本轨的结论分两半，缺一不可： + +A. 核对器本身已实现（本模块，纯函数 + 注入式 reader， + 有测试）：verify_gate_evidence(evidence, reader) 把 + 调用方证据与 reader 给出的实时数据逐项比对——CI + 状态、每项 required 检查的状态、分支保护状态；不 + 一致即 mismatch（逐条列出），一致即 verified。reader + 是调用方注入的 callable：reader(repo, ref) -> { + "ci_checks": {name: status}, "branch_protection": + "protected"|"unprotected"|... }，真实 reader 用 + github_app 的 _github_api_request 走 GET Checks / + GET 分支保护（只读，绝不写 PR、不改保护、不合并）。 + reader 为 None 或抛异常时结果是“未核对” + （checked=False, verified=False），绝不是“通过”—— + 调用方把 skipped 当 pass 用，是调用方的 bug，本 + 模块的返回值结构让这种误用一眼可见（verified 恒 + False，除非逐项比对全过）。 + +B. 实时求证在当前合并/发布路径上**接不上**（不可实现 + 证明，不是未做）：① 请求体里没有 repo 与 ref—— + merge/release 的证据字段只有 files_changed、ci_checks、 + branch_protection 等，没有仓库名、没有提交 SHA， + reader 不知道向哪个仓库的哪次提交求证，猜仓库 + 就是编造求证；② 服务端没有可信的求证通道保证—— + github_app.resolve_app_token 对未安装 App 的仓库 + 返回 None（调用方回落 PAT），PAT 是调用方自己的 + 凭证，用它“核对”调用方自己的证据不是独立求证； + ③ 任务行里没有 repo/ref 列，服务端无法从任务 id + 反推求证目标。接上实时求证须先改契约（请求体或任 + 务行加 repo + ref，安装保证或服务端侧可信 token）， + 那是契约决定，须协调员/用户在合轨时定；本轨不编 + 一个“看起来在求证”的假 reader、不把调用方证据 + 自己与自己比对后写成 verified。 + +当前真实控制（写清，不拔高）：门禁强制证据齐全全绿 +（硬模式 fail-closed）；审计记 outcome 与 blocks +（merge_gate_blocked / merge_task / release_task）， +但审计 detail 不保存证据体全文——事后追责能查到 +“谁在何时以何 blocks 通过/被挡”，查不到“当时提交 +的 ci_checks 原文”。谎报证据的发现仍靠事后审计与 +GitHub 侧记录比对，不是当场发现。本模块的核对器 +是为契约补齐后的接线准备的：届时服务端只需注入 +真实 reader 并把 verified=False（无论 skipped 还是 +mismatch）当作阻断，核对器本身不用改。 + +零第三方依赖，只用标准库。本模块不 import cloud +其它模块（reader 由调用方注入），不发网络请求。 +""" + + +def verify_gate_evidence(evidence, reader=None, repo=None, ref=None): + """只读核对调用方门禁证据与实时数据。 + + evidence：调用方提交的门禁证据 dict（与硬门禁同款 + 字段：ci_status、ci_checks[{name, status}]、 + required_checks、branch_protection）。 + reader：callable (repo, ref) -> {"ci_checks": {名: 状态}, + "branch_protection": str}，或 None（无可信求证通道）。 + repo/ref：求证目标；reader 非 None 但 repo/ref 缺失 + 时同样视为未核对（不猜目标）。 + + 返回 {"checked": bool, "verified": bool, "mismatches": [...], + "reason": str|None}： + - checked=False：未核对（无 reader、无目标、reader 失败）， + verified 恒 False，reason 说明原因；调用方绝不能把 + checked=False 当通过。 + - checked=True 且 mismatches 为空：逐项比对全过， + verified=True。 + - checked=True 且 mismatches 非空：证据与实时数据不 + 一致，verified=False，mismatches 逐条说明（ci 状态、 + 某项检查、分支保护）。 + 比对规则：只比对调用方声称的项——调用方没声称的 + 检查不判（缺失由硬门禁的证据缺失码处理，不归本 + 模块）；实时数据缺失某项时记 mismatch（声称全绿却 + 查无此项，不是“查不到所以通过”）。 + """ + if reader is None: + return {"checked": False, "verified": False, + "mismatches": [], + "reason": "no_trusted_reader"} + if not repo or not ref: + return {"checked": False, "verified": False, + "mismatches": [], + "reason": "no_verification_target"} + try: + live = reader(repo, ref) + except Exception as exc: + return {"checked": False, "verified": False, + "mismatches": [], + "reason": "reader_failed: %s" % exc} + if not isinstance(live, dict): + return {"checked": False, "verified": False, + "mismatches": [], + "reason": "reader_failed: non-dict result"} + mismatches = [] + live_checks = live.get("ci_checks") or {} + if not isinstance(live_checks, dict): + live_checks = {} + evidence = evidence or {} + if evidence.get("ci_status") == "success": + failed_live = sorted( + name for name, status in live_checks.items() + if status != "success") + if failed_live: + mismatches.append( + "ci_not_green_live: %s" % ", ".join(failed_live)) + for check in (evidence.get("ci_checks") or []): + if not isinstance(check, dict): + continue + name = check.get("name") + if not name: + continue + if check.get("status") == "success" \ + and live_checks.get(name) != "success": + mismatches.append( + "check_not_green_live: %s" % name) + for name in (evidence.get("required_checks") or []): + if live_checks.get(name) != "success": + mismatches.append( + "required_check_not_green_live: %s" % name) + claimed_protection = evidence.get("branch_protection") + live_protection = live.get("branch_protection") + if claimed_protection == "protected" \ + and live_protection != "protected": + mismatches.append( + "branch_protection_not_protected_live: %s" + % live_protection) + return {"checked": True, "verified": not mismatches, + "mismatches": mismatches, "reason": None} diff --git a/cloud/ledger_onboarding.py b/cloud/ledger_onboarding.py new file mode 100644 index 0000000..c80ada6 --- /dev/null +++ b/cloud/ledger_onboarding.py @@ -0,0 +1,190 @@ +"""台账钩子开箱闭合方案（集成 v2，代码层）。 + +首轮缺口（成本与台账 8.5 扣 0.5）：Worker(attempt_ledger_hook +=False) 默认关，CLI 须显式 --attempt-ledger-hook；生产 PG +上启用前须先手动执行台账 PG_DDL 与标记表 PG_MARKER_DDL +并验证。默认状态下指标率是 unknown（诚实），不是“已用 +真实台账”——“接通”只在“开 flag + 手动跑 DDL + 验证” +之后成立，开箱未闭合。 + +本轨的闭合方案（不改任何已定语义）： + +1. 默认行为不变：钩子默认仍关。默认关是 #255 的设计 + （显式启用才写行，关时行为与 gap 3 完全一致、无新表 + 无新行），不是 bug，不为“开箱即用”悄悄改成默认开。 + 把默认改成开，是用户/协调员在合轨时才能定的语义变 + 更，本轨不做。 +2. 显式开启路径唯一且可验证：Worker(attempt_ledger_hook + =True) 或 CLI --attempt-ledger-hook；开启后 run_one + 按 attempt_hook.py 规则 1-4 写行。本模块的 + ledger_setup_report() 把“表是否齐、钩子是否开”分开 + 报告：表齐但钩子关 ≠ 就绪，钩子开但表不齐 ≠ 就绪。 +3. DDL 清单不另造：清单直接引用 attempt_ledger.PG_DDL + 与 attempt_hook.PG_MARKER_DDL 两个常量（本模块不 + 复制 DDL 文本，复制就会漂移）。另含 gap 3 旧表升级句 + attempt_ledger.OLD_TABLE_ALTER（仅当旧 DDL 建过表时 + 需要；gap 3 未合并，生产尚无旧表，属防御项）。 +4. 校验函数 verify_ledger_ready(conn)：只读探针——查两 + 张表的存在与必需列，不建表、不写行（读路径绝不改 + schema，与 metrics-report 端点的只读纪律一致）。 +5. 生产 PG 手动 DDL 仍是运维侧动作：本轨无生产 PG、无 + 本地 PG，PG 证据与此前各轨同款——适配器级（? 占位符 + 经 cloud.db 翻译、DDL 仅用 PG 合法类型、无自增列）， + 绝不写成“生产 PG 已执行”。运维执行清单见 + LEDGER_DDL_CHECKLIST（顺序执行 + 每步验证命令）。 + +零第三方依赖，只用标准库。conn 约定与 attempt_ledger +相同：DBAPI 连接，? 占位符（PG 经 cloud.db 翻译）。 +""" + +from cloud.attempt_hook import MARKER_DDL as _MARKER_DDL +from cloud.attempt_hook import PG_MARKER_DDL as _PG_MARKER_DDL +from cloud.attempt_ledger import OLD_TABLE_ALTER as _OLD_TABLE_ALTER +from cloud.attempt_ledger import PG_DDL as _LEDGER_PG_DDL + +# 必需列：缺任一列即视为表不齐（旧 DDL 建的表缺 +# corrects_attempt_no 时走 OLD_TABLE_ALTER 补列）。 +LEDGER_REQUIRED_COLUMNS = ( + "task_id", "attempt_no", "outcome", "is_defect", + "defect_category", "created_at", "corrects_attempt_no", +) +MARKER_REQUIRED_COLUMNS = ("task_id", "origin", "queued_at") + +# 运维执行清单（顺序）：每步含“做什么 + 怎么验证”。本 +# 清单是文档数据，不是执行器——生产 PG 上的执行仍由 +# 运维手工完成，本模块不代执行、不连生产库。 +LEDGER_DDL_CHECKLIST = ( + { + "step": 1, + "action": "在生产 PG 上执行 attempt_ledger.PG_DDL" + "（建 task_attempts 表）", + "ddl_source": "cloud.attempt_ledger.PG_DDL", + "verify": "\\d task_attempts 或 information_schema" + " 查表与列齐全", + }, + { + "step": 2, + "action": "在生产 PG 上执行 attempt_hook.PG_MARKER_DDL" + "（建 task_attempt_origin_markers 表）", + "ddl_source": "cloud.attempt_hook.PG_MARKER_DDL", + "verify": "\\d task_attempt_origin_markers 或" + " information_schema 查表与列齐全", + }, + { + "step": 3, + "action": "仅当表由 gap 3 旧 DDL 建过时：执行" + " attempt_ledger.OLD_TABLE_ALTER 补" + " corrects_attempt_no 列（gap 3 未合并，" + "生产正常路径跳过此步）", + "ddl_source": "cloud.attempt_ledger.OLD_TABLE_ALTER", + "verify": "task_attempts 含 corrects_attempt_no 列", + }, + { + "step": 4, + "action": "启用钩子：worker 加 --attempt-ledger-hook" + "（或 Worker(attempt_ledger_hook=True)）后重启", + "ddl_source": None, + "verify": "新派发产生 task_attempts 行；" + "平台重派发不产生计数行（marker 生效）", + }, + { + "step": 5, + "action": "用 verify_ledger_ready(conn) 复核：" + "tables_ready 且 hook_enabled 方为就绪", + "ddl_source": None, + "verify": "ledger_setup_report(conn, hook_enabled=True)" + " 的 ready 为 True", + }, +) + + +def ledger_ddl_sources(): + """DDL 清单的权威来源（返回常量本身，供运维原样执行 + 与单测钉住“清单与实现同一来源”，防止清单漂移）。""" + return { + "task_attempts_pg_ddl": _LEDGER_PG_DDL, + "marker_pg_ddl": _PG_MARKER_DDL, + "old_table_alter": _OLD_TABLE_ALTER, + "marker_ddl": _MARKER_DDL, + } + + +def _table_columns(conn, table): + """只读探针：表不存在返回 None；存在返回列名集合。 + + PRAGMA table_info 经 cloud.db 适配层翻译（与 + CostStore._ensure_cached_column 同款做法）；探针失败 + （如 PG 适配差异）返回 None，按“不齐”处理，绝不 + 按“齐”冒充。 + """ + try: + rows = conn.execute( + "PRAGMA table_info(%s)" % table).fetchall() + except Exception: + return None + if not rows: + return None + try: + return {row[1] for row in rows} + except (IndexError, TypeError, KeyError): + return None + + +def verify_ledger_ready(conn): + """只读校验两张台账表：返回 {tables_ready, missing}。 + + missing 列出缺失项，如 "task_attempts table missing"、 + "task_attempts missing columns: corrects_attempt_no"、 + "task_attempt_origin_markers table missing"。本函数不 + 建表、不写行、不提交。 + """ + missing = [] + ledger_cols = _table_columns(conn, "task_attempts") + if ledger_cols is None: + missing.append("task_attempts table missing") + else: + absent = [c for c in LEDGER_REQUIRED_COLUMNS + if c not in ledger_cols] + if absent: + missing.append("task_attempts missing columns: %s" + % ", ".join(absent)) + marker_cols = _table_columns(conn, "task_attempt_origin_markers") + if marker_cols is None: + missing.append("task_attempt_origin_markers table missing") + else: + absent = [c for c in MARKER_REQUIRED_COLUMNS + if c not in marker_cols] + if absent: + missing.append( + "task_attempt_origin_markers missing columns: %s" + % ", ".join(absent)) + return {"tables_ready": not missing, "missing": missing} + + +def ledger_setup_report(conn, hook_enabled=False): + """开箱就绪报告：表齐 + 钩子开，两者缺一都不算就绪。 + + 返回 {ready, tables_ready, hook_enabled, missing, + next_steps}：next_steps 是运维下一步动作的文字清单 + （按 LEDGER_DDL_CHECKLIST 顺序，未做项才列出）， + 不是自动执行。本函数同样只读，不建表、不写行。 + """ + check = verify_ledger_ready(conn) + enabled = bool(hook_enabled) + steps = [] + if check["missing"]: + steps.append("run the PG DDL steps in LEDGER_DDL_CHECKLIST" + " (steps 1-3) on production Postgres and verify") + if not enabled: + steps.append("enable the hook with --attempt-ledger-hook" + " (Worker(attempt_ledger_hook=True)) and restart") + if not steps: + steps.append("ledger is live: counted dispatches write rows;" + " rates are real, not unknown") + return { + "ready": bool(check["tables_ready"] and enabled), + "tables_ready": check["tables_ready"], + "hook_enabled": enabled, + "missing": check["missing"], + "next_steps": steps, + } diff --git a/cloud/ledger_outbox.py b/cloud/ledger_outbox.py new file mode 100644 index 0000000..7a069ff --- /dev/null +++ b/cloud/ledger_outbox.py @@ -0,0 +1,178 @@ +"""台账写失败的 durable 记录与重放（集成 v2，代码层）。 + +首轮缺口（成本与台账 8.5 扣 0.5）：_record_ledger_attempt +写失败只记一条 worker_attempt_ledger_failed 日志，绝不 +影响任务状态与返回值（attempt_hook.py 规则 7）。这是 +设计（指标台账不应挡任务），但与 #241 审计账本的 +fail-closed 不对称：一次写失败会让率悄悄少算一行，只 +有日志可查，没有重放/补写路径。 + +本轨的对齐方式（写清边界，不悄悄改任一侧语义）： + +- 不把台账改成 fail-closed：台账写失败仍不挡任务、不 + 改任务状态、不改返回值（规则 7 原样保留）。让指标写 + 入能阻塞任务执行，是比丢一行更坏的 trade-off， + 本轨不做。 +- 不把审计改成 best-effort：审计账本 fail-closed 原样 + 保留（审计是决策级证据，丢一行即链不可信；台账是 + 指标，丢一行只影响率的分母/分子，且可事后补）。 +- 不对称的剩余部分用“不丢”补齐：写失败时除日志外， + 再向 worker 侧 outbox 文件追加一行 JSONL（task_id、 + outcome、error、failed_at）；运维事后调 + replay_failures() 把 outbox 行经 record_attempt 补写 + 进台账（append-only，attempt_no 按 MAX+1 续号）， + 成功补写的行从 outbox 清除，仍失败的行保留待下次。 + 于是失败不再是“只有日志可查”，而是“可数、可重放、 + 重放后可验证行数补齐”。 + +诚实边界： +- outbox 本身也是 best-effort：outbox 追加失败（如 + 磁盘满）只记日志，不抛、不挡任务——此时仍退回首轮 + 状态（仅日志可查），但这条路径本身已被测试钉住 + （追加失败不抛异常）。 +- 重放无幂等键：record_attempt 没有去重键；若原始写 + 入实际已提交、只是提交确认丢失（ambiguous commit）， + 重放会多写一行。重放是运维手动动作：重放前须核对 + task_attempts 是否已有该任务的对应行，确认缺失再 + 放行；本函数不假装能自动去重（去重键是新的 schema + 语义，须另立一轨定）。 +- outbox 路径：环境变量 CLOUD_LEDGER_FAILURE_OUTBOX + 显式指定时优先；否则取 worker db_path 同目录下的 + attempt_ledger_failures.jsonl；db_path 为 :memory: + 时无落盘位置，返回 None（调用方只记日志）。 + +零第三方依赖，只用标准库。 +""" + +import json +import os +import time + +OUTBOX_ENV = "CLOUD_LEDGER_FAILURE_OUTBOX" +OUTBOX_FILENAME = "attempt_ledger_failures.jsonl" + + +def outbox_path_for(db_path, env=None): + """解析 outbox 文件路径：显式 env > db 同目录 > None。 + + env 缺省读 os.environ（调用方可注入 dict 做单测）。 + db_path 为 :memory: 或空时无同目录可言，返回 None。 + """ + if env is None: + env = os.environ + explicit = env.get(OUTBOX_ENV) if env else None + if explicit: + return str(explicit) + if not db_path or str(db_path) == ":memory:": + return None + directory = os.path.dirname(os.path.abspath(str(db_path))) + return os.path.join(directory, OUTBOX_FILENAME) + + +def append_failure(path, task_id, outcome, error): + """向 outbox 追加一行失败记录。返回 True/False（写成功与否）。 + + 追加失败返回 False，不抛异常（调用方只记日志）。 + 行格式：{task_id, outcome, error, failed_at}，一行一 + 条 JSON，坏行在重放时跳过并保留（见 replay_failures）。 + """ + if not path: + return False + record = { + "task_id": task_id, + "outcome": outcome, + "error": str(error), + "failed_at": time.time(), + } + try: + directory = os.path.dirname(os.path.abspath(str(path))) + if directory: + os.makedirs(directory, exist_ok=True) + with open(path, "a", encoding="utf-8") as handle: + handle.write(json.dumps(record, ensure_ascii=False) + "\n") + return True + except Exception: + return False + + +def read_failures(path): + """读 outbox：返回 (records, bad_lines)。 + + records 为解析成功的 dict 列表（按文件顺序）；bad_lines + 为解析失败的行数（坏行不丢弃，重放时原样保留）。 + 文件不存在返回 ([], 0)，不是错误。 + """ + records = [] + bad = 0 + try: + with open(path, encoding="utf-8") as handle: + lines = handle.readlines() + except OSError: + return records, bad + for line in lines: + line = line.strip() + if not line: + continue + try: + item = json.loads(line) + except ValueError: + bad += 1 + continue + if not isinstance(item, dict) or not item.get("task_id") \ + or item.get("outcome") not in ( + "success", "failure", "interrupted"): + bad += 1 + continue + records.append(item) + return records, bad + + +def replay_failures(path, conn, record_fn=None): + """重放 outbox：经 record_attempt 补写，返回统计。 + + 返回 {replayed, failed, remaining, bad_lines}： + replayed 为成功补写行数，failed 为补写仍失败的行数， + remaining 为重放后 outbox 剩余行数（failed + bad 行）， + bad_lines 为跳过未补写的坏行数。成功补写的行从 + outbox 清除（文件重写为剩余行）；全部成功时 outbox + 被清空（文件保留为空文件，不是删除——删除会丢“曾 + 有失败”的痕迹，空文件 + 台账行即为补齐证据）。 + record_fn 缺省为 attempt_ledger.record_attempt（调用 + 方可注入假函数做单测）；record_fn 抛异常的行计入 + failed 并保留。调用方负责提交（本函数不 commit， + 与 record_attempt 的调用方提交约定一致）。 + """ + if record_fn is None: + from cloud.attempt_ledger import record_attempt as record_fn + records, bad = read_failures(path) + remaining_records = [] + replayed = 0 + failed = 0 + for item in records: + try: + record_fn(conn, item["task_id"], item["outcome"], + is_defect=False) + replayed += 1 + except Exception: + failed += 1 + remaining_records.append(item) + # 坏行无法解析出 task/outcome，不能补写也不能丢弃： + # 原样保留在 outbox 里（本函数读不到原文，故坏行数 + # 只计入 remaining，不重写坏行文本——重写会丢原文）。 + # 为避免重写丢坏行：仅当 bad == 0 时重写文件；有坏行 + # 时不重写，只返回统计，由运维人工处理坏行后再重放。 + remaining = failed + bad + if bad == 0: + try: + with open(path, "w", encoding="utf-8") as handle: + for item in remaining_records: + handle.write(json.dumps( + item, ensure_ascii=False) + "\n") + except OSError: + pass + return { + "replayed": replayed, + "failed": failed, + "remaining": remaining, + "bad_lines": bad, + } diff --git a/cloud/quota.py b/cloud/quota.py index 462fb10..83ce9e8 100644 --- a/cloud/quota.py +++ b/cloud/quota.py @@ -14,6 +14,18 @@ automations._automation_quota_exceeded），且三处已经 绝不能触发上限；服务端 _quota_unpriced_tokens 改为本 模块口径（行为变化仅限含 estimated 无价行的用户：此 前可能被估算行误拦，此后不再误拦；真实花费的拦阻不变）。 + 集成 v2 确认（临时口径，非原 PR 已定）：本收敛来自 + 集成基底（非 main），原 PR 语义（main 里 #86 计全部 + vs #90 actual-only 的分歧）并未在 main 上改动；合轨 + 时须由协调员另行确认 actual-only 是否成为原 PR 口径， + 本轨只把集成 v2 的口径钉死并补双侧回归测试（见 + cloud/tests/test_v2_evidence.py），绝不把临时口径写 + 成原 PR 已定。受影响调用方：server._quota_snapshot / + _check_unpriced_token_cap、worker._quota_limit_and_used / + _quota_exceeded、automations._automation_quota_exceeded， + 三处均经本模块 monthly_usage + decide，改一处即三处 + 同口径；USD 求和不受影响（estimated 行 usd 为 NULL， + SUM 本就跳过）。 - 默认配额解析：server、worker、automations 各一份 CLOUD_DEFAULT_QUOTA_USD_MICROS 解析，现统一为 default_quota_usd_micros（默认值 $10/月 = 10_000_000 diff --git a/cloud/server.py b/cloud/server.py index 9b5f08f..ed4449c 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -79,6 +79,9 @@ from cloud.peer_messages import ensure_followups_table as _peer_ensure_followups from cloud.peer_messages import peer_message_prefix as _peer_prefix from cloud.peer_messages import row_get as _row_get from cloud.marker_tools import parse_spawn_defaults as _parse_spawn_defaults +from cloud.evidence_binding import EvidenceNonceStore as _EvidenceNonceStore +from cloud.evidence_binding import binding_presence as _evidence_presence +from cloud.evidence_binding import validate_bound_evidence as _validate_evidence from cloud.merge_gate import evaluate_merge_gate as _merge_gate_evaluate from cloud.merge_gate import protection_from_env as _merge_gate_protection from cloud.audit_wiring import pushback_block_for_task as _pushback_block @@ -6431,7 +6434,14 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au reader 强制；deploy.py 是运维脚本，不在本接口门禁内。 本服务没有可信渠道直连 GitHub 核实 CI/保护：门禁强制的是 “调用方提交的证据齐全且全绿”，谎报证据由审计与 PR/Checks - 只读核对覆盖，不在本接口的验证范围内。 + 只读核对覆盖，不在本接口的验证范围内。集成 v2 补充 + （cloud/gate_verification）：只读核对器本身已实现 + （注入式 reader、逐项比对、mismatch 列出），但当前 + 合并/发布路径因请求体与任务行均无 repo/ref、且无可 + 信求证通道保证而接不上实时求证——见该模块头的不可 + 实现证明；当前真实控制仍是“证据齐全全绿 + 审计记 + outcome/blocks”，审计 detail 不存证据体全文，谎报 + 发现靠事后比对，不是当场发现。 """ row = self._fetch_task(task_id) if row is None: @@ -6992,7 +7002,16 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au 把它绑到任务链上会在最需要部署修复与回滚时反锁部署。 200 时写一条显式的 release_task 审计作为发布后动作记 录（检查本身是只读的，这条是检查完成后的显式动作， - 不是检查的副产品；锚点须在这条之后重取）。""" + 不是检查的副产品；锚点须在这条之后重取）。 + 证据绑定（集成 v2，cloud/evidence_binding）：请求体成 + 对带 evidence_ts / evidence_nonce 时，服务端校验新鲜 + 度窗口并单次消费 nonce，原样重交即 409 + release_blocked_evidence_replayed，陈旧/未来时间戳即 + 409 release_blocked_evidence_stale；两者皆无时走旧路径 + 并在成功响应标注 evidence_binding=absent。绑定防原样 + 重放与陈旧证据，不防重盖戳（无签名）；实时求证见 + cloud/gate_verification（当前路径因无 repo/ref 且无 + 可信求证通道而接不上，见该模块头的不可实现证明）。""" row = self._fetch_task(task_id) if row is None: self._send_json(404, {"error": "task not found"}) @@ -7014,6 +7033,58 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au if hard_err: self._send_json(400, {"error": hard_err}) return + # 证据新鲜度与单次使用绑定（集成 v2，cloud/evidence_binding）： + # 体里成对带 evidence_ts / evidence_nonce 时才校验；两者 + # 皆无走旧路径（#259 契约兼容，成功响应标注 absent）； + # 只带其一是格式错误 400。校验通过的 nonce 在此次提交 + # 即消费（无论后续门禁放行与否），原样重交同一 nonce + # 即 409 evidence_replayed；陈旧/未来时间戳即 409 + # evidence_stale。409 按本接口规则不写审计；nonce + # 消费写 evidence_nonces 表（与审计链无关，不移动链头）。 + # 绑定防原样重放与陈旧证据，不防重盖戳（无签名证据 + # 无法识破旧内容盖新戳——见 evidence_binding 模块头）。 + binding = _evidence_presence(body) + if binding == "partial": + self._send_json(400, {"error": + "evidence_ts and evidence_nonce" + " must be supplied together"}) + return + if binding == "bound": + ok, reason = _validate_evidence(body) + if not ok: + if reason in ("evidence_too_old", + "evidence_from_future"): + self._send_json(409, { + "error": "release blocked", + "release_gate": { + "allowed": False, "http_status": 409, + "code": "release_blocked_evidence_stale", + "blocked_by": "evidence_binding", + "blocked_sources": ["evidence_binding"], + }, + }) + else: + self._send_json(400, {"error": reason}) + return + _bconn = self._db() + try: + _store = _EvidenceNonceStore(_bconn) + if not _store.consume(task_id, + body["evidence_nonce"], + body["evidence_ts"]): + self._send_json(409, { + "error": "release blocked", + "release_gate": { + "allowed": False, "http_status": 409, + "code": "release_blocked_evidence_replayed", + "blocked_by": "evidence_binding", + "blocked_sources": ["evidence_binding"], + }, + }) + return + _bconn.commit() + finally: + _bconn.close() # 发布时刻的新证据：体里没给任何硬门禁字段时，hard_gate # 为 None -> 证据缺失 409，而不是用合并时的旧结论补。 has_evidence = any( @@ -7060,6 +7131,10 @@ class CloudControlPlaneHandler(OAuthMixin, IdentityMemoryMixin, ProjectMixin, Au store.audit(user_id, task["project_id"], "release_task", task_id) task["release_gate"] = decision + # 绑定出示标注（绝不谎称）：bound = 本次提交带时间戳 + # /nonce 且通过单次校验；absent = 未出示绑定，走 #259 + # 旧路径（证据仍经硬门禁重算，但新鲜度未绑定）。 + task["evidence_binding"] = binding self._send_json(200, task) # -- 运行时审批队列 ------------------------------------------------ diff --git a/cloud/tests/test_v2_evidence.py b/cloud/tests/test_v2_evidence.py new file mode 100644 index 0000000..8f9f225 --- /dev/null +++ b/cloud/tests/test_v2_evidence.py @@ -0,0 +1,487 @@ +"""集成 v2 证据与口径硬伤定向测试（本轨新增，5 项全覆盖）。 + +1. 证据重放绑定（cloud/evidence_binding + release 端点）： + 原样重交被拒、正常证据通过、陈旧/未来被拒、只带其一 + 400、无绑定走旧路径并标注 absent；另钉住诚实边界： + 绑定不鉴别内容（改 ci_status 后仍通过校验——防重盖 + 戳须签名或只读求证，不在本模块冒充）。 +2. 无价 token 口径统一（cloud/quota.monthly_usage 为 + server/worker/automations 三处共用 canonical 源）： + actual 无价行计入、estimated 无价行不计入，并与 + main 旧 SQL（计全部 NULL 行）做前后对比。 +3. 台账钩子开箱（cloud/ledger_onboarding）：表缺时 + not ready、表齐且钩子开时 ready、表齐但钩子关时 + 仍 not ready；DDL 清单引用常量本身，不复制文本。 +4. 台账写失败 outbox（cloud/ledger_outbox + worker + _record_ledger_attempt 接线）：失败进 outbox、重放 + 补写行、重放后 outbox 清空、二次重放 0 行；任务 + 侧返回值仍为 None（不挡任务，规则 7 不变）。 +5. 只读核对器（cloud/gate_verification）：假 reader + 下一致通过、谎报 CI/保护被 mismatch、无 reader / + 无目标 / reader 失败一律 checked=False 且 verified + 恒 False（skipped 绝不当 pass）；实时接线不可实现 + 的三点见 gate_verification 模块头，服务端 docstring + 已同步该边界（本文件不测不存在的实时求证）。 +""" + +import http.client +import json +import logging +import os +import sqlite3 +import tempfile +import threading +import time +import unittest +from pathlib import Path + +try: + from cloud import ledger_onboarding + from cloud import ledger_outbox + from cloud.attempt_hook import ensure_marker_table + from cloud.attempt_ledger import PG_DDL as LEDGER_PG_DDL + from cloud.attempt_ledger import record_attempt + from cloud.attempt_hook import PG_MARKER_DDL as MARKER_PG_DDL + from cloud.cost import CostStore + from cloud.evidence_binding import EvidenceNonceStore + from cloud.evidence_binding import bind_evidence + from cloud.evidence_binding import binding_presence + from cloud.evidence_binding import validate_bound_evidence + from cloud.gate_verification import verify_gate_evidence + from cloud.quota import decide as quota_decide + from cloud.quota import monthly_usage +except ImportError: + import sys + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud import ledger_onboarding + from cloud import ledger_outbox + from cloud.attempt_hook import ensure_marker_table + from cloud.attempt_ledger import PG_DDL as LEDGER_PG_DDL + from cloud.attempt_ledger import record_attempt + from cloud.attempt_hook import PG_MARKER_DDL as MARKER_PG_DDL + from cloud.cost import CostStore + from cloud.evidence_binding import EvidenceNonceStore + from cloud.evidence_binding import bind_evidence + from cloud.evidence_binding import binding_presence + from cloud.evidence_binding import validate_bound_evidence + from cloud.gate_verification import verify_gate_evidence + from cloud.quota import decide as quota_decide + from cloud.quota import monthly_usage + + +def _hard_body(): + return { + "files_changed": ["apps/web/a.py"], + "owned_dir": "apps/web", + "branch_protection": "protected", + "ci_status": "success", + "ci_checks": [{"name": "ci", "status": "success"}], + "required_checks": ["ci"], + "contract_version": "v2", + "contract_frozen": True, + "coordinator_approved": True, + "unresolved_review_comments": 0, + } + + +class EvidenceBindingTestCase(unittest.TestCase): + def test_bind_and_validate_fresh(self): + stamped = bind_evidence(_hard_body()) + self.assertEqual(binding_presence(stamped), "bound") + ok, reason = validate_bound_evidence(stamped) + self.assertTrue(ok) + self.assertIsNone(reason) + + def test_partial_binding_is_not_bound(self): + body = _hard_body() + body["evidence_ts"] = time.time() + self.assertEqual(binding_presence(body), "partial") + ok, reason = validate_bound_evidence(body) + self.assertFalse(ok) + self.assertEqual(reason, "evidence_binding_incomplete") + + def test_stale_and_future_rejected(self): + now = time.time() + old = bind_evidence(_hard_body(), now=now - 7200) + ok, reason = validate_bound_evidence(old, now=now) + self.assertFalse(ok) + self.assertEqual(reason, "evidence_too_old") + future = bind_evidence(_hard_body(), now=now + 3600) + ok, reason = validate_bound_evidence(future, now=now) + self.assertFalse(ok) + self.assertEqual(reason, "evidence_from_future") + + def test_nonce_single_use_via_store(self): + conn = sqlite3.connect(":memory:") + store = EvidenceNonceStore(conn) + stamped = bind_evidence(_hard_body(), nonce="nonce-1") + self.assertTrue(store.consume("t1", stamped["evidence_nonce"], + stamped["evidence_ts"])) + conn.commit() + # 原样重交：同一 nonce 第二次消费即重放。 + self.assertFalse(store.consume("t1", stamped["evidence_nonce"], + stamped["evidence_ts"])) + + def test_binding_does_not_authenticate_content(self): + """诚实边界：绑定防原样重放，不防重盖戳/篡改内容。 + + 盖戳后改 ci_status，校验仍通过——因为没有签名。 + 本项钉住该边界，防止后来人把 binding 误读成防篡改。 + """ + stamped = bind_evidence(_hard_body()) + stamped["ci_status"] = "failure" + ok, _reason = validate_bound_evidence(stamped) + self.assertTrue(ok) + + +class QuotaActualOnlyTestCase(unittest.TestCase): + def test_estimated_unpriced_rows_do_not_trip_cap(self): + tmp = tempfile.TemporaryDirectory() + try: + db_path = os.path.join(tmp.name, "control.db") + store = CostStore(db_path) + conn = sqlite3.connect(db_path) + try: + conn.execute("CREATE TABLE tasks(id TEXT PRIMARY KEY," + " created_by TEXT)") + conn.execute("INSERT INTO tasks(id, created_by)" + " VALUES(?, ?)", ("task-1", "user-1")) + conn.commit() + finally: + conn.close() + # actual 无价 100（60+40），estimated 无价 5000， + # actual 有价 1 行（usd 非 NULL，不进无价计数）。 + store.record("task-1", "proj", "terra-model", + input_tokens=60, output_tokens=40, + usd_micros=None, source="actual") + store.record("task-1", "proj", "terra-model", + input_tokens=5000, output_tokens=0, + usd_micros=None, source="estimated") + store.record("task-1", "proj", "gpt-priced", + input_tokens=10, output_tokens=10, + usd_micros=12345, source="actual") + month = time.strftime("%Y-%m", time.gmtime()) + conn = sqlite3.connect(db_path) + try: + used, unpriced = monthly_usage(conn, "user-1", month) + # 前后对比：main 旧 SQL（计全部 NULL 行）会得 + # 5100；actual-only 口径只得 100。 + old_row = conn.execute( + "SELECT COALESCE(SUM(c.input_tokens" + " + c.output_tokens), 0) FROM cost_records c" + " JOIN tasks t ON c.task_id = t.id" + " WHERE t.created_by = ? AND c.day LIKE ?" + " AND c.usd_micros IS NULL", + ("user-1", month + "%")).fetchone() + self.assertEqual(int(old_row[0]), 5100) + self.assertEqual(unpriced, 100) + self.assertEqual(used, 12345) + # 双侧同一口径：CostStore 报表 unpriced 亦为 100。 + report = store.report_by_task("task-1") + self.assertEqual(report["unpriced_actual_tokens"], 100) + # 上限判定：100 未超 1000 放行；5100（旧口径） + # 会误拦——这正是统一前服务端与执行侧的分歧点。 + self.assertIsNone(quota_decide(10000000, used, unpriced, + 1000, month)) + self.assertIsNotNone(quota_decide( + 10000000, used, 5100, 1000, month)) + finally: + conn.close() + finally: + tmp.cleanup() + + +class LedgerOnboardingTestCase(unittest.TestCase): + def test_missing_tables_not_ready(self): + conn = sqlite3.connect(":memory:") + report = ledger_onboarding.ledger_setup_report( + conn, hook_enabled=True) + self.assertFalse(report["ready"]) + self.assertFalse(report["tables_ready"]) + self.assertIn("task_attempts table missing", + report["missing"]) + self.assertIn("task_attempt_origin_markers table missing", + report["missing"]) + + def test_tables_ready_but_hook_off_not_ready(self): + conn = sqlite3.connect(":memory:") + record_attempt(conn, "task-1", "success", is_defect=False) + conn.commit() + ensure_marker_table(conn) + conn.commit() + report = ledger_onboarding.ledger_setup_report( + conn, hook_enabled=False) + self.assertTrue(report["tables_ready"]) + self.assertFalse(report["ready"]) + self.assertTrue(any("enable the hook" in s + for s in report["next_steps"])) + report_on = ledger_onboarding.ledger_setup_report( + conn, hook_enabled=True) + self.assertTrue(report_on["ready"]) + + def test_ddl_checklist_references_constants_not_copies(self): + sources = ledger_onboarding.ledger_ddl_sources() + self.assertEqual(sources["task_attempts_pg_ddl"], + LEDGER_PG_DDL) + self.assertEqual(sources["marker_pg_ddl"], MARKER_PG_DDL) + self.assertIn("task_attempts", LEDGER_PG_DDL) + self.assertIn("corrects_attempt_no", LEDGER_PG_DDL) + self.assertIn("task_attempt_origin_markers", MARKER_PG_DDL) + steps = [item["step"] for item in + ledger_onboarding.LEDGER_DDL_CHECKLIST] + self.assertEqual(steps, [1, 2, 3, 4, 5]) + + +class LedgerOutboxTestCase(unittest.TestCase): + def test_append_and_replay(self): + tmp = tempfile.TemporaryDirectory() + try: + path = os.path.join(tmp.name, "failures.jsonl") + self.assertTrue(ledger_outbox.append_failure( + path, "task-1", "success", "boom")) + records, bad = ledger_outbox.read_failures(path) + self.assertEqual(len(records), 1) + self.assertEqual(bad, 0) + conn = sqlite3.connect(":memory:") + stats = ledger_outbox.replay_failures(path, conn) + conn.commit() + self.assertEqual(stats["replayed"], 1) + self.assertEqual(stats["remaining"], 0) + rows = conn.execute( + "SELECT outcome FROM task_attempts" + " WHERE task_id = ?", ("task-1",)).fetchall() + self.assertEqual([row[0] for row in rows], ["success"]) + # 二次重放：outbox 已空，0 行，不会多写一行。 + stats2 = ledger_outbox.replay_failures(path, conn) + self.assertEqual(stats2["replayed"], 0) + rows = conn.execute( + "SELECT COUNT(*) FROM task_attempts" + " WHERE task_id = ?", ("task-1",)).fetchone() + self.assertEqual(rows[0], 1) + finally: + tmp.cleanup() + + def test_append_failure_never_raises(self): + # 路径父级是个文件时 makedirs 失败：返回 False，不抛。 + tmp = tempfile.TemporaryDirectory() + try: + blocker = os.path.join(tmp.name, "blocker") + with open(blocker, "w") as handle: + handle.write("x") + bad_path = os.path.join(blocker, "failures.jsonl") + self.assertFalse(ledger_outbox.append_failure( + bad_path, "task-1", "success", "boom")) + finally: + tmp.cleanup() + + def test_worker_record_failure_goes_to_outbox(self): + """worker 接线：_record_ledger_attempt 写失败时， + 除日志外 outbox 里留一行，且返回 None（不挡任务）。""" + import cloud.worker as worker_mod + tmp = tempfile.TemporaryDirectory() + try: + db_path = os.path.join(tmp.name, "control.db") + worker = worker_mod.Worker.__new__(worker_mod.Worker) + worker.db_path = db_path + worker.attempt_ledger_hook = True + worker.logger = logging.getLogger("test-v2-evidence") + original_connect = worker_mod._db_connect + + def _boom(_path): + raise RuntimeError("db down") + + worker_mod._db_connect = _boom + try: + result = worker._record_ledger_attempt( + "task-9", "success", True) + finally: + worker_mod._db_connect = original_connect + self.assertIsNone(result) + outbox = ledger_outbox.outbox_path_for(db_path) + records, bad = ledger_outbox.read_failures(outbox) + self.assertEqual(len(records), 1) + self.assertEqual(records[0]["task_id"], "task-9") + self.assertEqual(records[0]["outcome"], "success") + self.assertEqual(bad, 0) + finally: + tmp.cleanup() + + +class GateVerificationTestCase(unittest.TestCase): + def _live(self, checks, protection): + def reader(repo, ref): + self.assertEqual(repo, "o/r") + self.assertEqual(ref, "sha1") + return {"ci_checks": checks, + "branch_protection": protection} + return reader + + def test_matching_evidence_verified(self): + result = verify_gate_evidence( + _hard_body(), reader=self._live( + {"ci": "success"}, "protected"), + repo="o/r", ref="sha1") + self.assertTrue(result["checked"]) + self.assertTrue(result["verified"]) + self.assertEqual(result["mismatches"], []) + + def test_lied_ci_and_protection_mismatch(self): + result = verify_gate_evidence( + _hard_body(), reader=self._live( + {"ci": "failure"}, "unprotected"), + repo="o/r", ref="sha1") + self.assertTrue(result["checked"]) + self.assertFalse(result["verified"]) + self.assertTrue(any("ci_not_green_live" in m + for m in result["mismatches"])) + self.assertTrue(any("branch_protection" in m + for m in result["mismatches"])) + + def test_no_reader_is_not_a_pass(self): + result = verify_gate_evidence(_hard_body(), reader=None, + repo="o/r", ref="sha1") + self.assertFalse(result["checked"]) + self.assertFalse(result["verified"]) + self.assertEqual(result["reason"], "no_trusted_reader") + + def test_no_target_or_reader_failure_is_not_a_pass(self): + result = verify_gate_evidence( + _hard_body(), reader=self._live({}, "protected"), + repo=None, ref=None) + self.assertFalse(result["checked"]) + self.assertFalse(result["verified"]) + + def _boom(_repo, _ref): + raise RuntimeError("github down") + + result = verify_gate_evidence(_hard_body(), reader=_boom, + repo="o/r", ref="sha1") + self.assertFalse(result["checked"]) + self.assertFalse(result["verified"]) + self.assertTrue(result["reason"].startswith("reader_failed")) + + +class ReleaseEvidenceBindingServerTestCase(unittest.TestCase): + def setUp(self): + from cloud.auth import AuthStore + from cloud.server import create_server + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self._saved_env = { + k: os.environ.get(k) for k in ( + "CLOUD_MERGE_GATE_MODE", "CLOUD_BRANCH_PROTECTION", + "CODEX_GLOBAL_ADMINS")} + os.environ.pop("CLOUD_MERGE_GATE_MODE", None) + os.environ.pop("CLOUD_BRANCH_PROTECTION", None) + self.auth = AuthStore(self.db_path) + self.owner_id, self.owner_key = self.auth.create_user("owner") + os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + status, self.project = self._req( + "POST", "/projects", {"name": "demo"}, self.owner_key) + self.assertEqual(status, 201) + status, task = self._req( + "POST", "/tasks", + {"project_id": self.project["id"], "prompt": "x"}, + self.owner_key) + self.assertEqual(status, 201) + conn = sqlite3.connect(self.db_path) + try: + conn.execute( + "UPDATE tasks SET status = 'completed' WHERE id = ?", + (task["id"],)) + conn.commit() + finally: + conn.close() + status, _merged = self._req( + "POST", "/tasks/%s/merge" % task["id"], _hard_body(), + self.owner_key) + self.assertEqual(status, 200) + self.task_id = task["id"] + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + for key, value in self._saved_env.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + + def _req(self, method, path, body=None, key=None): + conn = http.client.HTTPConnection( + "127.0.0.1", self.port, timeout=10) + hdrs = {} + payload = None + if body is not None: + payload = json.dumps(body).encode("utf-8") + hdrs["Content-Type"] = "application/json" + if key is not None: + hdrs["X-API-Key"] = key + try: + conn.request(method, path, body=payload, 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 + + def test_replay_rejected_fresh_passes(self): + bound = bind_evidence(_hard_body()) + status, first = self._req( + "POST", "/tasks/%s/release" % self.task_id, bound, + self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(first["evidence_binding"], "bound") + # 原样重交同一份证据体：nonce 已消费，409 重放。 + status, replay = self._req( + "POST", "/tasks/%s/release" % self.task_id, bound, + self.owner_key) + self.assertEqual(status, 409) + self.assertEqual(replay["release_gate"]["code"], + "release_blocked_evidence_replayed") + # 新 nonce 的正常证据仍通过（其余字段完全相同）。 + fresh = bind_evidence(_hard_body()) + status, second = self._req( + "POST", "/tasks/%s/release" % self.task_id, fresh, + self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(second["evidence_binding"], "bound") + + def test_stale_partial_and_absent(self): + stale = bind_evidence(_hard_body(), now=time.time() - 7200) + status, resp = self._req( + "POST", "/tasks/%s/release" % self.task_id, stale, + self.owner_key) + self.assertEqual(status, 409) + self.assertEqual(resp["release_gate"]["code"], + "release_blocked_evidence_stale") + partial = _hard_body() + partial["evidence_ts"] = time.time() + status, _resp = self._req( + "POST", "/tasks/%s/release" % self.task_id, partial, + self.owner_key) + self.assertEqual(status, 400) + status, plain = self._req( + "POST", "/tasks/%s/release" % self.task_id, _hard_body(), + self.owner_key) + self.assertEqual(status, 200) + self.assertEqual(plain["evidence_binding"], "absent") + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/worker.py b/cloud/worker.py index 0b1f2a2..3373a9f 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -5668,6 +5668,22 @@ class Worker: except Exception as exc: log_event(self.logger, "worker_attempt_ledger_failed", message="attempt ledger write failed: %s" % exc) + # 集成 v2（cloud/ledger_outbox）：写失败除日志外再 + # 进 outbox，运维事后可重放补写——任务状态与返回值 + # 仍不受影响（规则 7 原样保留，台账不改 fail-closed， + # 审计 fail-closed 也不动；不对称的剩余部分见 + # ledger_outbox 模块头的边界说明）。outbox 追加本身 + # 也是 best-effort，失败只吞掉，不抛、不挡任务。 + try: + from cloud.ledger_outbox import append_failure \ + as _ledger_outbox_append + from cloud.ledger_outbox import outbox_path_for \ + as _ledger_outbox_path + _ledger_outbox_append( + _ledger_outbox_path(self.db_path), + task_id, outcome, exc) + except Exception: + pass return None def run_one(self, task=None):