harden ember-server — auth rate limiting, pubsub limits, integer safety

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 =

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))]),
                     ])
                 })

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`

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<Semaphore>, 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"),