rune/shared/audit_ledger.py

#!/usr/bin/env python3
"""audit_ledger.py — HMAC-chained tamper-evident promotion ledger.

Every promotion decision is recorded in a hash-chained ledger.
Tampering with any entry breaks the chain.

Research basis:
- PoEM (arXiv:2608.16032): HMAC-chained tamper-evident execution ledger
- Autogenesis (arXiv:2604.15034): versioned interfaces, auditable lineage

Run under ~/.venvs/memory/bin/python3.
"""

import hashlib
import hmac
import json
import os
import sys
from datetime import datetime, timezone
from typing import Dict, List, Optional

LEDGER_FILE = os.path.expanduser("~/.config/opencode/memory/promotion_ledger.jsonl")
SECRET_KEY = os.environ.get("AUDIT_LEDGER_KEY", "rune-drift-integrity-2024")


def _compute_hash(prev_hash: str, entry: dict) -> str:
    """Compute HMAC chain hash for a ledger entry."""
    payload = json.dumps(entry, sort_keys=True).encode()
    msg = (prev_hash + payload.decode()).encode()
    return hmac.new(SECRET_KEY.encode(), msg, hashlib.sha256).hexdigest()


def append_entry(entry_type: str, entry_id: int, details: dict) -> dict:
    """Append an entry to the promotion ledger.

    Returns the ledger entry with chain hash.
    """
    # Read last hash
    prev_hash = "genesis"
    if os.path.exists(LEDGER_FILE):
        with open(LEDGER_FILE, "r") as f:
            lines = f.readlines()
            if lines:
                try:
                    last = json.loads(lines[-1])
                    prev_hash = last.get("hash", "genesis")
                except json.JSONDecodeError:
                    prev_hash = "genesis"

    entry = {
        "type": entry_type,
        "entry_id": entry_id,
        "details": details,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "prev_hash": prev_hash,
    }

    # Compute chain hash
    entry["hash"] = _compute_hash(prev_hash, entry)

    # Append to ledger
    os.makedirs(os.path.dirname(LEDGER_FILE), exist_ok=True)
    with open(LEDGER_FILE, "a") as f:
        f.write(json.dumps(entry) + "\n")

    return entry


def verify_chain() -> dict:
    """Verify the integrity of the entire ledger chain.

    Returns dict with verification result.
    """
    if not os.path.exists(LEDGER_FILE):
        return {"valid": True, "entries": 0, "reason": "empty ledger"}

    with open(LEDGER_FILE, "r") as f:
        lines = f.readlines()

    prev_hash = "genesis"
    for i, line in enumerate(lines):
        try:
            entry = json.loads(line)
        except json.JSONDecodeError:
            return {"valid": False, "entries": i, "reason": f"invalid JSON at line {i+1}"}

        # Verify prev_hash chain
        if entry.get("prev_hash") != prev_hash:
            return {
                "valid": False,
                "entries": i,
                "reason": f"chain broken at line {i+1}: expected prev_hash={prev_hash}, got {entry.get('prev_hash')}"
            }

        # Verify HMAC
        stored_hash = entry.get("hash")
        computed = _compute_hash(prev_hash, {k: v for k, v in entry.items() if k != "hash"})
        if stored_hash != computed:
            return {
                "valid": False,
                "entries": i,
                "reason": f"HMAC mismatch at line {i+1}: stored={stored_hash[:16]}..., computed={computed[:16]}..."
            }

        prev_hash = stored_hash

    return {"valid": True, "entries": len(lines), "reason": "chain intact"}


def get_entries(limit: int = 50) -> List[dict]:
    """Get recent ledger entries."""
    if not os.path.exists(LEDGER_FILE):
        return []

    with open(LEDGER_FILE, "r") as f:
        lines = f.readlines()

    entries = []
    for line in lines[-limit:]:
        try:
            entries.append(json.loads(line))
        except json.JSONDecodeError:
            continue
    return entries


def log_promotion(entry_id: int, old_status: str, new_status: str, wash_count: int, reason: str) -> dict:
    """Log a promotion decision."""
    return append_entry("promotion", entry_id, {
        "old_status": old_status,
        "new_status": new_status,
        "wash_count": wash_count,
        "reason": reason,
    })


def log_gate_result(entry_id: int, verdict: str, critics: dict) -> dict:
    """Log a pre-commit gate result."""
    return append_entry("gate", entry_id, {
        "verdict": verdict,
        "critics": critics,
    })


def log_drift_alert(alert_type: str, message: str, severity: str) -> dict:
    """Log a drift alert."""
    return append_entry("alert", 0, {
        "alert_type": alert_type,
        "message": message,
        "severity": severity,
    })


# ═══════════════════════════════════════════════════════════════════════════════
# CLI
# ═══════════════════════════════════════════════════════════════════════════════

def main():
    import argparse
    parser = argparse.ArgumentParser(description="Promotion audit ledger")
    sub = parser.add_subparsers(dest="cmd")

    # verify
    sub.add_parser("verify", help="Verify ledger chain integrity")

    # list [--limit N]
    p_list = sub.add_parser("list", help="Show recent ledger entries")
    p_list.add_argument("--limit", type=int, default=20)

    # log-promotion <entry_id> <old> <new> <wash_count> <reason>
    p_log = sub.add_parser("log-promotion", help="Log a promotion")
    p_log.add_argument("entry_id", type=int)
    p_log.add_argument("old_status")
    p_log.add_argument("new_status")
    p_log.add_argument("wash_count", type=int)
    p_log.add_argument("reason")

    args = parser.parse_args()

    if not args.cmd:
        parser.print_help()
        sys.exit(1)

    if args.cmd == "verify":
        result = verify_chain()
        print(f"Chain valid: {result['valid']}")
        print(f"Entries: {result['entries']}")
        print(f"Reason: {result['reason']}")

    elif args.cmd == "list":
        entries = get_entries(args.limit)
        if entries:
            print(f"Recent ledger entries ({len(entries)}):")
            for e in entries:
                print(f"  [{e.get('type')}] entry_id={e.get('entry_id')} @ {e.get('timestamp', '')[:19]}")
                details = e.get('details', {})
                if e.get('type') == 'promotion':
                    print(f"    {details.get('old_status')} → {details.get('new_status')} (wash={details.get('wash_count')})")
                elif e.get('type') == 'gate':
                    print(f"    verdict={details.get('verdict')}")
        else:
            print("No ledger entries.")

    elif args.cmd == "log-promotion":
        entry = log_promotion(args.entry_id, args.old_status, args.new_status, args.wash_count, args.reason)
        print(f"Logged: {entry['hash'][:16]}...")


if __name__ == "__main__":
    main()