File size: 5,471 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 | """
Primordial OS Runtime — Audit Store (Phase 10)
Safe JSONL file writer and reader for hash-chained audit events.
Writes one JSON object per line (JSONL). Parent directory must exist before
writing. Hidden file paths are rejected. File is created on first append;
existing content is preserved on subsequent appends.
No raw prompts, user input, PHI, PII, or clinical content is written —
only the structural runtime fields that already exist in AuditEvent.
Pre-validation architecture. Not clinical software. Not a medical device.
No health-domain inference, treatment, or decision-making functionality
is present or intended.
"""
from __future__ import annotations
import json
from pathlib import Path
from primordial_os.audit_log import AuditEvent, verify_audit_chain
# ---------------------------------------------------------------------------
# Serialization
# ---------------------------------------------------------------------------
def audit_event_to_dict(event: AuditEvent) -> dict:
"""Return a plain dict of structural runtime fields only.
Explicit field enumeration ensures no unexpected keys are written.
No raw prompts, user input, PHI, PII, or clinical content is included.
"""
return {
"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,
"current_hash": event.current_hash,
}
def audit_event_from_dict(data: dict) -> AuditEvent:
"""Reconstruct an AuditEvent from a plain dict (e.g. parsed from JSONL)."""
return AuditEvent(
timestamp=data["timestamp"],
event_type=data["event_type"],
final_gate_state=data["final_gate_state"],
hard_stop=data["hard_stop"],
resonance=data["resonance"],
pressure_adjusted_stability=data["pressure_adjusted_stability"],
max_oam_severity=data["max_oam_severity"],
summary=data["summary"],
previous_hash=data["previous_hash"],
current_hash=data["current_hash"],
)
# ---------------------------------------------------------------------------
# Path safety
# ---------------------------------------------------------------------------
def _validate_write_path(path: Path) -> None:
"""Raise for unsafe or invalid write targets.
Checks:
- hidden file paths (name starts with '.')
- parent directory does not exist
"""
if path.name.startswith("."):
raise ValueError(f"Hidden file paths are not permitted: {path}")
if not path.parent.exists():
raise FileNotFoundError(
f"Parent directory does not exist: {path.parent}"
)
# ---------------------------------------------------------------------------
# JSONL I/O
# ---------------------------------------------------------------------------
def append_audit_event_jsonl(path: str | Path, event: AuditEvent) -> None:
"""Append one AuditEvent as a single JSON line to the JSONL file at path.
Safety requirements:
- Parent directory must exist; will not be created automatically.
- Hidden file paths (name starts with '.') are rejected.
- File is created if absent; existing lines are never modified.
- Encoding: UTF-8.
- Format: one compact JSON object per line, keys sorted, no trailing spaces.
- No raw prompts, PHI, PII, or clinical content is written.
"""
p = Path(path)
_validate_write_path(p)
line = json.dumps(
audit_event_to_dict(event),
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
)
with p.open("a", encoding="utf-8") as f:
f.write(line + "\n")
def read_audit_events_jsonl(path: str | Path) -> tuple[AuditEvent, ...]:
"""Read all AuditEvents from a JSONL file and return them as a tuple.
Events are returned in file order (append order).
Raises:
FileNotFoundError — if the file does not exist.
ValueError — if any line contains malformed JSON.
"""
p = Path(path)
if not p.exists():
raise FileNotFoundError(f"Audit log file not found: {p}")
events: list[AuditEvent] = []
with p.open("r", encoding="utf-8") as f:
for line_num, raw_line in enumerate(f, start=1):
line = raw_line.strip()
if not line:
continue
try:
data = json.loads(line)
except json.JSONDecodeError as exc:
raise ValueError(
f"Malformed JSON on line {line_num} of {p}: {exc}"
) from exc
events.append(audit_event_from_dict(data))
return tuple(events)
def verify_audit_jsonl(path: str | Path) -> bool:
"""Return True if the JSONL file at path contains a valid, unbroken audit chain.
Returns False — without raising — when:
- the file is missing
- the chain hash linkage is broken
- any event hash does not match its stored fields
An empty file is treated as a valid (empty) chain and returns True.
"""
p = Path(path)
try:
events = read_audit_events_jsonl(p)
except FileNotFoundError:
return False
return verify_audit_chain(events)
|