diff --git a/Cargo.lock b/Cargo.lock index 5cb4d1e..5f4e99c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -791,7 +791,7 @@ dependencies = [ [[package]] name = "rcache" -version = "0.11.5" +version = "0.11.6" dependencies = [ "bytes", "crc16", diff --git a/Cargo.toml b/Cargo.toml index 9245751..e8dcdcc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rcache" -version = "0.11.5" +version = "0.11.6" edition = "2024" [dependencies] diff --git a/src/command/acl.rs b/src/command/acl.rs index e4e59d8..aa43034 100644 --- a/src/command/acl.rs +++ b/src/command/acl.rs @@ -493,7 +493,7 @@ pub fn check_password(username: &str, password: &str) -> AuthOutcome { } use sha2::{Digest, Sha256}; let hash = format!("{:x}", Sha256::digest(password.as_bytes())); - if u.passwords.contains(&hash) { + if u.passwords.iter().any(|p| constant_time_eq(p.as_bytes(), hash.as_bytes())) { AuthOutcome::Ok } else { AuthOutcome::WrongPass @@ -505,6 +505,30 @@ pub fn check_password(username: &str, password: &str) -> AuthOutcome { } } +/// Constant-time byte-slice equality. The length comparison is unavoidable, but +/// content comparison does not short-circuit on the first differing byte, so it +/// leaks no per-byte timing signal. +pub fn constant_time_eq(a: &[u8], b: &[u8]) -> bool { + if a.len() != b.len() { + return false; + } + let mut diff = 0u8; + for (x, y) in a.iter().zip(b.iter()) { + diff |= x ^ y; + } + diff == 0 +} + +/// Constant-time verification of a plaintext secret (e.g. `requirepass`) by +/// comparing fixed-length SHA-256 digests, so neither the secret's length nor a +/// matching prefix leaks through response timing. +pub fn verify_secret(provided: &str, expected: &str) -> bool { + use sha2::{Digest, Sha256}; + let a = Sha256::digest(provided.as_bytes()); + let b = Sha256::digest(expected.as_bytes()); + constant_time_eq(&a, &b) +} + /// Whether `username` may run `cmd`. Unknown users are not blocked here (the /// connection layer gates unauthenticated access separately); the `default` /// user always exists. @@ -556,7 +580,13 @@ pub fn http_authenticate(token: &str) -> Option { let users = ACL_USERS.lock().ok()?; users .iter() - .find(|(_, u)| u.enabled && !u.no_pass && u.passwords.contains(&hash)) + .find(|(_, u)| { + u.enabled + && !u.no_pass + && u.passwords + .iter() + .any(|p| constant_time_eq(p.as_bytes(), hash.as_bytes())) + }) .map(|(name, _)| name.clone()) } @@ -579,3 +609,23 @@ static ACL_USERS: std::sync::LazyLock>> = ); Mutex::new(users) }); + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_constant_time_eq() { + assert!(constant_time_eq(b"abc", b"abc")); + assert!(!constant_time_eq(b"abc", b"abd")); + assert!(!constant_time_eq(b"abc", b"ab")); + assert!(constant_time_eq(b"", b"")); + } + + #[test] + fn test_verify_secret() { + assert!(verify_secret("hunter2", "hunter2")); + assert!(!verify_secret("hunter2", "hunter3")); + assert!(!verify_secret("", "x")); + } +} diff --git a/src/command/bitmap.rs b/src/command/bitmap.rs index a3ea99d..4c5b5b9 100644 --- a/src/command/bitmap.rs +++ b/src/command/bitmap.rs @@ -119,6 +119,12 @@ pub fn cmd_bitcount(ctx: &mut CommandContext) -> RespValue { let use_bit = ctx.args.len() > 4 && String::from_utf8_lossy(&ctx.args[4]).to_uppercase() == "BIT"; + // An empty value has no bits/bytes to count. Guard here so the range math + // below (`data.len() - 1` / `total_bits - 1`) can't underflow. + if data.is_empty() { + return RespValue::integer(0); + } + if use_bit { // BIT mode: start and end are bit offsets let total_bits = data.len() * 8; diff --git a/src/command/hash.rs b/src/command/hash.rs index 20df011..3a24c9c 100644 --- a/src/command/hash.rs +++ b/src/command/hash.rs @@ -4,6 +4,10 @@ use crate::protocol::RespValue; use crate::storage::RedisObject; use super::registry::CommandContext; +/// Upper bound on the magnitude of a negative HRANDFIELD `count` (the +/// with-duplicates form), preventing an unbounded allocation. +const MAX_RAND_COUNT: u64 = 1_000_000; + fn ensure_hash<'a>(ctx: &'a mut CommandContext, key: &Bytes) -> Result<&'a mut HashMap, RespValue> { match ctx.db().get_or_insert_with(key, || RedisObject::Hash(HashMap::new())) { RedisObject::Hash(h) => Ok(h), @@ -310,7 +314,12 @@ pub fn cmd_hrandfield(ctx: &mut CommandContext) -> RespValue { } } Some(n) => { - let n = (-n) as usize; + // Negative count allows duplicates; cap the magnitude and + // guard i64::MIN to avoid an unbounded allocation. + let n = match n.checked_neg() { + Some(v) if v as u64 <= MAX_RAND_COUNT => v as usize, + _ => return RespValue::error("ERR count value is out of range"), + }; use rand::seq::SliceRandom; let entries: Vec<(Bytes, Bytes)> = hash.iter().map(|(k, v)| (k.clone(), v.clone())).collect(); let mut items = Vec::new(); diff --git a/src/command/probabilistic.rs b/src/command/probabilistic.rs index d3d641a..65538dd 100644 --- a/src/command/probabilistic.rs +++ b/src/command/probabilistic.rs @@ -297,6 +297,13 @@ pub fn cmd_bf_reserve(ctx: &mut CommandContext) -> RespValue { _ => return RespValue::error("ERR (error) bad capacity"), }; + // Reject dimensions that would allocate more than the per-structure cap. + let num_bits = optimal_bits(capacity, error_rate).max(8); + let byte_count = (num_bits as usize + 7) / 8; + if byte_count > MAX_PROB_BYTES { + return RespValue::error("ERR Bloom filter dimensions too large"); + } + let db = ctx.db(); if db.exists(&key) { return RespValue::error("ERR item exists"); @@ -364,7 +371,18 @@ fn new_cms(width: u32, depth: u32) -> Vec { } fn is_cms(data: &[u8]) -> bool { - data.len() >= CMS_META_LEN && &data[0..CMS_HEADER_LEN] == CMS_HEADER + if data.len() < CMS_META_LEN || &data[0..CMS_HEADER_LEN] != CMS_HEADER { + return false; + } + // Reject a forged value whose declared dimensions don't fit the buffer. + // Without this, counter accessors index out of bounds (panic) and readers + // allocate from attacker-controlled dimensions. + let width = cms_width(data) as usize; + let depth = cms_depth(data) as usize; + match width.checked_mul(depth).and_then(|c| c.checked_mul(8)) { + Some(counter_bytes) => data.len() >= CMS_META_LEN + counter_bytes, + None => false, + } } fn cms_width(data: &[u8]) -> u32 { @@ -754,7 +772,17 @@ fn new_topk(k: u32, width: u32, depth: u32, decay: f64) -> Vec { } fn is_topk(data: &[u8]) -> bool { - data.len() >= TOPK_META_LEN && &data[0..TOPK_HEADER_LEN] == TOPK_HEADER + if data.len() < TOPK_META_LEN || &data[0..TOPK_HEADER_LEN] != TOPK_HEADER { + return false; + } + // Reject a forged value whose declared CMS dimensions don't fit the buffer, + // so `from_bytes` cannot read counters out of bounds. + let width = topk_width(data) as usize; + let depth = topk_depth(data) as usize; + match width.checked_mul(depth).and_then(|c| c.checked_mul(8)) { + Some(counter_bytes) => data.len() >= TOPK_META_LEN + counter_bytes, + None => false, + } } fn topk_k(data: &[u8]) -> u32 { @@ -810,7 +838,10 @@ impl TopKState { counters.push(val); } - let mut heap = Vec::with_capacity(num_items); + // Each heap entry needs at least 12 bytes (4-byte name length + 8-byte + // count), so bound the pre-allocation by what the buffer can actually + // hold — a forged `num_items` can't drive a huge reservation. + let mut heap = Vec::with_capacity(num_items.min(data.len() / 12 + 1)); let mut pos = TOPK_META_LEN + cms_size * 8; for _ in 0..num_items { if pos + 4 > data.len() { @@ -1206,6 +1237,30 @@ mod tests { assert_eq!(k, 10); } + #[test] + fn test_forged_cms_header_rejected() { + // Magic + oversized declared width/depth but a short buffer must be + // rejected so counter accessors can't index out of bounds. + let mut forged = Vec::new(); + forged.extend_from_slice(CMS_HEADER); + forged.extend_from_slice(&u32::MAX.to_le_bytes()); // width + forged.extend_from_slice(&u32::MAX.to_le_bytes()); // depth + forged.extend_from_slice(&0u64.to_le_bytes()); // total_count + assert!(!is_cms(&forged)); + } + + #[test] + fn test_forged_topk_header_rejected() { + let mut forged = Vec::new(); + forged.extend_from_slice(TOPK_HEADER); + forged.extend_from_slice(&3u32.to_le_bytes()); // k + forged.extend_from_slice(&u32::MAX.to_le_bytes()); // width + forged.extend_from_slice(&u32::MAX.to_le_bytes()); // depth + forged.extend_from_slice(&0.9f64.to_le_bytes()); // decay + forged.extend_from_slice(&0u32.to_le_bytes()); // num_items + assert!(!is_topk(&forged)); + } + #[test] fn test_cms_basic() { let mut data = new_cms(100, 5); diff --git a/src/command/set.rs b/src/command/set.rs index f407080..202ef48 100644 --- a/src/command/set.rs +++ b/src/command/set.rs @@ -4,6 +4,11 @@ use crate::protocol::RespValue; use crate::storage::RedisObject; use super::registry::CommandContext; +/// Upper bound on the magnitude of a negative `count` for SRANDMEMBER / +/// ZRANDMEMBER / HRANDFIELD (the with-duplicates form), preventing an +/// attacker from forcing a multi-billion-element allocation. +const MAX_RAND_COUNT: u64 = 1_000_000; + fn get_set<'a>(ctx: &'a mut CommandContext, key: &Bytes) -> Result>, RespValue> { match ctx.db().get(key) { Some(RedisObject::Set(s)) => Ok(Some(s)), @@ -154,8 +159,13 @@ pub fn cmd_srandmember(ctx: &mut CommandContext) -> RespValue { RespValue::array(members) } Some(n) => { - // Negative count: allow duplicates - let n = (-n) as usize; + // Negative count: allow duplicates. Cap the magnitude (and + // guard i64::MIN's negation) so a huge count can't allocate + // billions of elements. + let n = match n.checked_neg() { + Some(v) if v as u64 <= MAX_RAND_COUNT => v as usize, + _ => return RespValue::error("ERR count value is out of range"), + }; use rand::seq::SliceRandom; let items: Vec<&Bytes> = set.iter().collect(); let members: Vec = (0..n) diff --git a/src/command/sorted_set.rs b/src/command/sorted_set.rs index eb8dd38..6896fd4 100644 --- a/src/command/sorted_set.rs +++ b/src/command/sorted_set.rs @@ -4,6 +4,10 @@ use crate::storage::types::SortedSetData; use crate::storage::RedisObject; use super::registry::CommandContext; +/// Upper bound on the magnitude of a negative ZRANDMEMBER `count` (the +/// with-duplicates form), preventing an unbounded allocation. +const MAX_RAND_COUNT: u64 = 1_000_000; + fn get_zset<'a>(ctx: &'a mut CommandContext, key: &Bytes) -> Result, RespValue> { match ctx.db().get(key) { Some(RedisObject::SortedSet(z)) => Ok(Some(z)), @@ -588,7 +592,12 @@ pub fn cmd_zrandmember(ctx: &mut CommandContext) -> RespValue { } } Some(n) => { - let n = (-n) as usize; + // Negative count allows duplicates; cap the magnitude and + // guard i64::MIN to avoid an unbounded allocation. + let n = match n.checked_neg() { + Some(v) if v as u64 <= MAX_RAND_COUNT => v as usize, + _ => return RespValue::error("ERR count value is out of range"), + }; use rand::seq::SliceRandom; let all: Vec<(Bytes, f64)> = zset.range_by_index(0, -1); let mut items = Vec::new(); diff --git a/src/http.rs b/src/http.rs index 4b29546..ceabe29 100644 --- a/src/http.rs +++ b/src/http.rs @@ -48,7 +48,13 @@ pub async fn run_http_server(state: Arc, port: u16) -> Result<(), B } }); - if let Err(e) = http1::Builder::new().serve_connection(io, svc).await { + // Bound how long a client may take to send request headers, so a + // slow-loris client cannot hold a connection open indefinitely. + let mut builder = http1::Builder::new(); + builder + .timer(hyper_util::rt::TokioTimer::new()) + .header_read_timeout(std::time::Duration::from_secs(15)); + if let Err(e) = builder.serve_connection(io, svc).await { tracing::debug!("HTTP connection error: {}", e); } }); @@ -77,17 +83,21 @@ async fn handle_request( (Method::GET, "/info") => handle_info(&state).await, (Method::GET, "/metrics") => handle_metrics(&state).await, (Method::POST, "/api/v1/command") => { - let body = req.collect().await?.to_bytes(); - handle_command(&state, &user, &body).await + match collect_limited(req).await { + Ok(body) => handle_command(&state, &user, &body).await, + Err(resp) => Ok(resp), + } } (Method::GET, p) if p.starts_with("/api/v1/") => { let key = &p["/api/v1/".len()..]; handle_get_key(&state, &user, key).await } (Method::PUT, p) if p.starts_with("/api/v1/") => { - let key = &p["/api/v1/".len()..]; - let body = req.collect().await?.to_bytes(); - handle_put_key(&state, &user, key, &body).await + let key = p["/api/v1/".len()..].to_string(); + match collect_limited(req).await { + Ok(body) => handle_put_key(&state, &user, &key, &body).await, + Err(resp) => Ok(resp), + } } (Method::DELETE, p) if p.starts_with("/api/v1/") => { let key = &p["/api/v1/".len()..]; @@ -140,17 +150,48 @@ async fn check_auth( }; if let Some(ref req_pass) = state.config.requirepass { - if token == *req_pass { + if crate::command::acl::verify_secret(&token, req_pass) { return Ok("default".to_string()); } } match acl::http_authenticate(&token) { Some(user) => Ok(user), - None => Err(unauthorized_response()), + None => { + // Fixed delay on a bad credential to slow brute-force attempts. The + // HTTP path is stateless per request, so this is a per-attempt cost + // rather than a per-connection backoff. + tokio::time::sleep(std::time::Duration::from_millis(AUTH_FAIL_DELAY_MS)).await; + Err(unauthorized_response()) + } + } +} + +/// Maximum request-body size accepted by the HTTP API (64 MB). Bounds +/// memory used before a handler runs, preventing a body-flood DoS. +const MAX_HTTP_BODY: usize = 64 * 1024 * 1024; +/// Delay applied after a failed HTTP auth attempt to slow brute force. +const AUTH_FAIL_DELAY_MS: u64 = 250; + +/// Collect a request body, rejecting anything larger than `MAX_HTTP_BODY`. +async fn collect_limited(req: Request) -> Result>> { + use http_body_util::Limited; + let limited = Limited::new(req.into_body(), MAX_HTTP_BODY); + match limited.collect().await { + Ok(collected) => Ok(collected.to_bytes()), + Err(_) => Err(payload_too_large_response()), } } +fn payload_too_large_response() -> Response> { + let body = serde_json::to_vec(&json!({"error": "payload too large"})).unwrap_or_default(); + Response::builder() + .status(StatusCode::PAYLOAD_TOO_LARGE) + .header("Content-Type", "application/json") + .body(Full::new(Bytes::from(body))) + .expect("static response builder") +} + fn unauthorized_response() -> Response> { let body = serde_json::to_vec(&json!({"error": "unauthorized"})).unwrap_or_default(); Response::builder() diff --git a/src/protocol/parser.rs b/src/protocol/parser.rs index 91e4db5..54775a1 100644 --- a/src/protocol/parser.rs +++ b/src/protocol/parser.rs @@ -16,6 +16,11 @@ const MAX_BULK_LEN: usize = 512 * 1024 * 1024; const MAX_MULTIBULK_LEN: usize = 1_048_576; /// Maximum nesting depth for aggregates — guards against stack overflow. const MAX_DEPTH: u32 = 32; +/// Upper bound on how many aggregate slots to pre-reserve before any element +/// bytes are seen. The declared length is still validated against +/// `MAX_MULTIBULK_LEN`, but the container grows lazily from here so a tiny +/// message can't force a huge (or deeply nested, amplified) up-front allocation. +const PREALLOC_CAP: usize = 1024; impl Parser { /// Try to parse one complete RESP value from the buffer. @@ -127,7 +132,7 @@ impl Parser { return Err(ParseError::Invalid("multibulk length out of range".to_string())); } - let mut items = Vec::with_capacity(len); + let mut items = Vec::with_capacity(len.min(PREALLOC_CAP)); for _ in 0..len { let rest = buf.get(consumed..).ok_or(ParseError::Incomplete)?; @@ -230,7 +235,7 @@ impl Parser { if len > MAX_MULTIBULK_LEN { return Err(ParseError::Invalid("map length out of range".to_string())); } - let mut entries = Vec::with_capacity(len); + let mut entries = Vec::with_capacity(len.min(PREALLOC_CAP)); for _ in 0..len { let rest = buf.get(consumed..).ok_or(ParseError::Incomplete)?; @@ -262,7 +267,7 @@ impl Parser { if len > MAX_MULTIBULK_LEN { return Err(ParseError::Invalid("set length out of range".to_string())); } - let mut items = Vec::with_capacity(len); + let mut items = Vec::with_capacity(len.min(PREALLOC_CAP)); for _ in 0..len { let rest = buf.get(consumed..).ok_or(ParseError::Incomplete)?; @@ -289,7 +294,7 @@ impl Parser { if len > MAX_MULTIBULK_LEN { return Err(ParseError::Invalid("push length out of range".to_string())); } - let mut items = Vec::with_capacity(len); + let mut items = Vec::with_capacity(len.min(PREALLOC_CAP)); for _ in 0..len { let rest = buf.get(consumed..).ok_or(ParseError::Incomplete)?; diff --git a/src/server/connection.rs b/src/server/connection.rs index aaff19c..055ca05 100644 --- a/src/server/connection.rs +++ b/src/server/connection.rs @@ -20,6 +20,12 @@ use super::SharedState; /// client query-buffer cap. const MAX_QUERY_BUFFER: usize = 1024 * 1024 * 1024; +/// Idle read timeout. When it fires, a connection that is mid-command (has a +/// partial request buffered) or not yet authenticated is dropped — the +/// slow-loris shapes. A fully idle, authenticated client with an empty buffer +/// (a normal pooled connection) is left alone. +const READ_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + /// A stream that may or may not be TLS-wrapped. pub enum MaybeTls { Plain(TcpStream), @@ -180,8 +186,25 @@ impl Connection { self.stream.write_all(&data).await?; } - // Read more data from the socket - let n = self.stream.read_buf(&mut self.buffer).await?; + // Read more data from the socket, bounded by an idle timeout. + let n = match tokio::time::timeout( + READ_IDLE_TIMEOUT, + self.stream.read_buf(&mut self.buffer), + ) + .await + { + Ok(res) => res?, + Err(_) => { + // Idle timeout: reap slow/partial or pre-auth connections, + // but let an authenticated client with no pending data + // keep its connection open. + if self.buffer.is_empty() && self.authenticated { + continue; + } + self.cleanup_pubsub().await; + return Ok(()); + } + }; if n == 0 { self.cleanup_pubsub().await; return Ok(()); @@ -836,7 +859,7 @@ impl Connection { if username == "default" { if let Some(ref req_pass) = self.state.config.requirepass { - if password == *req_pass { + if crate::command::acl::verify_secret(&password, req_pass) { self.auth_success(username); return RespValue::ok(); } @@ -863,7 +886,7 @@ impl Connection { if username == "default" { if let Some(ref req_pass) = self.state.config.requirepass { - if password == *req_pass { + if crate::command::acl::verify_secret(&password, req_pass) { self.auth_success(username); return None; } diff --git a/src/server/mod.rs b/src/server/mod.rs index a5b441c..0f883df 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -381,8 +381,15 @@ impl Server { state.connected_clients.fetch_add(1, Ordering::Relaxed); tokio::spawn(async move { - match acceptor.accept(socket).await { - Ok(tls_stream) => { + // Bound the TLS handshake so a client that completes TCP + // but stalls the handshake can't hold a client slot open. + let handshake = tokio::time::timeout( + std::time::Duration::from_secs(10), + acceptor.accept(socket), + ) + .await; + match handshake { + Ok(Ok(tls_stream)) => { tracing::debug!("New TLS connection from {} (client_id={})", addr, client_id); let stream = connection::MaybeTls::Tls(tls_stream); let mut conn = connection::Connection::new(stream, state.clone(), client_id); @@ -390,9 +397,12 @@ impl Server { tracing::debug!("TLS connection {} error: {}", addr, e); } } - Err(e) => { + Ok(Err(e)) => { tracing::debug!("TLS handshake failed from {}: {}", addr, e); } + Err(_) => { + tracing::debug!("TLS handshake timed out from {}", addr); + } } state.connected_clients.fetch_sub(1, Ordering::Relaxed); drop(permit);