fix: comprehensive security audit hardening

crates/ember-cli/src/format.rs

@@ -20,13 +20,40 @@ pub fn format_response(frame: &Frame) -> String {
     format_frame(frame, 0)
 }
 
+/// Strips ANSI escape sequences and other control characters from
+/// server-supplied strings to prevent terminal manipulation attacks.
+/// Retains printable ASCII, tabs, and newlines (CR/LF).
+fn sanitize(s: &str) -> String {
+    let mut out = String::with_capacity(s.len());
+    let mut chars = s.chars();
+    while let Some(ch) = chars.next() {
+        if ch == '\x1b' {
+            // skip the ESC and the rest of the ANSI sequence
+            if let Some(next) = chars.next() {
+                if next == '[' {
+                    // CSI sequence — consume until a letter
+                    for c in chars.by_ref() {
+                        if c.is_ascii_alphabetic() {
+                            break;
+                        }
+                    }
+                }
+                // else: single-char escape, already consumed
+            }
+        } else if ch == '\t' || ch == '\n' || ch == '\r' || !ch.is_control() {
+            out.push(ch);
+        }
+    }
+    out
+}
+
 fn format_frame(frame: &Frame, indent: usize) -> String {
     let prefix = " ".repeat(indent);
 
     match frame {
-        Frame::Simple(s) => format!("{prefix}{}", s.green()),
+        Frame::Simple(s) => format!("{prefix}{}", sanitize(s).green()),
 
-        Frame::Error(e) => format!("{prefix}{} {}", "(error)".red(), e.red()),
+        Frame::Error(e) => format!("{prefix}{} {}", "(error)".red(), sanitize(e).red()),
 
         Frame::Integer(n) => format!(
             "{prefix}{} {}",
@@ -38,9 +65,9 @@ fn format_frame(frame: &Frame, indent: usize) -> String {
             match std::str::from_utf8(data) {
                 Ok(s) if s.contains("\r\n") || s.contains('\n') => {
                     // multiline output (like INFO) — print unquoted
-                    format!("{prefix}{}", s.green())
+                    format!("{prefix}{}", sanitize(s).green())
                 }
-                Ok(s) => format!("{prefix}{}", format!("\"{}\"", s).green()),
+                Ok(s) => format!("{prefix}{}", format!("\"{}\"", sanitize(s)).green()),
                 Err(_) => {
                     // binary data — show as hex
                     let hex: String = data.iter().map(|b| format!("{b:02x}")).collect();
@@ -187,4 +214,25 @@ mod tests {
         });
         assert_eq!(out, "1) key => (integer) 1");
     }
+
+    #[test]
+    fn sanitize_strips_ansi_escapes() {
+        assert_eq!(sanitize("hello\x1b[31mworld\x1b[0m"), "helloworld");
+    }
+
+    #[test]
+    fn sanitize_strips_control_chars() {
+        assert_eq!(sanitize("hello\x07\x08world"), "helloworld");
+    }
+
+    #[test]
+    fn sanitize_preserves_tabs_and_newlines() {
+        assert_eq!(sanitize("line1\nline2\ttab"), "line1\nline2\ttab");
+    }
+
+    #[test]
+    fn sanitize_server_response_with_escape() {
+        let out = no_color(|| format_response(&Frame::Simple("\x1b[31mfake-error\x1b[0m".into())));
+        assert_eq!(out, "fake-error");
+    }
 }

crates/ember-cli/src/repl.rs

@@ -87,7 +87,14 @@ pub fn run_repl(host: &str, port: u16, password: Option<&str>, tls: Option<&TlsC
                     continue;
                 }
 
-                let _ = rl.add_history_entry(trimmed);
+                // don't save AUTH commands to history — they contain passwords
+                if !trimmed
+                    .split_whitespace()
+                    .next()
+                    .is_some_and(|w| w.eq_ignore_ascii_case("auth"))
+                {
+                    let _ = rl.add_history_entry(trimmed);
+                }
 
                 // handle local commands
                 let first_word = trimmed.split_whitespace().next().unwrap_or("");
@@ -157,6 +164,13 @@ pub fn run_repl(host: &str, port: u16, password: Option<&str>, tls: Option<&TlsC
 
     if let Some(ref path) = history_path {
         let _ = rl.save_history(path);
+        // restrict history file permissions — it may contain key names
+        // and values from previous sessions
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::PermissionsExt;
+            let _ = std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600));
+        }
     }
 
     // graceful shutdown — send QUIT and close the TCP stream

crates/ember-cluster/src/gossip.rs

@@ -23,6 +23,11 @@ use tracing::{debug, info, trace, warn};
 use crate::message::{GossipMessage, MemberInfo, NodeUpdate};
 use crate::{NodeId, SlotRange};
 
+/// Maximum allowed incarnation value. Rejects gossip updates with
+/// incarnation numbers beyond this to prevent a malicious node from
+/// sending u64::MAX and permanently disabling suspicion refutation.
+const MAX_INCARNATION: u64 = u64::MAX / 2;
+
 /// Configuration for the gossip protocol.
 #[derive(Debug, Clone)]
 pub struct GossipConfig {
@@ -377,6 +382,13 @@ impl GossipEngine {
                     addr,
                     incarnation,
                 } => {
+                    if *incarnation > MAX_INCARNATION {
+                        warn!(
+                            "rejecting alive update for {} with excessive incarnation {}",
+                            node, incarnation
+                        );
+                        continue;
+                    }
                     if *node == self.local_id {
                         // Someone thinks we're alive, good
                         continue;
@@ -419,6 +431,13 @@ impl GossipEngine {
                 }
 
                 NodeUpdate::Suspect { node, incarnation } => {
+                    if *incarnation > MAX_INCARNATION {
+                        warn!(
+                            "rejecting suspect update for {} with excessive incarnation {}",
+                            node, incarnation
+                        );
+                        continue;
+                    }
                     if *node == self.local_id {
                         // Refute suspicion by incrementing our incarnation
                         if *incarnation >= self.incarnation {
@@ -445,6 +464,13 @@ impl GossipEngine {
                 }
 
                 NodeUpdate::Dead { node, incarnation } => {
+                    if *incarnation > MAX_INCARNATION {
+                        warn!(
+                            "rejecting dead update for {} with excessive incarnation {}",
+                            node, incarnation
+                        );
+                        continue;
+                    }
                     if *node == self.local_id {
                         // Refute death claim
                         self.incarnation = incarnation.saturating_add(1);

crates/ember-cluster/src/message.rs

@@ -170,8 +170,9 @@ impl GossipMessage {
             GossipMessage::Welcome { sender, members } => {
                 buf.put_u8(MSG_WELCOME);
                 encode_node_id(buf, sender);
-                buf.put_u16_le(members.len() as u16);
-                for member in members {
+                let count = members.len().min(MAX_COLLECTION_COUNT);
+                buf.put_u16_le(count as u16);
+                for member in &members[..count] {
                     encode_member_info(buf, member);
                 }
             }
@@ -324,8 +325,9 @@ fn decode_socket_addr(buf: &mut &[u8]) -> io::Result<SocketAddr> {
 }
 
 fn encode_updates(buf: &mut BytesMut, updates: &[NodeUpdate]) {
-    buf.put_u16_le(updates.len() as u16);
-    for update in updates {
+    let count = updates.len().min(MAX_COLLECTION_COUNT);
+    buf.put_u16_le(count as u16);
+    for update in &updates[..count] {
         encode_update(buf, update);
     }
 }
@@ -413,8 +415,9 @@ fn encode_member_info(buf: &mut BytesMut, member: &MemberInfo) {
     encode_socket_addr(buf, &member.addr);
     buf.put_u64_le(member.incarnation);
     buf.put_u8(if member.is_primary { 1 } else { 0 });
-    buf.put_u16_le(member.slots.len() as u16);
-    for slot in &member.slots {
+    let slot_count = member.slots.len().min(MAX_COLLECTION_COUNT);
+    buf.put_u16_le(slot_count as u16);
+    for slot in &member.slots[..slot_count] {
         buf.put_u16_le(slot.start);
         buf.put_u16_le(slot.end);
     }

crates/ember-cluster/src/raft.rs

@@ -18,6 +18,7 @@ use openraft::{
 use serde::{Deserialize, Serialize};
 use tokio::sync::RwLock;
 
+use crate::slots::SLOT_COUNT;
 use crate::{NodeId, SlotRange};
 
 /// Type configuration for openraft.
@@ -178,6 +179,17 @@ impl Storage {
             }
 
             ClusterCommand::AssignSlots { node_id, slots } => {
+                // validate all slot ranges before applying
+                for range in slots {
+                    if range.start > range.end || range.end >= SLOT_COUNT {
+                        return ClusterResponse::Error(format!(
+                            "invalid slot range {}..={} (max {})",
+                            range.start,
+                            range.end,
+                            SLOT_COUNT - 1
+                        ));
+                    }
+                }
                 let key = node_id.0.to_string();
                 if let Some(node) = state.nodes.get_mut(&key) {
                     node.slots = slots.clone();
@@ -203,6 +215,12 @@ impl Storage {
             }
 
             ClusterCommand::BeginMigration { slot, from, to } => {
+                if *slot >= SLOT_COUNT {
+                    return ClusterResponse::Error(format!(
+                        "slot {slot} out of range (max {})",
+                        SLOT_COUNT - 1
+                    ));
+                }
                 state.migrations.insert(
                     *slot,
                     MigrationState {
@@ -214,6 +232,11 @@ impl Storage {
             }
 
             ClusterCommand::CompleteMigration { slot, new_owner } => {
+                if !state.migrations.contains_key(slot) {
+                    return ClusterResponse::Error(format!(
+                        "no migration in progress for slot {slot}"
+                    ));
+                }
                 state.migrations.remove(slot);
                 let key = new_owner.0.to_string();
                 state.slots.insert(*slot, key);
@@ -564,6 +587,112 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn assign_slots_rejects_invalid_range() {
+        let storage = Arc::new(Storage::new());
+        let mut s = Arc::clone(&storage);
+
+        let node_id = NodeId::new();
+        let add = Entry {
+            log_id: log_id(1, 1),
+            payload: EntryPayload::Normal(ClusterCommand::AddNode {
+                node_id,
+                raft_id: 1,
+                addr: "127.0.0.1:6379".into(),
+                is_primary: true,
+            }),
+        };
+        s.apply_to_state_machine(&[add]).await.unwrap();
+
+        // craft a SlotRange with start > end (bypassing SlotRange::new)
+        let bad_range = SlotRange {
+            start: 100,
+            end: 50,
+        };
+        let assign = Entry {
+            log_id: log_id(1, 2),
+            payload: EntryPayload::Normal(ClusterCommand::AssignSlots {
+                node_id,
+                slots: vec![bad_range],
+            }),
+        };
+        let results = s.apply_to_state_machine(&[assign]).await.unwrap();
+        assert!(
+            matches!(&results[0], ClusterResponse::Error(msg) if msg.contains("invalid slot range"))
+        );
+    }
+
+    #[tokio::test]
+    async fn assign_slots_rejects_out_of_range() {
+        let storage = Arc::new(Storage::new());
+        let mut s = Arc::clone(&storage);
+
+        let node_id = NodeId::new();
+        let add = Entry {
+            log_id: log_id(1, 1),
+            payload: EntryPayload::Normal(ClusterCommand::AddNode {
+                node_id,
+                raft_id: 1,
+                addr: "127.0.0.1:6379".into(),
+                is_primary: true,
+            }),
+        };
+        s.apply_to_state_machine(&[add]).await.unwrap();
+
+        // slot end >= SLOT_COUNT
+        let bad_range = SlotRange {
+            start: 0,
+            end: 16384,
+        };
+        let assign = Entry {
+            log_id: log_id(1, 2),
+            payload: EntryPayload::Normal(ClusterCommand::AssignSlots {
+                node_id,
+                slots: vec![bad_range],
+            }),
+        };
+        let results = s.apply_to_state_machine(&[assign]).await.unwrap();
+        assert!(
+            matches!(&results[0], ClusterResponse::Error(msg) if msg.contains("invalid slot range"))
+        );
+    }
+
+    #[tokio::test]
+    async fn complete_migration_without_begin_errors() {
+        let storage = Arc::new(Storage::new());
+        let mut s = Arc::clone(&storage);
+
+        let node_id = NodeId::new();
+        let complete = Entry {
+            log_id: log_id(1, 1),
+            payload: EntryPayload::Normal(ClusterCommand::CompleteMigration {
+                slot: 100,
+                new_owner: node_id,
+            }),
+        };
+        let results = s.apply_to_state_machine(&[complete]).await.unwrap();
+        assert!(matches!(&results[0], ClusterResponse::Error(msg) if msg.contains("no migration")));
+    }
+
+    #[tokio::test]
+    async fn begin_migration_rejects_invalid_slot() {
+        let storage = Arc::new(Storage::new());
+        let mut s = Arc::clone(&storage);
+
+        let node1 = NodeId::new();
+        let node2 = NodeId::new();
+        let begin = Entry {
+            log_id: log_id(1, 1),
+            payload: EntryPayload::Normal(ClusterCommand::BeginMigration {
+                slot: 16384,
+                from: node1,
+                to: node2,
+            }),
+        };
+        let results = s.apply_to_state_machine(&[begin]).await.unwrap();
+        assert!(matches!(&results[0], ClusterResponse::Error(msg) if msg.contains("out of range")));
+    }
+
     #[tokio::test]
     async fn storage_log_operations() {
         let storage = Arc::new(Storage::new());

crates/ember-cluster/src/slots.rs

@@ -114,10 +114,10 @@ impl SlotRange {
     ///
     /// # Panics
     ///
-    /// Debug-panics if `start > end` or if `end >= SLOT_COUNT`.
+    /// Panics if `start > end` or if `end >= SLOT_COUNT`.
     pub fn new(start: u16, end: u16) -> Self {
-        debug_assert!(start <= end, "SlotRange requires start <= end");
-        debug_assert!(end < SLOT_COUNT, "slot must be < {SLOT_COUNT}");
+        assert!(start <= end, "SlotRange requires start <= end");
+        assert!(end < SLOT_COUNT, "slot must be < {SLOT_COUNT}");
         Self { start, end }
     }
 

crates/ember-core/src/engine.rs

@@ -56,6 +56,10 @@ impl Engine {
     /// Panics if `shard_count` is zero.
     pub fn with_config(shard_count: usize, config: EngineConfig) -> Self {
         assert!(shard_count > 0, "shard count must be at least 1");
+        assert!(
+            shard_count <= u16::MAX as usize,
+            "shard count must fit in u16"
+        );
 
         let drop_handle = DropHandle::spawn();