diff --git a/Cargo.lock b/Cargo.lock index 5f4e99c..166b5ed 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -791,7 +791,7 @@ dependencies = [ [[package]] name = "rcache" -version = "0.11.6" +version = "0.11.7" dependencies = [ "bytes", "crc16", diff --git a/Cargo.toml b/Cargo.toml index e8dcdcc..c07f14d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rcache" -version = "0.11.6" +version = "0.11.7" edition = "2024" [dependencies] diff --git a/README.md b/README.md index 222e0b9..bf23ee8 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,7 @@ redis-cli -p 6380 | `--bind` | 0.0.0.0 | Bind address | | `--port` | 6379 | RESP protocol port | | `--requirepass` | (none) | Password for AUTH | +| `--protected-mode` | yes | When enabled and no auth is configured, only loopback clients are accepted | | `--databases` | 16 | Number of databases | | `--maxclients` | 10000 | Max concurrent connections | | `--maxmemory` | 0 (unlimited) | Memory limit in bytes | diff --git a/src/command/acl.rs b/src/command/acl.rs index aa43034..aa1c65d 100644 --- a/src/command/acl.rs +++ b/src/command/acl.rs @@ -553,6 +553,39 @@ pub fn is_key_allowed(username: &str, key: &str) -> bool { } } +/// Return the argument values that are keys for `cmd`, for per-key ACL checks. +/// `args[0]` is the command name. This corrects the previous "always check +/// `args[1]`" behavior, which under-checked multi-key commands (MGET, MSET, +/// DEL, RENAME, ...) and mis-treated non-key arguments (PUBLISH channel, +/// SELECT index, ...) as keys. +pub fn command_keys<'a>(cmd: &str, args: &'a [Bytes]) -> Vec<&'a Bytes> { + match cmd { + // Commands with no key arguments. + "PING" | "ECHO" | "SELECT" | "AUTH" | "HELLO" | "QUIT" | "INFO" | "CONFIG" | "CLIENT" + | "COMMAND" | "DBSIZE" | "FLUSHDB" | "FLUSHALL" | "SWAPDB" | "TIME" | "DEBUG" + | "SLOWLOG" | "ACL" | "SUBSCRIBE" | "UNSUBSCRIBE" | "PSUBSCRIBE" | "PUNSUBSCRIBE" + | "PUBLISH" | "SPUBLISH" | "PUBSUB" | "SCRIPT" | "FUNCTION" | "FCALL" | "FCALL_RO" + | "EVAL" | "EVALSHA" | "EVAL_RO" | "EVALSHA_RO" | "WAIT" | "RESET" | "LOLWUT" + | "MEMORY" | "LATENCY" | "CLUSTER" | "REPLICAOF" | "SLAVEOF" | "MULTI" | "EXEC" + | "DISCARD" | "UNWATCH" | "SCAN" | "RANDOMKEY" | "KEYS" | "LASTSAVE" | "SAVE" + | "BGSAVE" | "BGREWRITEAOF" | "SHUTDOWN" | "REPLCONF" => Vec::new(), + + // Every argument after the command name is a key. + "MGET" | "DEL" | "UNLINK" | "EXISTS" | "WATCH" | "TOUCH" | "SUNION" | "SINTER" + | "SDIFF" | "PFCOUNT" | "PFMERGE" => args.iter().skip(1).collect(), + + // Alternating key/value pairs: k v k v ... + "MSET" | "MSETNX" => args.iter().skip(1).step_by(2).collect(), + + // Source and destination keys. + "RENAME" | "RENAMENX" | "SMOVE" | "COPY" | "LMOVE" | "BLMOVE" | "RPOPLPUSH" + | "BRPOPLPUSH" | "GEOSEARCHSTORE" | "ZRANGESTORE" => args.iter().skip(1).take(2).collect(), + + // Default: the first argument is the key, if present. + _ => args.get(1).into_iter().collect(), + } +} + /// 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() { @@ -614,6 +647,38 @@ static ACL_USERS: std::sync::LazyLock>> = mod tests { use super::*; + #[test] + fn test_command_keys() { + let mk = |parts: &[&str]| { + parts.iter().map(|s| Bytes::from(s.to_string())).collect::>() + }; + + // Every arg is a key. + let args = mk(&["MGET", "a", "b", "c"]); + assert_eq!(command_keys("MGET", &args).len(), 3); + + // Alternating key/value. + let args = mk(&["MSET", "k1", "v1", "k2", "v2"]); + let keys = command_keys("MSET", &args); + assert_eq!(keys.len(), 2); + assert_eq!(keys[0].as_ref(), b"k1"); + assert_eq!(keys[1].as_ref(), b"k2"); + + // Source + destination. + let args = mk(&["RENAME", "src", "dst"]); + assert_eq!(command_keys("RENAME", &args).len(), 2); + + // No-key command: the channel is not treated as a key. + let args = mk(&["PUBLISH", "chan", "msg"]); + assert!(command_keys("PUBLISH", &args).is_empty()); + + // Default: first argument is the key. + let args = mk(&["GET", "mykey"]); + let keys = command_keys("GET", &args); + assert_eq!(keys.len(), 1); + assert_eq!(keys[0].as_ref(), b"mykey"); + } + #[test] fn test_constant_time_eq() { assert!(constant_time_eq(b"abc", b"abc")); diff --git a/src/command/persistence_cmds.rs b/src/command/persistence_cmds.rs index c5d9e2c..1a54433 100644 --- a/src/command/persistence_cmds.rs +++ b/src/command/persistence_cmds.rs @@ -56,9 +56,12 @@ pub fn cmd_bgrewriteaof(ctx: &mut CommandContext) -> RespValue { // Since we can't access SharedState from the command handler directly, // we write a temp AOF from the store and let it be picked up. let aof_path = std::path::Path::new("appendonly.aof"); - let temp_path = aof_path.with_extension("aof.tmp"); + let (file, temp_path) = match crate::persistence::util::create_temp_file(aof_path) { + Ok(v) => v, + Err(e) => return RespValue::error(format!("ERR AOF rewrite failed: {}", e)), + }; - match rewrite_aof_from_store(ctx.store, &temp_path) { + match rewrite_aof_from_store(ctx.store, file) { Ok(()) => { // Atomically replace if let Err(e) = std::fs::rename(&temp_path, aof_path) { @@ -70,13 +73,12 @@ pub fn cmd_bgrewriteaof(ctx: &mut CommandContext) -> RespValue { } } -/// Helper to write all store data as AOF commands. -fn rewrite_aof_from_store(store: &crate::storage::Store, path: &std::path::Path) -> std::io::Result<()> { +/// Helper to write all store data as AOF commands into an already-open file. +fn rewrite_aof_from_store(store: &crate::storage::Store, file: std::fs::File) -> std::io::Result<()> { use bytes::Bytes; use std::io::{Write, BufWriter}; use crate::storage::types::RedisObject; - let file = std::fs::File::create(path)?; let mut writer = BufWriter::new(file); let now = std::time::Instant::now(); diff --git a/src/command/stream.rs b/src/command/stream.rs index a2d602f..1b44b1f 100644 --- a/src/command/stream.rs +++ b/src/command/stream.rs @@ -85,12 +85,12 @@ fn parse_stream_id_for_range_end(s: &str) -> Result { fn generate_id(stream: &StreamData) -> StreamId { let ms = current_time_ms(); let seq = if ms == stream.last_id.ms { - stream.last_id.seq + 1 + stream.last_id.seq.saturating_add(1) } else if ms > stream.last_id.ms { 0 } else { // Clock went backwards, use last_id.ms - stream.last_id.seq + 1 + stream.last_id.seq.saturating_add(1) }; let actual_ms = if ms >= stream.last_id.ms { ms @@ -369,11 +369,11 @@ pub fn cmd_xadd(ctx: &mut CommandContext) -> RespValue { if parts[1] == "*" { // Auto-sequence for given ms if ms == stream.last_id.ms { - stream.last_id.seq + 1 + stream.last_id.seq.saturating_add(1) } else if ms > stream.last_id.ms { 0 } else { - stream.last_id.seq + 1 + stream.last_id.seq.saturating_add(1) } } else { match parts[1].parse() { @@ -388,7 +388,7 @@ pub fn cmd_xadd(ctx: &mut CommandContext) -> RespValue { } else { // No seq specified, auto-assign if ms == stream.last_id.ms { - stream.last_id.seq + 1 + stream.last_id.seq.saturating_add(1) } else { 0 } diff --git a/src/command/strings.rs b/src/command/strings.rs index 484f046..f3c8c06 100644 --- a/src/command/strings.rs +++ b/src/command/strings.rs @@ -401,6 +401,9 @@ pub fn cmd_getrange(ctx: &mut CommandContext) -> RespValue { let db = ctx.db(); match db.get(&key) { Some(RedisObject::String(b)) => { + if b.is_empty() { + return RespValue::bulk_string(Bytes::new()); + } let len = b.len() as i64; let s = if start < 0 { (len + start).max(0) } else { start } as usize; let e = if end < 0 { (len + end).max(0) } else { end.min(len - 1) } as usize; diff --git a/src/config.rs b/src/config.rs index f1e128e..075da06 100644 --- a/src/config.rs +++ b/src/config.rs @@ -34,6 +34,9 @@ pub struct Config { pub slowlog_log_slower_than: i64, /// Maximum number of slow log entries to keep. pub slowlog_max_len: usize, + /// Protected mode: when enabled and no authentication is configured, only + /// loopback clients are accepted. Mirrors Redis's protected-mode default. + pub protected_mode: bool, } impl Default for Config { @@ -62,6 +65,7 @@ impl Default for Config { compression_enabled: false, slowlog_log_slower_than: 10000, slowlog_max_len: 128, + protected_mode: true, } } } @@ -157,6 +161,12 @@ impl Config { config.aof_enabled = args[i] == "yes"; } } + "--protected-mode" => { + i += 1; + if i < args.len() { + config.protected_mode = args[i] == "yes"; + } + } "--appendfilename" => { i += 1; if i < args.len() { diff --git a/src/http.rs b/src/http.rs index ceabe29..a1640a1 100644 --- a/src/http.rs +++ b/src/http.rs @@ -456,12 +456,14 @@ async fn handle_command( cmd_name.to_lowercase() )); } - if args.len() > 1 && !acl::user_has_all_keys(user) { - let key_str = String::from_utf8_lossy(&args[1]); - if !acl::is_key_allowed(user, &key_str) { - return forbidden_response( - "this user has no permissions to access one of the keys used as arguments", - ); + if !acl::user_has_all_keys(user) { + for key in acl::command_keys(&cmd_name, &args) { + let key_str = String::from_utf8_lossy(key); + if !acl::is_key_allowed(user, &key_str) { + return forbidden_response( + "this user has no permissions to access one of the keys used as arguments", + ); + } } } } diff --git a/src/persistence/aof.rs b/src/persistence/aof.rs index 018afdd..bcd5838 100644 --- a/src/persistence/aof.rs +++ b/src/persistence/aof.rs @@ -43,10 +43,10 @@ impl FsyncMode { impl AofWriter { /// Open or create the AOF file for appending. pub fn open(path: &Path, fsync_mode: FsyncMode) -> io::Result { - let file = OpenOptions::new() - .create(true) - .append(true) - .open(path)?; + let mut opts = OpenOptions::new(); + opts.create(true).append(true); + crate::persistence::util::with_secure_mode(&mut opts); + let file = opts.open(path)?; Ok(Self { path: path.to_path_buf(), @@ -90,9 +90,8 @@ impl AofWriter { /// Rewrite the AOF from the current store state. /// Creates a new temp file, writes all current data, then replaces the old file. pub fn rewrite(&mut self, store: &Store) -> io::Result<()> { - let temp_path = self.path.with_extension("aof.tmp"); + let (file, temp_path) = crate::persistence::util::create_temp_file(&self.path)?; { - let file = File::create(&temp_path)?; let mut writer = BufWriter::new(file); let now = Instant::now(); @@ -216,10 +215,10 @@ impl AofWriter { std::fs::rename(&temp_path, &self.path)?; // Re-open the file for appending - let file = OpenOptions::new() - .create(true) - .append(true) - .open(&self.path)?; + let mut opts = OpenOptions::new(); + opts.create(true).append(true); + crate::persistence::util::with_secure_mode(&mut opts); + let file = opts.open(&self.path)?; self.writer = BufWriter::new(file); Ok(()) diff --git a/src/persistence/mod.rs b/src/persistence/mod.rs index 6834c5d..523bb7a 100644 --- a/src/persistence/mod.rs +++ b/src/persistence/mod.rs @@ -1,2 +1,3 @@ pub mod rdb; pub mod aof; +pub mod util; diff --git a/src/persistence/rdb.rs b/src/persistence/rdb.rs index 50ad426..051d8f0 100644 --- a/src/persistence/rdb.rs +++ b/src/persistence/rdb.rs @@ -43,8 +43,7 @@ const RDB_TYPE_JSON: u8 = 6; /// Save the entire store to an RDB file at the given path. pub fn save(store: &Store, path: &Path) -> io::Result<()> { - let temp_path = path.with_extension("rdb.tmp"); - let mut file = std::fs::File::create(&temp_path)?; + let (mut file, temp_path) = crate::persistence::util::create_temp_file(path)?; // Magic header file.write_all(RDB_MAGIC)?; @@ -240,15 +239,22 @@ pub fn load(path: &Path, num_databases: usize) -> io::Result { // Verify CRC64 if present (8 bytes after EOF marker) if cursor + 8 <= data.len() { let stored_crc = u64::from_le_bytes(data[cursor..cursor + 8].try_into().unwrap()); + // A stored CRC of 0 means checksums were disabled when the + // file was written; only enforce when a checksum is present. if stored_crc != 0 { // Compute CRC64 of everything up to (but not including) the CRC bytes // cursor points to right after the EOF opcode, so data[..cursor] includes EOF let computed_crc = crc64_compute(&data[..cursor]); if stored_crc != computed_crc { - tracing::warn!( - "RDB CRC64 mismatch: stored={:#x}, computed={:#x} (continuing anyway)", - stored_crc, computed_crc - ); + // Refuse to load a corrupt/tampered snapshot rather + // than silently trusting mismatched data. + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "RDB CRC64 mismatch: stored={:#x}, computed={:#x}", + stored_crc, computed_crc + ), + )); } } } diff --git a/src/persistence/util.rs b/src/persistence/util.rs new file mode 100644 index 0000000..1f17e56 --- /dev/null +++ b/src/persistence/util.rs @@ -0,0 +1,54 @@ +use std::fs::{File, OpenOptions}; +use std::io; +use std::path::{Path, PathBuf}; + +/// Restrictive permission bits for persistence files (owner read/write only). +/// The RDB/AOF hold the full dataset in cleartext, so they must not be +/// world-readable on a multi-user host. +#[cfg(unix)] +pub const PERSIST_FILE_MODE: u32 = 0o600; + +/// Apply the restrictive persistence-file mode to an `OpenOptions` (no-op on +/// non-Unix targets). +pub fn with_secure_mode(opts: &mut OpenOptions) -> &mut OpenOptions { + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + opts.mode(PERSIST_FILE_MODE); + } + opts +} + +/// Create a fresh temp file next to `final_path` for an atomic +/// write-then-rename. Uses `O_EXCL` (`create_new`) with a randomized suffix so a +/// symlink pre-created at the path cannot be followed (TOCTOU), and restricts +/// permissions to owner-only on Unix. Returns the open handle and its path. +pub fn create_temp_file(final_path: &Path) -> io::Result<(File, PathBuf)> { + use rand::Rng; + let mut rng = rand::thread_rng(); + for _ in 0..8 { + let suffix: u64 = rng.r#gen(); + let temp_path = temp_path_with_suffix(final_path, suffix); + let mut opts = OpenOptions::new(); + opts.write(true).create_new(true); + with_secure_mode(&mut opts); + match opts.open(&temp_path) { + Ok(f) => return Ok((f, temp_path)), + Err(e) if e.kind() == io::ErrorKind::AlreadyExists => continue, + Err(e) => return Err(e), + } + } + Err(io::Error::new( + io::ErrorKind::AlreadyExists, + "could not create a unique temp file after several attempts", + )) +} + +fn temp_path_with_suffix(final_path: &Path, suffix: u64) -> PathBuf { + let mut name = final_path + .file_name() + .map(|s| s.to_os_string()) + .unwrap_or_default(); + name.push(format!(".{:016x}.tmp", suffix)); + final_path.with_file_name(name) +} diff --git a/src/scripting_engine.rs b/src/scripting_engine.rs index 645bea4..b030f3d 100644 --- a/src/scripting_engine.rs +++ b/src/scripting_engine.rs @@ -411,6 +411,21 @@ fn execute_redis_command( ); } + // Block non-deterministic commands whose output could drive a write, so a + // script's effect stays a pure function of its inputs. This keeps scripts + // deterministic for AOF/replication consistency (matches Redis's historical + // deterministic-script requirement). + if matches!( + cmd.as_str(), + "RANDOMKEY" | "SRANDMEMBER" | "ZRANDMEMBER" | "HRANDFIELD" | "TIME" + | "SCAN" | "SSCAN" | "HSCAN" | "ZSCAN" + ) { + return RespValue::error(format!( + "ERR '{}' is not allowed from script: it is non-deterministic", + cmd.to_lowercase() + )); + } + let byte_args: Vec = args.iter().map(|s| Bytes::from(s.clone())).collect(); // Use the command registry to execute diff --git a/src/server/connection.rs b/src/server/connection.rs index 055ca05..a76afb2 100644 --- a/src/server/connection.rs +++ b/src/server/connection.rs @@ -101,12 +101,17 @@ pub struct Connection { tracking_rx: Option>, // Multi-tenancy namespace namespace: Option, - // AUTH brute-force protection: count consecutive failures, reset on success. - auth_failures: u32, + // Peer IP, used for per-IP AUTH brute-force backoff that survives reconnects. + peer_ip: Option, } impl Connection { - pub fn new(stream: MaybeTls, state: Arc, client_id: u64) -> Self { + pub fn new( + stream: MaybeTls, + state: Arc, + client_id: u64, + peer_ip: Option, + ) -> Self { let authenticated = state.config.requirepass.is_none(); let (tx, rx) = mpsc::unbounded_channel(); let (tracking_tx, tracking_rx) = mpsc::unbounded_channel(); @@ -132,8 +137,8 @@ impl Connection { tracked_keys: HashSet::new(), tracking_tx: Some(tracking_tx), tracking_rx: Some(tracking_rx), - auth_failures: 0, namespace: None, + peer_ip, } } @@ -375,14 +380,16 @@ impl Connection { cmd_name.to_lowercase() )); } - // Check key patterns for commands that have keys - if args.len() > 1 && !acl::user_has_all_keys(&self.auth_username) { - let key_str = String::from_utf8_lossy(&args[1]); - if !acl::is_key_allowed(&self.auth_username, &key_str) { - return RespValue::error( - "NOPERM this user has no permissions to access one of the keys used as arguments" - .to_string(), - ); + // Check key patterns against every key argument of this command. + if !acl::user_has_all_keys(&self.auth_username) { + for key in acl::command_keys(&cmd_name, &args) { + let key_str = String::from_utf8_lossy(key); + if !acl::is_key_allowed(&self.auth_username, &key_str) { + return RespValue::error( + "NOPERM this user has no permissions to access one of the keys used as arguments" + .to_string(), + ); + } } } } @@ -820,18 +827,38 @@ impl Connection { } } - /// After `BACKOFF_AFTER` consecutive failed AUTH/HELLO attempts on this - /// connection, sleep before returning the failure response. Doubles each - /// time, capped at 5 s. Resets to zero on a successful auth. + /// Record a failed AUTH/HELLO attempt keyed by peer IP and, once more than + /// `BACKOFF_AFTER` failures have accumulated for that IP, sleep with + /// exponential backoff before returning the failure. Tracking per IP (rather + /// than per connection) means an attacker cannot reset the delay by opening + /// a new connection for each guess. The counter decays after an idle window. async fn auth_backoff(&mut self) { const BACKOFF_AFTER: u32 = 5; const MAX_BACKOFF: std::time::Duration = std::time::Duration::from_secs(5); + const DECAY: std::time::Duration = std::time::Duration::from_secs(60); + + let ip = match self.peer_ip { + Some(ip) => ip, + None => return, + }; + + let failures = { + let now = std::time::Instant::now(); + let mut map = self.state.auth_failures.lock().await; + let entry = map.entry(ip).or_insert((0, now)); + // Reset the counter if the last failure was long enough ago. + if now.duration_since(entry.1) > DECAY { + entry.0 = 0; + } + entry.0 = entry.0.saturating_add(1); + entry.1 = now; + entry.0 + }; - self.auth_failures = self.auth_failures.saturating_add(1); - if self.auth_failures <= BACKOFF_AFTER { + if failures <= BACKOFF_AFTER { return; } - let over = self.auth_failures - BACKOFF_AFTER; + let over = failures - BACKOFF_AFTER; let ms: u64 = 100u64.saturating_mul(1u64 << over.min(10)); let delay = std::time::Duration::from_millis(ms).min(MAX_BACKOFF); tokio::time::sleep(delay).await; @@ -840,7 +867,12 @@ impl Connection { fn auth_success(&mut self, username: String) { self.authenticated = true; self.auth_username = username; - self.auth_failures = 0; + // Clear this IP's accumulated failures on a successful auth. + if let Some(ip) = self.peer_ip { + if let Ok(mut map) = self.state.auth_failures.try_lock() { + map.remove(&ip); + } + } } async fn handle_auth(&mut self, args: &[Bytes]) -> RespValue { @@ -908,6 +940,9 @@ impl Connection { // === Pub/Sub handlers === async fn handle_subscribe(&mut self, args: &[Bytes]) -> RespValue { + if args.len() < 2 { + return RespValue::error("ERR wrong number of arguments for 'subscribe' command"); + } let channels: Vec = args[1..].to_vec(); let mut responses = Vec::new(); @@ -981,6 +1016,9 @@ impl Connection { } async fn handle_psubscribe(&mut self, args: &[Bytes]) -> RespValue { + if args.len() < 2 { + return RespValue::error("ERR wrong number of arguments for 'psubscribe' command"); + } let patterns: Vec = args[1..].to_vec(); let mut responses = Vec::new(); diff --git a/src/server/mod.rs b/src/server/mod.rs index 0f883df..47638b9 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -198,6 +198,9 @@ pub struct SharedState { pub slowlog_next_id: AtomicU64, /// Named namespaces (multi-tenancy). pub namespaces: Mutex>, + /// Per-IP AUTH failure tracking for brute-force backoff that survives + /// reconnects: maps a peer IP to (consecutive failures, last-failure time). + pub auth_failures: Mutex>, /// Client tracking state. A single mutex protects the forward index /// (client_id -> keys it tracks), the senders, and the reverse index /// (key -> client_ids interested in it). Consolidating into one mutex @@ -315,6 +318,7 @@ impl Server { slowlog_next_id: AtomicU64::new(0), namespaces: Mutex::new(HashMap::new()), tracking: Mutex::new(TrackingState::default()), + auth_failures: Mutex::new(HashMap::new()), }); // Start HTTP/REST API server if configured @@ -367,6 +371,13 @@ impl Server { }; let state = Arc::clone(&tls_state); + + // Protected mode: refuse non-loopback clients when no auth is set. + if protected_mode_reject(&state, addr.ip()) { + tracing::warn!("Protected mode: rejecting TLS connection from {}", addr); + continue; + } + let permit = match tls_semaphore.clone().try_acquire_owned() { Ok(p) => p, Err(_) => { @@ -392,7 +403,7 @@ impl Server { 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); + let mut conn = connection::Connection::new(stream, state.clone(), client_id, Some(addr.ip())); if let Err(e) = conn.handle().await { tracing::debug!("TLS connection {} error: {}", addr, e); } @@ -415,6 +426,19 @@ impl Server { loop { let (socket, addr) = listener.accept().await?; let state = Arc::clone(&state); + + // Protected mode: refuse non-loopback clients when no auth is set. + if protected_mode_reject(&state, addr.ip()) { + tracing::warn!("Protected mode: rejecting connection from {}", addr); + let mut socket = socket; + tokio::spawn(async move { + use tokio::io::AsyncWriteExt; + let _ = socket.write_all(PROTECTED_MODE_ERROR).await; + let _ = socket.shutdown().await; + }); + continue; + } + let permit = match semaphore.clone().try_acquire_owned() { Ok(p) => p, Err(_) => { @@ -430,7 +454,7 @@ impl Server { tokio::spawn(async move { tracing::debug!("New connection from {} (client_id={})", addr, client_id); let stream = connection::MaybeTls::Plain(socket); - let mut conn = connection::Connection::new(stream, state.clone(), client_id); + let mut conn = connection::Connection::new(stream, state.clone(), client_id, Some(addr.ip())); if let Err(e) = conn.handle().await { tracing::debug!("Connection {} error: {}", addr, e); } @@ -442,6 +466,21 @@ impl Server { } } +/// RESP error sent to a client rejected by protected mode. +const PROTECTED_MODE_ERROR: &[u8] = + b"-DENIED rCache is running in protected mode because protected mode is enabled \ +and no authentication is configured. Connect from the loopback interface, set a \ +password (requirepass / ACL), or disable protected mode with '--protected-mode no'.\r\n"; + +/// Whether a peer should be refused under protected mode: it is enabled, no +/// authentication is configured, and the peer is not on the loopback interface. +fn protected_mode_reject(state: &SharedState, ip: std::net::IpAddr) -> bool { + state.config.protected_mode + && state.config.requirepass.is_none() + && !crate::command::acl::any_password_required() + && !ip.is_loopback() +} + /// Build a TLS acceptor from PEM cert and key files. fn build_tls_acceptor( cert_path: &str,