diff --git a/Cargo.lock b/Cargo.lock index 166b5ed..f51b719 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -791,7 +791,7 @@ dependencies = [ [[package]] name = "rcache" -version = "0.11.7" +version = "0.11.8" dependencies = [ "bytes", "crc16", diff --git a/Cargo.toml b/Cargo.toml index c07f14d..ad2c6b1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rcache" -version = "0.11.7" +version = "0.11.8" edition = "2024" [dependencies] diff --git a/src/command/acl.rs b/src/command/acl.rs index aa1c65d..279889c 100644 --- a/src/command/acl.rs +++ b/src/command/acl.rs @@ -586,6 +586,44 @@ pub fn command_keys<'a>(cmd: &str, args: &'a [Bytes]) -> Vec<&'a Bytes> { } } +/// Return the argument values that are pub/sub channels (or patterns) for `cmd`, +/// for per-channel ACL checks. Returns an empty vec for non-pub/sub commands. +pub fn command_channels<'a>(cmd: &str, args: &'a [Bytes]) -> Vec<&'a Bytes> { + match cmd { + // Only the first argument is a channel; the rest is the payload. + "PUBLISH" | "SPUBLISH" => args.get(1).into_iter().collect(), + // Every argument after the command is a channel or pattern. + "SUBSCRIBE" | "UNSUBSCRIBE" | "SSUBSCRIBE" | "SUNSUBSCRIBE" | "PSUBSCRIBE" + | "PUNSUBSCRIBE" => args.iter().skip(1).collect(), + _ => Vec::new(), + } +} + +/// Whether `username`'s channel patterns cover every channel. +pub fn user_has_all_channels(username: &str) -> bool { + match ACL_USERS.lock() { + Ok(users) => users + .get(username) + .map(|u| u.channel_patterns.iter().any(|p| p == "*")) + .unwrap_or(true), + Err(_) => false, + } +} + +/// Whether `username` may use the pub/sub `channel`. +pub fn is_channel_allowed(username: &str, channel: &str) -> bool { + match ACL_USERS.lock() { + Ok(users) => match users.get(username) { + Some(u) => u + .channel_patterns + .iter() + .any(|pat| pat == "*" || crate::storage::db::glob_match(pat, channel)), + None => true, + }, + Err(_) => false, + } +} + /// Whether `username`'s key patterns cover every key (no per-key check needed). pub fn user_has_all_keys(username: &str) -> bool { match ACL_USERS.lock() { @@ -679,6 +717,27 @@ mod tests { assert_eq!(keys[0].as_ref(), b"mykey"); } + #[test] + fn test_command_channels() { + let mk = |parts: &[&str]| { + parts.iter().map(|s| Bytes::from(s.to_string())).collect::>() + }; + + // PUBLISH: only the channel, not the message payload. + let args = mk(&["PUBLISH", "news", "hello world"]); + let chans = command_channels("PUBLISH", &args); + assert_eq!(chans.len(), 1); + assert_eq!(chans[0].as_ref(), b"news"); + + // SUBSCRIBE: every argument is a channel. + let args = mk(&["SUBSCRIBE", "a", "b"]); + assert_eq!(command_channels("SUBSCRIBE", &args).len(), 2); + + // Non-pub/sub command: no channels. + let args = mk(&["GET", "key"]); + assert!(command_channels("GET", &args).is_empty()); + } + #[test] fn test_constant_time_eq() { assert!(constant_time_eq(b"abc", b"abc")); diff --git a/src/compression.rs b/src/compression.rs index 1c73955..bde6759 100644 --- a/src/compression.rs +++ b/src/compression.rs @@ -39,6 +39,22 @@ pub fn decompress(data: &[u8]) -> Result { let original_len = read_original_len(data).map_err(|e| e.to_string())?; let compressed_data = &data[8..]; + // Guard against a decompression bomb: `decompress_size_prepended` allocates + // from the size prefix embedded in `compressed_data`, which is attacker- + // controlled if the value was crafted. Reject anything whose declared output + // exceeds the cap before decompressing. + const MAX_DECOMPRESSED: usize = 512 * 1024 * 1024; + if original_len > MAX_DECOMPRESSED { + return Err("decompressed size exceeds maximum".to_string()); + } + if compressed_data.len() >= 4 { + let prepended = + u32::from_le_bytes(compressed_data[0..4].try_into().unwrap()) as usize; + if prepended > MAX_DECOMPRESSED { + return Err("decompressed size exceeds maximum".to_string()); + } + } + match lz4_flex::decompress_size_prepended(compressed_data) { Ok(decompressed) => { if decompressed.len() != original_len { @@ -106,6 +122,17 @@ mod tests { assert_eq!(original_size(data), data.len()); } + #[test] + fn test_decompress_rejects_oversized_header() { + // Forge a value whose header claims a huge original length; decompress + // must reject it rather than attempting a giant allocation. + let mut forged = Vec::new(); + forged.extend_from_slice(LZ4_MAGIC); + forged.extend_from_slice(&u32::MAX.to_le_bytes()); // original_len + forged.extend_from_slice(&[0u8; 8]); // dummy compressed body + assert!(decompress(&forged).is_err()); + } + #[test] fn test_maybe_compress_below_threshold() { let data = b"short"; diff --git a/src/config.rs b/src/config.rs index 075da06..e2ac341 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,5 +1,5 @@ /// Server configuration. -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct Config { pub bind: String, pub port: u16, @@ -39,6 +39,39 @@ pub struct Config { pub protected_mode: bool, } +impl std::fmt::Debug for Config { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + // Redact requirepass so it never lands in a `{:?}` log line. + let requirepass = self.requirepass.as_ref().map(|_| "***"); + f.debug_struct("Config") + .field("bind", &self.bind) + .field("port", &self.port) + .field("databases", &self.databases) + .field("maxclients", &self.maxclients) + .field("requirepass", &requirepass) + .field("hz", &self.hz) + .field("maxmemory", &self.maxmemory) + .field("maxmemory_policy", &self.maxmemory_policy) + .field("maxmemory_samples", &self.maxmemory_samples) + .field("rdb_filename", &self.rdb_filename) + .field("aof_enabled", &self.aof_enabled) + .field("aof_filename", &self.aof_filename) + .field("appendfsync", &self.appendfsync) + .field("lfu_log_factor", &self.lfu_log_factor) + .field("lfu_decay_time", &self.lfu_decay_time) + .field("http_port", &self.http_port) + .field("tls_port", &self.tls_port) + .field("tls_cert_file", &self.tls_cert_file) + .field("tls_key_file", &self.tls_key_file) + .field("compression_threshold", &self.compression_threshold) + .field("compression_enabled", &self.compression_enabled) + .field("slowlog_log_slower_than", &self.slowlog_log_slower_than) + .field("slowlog_max_len", &self.slowlog_max_len) + .field("protected_mode", &self.protected_mode) + .finish() + } +} + impl Default for Config { fn default() -> Self { Self { diff --git a/src/persistence/aof.rs b/src/persistence/aof.rs index bcd5838..46826f1 100644 --- a/src/persistence/aof.rs +++ b/src/persistence/aof.rs @@ -424,7 +424,7 @@ pub fn replay(path: &Path, store: &mut Store) -> io::Result { if let Ok(ts) = String::from_utf8_lossy(&args[2]).parse::() { let now_secs = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) - .unwrap() + .unwrap_or_default() .as_secs(); if ts > now_secs { let db = store.db_mut(current_db); @@ -441,7 +441,7 @@ pub fn replay(path: &Path, store: &mut Store) -> io::Result { if let Ok(ts_ms) = String::from_utf8_lossy(&args[2]).parse::() { let now_ms = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) - .unwrap() + .unwrap_or_default() .as_millis() as u64; if ts_ms > now_ms { let db = store.db_mut(current_db); diff --git a/src/scripting_engine.rs b/src/scripting_engine.rs index b030f3d..bf9cf9c 100644 --- a/src/scripting_engine.rs +++ b/src/scripting_engine.rs @@ -27,26 +27,26 @@ impl ScriptCache { /// Load a script into the cache, returning its SHA1 hash. pub fn load(&self, script: &str) -> String { let sha = sha1_hex(script); - let mut scripts = self.scripts.lock().unwrap(); + let mut scripts = self.scripts.lock().unwrap_or_else(|e| e.into_inner()); scripts.insert(sha.clone(), script.to_string()); sha } /// Check if a script exists by SHA1. pub fn exists(&self, sha: &str) -> bool { - let scripts = self.scripts.lock().unwrap(); + let scripts = self.scripts.lock().unwrap_or_else(|e| e.into_inner()); scripts.contains_key(sha) } /// Get a script by SHA1. pub fn get(&self, sha: &str) -> Option { - let scripts = self.scripts.lock().unwrap(); + let scripts = self.scripts.lock().unwrap_or_else(|e| e.into_inner()); scripts.get(sha).cloned() } /// Flush all cached scripts. pub fn flush(&self) { - let mut scripts = self.scripts.lock().unwrap(); + let mut scripts = self.scripts.lock().unwrap_or_else(|e| e.into_inner()); scripts.clear(); } } @@ -84,7 +84,7 @@ impl FunctionLibrary { .map(|s| s.trim().to_string()) .ok_or_else(|| "ERR Library name not found in header".to_string())?; - let mut libs = self.libraries.lock().unwrap(); + let mut libs = self.libraries.lock().unwrap_or_else(|e| e.into_inner()); if libs.contains_key(&name) && !replace { return Err(format!("ERR Library '{}' already exists", name)); } @@ -130,7 +130,7 @@ impl FunctionLibrary { } pub fn delete(&self, name: &str) -> Result<(), String> { - let mut libs = self.libraries.lock().unwrap(); + let mut libs = self.libraries.lock().unwrap_or_else(|e| e.into_inner()); if libs.remove(name).is_some() { Ok(()) } else { @@ -139,7 +139,7 @@ impl FunctionLibrary { } pub fn list(&self) -> Vec<(String, String, Vec)> { - let libs = self.libraries.lock().unwrap(); + let libs = self.libraries.lock().unwrap_or_else(|e| e.into_inner()); libs.values() .map(|lib| { let fnames: Vec = lib.functions.keys().cloned().collect(); @@ -149,7 +149,7 @@ impl FunctionLibrary { } pub fn find_function(&self, fname: &str) -> Option { - let libs = self.libraries.lock().unwrap(); + let libs = self.libraries.lock().unwrap_or_else(|e| e.into_inner()); for lib in libs.values() { if lib.functions.contains_key(fname) { return Some(lib.code.clone()); @@ -159,7 +159,7 @@ impl FunctionLibrary { } pub fn flush(&self) { - let mut libs = self.libraries.lock().unwrap(); + let mut libs = self.libraries.lock().unwrap_or_else(|e| e.into_inner()); libs.clear(); } } diff --git a/src/server/connection.rs b/src/server/connection.rs index a76afb2..2683de8 100644 --- a/src/server/connection.rs +++ b/src/server/connection.rs @@ -392,6 +392,26 @@ impl Connection { } } } + // Check channel patterns for pub/sub commands. + if !acl::user_has_all_channels(&self.auth_username) { + for channel in acl::command_channels(&cmd_name, &args) { + let chan_str = String::from_utf8_lossy(channel); + if !acl::is_channel_allowed(&self.auth_username, &chan_str) { + return RespValue::error( + "NOPERM this user has no permissions to access one of the channels used as arguments" + .to_string(), + ); + } + } + } + } + + // ACL WHOAMI reflects the actual authenticated user (the command handler + // has no connection state, so it is answered here). + if cmd_name == "ACL" && args.len() >= 2 + && String::from_utf8_lossy(&args[1]).eq_ignore_ascii_case("WHOAMI") + { + return RespValue::bulk_string(Bytes::from(self.auth_username.clone())); } // Handle CLIENT TRACKING specially