From ab91cded9405bd75cfa638b56c0efbd3000ddb5f Mon Sep 17 00:00:00 2001 From: s0u7a Date: Wed, 11 Nov 2026 08:53:53 +0000 Subject: [PATCH] fix: namespace Valkey cache keys so clear() removes this cache's entries The Valkey backend stored entries under the raw "METHOD URL BODY" key, but clear() deleted `KEYS searxng:*`. It never matched this cache's own entries, and on a database shared with an upstream SearXNG it deleted that instance's keys instead (audit A12). ValkeyCache now prefixes every key with "rustxng:cache:" for GET, SET and clear(). clear() walks the keyspace with SCAN ... MATCH instead of a blocking KEYS and DELs each page. Entries written by older versions use the old key format; they are no longer reachable and expire via their TTL. Co-Authored-By: Claude Opus 5.5 --- src/cache.rs | 243 +++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 217 insertions(+), 26 deletions(-) diff --git a/src/cache.rs b/src/cache.rs index 30b1734..7883385 100644 --- a/src/cache.rs +++ b/src/cache.rs @@ -309,6 +309,11 @@ impl Cache { } } +/// Namespace for every key this cache stores in Valkey / Redis, so `clear()` +/// removes exactly what `set()` wrote and nothing owned by other applications +/// (such as an upstream SearXNG) sharing the same database. +const VALKEY_KEY_PREFIX: &str = "rustxng:cache:"; + /// A cache stored in Valkey / Redis, like the upstream valkey cache backend. pub struct ValkeyCache { url: String, @@ -333,22 +338,30 @@ impl ValkeyCache { } } - async fn command(&self, args: &[&str]) -> Result, String> { + async fn request(&self, args: &[&str]) -> Result { let mut pooled = self.connection.lock().await; // Cancellation must discard the connection. let mut connection = match pooled.take() { Some(connection) => connection, None => ValkeyConnection::connect(&self.url).await?, }; - let response = connection.command(args).await; + let response = connection.request(args).await; if response.is_ok() { *pooled = Some(connection); } response } + async fn command(&self, args: &[&str]) -> Result, String> { + self.request(args).await.map(ValkeyReply::flatten) + } + + fn namespaced(key: &str) -> String { + format!("{VALKEY_KEY_PREFIX}{key}") + } + pub async fn get(&self, key: &str) -> Option { - match self.command(&["GET", key]).await { + match self.command(&["GET", &Self::namespaced(key)]).await { Ok(Some(payload)) => match decode_response(&payload) { Some(response) => { self.hit.fetch_add(1, Ordering::Relaxed); @@ -378,7 +391,8 @@ impl ValkeyCache { return; }; let ttl = self.ttl.as_secs().max(1).to_string(); - match self.command(&["SET", key, &payload, "EX", &ttl]).await { + let key = Self::namespaced(key); + match self.command(&["SET", &key, &payload, "EX", &ttl]).await { Ok(_) => { self.entries.fetch_add(1, Ordering::Relaxed); } @@ -388,13 +402,44 @@ impl ValkeyCache { } } + /// Delete this cache's entries, walking the keyspace with SCAN so a large + /// database is not blocked the way a single KEYS call would block it. pub async fn clear(&self) { - if let Ok(Some(keys)) = self.command(&["KEYS", "searxng:*"]).await { - for key in keys.split('\n') { - if !key.is_empty() { - let _ = self.command(&["DEL", key]).await; + let pattern = format!("{VALKEY_KEY_PREFIX}*"); + let mut cursor = "0".to_string(); + loop { + let scan: [&str; 6] = ["SCAN", &cursor, "MATCH", &pattern, "COUNT", "100"]; + let page = match self.request(&scan).await { + Ok(ValkeyReply::Array(page)) => <[ValkeyReply; 2]>::try_from(page), + _ => { + self.error.fetch_add(1, Ordering::Relaxed); + return; + } + }; + // A SCAN reply is [next cursor, [keys...]]. + let Ok([ValkeyReply::Bulk(Some(next)), ValkeyReply::Array(keys)]) = page else { + self.error.fetch_add(1, Ordering::Relaxed); + return; + }; + let keys: Vec = keys + .into_iter() + .filter_map(|key| match key { + ValkeyReply::Bulk(Some(key)) => Some(key), + _ => None, + }) + .collect(); + if !keys.is_empty() { + let mut args = vec!["DEL"]; + args.extend(keys.iter().map(String::as_str)); + if self.command(&args).await.is_err() { + self.error.fetch_add(1, Ordering::Relaxed); + return; } } + if next == "0" { + return; + } + cursor = next; } } @@ -490,6 +535,11 @@ impl ValkeyConnection { /// Run one command; bulk replies come back as Some(String), nil as None. async fn command(&mut self, args: &[&str]) -> Result, String> { + self.request(args).await.map(ValkeyReply::flatten) + } + + /// Run one command and return the reply with its structure intact. + async fn request(&mut self, args: &[&str]) -> Result { use tokio::io::AsyncWriteExt as _; let mut request = format!("*{}\r\n", args.len()); for argument in args { @@ -503,22 +553,7 @@ impl ValkeyConnection { .flush() .await .map_err(|error| error.to_string())?; - match self.read_reply().await? { - ValkeyReply::Bulk(value) => Ok(value), - ValkeyReply::Simple(value) => Ok(Some(value)), - ValkeyReply::Integer(value) => Ok(Some(value.to_string())), - ValkeyReply::Nil => Ok(None), - ValkeyReply::Array(items) => Ok(Some( - items - .into_iter() - .filter_map(|item| match item { - ValkeyReply::Bulk(Some(value)) | ValkeyReply::Simple(value) => Some(value), - _ => None, - }) - .collect::>() - .join("\n"), - )), - } + self.read_reply().await } fn read_reply( @@ -579,6 +614,28 @@ enum ValkeyReply { Nil, } +impl ValkeyReply { + /// Flatten a reply to a string; array items are joined with newlines. + fn flatten(self) -> Option { + match self { + ValkeyReply::Bulk(value) => value, + ValkeyReply::Simple(value) => Some(value), + ValkeyReply::Integer(value) => Some(value.to_string()), + ValkeyReply::Nil => None, + ValkeyReply::Array(items) => Some( + items + .into_iter() + .filter_map(|item| match item { + ValkeyReply::Bulk(Some(value)) | ValkeyReply::Simple(value) => Some(value), + _ => None, + }) + .collect::>() + .join("\n"), + ), + } + } +} + #[cfg(test)] mod valkey_tests { use super::*; @@ -610,7 +667,10 @@ mod valkey_tests { let server = tokio::spawn(async move { let (stream, _) = listener.accept().await.unwrap(); let mut first = tokio::io::BufReader::new(stream); - assert_eq!(read_command(&mut first).await, ["GET", "first"]); + assert_eq!( + read_command(&mut first).await, + ["GET", "rustxng:cache:first"] + ); requested.send(()).unwrap(); cancelled.await.unwrap(); let payload = @@ -622,7 +682,10 @@ mod valkey_tests { let (stream, _) = listener.accept().await.unwrap(); let mut second = tokio::io::BufReader::new(stream); - assert_eq!(read_command(&mut second).await, ["GET", "second"]); + assert_eq!( + read_command(&mut second).await, + ["GET", "rustxng:cache:second"] + ); let payload = r#"{"status":200,"final_url":"https://second.invalid/","body":"c2Vjb25k"}"#; second @@ -708,6 +771,134 @@ mod valkey_tests { } } + fn bulk(value: &str) -> String { + format!("${}\r\n{value}\r\n", value.len()) + } + + fn glob_matches(pattern: &str, key: &str) -> bool { + match pattern.strip_suffix('*') { + Some(prefix) => key.starts_with(prefix), + None => key == pattern, + } + } + + type Keyspace = std::collections::BTreeMap; + + /// Answer one command against a fake keyspace. SCAN visits one key per + /// page, and its cursor names the last visited key so deleting keys + /// between pages skips nothing (as with a real server). + fn reply_to(store: &mut Keyspace, command: &[String]) -> String { + match command[0].as_str() { + "GET" => match store.get(&command[1]) { + Some(value) => bulk(value), + None => "$-1\r\n".to_string(), + }, + "SET" => { + store.insert(command[1].clone(), command[2].clone()); + "+OK\r\n".to_string() + } + "DEL" => { + let removed = command[1..] + .iter() + .filter(|key| store.remove(*key).is_some()) + .count(); + format!(":{removed}\r\n") + } + "KEYS" => { + let keys: Vec<_> = store + .keys() + .filter(|key| glob_matches(&command[1], key)) + .map(|key| bulk(key)) + .collect(); + format!("*{}\r\n{}", keys.len(), keys.concat()) + } + "SCAN" => { + let after = command[1].strip_prefix("after:"); + let pattern = match command.iter().position(|argument| argument == "MATCH") { + Some(index) => command[index + 1].as_str(), + None => "*", + }; + let mut candidates = store + .keys() + .filter(|key| after.is_none_or(|after| key.as_str() > after)); + let current = candidates.next(); + let next = match (current, candidates.next()) { + (Some(key), Some(_)) => format!("after:{key}"), + _ => "0".to_string(), + }; + let page: Vec<_> = current + .filter(|key| glob_matches(pattern, key)) + .map(|key| bulk(key)) + .into_iter() + .collect(); + let cursor = bulk(&next); + format!("*2\r\n{cursor}*{}\r\n{}", page.len(), page.concat()) + } + other => format!("-ERR unknown command {other}\r\n"), + } + } + + /// Serve a fake Valkey keyspace on every accepted connection. + async fn serve_keyspace( + listener: tokio::net::TcpListener, + store: std::sync::Arc>, + ) { + loop { + let (stream, _) = listener.accept().await.unwrap(); + let store = std::sync::Arc::clone(&store); + tokio::spawn(async move { + let mut stream = tokio::io::BufReader::new(stream); + loop { + match stream.fill_buf().await { + Ok(buffer) if !buffer.is_empty() => {} + _ => return, + } + let command = read_command(&mut stream).await; + let reply = reply_to(&mut store.lock().unwrap(), &command); + stream.get_mut().write_all(reply.as_bytes()).await.unwrap(); + } + }); + } + } + + #[tokio::test] + async fn clear_removes_own_entries_and_leaves_other_namespaces() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("redis://{}", listener.local_addr().unwrap()); + // Keys owned by another application sharing the same database, + // e.g. an upstream SearXNG instance. + let foreign = ["searxng:limiter", "searxng:cache", "other"] + .map(|key| (key.to_string(), "foreign".to_string())); + let store = std::sync::Arc::new(Mutex::new(Keyspace::from(foreign))); + let server = tokio::spawn(serve_keyspace(listener, std::sync::Arc::clone(&store))); + + let cache = Cache::valkey(&url, Duration::from_secs(60)); + let first = Cache::key("GET", "https://example.test/a", ""); + let second = Cache::key("POST", "https://example.test/b", "q=1"); + cache.set(&first, response_with_body("first")).await; + cache.set(&second, response_with_body("second")).await; + assert!(cache.get(&first).await.is_some()); + + cache.clear().await; + + // Both of this cache's entries are gone ... + assert!(cache.get(&first).await.is_none()); + assert!(cache.get(&second).await.is_none()); + // ... and every key outside its namespace is untouched. + let remaining: Vec<_> = store.lock().unwrap().keys().cloned().collect(); + assert_eq!(remaining, ["other", "searxng:cache", "searxng:limiter"]); + server.abort(); + } + + fn response_with_body(body: &str) -> CachedResponse { + CachedResponse { + status: 200, + final_url: "https://example.test/".into(), + headers: Default::default(), + body: std::sync::Arc::new(body.as_bytes().to_vec()), + } + } + #[test] fn responses_round_trip_through_the_valkey_payload() { let response = CachedResponse {