Add hash chained event log
This commit is contained in:
+288
-1
@@ -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<String>,
|
||||
pub runtime_state: RuntimeState,
|
||||
pub payload_json: Value,
|
||||
pub previous_hash: Option<String>,
|
||||
pub hash: String,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct EventLog {
|
||||
path: PathBuf,
|
||||
next_sequence: u64,
|
||||
previous_hash: Option<String>,
|
||||
}
|
||||
|
||||
impl EventLog {
|
||||
pub fn open(path: impl AsRef<Path>) -> Result<Self> {
|
||||
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<String>,
|
||||
payload_json: Value,
|
||||
) -> Result<EventRecord> {
|
||||
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<String> {
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user