File size: 6,521 Bytes
c32102b ef02f17 c32102b ef02f17 c32102b ef02f17 c32102b | 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 189 190 191 192 193 194 195 196 197 | """
Rhodawk AI — Immutable Audit Trail Engine
==========================================
Every AI action is appended to an append-only JSONL file with SHA-256 chaining.
Each entry references the hash of the previous entry, creating a tamper-evident
chain of custody for every line of AI-generated code. Required for SOC 2 / ISO 27001.
"""
import hashlib
import json
import os
import threading
import time
from typing import Optional
AUDIT_LOG_PATH = "/data/audit_trail.jsonl"
_audit_write_lock = threading.Lock()
_last_hash: Optional[str] = None
def _compute_hash(entry: dict) -> str:
canonical = json.dumps(entry, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(canonical.encode()).hexdigest()
def _get_last_hash() -> str:
global _last_hash
if _last_hash:
return _last_hash
if not os.path.exists(AUDIT_LOG_PATH):
return "GENESIS"
try:
with open(AUDIT_LOG_PATH, "rb") as f:
lines = f.read().splitlines()
if not lines:
return "GENESIS"
last_line = lines[-1].decode("utf-8").strip()
if not last_line:
return "GENESIS"
last_entry = json.loads(last_line)
_last_hash = last_entry.get("entry_hash", "GENESIS")
return _last_hash
except Exception:
return "GENESIS"
def log_audit_event(
event_type: str,
job_id: str,
repo: str,
model: str,
details: dict,
outcome: str = "PENDING",
) -> str:
"""
Append an audit event to the immutable JSONL chain.
Returns the entry hash for cross-referencing.
"""
global _last_hash
with _audit_write_lock:
prev_hash = _get_last_hash()
entry = {
"schema_version": "1.0",
"timestamp_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"unix_ts": time.time(),
"event_type": event_type,
"job_id": job_id,
"repo": repo,
"model_version": model,
"outcome": outcome,
"details": details,
"prev_hash": prev_hash,
}
entry_hash = _compute_hash(entry)
entry["entry_hash"] = entry_hash
os.makedirs(os.path.dirname(AUDIT_LOG_PATH), exist_ok=True)
with open(AUDIT_LOG_PATH, "a") as f:
f.write(json.dumps(entry) + "\n")
_last_hash = entry_hash
return entry_hash
def read_audit_trail(limit: int = 50) -> list[dict]:
"""Return the last N audit events for dashboard display."""
if not os.path.exists(AUDIT_LOG_PATH):
return []
events = []
try:
with open(AUDIT_LOG_PATH, "r") as f:
for line in f:
line = line.strip()
if line:
try:
events.append(json.loads(line))
except json.JSONDecodeError:
pass
except OSError:
return []
return events[-limit:]
def verify_chain_integrity() -> tuple[bool, str]:
"""
Walk the ENTIRE audit chain and verify each entry's hash.
Returns (is_valid, summary_message).
Used for compliance attestation.
MINOR BUG FIX: Previously read_audit_trail(1000) was called which truncated
the chain — a log with >1000 entries would appear verified even if early entries
were tampered. Now the full file is always read for integrity checks.
"""
if not os.path.exists(AUDIT_LOG_PATH):
return True, "No audit log yet — chain is clean."
events = []
try:
with open(AUDIT_LOG_PATH, "r") as f:
for line in f:
line = line.strip()
if line:
try:
events.append(json.loads(line))
except json.JSONDecodeError:
return False, f"CHAIN BROKEN: malformed JSON entry at line {len(events) + 1}."
except OSError as e:
return False, f"Could not read audit log: {e}"
if not events:
return True, "Empty log — chain is clean."
for i, entry in enumerate(events):
stored_hash = entry.pop("entry_hash", None)
computed = _compute_hash(entry)
entry["entry_hash"] = stored_hash
if computed != stored_hash:
return False, f"CHAIN BROKEN at entry {i} (event: {entry.get('event_type')}). Possible tampering detected."
if i > 0:
expected_prev = events[i - 1]["entry_hash"]
if entry["prev_hash"] != expected_prev:
return False, f"HASH CHAIN BROKEN between entries {i-1} and {i}."
return True, f"Chain VERIFIED — {len(events)} entries, all hashes valid."
def export_compliance_report(output_path: str = "/data/rhodawk_soc2_audit_summary.md") -> str:
events = read_audit_trail(limit=100000)
valid, integrity_msg = verify_chain_integrity()
by_type: dict[str, int] = {}
by_outcome: dict[str, int] = {}
repos: dict[str, int] = {}
for event in events:
by_type[event.get("event_type", "UNKNOWN")] = by_type.get(event.get("event_type", "UNKNOWN"), 0) + 1
by_outcome[event.get("outcome", "UNKNOWN")] = by_outcome.get(event.get("outcome", "UNKNOWN"), 0) + 1
repos[event.get("repo", "unknown")] = repos.get(event.get("repo", "unknown"), 0) + 1
report = [
"# Rhodawk AI SOC 2 Audit Evidence Summary",
"",
f"Generated: {time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())}",
f"Chain integrity: {'VERIFIED' if valid else 'COMPROMISED'}",
f"Integrity detail: {integrity_msg}",
f"Total audit events: {len(events)}",
"",
"## Event Types",
"",
*[f"- {name}: {count}" for name, count in sorted(by_type.items())],
"",
"## Outcomes",
"",
*[f"- {name}: {count}" for name, count in sorted(by_outcome.items())],
"",
"## Repository Coverage",
"",
*[f"- {name}: {count} event(s)" for name, count in sorted(repos.items())],
"",
"## Latest Evidence Entries",
"",
]
for event in events[-25:]:
report.append(
f"- `{event.get('timestamp_utc')}` `{event.get('event_type')}` "
f"`{event.get('outcome')}` hash `{event.get('entry_hash', '')[:16]}`"
)
os.makedirs(os.path.dirname(output_path), exist_ok=True)
with open(output_path, "w", encoding="utf-8") as f:
f.write("\n".join(report))
return output_path
|