harden server — startup, shutdown, overflow guards

crates/ember-cluster/src/message.rs

@@ -403,7 +403,7 @@ fn decode_member_info(buf: &mut &[u8]) -> io::Result<MemberInfo> {
     for _ in 0..slot_count {
         let start = buf.get_u16_le();
         let end = buf.get_u16_le();
-        slots.push(SlotRange::new(start, end));
+        slots.push(SlotRange::try_new(start, end)?);
     }
     Ok(MemberInfo {
         id,
@@ -579,4 +579,42 @@ mod tests {
         let result = GossipMessage::decode(&[255]);
         assert!(result.is_err());
     }
+
+    #[test]
+    fn invalid_slot_range_in_welcome_rejected() {
+        // craft a Welcome message with an invalid slot range (start > end)
+        let mut buf = BytesMut::new();
+        buf.put_u8(MSG_WELCOME);
+        encode_node_id(&mut buf, &NodeId::new());
+        buf.put_u16_le(1); // 1 member
+        encode_node_id(&mut buf, &NodeId::new());
+        encode_socket_addr(&mut buf, &test_addr());
+        buf.put_u64_le(1); // incarnation
+        buf.put_u8(1); // is_primary
+        buf.put_u16_le(1); // 1 slot range
+        buf.put_u16_le(5000); // start
+        buf.put_u16_le(100); // end < start — invalid
+
+        let result = GossipMessage::decode(&buf);
+        assert!(result.is_err(), "should reject inverted slot range");
+    }
+
+    #[test]
+    fn out_of_range_slot_in_welcome_rejected() {
+        // craft a Welcome message with a slot >= 16384
+        let mut buf = BytesMut::new();
+        buf.put_u8(MSG_WELCOME);
+        encode_node_id(&mut buf, &NodeId::new());
+        buf.put_u16_le(1);
+        encode_node_id(&mut buf, &NodeId::new());
+        encode_socket_addr(&mut buf, &test_addr());
+        buf.put_u64_le(1);
+        buf.put_u8(1);
+        buf.put_u16_le(1); // 1 slot range
+        buf.put_u16_le(0);
+        buf.put_u16_le(16384); // out of range
+
+        let result = GossipMessage::decode(&buf);
+        assert!(result.is_err(), "should reject slot >= 16384");
+    }
 }

crates/ember-cluster/src/slots.rs

@@ -121,6 +121,26 @@ impl SlotRange {
         Self { start, end }
     }
 
+    /// Creates a new slot range with runtime validation.
+    ///
+    /// Returns an error if `start > end` or `end >= SLOT_COUNT`.
+    /// Use this for untrusted input (e.g. network-decoded data).
+    pub fn try_new(start: u16, end: u16) -> Result<Self, std::io::Error> {
+        if start > end {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                format!("SlotRange requires start <= end, got {start}..{end}"),
+            ));
+        }
+        if end >= SLOT_COUNT {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                format!("slot {end} out of range (max {})", SLOT_COUNT - 1),
+            ));
+        }
+        Ok(Self { start, end })
+    }
+
     /// Creates a range containing a single slot.
     pub fn single(slot: u16) -> Self {
         Self::new(slot, slot)
@@ -391,6 +411,17 @@ mod tests {
         assert_eq!(map.unassigned_count(), 1);
     }
 
+    #[test]
+    fn slot_range_try_new_validates() {
+        assert!(SlotRange::try_new(0, 5460).is_ok());
+        assert!(SlotRange::try_new(100, 100).is_ok());
+        // start > end
+        assert!(SlotRange::try_new(5000, 100).is_err());
+        // end >= SLOT_COUNT
+        assert!(SlotRange::try_new(0, 16384).is_err());
+        assert!(SlotRange::try_new(0, u16::MAX).is_err());
+    }
+
     #[test]
     fn slots_for_node_ranges() {
         let node = NodeId(Uuid::new_v4());

crates/ember-core/src/dropper.rs

@@ -42,16 +42,23 @@ pub struct DropHandle {
 
 impl DropHandle {
     /// Spawns the background drop thread and returns a handle.
+    ///
+    /// If the thread fails to spawn (resource exhaustion), logs a warning
+    /// and returns a handle that drops everything inline. The channel
+    /// disconnects immediately since the receiver is never started, and
+    /// `try_send` gracefully falls back to inline dropping.
     pub fn spawn() -> Self {
         let (tx, rx) = mpsc::sync_channel::<Droppable>(DROP_CHANNEL_CAPACITY);
 
-        std::thread::Builder::new()
+        if let Err(e) = std::thread::Builder::new()
             .name("ember-drop".into())
             .spawn(move || {
                 // just drain the channel — dropping each item frees the memory
                 while rx.recv().is_ok() {}
             })
-            .expect("failed to spawn drop thread");
+        {
+            tracing::warn!("failed to spawn drop thread, large values will be freed inline: {e}");
+        }
 
         Self { tx }
     }

crates/ember-persistence/src/recovery.rs

@@ -259,7 +259,7 @@ fn replay_aof(
             }
             AofRecord::Expire { key, seconds } => {
                 if let Some(entry) = map.get_mut(&key) {
-                    entry.1 = (seconds * 1000) as i64;
+                    entry.1 = seconds.saturating_mul(1000).min(i64::MAX as u64) as i64;
                 }
             }
             AofRecord::LPush { key, values } => {
@@ -348,7 +348,7 @@ fn replay_aof(
             }
             AofRecord::Pexpire { key, milliseconds } => {
                 if let Some(entry) = map.get_mut(&key) {
-                    entry.1 = milliseconds as i64;
+                    entry.1 = milliseconds.min(i64::MAX as u64) as i64;
                 }
             }
             AofRecord::Incr { key } => {

crates/ember-protocol/src/error.rs

@@ -33,4 +33,8 @@ pub enum ProtocolError {
     /// Command received with the wrong number of arguments.
     #[error("wrong number of arguments for '{0}' command")]
     WrongArity(String),
+
+    /// The frame exceeds the maximum nesting depth.
+    #[error("frame nesting depth exceeds limit of {0}")]
+    NestingTooDeep(usize),
 }

crates/ember-protocol/src/parse.rs

@@ -14,6 +14,10 @@ use bytes::Bytes;
 use crate::error::ProtocolError;
 use crate::types::Frame;
 
+/// Maximum nesting depth for arrays and maps. Prevents stack overflow
+/// from malicious or malformed deeply-nested frames.
+const MAX_NESTING_DEPTH: usize = 64;
+
 /// Checks whether `buf` contains a complete RESP3 frame and parses it.
 ///
 /// Returns `Ok(Some(frame))` if a complete frame was parsed,
@@ -26,11 +30,11 @@ pub fn parse_frame(buf: &[u8]) -> Result<Option<(Frame, usize)>, ProtocolError>
 
     let mut cursor = Cursor::new(buf);
 
-    match check(&mut cursor) {
+    match check(&mut cursor, 0) {
         Ok(()) => {
             // we know a complete frame exists — reset and parse it
             cursor.set_position(0);
-            let frame = parse(&mut cursor)?;
+            let frame = parse(&mut cursor, 0)?;
             let consumed = cursor.position() as usize;
             Ok(Some((frame, consumed)))
         }
@@ -45,16 +49,16 @@ pub fn parse_frame(buf: &[u8]) -> Result<Option<(Frame, usize)>, ProtocolError>
 
 /// Peeks through the buffer to verify a complete frame is present.
 /// Advances the cursor past the frame on success.
-fn check(cursor: &mut Cursor<&[u8]>) -> Result<(), ProtocolError> {
+fn check(cursor: &mut Cursor<&[u8]>, depth: usize) -> Result<(), ProtocolError> {
     let prefix = read_byte(cursor)?;
 
     match prefix {
         b'+' | b'-' => check_line(cursor),
         b':' => check_line(cursor),
         b'$' => check_bulk(cursor),
-        b'*' => check_array(cursor),
+        b'*' => check_array(cursor, depth),
         b'_' => check_line(cursor),
-        b'%' => check_map(cursor),
+        b'%' => check_map(cursor, depth),
         other => Err(ProtocolError::InvalidPrefix(other)),
     }
 }
@@ -88,27 +92,37 @@ fn check_bulk(cursor: &mut Cursor<&[u8]>) -> Result<(), ProtocolError> {
     Ok(())
 }
 
-fn check_array(cursor: &mut Cursor<&[u8]>) -> Result<(), ProtocolError> {
+fn check_array(cursor: &mut Cursor<&[u8]>, depth: usize) -> Result<(), ProtocolError> {
+    let next_depth = depth + 1;
+    if next_depth > MAX_NESTING_DEPTH {
+        return Err(ProtocolError::NestingTooDeep(MAX_NESTING_DEPTH));
+    }
+
     let count = read_integer_line(cursor)?;
     if count < 0 {
         return Err(ProtocolError::InvalidFrameLength(count));
     }
 
     for _ in 0..count {
-        check(cursor)?;
+        check(cursor, next_depth)?;
     }
     Ok(())
 }
 
-fn check_map(cursor: &mut Cursor<&[u8]>) -> Result<(), ProtocolError> {
+fn check_map(cursor: &mut Cursor<&[u8]>, depth: usize) -> Result<(), ProtocolError> {
+    let next_depth = depth + 1;
+    if next_depth > MAX_NESTING_DEPTH {
+        return Err(ProtocolError::NestingTooDeep(MAX_NESTING_DEPTH));
+    }
+
     let count = read_integer_line(cursor)?;
     if count < 0 {
         return Err(ProtocolError::InvalidFrameLength(count));
     }
 
     for _ in 0..count {
-        check(cursor)?; // key
-        check(cursor)?; // value
+        check(cursor, next_depth)?; // key
+        check(cursor, next_depth)?; // value
     }
     Ok(())
 }
@@ -117,7 +131,7 @@ fn check_map(cursor: &mut Cursor<&[u8]>) -> Result<(), ProtocolError> {
 // parse: actually builds Frame values (only called after check succeeds)
 // ---------------------------------------------------------------------------
 
-fn parse(cursor: &mut Cursor<&[u8]>) -> Result<Frame, ProtocolError> {
+fn parse(cursor: &mut Cursor<&[u8]>, depth: usize) -> Result<Frame, ProtocolError> {
     let prefix = read_byte(cursor)?;
 
     match prefix {
@@ -147,10 +161,11 @@ fn parse(cursor: &mut Cursor<&[u8]>) -> Result<Frame, ProtocolError> {
             Ok(Frame::Bulk(Bytes::copy_from_slice(data)))
         }
         b'*' => {
+            let next_depth = depth + 1;
             let count = read_integer_line(cursor)? as usize;
             let mut frames = Vec::with_capacity(count);
             for _ in 0..count {
-                frames.push(parse(cursor)?);
+                frames.push(parse(cursor, next_depth)?);
             }
             Ok(Frame::Array(frames))
         }
@@ -160,11 +175,12 @@ fn parse(cursor: &mut Cursor<&[u8]>) -> Result<Frame, ProtocolError> {
             Ok(Frame::Null)
         }
         b'%' => {
+            let next_depth = depth + 1;
             let count = read_integer_line(cursor)? as usize;
             let mut pairs = Vec::with_capacity(count);
             for _ in 0..count {
-                let key = parse(cursor)?;
-                let val = parse(cursor)?;
+                let key = parse(cursor, next_depth)?;
+                let val = parse(cursor, next_depth)?;
                 pairs.push((key, val));
             }
             Ok(Frame::Map(pairs))
@@ -396,4 +412,34 @@ mod tests {
         assert_eq!(frame, Frame::Simple("OK".into()));
         assert_eq!(consumed, 5);
     }
+
+    #[test]
+    fn deeply_nested_array_rejected() {
+        // build a frame nested 65 levels deep (exceeds MAX_NESTING_DEPTH of 64)
+        let mut buf = Vec::new();
+        for _ in 0..65 {
+            buf.extend_from_slice(b"*1\r\n");
+        }
+        buf.extend_from_slice(b":1\r\n"); // leaf value
+
+        let err = parse_frame(&buf).unwrap_err();
+        assert!(
+            matches!(err, ProtocolError::NestingTooDeep(64)),
+            "expected NestingTooDeep, got {err:?}"
+        );
+    }
+
+    #[test]
+    fn nesting_at_limit_accepted() {
+        // exactly 64 levels deep — should succeed
+        let mut buf = Vec::new();
+        for _ in 0..64 {
+            buf.extend_from_slice(b"*1\r\n");
+        }
+        buf.extend_from_slice(b":1\r\n");
+
+        let result = parse_frame(&buf);
+        assert!(result.is_ok(), "64 levels of nesting should be accepted");
+        assert!(result.unwrap().is_some());
+    }
 }

crates/ember-server/src/main.rs

@@ -361,10 +361,7 @@ async fn main() {
         let tls_addr: SocketAddr = match format!("{}:{}", args.host, tls_port).parse() {
             Ok(a) => a,
             Err(e) => {
-                eprintln!(
-                    "invalid TLS bind address '{}:{tls_port}': {e}",
-                    args.host
-                );
+                eprintln!("invalid TLS bind address '{}:{tls_port}': {e}", args.host);
                 std::process::exit(1);
             }
         };