Watch must not trust unauthenticated footer body_len

crates/engine/src/log/file.rs

@@ -326,23 +326,25 @@ impl LogFile {
 
     /// Returns the byte offset where record data ends — stops before the footer
     /// in sealed files so that `scan_since` doesn't misparse footer bytes as records.
+    ///
+    /// `body_len` in the trailer is outside the CRC. Trusting it unauthenticated
+    /// lets a flip collapse this to 0 and blind watch catch-up. Reuse
+    /// [`read_footer`]'s range+CRC check; on any miss scan to EOF.
     pub async fn data_end_offset(&self) -> u64 {
         let total = self.write_offset.get();
         if total < FOOTER_TRAILER_LEN {
             return total;
         }
-        let Ok(magic_bytes) = self.read_exact(total - 8, 8).await else {
-            return total;
-        };
-        let magic = u64::from_le_bytes(<[u8; 8]>::try_from(&magic_bytes[..]).unwrap_or([0u8; 8]));
-        if magic != FOOTER_MAGIC {
-            return total;
+        match self.read_footer().await {
+            Ok(Some(entries)) => {
+                let body_len: u64 = entries.iter().map(|e| e.encoded_size() as u64).sum();
+                match FOOTER_TRAILER_LEN.checked_add(body_len) {
+                    Some(n) => total.saturating_sub(n),
+                    None => total,
+                }
+            }
+            Ok(None) | Err(_) => total,
         }
-        let Ok(blen_bytes) = self.read_exact(total - FOOTER_TRAILER_LEN, 8).await else {
-            return total;
-        };
-        let body_len = u64::from_le_bytes(<[u8; 8]>::try_from(&blen_bytes[..]).unwrap_or([0u8; 8]));
-        total.saturating_sub(FOOTER_TRAILER_LEN + body_len)
     }
 
     /// Append a buffer at an offset reserved atomically *before* awaiting the
@@ -598,4 +600,33 @@ mod enospc_tests {
             );
         });
     }
+
+    /// `body_len` sits outside the footer CRC. Trusting it blinds
+    /// `scan_since`: `body_len = records_end` → `data_end_offset = 0`.
+    #[test]
+    fn footer_body_len_inflate_must_not_blind_data_end() {
+        run(async {
+            let dir = TempDir::new().unwrap();
+            let path = dir.path().join("data-0000000000.log");
+            let f = LogFile::open_rw(path.clone(), 0).await.unwrap();
+            let (_, _) = f.append(b"keep-record-bytes".to_vec()).await.unwrap();
+            let records_end = f.write_offset();
+            f.write_footer(&[]).await.unwrap();
+            let total = f.write_offset();
+            assert!(total > records_end);
+
+            // Flip the trailer body_len (first 8 bytes of the 24-byte trailer)
+            // to claim the whole file is footer. Magic stays intact.
+            let mut bytes = std::fs::read(&path).unwrap();
+            let blen_at = bytes.len() - FOOTER_TRAILER_LEN as usize;
+            bytes[blen_at..blen_at + 8].copy_from_slice(&records_end.to_le_bytes());
+            std::fs::write(&path, &bytes).unwrap();
+
+            let end = f.data_end_offset().await;
+            assert_eq!(
+                end, total,
+                "unauthenticated body_len must not collapse data_end to 0 (got {end}, total {total}, records_end {records_end})"
+            );
+        });
+    }
 }