sovereign_audit_chain.py

# sovereign_audit_chain.py
from __future__ import annotations

import json
import hashlib
import os
from typing import Dict, Any, List


AUDIT_CHAIN_FILE = "sovereign_audit_chain.jsonl"


def _hash(data: str) -> str:
    return hashlib.sha256(data.encode("utf-8")).hexdigest()


def _serialize(obj: Dict[str, Any]) -> str:
    return json.dumps(obj, sort_keys=True, ensure_ascii=False)


def _load_chain() -> List[Dict[str, Any]]:
    if not os.path.exists(AUDIT_CHAIN_FILE):
        return []
    with open(AUDIT_CHAIN_FILE, "r", encoding="utf-8") as f:
        return [json.loads(line) for line in f if line.strip()]


def _get_last_hash(chain: List[Dict[str, Any]]) -> str:
    if not chain:
        return "GENESIS"
    return chain[-1]["hash"]


def append_audit_event(event: Dict[str, Any]) -> Dict[str, Any]:
    chain = _load_chain()

    prev_hash = _get_last_hash(chain)

    record = {
        "event": event,
        "prev_hash": prev_hash,
    }

    record_str = _serialize(record)
    current_hash = _hash(record_str)

    record["hash"] = current_hash

    with open(AUDIT_CHAIN_FILE, "a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

    return {
        "ok": True,
        "hash": current_hash,
        "prev_hash": prev_hash,
    }


def verify_chain() -> Dict[str, Any]:
    chain = _load_chain()

    if not chain:
        return {"ok": True, "message": "empty_chain"}

    prev_hash = "GENESIS"

    for idx, record in enumerate(chain):
        expected_prev = record.get("prev_hash")
        if expected_prev != prev_hash:
            return {
                "ok": False,
                "error": "broken_chain",
                "index": idx,
            }

        record_copy = {
            "event": record["event"],
            "prev_hash": record["prev_hash"],
        }

        recalculated = _hash(_serialize(record_copy))

        if recalculated != record.get("hash"):
            return {
                "ok": False,
                "error": "tampered_record",
                "index": idx,
            }

        prev_hash = record["hash"]

    return {
        "ok": True,
        "length": len(chain),
        "status": "valid",
    }