diff --git a/src/event_log.rs b/src/event_log.rs index 693a1ef..63f82af 100644 --- a/src/event_log.rs +++ b/src/event_log.rs @@ -1 +1,288 @@ -// Placeholder for upcoming event_log module. +use crate::domain::{unix_secs_now, RuntimeState, POLICY_SCHEMA_VERSION}; +use anyhow::{anyhow, Context, Result}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sha2::{Digest, Sha256}; +use std::fs::{self, File, OpenOptions}; +use std::io::{BufRead, BufReader, Write}; +use std::path::{Path, PathBuf}; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum EventType { + CommitmentCreated, + RuntimeTransition, + CommitmentTransition, + PolicyApplied, + EvidenceObserved, + ViolationObserved, + TransitionStarted, + ReviewCompleted, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct EventRecord { + pub schema_version: u16, + pub sequence: u64, + pub timestamp_unix_secs: u64, + pub event_type: EventType, + pub commitment_id: Option, + pub runtime_state: RuntimeState, + pub payload_json: Value, + pub previous_hash: Option, + pub hash: String, +} + +#[derive(Debug)] +pub struct EventLog { + path: PathBuf, + next_sequence: u64, + previous_hash: Option, +} + +impl EventLog { + pub fn open(path: impl AsRef) -> Result { + let path = path.as_ref().to_path_buf(); + if let Some(parent) = path.parent() { + fs::create_dir_all(parent).with_context(|| { + format!("create parent directories for event log {}", path.display()) + })?; + } + + let mut next_sequence = 1; + let mut previous_hash = None; + + if path.exists() { + let file = + File::open(&path).with_context(|| format!("open event log {}", path.display()))?; + for (line_index, line) in BufReader::new(file).lines().enumerate() { + let line_number = line_index + 1; + let line = line.with_context(|| { + format!("read event log line {line_number} from {}", path.display()) + })?; + if line.trim().is_empty() { + continue; + } + + let record: EventRecord = serde_json::from_str(&line).with_context(|| { + format!("parse event log line {line_number} from {}", path.display()) + })?; + validate_record(&record, next_sequence, previous_hash.as_deref()) + .map_err(|err| anyhow!("validate event log line {line_number}: {err}"))?; + + next_sequence += 1; + previous_hash = Some(record.hash); + } + } + + Ok(Self { + path, + next_sequence, + previous_hash, + }) + } + + pub fn append( + &mut self, + event_type: EventType, + runtime_state: RuntimeState, + commitment_id: Option, + payload_json: Value, + ) -> Result { + if let Some(parent) = self.path.parent() { + fs::create_dir_all(parent).with_context(|| { + format!( + "create parent directories for event log {}", + self.path.display() + ) + })?; + } + + let mut record = EventRecord { + schema_version: POLICY_SCHEMA_VERSION, + sequence: self.next_sequence, + timestamp_unix_secs: unix_secs_now(), + event_type, + commitment_id, + runtime_state, + payload_json, + previous_hash: self.previous_hash.clone(), + hash: String::new(), + }; + record.hash = record_hash(&record)?; + + let mut file = OpenOptions::new() + .create(true) + .append(true) + .open(&self.path) + .with_context(|| format!("open event log for append {}", self.path.display()))?; + serde_json::to_writer(&mut file, &record) + .with_context(|| format!("serialize event log record to {}", self.path.display()))?; + file.write_all(b"\n") + .with_context(|| format!("write event log newline to {}", self.path.display()))?; + file.flush() + .with_context(|| format!("flush event log {}", self.path.display()))?; + + self.next_sequence += 1; + self.previous_hash = Some(record.hash.clone()); + + Ok(record) + } +} + +fn validate_record( + record: &EventRecord, + expected_sequence: u64, + expected_previous_hash: Option<&str>, +) -> Result<()> { + if record.sequence != expected_sequence { + return Err(anyhow!( + "sequence mismatch: expected {}, found {}", + expected_sequence, + record.sequence + )); + } + if record.previous_hash.as_deref() != expected_previous_hash { + return Err(anyhow!( + "previous hash mismatch at sequence {}", + record.sequence + )); + } + + let expected_hash = record_hash(record)?; + if record.hash != expected_hash { + return Err(anyhow!("hash mismatch at sequence {}", record.sequence)); + } + + Ok(()) +} + +fn record_hash(record: &EventRecord) -> Result { + let mut hashable = record.clone(); + hashable.hash.clear(); + let bytes = serde_json::to_vec(&hashable).context("serialize event log record for hash")?; + Ok(hex::encode(Sha256::digest(bytes))) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::domain::RuntimeState; + use serde_json::json; + use std::fs; + use std::path::PathBuf; + use std::time::{SystemTime, UNIX_EPOCH}; + + fn temp_log_path(name: &str) -> PathBuf { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + std::env::temp_dir().join(format!( + "antidrift-event-log-{name}-{}-{nonce}.jsonl", + std::process::id() + )) + } + + #[test] + fn appends_hash_chained_jsonl_records() { + let path = temp_log_path("append"); + let mut log = EventLog::open(&path).unwrap(); + + let first = log + .append( + EventType::RuntimeTransition, + RuntimeState::Transition, + Some("commitment-1".to_string()), + json!({"from": "active", "to": "transition"}), + ) + .unwrap(); + let second = log + .append( + EventType::PolicyApplied, + RuntimeState::Active, + Some("commitment-1".to_string()), + json!({"policy_id": "policy-1"}), + ) + .unwrap(); + + assert_eq!(first.sequence, 1); + assert_eq!(second.sequence, 2); + assert_eq!(second.previous_hash.as_deref(), Some(first.hash.as_str())); + assert_ne!(first.hash, second.hash); + + let contents = fs::read_to_string(&path).unwrap(); + assert_eq!(contents.lines().count(), 2); + + fs::remove_file(path).unwrap(); + } + + #[test] + fn opening_existing_log_validates_chain_and_continues_sequence() { + let path = temp_log_path("reopen"); + { + let mut log = EventLog::open(&path).unwrap(); + log.append( + EventType::RuntimeTransition, + RuntimeState::Transition, + None, + json!({"from": "active"}), + ) + .unwrap(); + } + + let mut reopened = EventLog::open(&path).unwrap(); + let next = reopened + .append( + EventType::PolicyApplied, + RuntimeState::Active, + None, + json!({"policy_id": "policy-1"}), + ) + .unwrap(); + + assert_eq!(next.sequence, 2); + + fs::remove_file(path).unwrap(); + } + + #[test] + fn opening_malformed_json_log_fails() { + let path = temp_log_path("malformed"); + fs::write(&path, "{not json}\n").unwrap(); + + let err = EventLog::open(&path).expect_err("malformed JSONL must fail"); + + assert!(err.to_string().contains("parse event log")); + fs::remove_file(path).unwrap(); + } + + #[test] + fn opening_tampered_log_chain_fails() { + let path = temp_log_path("tampered"); + { + let mut log = EventLog::open(&path).unwrap(); + log.append( + EventType::RuntimeTransition, + RuntimeState::Transition, + None, + json!({"from": "active"}), + ) + .unwrap(); + log.append( + EventType::PolicyApplied, + RuntimeState::Active, + None, + json!({"policy_id": "policy-1"}), + ) + .unwrap(); + } + let contents = fs::read_to_string(&path).unwrap(); + fs::write(&path, contents.replace("\"sequence\":2", "\"sequence\":3")).unwrap(); + + let err = EventLog::open(&path).expect_err("tampered chain must fail"); + + assert!(err.to_string().contains("sequence mismatch")); + fs::remove_file(path).unwrap(); + } +}