diff --git a/cloud/release_gate.py b/cloud/release_gate.py new file mode 100644 index 0000000..6c64409 --- /dev/null +++ b/cloud/release_gate.py @@ -0,0 +1,203 @@ +"""Future release endpoint: the gate check, without the endpoint. + +This module is the wiring preparation for the release endpoint +that PR #247 explicitly left open: ``release_blocked`` is returned +alongside the merge response today, but nothing enforces it yet, +because there is no server-side release endpoint in this codebase +and ``deploy.py`` is an operator script, not a release endpoint +(see ``docs/release-gate-and-ledger-anchor.md``). + +What this module IS: + +- A pure, read-only evaluator: ``evaluate_release`` re-reads the + hash chain through the SAME reader the merge endpoint uses + (``cloud/audit_wiring.pushback_block_for_task``) and ANDs it + with a FRESH hard-gate evaluation supplied by the caller. It + never appends to the chain, never calls ``AuthStore.audit`` + (which writes a chain row in the same transaction), never + commits, and never deploys anything. Checking must not move + the chain head: otherwise the check itself would invalidate + the external anchor taken before it. +- The contract the future endpoint must implement. The endpoint + itself (route, HTTP plumbing, task-status transition, actual + publication) is NOT built here: building a live endpoint in + this track would be exactly the "new live release endpoint + touching production" the task forbids. The endpoint, when it + is approved and built, calls this evaluator and maps its + decision to HTTP exactly as documented in the companion doc. + +What this module is NOT: + +- It is not a merge gate. Merge (``POST /tasks/{id}/merge``) + and release are separate decisions with separate re-checks: + a merge-time 200 is never a release pass. The future endpoint + must call this evaluator at release time, with fresh evidence, + even if the task was merged minutes ago. +- It is not a deploy gate. ``deploy.py`` deploys the platform + itself (host release directory + unit restarts) and is + deliberately outside this gate; see the companion doc. +- It does not decide task status, auth, or publication. Status + (``task_status``) and the hard-gate evidence (``hard_gate``) + are caller-supplied inputs, so the evaluator can be tested + without a server, a task row, or a publication step. When the + caller omits an input, the evaluator fails closed on that + input instead of assuming it (missing hard-gate evidence + blocks; an unmerged status blocks when a status is given). + +Decision shape (``evaluate_release`` return value):: + + { + "allowed": bool, # True only if every check passes + "http_status": 200 | 409, # the status the endpoint maps to + "code": str, # machine code, see below + "blocked_by": str | None, # primary source, or None if allowed + "blocked_sources": [str], # every applicable source, in order + "blocks": [str], # hard-gate block codes, if any + "failed_checks": [str], # audit-pushback failed checks, if any + "round": int | None, # pushback round, if any + "modules": dict, # per-module block detail, if any + } + +Machine codes (``code`` / ``blocked_sources``): + +- ``release_allowed`` -- every check passed. +- ``release_requires_merged`` -- a task status was supplied and + it is not ``"merged"``. Release is a post-merge act; an + unmerged task has nothing to release. +- ``release_blocked_state_unavailable`` -- the chain could not + be read at all (anything other than the reader's own + missing-table rule, which means "no chain has ever been + written" and is inherited from #247 unchanged). Fail-closed: + an unreadable chain is never read as "no block". +- ``release_blocked_hard_gate`` -- the fresh hard-gate + evaluation (five checks: CI, directory, contract, coordinator, + reviews, plus branch protection) did not allow. The failing + codes travel in ``blocks`` verbatim, so the caller can tell + a hard-gate block from an audit block without parsing text. +- ``release_blocked_audit_pushback`` -- the re-read chain says + a pushback is still open (``release_blocked`` from the same + reader the merge endpoint uses). This is the #247 reader's + verdict at release time, not a cached merge-time value. + +When several sources block at once, ``blocked_sources`` lists +all of them (chain-unavailable dominates and short-circuits, +because without a chain read no other verdict is trustworthy), +while ``blocked_by``/``code`` name the first source in +evaluation order: chain read, then task status, then hard gate, +then audit pushback. The endpoint must return the full +decision object, not just the first code, so an operator can +see that fixing one source still leaves another blocking. + +CORRECTED-to-pass does NOT unblock (inherited, not invented): + +The reader already implements the #247/#244 resolved semantic: +a meta-review correction to ``pass`` schedules a re-audit, it +does not release the gate; only a fresh passing verdict or a +passing re-reviewed pushback row unblocks. This evaluator adds +no special case for corrections, so the release path cannot +drift from the merge path: the same chain, read through the +same reader, gives the same answer at both gates. The contract +tests pin this: corrected-to-pass with no new passing +conclusion stays blocked at release, exactly as at merge. +""" + +from cloud.audit_wiring import pushback_block_for_task + +# Machine codes (kept as module constants so the future endpoint +# and its tests spell them identically, never by string literal). +CODE_ALLOWED = "release_allowed" +CODE_REQUIRES_MERGED = "release_requires_merged" +CODE_STATE_UNAVAILABLE = "release_blocked_state_unavailable" +CODE_HARD_GATE = "release_blocked_hard_gate" +CODE_AUDIT_PUSHBACK = "release_blocked_audit_pushback" +CODE_HARD_GATE_EVIDENCE_MISSING = "release_hard_gate_evidence_missing" + + +def _decision(allowed, code, blocked_by, sources, blocks=None, + failed_checks=None, round_no=None, modules=None): + return { + "allowed": allowed, + "http_status": 200 if allowed else 409, + "code": code, + "blocked_by": blocked_by, + "blocked_sources": list(sources), + "blocks": list(blocks or []), + "failed_checks": list(failed_checks or []), + "round": round_no, + "modules": dict(modules or {}), + } + + +def evaluate_release(task_id, conn, hard_gate=None, task_status=None): + """Decide whether a task may be released, right now. + + ``task_id`` is the task whose chain is re-read. ``conn`` is a + DB connection the caller owns (the evaluator never closes + it, never commits on it, never writes through it). + ``hard_gate`` must be a FRESH evaluation of the hard merge + gate (``cloud/merge_gate_hard.evaluate_hard_merge_gate``) + against fresh evidence: ``{"allowed": bool, "blocks": [...]}``. + Passing ``None`` blocks with + ``release_hard_gate_evidence_missing`` -- a merge-time 200, + a cached gate object, or "the task is already merged" is + never accepted as fresh evidence. ``task_status``, when + given, must be ``"merged"``; when omitted, the status check + is skipped and the caller owns it (the future endpoint must + not omit it). + + Order: chain read first (unavailable short-circuits, because + every other verdict would be untrustworthy without it), then + task status, then hard gate, then audit pushback. Hard-gate + and pushback blocks are both collected, so a decision blocked + by both says so instead of hiding one behind the other. + """ + # 1. Re-read the chain, now -- never a cached merge-time value. + # The reader itself returns "allow" when the audit_ledger + # table does not exist yet (no chain ever written, #247 rule, + # inherited unchanged). Any other read failure propagates + # out of the reader and is fail-closed here. + try: + pushback = pushback_block_for_task(conn, task_id) + except Exception: + return _decision( + False, CODE_STATE_UNAVAILABLE, + "audit_pushback_state_unavailable", + [CODE_STATE_UNAVAILABLE]) + sources = [] + blocks = [] + failed_checks = [] + round_no = None + modules = {} + primary_code = None + primary_by = None + + def _note(code, by): + nonlocal primary_code, primary_by + sources.append(code) + if primary_code is None: + primary_code = code + primary_by = by + + # 2. Task status, when the caller supplies it. + if task_status is not None and task_status != "merged": + _note(CODE_REQUIRES_MERGED, "task_status") + + # 3. Fresh hard-gate evidence (five checks + protection). + if hard_gate is None: + _note(CODE_HARD_GATE_EVIDENCE_MISSING, "hard_gate") + elif not hard_gate.get("allowed"): + blocks = list(hard_gate.get("blocks") or []) + _note(CODE_HARD_GATE, "hard_gate") + + # 4. Audit pushback, re-read above. + if pushback.get("release_blocked"): + failed_checks = list(pushback.get("failed_checks") or []) + round_no = pushback.get("round") + modules = dict(pushback.get("modules") or {}) + _note(CODE_AUDIT_PUSHBACK, "audit_pushback") + + if not sources: + return _decision(True, CODE_ALLOWED, None, []) + return _decision(False, primary_code, primary_by, sources, + blocks=blocks, failed_checks=failed_checks, + round_no=round_no, modules=modules) diff --git a/cloud/tests/test_ledger_anchor_check.py b/cloud/tests/test_ledger_anchor_check.py new file mode 100644 index 0000000..767bbfd --- /dev/null +++ b/cloud/tests/test_ledger_anchor_check.py @@ -0,0 +1,152 @@ +"""Anchor script: exit-code contract for tools/ledger_anchor_check.py. + +Runs the script as a subprocess against throwaway SQLite chains +in a TemporaryDirectory (deleted with the test). Pins only what +the operator depends on: 0 pass, 1 verification failure with the +failure named, 2 usage error, and --write-anchor refused on a +failing chain. The script itself never writes the chain; the +tamper/tail cases simulate privileged tampering by dropping the +SQLite append-only triggers first (a normal UPDATE is already +blocked by the trigger, which is itself asserted here). +""" + +import json +import os +import sqlite3 +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPT = REPO_ROOT / "tools" / "ledger_anchor_check.py" + +try: + from cloud.audit_wiring import record_audit_verdict + from cloud.ledger import LedgerStore +except ImportError: + sys.path.insert(0, str(REPO_ROOT)) + from cloud.audit_wiring import record_audit_verdict + from cloud.ledger import LedgerStore + + +def _run(*args): + proc = subprocess.run( + [sys.executable, str(SCRIPT), *args], + capture_output=True, text=True, timeout=60) + try: + report = json.loads(proc.stdout) if proc.stdout.strip() else None + except ValueError: + report = None + return proc.returncode, report, proc.stderr + + +class LedgerAnchorCheckTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + store = LedgerStore(self.db_path) + record_audit_verdict(store, "t1", "p1", "apps/web", "v2", + "fail", "audit", failures=["ci_green"]) + record_audit_verdict(store, "t1", "p1", "apps/web", "v2", + "pass", "audit") + head = store.export()["head_hash"] + store.close() + self.anchor_path = os.path.join(self.tmp.name, "anchor.json") + with open(self.anchor_path, "w", encoding="utf-8") as handle: + json.dump({"version": 1, "head_hash": head, "count": 2}, + handle) + + def tearDown(self): + self.tmp.cleanup() + + def _copy_db(self, name): + import shutil + path = os.path.join(self.tmp.name, name) + shutil.copyfile(self.db_path, path) + return path + + def test_pass_without_and_with_anchor(self): + code, report, _ = _run("--db", self.db_path) + self.assertEqual(code, 0) + self.assertTrue(report["ok"]) + code, report, _ = _run("--db", self.db_path, + "--anchor", self.anchor_path) + self.assertEqual(code, 0) + self.assertTrue(report["ok"]) + self.assertEqual(report["head_advanced_by"], 0) + + def test_normal_update_blocked_by_trigger_then_privileged_tamper_fails(self): + path = self._copy_db("tamper.db") + conn = sqlite3.connect(path) + with self.assertRaises(sqlite3.IntegrityError): + conn.execute( + "UPDATE audit_ledger SET detail='x' WHERE id=1") + conn.rollback() + conn.execute("DROP TRIGGER audit_ledger_no_update") + conn.execute("UPDATE audit_ledger SET detail='tampered'" + " WHERE id=1") + conn.commit() + conn.close() + code, report, _ = _run("--db", path, + "--anchor", self.anchor_path) + self.assertEqual(code, 1) + self.assertFalse(report["ok"]) + self.assertTrue(any("entry_hash mismatch" in f + for f in report["failures"])) + + def test_tail_deletion_fails_against_old_anchor(self): + path = self._copy_db("tail.db") + conn = sqlite3.connect(path) + conn.execute("DROP TRIGGER audit_ledger_no_delete") + conn.execute("DELETE FROM audit_ledger WHERE id=2") + conn.commit() + conn.close() + code, report, _ = _run("--db", path, + "--anchor", self.anchor_path) + self.assertEqual(code, 1) + self.assertFalse(report["ok"]) + self.assertTrue(any("shorter than anchor" in f + for f in report["failures"])) + + def test_extended_chain_still_passes_anchor(self): + store = LedgerStore(self.db_path) + record_audit_verdict(store, "t1", "p1", "apps/web", "v2", + "pass", "audit") + store.close() + code, report, _ = _run("--db", self.db_path, + "--anchor", self.anchor_path) + self.assertEqual(code, 0) + self.assertTrue(report["ok"]) + self.assertEqual(report["head_advanced_by"], 1) + + def test_write_anchor_refused_on_failing_export(self): + path = self._copy_db("tail2.db") + conn = sqlite3.connect(path) + conn.execute("DROP TRIGGER audit_ledger_no_delete") + conn.execute("DELETE FROM audit_ledger WHERE id=2") + conn.commit() + conn.close() + store = LedgerStore(path) + export_path = os.path.join(self.tmp.name, "export.json") + with open(export_path, "w", encoding="utf-8") as handle: + json.dump(store.export(), handle) + store.close() + out_path = os.path.join(self.tmp.name, "new-anchor.json") + code, _report, _ = _run("--export", export_path, + "--anchor", self.anchor_path, + "--write-anchor", out_path) + self.assertEqual(code, 1) + self.assertFalse(os.path.exists(out_path)) + + def test_usage_errors_exit_2(self): + code, _report, _ = _run( + "--db", os.path.join(self.tmp.name, "nope.db")) + self.assertEqual(code, 2) + code, _report, _ = _run("--db", "postgres://host/db") + self.assertEqual(code, 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_release_gate.py b/cloud/tests/test_release_gate.py new file mode 100644 index 0000000..16ea325 --- /dev/null +++ b/cloud/tests/test_release_gate.py @@ -0,0 +1,268 @@ +"""Future release endpoint: contract tests for cloud/release_gate. + +These tests pin the CONTRACT, not an endpoint: there is no live +release endpoint in this codebase (building one here would be +the "new live endpoint touching production" this track forbids), +so every test calls ``evaluate_release`` directly with a local +SQLite chain. Nothing here deploys, publishes, or touches a +production flow; the chain lives in a TemporaryDirectory and is +deleted with the test. + +What is pinned (and nowhere else): + +- A blocked audit pushback blocks release (409, audit code), + even when the fresh hard gate passes. +- A passing re-audit unblocks release -- but only with a FRESH + passing hard gate supplied by the caller; a merge-time 200 is + never an input, and a missing hard gate blocks fail-closed. +- A chain-read failure blocks fail-closed (state unavailable), + distinguishable from an audit block and from a hard-gate block. +- A hard-gate failure blocks release even when the chain is + clear, and both sources are listed when both block. +- CORRECTED-to-pass does NOT unblock: a meta-review correction + to ``pass`` with no new passing conclusion stays blocked; only + a fresh passing verdict unblocks (the #247/#244 semantic, + re-checked at release time through the same reader). +- A fix alone does not unblock. +- A merge-time pass is not reused: a chain that was clear at + "merge time" and is blocked by the time release is evaluated + blocks release (the evaluator re-reads the chain every call). +- The evaluator is read-only: evaluating never moves the chain + head (no row is appended by a check). +""" + +import os +import sqlite3 +import sys +import tempfile +import unittest +from pathlib import Path + +try: + from cloud.audit_pushback import GATE_CHECKS # noqa: F401 + from cloud.audit_wiring import pushback_package_from_gate + from cloud.audit_wiring import record_audit_verdict + from cloud.audit_wiring import record_fix + from cloud.audit_wiring import record_meta_review + from cloud.audit_wiring import record_pushback + from cloud.audit_wiring import record_reaudit + from cloud.ledger import LedgerStore + from cloud.merge_gate_hard import evaluate_hard_merge_gate + from cloud.release_gate import evaluate_release +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.audit_pushback import GATE_CHECKS # noqa: F401 + from cloud.audit_wiring import pushback_package_from_gate + from cloud.audit_wiring import record_audit_verdict + from cloud.audit_wiring import record_fix + from cloud.audit_wiring import record_meta_review + from cloud.audit_wiring import record_pushback + from cloud.audit_wiring import record_reaudit + from cloud.ledger import LedgerStore + from cloud.merge_gate_hard import evaluate_hard_merge_gate + from cloud.release_gate import evaluate_release + +MODULE = "apps/web" +CONTRACT = "v2" + + +def _passing_gate(): + return evaluate_hard_merge_gate( + files_changed=["apps/web/a.py"], owned_dir="apps/web", + 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) + + +def _failing_gate(): + return evaluate_hard_merge_gate( + files_changed=["apps/web/a.py"], owned_dir="apps/web", + protection="protected", ci_status="pending", + ci_checks=[{"name": "ci", "status": "pending"}], + required_checks=["ci"], contract_version="v2", + contract_frozen=True, coordinator_approved=True, + unresolved_review_comments=0) + + +class ReleaseGateTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.store = LedgerStore(self.db_path) + self.task = "task-1" + self.project = "proj-1" + + def tearDown(self): + try: + self.store.close() + except Exception: + pass + self.tmp.cleanup() + + def _conn(self): + return sqlite3.connect(self.db_path) + + def _decide(self, hard_gate=None, task_status="merged"): + conn = self._conn() + try: + return evaluate_release( + self.task, conn, hard_gate=hard_gate, + task_status=task_status) + finally: + conn.close() + + def _block_task(self): + record_audit_verdict( + self.store, self.task, self.project, MODULE, CONTRACT, + "fail", "audit", failures=["ci_green"]) + package = pushback_package_from_gate( + _failing_gate(), self.task, 0) + record_pushback(self.store, self.task, self.project, MODULE, + CONTRACT, 1, package["failed_checks"], package, + failure_source="audit") + return package + + def _head(self): + return self.store.verify()["head_hash"] + + def test_audit_pushback_blocks_release_even_with_passing_gate(self): + self._block_task() + decision = self._decide(hard_gate=_passing_gate()) + self.assertFalse(decision["allowed"]) + self.assertEqual(decision["http_status"], 409) + self.assertEqual(decision["code"], + "release_blocked_audit_pushback") + self.assertEqual(decision["blocked_by"], "audit_pushback") + self.assertIn("release_blocked_audit_pushback", + decision["blocked_sources"]) + self.assertIn("ci_green", decision["failed_checks"]) + self.assertEqual(decision["round"], 1) + + def test_reaudit_pass_allows_release_with_fresh_passing_gate(self): + package = self._block_task() + record_reaudit(self.store, self.task, self.project, MODULE, + CONTRACT, 1, {"verdict": "pass"}, + package=package) + decision = self._decide(hard_gate=_passing_gate()) + self.assertTrue(decision["allowed"]) + self.assertEqual(decision["http_status"], 200) + self.assertEqual(decision["code"], "release_allowed") + self.assertIsNone(decision["blocked_by"]) + self.assertEqual(decision["blocked_sources"], []) + + def test_corrected_to_pass_still_blocks_until_fresh_pass(self): + # One failing verdict is the ONLY block here, so the + # correction below is the only thing that could unblock + # -- if it did, this test would catch it. (A second fail + # or an open pushback would mask a wrong unblock.) + verdict = record_audit_verdict( + self.store, self.task, self.project, MODULE, CONTRACT, + "fail", "audit", failures=["ci_green"]) + self.assertFalse( + self._decide(hard_gate=_passing_gate())["allowed"]) + # Audit-of-audit correction: the original fail was wrong, + # corrected to pass. That schedules a re-audit; it does + # NOT release the gate by itself. + record_meta_review( + self.store, self.task, self.project, MODULE, CONTRACT, + verdict["id"], "fail", "pass", True, reviewer="auditor-b") + decision = self._decide(hard_gate=_passing_gate()) + self.assertFalse(decision["allowed"]) + self.assertEqual(decision["code"], + "release_blocked_audit_pushback") + # Only a fresh passing verdict unblocks. + record_audit_verdict( + self.store, self.task, self.project, MODULE, CONTRACT, + "pass", "audit") + decision = self._decide(hard_gate=_passing_gate()) + self.assertTrue(decision["allowed"], decision) + + def test_hard_gate_blocks_release_when_chain_is_clear(self): + record_audit_verdict( + self.store, self.task, self.project, MODULE, CONTRACT, + "pass", "audit") + decision = self._decide(hard_gate=_failing_gate()) + self.assertFalse(decision["allowed"]) + self.assertEqual(decision["http_status"], 409) + self.assertEqual(decision["code"], "release_blocked_hard_gate") + self.assertEqual(decision["blocked_by"], "hard_gate") + self.assertTrue(decision["blocks"]) + self.assertNotIn("release_blocked_audit_pushback", + decision["blocked_sources"]) + + def test_both_sources_listed_when_both_block(self): + self._block_task() + decision = self._decide(hard_gate=_failing_gate()) + self.assertFalse(decision["allowed"]) + self.assertIn("release_blocked_hard_gate", + decision["blocked_sources"]) + self.assertIn("release_blocked_audit_pushback", + decision["blocked_sources"]) + self.assertTrue(decision["blocks"]) + self.assertTrue(decision["failed_checks"]) + + def test_chain_read_failure_fails_closed(self): + conn = self._conn() + conn.close() + decision = evaluate_release( + self.task, conn, hard_gate=_passing_gate(), + task_status="merged") + self.assertFalse(decision["allowed"]) + self.assertEqual(decision["http_status"], 409) + self.assertEqual(decision["code"], + "release_blocked_state_unavailable") + self.assertEqual( + decision["blocked_by"], + "audit_pushback_state_unavailable") + + def test_missing_hard_gate_evidence_blocks(self): + record_audit_verdict( + self.store, self.task, self.project, MODULE, CONTRACT, + "pass", "audit") + decision = self._decide(hard_gate=None) + self.assertFalse(decision["allowed"]) + self.assertEqual(decision["code"], + "release_hard_gate_evidence_missing") + + def test_unmerged_task_cannot_release(self): + record_audit_verdict( + self.store, self.task, self.project, MODULE, CONTRACT, + "pass", "audit") + decision = self._decide(hard_gate=_passing_gate(), + task_status="completed") + self.assertFalse(decision["allowed"]) + self.assertEqual(decision["code"], "release_requires_merged") + + def test_merge_time_pass_is_not_reused_at_release(self): + # "Merge time": chain clear, fresh gate passes. + first = self._decide(hard_gate=_passing_gate()) + self.assertTrue(first["allowed"], first) + # Between merge and release, a pushback lands on the chain. + self._block_task() + second = self._decide(hard_gate=_passing_gate()) + self.assertFalse(second["allowed"]) + self.assertEqual(second["code"], + "release_blocked_audit_pushback") + + def test_fix_alone_does_not_unblock_release(self): + package = self._block_task() + record_fix(self.store, self.task, self.project, MODULE, + CONTRACT, 1, + {"summary": "fix ci", "diff_sha256": "b" * 64, + "files_changed": ["apps/web/a.py"]}, package) + decision = self._decide(hard_gate=_passing_gate()) + self.assertFalse(decision["allowed"]) + + def test_evaluator_is_read_only(self): + self._block_task() + before = self._head() + self._decide(hard_gate=_passing_gate()) + self._decide(hard_gate=_failing_gate()) + self.assertEqual(self._head(), before) + self.assertTrue(self.store.verify()["ok"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/release-gate-and-ledger-anchor.md b/docs/release-gate-and-ledger-anchor.md new file mode 100644 index 0000000..745996c --- /dev/null +++ b/docs/release-gate-and-ledger-anchor.md @@ -0,0 +1,401 @@ +# 发布端点接线准备 + 账本锚点运维流程 + +> 本轨性质:**接线准备,不是发布端点本身**。本轨不新建 live +> 发布端点、不碰生产、不跑 `deploy.py`、不合并、不部署。 +> 发布端点是否立项、锚点存哪里、校验周期多长,均属用户侧 +> 拍板项(见 §6),本文只给推荐值与判断依据,不替用户定。 +> +> 基底:堆叠于 PR #247(`feat-audit-wiring`,`217a27f`), +> 合并顺序:#209 → #240 → #244 → #241 → #247 → 本轨。 +> 本轨代码:`cloud/release_gate.py`(纯只读 evaluator)、 +> `cloud/tests/test_release_gate.py`(11 项契约测试)、 +> `tools/ledger_anchor_check.py`(只读校验脚本)、 +> `cloud/tests/test_ledger_anchor_check.py`(脚本退出码测试)。 +> 本轨不改 #247 / #241 / #244 / #240 / #209 的任何语义: +> 硬门禁五项、推回阻塞、CORRECTED-to-pass 不解封、链 +> fail-closed 规则,全部沿用,本文只说明它们在发布时 +> 如何被**重新**执行。 + +## 1. 现状缺口(#247 留下的原话) + +- 合并端点 `POST /tasks/{id}/merge` 已在链上推回阻塞与硬门禁 + 五项之间做 AND:任一阻塞即 409、状态不变,并在响应里附 + `audit_pushback` 对象(`merge_blocked` / `release_blocked` / + `blocked_by="audit_pushback"` / `round` / `failed_checks`)。 +- 但本仓库**没有独立的发布端点**:`release_blocked` 今天只是 + 随合并响应返回的一个字段,由“未来的发布端点用同一 reader + 强制”。在那之前,合并放行后、发布前没有任何服务端检查 + 会再读一次链。 +- 账本(#241)已有导出头哈希,但**没有库外锚点存放与定期 + 校验的运维流程**:头哈希只在导出响应里出现一次,不存下来 + 就等于没有锚点;尾巴删除单靠库内校验发现不了(#241 已明 + 说),必须靠“旧锚点 vs 当前链”的比对。 + +## 2. 未来发布端点接口契约(建议,待立项后实现) + +### 2.1 端点 + +| 项 | 建议值 | 说明 | +|---|---|---| +| 方法 + 路径 | `POST /tasks/{id}/release` | 与 `POST /tasks/{id}/merge` 对仗;`{id}` 为任务 id,路径风格与现有任务端点一致 | +| 请求体 | `{}` 或 `{"confirm": true}`(二选一,立项时定) | 本轨不规定体字段;关键在服务端**不信任**体里的任何门禁结论,只信任自己重算的结果 | +| 成功 | `200` + 任务对象 + `release_gate` 决策对象 | 决策对象即 `evaluate_release` 的返回值原样返回,不另起一套字段 | +| 阻塞 | `409` + `{"error": "release blocked", "release_gate": {...}}` | 见 §2.4 的阻塞码;阻塞时**不改任务状态、不发布、不写链** | +| 任务不存在 | `404` | 与 merge 端点一致 | +| 无权 | `403` | 见 §2.2 | + +本轨**没有实现这个端点**:`cloud/release_gate.py` 只有 evaluator, +没有路由、没有 HTTP plumbing、没有任务状态变更、没有发布动作。 +契约测试直接测 evaluator(见 §4),不测一个不存在的端点。 + +### 2.2 鉴权:与 verify/export 同源、但为写动作 + +- 身份来源与 `GET /admin/audit-ledger/verify`、 + `GET /admin/audit-ledger/export` **一致**:只认环境变量 + `CODEX_GLOBAL_ADMINS`(`_is_global_admin`),未配置即全体 + 403 fail-closed,绝不回落到项目角色推导。 +- 但动作性质**不一致**:verify/export 是全局管理员**只读**; + 发布是**写动作**(对外发布),所以发布端点必须是 + “全局管理员写”,不能是“全局管理员只读”,更不能只凭 + 项目 `approve_merge`(合并权)就放行发布。合并与发布是 + 两道不同的授权:项目 maintainer 能合并不等于能发布全系统。 +- 被拒(403)的审计口径沿用现有 `_audit_admin` 惯例(成功与 + 被拒都记;注意 `AuthStore.audit` 会同事务写链,见 §2.5 的 + 头移动说明)。409 阻塞本身**不写审计**:阻塞只是只读检查 + 的结果,每次被拒查询都写链会无意义地移动链头、污染锚点。 + +### 2.3 接线点:发布前必做的三件事(顺序即 §3 的求值顺序) + +未来端点在执行任何发布动作之前,必须按顺序做: + +1. **重读链**:调 `evaluate_release`,它内部调 + `cloud/audit_wiring.pushback_block_for_task`(与 merge 端点 + 同一个 reader)。读到的必须是**发布时刻**的链状态。 +2. **重算硬门禁**:用**发布时刻的新证据**重新跑 + `evaluate_hard_merge_gate`(CI、目录、合同、协调员、评审、 + 分支保护五项 + 保护),把结果传给 evaluator。合并时的 + gate 对象、合并时的 200,一律不作输入。 +3. **核任务状态**:任务状态须为 `merged` 才可发布(evaluator + 的 `task_status` 参数;端点不得省略它)。 + +三件缺一即 fail-closed:链读不到 → 409;硬门禁证据缺失 → +409;状态不是 `merged` → 409。发布动作只在 evaluator 返回 +`allowed=true` 后执行,且发布动作本身的成功记录(如需记账) +必须是**显式的新动作**,不是检查的副产品。 + +### 2.4 阻塞源区分(错误码与阻塞码对照) + +`release_gate` 决策对象同时返回 `code`(首要阻塞码)、 +`blocked_by`(首要阻塞源)、`blocked_sources`(全部阻塞源), +端点必须原样返回整个对象,不能只返回首要码——否则修好 +一个源后才发现另一个源仍阻塞,运维会被误导。 + +| 阻塞源 | `code` | `blocked_by` | HTTP | 附加字段 | 修复路径 | +|---|---|---|---|---|---| +| 链读失败 fail-closed | `release_blocked_state_unavailable` | `audit_pushback_state_unavailable` | 409 | 无(读不到即无可信细节) | 修读链通路(DB/权限),重调;绝不能“先放行、后补查” | +| 任务未合并 | `release_requires_merged` | `task_status` | 409 | — | 先走合并流程到 `merged` | +| 硬门禁五项失败 | `release_blocked_hard_gate` | `hard_gate` | 409 | `blocks`(硬门禁码原样,如 `ci_not_green`) | 补齐五项证据后重调;与 merge 的 blocks 码同一套,不解析文本 | +| 审计推回阻塞 | `release_blocked_audit_pushback` | `audit_pushback` | 409 | `failed_checks`、`round`、`modules`(与 merge 响应里的 `audit_pushback` 对象同源) | 修问题 → 重审通过(见 §2.6),重调 | +| 硬门禁证据缺失 | `release_hard_gate_evidence_missing` | `hard_gate` | 409 | — | 调用方必须提交发布时刻的新证据;这不是“证据不足放行”,是直接阻塞 | + +例外(沿用 #247,不扩大):`audit_ledger` 表根本不存在时, +reader 返回“无推回状态”(链从未写过),不算读链失败。 +这条沿用 merge 的口径;若未来组织要求发布比合并更严 +(无链即不发布),那是立项时另议的语义变更,不在本轨 +偷偷改(见 §6 待拍板项)。 + +### 2.5 只读性:检查绝不能写链 + +- evaluator 不 append、不 commit、不调 `AuthStore.audit`、 + 不关调用方的连接。契约测试 `test_evaluator_is_read_only` + 钉住:两次求值后链头不变、链仍 verify ok。 +- 为什么必须如此:`AuthStore.audit` 会同事务写一条链记录; + 若检查写审计,检查本身就移动链头,检查前取的锚点立刻 + 失效,锚点校验将永远对不上——检查就不再是检查。 + 现有 `GET /admin/audit-ledger/verify` 成功路径同样不写审计, + 正是同一道理。 +- 推论:409 阻塞不写审计(§2.2);403 被拒沿用 `_audit_admin` + 惯例写审计时,运维须知**链头已移动**,下一次锚点必须在 + 该行之后重取(锚点轮换见 §5.4)。 + +### 2.6 `release_blocked` 与 `merge_blocked` 的关系 + +- **合并放行 ≠ 发布放行**。`merge_blocked=false` 只说明 + “合并那一刻链上无阻塞且五项全过”;发布时刻链可能已 + 新增推回、CI 可能已变红、合同可能已变——发布必须重新 + 读链、重新判定。拿合并时的 200 当发布通行证是本设计 + 明确禁止的用法,evaluator 的签名即防住了它:它不接受 + “合并时的 gate 对象”作为输入,只接受调用方现算的 + `hard_gate`,且 `hard_gate=None` 直接阻塞。 +- 反向亦然:发布阻塞不回滚合并状态。发布 409 时任务仍为 + `merged`;修复后重调发布即可,不必重新合并。 +- `release_blocked` 与 `merge_blocked` 在 reader 里是同一 + 布尔值(`gate_block` 同时置两字段),但**求值时刻不同**: + 同一个 reader、两次调用、两次时刻——这才是两字段并存的 + 意义,而不是“合并时顺带算好发布结论”。 + +### 2.7 CORRECTED 为 pass 在发布端如何体现 + +- 沿用 #247/#244 已裁决的语义,不发明新语义:审计的审计 + 把原结论纠正为 `pass`,**本身不解封**——它只表示“原判错了, + 应安排重审”,不是一次重审。发布端与合并端共用同一 reader, + 因此发布端自动继承该语义:纠正后、无新通过结论时,发布 + 仍 409(`release_blocked_audit_pushback`)。 +- 解封的唯一路径:一条**新的**通过结论进链——新的 + `audit_verdict=pass`,或一次通过的 `re_reviewed` 推回轮次 + (且 QA 失败链接须按 #247 的 QA 护栏全部重跑通过,否则 + 该通过结论写不进链)。 +- 契约测试 `test_corrected_to_pass_still_blocks_until_fresh_pass` + 用**唯一阻塞**构造钉住这一点:若纠正本身能解封,该测试 + 必失败(测试注释说明了为何不用“纠正 + 另一失败”的掩盖构造)。 + +## 3. evaluator 求值顺序(`cloud/release_gate.py`) + +1. 读链。异常 → `release_blocked_state_unavailable`,短路 + (没有可信链读数时,其它结论都不可信)。 +2. 任务状态(仅当调用方给了 `task_status` 时)。非 `merged` + → `release_requires_merged`。 +3. 硬门禁。`hard_gate=None` → `release_hard_gate_evidence_missing`; + `allowed=false` → `release_blocked_hard_gate`,`blocks` 原样带出。 +4. 审计推回。reader 说阻塞 → `release_blocked_audit_pushback`, + `failed_checks` / `round` / `modules` 原样带出。 +5. 以上全过 → `allowed=true`,`http_status=200`, + `code=release_allowed`。 + +多源同时阻塞时,`blocked_sources` 列出全部,`code` / +`blocked_by` 只取求值顺序第一项。端点返回时不得丢弃 +`blocked_sources`。 + +## 4. 契约测试(`cloud/tests/test_release_gate.py`,11 项) + +测试只测 evaluator + 本地 SQLite 链(`TemporaryDirectory`, +跑完删除),**不部署、不调用生产发布流程**(生产本就没有 +发布端点可调)。覆盖: + +| 测试 | 断言 | +|---|---| +| 审计推回阻塞(硬门禁全过时) | 409 + `release_blocked_audit_pushback` + `failed_checks` 含 `ci_green` + `round=1` | +| 重审通过后放行(新硬门禁全过) | 200 + `release_allowed` | +| 纠正为 pass 但无新通过结论时仍阻塞;新通过结论后放行 | 前 409、后 200(唯一阻塞构造,见 §2.7) | +| 链读失败(已关闭的连接) | 409 + `release_blocked_state_unavailable` | +| 硬门禁失败而链已清 | 409 + `release_blocked_hard_gate` + `blocks` 非空,且不含审计阻塞码 | +| 两源同时阻塞 | `blocked_sources` 同时含两码,`blocks` 与 `failed_checks` 均非空 | +| 硬门禁证据缺失 | 409 + `release_hard_gate_evidence_missing` | +| 任务未合并 | 409 + `release_requires_merged` | +| 合并时放行、发布时链已阻塞 | 前 200、后 409(证明每次调用都重读链) | +| 仅修复不解封 | fix 行后仍 409 | +| evaluator 只读 | 两次求值后链头不变、链 verify ok | + +## 5. 账本锚点运维流程 + +### 5.1 存放位置(推荐一支 + 取舍 + 权限) + +**推荐:独立对象存储(与生产库不同机、不同账号、不同权限)。** + +- 每个锚点一个对象,命名 + `anchor---.json`,内容即 + §5.2 的锚点文件(`{version, head_hash, count, anchored_at, + source}`,仅 `head_hash` + `count` 参与校验)。 +- 存储桶开启**版本控制 + 对象锁定/保留(WORM)或至少禁止 + 删除的桶策略**:写锚点的账号只有 `PutObject`,没有 + `DeleteObject`、没有覆盖保留期的权限;读锚点的账号(审计 + 员)只有 `GetObject`。写与读、锚点与生产库,账号全部分开。 +- `current` 指针(如有)只是一个方便引用的独立小对象, + 更新指针绝不删除旧锚点对象;历史锚点对象本身就是真相, + 指针丢了可以从对象列表重建,对象丢了指针毫无意义。 + +取舍(为什么不选其它位置): + +| 候选位置 | 否决/降级原因 | +|---|---| +| 生产库同一张表/同库另一张表 | 与生产库同权限:能改链的人就能改锚点,锚点失去独立见证意义 | +| 生产主机上的文件 | 与生产库同机:主机被拿下则锚点与链一起被改写;且主机文件无版本控制,易被覆盖 | +| 运维人员个人机器文件 | 可作**过渡**(见下),但单点、无审计、无保留策略,人员变动即丢失 | +| Git 仓库 | 历史可查是优点,但锚点文件进仓等于把“见证”交给与代码同一套合并权限,且 head 频繁更新会污染提交历史;不推荐为主存 | + +**过渡方案(在对象存储未就绪前,须用户拍板后才用)**: +锚点文件存到一台**非生产主机**的运维账号目录(`0600`), +该账号与生产库账号不同、无生产库写权限;同时把锚点 +`head_hash` 抄送到第二渠道(如运维群/邮件归档)作人工见证。 +过渡不是终态:对象存储就绪后,历史锚点文件应原样迁入 +对象存储并保留过渡期的抄送记录,不得“迁入时重算头哈希”。 + +### 5.2 校验什么、多久一次、谁触发 + +每次校验(脚本 `tools/ledger_anchor_check.py`)做两层: + +1. **链完整性**:`verify_chain` / `verify_export`——逐条重算 + `entry_hash` + `prev_hash` 链接。中间篡改、重排、删中间 + 记录,在此层失败。 +2. **锚点前缀比对**:当前链条数 ≥ 锚点 `count`,且第 `count` + 条的 `entry_hash` 仍等于锚点 `head_hash`。条数变少 → + 锚定后的尾巴被截断;该位置哈希变了 → 锚点前的链被 + 替换/重写。头哈希**前进**(条数更多、前缀一致)是正常 + 通过,不是失败。 + +周期(推荐值,待用户拍板,见 §6): + +| 触发 | 动作 | +|---|---| +| 每日定时(cron/定时任务,运维账号) | 取最新锚点 → 全链校验 + 前缀比对;通过则取新锚点(轮换见 §5.4) | +| 每次发布前(未来发布端点立项后) | 发布流程先跑一次锚点校验;校验失败即发布 409(与 §2.4 的链读失败同级处理,属未来端点行为,本轨不实现) | +| 每次合并门禁异常/人工介入后 | 运维手动跑一次,确认链未被介入改动 | +| 锚点轮换时 | 新锚点必须先通过旧锚点的前缀校验,才能成为“当前锚点”(见 §5.4) | + +校验脚本本身不触发任何业务动作:它只读(SQLite 用 +`mode=ro` 打开;导出模式只读文件),不写链、不发通知—— +通知由调用它的定时任务/运维流程负责(见 §5.3)。 + +### 5.3 校验失败的告警与处理(绝不能静默重建) + +失败分两类,处理不同,但有一条共同红线: + +> **红线:绝不能删除旧链、重建整链、重算头哈希后冒充 +> “没发生”。** 任何恢复都必须保留原始链与旧锚点,恢复 +> 动作本身也要留痕(新锚点系列注明事件编号,见 §5.4)。 + +- **告警给谁**:全局管理员 + 当值运维(渠道由用户定: + 本轨不假设已有告警通道;定时任务失败时至少要在定时 + 任务日志与运维值班渠道留一条可查记录,不能只打印到 + 没人看的日志文件里)。 +- **是否冻结**:校验失败期间,**发布立即冻结**(不发布; + 未来发布端点立项后以校验失败作 409 条件之一;立项前由 + 运维人工冻结发布操作)。合并是否冻结由当值全局管理员 + 当场决定并记录——本轨不以代码自动冻结合并(那会是 + 新的门禁语义,须另行立项),但 runbook 要求管理员在 + 收到告警后 1 个工作日内给出“冻结合并与否”的书面决定, + 不能无决定地继续合并。 +- **恢复步骤**: + 1. 立即对生产库做只读快照/备份,冻结现场(不要先修链)。 + 2. 区分失败类:锚点文件笔误/取错(旧锚点系列中存在能 + 通过前缀校验的锚点)→ 以那个锚点为准重取当前锚点, + 并记录“锚点取错事件”;链本身失败(重算哈希不符、 + 链接断裂、前缀对不上任何历史锚点)→ 链不可信。 + 3. 链不可信时:旧链**原样封存**(只读,不删不改), + 新链只能以“事件后新系列”起头(新 genesis 标记注明 + 事件编号与旧链封存位置),新锚点系列与旧系列分开存放、 + 不得混写;旧系列永不删除。 + 4. 恢复后重新锚定:新锚点只能取自**已通过内部校验且 + 通过最老可信锚点前缀校验**的链;若无任何可信锚点, + 新系列首锚点必须注明“无可信前缀、事件后重建”,不得 + 写成与旧系列连续的样子。 + +### 5.4 锚点轮换与历史锚点保留 + +- **轮换**:每次校验通过后取新锚点(新文件、新对象), + 旧锚点文件/对象**永不覆盖、永不删除**。新锚点生效前必须 + 通过旧锚点的前缀校验——通不过的链不配产生新锚点。 +- **保留期**:锚点文件极小(约 200 字节),**建议永久保留 + 全部历史锚点**,不设删除期;对象存储侧以版本控制 + + 保留策略保证“删不掉”,而不是靠运维自觉不删。若存储 + 成本政策强制设限,则下限为:至少保留最近 12 个月的 + 每日锚点 + 每次发布前的锚点,且删除须全局管理员双人 + 确认并留审计记录——该下限须用户拍板(见 §6),本轨 + 不擅自把“永久”降级为某个期限。 +- **交接**:新旧锚点交接只认“新锚点通过旧锚点前缀校验” + 这一条;指针(`current`)更新失败不影响历史锚点的有效性, + 下次校验可直接用任意历史锚点重跑。 + +### 5.5 校验脚本用法与退出码 + +``` +# SQLite 本地链(只读打开,不写链): +python3 tools/ledger_anchor_check.py --db /path/to/control.db +python3 tools/ledger_anchor_check.py --db /path/to/control.db --anchor /path/to/anchor.json + +# 生产 PG:先经 GET /admin/audit-ledger/export 导出,再验导出(脚本不开 PG 连接): +python3 tools/ledger_anchor_check.py --export /path/to/export.json --anchor /path/to/anchor.json +# 校验通过后取新锚点(仅 --export 模式;失败时拒绝写文件,见 §5.5 末): +python3 tools/ledger_anchor_check.py --export /path/to/export.json --write-anchor /path/to/new-anchor.json +# --db 模式取新锚点:用通过的那次检查输出的 head_hash/count,按 §5.2 格式写新锚点文件 +# (脚本不代写,避免“检查脚本写文件”与“只读检查”的语义混淆;--export 模式的 --write-anchor +# 只写锚点文件、从不写链,失败时同样拒绝)。 +``` + +退出码:`0` 通过;`1` 校验失败(stdout JSON 的 `failures` +逐条列出失败项,`head_advanced_by` 为负即链比锚点短); +`2` 用法/IO 错误(文件不存在、锚点格式非法、`--db` 传了 +Postgres URL)。输出为单 JSON 对象,便于定时任务解析。 + +脚本的三类本地实测(测试链在 `/tmp`,跑完删除,不进提交): + +| 用例 | 结果 | +|---|---| +| 正常链(2 条)+ 锚点(同链头) | 退出 0,`head_advanced_by=0`;再追加 1 条后重验仍退出 0,`head_advanced_by=1`(头前进属正常) | +| 特权篡改(删触发器后改第 1 条 detail;注:正常 SQL 的 UPDATE 本就被 SQLite 触发器挡下,实测先确认触发器拦截,再模拟特权篡改) | 退出 1,`failures=["entry_hash mismatch at id 1 (record tampered)"]` | +| 删尾巴(删触发器后删最后 1 条)后与旧锚点比对 | 退出 1,`failures=["chain shorter than anchor: current count 1 < anchored count 2 ..."]` | +| 附加:导出模式失败时 `--write-anchor` 被拒绝(无文件生成);不存在的 DB 与 Postgres URL 均退出 2 | 均符合 | + +脚本退出码另有单元测试钉住(`cloud/tests/test_ledger_anchor_check.py`, +子进程跑脚本,断言三类用例的退出码与关键失败文案)。 + +### 5.6 真实 PG:本轨不做、只以常量/适配层证据写清 + +- 本机无真实 PG,PG 触发器(`cloud/ledger.py` 的 + `PG_GUARD_SQL` 常量)**未在真实 PG 上验证**,沿用 #241 + 的标注,不在本轨改口称已验证。 +- PG 路径的操作流程(生产 PG 须运维手动执行并验证后才认): + 1. 在生产 PG 上手动执行 `PG_GUARD_SQL`(函数 + 两触发器), + 并实测 UPDATE/DELETE 被拒绝; + 2. 经 `GET /admin/audit-ledger/verify`(全局管理员)确认 + `ok=true`; + 3. 经 `GET /admin/audit-ledger/export` 取导出,用本脚本 + `--export` 模式 + 历史锚点校验; + 4. 通过后按 §5.4 取新锚点存入对象存储。 +- 在 PG 验证完成前,生产 PG 的“已加固”状态一律写为 + “未验证”,不得在 PR 正文或摘要里写成已验证。 + +## 6. deploy.py 口径:为什么不在门禁内、未来是否纳入 + +**判断:deploy.py 不纳入合并/发布门禁,且未来也不应纳入。** + +- **走什么路径**:`deploy/deploy.py` 是运维人员在本机跑的 + CLI:打包/传包(tar 或主机 git 直拉)→ 主机 + `host_deploy.sh` 换 release 目录 → 重启控制面 unit。 + 它部署的是**平台本身**(服务端代码),不是某个任务的 + 合并产物;它没有任务 id、没有调用方提交的门禁证据、 + 不走服务端 HTTP 鉴权(走 SSH deploy key),与 + `POST /tasks/{id}/merge` / 未来 `POST /tasks/{id}/release` + 根本不在同一层。 +- **为什么不在门禁内**:门禁的客体是“任务能否合并/发布”, + deploy.py 的客体是“平台代码能否上线”。把 deploy.py 绑到 + 任务链状态上,会在平台本身出问题、最需要部署修复与回滚 + 的时候反锁部署(回滚 `rollback.sh` 同样在门禁外,理由相同)。 + 紧急修复被门禁挡住,比“部署时没查任务链”危险得多。 +- **未来是否纳入**:不纳入为原则;只加两条不阻塞的运维 + 配合(均不改 deploy.py 行为、不设卡): + 1. 部署完成后,运维手动跑一次 §5.2 的锚点校验(部署不 + 应改链;若校验失败,按 §5.3 处理,与部署成功与否分开判); + 2. 部署说明里写明“部署 ≠ 任务发布”:推新平台版本不等于 + 发布任何任务,任务发布仍走未来发布端点。 +- 若未来有人提议“deploy.py 发布前查链”,须先回答:查哪 + 个任务的链?以谁的身份?回滚时查不查?答不上来就不纳入 + ——本轨已把这三个问题写进 PR 正文,供立项时复核。 + +## 7. 未做项与诚实边界(本轨明确不声称的) + +- 未建发布端点:无路由、无 HTTP 测试、无任务状态变更、 + 无真实发布动作;`release_gate.py` 只是 evaluator + 契约。 +- 未部署、未跑 `deploy.py`、未动生产、主机 env、系统 key、 + sub2api、生产网关、仓库可见性、GitHub App 权限、生产容器。 +- 未在真实 PG 上验证:PG 触发器语法、PG 上的链校验与锚点 + 流程,均只以常量/导出模式写清,须生产 PG 手动执行并验证。 +- 未替用户拍板:发布端点是否立项、锚点对象存储选型与 + 账号、校验周期(推荐每日 + 每次发布前)、告警渠道、 + 锚点保留期下限、合并冻结与否的默认策略——见 §8。 +- 未改 v1.1 契约、分支保护决策、已有 PR 分支与主 checkout。 + +## 8. 用户侧仍需拍板的点(本轨不擅自决定) + +1. 发布端点是否立项:立项则按 §2 实现 `POST /tasks/{id}/release`; + 不立项则 `release_blocked` 继续只随合并响应返回,发布仍无 + 服务端强制——该风险须用户知晓。 +2. 锚点存放:对象存储选型(厂商/桶/账号)与过渡方案是否启用。 +3. 校验周期:推荐每日定时 + 每次发布前;是否接受该周期。 +4. 告警渠道:校验失败通知全局管理员与当值运维的具体渠道。 +5. 锚点保留期:推荐永久保留;若设下限,接受“12 个月每日 + + 每次发布前锚点、双人确认删除”与否。 +6. 合并冻结默认策略:校验失败时合并是“管理员当场决定”还是 + “默认冻结”(本轨推荐前者、代码不自动冻结)。 +7. 发布比合并更严的口径:无链(表不存在)时发布是否也应阻塞 + (本轨沿用 merge 的“无链即无阻塞”,改严须立项时定)。 diff --git a/tools/ledger_anchor_check.py b/tools/ledger_anchor_check.py new file mode 100644 index 0000000..75eb9c9 --- /dev/null +++ b/tools/ledger_anchor_check.py @@ -0,0 +1,264 @@ +#!/usr/bin/env python3 +"""Ledger anchor check: verify a hash chain against an external anchor. + +Read-only by construction, and that is the point: + +- ``--db`` opens SQLite with ``mode=ro`` (read-only URI). The + script never appends, never updates, never deletes, and never + calls ``AuthStore.audit`` -- writing a row would move the chain + head and make the check itself unreproducible (the same rule + the ``GET /admin/audit-ledger/verify`` endpoint follows on its + success path). A check that writes is not a check. +- ``--export`` reads a JSON export (e.g. from + ``GET /admin/audit-ledger/export``) and verifies it in memory. + This is the path for production Postgres: this script does NOT + open a Postgres connection, and it never pretends to. PG + verification is: export via the admin endpoint, then run this + script in ``--export`` mode against the stored anchor. +- ``--write-anchor`` (``--export`` mode only) writes an ANCHOR + FILE (outside the DB), never the chain, and only after the + internal chain verification has passed. Writing an anchor from + a failing chain would launder tampering into a fresh "trusted" + head, so the script refuses (no file written) when verification + fails. With ``--db``, take the anchor from the passing check's + reported head/count via the documented procedure instead. + +What is checked, in order (all failures are reported, exit 1): + +1. Internal integrity: linkage (each ``prev_hash`` equals the + previous ``entry_hash``) + recomputed ``entry_hash`` per row, + via ``cloud.ledger.verify_chain`` / ``verify_export``. This + catches middle tampering, reordering, and middle deletion. +2. Anchor prefix (only when ``--anchor`` is given): the anchor + records ``{head_hash, count}`` taken at an earlier time. The + current chain must still contain that prefix: it must have + at least ``count`` entries, and the entry at position + ``count`` (1-based) must still hash to ``head_hash``. A + shorter chain means the anchored tail was truncated; a + different hash at that position means the chain was replaced + or rewritten before the anchor point. A head that has ADVANCED + past the anchor (more entries, same prefix) is NORMAL and + passes -- anchors are points in history, not a frozen head. + The honest limit, stated not hidden: an anchor cannot detect + deletion of entries added AFTER that anchor was taken (the + chain then still ends exactly at the anchor head). That is + why anchors must be taken frequently: the anchor cadence + bounds the undetectable window, and the next anchor catches + what this one could not. + +Anchor file format (JSON):: + + {"version": 1, "head_hash": "<64 hex>", "count": , + "anchored_at": "", + "source": ""} + +``anchored_at`` and ``source`` are never trusted by the check: +only ``head_hash`` + ``count`` decide pass/fail. + +Exit codes: 0 = pass, 1 = verification failure (failures listed +in the JSON output), 2 = usage / I/O error (missing file, bad +anchor format, Postgres URL passed as ``--db``). Output is one +JSON object on stdout, so a cron job or operator can parse it +without scraping prose. +""" + +import argparse +import json +import os +import re +import sqlite3 +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname( + os.path.abspath(__file__)))) + +from cloud.ledger import verify_chain, verify_export # noqa: E402 + +_HEX64 = re.compile(r"^[0-9a-f]{64}$") + + +def _load_anchor(path): + try: + with open(path, "r", encoding="utf-8") as handle: + data = json.load(handle) + except (OSError, ValueError) as exc: + raise ValueError("cannot read anchor %r: %s" % (path, exc)) + if not isinstance(data, dict): + raise ValueError("anchor %r is not a JSON object" % path) + head = data.get("head_hash") + count = data.get("count") + if not isinstance(head, str) or not _HEX64.match(head): + raise ValueError( + "anchor %r: head_hash must be a 64-char lowercase hex" + " sha256" % path) + if isinstance(count, bool) or not isinstance(count, int) \ + or count < 0: + raise ValueError( + "anchor %r: count must be a non-negative int" % path) + if count == 0: + raise ValueError( + "anchor %r: count 0 anchors nothing (empty chain); " + "take an anchor only after at least one entry exists" + % path) + return {"head_hash": head, "count": count} + + +def _check_prefix(entry_hashes, anchor): + """Anchor-prefix check. Returns a list of failures (empty = pass).""" + count = anchor["count"] + if len(entry_hashes) < count: + return [ + "chain shorter than anchor: current count %d < anchored" + " count %d (anchored tail truncated or chain replaced)" + % (len(entry_hashes), count)] + if entry_hashes[count - 1] != anchor["head_hash"]: + return [ + "anchor prefix mismatch: entry %d hashes to %s, not the" + " anchored head %s (chain replaced or rewritten before" + " the anchor point)" + % (count, entry_hashes[count - 1], anchor["head_hash"])] + return [] + + +def _report(ok, failures, count, head_hash, anchor): + report = { + "ok": ok, + "failures": failures, + "count": count, + "head_hash": head_hash, + "anchor": anchor, + "head_advanced_by": ( + count - anchor["count"] if anchor is not None else None), + } + print(json.dumps(report, indent=2, sort_keys=True)) + return 0 if ok else 1 + + +def _run_db(db_path, anchor): + if "://" in db_path or db_path.startswith("postgres"): + raise ValueError( + "--db takes a SQLite file path, not a Postgres URL. " + "For Postgres, export via GET /admin/audit-ledger/export" + " and run this script with --export.") + if not os.path.exists(db_path): + raise ValueError("db file not found: %r" % db_path) + uri = "file:%s?mode=ro" % _quote(db_path) + conn = sqlite3.connect(uri, uri=True) + try: + result = verify_chain(conn) + rows = conn.execute( + "SELECT entry_hash FROM audit_ledger ORDER BY id ASC" + ).fetchall() if result["count"] else [] + finally: + conn.close() + failures = list(result.get("failures") or []) + entry_hashes = [row[0] for row in rows] + if anchor is not None: + failures.extend(_check_prefix(entry_hashes, anchor)) + return _report(not failures, failures, result["count"], + result["head_hash"], anchor) + + +def _quote(path): + # Minimal URI escaping for the read-only open: '#' and '?' + # would otherwise terminate the URI. A path containing them + # is unusual; escaping keeps the open honest instead of + # silently opening a different file. + return path.replace("?", "%3F").replace("#", "%23") + + +def _run_export(export_path, anchor): + try: + with open(export_path, "r", encoding="utf-8") as handle: + data = json.load(handle) + except (OSError, ValueError) as exc: + raise ValueError( + "cannot read export %r: %s" % (export_path, exc)) + result = verify_export(data) + entries = data.get("entries") if isinstance(data, dict) else None + entry_hashes = [] + if isinstance(entries, list): + entry_hashes = [e.get("entry_hash") for e in entries + if isinstance(e, dict)] + failures = list(result.get("failures") or []) + failures.extend(_check_prefix(entry_hashes, anchor) + if anchor is not None else []) + return _report(not failures, failures, result["count"], + result["head_hash"], anchor) + + +def _write_anchor(out_path, head_hash, count, source): + import datetime + payload = { + "version": 1, + "head_hash": head_hash, + "count": count, + "anchored_at": datetime.datetime.now( + datetime.timezone.utc).isoformat(), + "source": source, + } + tmp_path = out_path + ".tmp" + with open(tmp_path, "w", encoding="utf-8") as handle: + json.dump(payload, handle, indent=2, sort_keys=True) + handle.write("\n") + os.replace(tmp_path, out_path) + print(json.dumps({"anchor_written": out_path, **payload}, + indent=2, sort_keys=True)) + + +def main(argv=None): + parser = argparse.ArgumentParser( + description="Verify an audit hash chain against an external" + " anchor (read-only; exit 0 pass, 1 fail, 2 usage).") + group = parser.add_mutually_exclusive_group(required=True) + group.add_argument("--db", help="SQLite control.db file path") + group.add_argument("--export", + help="JSON file from /admin/audit-ledger/export") + parser.add_argument("--anchor", + help="anchor JSON file ({head_hash, count})") + parser.add_argument("--write-anchor", + help="after a passing check, write a fresh" + " anchor JSON to this path (anchor file" + " only, never the chain; refused on failure)") + args = parser.parse_args(argv) + try: + anchor = _load_anchor(args.anchor) if args.anchor else None + except ValueError as exc: + print("error: %s" % exc, file=sys.stderr) + return 2 + if args.db and args.write_anchor: + # Refused before any check runs, so a passing report is + # never followed by a contradictory exit code. Anchor + # taking from a DB check uses the documented procedure: + # run the check, then write the reported head/count as a + # new anchor file (the file, never the chain). + print("error: --write-anchor is not combined with --db;" + " take the anchor from the passing check's" + " head_hash/count via the documented procedure", + file=sys.stderr) + return 2 + try: + if args.db: + return _run_db(args.db, anchor) + code = _run_export(args.export, anchor) + if code == 0 and args.write_anchor: + # Re-read the export to obtain the verified head. + with open(args.export, "r", encoding="utf-8") as handle: + data = json.load(handle) + _write_anchor(args.write_anchor, data["head_hash"], + len(data["entries"]), args.export) + elif args.write_anchor: + print("error: refusing --write-anchor: verification" + " failed, an anchor from a failing chain would" + " launder tampering", file=sys.stderr) + return code + except ValueError as exc: + print("error: %s" % exc, file=sys.stderr) + return 2 + except sqlite3.Error as exc: + print("error: sqlite: %s" % exc, file=sys.stderr) + return 2 + + +if __name__ == "__main__": + sys.exit(main())