security: Tier 4b — WAL per-entry CRC32 checksum (WAL_VERSION 2)

Bump WAL_VERSION to 2: every entry (Save and Tombstone) now ends with a
4-byte CRC32 trailer computed over its type+timestamp+payload bytes, using
the existing clawhdf5_format::checksum::crc32 (already available since
clawhdf5-agent depends on clawhdf5-format with fast-checksum enabled).
A bit-flip inside an entry is now detected and replay stops there, instead
of silently accepting corrupted data as before.

Write side needed no restructuring — append_save/append_tombstone already
buffer an entry's bytes before a single write_all, so the CRC is just
appended to that buffer first.

Read side: read_len_prefixed_str/read_embedding are generalized from
&mut File to R: Read, and a new TeeReader<R> wraps the file handle for one
entry at a time, accumulating every byte actually consumed (via read_exact)
into a buffer. This lets read_entries compute the CRC over exactly the
bytes read for a Save entry without needing to know its length up front
(its sub-fields are length-prefixed and interleaved with the length itself
only becoming known as parsing proceeds). A new read_one_entry<R: Read>
factors the per-entry-type field parsing shared by both the legacy and
current read paths.

Backward compatibility: WAL_VERSION_LEGACY_NO_CRC (1) files are still
readable via WalFile::read_entries (old field-by-file-handle path,
unchanged, no CRC expected). WalFile::open migrates a legacy file by
recreating it fresh in the current format — safe because the only two
real call sites (HDF5Memory::open/create) always call read_entries before
open, so entries are already replayed by the time migration happens.

New tests: a corrupted-payload-byte test confirming replay stops cleanly
at the corrupted entry (no prior coverage existed for mid-entry bit-flip
detection), a legacy-v1-format read test, and an open()-migration test.

crates/clawhdf5-agent/src/wal.rs

@@ -7,10 +7,24 @@ use std::fs::{File, OpenOptions};
 use std::io::{Read, Seek, SeekFrom, Write};
 use std::path::{Path, PathBuf};
 
+use clawhdf5_format::checksum::crc32;
+
 use crate::MemoryError;
 
 const WAL_MAGIC: [u8; 4] = [0x45, 0x48, 0x57, 0x4C]; // "EHWL"
-const WAL_VERSION: u8 = 1;
+
+/// Current WAL format version: every entry ends with a 4-byte CRC32 trailer
+/// (see [`TeeReader`]) so a bit-flip is detected and replay stops there
+/// instead of silently accepting corrupted data.
+const WAL_VERSION: u8 = 2;
+
+/// The only other WAL version this crate still knows how to *read*: no
+/// per-entry CRC trailer. Written by versions of this crate before the CRC32
+/// hardening. `WalFile::open` migrates a legacy file to [`WAL_VERSION`] by
+/// recreating it fresh — safe because every real call site reads existing
+/// entries via [`WalFile::read_entries`] before calling `open` (see
+/// `HDF5Memory::open`), so no data is lost.
+const WAL_VERSION_LEGACY_NO_CRC: u8 = 1;
 
 /// Upper bound on a single length-prefixed WAL field (string bytes, or
 /// embedding element count), to reject a corrupted/truncated WAL length
