diff --git a/scripts/evidence_writer.py b/scripts/evidence_writer.py new file mode 100755 index 0000000..09472bc --- /dev/null +++ b/scripts/evidence_writer.py @@ -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": , "stage": "...", "event": "...", + "prev_hash": "", "hash": ""} + +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 (required) + --event "" (required) + --audit (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()) \ No newline at end of file