diff --git a/README.md b/README.md index db157bc5..8aff7a92 100644 --- a/README.md +++ b/README.md @@ -15,8 +15,10 @@ a low-latency, memory-efficient, distributed cache written in Rust. designed to - **hashes** — HSET, HGET, HGETALL, HDEL, HEXISTS, HLEN, HINCRBY, HKEYS, HVALS, HMGET - **sets** — SADD, SREM, SMEMBERS, SISMEMBER, SCARD - **key commands** — DEL, EXISTS, EXPIRE, TTL, PEXPIRE, PTTL, PERSIST, TYPE, SCAN, KEYS, RENAME -- **server commands** — PING, ECHO, INFO, DBSIZE, FLUSHDB, BGSAVE, BGREWRITEAOF +- **server commands** — PING, ECHO, INFO, DBSIZE, FLUSHDB, BGSAVE, BGREWRITEAOF, AUTH, QUIT - **pub/sub** — SUBSCRIBE, UNSUBSCRIBE, PSUBSCRIBE, PUNSUBSCRIBE, PUBLISH, plus PUBSUB introspection +- **authentication** — `--requirepass` for redis-compatible AUTH (legacy and username/password forms) +- **protected mode** — rejects non-loopback connections when no password is set on public binds - **observability** — prometheus metrics (`--metrics-port`), enriched INFO with 6 sections, SLOWLOG command - **sharded engine** — shared-nothing, thread-per-core design with no cross-shard locking - **concurrent mode** — experimental DashMap-backed keyspace for lock-free GET/SET (2x faster than Redis) @@ -108,6 +110,7 @@ redis-cli FLUSHDB # => OK | `--slowlog-log-slower-than` | 10000 | log commands slower than N microseconds (-1 disables) | | `--slowlog-max-len` | 128 | max entries in slow log ring buffer | | `--concurrent` | false | use DashMap-backed keyspace (experimental, faster GET/SET) | +| `--requirepass` | — | require AUTH with this password before running commands | ## build & development @@ -184,7 +187,7 @@ contributions welcome — see [CONTRIBUTING.md](CONTRIBUTING.md). | 4 | clustering (raft, gossip, slots, migration) | ✅ complete | | 5 | developer experience (observability, CLI, clients) | 🚧 in progress | -**current**: 83 commands, 695 tests, ~22k lines of code +**current**: 85 commands, 701 tests, ~23k lines of code ## security @@ -193,7 +196,7 @@ see [SECURITY.md](SECURITY.md) for: - security considerations for deployment - recommended configuration -**note**: ember does not currently support authentication. always run behind a firewall or in a trusted network. +**note**: use `--requirepass` to enable authentication. protected mode is active by default when no password is set, rejecting non-loopback connections on public binds. ## license diff --git a/crates/ember-protocol/src/command.rs b/crates/ember-protocol/src/command.rs index 007f2be8..dbf3a887 100644 --- a/crates/ember-protocol/src/command.rs +++ b/crates/ember-protocol/src/command.rs @@ -318,6 +318,17 @@ pub enum Command { /// PUBSUB NUMPAT. Returns the number of active pattern subscriptions. PubSubNumPat, + /// AUTH \[username\] password. Authenticate the connection. + Auth { + /// Username for ACL-style auth. None for legacy AUTH. + username: Option, + /// The password to validate. + password: String, + }, + + /// QUIT. Requests the server to close the connection. + Quit, + /// A command we don't recognize (yet). Unknown(String), } @@ -432,6 +443,8 @@ impl Command { Command::PubSubChannels { .. } => "pubsub", Command::PubSubNumSub { .. } => "pubsub", Command::PubSubNumPat => "pubsub", + Command::Auth { .. } => "auth", + Command::Quit => "quit", Command::Unknown(_) => "unknown", } } @@ -526,6 +539,8 @@ impl Command { "PUNSUBSCRIBE" => parse_punsubscribe(&frames[1..]), "PUBLISH" => parse_publish(&frames[1..]), "PUBSUB" => parse_pubsub(&frames[1..]), + "AUTH" => parse_auth(&frames[1..]), + "QUIT" => parse_quit(&frames[1..]), _ => Ok(Command::Unknown(name)), } } @@ -1657,6 +1672,34 @@ fn parse_pubsub(args: &[Frame]) -> Result { } } +fn parse_auth(args: &[Frame]) -> Result { + match args.len() { + 1 => { + let password = extract_string(&args[0])?; + Ok(Command::Auth { + username: None, + password, + }) + } + 2 => { + let username = extract_string(&args[0])?; + let password = extract_string(&args[1])?; + Ok(Command::Auth { + username: Some(username), + password, + }) + } + _ => Err(ProtocolError::WrongArity("AUTH".into())), + } +} + +fn parse_quit(args: &[Frame]) -> Result { + if !args.is_empty() { + return Err(ProtocolError::WrongArity("QUIT".into())); + } + Ok(Command::Quit) +} + #[cfg(test)] mod tests { use super::*; @@ -3764,4 +3807,53 @@ mod tests { let err = Command::from_frame(cmd(&["RENAME", "only"])).unwrap_err(); assert!(matches!(err, ProtocolError::WrongArity(_))); } + + // --- AUTH --- + + #[test] + fn auth_legacy() { + assert_eq!( + Command::from_frame(cmd(&["AUTH", "secret"])).unwrap(), + Command::Auth { + username: None, + password: "secret".into() + }, + ); + } + + #[test] + fn auth_with_username() { + assert_eq!( + Command::from_frame(cmd(&["AUTH", "default", "secret"])).unwrap(), + Command::Auth { + username: Some("default".into()), + password: "secret".into() + }, + ); + } + + #[test] + fn auth_no_args() { + let err = Command::from_frame(cmd(&["AUTH"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + } + + #[test] + fn auth_too_many_args() { + let err = Command::from_frame(cmd(&["AUTH", "a", "b", "c"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + } + + // --- QUIT --- + + #[test] + fn quit_basic() { + assert_eq!(Command::from_frame(cmd(&["QUIT"])).unwrap(), Command::Quit,); + } + + #[test] + fn quit_wrong_arity() { + let err = Command::from_frame(cmd(&["QUIT", "extra"])).unwrap_err(); + assert!(matches!(err, ProtocolError::WrongArity(_))); + } } diff --git a/crates/ember-server/src/concurrent_handler.rs b/crates/ember-server/src/concurrent_handler.rs index 05a231d4..17438fa6 100644 --- a/crates/ember-server/src/concurrent_handler.rs +++ b/crates/ember-server/src/concurrent_handler.rs @@ -22,7 +22,9 @@ use ember_protocol::{parse_frame, Command, Frame, SetExpire}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; -use crate::connection_common::{BUF_CAPACITY, IDLE_TIMEOUT, MAX_BUF_SIZE}; +use crate::connection_common::{ + is_allowed_before_auth, is_auth_frame, try_auth, BUF_CAPACITY, IDLE_TIMEOUT, MAX_BUF_SIZE, +}; use crate::pubsub::PubSubManager; use crate::server::ServerContext; use crate::slowlog::SlowLog; @@ -38,6 +40,8 @@ pub async fn handle( ) -> Result<(), Box> { stream.set_nodelay(true)?; + let mut authenticated = ctx.requirepass.is_none(); + let mut buf = BytesMut::with_capacity(BUF_CAPACITY); let mut out = BytesMut::with_capacity(BUF_CAPACITY); @@ -62,8 +66,27 @@ pub async fn handle( match parse_frame(&buf) { Ok(Some((frame, consumed))) => { let _ = buf.split_to(consumed); - let response = process(frame, &keyspace, &engine, ctx, slow_log, pubsub).await; - response.serialize(&mut out); + + if !authenticated { + if is_auth_frame(&frame) { + let (response, success) = try_auth(frame, ctx); + response.serialize(&mut out); + if success { + authenticated = true; + } + } else if is_allowed_before_auth(&frame) { + let response = + process(frame, &keyspace, &engine, ctx, slow_log, pubsub).await; + response.serialize(&mut out); + } else { + Frame::Error("NOAUTH Authentication required.".into()) + .serialize(&mut out); + } + } else { + let response = + process(frame, &keyspace, &engine, ctx, slow_log, pubsub).await; + response.serialize(&mut out); + } } Ok(None) => break, Err(e) => { @@ -99,7 +122,7 @@ async fn process( None }; - let response = execute_concurrent(cmd, keyspace, engine, pubsub).await; + let response = execute_concurrent(cmd, keyspace, engine, ctx, pubsub).await; ctx.commands_processed.fetch_add(1, Ordering::Relaxed); if let Some(start) = start { @@ -122,6 +145,7 @@ async fn execute_concurrent( cmd: Command, keyspace: &Arc, _engine: &Engine, + ctx: &Arc, pubsub: &Arc, ) -> Frame { match cmd { @@ -233,6 +257,37 @@ async fn execute_concurrent( Frame::Error("ERR pub/sub not supported in concurrent mode yet".into()) } + // AUTH on an already-authenticated connection (re-auth) + Command::Auth { username, password } => match &ctx.requirepass { + None => Frame::Error( + "ERR Client sent AUTH, but no password is set. \ + Did you mean ACL SETUSER with >password?" + .into(), + ), + Some(expected) => { + if let Some(ref user) = username { + if user != "default" { + return Frame::Error( + "WRONGPASS invalid username-password pair \ + or user is disabled." + .into(), + ); + } + } + if password == *expected { + Frame::Simple("OK".into()) + } else { + Frame::Error( + "WRONGPASS invalid username-password pair \ + or user is disabled." + .into(), + ) + } + } + }, + + Command::Quit => Frame::Simple("OK".into()), + // For unsupported commands, return an error Command::Unknown(name) => Frame::Error(format!("ERR unknown command '{name}'")), diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs index dd25a75c..ce181e57 100644 --- a/crates/ember-server/src/connection.rs +++ b/crates/ember-server/src/connection.rs @@ -18,7 +18,9 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; use tokio::sync::broadcast; -use crate::connection_common::{BUF_CAPACITY, IDLE_TIMEOUT, MAX_BUF_SIZE}; +use crate::connection_common::{ + is_allowed_before_auth, is_auth_frame, try_auth, BUF_CAPACITY, IDLE_TIMEOUT, MAX_BUF_SIZE, +}; use crate::pubsub::{PubMessage, PubSubManager}; use crate::server::ServerContext; use crate::slowlog::SlowLog; @@ -39,6 +41,9 @@ pub async fn handle( // and we already batch responses from pipelining into a single write stream.set_nodelay(true)?; + // per-connection auth state. auto-authenticated when no password is set. + let mut authenticated = ctx.requirepass.is_none(); + let mut buf = BytesMut::with_capacity(BUF_CAPACITY); let mut out = BytesMut::with_capacity(BUF_CAPACITY); @@ -82,6 +87,29 @@ pub async fn handle( } } + // when not yet authenticated, process frames serially so that an + // AUTH command in a pipeline takes effect for subsequent frames + if !authenticated { + for frame in frames { + if is_auth_frame(&frame) { + let (response, success) = try_auth(frame, ctx); + response.serialize(&mut out); + if success { + authenticated = true; + } + } else if is_allowed_before_auth(&frame) { + let response = process(frame, &engine, ctx, slow_log, pubsub).await; + response.serialize(&mut out); + } else { + Frame::Error("NOAUTH Authentication required.".into()).serialize(&mut out); + } + } + if !out.is_empty() { + stream.write_all(&out).await?; + } + continue; + } + // check if any frame is a subscribe command — if so, we need // to enter subscriber mode which changes the connection loop let enter_sub = frames.iter().any(is_subscribe_frame); @@ -1414,6 +1442,37 @@ async fn execute( Frame::Error("ERR subscribe commands should not reach execute".into()) } + // AUTH on an already-authenticated connection (re-auth) + Command::Auth { username, password } => match &ctx.requirepass { + None => Frame::Error( + "ERR Client sent AUTH, but no password is set. \ + Did you mean ACL SETUSER with >password?" + .into(), + ), + Some(expected) => { + if let Some(ref user) = username { + if user != "default" { + return Frame::Error( + "WRONGPASS invalid username-password pair \ + or user is disabled." + .into(), + ); + } + } + if password == *expected { + Frame::Simple("OK".into()) + } else { + Frame::Error( + "WRONGPASS invalid username-password pair \ + or user is disabled." + .into(), + ) + } + } + }, + + Command::Quit => Frame::Simple("OK".into()), + Command::Unknown(name) => Frame::Error(format!("ERR unknown command '{name}'")), } } diff --git a/crates/ember-server/src/connection_common.rs b/crates/ember-server/src/connection_common.rs index a3f03783..e450b7dd 100644 --- a/crates/ember-server/src/connection_common.rs +++ b/crates/ember-server/src/connection_common.rs @@ -6,6 +6,11 @@ use std::time::Duration; +use ember_protocol::types::Frame; +use ember_protocol::Command; + +use crate::server::ServerContext; + /// Initial read buffer capacity. 4KB covers most commands comfortably /// without over-allocating for simple PING/SET/GET workloads. pub const BUF_CAPACITY: usize = 4096; @@ -20,3 +25,84 @@ pub const MAX_BUF_SIZE: usize = 64 * 1024 * 1024; /// close it. Prevents abandoned connections from leaking resources. /// 5 minutes matches Redis default behavior. pub const IDLE_TIMEOUT: Duration = Duration::from_secs(300); + +/// Checks if a raw frame is an AUTH command (before full parsing). +/// +/// Peeks at the first bulk element to avoid a full `Command::from_frame` +/// round-trip on unauthenticated connections. +pub fn is_auth_frame(frame: &Frame) -> bool { + if let Frame::Array(parts) = frame { + if let Some(Frame::Bulk(name)) = parts.first() { + return name.eq_ignore_ascii_case(b"AUTH"); + } + } + false +} + +/// Checks if a raw frame represents a command allowed before authentication. +/// +/// Per Redis semantics, only AUTH, PING, ECHO, and QUIT are permitted +/// on unauthenticated connections. +pub fn is_allowed_before_auth(frame: &Frame) -> bool { + if let Frame::Array(parts) = frame { + if let Some(Frame::Bulk(name)) = parts.first() { + return name.eq_ignore_ascii_case(b"AUTH") + || name.eq_ignore_ascii_case(b"PING") + || name.eq_ignore_ascii_case(b"ECHO") + || name.eq_ignore_ascii_case(b"QUIT"); + } + } + false +} + +/// Attempts to authenticate using an AUTH frame. +/// +/// Returns `(response_frame, authenticated)`. The caller should flip +/// their per-connection auth state when `authenticated` is true. +pub fn try_auth(frame: Frame, ctx: &ServerContext) -> (Frame, bool) { + let cmd = match Command::from_frame(frame) { + Ok(cmd) => cmd, + Err(e) => return (Frame::Error(format!("ERR {e}")), false), + }; + + match cmd { + Command::Auth { username, password } => match &ctx.requirepass { + None => ( + Frame::Error( + "ERR Client sent AUTH, but no password is set. \ + Did you mean ACL SETUSER with >password?" + .into(), + ), + false, + ), + Some(expected) => { + // only the "default" username is accepted (no full ACL yet) + if let Some(ref user) = username { + if user != "default" { + return ( + Frame::Error( + "WRONGPASS invalid username-password pair \ + or user is disabled." + .into(), + ), + false, + ); + } + } + if password == *expected { + (Frame::Simple("OK".into()), true) + } else { + ( + Frame::Error( + "WRONGPASS invalid username-password pair \ + or user is disabled." + .into(), + ), + false, + ) + } + } + }, + _ => (Frame::Error("ERR expected AUTH command".into()), false), + } +} diff --git a/crates/ember-server/src/main.rs b/crates/ember-server/src/main.rs index 7ec3c698..a50bfe33 100644 --- a/crates/ember-server/src/main.rs +++ b/crates/ember-server/src/main.rs @@ -76,6 +76,11 @@ struct Args { /// experimental: bypasses channel overhead for GET/SET commands. #[arg(long)] concurrent: bool, + + /// require clients to AUTH with this password before running commands. + /// when set, connections must authenticate before executing any data commands. + #[arg(long)] + requirepass: Option, } #[tokio::main] @@ -182,6 +187,10 @@ async fn main() { "ember server starting..." ); + if args.requirepass.is_some() { + info!("authentication enabled (requirepass set)"); + } + let result = if args.concurrent { server::run_concurrent( addr, @@ -192,6 +201,7 @@ async fn main() { None, args.metrics_port.is_some(), slowlog_config, + args.requirepass, ) .await } else { @@ -202,6 +212,7 @@ async fn main() { None, args.metrics_port.is_some(), slowlog_config, + args.requirepass, ) .await }; diff --git a/crates/ember-server/src/server.rs b/crates/ember-server/src/server.rs index 7badf5af..7d39642b 100644 --- a/crates/ember-server/src/server.rs +++ b/crates/ember-server/src/server.rs @@ -9,6 +9,7 @@ use std::sync::Arc; use std::time::Instant; use ember_core::{ConcurrentKeyspace, Engine, EngineConfig, EvictionPolicy}; +use tokio::io::AsyncWriteExt; use tokio::net::TcpListener; use tokio::sync::Semaphore; use tracing::{error, info, warn}; @@ -36,6 +37,10 @@ pub struct ServerContext { pub connections_accepted: AtomicU64, pub connections_active: AtomicU64, pub commands_processed: AtomicU64, + /// Password required for AUTH. None means no authentication needed. + pub requirepass: Option, + /// The address the server is bound to (for protected mode checks). + pub bind_addr: SocketAddr, } /// Binds to `addr` and runs the accept loop. @@ -54,6 +59,7 @@ pub async fn run( max_connections: Option, metrics_enabled: bool, slowlog_config: SlowLogConfig, + requirepass: Option, ) -> Result<(), Box> { // ensure data directory exists if persistence is configured if let Some(ref pcfg) = config.persistence { @@ -91,6 +97,8 @@ pub async fn run( connections_accepted: AtomicU64::new(0), connections_active: AtomicU64::new(0), commands_processed: AtomicU64::new(0), + requirepass, + bind_addr: addr, }); let slow_log = Arc::new(SlowLog::new(slowlog_config)); @@ -114,7 +122,20 @@ pub async fn run( } result = listener.accept() => { - let (stream, peer) = result?; + let (mut stream, peer) = result?; + + // protected mode: reject non-loopback connections when no + // password is set and the server is bound to a public address + if is_protected_mode_violation(&ctx, &peer) { + let msg = "-DENIED Ember is running in protected mode \ + because no password is set. In this mode \ + connections are only accepted from the loopback \ + interface. Set a password with --requirepass or \ + bind to 127.0.0.1 to resolve this.\r\n"; + let _ = stream.write_all(msg.as_bytes()).await; + let _ = stream.shutdown().await; + continue; + } let permit = match semaphore.clone().try_acquire_owned() { Ok(permit) => permit, @@ -177,6 +198,7 @@ pub async fn run_concurrent( max_connections: Option, metrics_enabled: bool, slowlog_config: SlowLogConfig, + requirepass: Option, ) -> Result<(), Box> { let aof_enabled = config .persistence @@ -209,6 +231,8 @@ pub async fn run_concurrent( connections_accepted: AtomicU64::new(0), connections_active: AtomicU64::new(0), commands_processed: AtomicU64::new(0), + requirepass, + bind_addr: addr, }); let slow_log = Arc::new(SlowLog::new(slowlog_config)); @@ -229,7 +253,18 @@ pub async fn run_concurrent( } result = listener.accept() => { - let (stream, peer) = result?; + let (mut stream, peer) = result?; + + if is_protected_mode_violation(&ctx, &peer) { + let msg = "-DENIED Ember is running in protected mode \ + because no password is set. In this mode \ + connections are only accepted from the loopback \ + interface. Set a password with --requirepass or \ + bind to 127.0.0.1 to resolve this.\r\n"; + let _ = stream.write_all(msg.as_bytes()).await; + let _ = stream.shutdown().await; + continue; + } let permit = match semaphore.clone().try_acquire_owned() { Ok(permit) => permit, @@ -277,3 +312,19 @@ pub async fn run_concurrent( Ok(()) } + +/// Returns true if the connection should be rejected by protected mode. +/// +/// Protected mode activates when all three conditions hold: +/// 1. No password is configured (requirepass is None) +/// 2. The server is bound to a non-loopback address (e.g. 0.0.0.0) +/// 3. The connecting client is from a non-loopback address +fn is_protected_mode_violation(ctx: &ServerContext, peer: &SocketAddr) -> bool { + if ctx.requirepass.is_some() { + return false; + } + if ctx.bind_addr.ip().is_loopback() { + return false; + } + !peer.ip().is_loopback() +}