diff --git a/crates/ember-server/src/concurrent_handler.rs b/crates/ember-server/src/concurrent_handler.rs index a84c4acf..79f247bb 100644 --- a/crates/ember-server/src/concurrent_handler.rs +++ b/crates/ember-server/src/concurrent_handler.rs @@ -25,7 +25,8 @@ use subtle::ConstantTimeEq; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use crate::connection_common::{ - is_allowed_before_auth, is_auth_frame, try_auth, BUF_CAPACITY, IDLE_TIMEOUT, MAX_BUF_SIZE, + is_allowed_before_auth, is_auth_frame, try_auth, BUF_CAPACITY, IDLE_TIMEOUT, MAX_AUTH_FAILURES, + MAX_BUF_SIZE, }; use crate::pubsub::PubSubManager; use crate::server::ServerContext; @@ -47,6 +48,7 @@ where S: AsyncRead + AsyncWrite + Unpin, { let mut authenticated = ctx.requirepass.is_none(); + let mut auth_failures: u32 = 0; let mut buf = BytesMut::with_capacity(BUF_CAPACITY); let mut out = BytesMut::with_capacity(BUF_CAPACITY); @@ -79,6 +81,16 @@ where response.serialize(&mut out); if success { authenticated = true; + } else { + auth_failures += 1; + if auth_failures >= MAX_AUTH_FAILURES { + Frame::Error( + "ERR too many AUTH failures, closing connection".into(), + ) + .serialize(&mut out); + let _ = stream.write_all(&out).await; + return Ok(()); + } } } else if is_allowed_before_auth(&frame) { let response = diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs index 4e015c15..def8f565 100644 --- a/crates/ember-server/src/connection.rs +++ b/crates/ember-server/src/connection.rs @@ -19,7 +19,8 @@ use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tokio::sync::broadcast; use crate::connection_common::{ - is_allowed_before_auth, is_auth_frame, try_auth, BUF_CAPACITY, IDLE_TIMEOUT, MAX_BUF_SIZE, + is_allowed_before_auth, is_auth_frame, try_auth, BUF_CAPACITY, IDLE_TIMEOUT, MAX_AUTH_FAILURES, + MAX_BUF_SIZE, MAX_PATTERN_LEN, MAX_SUBSCRIPTIONS_PER_CONN, }; use crate::pubsub::{PubMessage, PubSubManager}; use crate::server::ServerContext; @@ -45,6 +46,7 @@ where { // per-connection auth state. auto-authenticated when no password is set. let mut authenticated = ctx.requirepass.is_none(); + let mut auth_failures: u32 = 0; let mut buf = BytesMut::with_capacity(BUF_CAPACITY); let mut out = BytesMut::with_capacity(BUF_CAPACITY); @@ -98,6 +100,14 @@ where response.serialize(&mut out); if success { authenticated = true; + } else { + auth_failures += 1; + if auth_failures >= MAX_AUTH_FAILURES { + Frame::Error("ERR too many AUTH failures, closing connection".into()) + .serialize(&mut out); + let _ = stream.write_all(&out).await; + return Ok(()); + } } } else if is_allowed_before_auth(&frame) { let response = process(frame, &engine, ctx, slow_log, pubsub).await; @@ -312,6 +322,12 @@ fn handle_sub_command( match cmd { Command::Subscribe { channels } => { for ch in channels { + let total = channel_rxs.len() + pattern_rxs.len(); + if total >= MAX_SUBSCRIPTIONS_PER_CONN { + Frame::Error("ERR max subscriptions per connection reached".into()) + .serialize(out); + continue; + } let rx = pubsub.subscribe(&ch); channel_rxs.insert(ch.clone(), rx); let count = channel_rxs.len() + pattern_rxs.len(); @@ -343,6 +359,21 @@ fn handle_sub_command( } Command::PSubscribe { patterns } => { for pat in patterns { + if pat.len() > MAX_PATTERN_LEN { + Frame::Error(format!( + "ERR pattern too long ({} bytes, max {})", + pat.len(), + MAX_PATTERN_LEN + )) + .serialize(out); + continue; + } + let total = channel_rxs.len() + pattern_rxs.len(); + if total >= MAX_SUBSCRIPTIONS_PER_CONN { + Frame::Error("ERR max subscriptions per connection reached".into()) + .serialize(out); + continue; + } let rx = pubsub.psubscribe(&pat); pattern_rxs.insert(pat.clone(), rx); let count = channel_rxs.len() + pattern_rxs.len(); @@ -1602,9 +1633,9 @@ async fn execute( .unwrap_or_default() .as_secs(); Frame::Array(vec![ - Frame::Integer(e.id as i64), - Frame::Integer(ts as i64), - Frame::Integer(e.duration.as_micros() as i64), + Frame::Integer(e.id.min(i64::MAX as u64) as i64), + Frame::Integer(ts.min(i64::MAX as u64) as i64), + Frame::Integer(e.duration.as_micros().min(i64::MAX as u128) as i64), Frame::Array(vec![Frame::Bulk(Bytes::from(e.command))]), ]) }) diff --git a/crates/ember-server/src/connection_common.rs b/crates/ember-server/src/connection_common.rs index abc2d5e8..31f9e3d1 100644 --- a/crates/ember-server/src/connection_common.rs +++ b/crates/ember-server/src/connection_common.rs @@ -27,6 +27,20 @@ pub const MAX_BUF_SIZE: usize = 64 * 1024 * 1024; /// 5 minutes matches Redis default behavior. pub const IDLE_TIMEOUT: Duration = Duration::from_secs(300); +/// Maximum number of failed AUTH attempts before the connection is closed. +/// Prevents brute-force password guessing. Matches Redis 6.2+ behavior. +pub const MAX_AUTH_FAILURES: u32 = 10; + +/// Maximum number of pub/sub subscriptions a single connection can hold. +/// Prevents a malicious client from exhausting memory with thousands of +/// broadcast channels. 10,000 is generous for legitimate use cases. +pub const MAX_SUBSCRIPTIONS_PER_CONN: usize = 10_000; + +/// Maximum length of a PSUBSCRIBE pattern string. Very long patterns can +/// cause pathological backtracking in glob matching. 256 bytes covers any +/// reasonable channel naming scheme. +pub const MAX_PATTERN_LEN: usize = 256; + /// Checks if a raw frame is an AUTH command (before full parsing). /// /// Peeks at the first bulk element to avoid a full `Command::from_frame` diff --git a/crates/ember-server/src/server.rs b/crates/ember-server/src/server.rs index 62624e2b..7983f15c 100644 --- a/crates/ember-server/src/server.rs +++ b/crates/ember-server/src/server.rs @@ -284,7 +284,8 @@ fn on_connection_done(ctx: &ServerContext) { /// Waits for all connections to drain with a 30-second timeout. async fn drain_connections(semaphore: &Arc, max_conn: usize) { info!("waiting for active connections to close..."); - let drain = semaphore.acquire_many(max_conn as u32); + let max_conn_u32 = u32::try_from(max_conn).unwrap_or(u32::MAX); + let drain = semaphore.acquire_many(max_conn_u32); match tokio::time::timeout(Duration::from_secs(30), drain).await { Ok(_) => info!("all connections drained, shutting down"), Err(_) => warn!("shutdown timeout after 30s, forcing exit"),