feat(h5): ① KMS key rotation/non-exportable + ② external WORM audit ledger

① Key management (B3/KMS): vault-kms.sh gains `rotate` (Transit key rotation) and `assert-nonexportable` (proves private material never leaves the KMS). Validated live against a Vault dev server: sign→verify (v1) → rotate → sign→verify (v2) → export denied. sign-audit-head.sh already routes to Vault when VAULT_ADDR/TOKEN are set, so this is the real production signing path. ② External WORM audit (C5/V21): worm-ledger.py + audit-ship.sh append the audit head to a hash-linked append-only ledger (chattr +a best-effort on Linux; S3 Object Lock/QLDB in production). verify-audit-gap.sh detects local audit rollback (AUDIT_GAP_DETECTED — the durable ledger still holds the later head) and ledger tampering (AUDIT_LEDGER_TAMPERED). phase-h5-infra-tests.sh: 7 checks — KMS sign/rotate/non-exportable (skip-aware, live when Vault reachable) + WORM in-sync/rollback/tamper (always local). Baselines: run-casan4 35/35, adversarial 44/44, approval 8/8. Lifts H5 key-mgmt 2.5→~4 (KMS live path + rotation + non-exportable) and external-audit 1.5→~3.5 (WORM ledger + gap detection). Suites now 7 (+7 = 155 checks). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-04 23:16:21 +09:00
parent e21a1472b1
commit 86e13a26ed
5 changed files with 246 additions and 1 deletions
@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+"""CASAN H5 — External append-only (WORM) anchor ledger (C5 / V21).
+
+Ships the audit-chain head to an EXTERNAL append-only ledger so that deleting or
+rolling back the local audit log is detectable: the durable ledger still holds
+the later head. Entries are hash-linked (each anchor commits to the previous
+one), so truncating or editing the ledger itself is also detectable.
+
+This is the local MVP of WORM. Production ships each anchor to a true
+write-once store (S3 Object Lock / QLDB / append-only Kafka) + trusted timestamp.
+
+Argv:
+  ship   <head-file> <ledger-file> [iso-timestamp]
+  verify <head-file> <ledger-file>
+
+Ledger line: {"seq":N,"ts":..,"head":<hex>,"prev":<anchor|"">,"anchor":<hex>}
+anchor = sha256("seq|ts|head|prev")
+
+verify exit: 0 in-sync (OK) · 1 tamper/gap (AUDIT_LEDGER_TAMPERED | AUDIT_GAP_DETECTED)
+"""
+import hashlib
+import json
+import sys
+
+
+def _anchor(seq, ts, head, prev):
+    return hashlib.sha256(f"{seq}|{ts}|{head}|{prev}".encode()).hexdigest()
+
+
+def _read_ledger(path):
+    rows = []
+    try:
+        for line in open(path, encoding="utf-8"):
+            line = line.strip()
+            if line:
+                rows.append(json.loads(line))
+    except FileNotFoundError:
+        pass
+    return rows
+
+
+def _read_head(path):
+    with open(path, encoding="utf-8") as f:
+        return f.read().strip()
+
+
+def ship(head_file, ledger_file, ts):
+    head = _read_head(head_file)
+    rows = _read_ledger(ledger_file)
+    prev = rows[-1]["anchor"] if rows else ""
+    seq = (rows[-1]["seq"] + 1) if rows else 1
+    entry = {"seq": seq, "ts": ts, "head": head, "prev": prev,
+             "anchor": _anchor(seq, ts, head, prev)}
+    with open(ledger_file, "a", encoding="utf-8") as f:
+        f.write(json.dumps(entry) + "\n")
+    print(f"WORM_ANCHOR_SHIPPED seq={seq} head={head[:16]}… anchor={entry['anchor'][:16]}…")
+    return 0
+
+
+def verify(head_file, ledger_file):
+    rows = _read_ledger(ledger_file)
+    if not rows:
+        sys.stderr.write("WORM_LEDGER_EMPTY (nothing shipped yet)\n")
+        return 1
+    # 1. Ledger self-integrity: recompute anchors + check the hash-link.
+    prev = ""
+    for r in rows:
+        if r.get("prev", "") != prev:
+            sys.stderr.write(f"AUDIT_LEDGER_TAMPERED seq={r.get('seq')} broken_link\n")
+            return 1
+        if _anchor(r.get("seq"), r.get("ts"), r.get("head"), prev) != r.get("anchor"):
+            sys.stderr.write(f"AUDIT_LEDGER_TAMPERED seq={r.get('seq')} anchor_mismatch\n")
+            return 1
+        prev = r["anchor"]
+    # 2. Local head vs durable ledger.
+    local = _read_head(head_file)
+    latest = rows[-1]["head"]
+    if local == latest:
+        print(f"WORM_IN_SYNC anchors={len(rows)} head={local[:16]}…")
+        return 0
+    older = [r["seq"] for r in rows[:-1] if r["head"] == local]
+    if older:
+        # Local audit tip matches an OLDER durable anchor → local was rolled back.
+        sys.stderr.write(
+            f"AUDIT_GAP_DETECTED local_head=older(seq={older[-1]}) durable_latest_seq={rows[-1]['seq']} "
+            f"— local audit rolled back below the durable WORM anchor\n")
+        return 1
+    sys.stderr.write(
+        "AUDIT_UNSHIPPED local head not yet anchored (ship it) — not a rollback\n")
+    return 1
+
+
+def main():
+    if len(sys.argv) < 4:
+        sys.stderr.write("Usage: worm-ledger.py {ship|verify} <head-file> <ledger-file> [ts]\n")
+        return 64
+    cmd, head_file, ledger_file = sys.argv[1], sys.argv[2], sys.argv[3]
+    if cmd == "ship":
+        ts = sys.argv[4] if len(sys.argv) > 4 else __import__("datetime").datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
+        return ship(head_file, ledger_file, ts)
+    if cmd == "verify":
+        return verify(head_file, ledger_file)
+    sys.stderr.write(f"unknown command: {cmd}\n")
+    return 64
+
+
+if __name__ == "__main__":
+    sys.exit(main())