@@ -67,6 +81,11 @@ pub struct WalFile {
 
 impl WalFile {
     /// Open or create a WAL file. If it exists, read the header and entry count.
+    ///
+    /// A legacy (pre-CRC) WAL file is migrated to the current format by
+    /// recreating it fresh — see [`WAL_VERSION_LEGACY_NO_CRC`]. Callers that
+    /// need the legacy file's entries must call [`WalFile::read_entries`]
+    /// first, before calling `open`.
     pub fn open(path: &Path) -> Result<Self, MemoryError> {
         if path.exists() {
             // Read existing header
@@ -82,30 +101,34 @@ impl WalFile {
             }
             let mut ver = [0u8; 1];
             f.read_exact(&mut ver)?;
-            if ver[0] != WAL_VERSION {
-                return Err(MemoryError::Schema(format!(
-                    "unsupported WAL version {}",
-                    ver[0]
-                )));
+            match ver[0] {
+                WAL_VERSION => {
+                    let mut count_buf = [0u8; 4];
+                    f.read_exact(&mut count_buf)?;
+                    let entry_count = u32::from_le_bytes(count_buf);
+                    // Seek to end for appending
+                    f.seek(SeekFrom::End(0))?;
+                    Ok(Self {
+                        path: path.to_path_buf(),
+                        file: Some(f),
+                        entry_count,
+                        pending_header_sync: 0,
+                    })
+                }
+                WAL_VERSION_LEGACY_NO_CRC => {
+                    drop(f);
+                    let f = create_fresh_wal_file(path)?;
+                    Ok(Self {
+                        path: path.to_path_buf(),
+                        file: Some(f),
+                        entry_count: 0,
+                        pending_header_sync: 0,
+                    })
+                }
+                v => Err(MemoryError::Schema(format!("unsupported WAL version {v}"))),
             }
-            let mut count_buf = [0u8; 4];
-            f.read_exact(&mut count_buf)?;
-            let entry_count = u32::from_le_bytes(count_buf);
-            // Seek to end for appending
-            f.seek(SeekFrom::End(0))?;
-            Ok(Self {
-                path: path.to_path_buf(),
-                file: Some(f),
-                entry_count,
-                pending_header_sync: 0,
-            })
         } else {
-            // Create new WAL
-            let mut f = File::create(path)?;
-            f.write_all(&WAL_MAGIC)?;
-            f.write_all(&[WAL_VERSION])?;
-            f.write_all(&0u32.to_le_bytes())?;
-            f.flush()?;
+            let f = create_fresh_wal_file(path)?;
             Ok(Self {
                 path: path.to_path_buf(),
                 file: Some(f),
@@ -145,6 +168,9 @@ impl WalFile {
         serialize_str(&mut buf, &entry.session_id);
         serialize_str(&mut buf, &entry.tags);
 
+        let crc = crc32(&buf);
+        buf.extend_from_slice(&crc.to_le_bytes());
+
         let f = self
             .file
             .as_mut()
@@ -161,10 +187,12 @@ impl WalFile {
 
     /// Append a tombstone entry (deletion).
     pub fn append_tombstone(&mut self, index: usize, timestamp: f64) -> Result<(), MemoryError> {
-        let mut buf = [0u8; 1 + 8 + 4]; // type + timestamp + index
+        let mut buf = [0u8; 1 + 8 + 4 + 4]; // type + timestamp + index + crc32
         buf[0] = WalEntryType::Tombstone as u8;
         buf[1..9].copy_from_slice(&timestamp.to_le_bytes());
         buf[9..13].copy_from_slice(&(index as u32).to_le_bytes());
+        let crc = crc32(&buf[..13]);
+        buf[13..17].copy_from_slice(&crc.to_le_bytes());
 
         let f = self
             .file
@@ -185,7 +213,9 @@ impl WalFile {
     /// Reads until EOF — the header `entry_count` is used only for pre-allocation
     /// (and may be stale if written with deferred group-commit updates). This
     /// tolerates both truncated files (crash mid-write) and stale header counts
-    /// (crash before the next group-commit header sync).
+    /// (crash before the next group-commit header sync). On a `WAL_VERSION`
+    /// file, a CRC32 mismatch on an entry is treated the same way — replay
+    /// stops there rather than accepting corrupted data.
     pub fn read_entries(path: &Path) -> Result<Vec<WalEntry>, MemoryError> {
         if !path.exists() {
             return Ok(Vec::new());
@@ -197,81 +227,45 @@ impl WalFile {
         if header[0..4] != WAL_MAGIC {
             return Err(MemoryError::Schema("invalid WAL magic bytes".into()));
         }
-        if header[4] != WAL_VERSION {
-            return Err(MemoryError::Schema(format!(
-                "unsupported WAL version {}",
-                header[4]
-            )));
-        }
         // entry_count is a pre-allocation hint only — we read until EOF.
         let entry_count_hint = u32::from_le_bytes([header[5], header[6], header[7], header[8]]);
         let mut entries = Vec::with_capacity(entry_count_hint as usize);
 
-        loop {
-            // Read entry type — EOF here is normal end-of-log, not an error
-            let mut type_buf = [0u8; 1];
-            if f.read_exact(&mut type_buf).is_err() {
-                break;
-            }
-            let entry_type = match WalEntryType::from_u8(type_buf[0]) {
-                Some(et) => et,
-                None => break,
-            };
-
-            let mut ts_buf = [0u8; 8];
-            if f.read_exact(&mut ts_buf).is_err() {
-                break;
-            }
-            let timestamp = f64::from_le_bytes(ts_buf);
-
-            match entry_type {
-                WalEntryType::Save => {
-                    let Ok(chunk) = read_len_prefixed_str(&mut f) else {
-                        break;
-                    };
-                    let Ok(embedding) = read_embedding(&mut f) else {
-                        break;
-                    };
-                    let Ok(source_channel) = read_len_prefixed_str(&mut f) else {
-                        break;
-                    };
-                    let Ok(session_id) = read_len_prefixed_str(&mut f) else {
-                        break;
-                    };
-                    let Ok(tags) = read_len_prefixed_str(&mut f) else {
-                        break;
-                    };
-                    entries.push(WalEntry {
-                        entry_type,
-                        timestamp,
-                        chunk,
-                        embedding,
-                        source_channel,
-                        session_id,
-                        tags,
-                        tombstone_index: None,
-                    });
+        match header[4] {
+            WAL_VERSION => loop {
+                let raw_and_result = {
+                    let mut tee = TeeReader::new(&mut f);
+                    let result = read_one_entry(&mut tee);
+                    (tee.into_buf(), result)
+                };
+                let (raw, result) = raw_and_result;
+                let entry_opt = match result {
+                    Err(()) => break,
+                    Ok(v) => v,
+                };
+                let mut crc_buf = [0u8; 4];
+                if f.read_exact(&mut crc_buf).is_err() {
+                    break;
                 }
-                WalEntryType::Tombstone => {
-                    let mut idx_buf = [0u8; 4];
-                    if f.read_exact(&mut idx_buf).is_err() {
-                        break;
-                    }
-                    let idx = u32::from_le_bytes(idx_buf) as usize;
-                    entries.push(WalEntry {
-                        entry_type,
-                        timestamp,
-                        chunk: String::new(),
-                        embedding: Vec::new(),
-                        source_channel: String::new(),
-                        session_id: String::new(),
-                        tags: String::new(),
-                        tombstone_index: Some(idx),
-                    });
+                let stored_crc = u32::from_le_bytes(crc_buf);
+                if crc32(&raw) != stored_crc {
+                    // Corruption detected — stop replay here, same as a clean
+                    // truncation/EOF, rather than accepting the bad entry.
+                    break;
                 }
-                WalEntryType::ActivationUpdate => {
-                    // Reserved for future use
+                if let Some(entry) = entry_opt {
+                    entries.push(entry);
                 }
+            },
+            WAL_VERSION_LEGACY_NO_CRC => loop {
+                match read_one_entry(&mut f) {
+                    Err(()) => break,
+                    Ok(Some(entry)) => entries.push(entry),
+                    Ok(None) => {}
+                }
+            },
+            v => {
+                return Err(MemoryError::Schema(format!("unsupported WAL version {v}")));
             }
         }
         Ok(entries)
@@ -281,11 +275,7 @@ impl WalFile {
     pub fn truncate(&mut self) -> Result<(), MemoryError> {
         // Close existing handle and recreate
         self.file = None;
-        let mut f = File::create(&self.path)?;
-        f.write_all(&WAL_MAGIC)?;
-        f.write_all(&[WAL_VERSION])?;
-        f.write_all(&0u32.to_le_bytes())?;
-        f.flush()?;
+        let f = create_fresh_wal_file(&self.path)?;
         self.file = Some(f);
         self.entry_count = 0;
         self.pending_header_sync = 0;
@@ -350,7 +340,7 @@ fn serialize_str(buf: &mut Vec<u8>, s: &str) {
     buf.extend_from_slice(bytes);
 }
 
-fn read_len_prefixed_str(f: &mut File) -> Result<String, MemoryError> {
+fn read_len_prefixed_str<R: Read>(f: &mut R) -> Result<String, MemoryError> {
     let mut len_buf = [0u8; 4];
     f.read_exact(&mut len_buf)?;
     let len = u32::from_le_bytes(len_buf) as usize;
@@ -364,7 +354,7 @@ fn read_len_prefixed_str(f: &mut File) -> Result<String, MemoryError> {
     String::from_utf8(buf).map_err(|e| MemoryError::Schema(format!("invalid UTF-8 in WAL: {e}")))
 }
 
-fn read_embedding(f: &mut File) -> Result<Vec<f32>, MemoryError> {
+fn read_embedding<R: Read>(f: &mut R) -> Result<Vec<f32>, MemoryError> {
     let mut len_buf = [0u8; 4];
     f.read_exact(&mut len_buf)?;
     let count = u32::from_le_bytes(len_buf) as usize;
@@ -383,6 +373,99 @@ fn read_embedding(f: &mut File) -> Result<Vec<f32>, MemoryError> {
     Ok(vals)
 }
 
+/// Create a fresh WAL file at `path` with the current-version header,
+/// truncating/overwriting anything already there.
+fn create_fresh_wal_file(path: &Path) -> Result<File, MemoryError> {
+    let mut f = File::create(path)?;
+    f.write_all(&WAL_MAGIC)?;
+    f.write_all(&[WAL_VERSION])?;
+    f.write_all(&0u32.to_le_bytes())?;
+    f.flush()?;
+    Ok(f)
+}
+
+/// Wraps a [`Read`]er, accumulating every byte actually consumed (including
+/// via `read_exact`, which is implemented in terms of `read`) into an
+/// internal buffer — used to capture a WAL entry's raw bytes for CRC32
+/// verification without needing to know its length up front.
+struct TeeReader<'a, R: Read> {
+    inner: &'a mut R,
+    buf: Vec<u8>,
+}
+
+impl<'a, R: Read> TeeReader<'a, R> {
+    fn new(inner: &'a mut R) -> Self {
+        Self {
+            inner,
+            buf: Vec::new(),
+        }
+    }
+
+    fn into_buf(self) -> Vec<u8> {
+        self.buf
+    }
+}
+
+impl<R: Read> Read for TeeReader<'_, R> {
+    fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
+        let n = self.inner.read(out)?;
+        self.buf.extend_from_slice(&out[..n]);
+        Ok(n)
+    }
+}
+
+/// Read one WAL entry (type + timestamp + type-specific payload) from `r`.
+///
+/// Returns `Ok(None)` for entry types with no representable `WalEntry` (only
+/// `ActivationUpdate`, reserved for future use). Returns `Err(())` on any
+/// read failure or unrecognized entry type — the caller treats this the same
+/// as a clean end-of-log (crash-mid-write tolerance).
+fn read_one_entry<R: Read>(r: &mut R) -> Result<Option<WalEntry>, ()> {
+    let mut type_buf = [0u8; 1];
+    r.read_exact(&mut type_buf).map_err(|_| ())?;
+    let entry_type = WalEntryType::from_u8(type_buf[0]).ok_or(())?;
+
+    let mut ts_buf = [0u8; 8];
+    r.read_exact(&mut ts_buf).map_err(|_| ())?;
+    let timestamp = f64::from_le_bytes(ts_buf);
+
+    match entry_type {
+        WalEntryType::Save => {
+            let chunk = read_len_prefixed_str(r).map_err(|_| ())?;
+            let embedding = read_embedding(r).map_err(|_| ())?;
+            let source_channel = read_len_prefixed_str(r).map_err(|_| ())?;
+            let session_id = read_len_prefixed_str(r).map_err(|_| ())?;
+            let tags = read_len_prefixed_str(r).map_err(|_| ())?;
+            Ok(Some(WalEntry {
+                entry_type,
+                timestamp,
+                chunk,
+                embedding,
+                source_channel,
+                session_id,
+                tags,
+                tombstone_index: None,
+            }))
+        }
+        WalEntryType::Tombstone => {
+            let mut idx_buf = [0u8; 4];
+            r.read_exact(&mut idx_buf).map_err(|_| ())?;
+            let idx = u32::from_le_bytes(idx_buf) as usize;
+            Ok(Some(WalEntry {
+                entry_type,
+                timestamp,
+                chunk: String::new(),
+                embedding: Vec::new(),
+                source_channel: String::new(),
+                session_id: String::new(),
+                tags: String::new(),
+                tombstone_index: Some(idx),
+            }))
+        }
+        WalEntryType::ActivationUpdate => Ok(None),
+    }
+}
+
 // --- Tests ---
 
 #[cfg(test)]
@@ -799,6 +882,86 @@ mod tests {
         assert!(err.contains("unsupported WAL version"), "got: {err}");
     }
 
+    #[test]
+    fn test_wal_v2_detects_corrupted_payload_and_stops_replay() {
+        let dir = TempDir::new().unwrap();
+        let wal_path = dir.path().join("test.h5.wal");
+        let mut wal = WalFile::open(&wal_path).unwrap();
+        wal.append_save(&make_wal_entry("first", &[1.0, 2.0]))
+            .unwrap();
+        let len_after_first = std::fs::metadata(&wal_path).unwrap().len();
+        wal.append_save(&make_wal_entry("second", &[3.0, 4.0]))
+            .unwrap();
+        drop(wal);
+
+        // Flip one byte inside the second entry's "second" chunk string
+        // (well past the header and the first entry, and not touching any
+        // length-prefix field) — this must be caught by the CRC32 trailer,
+        // not by any length-cap guard.
+        let mut bytes = std::fs::read(&wal_path).unwrap();
+        let corrupt_at = len_after_first as usize + 15;
+        bytes[corrupt_at] ^= 0xFF;
+        std::fs::write(&wal_path, &bytes).unwrap();
+
+        let entries = WalFile::read_entries(&wal_path).unwrap();
+        assert_eq!(
+            entries.len(),
+            1,
+            "the corrupted second entry must not be returned"
+        );
+        assert_eq!(entries[0].chunk, "first");
+    }
+
+    #[test]
+    fn test_wal_reads_legacy_v1_format_without_crc() {
+        let dir = TempDir::new().unwrap();
+        let wal_path = dir.path().join("legacy.h5.wal");
+        let mut buf = Vec::new();
+        buf.extend_from_slice(&WAL_MAGIC);
+        buf.push(WAL_VERSION_LEGACY_NO_CRC);
+        buf.extend_from_slice(&1u32.to_le_bytes());
+        // One Save entry in the old format: type + timestamp + fields, with
+        // no trailing CRC32.
+        buf.push(WalEntryType::Save as u8);
+        buf.extend_from_slice(&42.0f64.to_le_bytes());
+        serialize_str(&mut buf, "legacy-chunk");
+        let embedding = [1.0f32, 2.0];
+        buf.extend_from_slice(&(embedding.len() as u32).to_le_bytes());
+        for v in embedding {
+            buf.extend_from_slice(&v.to_le_bytes());
+        }
+        serialize_str(&mut buf, "chan");
+        serialize_str(&mut buf, "sess");
+        serialize_str(&mut buf, "tags");
+        std::fs::write(&wal_path, &buf).unwrap();
+
+        let entries = WalFile::read_entries(&wal_path).unwrap();
+        assert_eq!(entries.len(), 1);
+        assert_eq!(entries[0].chunk, "legacy-chunk");
+        assert_eq!(entries[0].embedding, vec![1.0, 2.0]);
+    }
+
+    #[test]
+    fn test_wal_open_migrates_legacy_v1_to_current_version() {
+        let dir = TempDir::new().unwrap();
+        let wal_path = dir.path().join("legacy.h5.wal");
+        let mut buf = Vec::new();
+        buf.extend_from_slice(&WAL_MAGIC);
+        buf.push(WAL_VERSION_LEGACY_NO_CRC);
+        buf.extend_from_slice(&0u32.to_le_bytes());
+        std::fs::write(&wal_path, &buf).unwrap();
+
+        let wal = WalFile::open(&wal_path).unwrap();
+        assert!(wal.is_empty());
+        drop(wal);
+
+        let bytes = std::fs::read(&wal_path).unwrap();
+        assert_eq!(
+            bytes[4], WAL_VERSION,
+            "legacy file must be migrated to the current version"
+        );
+    }
+
     #[test]
     fn test_wal_disabled() {
         let dir = TempDir::new().unwrap();