File size: 5,804 Bytes
ab48644 | 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 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 | """
Bifrost WORM-Link — Append-only hash chain for Guardian Omega FSM.
Every state transition is sealed. No mutation. No deletion.
Authors: Ahmad Ali Parr, Jessica L. Williams (SNAPKITTYWEST)
"""
import hashlib
import json
import time
from dataclasses import dataclass, field
from typing import List, Optional
@dataclass(frozen=True)
class WORMEntry:
seq: int
timestamp: float
state_from: str
event: str
state_to: str
action: str
record_id: Optional[str]
prev_hash: str
entry_hash: str
class BifrostWORMChain:
"""
Append-only chain for FSM state transitions.
Properties:
- Monotonic: seq always increases
- Immutable: entries cannot be modified after append
- Verifiable: hash chain links every entry to genesis
- Deterministic: same inputs produce same hashes
"""
def __init__(self):
self._chain: List[WORMEntry] = []
self._genesis_hash = hashlib.sha256(b"BIFROST_GENESIS_2026").hexdigest()
@property
def tip(self) -> str:
if not self._chain:
return self._genesis_hash
return self._chain[-1].entry_hash
@property
def length(self) -> int:
return len(self._chain)
def append(self, state_from: str, event: str, state_to: str,
action: str, record_id: Optional[str] = None) -> WORMEntry:
"""Append a new entry. Returns the sealed entry."""
seq = len(self._chain)
ts = time.time()
prev = self.tip
# Compute entry hash deterministically
payload = f"{seq}|{state_from}|{event}|{state_to}|{action}|{record_id}|{prev}"
entry_hash = hashlib.sha256(payload.encode()).hexdigest()
entry = WORMEntry(
seq=seq,
timestamp=ts,
state_from=state_from,
event=event,
state_to=state_to,
action=action,
record_id=record_id,
prev_hash=prev,
entry_hash=entry_hash
)
self._chain.append(entry)
return entry
def verify(self) -> bool:
"""Verify entire chain integrity."""
if not self._chain:
return True
# First entry must link to genesis
if self._chain[0].prev_hash != self._genesis_hash:
return False
# Each entry must link to previous
for i in range(1, len(self._chain)):
if self._chain[i].prev_hash != self._chain[i - 1].entry_hash:
return False
# Verify hash computation
e = self._chain[i]
payload = f"{e.seq}|{e.state_from}|{e.event}|{e.state_to}|{e.action}|{e.record_id}|{e.prev_hash}"
expected = hashlib.sha256(payload.encode()).hexdigest()
if e.entry_hash != expected:
return False
# Monotonic timestamps
for i in range(1, len(self._chain)):
if self._chain[i].timestamp < self._chain[i - 1].timestamp:
return False
return True
def export_json(self) -> str:
"""Export chain as JSON (for audit)."""
return json.dumps([
{
"seq": e.seq,
"timestamp": e.timestamp,
"state_from": e.state_from,
"event": e.event,
"state_to": e.state_to,
"action": e.action,
"record_id": e.record_id,
"prev_hash": e.prev_hash,
"entry_hash": e.entry_hash
}
for e in self._chain
], indent=2)
def rollback_point(self, seq: int) -> Optional[str]:
"""
Find the hash at a given sequence number for recovery.
Does NOT mutate the chain — rollback creates a NEW branch.
"""
if seq < 0 or seq >= len(self._chain):
return None
return self._chain[seq].entry_hash
class GuardianOmegaFSM:
"""
The sovereign FSM with WORM-sealed transitions.
Every state change is recorded immutably.
"""
TRANSITIONS = {
("idle", "record_arrived"): ("absorbing", "absorb"),
("absorbing", "schema_valid"): ("classifying", "classify"),
("absorbing", "schema_invalid"): ("idle", "reject"),
("classifying", "factual_pass"): ("gating", "noop"),
("classifying", "factual_fail"): ("idle", "reject"),
("gating", "plasma_pass"): ("sealing", "seal_worm"),
("gating", "plasma_fail"): ("idle", "reject"),
("sealing", "worm_sealed"): ("idle", "approve"),
("sealing", "worm_failed"): ("recovering", "rollback"),
("quarantined", "recovery_complete"): ("idle", "noop"),
("recovering", "recovery_complete"): ("idle", "noop"),
}
def __init__(self):
self.state = "idle"
self.chain = BifrostWORMChain()
def process_event(self, event: str, record_id: Optional[str] = None) -> str:
"""Process event, transition state, seal to WORM chain."""
# Byzantine detection fires from any state
if event == "byzantine_detected":
old_state = self.state
self.state = "quarantined"
self.chain.append(old_state, event, "quarantined", "quarantine", record_id)
return "quarantine"
key = (self.state, event)
if key not in self.TRANSITIONS:
# No transition defined — stay in current state
self.chain.append(self.state, event, self.state, "noop", record_id)
return "noop"
new_state, action = self.TRANSITIONS[key]
self.chain.append(self.state, event, new_state, action, record_id)
self.state = new_state
return action
def verify_integrity(self) -> bool:
"""Verify the WORM chain is intact."""
return self.chain.verify()
|