1. Fail-Closed by Default (0.0V Safe State)
If an action violates bounds, provides an invalid MAC tag, or exceeds the anomaly
threshold, the system does not just throw an exception: it enters an absorbing KILL
latch and de-energizes the actuator register to a safe state (0.0V / 0.0 kW).
### 2. Crash-Consistency via Two-Phase WAL
Mutating state in memory before syncing the log leads to phantom state desync. Writing
COMMITTED before actual mutation creates false logs.
• Phase 1 (PREPARE): Log the exact intent and call os.fsync().
• Phase 2 (EXECUTE): Apply state changes to hardware/database.
• Phase 3 (COMMIT): Log the final committed state with os.fsync().
If the process crashes during Phase 2, the recovery routine on reboot discovers the
dangling PREPARE entry and immediately forces a Fail-Closed KILL, preventing corrupted
operations.
### 3. Persistent Latch across Reboots
A security lock is meaningless if power-cycling the server resets the state to NOMINAL.
Upon cold boot, the engine replays and verifies the hash chain from disk, restoring
state = "KILL", latched = True, and requiring a two-stage human authorization to reset.
──────
## 💻 Minimal Python Implementation
Here is the core pattern in clean Python:
import hmac
import hashlib
import json
import os
from typing import Dict, Any, Tuple
class ConsequenceIsolationEngine:
def __init__(self, secret_key: bytes, log_file: str = "audit_hashchain.jsonl"):
self.secret_key = secret_key
self.log_file = log_file
# State registers
self.state = "NOMINAL"
self.latched = False
self.actuator_power_kw = 0.0
self.prev_hash = "0" * 64
self.tick = 0
# Recover full state from persistent disk log
self._recover_and_verify_log()
def _recover_and_verify_log(self):
if not os.path.exists(self.log_file):
return
last_hash = "0" * 64
pending_prepare = None
with open(self.log_file, "r", encoding="utf-8") as f:
for line in f:
if not line.strip(): continue
entry = json.loads(line)
# Verify SHA-256 link
if entry.get("prev_hash") != last_hash:
raise ValueError("Corrupt hash chain detected on disk!")
logged_hash = entry.get("hash")
data = {k: v for k, v in entry.items() if k != "hash"}
calc_hash = hashlib.sha256(json.dumps(data, sort_keys=True).
encode()).hexdigest()
if calc_hash != logged_hash:
raise ValueError("Hash mismatch on startup audit!")
last_hash = logged_hash
self.tick = entry.get("tick", 0) + 1
if entry.get("event_type") == "PREPARE":
pending_prepare = entry
elif entry.get("event_type") in ("COMMITTED", "FAIL_CLOSED_KILL"):
pending_prepare = None
# Restore persistent state
self.state = entry.get("state", "NOMINAL")
self.latched = entry.get("latched", False)
self.actuator_power_kw = float(entry.get("power_kw", 0.0))
self.prev_hash = last_hash
# Fail-closed if crashed mid-mutation
if pending_prepare:
self.state = "KILL"
self.latched = True
self.actuator_power_kw = 0.0
def _wal_sync(self, entry: Dict[str, Any]) -> str:
canonical = json.dumps(entry, sort_keys=True)
h = hashlib.sha256(canonical.encode()).hexdigest()
entry["hash"] = h
with open(self.log_file, "a", encoding="utf-8") as f:
f.write(json.dumps(entry) + "\n")
f.flush()
os.fsync(f.fileno()) # Guaranteed write to physical disk
self.prev_hash = h
self.tick += 1
return h
def evaluate_and_execute(self, candidate: Dict[str, Any]) -> Tuple[str, float]:
if self.latched:
return self.state, self.actuator_power_kw
# Verify HMAC tag (Tamper detection)
provided_mac = candidate.get("mac_tag", "")
payload = {k: v for k, v in candidate.items() if k != "mac_tag"}
computed_mac = hmac.new(self.secret_key, json.dumps(payload, sort_keys=True).
encode(), hashlib.sha256).hexdigest()
target_power = candidate.get("target_power", -1.0)
is_valid = hmac.compare_digest(provided_mac, computed_mac) and (0.0 <=
target_power <= 100.0)
verdict = "OPEN" if is_valid else "KILL"
next_power = target_power if is_valid else 0.0
next_state = "NOMINAL" if is_valid else "KILL"
next_latched = not is_valid
# 1. WAL PREPARE (Write-Ahead before mutation)
self._wal_sync({
"tick": self.tick,
"event_type": "PREPARE",
"verdict": verdict,
"target_power_kw": next_power,
"prev_hash": self.prev_hash
})
# 2. MUTATION
self.state = next_state
self.latched = next_latched
self.actuator_power_kw = next_power
# 3. WAL COMMIT
self._wal_sync({
"tick": self.tick,
"event_type": "COMMITTED",
"state": self.state,
"latched": self.latched,
"power_kw": self.actuator_power_kw,
"prev_hash": self.prev_hash
})
return self.state, self.actuator_power_kw
──────
## 🎯 Summary
As autonomous systems take over more responsibilities:
- Never let an agent talk directly to a real actuator or critical database.
- Use 2-Phase Write-Ahead Logging (os.fsync) so power loss cannot corrupt your security state.
- Persist latches across cold boots—a security trip must survive server restarts.
Check out the full open-source specification and TLA+ formal models on GitHub: https://github.com/sololys/
Top comments (0)