File size: 4,997 Bytes
9abace2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """
Layer 0: WORM Evidence Ledger
Part of SOVEREIGN PYTHON LLM ENGINE
Append-only binary ledger backed by storage.WORMFile (struct-packed records).
No JSON, no text, no injection surface.
"""
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterator
from .crypto import ContentHash, Signature, SigningKey, hash_content, sign_artifact
from .storage import WORMFile, WORMRecord
# ─────────────────────────────────────────────
# Evidence Record (public API type)
# ─────────────────────────────────────────────
@dataclass(frozen=True)
class EvidenceRecord:
timestamp_ns: int
event_type: str
content_hash: bytes
signature: bytes
payload: bytes
metadata: bytes
prev_hash: bytes
@property
def timestamp(self) -> float:
return self.timestamp_ns / 1_000_000_000
def _record_from_worm(w: WORMRecord) -> EvidenceRecord:
return EvidenceRecord(
timestamp_ns=w.timestamp_ns,
event_type=w.event_type,
content_hash=w.content_hash,
signature=w.signature,
payload=w.payload,
metadata=w.metadata,
prev_hash=w.prev_hash
)
# ─────────────────────────────────────────────
# WORM Ledger
# ─────────────────────────────────────────────
class WORMLedger:
"""
Write-Once-Read-Many append-only ledger.
Backed by binary struct-packed records (WORMFile).
No JSONL, no text encoding, no injection surface.
Public API is unchanged from the JSONL version —
callers pass raw bytes for payload, raw bytes for metadata.
"""
def __init__(self, ledger_path: Path, signing_key: SigningKey | None = None):
self._file = WORMFile(ledger_path, signing_key=signing_key)
self.ledger_path = ledger_path
self.signing_key = signing_key
def append(
self,
event_type: str,
data: bytes | dict | str | None = None,
metadata: bytes | dict | str | None = None
) -> EvidenceRecord:
"""
Append evidence record.
data and metadata accept:
bytes — written as-is
dict — encoded as UTF-8 key=value pairs (no JSON)
str — encoded as UTF-8
None — empty bytes
"""
payload = self._coerce(data)
meta = self._coerce(metadata)
rec = self._file.append(event_type, payload, meta)
return _record_from_worm(rec)
def read_all(self) -> list[EvidenceRecord]:
return [_record_from_worm(r) for r in self._file.scan()]
def scan(self) -> Iterator[EvidenceRecord]:
for r in self._file.scan():
yield _record_from_worm(r)
def verify_chain(self) -> bool:
valid, _ = self._file.verify_chain()
return valid
def get_records_by_type(self, event_type: str) -> list[EvidenceRecord]:
return [_record_from_worm(r) for r in self._file.scan_by_type(event_type)]
def count(self) -> int:
return self._file.count()
@staticmethod
def _coerce(value: bytes | dict | str | None) -> bytes:
if value is None:
return b''
if isinstance(value, bytes):
return value
if isinstance(value, str):
return value.encode('utf-8')
if isinstance(value, dict):
# Flat key=value encoding — no JSON, no injection
parts = []
for k, v in value.items():
k_s = str(k).replace('=', '_').replace('\n', '_')
v_s = str(v).replace('\n', '_')
parts.append(f"{k_s}={v_s}")
return '\n'.join(parts).encode('utf-8')
return str(value).encode('utf-8')
# ─────────────────────────────────────────────
# Stats helper
# ─────────────────────────────────────────────
def ledger_stats(ledger: WORMLedger) -> dict[str, Any]:
records = ledger.read_all()
event_counts: dict[str, int] = {}
for r in records:
event_counts[r.event_type] = event_counts.get(r.event_type, 0) + 1
return {
'total_records': len(records),
'event_counts': event_counts,
'first_timestamp': records[0].timestamp if records else None,
'last_timestamp': records[-1].timestamp if records else None,
'chain_valid': ledger.verify_chain()
}
|