File size: 4,748 Bytes
ecf9cb9 | 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 | """
audit/worm.py
Append-only WORM audit chain. Each record is chained via SHA-256.
The Semantic Hash from the NAND gate is the immutable identity anchor.
"""
import hashlib
import json
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class AuditRecord:
timestamp: str
semantic_hash: str
assertion_id: str
issuer: str
subject: str
gate_result: bool
entropy: float
attributes_hash: str # SHA-256 of sorted attr JSON — no raw values
event: str # ASSERTION_ACCEPTED | ASSERTION_REJECTED | ...
def canonical_json(self) -> str:
d = {
'assertion_id': self.assertion_id,
'attributes_hash': self.attributes_hash,
'entropy': round(self.entropy, 6),
'event': self.event,
'gate_result': self.gate_result,
'issuer': self.issuer,
'semantic_hash': self.semantic_hash,
'subject': self.subject,
'timestamp': self.timestamp,
}
return json.dumps(d, sort_keys=True, separators=(',', ':'))
@dataclass
class ChainEntry:
sequence: int
prev_hash: str
entry_hash: str
record: AuditRecord
@dataclass
class ChainVerifyResult:
valid: bool
broken_at: Optional[int]
total_entries: int
error: Optional[str] = None
class WORMAuditChain:
def __init__(self, path: str):
self._path = path
self._sequence = 0
self._prev_hash = '0' * 64
# Load last entry to restore chain state
try:
with open(path, 'r', encoding='utf-8') as f:
lines = [l.strip() for l in f if l.strip()]
if lines:
last = json.loads(lines[-1])
self._prev_hash = last['entry_hash']
self._sequence = last['sequence'] + 1
except FileNotFoundError:
pass
def append(self, record: AuditRecord) -> ChainEntry:
entry_hash = hashlib.sha256(
(self._prev_hash + record.canonical_json()).encode('utf-8')
).hexdigest()
entry = ChainEntry(
sequence=self._sequence,
prev_hash=self._prev_hash,
entry_hash=entry_hash,
record=record,
)
line = json.dumps({
'sequence': entry.sequence,
'prev_hash': entry.prev_hash,
'entry_hash': entry.entry_hash,
'record': json.loads(record.canonical_json()),
}, sort_keys=True, separators=(',', ':'))
with open(self._path, 'a', encoding='utf-8') as f:
f.write(line + '\n')
self._prev_hash = entry_hash
self._sequence += 1
return entry
@staticmethod
def verify_chain(path: str) -> ChainVerifyResult:
try:
with open(path, 'r', encoding='utf-8') as f:
lines = [l.strip() for l in f if l.strip()]
except FileNotFoundError:
return ChainVerifyResult(valid=True, broken_at=None, total_entries=0)
prev_hash = '0' * 64
for i, line in enumerate(lines):
try:
entry = json.loads(line)
except json.JSONDecodeError as e:
return ChainVerifyResult(valid=False, broken_at=i, total_entries=len(lines),
error=f"JSON parse error: {e}")
if entry.get('prev_hash') != prev_hash:
return ChainVerifyResult(valid=False, broken_at=i, total_entries=len(lines),
error="prev_hash mismatch")
record_json = json.dumps(entry['record'], sort_keys=True, separators=(',', ':'))
expected_hash = hashlib.sha256(
(prev_hash + record_json).encode('utf-8')
).hexdigest()
if entry.get('entry_hash') != expected_hash:
return ChainVerifyResult(valid=False, broken_at=i, total_entries=len(lines),
error="entry_hash mismatch")
prev_hash = entry['entry_hash']
return ChainVerifyResult(valid=True, broken_at=None, total_entries=len(lines))
def attributes_hash(attrs: dict) -> str:
"""SHA-256 of sorted attribute JSON. No raw values stored in audit records."""
sorted_attrs = {k: sorted(v) for k, v in sorted(attrs.items())}
return hashlib.sha256(
json.dumps(sorted_attrs, sort_keys=True, separators=(',', ':')).encode('utf-8')
).hexdigest()
|