File size: 6,399 Bytes
5729d24 | 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 179 180 181 182 183 184 185 186 187 188 | """
Primordial OS Runtime — Audit Log (Phase 6 / Phase 9)
Append-only hash-chained audit logging for runtime decisions.
Each event carries a SHA-256 hash of its own content plus the preceding
event's hash, forming a tamper-evident chain. Phase 9 adds ergonomic
chain helpers: create_runtime_audit_event and append_to_audit_chain.
In-memory audit demonstration only. Not clinical software. Not a medical
device. No health-domain inference, treatment, or decision-making
functionality is present or intended. Does not persist protected health
information or personally identifiable information.
"""
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass
from datetime import datetime, timezone
from primordial_os.runtime import RuntimeDecision
# ---------------------------------------------------------------------------
# Hash computation
# ---------------------------------------------------------------------------
def compute_event_hash(
timestamp: str,
event_type: str,
final_gate_state: str,
hard_stop: bool,
resonance: float,
pressure_adjusted_stability: float,
max_oam_severity: float,
summary: str,
previous_hash: str | None,
) -> str:
"""Return the SHA-256 hex digest of the canonical serialization of event fields."""
payload = json.dumps(
{
"timestamp": timestamp,
"event_type": event_type,
"final_gate_state": final_gate_state,
"hard_stop": hard_stop,
"resonance": resonance,
"pressure_adjusted_stability": pressure_adjusted_stability,
"max_oam_severity": max_oam_severity,
"summary": summary,
"previous_hash": previous_hash,
},
sort_keys=True,
separators=(",", ":"),
)
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
# ---------------------------------------------------------------------------
# Audit event model
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class AuditEvent:
"""
A single immutable audit record in the hash chain.
previous_hash is None for the first event in a chain.
current_hash is SHA-256 of all other fields plus previous_hash.
"""
timestamp: str
event_type: str
final_gate_state: str
hard_stop: bool
resonance: float
pressure_adjusted_stability: float
max_oam_severity: float
summary: str
previous_hash: str | None
current_hash: str
# ---------------------------------------------------------------------------
# Factory
# ---------------------------------------------------------------------------
def create_audit_event(
runtime_decision: RuntimeDecision,
event_type: str = "runtime_evaluation",
previous_hash: str | None = None,
) -> AuditEvent:
"""Create an AuditEvent from a RuntimeDecision and link it to the chain."""
timestamp = datetime.now(timezone.utc).isoformat()
final_gate_state = runtime_decision.final_gate_state.value
hard_stop = runtime_decision.hard_stop
resonance = runtime_decision.hir_result.resonance
pas = runtime_decision.hir_result.pressure_adjusted_stability
max_oam_severity = runtime_decision.oam_assessment.max_severity
summary = runtime_decision.summary
current_hash = compute_event_hash(
timestamp=timestamp,
event_type=event_type,
final_gate_state=final_gate_state,
hard_stop=hard_stop,
resonance=resonance,
pressure_adjusted_stability=pas,
max_oam_severity=max_oam_severity,
summary=summary,
previous_hash=previous_hash,
)
return AuditEvent(
timestamp=timestamp,
event_type=event_type,
final_gate_state=final_gate_state,
hard_stop=hard_stop,
resonance=resonance,
pressure_adjusted_stability=pas,
max_oam_severity=max_oam_severity,
summary=summary,
previous_hash=previous_hash,
current_hash=current_hash,
)
# ---------------------------------------------------------------------------
# Chain verification
# ---------------------------------------------------------------------------
def create_runtime_audit_event(
runtime_decision: RuntimeDecision,
event_type: str = "runtime_evaluation",
) -> AuditEvent:
"""Create a standalone AuditEvent with previous_hash=None."""
return create_audit_event(runtime_decision, event_type=event_type, previous_hash=None)
def append_to_audit_chain(
events: tuple[AuditEvent, ...] | list[AuditEvent],
runtime_decision: RuntimeDecision,
event_type: str = "runtime_evaluation",
) -> tuple[AuditEvent, ...]:
"""
Append a new AuditEvent to an existing chain and return the new chain as a tuple.
Does not mutate the input chain. previous_hash is None when the chain is
empty; otherwise it is the current_hash of the last event in the chain.
"""
previous_hash = events[-1].current_hash if events else None
new_event = create_audit_event(
runtime_decision,
event_type=event_type,
previous_hash=previous_hash,
)
return tuple(events) + (new_event,)
# ---------------------------------------------------------------------------
# Chain verification
# ---------------------------------------------------------------------------
def verify_audit_chain(events: tuple[AuditEvent, ...] | list[AuditEvent]) -> bool:
"""
Return True if every event's hash is internally consistent and each
event is correctly linked to its predecessor.
An empty list is valid. A single event is valid if its hash is correct.
"""
for i, event in enumerate(events):
expected = compute_event_hash(
timestamp=event.timestamp,
event_type=event.event_type,
final_gate_state=event.final_gate_state,
hard_stop=event.hard_stop,
resonance=event.resonance,
pressure_adjusted_stability=event.pressure_adjusted_stability,
max_oam_severity=event.max_oam_severity,
summary=event.summary,
previous_hash=event.previous_hash,
)
if event.current_hash != expected:
return False
if i > 0 and event.previous_hash != events[i - 1].current_hash:
return False
return True
|