build(agent): weasel-1#856f80 iteration
This commit is contained in:
parent
6aa5f058c0
commit
a31e6f1a38
|
|
@ -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
|
||||||
|
|
@ -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
|
||||||
Loading…
Reference in New Issue