Security: fix low-severity findings (v0.11.8)

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rcache"
-version = "0.11.7"
+version = "0.11.8"
 edition = "2024"
 
 [dependencies]

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::<Vec<_>>()
+        };
+
+        // 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"));

src/compression.rs

@@ -39,6 +39,22 @@ pub fn decompress(data: &[u8]) -> Result<Bytes, String> {
     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";

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 {

src/persistence/aof.rs

@@ -424,7 +424,7 @@ pub fn replay(path: &Path, store: &mut Store) -> io::Result<usize> {
                             if let Ok(ts) = String::from_utf8_lossy(&args[2]).parse::<u64>() {
                                 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<usize> {
                             if let Ok(ts_ms) = String::from_utf8_lossy(&args[2]).parse::<u64>() {
                                 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);

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<String> {
-        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<String>)> {
-        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<String> = lib.functions.keys().cloned().collect();
@@ -149,7 +149,7 @@ impl FunctionLibrary {
     }
 
     pub fn find_function(&self, fname: &str) -> Option<String> {
-        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();
     }
 }

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