feat(P03): evidence_writer.py (T-3.7)

---ci--- phase: 3 milestone: v1.0 status: execute persona: backend-engineer task: T-3.7 requirements: covered: [REQ-11] ---/ci--- Wave 2, task T-3.7. Hash-chained audit event writer (D-023/D-005) with auto-genesis and atomic writes.
2026-07-21 13:24:44 +00:00
parent 2be16f72b9
commit 4c37ad1d03
1 changed files with 123 additions and 0 deletions
@@ -0,0 +1,123 @@
+#!/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())