136 lines
5.0 KiB
Python
136 lines
5.0 KiB
Python
"""Minimal sidecar runtime helpers.
|
|
|
|
Provides:
|
|
- AuditLogStore: SQLite-backed append-only log with chained hashes (Merkle-like linear chain).
|
|
- merge_plan_deltas: simple CRDT-style merge for PlanDelta patches.
|
|
|
|
This is intentionally small and dependency-free so it can run on edge devices.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import hashlib
|
|
import json
|
|
from typing import Dict, Any, List, Optional
|
|
|
|
|
|
def _hash_bytes(b: bytes) -> str:
|
|
return hashlib.sha256(b).hexdigest()
|
|
|
|
|
|
class AuditLogStore:
|
|
"""Append-only SQLite-backed audit log with chained hashes.
|
|
|
|
Each row contains (id INTEGER PRIMARY KEY AUTOINCREMENT, entry_json TEXT, prev_hash TEXT, hash TEXT)
|
|
The hash is sha256(prev_hash || entry_json) so the chain can be verified deterministically.
|
|
"""
|
|
|
|
def __init__(self, db_path: str = ":memory:"):
|
|
self.db_path = db_path
|
|
self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
|
|
self._init_db()
|
|
|
|
def _init_db(self) -> None:
|
|
cur = self.conn.cursor()
|
|
cur.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS audit_log (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
entry_json TEXT NOT NULL,
|
|
prev_hash TEXT,
|
|
hash TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
self.conn.commit()
|
|
|
|
def append(self, actor: str, action: str, metadata: Optional[Dict[str, Any]] = None) -> int:
|
|
entry = {"actor": actor, "action": action, "metadata": metadata or {}}
|
|
entry_json = json.dumps(entry, sort_keys=True, separators=(",", ":"))
|
|
prev_hash = self._get_latest_hash()
|
|
h = _hash_bytes((prev_hash or "").encode() + entry_json.encode())
|
|
cur = self.conn.cursor()
|
|
cur.execute(
|
|
"INSERT INTO audit_log (entry_json, prev_hash, hash) VALUES (?, ?, ?)",
|
|
(entry_json, prev_hash, h),
|
|
)
|
|
self.conn.commit()
|
|
return cur.lastrowid
|
|
|
|
def _get_latest_hash(self) -> Optional[str]:
|
|
cur = self.conn.cursor()
|
|
cur.execute("SELECT hash FROM audit_log ORDER BY id DESC LIMIT 1")
|
|
row = cur.fetchone()
|
|
return row[0] if row is not None else None
|
|
|
|
def iter_entries(self) -> List[Dict[str, Any]]:
|
|
cur = self.conn.cursor()
|
|
cur.execute("SELECT id, entry_json, prev_hash, hash FROM audit_log ORDER BY id ASC")
|
|
out = []
|
|
for row in cur.fetchall():
|
|
out.append({"id": row[0], "entry": json.loads(row[1]), "prev_hash": row[2], "hash": row[3]})
|
|
return out
|
|
|
|
def verify_chain(self) -> bool:
|
|
"""Verify the integrity of the chain.
|
|
|
|
Returns True if every row's hash equals sha256(prev_hash || entry_json) and linkage is intact.
|
|
"""
|
|
cur = self.conn.cursor()
|
|
cur.execute("SELECT entry_json, prev_hash, hash FROM audit_log ORDER BY id ASC")
|
|
prev = None
|
|
for entry_json, prev_hash, h in cur.fetchall():
|
|
expected = _hash_bytes((prev_hash or "").encode() + entry_json.encode())
|
|
if expected != h:
|
|
return False
|
|
# Also check that prev_hash matches previous row's hash
|
|
if prev is not None and prev != (prev_hash or None):
|
|
return False
|
|
prev = h
|
|
return True
|
|
|
|
|
|
def merge_plan_deltas(deltas: List[Dict[str, Any]]) -> Dict[str, Any]:
|
|
"""Merge a list of PlanDelta-like dicts into a single combined PlanDelta.
|
|
|
|
Strategy (simple CRDT-style):
|
|
- Each delta has an `id`, optional `base_plan_id`, and `patches` list where each patch is a dict
|
|
that must include an `op_id` unique identifier.
|
|
- We combine all patches and deduplicate by `op_id`, ordering by delta timestamp if present.
|
|
- This is intentionally simple; real-world use should use signed ops and version vectors.
|
|
"""
|
|
# Sort deltas by timestamp if provided, otherwise by id to ensure deterministic order
|
|
def _sort_key(d):
|
|
return (d.get("timestamp") or "", d.get("id") or "")
|
|
|
|
sorted_d = sorted(deltas, key=_sort_key)
|
|
seen = set()
|
|
combined_patches = []
|
|
for d in sorted_d:
|
|
for p in d.get("patches", []):
|
|
op_id = p.get("op_id") or json.dumps(p, sort_keys=True)
|
|
if op_id in seen:
|
|
continue
|
|
seen.add(op_id)
|
|
combined_patches.append(p)
|
|
|
|
merged = {"id": "merged-" + _hash_bytes(json.dumps([d.get("id") for d in sorted_d]).encode()),
|
|
"base_plan_id": None,
|
|
"patches": combined_patches}
|
|
return merged
|
|
|
|
|
|
def emit_rollback_event(store: AuditLogStore, plan_cert: "object", reason: str) -> int:
|
|
"""Emit a watchdog/rollback event into the audit log.
|
|
|
|
This is a small helper used by runtimes/watchdogs to persist an operator/verifier
|
|
initiated rollback. The function records the plan_id and any rollback_checkpoint
|
|
present on the PlanCert.
|
|
"""
|
|
metadata = {
|
|
"plan_id": getattr(plan_cert, "plan_id", None),
|
|
"rollback_checkpoint": getattr(plan_cert, "rollback_checkpoint", None),
|
|
"reason": reason,
|
|
}
|
|
return store.append("watchdog", "rollback", metadata)
|