diff --git a/src/event_log.rs b/src/event_log.rs index 63f82af..e628712 100644 --- a/src/event_log.rs +++ b/src/event_log.rs @@ -21,6 +21,7 @@ pub enum EventType { } #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] pub struct EventRecord { pub schema_version: u16, pub sequence: u64, @@ -49,31 +50,7 @@ impl EventLog { })?; } - 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); - } - } + let (next_sequence, previous_hash) = load_tail(&path)?; Ok(Self { path, @@ -98,6 +75,10 @@ impl EventLog { })?; } + let (next_sequence, previous_hash) = load_tail(&self.path)?; + self.next_sequence = next_sequence; + self.previous_hash = previous_hash; + let mut record = EventRecord { schema_version: POLICY_SCHEMA_VERSION, sequence: self.next_sequence, @@ -130,11 +111,49 @@ impl EventLog { } } +fn load_tail(path: &Path) -> Result<(u64, Option)> { + 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((next_sequence, previous_hash)) +} + fn validate_record( record: &EventRecord, expected_sequence: u64, expected_previous_hash: Option<&str>, ) -> Result<()> { + if record.schema_version != POLICY_SCHEMA_VERSION { + return Err(anyhow!( + "unsupported schema version at sequence {}: expected {}, found {}", + record.sequence, + POLICY_SCHEMA_VERSION, + record.schema_version + )); + } if record.sequence != expected_sequence { return Err(anyhow!( "sequence mismatch: expected {}, found {}", @@ -285,4 +304,89 @@ mod tests { assert!(err.to_string().contains("sequence mismatch")); fs::remove_file(path).unwrap(); } + + #[test] + fn opening_log_with_unknown_record_field_fails() { + let path = temp_log_path("unknown-field"); + { + let mut log = EventLog::open(&path).unwrap(); + log.append( + EventType::RuntimeTransition, + RuntimeState::Transition, + None, + json!({"from": "active"}), + ) + .unwrap(); + } + let contents = fs::read_to_string(&path).unwrap(); + let mut record_json: Value = + serde_json::from_str(contents.lines().next().unwrap()).unwrap(); + record_json["unexpected_field"] = json!("ignored-before-hashing"); + fs::write( + &path, + format!("{}\n", serde_json::to_string(&record_json).unwrap()), + ) + .unwrap(); + + let err = EventLog::open(&path).expect_err("unknown record fields must fail"); + + assert!(err.to_string().contains("parse event log")); + fs::remove_file(path).unwrap(); + } + + #[test] + fn opening_log_with_unsupported_schema_version_fails() { + let path = temp_log_path("unsupported-schema"); + let mut record = EventRecord { + schema_version: 999, + sequence: 1, + timestamp_unix_secs: unix_secs_now(), + event_type: EventType::RuntimeTransition, + commitment_id: None, + runtime_state: RuntimeState::Transition, + payload_json: json!({"from": "active"}), + previous_hash: None, + hash: String::new(), + }; + record.hash = record_hash(&record).unwrap(); + fs::write( + &path, + format!("{}\n", serde_json::to_string(&record).unwrap()), + ) + .unwrap(); + + let err = EventLog::open(&path).expect_err("unsupported schema version must fail"); + + assert!(err.to_string().contains("unsupported schema version")); + fs::remove_file(path).unwrap(); + } + + #[test] + fn stale_handle_reloads_tail_before_append() { + let path = temp_log_path("stale-handle"); + let mut log1 = EventLog::open(&path).unwrap(); + let mut log2 = EventLog::open(&path).unwrap(); + + let first = log1 + .append( + EventType::RuntimeTransition, + RuntimeState::Transition, + None, + json!({"from": "active"}), + ) + .unwrap(); + let second = log2 + .append( + EventType::PolicyApplied, + RuntimeState::Active, + None, + json!({"policy_id": "policy-1"}), + ) + .unwrap(); + + assert_eq!(second.sequence, 2); + assert_eq!(second.previous_hash.as_deref(), Some(first.hash.as_str())); + EventLog::open(&path).unwrap(); + fs::remove_file(path).unwrap(); + } }