diff --git a/src/robotproof_studio/audit.py b/src/robotproof_studio/audit.py new file mode 100644 index 0000000..b1b0525 --- /dev/null +++ b/src/robotproof_studio/audit.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Dict, List + +from .contracts import PlanDelta + + +def _now_iso() -> str: + return datetime.utcnow().isoformat() + "Z" + + +@dataclass +class AuditEntry: + index: int + timestamp: str + delta_id: str + payload_hash: str + prev_hash: str + entry_hash: str + + +class AuditLog: + """Simple tamper-evident audit log implemented as a hash chain. + + Each entry includes the previous entry hash so that any modification + is detectable by re-checking the chain. + """ + + def __init__(self) -> None: + self.entries: List[AuditEntry] = [] + + @staticmethod + def _hash(obj: Any) -> str: + s = json.dumps(obj, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(s.encode("utf-8")).hexdigest() + + def append(self, delta: PlanDelta) -> AuditEntry: + # hash the important parts of the delta (id, timestamp, data, proof) + payload = {"delta_id": delta.delta_id, "timestamp": delta.timestamp, "data": delta.data, "proof": delta.proof} + payload_hash = self._hash(payload) + + prev_hash = self.entries[-1].entry_hash if self.entries else "" + + entry_obj = {"index": len(self.entries), "timestamp": _now_iso(), "delta_id": delta.delta_id, "payload_hash": payload_hash, "prev_hash": prev_hash} + entry_hash = self._hash(entry_obj) + + entry = AuditEntry(index=entry_obj["index"], timestamp=entry_obj["timestamp"], delta_id=entry_obj["delta_id"], payload_hash=payload_hash, prev_hash=prev_hash, entry_hash=entry_hash) + self.entries.append(entry) + return entry + + def verify_chain(self) -> bool: + prev_hash = "" + for i, e in enumerate(self.entries): + # recompute payload hash is not possible here, but we can recompute the entry hash from stored fields + entry_obj = {"index": e.index, "timestamp": e.timestamp, "delta_id": e.delta_id, "payload_hash": e.payload_hash, "prev_hash": prev_hash} + expected_hash = self._hash(entry_obj) + if expected_hash != e.entry_hash: + return False + prev_hash = e.entry_hash + return True diff --git a/tests/test_audit.py b/tests/test_audit.py new file mode 100644 index 0000000..6da7503 --- /dev/null +++ b/tests/test_audit.py @@ -0,0 +1,27 @@ +from datetime import datetime + +from robotproof_studio.contracts import PlanDelta +from robotproof_studio.audit import AuditLog + + +def make_delta(id_suffix: str) -> PlanDelta: + return PlanDelta(delta_id=f"d-{id_suffix}", timestamp=datetime.utcnow().isoformat() + "Z", data={"val": id_suffix}) + + +def test_audit_log_chain_and_detection(): + log = AuditLog() + + d1 = make_delta("1") + d1.sign("tester") + e1 = log.append(d1) + + d2 = make_delta("2") + d2.sign("tester") + e2 = log.append(d2) + + # chain should be valid + assert log.verify_chain() is True + + # tamper with second entry's delta_id (simulates external tampering) + log.entries[1].delta_id = "tampered" + assert log.verify_chain() is False