From 88b2fe591b3f1de8360237effc86b0361c806125 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Wed, 11 Feb 2026 19:11:38 -0500 Subject: [PATCH 1/5] fix: add nesting depth limit to RESP3 parser the recursive check/parse functions had no depth limit, allowing a malicious client to stack-overflow the server with deeply nested arrays or maps. adds a MAX_NESTING_DEPTH of 64 and threads a depth counter through both check() and parse(). returns NestingTooDeep error when exceeded. --- crates/ember-protocol/src/error.rs | 4 ++ crates/ember-protocol/src/parse.rs | 74 ++++++++++++++++++++++++------ 2 files changed, 64 insertions(+), 14 deletions(-) diff --git a/crates/ember-protocol/src/error.rs b/crates/ember-protocol/src/error.rs index 1552ec28..50e602e9 100644 --- a/crates/ember-protocol/src/error.rs +++ b/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), } diff --git a/crates/ember-protocol/src/parse.rs b/crates/ember-protocol/src/parse.rs index 416b593d..e12eeef3 100644 --- a/crates/ember-protocol/src/parse.rs +++ b/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, 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, 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 { +fn parse(cursor: &mut Cursor<&[u8]>, depth: usize) -> Result { let prefix = read_byte(cursor)?; match prefix { @@ -147,10 +161,11 @@ fn parse(cursor: &mut Cursor<&[u8]>) -> Result { 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 { 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()); + } } From 562f1f59bc080820a51d4a168b301a55fe8aff9f Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Wed, 11 Feb 2026 19:12:02 -0500 Subject: [PATCH 2/5] fix: clamp ttl arithmetic in aof recovery to prevent overflow expire replay computed (seconds * 1000) as i64 which silently wrapped for very large u64 values, and pexpire cast u64 directly to i64 which sign-corrupted values above i64::MAX. both now clamp to i64::MAX. --- crates/ember-persistence/src/recovery.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/ember-persistence/src/recovery.rs b/crates/ember-persistence/src/recovery.rs index 97e325b0..d8597bd8 100644 --- a/crates/ember-persistence/src/recovery.rs +++ b/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 } => { From d1ee0e79befafa24cc6a907092d122b32f1c51ea Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Wed, 11 Feb 2026 19:12:25 -0500 Subject: [PATCH 3/5] fix: graceful degradation when drop thread fails to spawn DropHandle::spawn() used .expect() which panics if the OS can't create the thread. now logs a warning and returns a handle that falls back to inline dropping via the existing try_send Disconnected handling. --- crates/ember-core/src/dropper.rs | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/crates/ember-core/src/dropper.rs b/crates/ember-core/src/dropper.rs index 50112d8f..51527010 100644 --- a/crates/ember-core/src/dropper.rs +++ b/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::(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 } } From 04a43d19c3b7fe7e90a3508a40935ecad2bcbe30 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Wed, 11 Feb 2026 19:13:11 -0500 Subject: [PATCH 4/5] fix: validate slot ranges at runtime in gossip decode path SlotRange::new() only checked invariants via debug_assert!, which is stripped in release builds. adds try_new() with runtime validation and uses it in decode_member_info() where data comes from the network. internal callers with known-valid ranges keep using new(). --- crates/ember-cluster/src/message.rs | 40 ++++++++++++++++++++++++++++- crates/ember-cluster/src/slots.rs | 31 ++++++++++++++++++++++ 2 files changed, 70 insertions(+), 1 deletion(-) diff --git a/crates/ember-cluster/src/message.rs b/crates/ember-cluster/src/message.rs index 8e39bd33..ff3120b7 100644 --- a/crates/ember-cluster/src/message.rs +++ b/crates/ember-cluster/src/message.rs @@ -403,7 +403,7 @@ fn decode_member_info(buf: &mut &[u8]) -> io::Result { 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"); + } } diff --git a/crates/ember-cluster/src/slots.rs b/crates/ember-cluster/src/slots.rs index 7f52147f..6b4f995f 100644 --- a/crates/ember-cluster/src/slots.rs +++ b/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 { + 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()); From d7da6093f4304b696b174a6dba2da47c1d509cf1 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Wed, 11 Feb 2026 19:14:15 -0500 Subject: [PATCH 5/5] chore: cargo fmt --- crates/ember-server/src/main.rs | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/crates/ember-server/src/main.rs b/crates/ember-server/src/main.rs index 18afb031..34a31030 100644 --- a/crates/ember-server/src/main.rs +++ b/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); } };