Add checksum validation to state entry storage (#1048)

Author: spetz

Cargo.lock

@@ -42,6 +42,7 @@ async fn should_apply_single_entry() {
     assert_eq!(entry.term, 0);
     assert_eq!(entry.version, setup.version());
     assert_eq!(entry.flags, 0);
+    assert!(entry.checksum > 0);
     assert!(entry.timestamp.as_micros() > 0);
     assert_eq!(entry.user_id, user_id);
     assert_eq!(entry.command, command_clone);
@@ -101,6 +102,7 @@ async fn should_apply_multiple_entries() {
     assert_eq!(create_user_entry.term, 0);
     assert_eq!(create_user_entry.version, setup.version());
     assert_eq!(create_user_entry.flags, 0);
+    assert!(create_user_entry.checksum > 0);
     assert!(create_user_entry.timestamp.as_micros() > 0);
     assert_eq!(create_user_entry.user_id, 1);
     assert!(create_user_entry.context.is_empty());
@@ -111,6 +113,7 @@ async fn should_apply_multiple_entries() {
     assert_eq!(create_stream_entry.term, 0);
     assert_eq!(create_stream_entry.version, setup.version());
     assert_eq!(create_stream_entry.flags, 0);
+    assert!(create_stream_entry.checksum > 0);
     assert!(create_stream_entry.timestamp.as_micros() > 0);
     assert!(create_stream_entry.timestamp.as_micros() > create_user_entry.timestamp.as_micros());
     assert_eq!(create_stream_entry.user_id, 2);

integration/tests/state/file.rs

@@ -36,6 +36,8 @@ pub enum IggyError {
     StateFileNotFound = 14,
     #[error("State file corrupted")]
     StateFileCorrupted = 15,
+    #[error("Invalid state entry checksum: {0}, expected: {1}, for index: {2}")]
+    InvalidStateEntryChecksum(u32, u32, u64) = 16,
     #[error("Cannot open database, Path: {0}")]
     CannotOpenDatabase(String) = 19,
     #[error("Resource with key: {0} was not found.")]

sdk/src/error.rs

@@ -1,6 +1,6 @@
 [package]
 name = "server"
-version = "0.3.0"
+version = "0.3.1"
 edition = "2021"
 build = "src/build.rs"
 

server/Cargo.toml

@@ -2,6 +2,7 @@ use crate::state::command::EntryCommand;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 use iggy::bytes_serializable::BytesSerializable;
 use iggy::error::IggyError;
+use iggy::utils::checksum;
 use iggy::utils::timestamp::IggyTimestamp;
 use std::fmt::{Display, Formatter};
 
@@ -13,6 +14,7 @@ use std::fmt::{Display, Formatter};
 /// - `flags` - Reserved for future use
 /// - `timestamp` - Timestamp when the command was issued
 /// - `user_id` - User ID of the user who issued the command
+/// - `checksum` - Checksum of the entry
 /// - `code` - Command code
 /// - `command` - Payload of the command
 /// - `context` - Optional context e.g. used to enrich the payload with additional data
@@ -25,8 +27,9 @@ pub struct StateEntry {
     pub flags: u64,
     pub timestamp: IggyTimestamp,
     pub user_id: u32,
-    pub command: EntryCommand,
+    pub checksum: u32,
     pub context: Bytes,
+    pub command: EntryCommand,
 }
 
 impl StateEntry {
@@ -50,25 +53,57 @@ impl StateEntry {
             flags,
             timestamp,
             user_id,
+            checksum: Self::calculate_checksum(
+                index, term, leader_id, version, flags, timestamp, user_id, &context, &command,
+            ),
             context,
             command,
         }
     }
+
+    #[allow(clippy::too_many_arguments)]
+    fn calculate_checksum(
+        index: u64,
+        term: u64,
+        leader_id: u32,
+        version: u32,
+        flags: u64,
+        timestamp: IggyTimestamp,
+        user_id: u32,
+        context: &Bytes,
+        command: &EntryCommand,
+    ) -> u32 {
+        let command = command.to_bytes();
+        let mut bytes =
+            BytesMut::with_capacity(8 + 8 + 4 + 4 + 8 + 8 + 4 + 4 + context.len() + command.len());
+        bytes.put_u64_le(index);
+        bytes.put_u64_le(term);
+        bytes.put_u32_le(leader_id);
+        bytes.put_u32_le(version);
+        bytes.put_u64_le(flags);
+        bytes.put_u64_le(timestamp.into());
+        bytes.put_u32_le(user_id);
+        bytes.put_u32_le(context.len() as u32);
+        bytes.put_slice(context);
+        bytes.extend(command);
+        checksum::calculate(&bytes.freeze())
+    }
 }
 
 impl Display for StateEntry {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "StateEntry {{ index: {}, term: {}, leader ID: {}, version: {}, flags: {}, timestamp: {}, user ID: {}, command: {:?} }}",
+            "StateEntry {{ index: {}, term: {}, leader ID: {}, version: {}, flags: {}, timestamp: {}, user ID: {}, checksum: {}, command: {:?} }}",
             self.index,
             self.term,
             self.leader_id,
             self.version,
             self.flags,
             self.timestamp,
             self.user_id,
-            self.command,
+            self.checksum,
+            self.command
         )
     }
 }
@@ -77,18 +112,19 @@ impl BytesSerializable for StateEntry {
     fn to_bytes(&self) -> Bytes {
         let command = self.command.to_bytes();
         let mut bytes = BytesMut::with_capacity(
-            8 + 8 + 4 + 4 + 8 + 8 + 4 + 4 + self.context.len() + command.len(),
+            8 + 8 + 4 + 4 + 8 + 8 + 4 + 4 + 4 + self.context.len() + command.len(),
         );
         bytes.put_u64_le(self.index);
         bytes.put_u64_le(self.term);
         bytes.put_u32_le(self.leader_id);
         bytes.put_u32_le(self.version);
         bytes.put_u64_le(self.flags);
-        bytes.put_u64_le(self.timestamp.as_micros());
+        bytes.put_u64_le(self.timestamp.into());
         bytes.put_u32_le(self.user_id);
+        bytes.put_u32_le(self.checksum);
         bytes.put_u32_le(self.context.len() as u32);
         bytes.put_slice(&self.context);
-        bytes.extend(self.command.to_bytes());
+        bytes.extend(command);
         bytes.freeze()
     }
 
@@ -103,9 +139,10 @@ impl BytesSerializable for StateEntry {
         let flags = bytes.slice(24..32).get_u64_le();
         let timestamp = IggyTimestamp::from(bytes.slice(32..40).get_u64_le());
         let user_id = bytes.slice(40..44).get_u32_le();
-        let context_length = bytes.slice(44..48).get_u32_le() as usize;
-        let context = bytes.slice(48..48 + context_length);
-        let command = EntryCommand::from_bytes(bytes.slice(48 + context_length..))?;
+        let checksum = bytes.slice(44..48).get_u32_le();
+        let context_length = bytes.slice(48..52).get_u32_le() as usize;
+        let context = bytes.slice(52..52 + context_length);
+        let command = EntryCommand::from_bytes(bytes.slice(52 + context_length..))?;
 
         Ok(StateEntry {
             index,
@@ -115,6 +152,7 @@ impl BytesSerializable for StateEntry {
             flags,
             timestamp,
             user_id,
+            checksum,
             context,
             command,
         })

server/src/state/entry.rs

@@ -117,6 +117,7 @@ impl State for FileState {
             let flags = reader.read_u64_le().await?;
             let timestamp = IggyTimestamp::from(reader.read_u64_le().await?);
             let user_id = reader.read_u32_le().await?;
+            let checksum = reader.read_u32_le().await?;
             let context_length = reader.read_u32_le().await? as usize;
             let mut context = BytesMut::with_capacity(context_length);
             context.put_bytes(0, context_length);
@@ -143,6 +144,14 @@ impl State for FileState {
                 command,
             );
             debug!("Read state entry: {entry}");
+            if entry.checksum != checksum {
+                return Err(IggyError::InvalidStateEntryChecksum(
+                    entry.checksum,
+                    checksum,
+                    entry.index,
+                ));
+            }
+
             entries.push(entry);
             total_size += 8
                 + 8
@@ -152,6 +161,7 @@ impl State for FileState {
                 + 8
                 + 4
                 + 4
+                + 4
                 + context_length as u64
                 + 4
                 + 4
@@ -167,22 +177,21 @@ impl State for FileState {
 
     async fn apply(&self, user_id: u32, command: EntryCommand) -> Result<(), IggyError> {
         debug!("Applying state entry with command: {command}, user ID: {user_id}");
-        let entry = StateEntry {
-            index: if self.entries_count.load(Ordering::SeqCst) == 0 {
+        let entry = StateEntry::new(
+            if self.entries_count.load(Ordering::SeqCst) == 0 {
                 0
             } else {
                 self.current_index.fetch_add(1, Ordering::SeqCst) + 1
             },
-            term: self.term.load(Ordering::SeqCst),
-            leader_id: self.current_leader.load(Ordering::SeqCst),
-            version: self.version,
-            flags: 0,
-            timestamp: IggyTimestamp::now(),
+            self.term.load(Ordering::SeqCst),
+            self.current_leader.load(Ordering::SeqCst),
+            self.version,
+            0,
+            IggyTimestamp::now(),
             user_id,
+            Bytes::new(),
             command,
-            context: Bytes::new(),
-        };
-
+        );
         self.entries_count.fetch_add(1, Ordering::SeqCst);
         self.persister.append(&self.path, &entry.to_bytes()).await?;
         debug!("Applied state entry: {entry}");