release: gate-check contract for future release endpoint + ledger anchor ops (prep, no endpoint, no deploy)

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()