grimacing
3ea36ef3ab
Squash merge of phase/03-l2-modules-and-core-scripts; 4 L2s + 5 core scripts; verify_phase03.sh green.
123 lines
3.7 KiB
Python
Executable File
123 lines
3.7 KiB
Python
Executable File
#!/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()) |