Harden event log validation

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<String>)> {
+    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();
+    }
 }