acdl/demo/scripts/evidence_writer.py

#!/usr/bin/env python3
"""evidence_writer.py — REQ-11 / D-023 / D-005

Appends a hash-chained event to audit.json.

Each event: {"seq": N, "ts": <iso8601 UTC>, "stage": "...", "event": "...",
             "prev_hash": "<sha256 or GENESIS>", "hash": "<sha256 of canonical json of this event with hash empty>"}

Hash chain (D-023):
  1. Build event dict with hash = "" (empty string).
  2. canonical = json.dumps(event, sort_keys=True, separators=(",", ":"))
  3. hash = sha256(canonical.encode("utf-8")).hexdigest()
  4. event["hash"] = hash
  5. append to audit.json

Auto-genesis: if audit.json is empty/missing and --stage is not "genesis",
a genesis event (seq 0, prev_hash "GENESIS") is inserted first.

Input:
  --stage <dev|qa|prod|finalize|genesis>  (required)
  --event "<text>"                        (required)
  --audit <path>                          (optional, default ./audit.json)
Output: stdout {"seq": N, "hash": "..."}
Exit:   0 on success, 1 on I/O error.
"""
import argparse
import datetime
import hashlib
import json
import os
import sys

GENESIS_EVENT_TEXT = "audit log initialized"


def now_iso8601_utc() -> str:
    return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def compute_hash(event: dict) -> str:
    """Compute the sha256 hash of an event using canonical JSON (D-023)."""
    tmp = dict(event)
    tmp["hash"] = ""
    canonical = json.dumps(tmp, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def make_event(seq: int, stage: str, event_text: str, prev_hash: str) -> dict:
    event = {
        "seq": seq,
        "ts": now_iso8601_utc(),
        "stage": stage,
        "event": event_text,
        "prev_hash": prev_hash,
        "hash": "",
    }
    event["hash"] = compute_hash(event)
    return event


def load_audit(audit_path: str) -> list:
    if not os.path.exists(audit_path):
        return []
    try:
        with open(audit_path, "r", encoding="utf-8") as fh:
            data = json.load(fh)
    except (json.JSONDecodeError, ValueError):
        return []
    if not isinstance(data, list):
        return []
    return data


def atomic_write(audit_path: str, data: list) -> None:
    tmp_path = audit_path + ".tmp"
    with open(tmp_path, "w", encoding="utf-8") as fh:
        json.dump(data, fh, indent=2)
        fh.write("\n")
    os.replace(tmp_path, audit_path)


def main() -> int:
    parser = argparse.ArgumentParser(description="Append a hash-chained event to audit.json")
    parser.add_argument("--stage", required=True,
                        choices=["dev", "qa", "prod", "finalize", "genesis"])
    parser.add_argument("--event", required=True)
    parser.add_argument("--audit", default="./audit.json")
    args = parser.parse_args()

    events = load_audit(args.audit)

    # Auto-genesis: if the log is empty and the caller did not ask for a
    # genesis event, seed one first.
    if len(events) == 0 and args.stage != "genesis":
        genesis = make_event(seq=0, stage="genesis", event_text=GENESIS_EVENT_TEXT,
                             prev_hash="GENESIS")
        events.append(genesis)

    # Determine the new seq + prev_hash.
    if events:
        last = events[-1]
        seq = last["seq"] + 1
        prev_hash = last["hash"]
    else:
        seq = 0
        prev_hash = "GENESIS"

    new_event = make_event(seq=seq, stage=args.stage, event_text=args.event,
                           prev_hash=prev_hash)
    events.append(new_event)

    try:
        atomic_write(args.audit, events)
    except OSError as exc:
        print(f"evidence_writer: I/O error: {exc}", file=sys.stderr)
        return 1

    print(json.dumps({"seq": new_event["seq"], "hash": new_event["hash"]}))
    return 0


if __name__ == "__main__":
    sys.exit(main())