From 755b83bc06b47f3e1055ceda793b4104f73a8043 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 30 Jul 2026 23:48:28 +0700 Subject: [PATCH 1/5] fix(security): run the ACL gate before every privileged intercept; deny unknown users; track post-AUTH identity (c10k B1/B2/B3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three ACL/auth defects from the c10k hardening review, all in the same blast radius: an authenticated account with minimal grants could take the server over, and revoking that account could not stop it. B1 — privileged intercepts ran ABOVE the ACL gate. Both connection handlers intercept EVAL/EVALSHA/SCRIPT, ACL and CLUSTER before the permission check, and each intercept `continue`s on a match, so the gate below never ran. The comment attached to that gate claimed the exact invariant the code above it violated ("must run before any command-specific handlers ... so that low-privilege users cannot reach admin commands"), which is presumably why it survived so long. Verified end-to-end before the fix: a user holding nothing but `~app:* +get` is correctly refused a plain SET, and then succeeds at ACL SETUSER evil on nopass ~* +@all -> +OK, persisted CONFIG SET maxmemory 999999999 -> +OK, persisted EVAL "return 'pwned'" 0 -> "pwned" SCRIPT LOAD / REPLICAOF / BGSAVE -> all accepted The gate now sits above every privileged intercept in both handlers, with AUTH and HELLO lifted above IT — Redis marks both NO_AUTH, and gating them would strand a restricted client with no way to re-authenticate and break the RESP3 handshake. Existing consistency suites authenticate as `default` (unrestricted), which takes the `acl_skip_allowed` fast path, so they were blind to all of this. B2 — unknown users failed OPEN. check_command_permission, check_key_permission and check_channel_permission each began with `self.users.get(username)?`, and None is their "allowed" answer. Nothing closes a live session when its account disappears, so `ACL DELUSER alice` or an `ACL LOAD` that drops her silently PROMOTED every connection alice still had open to full `~* +@all`. Unknown users are denied now. The matching guard is `ensure_default_user`: `default` is inserted (from requirepass, exactly as the no-aclfile path does) after any load that could omit it, so fail-closed cannot brick a server whose ACL file never mentions `default`. B3 — CLIENT KILL USER was inert. `client_registry::register` captures the username once, at accept time, when it is always `default`; every AUTH/HELLO success updated only the connection-local copy. So CLIENT LIST reported `user=default` for every session and `CLIENT KILL USER alice` returned 0 — the primary incident-response lever for a compromised credential, useless. All 13 assignment sites across the three handlers now go through one `ConnectionState::adopt_user`, which publishes to the registry as well as refreshing the ACL cache. With identity tracked, ACL DELUSER also disconnects the sessions the deleted user still holds, as Redis does. Tests (red proven by reverting src/ and re-running, both shard configs): - tests/acl_privileged_intercepts.rs — 9 escalation vectors must be NOPERM, with a GET positive control, no-side-effect checks on the denied ACL SETUSER / CONFIG SET, and AUTH/HELLO still exempt. RED: `ACL SETUSER evil ...` returned +OK at shards=1 and shards=4. - tests/acl_user_revocation.rs — CLIENT LIST reports the post-AUTH user, CLIENT KILL USER tears the session down and leaves others alone, ACL DELUSER disconnects. RED: CLIENT LIST showed `user=default`; DELUSER left bob connected. - src/acl/table.rs unit tests for unknown-user denial and ensure_default_user (creates only when missing, honours requirepass). Gates: fmt; clippy -D warnings on default and runtime-tokio,jemalloc (macOS + Linux VM); lib 4469 monoio / 3631 tokio; the conn/auth integration suites (maxclients_reject_parity, parked_idle_parity, idle_downshift_parity, tls_idle_downshift_parity, client_tracking_invalidation) green on both runtimes. Refs: tmp/C10K-HARDENING-REVIEW.md cluster B. author: Tin Dang --- CHANGELOG.md | 44 ++++ src/acl/io.rs | 7 +- src/acl/table.rs | 113 ++++++++- src/command/acl.rs | 20 ++ src/server/conn/core.rs | 24 ++ src/server/conn/handler_monoio/dispatch.rs | 12 +- src/server/conn/handler_monoio/mod.rs | 78 +++++-- src/server/conn/handler_sharded/mod.rs | 176 +++++++------- src/server/conn/handler_single.rs | 15 +- tests/acl_privileged_intercepts.rs | 259 +++++++++++++++++++++ tests/acl_user_revocation.rs | 254 ++++++++++++++++++++ 11 files changed, 877 insertions(+), 125 deletions(-) create mode 100644 tests/acl_privileged_intercepts.rs create mode 100644 tests/acl_user_revocation.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 524dcd164..90ec65351 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,51 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Security +- **Privileged commands ran BEFORE the ACL permission check (c10k B1).** + Both connection handlers intercepted `EVAL`/`EVALSHA`/`SCRIPT`, `ACL`, and + `CLUSTER` above the ACL gate, and each intercept `continue`s on a match, so + the gate below it never ran — even though the comment attached to that gate + claimed it "must run before any command-specific handlers ... so that + low-privilege users cannot reach admin commands". Any authenticated account + could escalate to full admin: a user holding nothing but `~app:* +get` was + correctly refused a plain `SET` yet could run + `ACL SETUSER evil on nopass ~* +@all` (persisted), `CONFIG SET` (persisted), + arbitrary Lua via `EVAL`, `SCRIPT LOAD`, `REPLICAOF` and `BGSAVE`. The gate + now sits above every privileged intercept in both handlers, with `AUTH` and + `HELLO` lifted above it as Redis's `NO_AUTH` commands. Verified end-to-end at + 1 and 4 shards (`tests/acl_privileged_intercepts.rs`). +- **Unknown ACL users failed OPEN (c10k B2).** All three permission checks + started with `users.get(username)?`, and `None` is their "allowed" answer, so + a username missing from the table was granted everything. Nothing closes a + live session when its account disappears, so `ACL DELUSER alice` / `ACL LOAD` + silently PROMOTED every connection alice still had open to `~* +@all`. + Unknown users are now denied; `default` is guaranteed present after any load + (`ensure_default_user`) so a server whose ACL file omits it is not bricked. +- **`CLIENT KILL USER` was inert and `CLIENT LIST` reported `user=default` for + everyone (c10k B3).** The client registry captured the username once, at + accept time; AUTH/HELLO updated only the connection-local copy. The primary + incident-response lever for a compromised credential matched nothing. AUTH + and HELLO now publish the adopted identity to the registry, and — matching + Redis — `ACL DELUSER` disconnects the sessions the deleted user still holds. +- **The experimental io_uring bridge served commands with no auth (c10k B4).** + `MOON_URING=1` (tokio, Linux) binds a second `SO_REUSEPORT` listener on the + server's own port whose accept path has no auth gate, no ACL check and no + client registry — so on a server with `requirepass`/`aclfile` configured the + kernel load-balanced roughly half of all new connections onto a listener that + skipped authentication entirely. The documented limitation named maxclients + and `CLIENT LIST`/`KILL`, not this. The bridge now refuses to arm (loudly) + when authentication is configured and the shard stays on the tokio path. + ### Fixed +- **TLS park veto samples `wants_write()` after processing, not before (c10k + B5).** `Stream::task_park_safe` read `wants_write()` before + `process_new_packets()`, which can queue outbound bytes and still return + `Ok` — reachable in rustls 0.23 via the TLS 1.2 renegotiation rebuff + (`NoRenegotiation` warning alert) — so the connection could park owing the + peer a reply. (The TLS 1.3 KeyUpdate variant is *not* reachable: rustls + defers that reply until the next outbound record; `tests/tls_park_keyupdate.rs` + pins the behaviour so a future rustls change is caught.) - **A client that disconnects while blocked is now reaped (c10k hardening A1).** The infinite-wait `select!` behind `BLPOP`/`BRPOP`/`BLMOVE`/ `BZPOPMIN`/`BZPOPMAX`/`BLMPOP`/`BZMPOP`/`BRPOPLPUSH` had exactly two arms, diff --git a/src/acl/io.rs b/src/acl/io.rs index 1b2b3bd98..f00ec87af 100644 --- a/src/acl/io.rs +++ b/src/acl/io.rs @@ -113,7 +113,12 @@ pub fn acl_load(path: &str) -> std::io::Result { pub fn acl_table_from_config(config: &crate::config::ServerConfig) -> AclTable { if let Some(ref path) = config.aclfile { match acl_load(path) { - Ok(table) => { + Ok(mut table) => { + // c10k B2: the permission checks now DENY an unknown user, so + // an ACL file that never defines `default` would lock out + // every connection. Redis guarantees `default` exists; so do + // we, seeded from requirepass exactly as the no-file path below. + table.ensure_default_user(config.requirepass.as_deref()); return table; } Err(e) if e.kind() == std::io::ErrorKind::NotFound => { diff --git a/src/acl/table.rs b/src/acl/table.rs index e88ed78d2..bee991e74 100644 --- a/src/acl/table.rs +++ b/src/acl/table.rs @@ -299,6 +299,26 @@ impl AclTable { self.bump_version(); } + /// Guarantee the `default` user exists, creating it from `requirepass` + /// (or nopass) if an ACL file did not define it. No-op when present. + /// + /// c10k hardening B2 support. The permission checks below now DENY an + /// unknown user instead of falling open, so `default` missing from the + /// table would lock out every connection on a server with no auth + /// configured at all. Redis likewise guarantees `default` always exists. + /// Call after any load that can produce an arbitrary user set (startup + /// aclfile load, `ACL LOAD`). + pub fn ensure_default_user(&mut self, requirepass: Option<&str>) { + if self.users.contains_key("default") { + return; + } + let user = match requirepass { + Some(p) if !p.is_empty() => AclUser::new_default_with_password(p), + _ => AclUser::new_default_nopass(), + }; + self.set_user("default".to_string(), user); + } + /// Bootstrap from ServerConfig. Loads aclfile if configured, otherwise creates /// the default user from requirepass (or nopass). pub fn load_or_default(config: &ServerConfig) -> Self { @@ -386,7 +406,16 @@ impl AclTable { cmd: &[u8], _args: &[Frame], ) -> Option { - let user = self.users.get(username)?; + // c10k hardening B2: an unknown user is DENIED, never allowed. + // This used to be `self.users.get(username)?` - `None` means + // "allowed", so a user that vanished from the table fell open to + // full `~* +@all`. `ACL DELUSER alice` / `ACL LOAD` do not close + // alice's live sessions, so revoking her account silently + // PROMOTED every connection she still had open. `default` is + // guaranteed present by `ensure_default_user`. + let Some(user) = self.users.get(username) else { + return Some(format!("user {} no longer exists", username)); + }; // Hot path: unrestricted user (default `on nopass ~* &* +@all`) // short-circuits before any per-command allocation. Profile showed // ~1% of CPU here for the lowercasing + HashSet probe; the @@ -416,7 +445,16 @@ impl AclTable { args: &[Frame], is_write: bool, ) -> Option { - let user = self.users.get(username)?; + // c10k hardening B2: an unknown user is DENIED, never allowed. + // This used to be `self.users.get(username)?` - `None` means + // "allowed", so a user that vanished from the table fell open to + // full `~* +@all`. `ACL DELUSER alice` / `ACL LOAD` do not close + // alice's live sessions, so revoking her account silently + // PROMOTED every connection she still had open. `default` is + // guaranteed present by `ensure_default_user`. + let Some(user) = self.users.get(username) else { + return Some(format!("user {} no longer exists", username)); + }; // Hot path: unrestricted user skips extract_command_keys + the // O(patterns*keys) glob match loop. Profile showed ~1.2% of CPU // here, most of it in glob_match and Vec allocation for the @@ -456,7 +494,16 @@ impl AclTable { /// Check channel access for pub/sub. pub fn check_channel_permission(&self, username: &str, channel: &[u8]) -> Option { - let user = self.users.get(username)?; + // c10k hardening B2: an unknown user is DENIED, never allowed. + // This used to be `self.users.get(username)?` - `None` means + // "allowed", so a user that vanished from the table fell open to + // full `~* +@all`. `ACL DELUSER alice` / `ACL LOAD` do not close + // alice's live sessions, so revoking her account silently + // PROMOTED every connection she still had open. `default` is + // guaranteed present by `ensure_default_user`. + let Some(user) = self.users.get(username) else { + return Some(format!("user {} no longer exists", username)); + }; if user.unrestricted { return None; } @@ -571,6 +618,66 @@ mod tests { ServerConfig::parse_from(args) } + /// c10k hardening B2. All three permission checks used to start with + /// `self.users.get(username)?` — and `None` is their "allowed" answer, so + /// a username that is not in the table was granted everything. Nothing + /// closes a live session when its account goes away (`ACL DELUSER`, `ACL + /// LOAD` dropping a user), so revoking an account silently PROMOTED every + /// connection it had already authenticated to full `~* +@all`. + #[test] + fn unknown_user_is_denied_by_every_check() { + let table = AclTable::new(); // no users at all + let args: Vec = vec![Frame::BulkString(Bytes::from_static(b"k"))]; + + assert!( + table + .check_command_permission("ghost", b"SET", &args) + .is_some(), + "unknown user must be denied commands, not allowed" + ); + assert!( + table + .check_key_permission("ghost", b"SET", &args, true) + .is_some(), + "unknown user must be denied keys, not allowed" + ); + assert!( + table.check_channel_permission("ghost", b"news").is_some(), + "unknown user must be denied channels, not allowed" + ); + } + + /// The flip side of the fail-closed change: `default` must always be + /// present, or a server whose ACL file never mentions it would refuse + /// every connection. + #[test] + fn ensure_default_user_creates_only_when_missing() { + let mut table = AclTable::new(); + table.ensure_default_user(None); + let user = table.get_user("default").expect("default was created"); + assert!(user.unrestricted(), "nopass default must be unrestricted"); + + // With requirepass, the created default carries the password. + let mut table = AclTable::new(); + table.ensure_default_user(Some("hunter2")); + assert_eq!( + table.authenticate("default", "hunter2"), + Some("default".to_string()) + ); + assert_eq!(table.authenticate("default", "wrong"), None); + + // Present already: left exactly as the file defined it. + let mut table = AclTable::new(); + table.apply_setuser("default", &["on", "~app:*", "-@all", "+get"]); + table.ensure_default_user(Some("hunter2")); + assert!( + table + .check_command_permission("default", b"SET", &[]) + .is_some(), + "an existing restricted default must not be overwritten by a fresh unrestricted one" + ); + } + #[test] fn default_user_is_unrestricted() { // Every construction path that yields a "fully open" default diff --git a/src/command/acl.rs b/src/command/acl.rs index 6dd2a1c33..af6eaef1c 100644 --- a/src/command/acl.rs +++ b/src/command/acl.rs @@ -170,6 +170,7 @@ pub fn handle_acl( )); } let mut count = 0i64; + let mut revoked: Vec = Vec::new(); let Ok(mut table) = acl_table.write() else { return Frame::Error(Bytes::from_static(b"ERR internal ACL error")); }; @@ -182,9 +183,24 @@ pub fn handle_acl( } if table.del_user(name) { count += 1; + revoked.push(name.to_string()); } } } + // c10k B2/B3: Redis disconnects clients authenticated as a deleted + // user, and so do we now that the registry tracks the post-AUTH + // identity. Deleting the account alone never closed the sessions + // it had already granted; the permission checks fail those + // sessions closed since B2, but leaving them connected would keep + // holding maxclients slots on a credential that no longer exists. + // Released before the kill so a self-kill cannot deadlock on it. + drop(table); + for name in revoked { + crate::client_registry::kill_clients( + &crate::client_registry::KillFilter::User(name), + None, + ); + } Frame::Integer(count) } @@ -318,6 +334,10 @@ pub fn handle_acl( new_table.set_user(user.username.clone(), user); } } + // c10k B2: unknown users are denied now, so a file + // without a `default` line must not brick the server. + let requirepass = runtime_config.read().requirepass.clone(); + new_table.ensure_default_user(requirepass.as_deref()); let Ok(mut table) = acl_table.write() else { return Frame::Error(Bytes::from_static(b"ERR internal ACL error")); }; diff --git a/src/server/conn/core.rs b/src/server/conn/core.rs index 8e9cd06b6..7c77c803e 100644 --- a/src/server/conn/core.rs +++ b/src/server/conn/core.rs @@ -360,6 +360,30 @@ impl ConnectionState { } } + /// Adopt a new authenticated identity — the ONLY way to change + /// `current_user`. + /// + /// c10k hardening B3. Beyond the ACL cache refresh, this publishes the + /// new username to the client registry. `register` captures `user` once, + /// at accept time, when it is always `default`; every AUTH/HELLO success + /// used to update only the connection-local copy. The registry's copy is + /// what `CLIENT LIST` reports and what `CLIENT KILL USER ` matches + /// on, so both lied about every authenticated session: `CLIENT LIST` + /// showed `user=default` for everyone and `CLIENT KILL USER alice` + /// returned 0 — the primary incident-response lever for a compromised + /// credential, inert. It is also what makes revocation reachable, since + /// dropping a user from the table cannot by itself close their live + /// sessions. + /// + /// The registry write takes a stripe lock, which is fine here: AUTH and + /// HELLO are per-session events, never the steady-state batch loop. + pub fn adopt_user(&mut self, username: String, acl_table: &StdRwLock) { + self.current_user = username; + self.refresh_acl_cache(acl_table); + let user = self.current_user.clone(); + crate::client_registry::update(self.client_id, move |e| e.user = user); + } + /// Resolve and cache the unrestricted flag from the AclTable. /// Called once on connection init and after AUTH / HELLO. /// diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index e3d607645..8b1ebcee8 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -58,8 +58,7 @@ pub(super) fn check_auth_gate( let (response, opt_user) = conn_cmd::auth_acl(cmd_args, &ctx.acl_table); if let Some(uname) = opt_user { conn.authenticated = true; - conn.current_user = uname; - conn.refresh_acl_cache(&ctx.acl_table); + conn.adopt_user(uname, &ctx.acl_table); if let Ok(addr) = peer_addr.parse::() { crate::auth_ratelimit::record_success(addr.ip()); } @@ -99,8 +98,7 @@ pub(super) fn check_auth_gate( conn.client_name = Some(name); } if let Some(ref uname) = opt_user { - conn.current_user = uname.clone(); - conn.refresh_acl_cache(&ctx.acl_table); + conn.adopt_user(uname.clone(), &ctx.acl_table); } // HELLO AUTH rate limiting if matches!(&response, Frame::Error(_)) { @@ -337,8 +335,7 @@ pub(super) fn try_handle_auth( } let (response, opt_user) = conn_cmd::auth_acl(cmd_args, &ctx.acl_table); if let Some(uname) = opt_user { - conn.current_user = uname; - conn.refresh_acl_cache(&ctx.acl_table); + conn.adopt_user(uname, &ctx.acl_table); if let Ok(addr) = peer_addr.parse::() { crate::auth_ratelimit::record_success(addr.ip()); } @@ -382,8 +379,7 @@ pub(super) fn try_handle_hello( conn.client_name = Some(name); } if let Some(ref uname) = opt_user { - conn.current_user = uname.clone(); - conn.refresh_acl_cache(&ctx.acl_table); + conn.adopt_user(uname.clone(), &ctx.acl_table); } if matches!(&response, Frame::Error(_)) { if let Ok(addr) = peer_addr.parse::() { diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index e8a2ba776..19b741b3f 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/src/server/conn/handler_monoio/mod.rs @@ -1075,24 +1075,14 @@ pub(crate) async fn handle_connection_sharded_monoio< // connection-level command names. Cut per-command dispatch cost // from ~14 non-matching function calls to ~1 on SET/GET workloads. let cmd_len = cmd.len(); - if cmd_len == 7 && dispatch::try_handle_cluster(cmd, cmd_args, ctx, &mut responses) { - continue; - } - if cmd_len == 7 - && dispatch::try_handle_evalsha(cmd, cmd_args, &conn, ctx, &mut responses) - { - continue; - } - if cmd_len == 4 && dispatch::try_handle_eval(cmd, cmd_args, &conn, ctx, &mut responses) - { - continue; - } - if cmd_len == 6 && dispatch::try_handle_script(cmd, cmd_args, ctx, &mut responses) { - continue; - } - if dispatch::try_handle_cluster_routing(cmd, cmd_args, &mut conn, ctx, &mut responses) { - continue; - } + // === ACL-EXEMPT COMMANDS === + // + // AUTH and HELLO carry Redis's `NO_AUTH` flag and are permitted + // regardless of the current user's permissions - a restricted user + // must always be able to re-authenticate as somebody else, and + // gating HELLO would break the RESP3 handshake. `check_auth_gate` + // above only handles them while UNauthenticated, so these are the + // post-authentication path and must stay ahead of the gate. if cmd_len == 4 && dispatch::try_handle_auth( cmd, @@ -1121,6 +1111,48 @@ pub(crate) async fn handle_connection_sharded_monoio< { continue; } + + // === ACL GATE - every privileged intercept MUST sit below this === + // + // c10k hardening B1. This gate used to sit ~200 lines further + // down, below the script/ACL/CONFIG/REPLICAOF/BGSAVE/SHUTDOWN + // intercepts, while the comment attached to it claimed the + // opposite invariant. Because those intercepts `continue` on a + // match, they returned before any permission check ever ran: a + // user holding nothing but `~app:* +get` was correctly refused a + // plain SET yet could still run `ACL SETUSER evil on nopass ~* + // +@all`, `CONFIG SET`, arbitrary Lua, `REPLICAOF` and `BGSAVE` + // - verified end-to-end, see tests/acl_privileged_intercepts.rs. + // Authenticated-restricted-user escalation to full admin. + // + // The auth gate runs earlier still, so unauthenticated clients + // get NOAUTH rather than NOPERM; only AUTH/HELLO above are exempt + // (Redis marks both NO_AUTH). + // + // If you add a privileged intercept, add it BELOW this line. + if dispatch::try_enforce_acl(cmd, cmd_args, &mut conn, ctx, &peer_addr, &mut responses) + { + continue; + } + + if cmd_len == 7 && dispatch::try_handle_cluster(cmd, cmd_args, ctx, &mut responses) { + continue; + } + if cmd_len == 7 + && dispatch::try_handle_evalsha(cmd, cmd_args, &conn, ctx, &mut responses) + { + continue; + } + if cmd_len == 4 && dispatch::try_handle_eval(cmd, cmd_args, &conn, ctx, &mut responses) + { + continue; + } + if cmd_len == 6 && dispatch::try_handle_script(cmd, cmd_args, ctx, &mut responses) { + continue; + } + if dispatch::try_handle_cluster_routing(cmd, cmd_args, &mut conn, ctx, &mut responses) { + continue; + } if cmd_len == 3 && dispatch::try_handle_acl( cmd, @@ -1277,13 +1309,9 @@ pub(crate) async fn handle_connection_sharded_monoio< break; } } - // ACL gate MUST run before any privileged intercept (SWAPDB included) - // — otherwise unauthenticated clients can mutate cross-DB state. - // handler_sharded already enforces this ordering; this matches it. - if dispatch::try_enforce_acl(cmd, cmd_args, &mut conn, ctx, &peer_addr, &mut responses) - { - continue; - } + // (B1) The ACL gate that used to sit here now runs far above, ahead + // of every privileged intercept. Its old comment claimed exactly + // the invariant the code above it violated. // --- SWAPDB: handler-layer intercept (needs async + multi-db access) --- if dispatch::try_handle_swapdb(cmd, cmd_args, &conn, ctx, &mut responses).await { continue; diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index b7a55fcb9..1fc8621be 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -469,8 +469,7 @@ pub(crate) async fn handle_connection_sharded_inner< let (response, opt_user) = conn_cmd::auth_acl(cmd_args, &ctx.acl_table); if let Some(uname) = opt_user { conn.authenticated = true; - conn.current_user = uname; - conn.refresh_acl_cache(&ctx.acl_table); + conn.adopt_user(uname, &ctx.acl_table); if let Ok(addr) = peer_addr.parse::() { crate::auth_ratelimit::record_success(addr.ip()); } @@ -507,8 +506,7 @@ pub(crate) async fn handle_connection_sharded_inner< conn.client_name = Some(name); } if let Some(ref uname) = opt_user { - conn.current_user = uname.clone(); - conn.refresh_acl_cache(&ctx.acl_table); + conn.adopt_user(uname.clone(), &ctx.acl_table); } // HELLO AUTH rate limiting (same as AUTH gate) if matches!(&response, Frame::Error(_)) { @@ -561,6 +559,99 @@ pub(crate) async fn handle_connection_sharded_inner< continue; } + // === ACL-EXEMPT COMMANDS === + // + // AUTH and HELLO carry Redis's `NO_AUTH` flag and are + // allowed whatever the current user's permissions are: a + // restricted user must always be able to re-authenticate, + // and gating HELLO would break the RESP3 handshake. The + // pre-auth gate above handles them only while + // UNauthenticated; this is the post-authentication path. + // --- AUTH (already conn.authenticated) --- + if cmd.eq_ignore_ascii_case(b"AUTH") { + let (response, opt_user) = conn_cmd::auth_acl(cmd_args, &ctx.acl_table); + if let Some(uname) = opt_user { + conn.adopt_user(uname, &ctx.acl_table); + if let Ok(addr) = peer_addr.parse::() { + crate::auth_ratelimit::record_success(addr.ip()); + } + } else if let Ok(addr) = peer_addr.parse::() { + auth_delay_ms += crate::auth_ratelimit::record_failure(addr.ip()); + } + responses.push(response); + continue; + } + + // --- HELLO --- + if cmd.eq_ignore_ascii_case(b"HELLO") { + let (response, new_proto, new_name, opt_user) = conn_cmd::hello_acl( + cmd_args, conn.protocol_version, client_id, &ctx.acl_table, &mut conn.authenticated, + ); + if !matches!(&response, Frame::Error(_)) { conn.protocol_version = new_proto; } + if let Some(name) = new_name { conn.client_name = Some(name); } + if let Some(ref uname) = opt_user { + conn.adopt_user(uname.clone(), &ctx.acl_table); + } + if matches!(&response, Frame::Error(_)) { + if let Ok(addr) = peer_addr.parse::() { + auth_delay_ms += crate::auth_ratelimit::record_failure(addr.ip()); + } + } else if opt_user.is_some() { + if let Ok(addr) = peer_addr.parse::() { + crate::auth_ratelimit::record_success(addr.ip()); + } + } + responses.push(response); + continue; + } + + // === ACL GATE - every privileged intercept MUST sit below === + // + // c10k hardening B1. This gate used to sit ~180 lines + // further down, below the scripting and ACL intercepts, + // while its own comment claimed it ran "before any + // command-specific handlers". That held for CONFIG and + // REPLICAOF, but the Lua and ACL intercepts `continue`d + // above it and so never reached a permission check: a user + // holding only `~app:* +get` was correctly refused a plain + // SET yet could run `ACL SETUSER evil on nopass ~* +@all` + // and arbitrary Lua. See tests/acl_privileged_intercepts.rs. + // + // If you add a privileged intercept, add it BELOW this line. + // + // Fast path: skip RwLock + HashMap for unrestricted users + // with a fresh cache. Stale caches (after ACL SETUSER / + // DELUSER / LOAD) fall through to the full check. + if !conn.acl_skip_allowed() { + #[allow(clippy::unwrap_used)] // std RwLock: poison = prior panic = unrecoverable + let acl_guard = ctx.acl_table.read().unwrap(); + if let Some(deny_reason) = acl_guard.check_command_permission(&conn.current_user, cmd, cmd_args) { + drop(acl_guard); + conn.acl_log.push(crate::acl::AclLogEntry { + reason: "command".to_string(), + object: String::from_utf8_lossy(cmd).to_ascii_lowercase(), + username: conn.current_user.clone(), + client_addr: peer_addr.clone(), + timestamp_ms: std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_millis() as u64, + }); + responses.push(Frame::Error(Bytes::from(format!("NOPERM {}", deny_reason)))); + continue; + } + let is_write_for_acl = metadata::is_write(cmd); + if let Some(deny_reason) = acl_guard.check_key_permission(&conn.current_user, cmd, cmd_args, is_write_for_acl) { + drop(acl_guard); + conn.acl_log.push(crate::acl::AclLogEntry { + reason: "command".to_string(), + object: String::from_utf8_lossy(cmd).to_ascii_lowercase(), + username: conn.current_user.clone(), + client_addr: peer_addr.clone(), + timestamp_ms: std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_millis() as u64, + }); + responses.push(Frame::Error(Bytes::from(format!("NOPERM {}", deny_reason)))); + continue; + } + } + // --- CLUSTER subcommands --- if cmd.eq_ignore_ascii_case(b"CLUSTER") { if let Some(ref cs) = ctx.cluster_state { @@ -670,45 +761,9 @@ pub(crate) async fn handle_connection_sharded_inner< } } - // --- AUTH (already conn.authenticated) --- - if cmd.eq_ignore_ascii_case(b"AUTH") { - let (response, opt_user) = conn_cmd::auth_acl(cmd_args, &ctx.acl_table); - if let Some(uname) = opt_user { - conn.current_user = uname; - conn.refresh_acl_cache(&ctx.acl_table); - if let Ok(addr) = peer_addr.parse::() { - crate::auth_ratelimit::record_success(addr.ip()); - } - } else if let Ok(addr) = peer_addr.parse::() { - auth_delay_ms += crate::auth_ratelimit::record_failure(addr.ip()); - } - responses.push(response); - continue; - } - - // --- HELLO --- - if cmd.eq_ignore_ascii_case(b"HELLO") { - let (response, new_proto, new_name, opt_user) = conn_cmd::hello_acl( - cmd_args, conn.protocol_version, client_id, &ctx.acl_table, &mut conn.authenticated, - ); - if !matches!(&response, Frame::Error(_)) { conn.protocol_version = new_proto; } - if let Some(name) = new_name { conn.client_name = Some(name); } - if let Some(ref uname) = opt_user { - conn.current_user = uname.clone(); - conn.refresh_acl_cache(&ctx.acl_table); - } - if matches!(&response, Frame::Error(_)) { - if let Ok(addr) = peer_addr.parse::() { - auth_delay_ms += crate::auth_ratelimit::record_failure(addr.ip()); - } - } else if opt_user.is_some() { - if let Ok(addr) = peer_addr.parse::() { - crate::auth_ratelimit::record_success(addr.ip()); - } - } - responses.push(response); - continue; - } + // (B1) The ACL gate that used to sit here now runs far above, + // ahead of every privileged intercept. Its old comment claimed + // exactly the invariant the code above it violated. // --- ACL --- if cmd.eq_ignore_ascii_case(b"ACL") { @@ -762,41 +817,6 @@ pub(crate) async fn handle_connection_sharded_inner< } } - // === ACL permission check === - // Must run before any command-specific handlers (CONFIG, REPLICAOF, etc.) - // so that low-privilege users cannot reach admin commands. - // Fast path: skip RwLock + HashMap for unrestricted users - // with a fresh cache. Stale caches (after ACL SETUSER / - // DELUSER / LOAD) fall through to the full check. - if !conn.acl_skip_allowed() { - #[allow(clippy::unwrap_used)] // std RwLock: poison = prior panic = unrecoverable - let acl_guard = ctx.acl_table.read().unwrap(); - if let Some(deny_reason) = acl_guard.check_command_permission(&conn.current_user, cmd, cmd_args) { - drop(acl_guard); - conn.acl_log.push(crate::acl::AclLogEntry { - reason: "command".to_string(), - object: String::from_utf8_lossy(cmd).to_ascii_lowercase(), - username: conn.current_user.clone(), - client_addr: peer_addr.clone(), - timestamp_ms: std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_millis() as u64, - }); - responses.push(Frame::Error(Bytes::from(format!("NOPERM {}", deny_reason)))); - continue; - } - let is_write_for_acl = metadata::is_write(cmd); - if let Some(deny_reason) = acl_guard.check_key_permission(&conn.current_user, cmd, cmd_args, is_write_for_acl) { - drop(acl_guard); - conn.acl_log.push(crate::acl::AclLogEntry { - reason: "command".to_string(), - object: String::from_utf8_lossy(cmd).to_ascii_lowercase(), - username: conn.current_user.clone(), - client_addr: peer_addr.clone(), - timestamp_ms: std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_millis() as u64, - }); - responses.push(Frame::Error(Bytes::from(format!("NOPERM {}", deny_reason)))); - continue; - } - } // --- Functions API: FUNCTION/FCALL/FCALL_RO --- // Placed AFTER ACL check. Respects MULTI queue — if conn.in_multi, diff --git a/src/server/conn/handler_single.rs b/src/server/conn/handler_single.rs index 1c22b3edf..301635c69 100644 --- a/src/server/conn/handler_single.rs +++ b/src/server/conn/handler_single.rs @@ -330,8 +330,7 @@ pub async fn handle_connection( conn.client_name = Some(name); } if let Some(uname) = opt_user { - conn.current_user = uname; - conn.refresh_acl_cache(&acl_table); + conn.adopt_user(uname, &acl_table); } let _ = framed.send(response).await; } @@ -450,8 +449,7 @@ pub async fn handle_connection( let (response, opt_user) = conn_cmd::auth_acl(cmd_args, &acl_table); if let Some(uname) = opt_user { conn.authenticated = true; - conn.current_user = uname; - conn.refresh_acl_cache(&acl_table); + conn.adopt_user(uname, &acl_table); } else { // Log failed auth attempt conn.acl_log.push(crate::acl::AclLogEntry { @@ -485,8 +483,7 @@ pub async fn handle_connection( conn.client_name = Some(name); } if let Some(uname) = opt_user { - conn.current_user = uname; - conn.refresh_acl_cache(&acl_table); + conn.adopt_user(uname, &acl_table); } responses.push(response); continue; @@ -511,8 +508,7 @@ pub async fn handle_connection( if cmd.eq_ignore_ascii_case(b"AUTH") { let (response, opt_user) = conn_cmd::auth_acl(cmd_args, &acl_table); if let Some(uname) = opt_user { - conn.current_user = uname; - conn.refresh_acl_cache(&acl_table); + conn.adopt_user(uname, &acl_table); } responses.push(response); continue; @@ -534,8 +530,7 @@ pub async fn handle_connection( conn.client_name = Some(name); } if let Some(uname) = opt_user { - conn.current_user = uname; - conn.refresh_acl_cache(&acl_table); + conn.adopt_user(uname, &acl_table); } responses.push(response); continue; diff --git a/tests/acl_privileged_intercepts.rs b/tests/acl_privileged_intercepts.rs new file mode 100644 index 000000000..cdcd911bb --- /dev/null +++ b/tests/acl_privileged_intercepts.rs @@ -0,0 +1,259 @@ +//! c10k hardening B1 — privileged intercepts must sit BELOW the ACL gate. +//! +//! Both connection handlers used to run their command-specific intercepts +//! (Lua `EVAL`/`EVALSHA`/`SCRIPT`, `ACL`, `CLUSTER`) BEFORE the ACL +//! permission check, even though the comment attached to that check claimed +//! it "must run before any command-specific handlers ... so that +//! low-privilege users cannot reach admin commands". Each intercept +//! `continue`s on a match, so the gate below it never ran. +//! +//! The result was full privilege escalation from any authenticated account: +//! a user holding nothing but `~app:* +get` was correctly refused a plain +//! `SET` yet could run `ACL SETUSER evil on nopass ~* +@all` (persisted), +//! `CONFIG SET maxmemory ...` (persisted) and arbitrary Lua via `EVAL`. +//! +//! Runs at `--shards 1` (monoio TopLevel handler) and `--shards 4` (sharded +//! handler) because the two handlers carry independent copies of the +//! ordering. Skips gracefully when the moon binary is missing. + +mod common; + +use std::io::{Read, Write}; +use std::net::TcpStream; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +fn moon_binary() -> Option { + if let Ok(p) = std::env::var("MOON_BIN") { + return Some(std::path::PathBuf::from(p)); + } + let cargo_bin = std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")); + if cargo_bin.exists() { + return Some(cargo_bin); + } + None +} + +struct Moon { + child: Child, + port: u16, + tmp_dir: std::path::PathBuf, +} + +impl Drop for Moon { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + let _ = std::fs::remove_dir_all(&self.tmp_dir); + } +} + +fn spawn_moon(shards: &str) -> Option { + let bin = moon_binary()?; + let tmp_dir = std::env::temp_dir().join(format!( + "moon-acl-intercepts-{}-{shards}", + std::process::id() + )); + let _ = std::fs::create_dir_all(&tmp_dir); + let (child, port) = common::spawn_listening(|port| { + Command::new(&bin) + .args([ + "--port", + &port.to_string(), + "--shards", + shards, + "--admin-port", + "0", + "--appendonly", + "no", + "--disk-free-min-pct", + "0", + // Keep the per-shard page cache small: a default auto-sized + // maxmemory makes startup slow enough to trip readiness on a + // contended host. + "--maxmemory", + "268435456", + "--dir", + tmp_dir.to_str().expect("utf8 tmp dir"), + ]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon") + }); + let moon = Moon { + child, + port, + tmp_dir, + }; + let deadline = Instant::now() + Duration::from_secs(30); + while Instant::now() < deadline { + if let Ok(mut c) = TcpStream::connect(("127.0.0.1", moon.port)) { + let _ = c.set_read_timeout(Some(Duration::from_millis(500))); + if c.write_all(b"*1\r\n$4\r\nPING\r\n").is_ok() { + let mut buf = [0u8; 64]; + if let Ok(n) = c.read(&mut buf) + && n > 0 + && buf.starts_with(b"+PONG") + { + return Some(moon); + } + } + } + std::thread::sleep(Duration::from_millis(100)); + } + eprintln!("skipping: moon did not become ready on port {port}"); + None +} + +/// Minimal RESP client over a blocking TcpStream (rolling receive buffer). +struct Resp { + stream: TcpStream, + buf: Vec, +} + +impl Resp { + fn connect(port: u16) -> Self { + let stream = TcpStream::connect(("127.0.0.1", port)).expect("connect"); + stream + .set_read_timeout(Some(Duration::from_millis(100))) + .expect("set read timeout"); + Self { + stream, + buf: Vec::new(), + } + } + + fn send(&mut self, args: &[&str]) { + let mut out = format!("*{}\r\n", args.len()).into_bytes(); + for a in args { + out.extend_from_slice(format!("${}\r\n{a}\r\n", a.len()).as_bytes()); + } + self.stream.write_all(&out).expect("write"); + } + + fn pump(&mut self, total: Duration) { + let deadline = Instant::now() + total; + let mut chunk = [0u8; 4096]; + while Instant::now() < deadline { + match self.stream.read(&mut chunk) { + Ok(0) => break, + Ok(n) => self.buf.extend_from_slice(&chunk[..n]), + Err(_) => {} + } + } + } + + /// Send one command and return everything that came back for it. + fn cmd(&mut self, args: &[&str]) -> String { + self.buf.clear(); + self.send(args); + self.pump(Duration::from_millis(250)); + String::from_utf8_lossy(&self.buf).into_owned() + } +} + +/// Every privileged intercept must answer NOPERM for a `~app:* +get` user, +/// and must leave no trace behind when it does. +fn run_privileged_intercepts(shards: &str) { + let Some(moon) = spawn_moon(shards) else { + return; // binary missing — skip + }; + let tag = format!("[shards={shards}]"); + + // Admin (default user) provisions the restricted account. + let mut admin = Resp::connect(moon.port); + let r = admin.cmd(&[ + "ACL", + "SETUSER", + "lowpriv", + "on", + ">pw", + "resetkeys", + "~app:*", + "-@all", + "+get", + ]); + assert!(r.contains("+OK"), "{tag} ACL SETUSER lowpriv failed: {r:?}"); + + let mut c = Resp::connect(moon.port); + let r = c.cmd(&["AUTH", "lowpriv", "pw"]); + assert!(r.contains("+OK"), "{tag} AUTH lowpriv failed: {r:?}"); + + // Positive control: the one thing this user IS allowed to do. If this + // fails the gate is over-broad and the NOPERM assertions below prove + // nothing. + let r = c.cmd(&["GET", "app:hello"]); + assert!( + !r.contains("NOPERM") && !r.contains("NOAUTH"), + "{tag} GET app:* must stay allowed: {r:?}" + ); + + // Baseline: a command with no intercept in front of it was ALWAYS gated. + let r = c.cmd(&["SET", "app:hello", "1"]); + assert!(r.contains("NOPERM"), "{tag} SET must be NOPERM: {r:?}"); + + // --- The escalation vectors --- + for args in [ + // Privilege escalation: mint a full-admin account. + &["ACL", "SETUSER", "evil", "on", "nopass", "~*", "+@all"][..], + // Server reconfiguration. + &["CONFIG", "SET", "maxmemory", "999999999"][..], + // Arbitrary Lua. + &["EVAL", "return 'pwned'", "0"][..], + &["EVALSHA", "ffffffffffffffffffffffffffffffffffffffff", "0"][..], + &["SCRIPT", "LOAD", "return 1"][..], + // Replication takeover. + &["REPLICAOF", "no", "one"][..], + &["SLAVEOF", "no", "one"][..], + // Persistence / cluster admin. + &["BGSAVE"][..], + &["CLUSTER", "INFO"][..], + ] { + let r = c.cmd(args); + assert!( + r.contains("NOPERM"), + "{tag} {args:?} must be NOPERM for a `~app:* +get` user, got: {r:?}" + ); + } + + // --- And the denials must have had no side effect --- + let r = admin.cmd(&["ACL", "GETUSER", "evil"]); + assert!( + r.starts_with("$-1") || r.starts_with("*-1") || r.starts_with("_") || r.starts_with("*0"), + "{tag} the denied ACL SETUSER must not have created `evil`: {r:?}" + ); + let r = admin.cmd(&["CONFIG", "GET", "maxmemory"]); + assert!( + !r.contains("999999999"), + "{tag} the denied CONFIG SET must not have persisted: {r:?}" + ); + + // --- ACL-exempt commands must still work for the restricted user --- + // Redis marks AUTH and HELLO NO_AUTH; gating them would strand a + // restricted client with no way to re-authenticate, and would break the + // RESP3 handshake. + let r = c.cmd(&["HELLO", "2"]); + assert!( + !r.contains("NOPERM"), + "{tag} HELLO must stay ACL-exempt: {r:?}" + ); + let r = c.cmd(&["AUTH", "lowpriv", "pw"]); + assert!( + r.contains("+OK"), + "{tag} AUTH must stay ACL-exempt (re-auth): {r:?}" + ); + // ...and re-authenticating as an unrestricted user restores the powers. + let r = c.cmd(&["ACL", "WHOAMI"]); + assert!(r.contains("NOPERM"), "{tag} still restricted: {r:?}"); +} + +#[test] +fn privileged_intercepts_are_acl_gated_single_shard() { + run_privileged_intercepts("1"); +} + +#[test] +fn privileged_intercepts_are_acl_gated_multi_shard() { + run_privileged_intercepts("4"); +} diff --git a/tests/acl_user_revocation.rs b/tests/acl_user_revocation.rs new file mode 100644 index 000000000..394234369 --- /dev/null +++ b/tests/acl_user_revocation.rs @@ -0,0 +1,254 @@ +//! c10k hardening B3 (+ the B2 revocation story) — the client registry must +//! track the POST-AUTH identity, and revoking an account must close the +//! sessions it already granted. +//! +//! `client_registry::register` captures `user` once, at accept time, when it +//! is always `default`. Every AUTH/HELLO success updated only the +//! connection-local copy, so the registry — the thing `CLIENT LIST` reports +//! and `CLIENT KILL USER ` matches on — never learned who anybody was: +//! `CLIENT LIST` showed `user=default` for every session and `CLIENT KILL +//! USER alice` returned 0. That is the primary incident-response lever for a +//! compromised credential, and it was inert. +//! +//! It is also what makes revocation reachable at all: dropping a user from +//! the ACL table cannot by itself close the sessions that user already holds +//! (Redis disconnects them; we did not). +//! +//! Runs at `--shards 1` and `--shards 4` — the two handlers carry +//! independent AUTH/HELLO paths. Skips gracefully when the binary is missing. + +mod common; + +use std::io::{Read, Write}; +use std::net::TcpStream; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +fn moon_binary() -> Option { + if let Ok(p) = std::env::var("MOON_BIN") { + return Some(std::path::PathBuf::from(p)); + } + let cargo_bin = std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")); + if cargo_bin.exists() { + return Some(cargo_bin); + } + None +} + +struct Moon { + child: Child, + port: u16, + tmp_dir: std::path::PathBuf, +} + +impl Drop for Moon { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + let _ = std::fs::remove_dir_all(&self.tmp_dir); + } +} + +fn spawn_moon(tag: &str, shards: &str) -> Option { + let bin = moon_binary()?; + let tmp_dir = std::env::temp_dir().join(format!( + "moon-acl-revoke-{}-{tag}-{shards}", + std::process::id() + )); + let _ = std::fs::create_dir_all(&tmp_dir); + let (child, port) = common::spawn_listening(|port| { + Command::new(&bin) + .args([ + "--port", + &port.to_string(), + "--shards", + shards, + "--admin-port", + "0", + "--appendonly", + "no", + "--disk-free-min-pct", + "0", + "--maxmemory", + "268435456", + "--dir", + tmp_dir.to_str().expect("utf8 tmp dir"), + ]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn moon") + }); + let moon = Moon { + child, + port, + tmp_dir, + }; + let deadline = Instant::now() + Duration::from_secs(30); + while Instant::now() < deadline { + if let Ok(mut c) = TcpStream::connect(("127.0.0.1", moon.port)) { + let _ = c.set_read_timeout(Some(Duration::from_millis(500))); + if c.write_all(b"*1\r\n$4\r\nPING\r\n").is_ok() { + let mut buf = [0u8; 64]; + if let Ok(n) = c.read(&mut buf) + && n > 0 + && buf.starts_with(b"+PONG") + { + return Some(moon); + } + } + } + std::thread::sleep(Duration::from_millis(100)); + } + eprintln!("skipping: moon did not become ready on port {port}"); + None +} + +struct Resp { + stream: TcpStream, + buf: Vec, +} + +impl Resp { + fn connect(port: u16) -> Self { + let stream = TcpStream::connect(("127.0.0.1", port)).expect("connect"); + stream + .set_read_timeout(Some(Duration::from_millis(100))) + .expect("set read timeout"); + Self { + stream, + buf: Vec::new(), + } + } + + fn cmd(&mut self, args: &[&str]) -> String { + self.buf.clear(); + let mut out = format!("*{}\r\n", args.len()).into_bytes(); + for a in args { + out.extend_from_slice(format!("${}\r\n{a}\r\n", a.len()).as_bytes()); + } + self.stream.write_all(&out).expect("write"); + let deadline = Instant::now() + Duration::from_millis(300); + let mut chunk = [0u8; 8192]; + while Instant::now() < deadline { + match self.stream.read(&mut chunk) { + Ok(0) => break, + Ok(n) => self.buf.extend_from_slice(&chunk[..n]), + Err(_) => {} + } + } + String::from_utf8_lossy(&self.buf).into_owned() + } + + /// True once the peer has closed (or reset) this socket. Polls for up to + /// 2s so a cooperative teardown has time to land. + fn is_closed(&mut self) -> bool { + let deadline = Instant::now() + Duration::from_secs(2); + let mut chunk = [0u8; 4096]; + while Instant::now() < deadline { + match self.stream.read(&mut chunk) { + Ok(0) => return true, + Ok(_) => {} + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {} + Err(e) if e.kind() == std::io::ErrorKind::TimedOut => {} + Err(_) => return true, // ECONNRESET and friends + } + } + false + } +} + +/// `CLIENT LIST` must report who a session actually authenticated as, and +/// `CLIENT KILL USER` must find it. +fn run_kill_by_user(shards: &str) { + let Some(moon) = spawn_moon("kill", shards) else { + return; // binary missing — skip + }; + let tag = format!("[shards={shards}]"); + + let mut admin = Resp::connect(moon.port); + let r = admin.cmd(&["ACL", "SETUSER", "alice", "on", ">pw", "~*", "+@all"]); + assert!(r.contains("+OK"), "{tag} ACL SETUSER alice failed: {r:?}"); + + let mut alice = Resp::connect(moon.port); + let r = alice.cmd(&["AUTH", "alice", "pw"]); + assert!(r.contains("+OK"), "{tag} AUTH alice failed: {r:?}"); + // A command after AUTH, so the session is unambiguously established. + let r = alice.cmd(&["PING"]); + assert!(r.contains("+PONG"), "{tag} alice PING failed: {r:?}"); + + let r = admin.cmd(&["CLIENT", "LIST"]); + assert!( + r.contains("user=alice"), + "{tag} CLIENT LIST must report the post-AUTH user, got: {r:?}" + ); + + let r = admin.cmd(&["CLIENT", "KILL", "USER", "alice"]); + assert!( + r.starts_with(":1") || r.starts_with(":2"), + "{tag} CLIENT KILL USER alice must match alice's session, got: {r:?}" + ); + assert!( + alice.is_closed(), + "{tag} alice's session must actually be torn down" + ); + + // The admin (still `default`) must NOT have been caught by that kill. + let r = admin.cmd(&["PING"]); + assert!( + r.contains("+PONG"), + "{tag} CLIENT KILL USER alice must not touch other users: {r:?}" + ); +} + +/// Deleting a user closes the sessions that user already holds — otherwise a +/// revoked credential keeps its connection (and its maxclients slot) alive. +fn run_deluser_disconnects(shards: &str) { + let Some(moon) = spawn_moon("del", shards) else { + return; // binary missing — skip + }; + let tag = format!("[shards={shards}]"); + + let mut admin = Resp::connect(moon.port); + let r = admin.cmd(&["ACL", "SETUSER", "bob", "on", ">pw", "~*", "+@all"]); + assert!(r.contains("+OK"), "{tag} ACL SETUSER bob failed: {r:?}"); + + let mut bob = Resp::connect(moon.port); + let r = bob.cmd(&["AUTH", "bob", "pw"]); + assert!(r.contains("+OK"), "{tag} AUTH bob failed: {r:?}"); + let r = bob.cmd(&["SET", "k", "v"]); + assert!(r.contains("+OK"), "{tag} bob SET failed: {r:?}"); + + let r = admin.cmd(&["ACL", "DELUSER", "bob"]); + assert!(r.starts_with(":1"), "{tag} ACL DELUSER bob failed: {r:?}"); + + assert!( + bob.is_closed(), + "{tag} deleting bob must disconnect bob's live session" + ); + let r = admin.cmd(&["PING"]); + assert!( + r.contains("+PONG"), + "{tag} ACL DELUSER must not disturb other sessions: {r:?}" + ); +} + +#[test] +fn client_kill_by_user_finds_authenticated_sessions_single_shard() { + run_kill_by_user("1"); +} + +#[test] +fn client_kill_by_user_finds_authenticated_sessions_multi_shard() { + run_kill_by_user("4"); +} + +#[test] +fn acl_deluser_disconnects_live_sessions_single_shard() { + run_deluser_disconnects("1"); +} + +#[test] +fn acl_deluser_disconnects_live_sessions_multi_shard() { + run_deluser_disconnects("4"); +} From 3237f12df750f8bef4834f3150df502b2177bbc6 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 30 Jul 2026 23:48:39 +0700 Subject: [PATCH 2/5] fix(security): refuse the io_uring bridge when authentication is configured (c10k B4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `MOON_URING=1` (tokio, Linux) arms an experimental io_uring bridge that binds a SECOND SO_REUSEPORT listener on the server's own port and serves commands from its own accept path — one with no auth gate, no ACL check and no client registry. On a server that HAS configured authentication the kernel then load-balances new connections between a listener that enforces auth and one that does not, so roughly half of them skip authentication entirely. The bridge's documented limitation named maxclients and CLIENT LIST/KILL. It did not name this. The bridge now refuses to arm when requirepass or aclfile is set, logging at error level, and the shard stays on the stable pure-tokio path (which does enforce auth). Refusing is the fail-closed answer and matches how every other failure in this block degrades; killing the process over an experimental opt-in would be worse. The rule is a free function so it is unit-testable off Linux and under both runtimes. Test: src/shard/event_loop.rs — no auth allows the bridge, requirepass and aclfile each refuse it. Refs: tmp/C10K-HARDENING-REVIEW.md B4. author: Tin Dang --- src/shard/event_loop.rs | 68 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 68 insertions(+) diff --git a/src/shard/event_loop.rs b/src/shard/event_loop.rs index 7ab0c1c57..db428ec60 100644 --- a/src/shard/event_loop.rs +++ b/src/shard/event_loop.rs @@ -36,6 +36,22 @@ use super::shared_databases::ShardDatabases; use super::uring_handler; use super::{conn_accept, persistence_tick, spsc_handler, timers}; +/// c10k hardening B4: may the experimental io_uring bridge be armed? +/// +/// The bridge binds a SECOND `SO_REUSEPORT` listener on the server's own port +/// and dispatches commands from its own accept path — one with no auth gate, +/// no ACL check and no client registry. On a server that has configured +/// authentication the kernel would then load-balance new connections between +/// a listener that enforces auth and one that does not. Answer `false` there +/// and stay on the pure-tokio path. +/// +/// Free function (not `cfg`-gated like its only call site) so the rule is +/// unit-testable on every platform and under both runtimes. +#[allow(dead_code)] // Called only under cfg(linux + runtime-tokio); tests use it everywhere. +pub(crate) fn uring_bridge_allowed(config: &crate::config::ServerConfig) -> bool { + config.requirepass.is_none() && config.aclfile.is_none() +} + impl super::Shard { /// Run the shard event loop on its dedicated current_thread runtime. /// @@ -104,6 +120,28 @@ impl super::Shard { self.id ); None + } else if !uring_bridge_allowed(&server_config) { + // c10k hardening B4. The bridge's own accept path + // (`uring_handler`) dispatches commands directly — it has no + // auth gate, no ACL check and no client registry. On a server + // that HAS configured authentication it therefore binds a + // second SO_REUSEPORT socket on the very same port that serves + // every command unauthenticated: the kernel load-balances new + // connections between the two listeners, so roughly half of + // them skip auth entirely. The documented limitation named + // maxclients and CLIENT LIST/KILL, not this. + // + // Refusing to arm is the fail-closed answer and leaves the + // shard on the stable pure-tokio path (which does enforce + // auth), rather than killing a server over an experimental + // opt-in. + tracing::error!( + "Shard {} REFUSING io_uring bridge: MOON_URING=1 with requirepass/aclfile \ + configured would serve unauthenticated commands on the same port. \ + Falling back to tokio I/O; unset MOON_URING or remove auth to use it.", + self.id + ); + None } else { match UringDriver::new(UringConfig { sqpoll_idle_ms: server_config.uring_sqpoll_ms, @@ -2620,3 +2658,33 @@ impl super::Shard { self.pubsub_registry = std::mem::take(&mut *pubsub_arc.write()); } } + +#[cfg(test)] +mod tests { + use super::uring_bridge_allowed; + use crate::config::ServerConfig; + use clap::Parser; + + /// c10k B4. The bridge is only allowed when the server has no + /// authentication to bypass in the first place. + #[test] + fn uring_bridge_is_refused_when_auth_is_configured() { + let no_auth = ServerConfig::parse_from(["moon"]); + assert!( + uring_bridge_allowed(&no_auth), + "no auth configured: the bridge has nothing to bypass" + ); + + let with_pass = ServerConfig::parse_from(["moon", "--requirepass", "hunter2"]); + assert!( + !uring_bridge_allowed(&with_pass), + "requirepass set: the bridge would serve unauthenticated commands on the same port" + ); + + let with_aclfile = ServerConfig::parse_from(["moon", "--aclfile", "/tmp/users.acl"]); + assert!( + !uring_bridge_allowed(&with_aclfile), + "aclfile set: same bypass, via ACL users instead of requirepass" + ); + } +} From 41edd9104d3bafec5af7140dd1fa0992e7dffec2 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Thu, 30 Jul 2026 23:48:53 +0700 Subject: [PATCH 3/5] fix(tls): process TLS records before vetoing a task park (c10k B5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `Stream::task_park_safe` (vendored monoio-rustls) decides whether an idle TLS connection may be task-parked on raw-fd readability. It sampled `session.wants_write()` BEFORE calling `process_new_packets()` — but processing a buffered record can itself queue outbound bytes and still return `Ok`, so the veto read a state from before those bytes existed, passed, and parked the connection owing the peer a reply that nothing would flush. The reachable case in rustls 0.23 is the TLS 1.2 renegotiation rebuff: a post-handshake ClientHello makes rustls queue a `NoRenegotiation` warning alert into `sendable_tls` and return Ok. moon builds rustls with `tls12` enabled, so a 1.2 client reaches it. Correcting the review's own claim: the TLS 1.3 KeyUpdate(update_requested) case it named is NOT reachable. rustls parks its KeyUpdate reply in `queued_key_update_message` and only moves it into `sendable_tls` on the next outbound record, so `wants_write()` reads false under either ordering. That is a finding, not an oversight, and the new test pins it: if a future rustls starts queueing eagerly the test fails loudly and the KeyUpdate deadlock becomes live — at which point the reordered veto is already what handles it. Test: tests/tls_park_keyupdate.rs — drives a real in-memory rustls 1.3 session pair (no live server, no monoio), asserts the deferral and that the deferred reply does ride out with the next record. Refs: tmp/C10K-HARDENING-REVIEW.md B5. author: Tin Dang --- tests/tls_park_keyupdate.rs | 229 +++++++++++++++++++++++++++++ vendor/monoio-rustls/src/stream.rs | 29 +++- 2 files changed, 252 insertions(+), 6 deletions(-) create mode 100644 tests/tls_park_keyupdate.rs diff --git a/tests/tls_park_keyupdate.rs b/tests/tls_park_keyupdate.rs new file mode 100644 index 000000000..949378ef0 --- /dev/null +++ b/tests/tls_park_keyupdate.rs @@ -0,0 +1,229 @@ +//! c10k hardening B5 — pins rustls's KeyUpdate deferral, the assumption the +//! reordered TLS park veto rests on. +//! +//! `Stream::task_park_safe` (vendored monoio-rustls) decides whether a TLS +//! connection may be task-parked on raw-fd readability. Parking is only +//! correct when nothing is buffered anywhere in the TLS stack — anything we +//! still owe the peer would wait forever behind a `readable()` that never +//! fires. The veto used to sample `session.wants_write()` BEFORE calling +//! `process_new_packets()`, which reads the state from before processing +//! could queue anything; it now processes first. The reachable case that +//! makes the order matter is the TLS 1.2 renegotiation rebuff, where rustls +//! queues a `NoRenegotiation` warning alert and returns `Ok` — that one can +//! only be driven by forging encrypted records, so it is not exercised here. +//! +//! What this test DOES establish is the negative result: the TLS 1.3 +//! KeyUpdate(update_requested) case named in the c10k review is **not** +//! reachable. rustls stashes its KeyUpdate reply in +//! `queued_key_update_message` and only moves it into `sendable_tls` on the +//! next outbound record, so `wants_write()` reads false under either +//! ordering. If a future rustls starts queueing eagerly, this test fails and +//! the KeyUpdate case becomes live — at which point the reordered veto is +//! already what handles it correctly. +//! +//! Runs on a real in-memory rustls session pair; no live server, no monoio. + +use std::io::Write; +use std::process::{Command, Stdio}; +use std::sync::Arc; + +/// Accept any server certificate — the cert is a throwaway generated per run; +/// transport privacy is irrelevant to the ordering question. +#[derive(Debug)] +struct AcceptAnyCert(rustls::crypto::CryptoProvider); + +impl rustls::client::danger::ServerCertVerifier for AcceptAnyCert { + fn verify_server_cert( + &self, + _end_entity: &rustls::pki_types::CertificateDer<'_>, + _intermediates: &[rustls::pki_types::CertificateDer<'_>], + _server_name: &rustls::pki_types::ServerName<'_>, + _ocsp_response: &[u8], + _now: rustls::pki_types::UnixTime, + ) -> Result { + Ok(rustls::client::danger::ServerCertVerified::assertion()) + } + + fn verify_tls12_signature( + &self, + _message: &[u8], + _cert: &rustls::pki_types::CertificateDer<'_>, + _dss: &rustls::DigitallySignedStruct, + ) -> Result { + Ok(rustls::client::danger::HandshakeSignatureValid::assertion()) + } + + fn verify_tls13_signature( + &self, + _message: &[u8], + _cert: &rustls::pki_types::CertificateDer<'_>, + _dss: &rustls::DigitallySignedStruct, + ) -> Result { + Ok(rustls::client::danger::HandshakeSignatureValid::assertion()) + } + + fn supported_verify_schemes(&self) -> Vec { + self.0.signature_verification_algorithms.supported_schemes() + } +} + +/// Generate a throwaway self-signed cert. Returns false (test skips) when the +/// `openssl` CLI is unavailable. +fn generate_cert(dir: &std::path::Path) -> bool { + let status = Command::new("openssl") + .args([ + "req", + "-x509", + "-newkey", + "rsa:2048", + "-nodes", + "-days", + "1", + "-subj", + "/CN=localhost", + ]) + .arg("-keyout") + .arg(dir.join("key.pem")) + .arg("-out") + .arg(dir.join("cert.pem")) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); + matches!(status, Ok(s) if s.success()) +} + +/// Pump one side's outbound bytes into the other until both are quiet. +fn drive_handshake( + client: &mut rustls::ClientConnection, + server: &mut rustls::ServerConnection, +) -> Result<(), rustls::Error> { + for _ in 0..16 { + let mut c2s = Vec::new(); + while client.wants_write() { + client.write_tls(&mut c2s).expect("write_tls to Vec"); + } + if !c2s.is_empty() { + let mut cursor = &c2s[..]; + while !cursor.is_empty() { + server.read_tls(&mut cursor).expect("read_tls from slice"); + server.process_new_packets()?; + } + } + let mut s2c = Vec::new(); + while server.wants_write() { + server.write_tls(&mut s2c).expect("write_tls to Vec"); + } + if !s2c.is_empty() { + let mut cursor = &s2c[..]; + while !cursor.is_empty() { + client.read_tls(&mut cursor).expect("read_tls from slice"); + client.process_new_packets()?; + } + } + if !client.is_handshaking() && !server.is_handshaking() { + return Ok(()); + } + } + panic!("handshake did not converge"); +} + +/// rustls defers its KeyUpdate reply, so a parked connection owes the peer +/// nothing observable through `wants_write()`. Canary: if this ever flips, +/// the KeyUpdate deadlock the c10k review described becomes real. +#[test] +fn keyupdate_reply_is_deferred_until_the_next_outbound_record() { + let tmp = std::env::temp_dir().join(format!("moon-tls-keyupdate-{}", std::process::id())); + let _ = std::fs::create_dir_all(&tmp); + if !generate_cert(&tmp) { + eprintln!("skipping: openssl unavailable"); + let _ = std::fs::remove_dir_all(&tmp); + return; + } + + let server_cfg = moon::tls::build_tls_config( + tmp.join("cert.pem").to_str().expect("utf8 path"), + tmp.join("key.pem").to_str().expect("utf8 path"), + None, + None, + ) + .expect("build server TLS config"); + + let provider = rustls::crypto::aws_lc_rs::default_provider(); + let client_cfg = rustls::ClientConfig::builder() + .dangerous() + .with_custom_certificate_verifier(Arc::new(AcceptAnyCert(provider))) + .with_no_client_auth(); + + let mut client = rustls::ClientConnection::new( + Arc::new(client_cfg), + "localhost".try_into().expect("server name"), + ) + .expect("client conn"); + let mut server = rustls::ServerConnection::new(server_cfg).expect("server conn"); + + drive_handshake(&mut client, &mut server).expect("handshake"); + assert_eq!( + client.protocol_version(), + Some(rustls::ProtocolVersion::TLSv1_3), + "KeyUpdate is a TLS 1.3 message; the test needs a 1.3 session" + ); + + // Quiescent: nothing owed either way. This is the state the park veto is + // designed to say "yes" to. + assert!(!server.wants_write(), "server owes nothing after handshake"); + + // The client asks us to rotate keys AND to answer with our own KeyUpdate. + client.refresh_traffic_keys().expect("refresh_traffic_keys"); + let mut c2s = Vec::new(); + while client.wants_write() { + client.write_tls(&mut c2s).expect("write_tls"); + } + assert!(!c2s.is_empty(), "KeyUpdate must produce bytes on the wire"); + + // The record is now in our TLS buffer but NOT yet processed — exactly the + // moment the idle sweep evaluates `task_park_safe`. + let mut cursor = &c2s[..]; + while !cursor.is_empty() { + server.read_tls(&mut cursor).expect("read_tls"); + } + + // OLD ORDERING: `wants_write()` sampled here, before processing. + let old_wants_write = server.wants_write(); + let state = server.process_new_packets().expect("process_new_packets"); + let plaintext = state.plaintext_bytes_to_read(); + let closed = state.peer_has_closed(); + // NEW ORDERING: sampled after. + let new_wants_write = server.wants_write(); + + assert_eq!(plaintext, 0, "KeyUpdate carries no plaintext"); + assert!(!closed, "peer has not closed"); + assert!(!old_wants_write, "nothing was queued before processing"); + + // THE CANARY. rustls parks its reply in `queued_key_update_message` and + // flushes it lazily, so processing the record queues nothing visible. Both + // orderings therefore agree that parking is safe — the KeyUpdate deadlock + // the review described is not reachable with this rustls. + assert!( + !new_wants_write, + "rustls now queues the KeyUpdate reply eagerly — the KeyUpdate park \ + deadlock just became REACHABLE. The reordered veto in \ + vendor/monoio-rustls/src/stream.rs (c10k B5) already handles it; \ + update that comment and this test rather than reverting." + ); + + // The reply is real, just deferred: it appears once anything else goes out. + server + .writer() + .write_all(b"x") + .expect("write plaintext to trigger the deferred flush"); + let mut s2c = Vec::new(); + while server.wants_write() { + server.write_tls(&mut s2c).expect("write_tls"); + } + assert!( + !s2c.is_empty(), + "the deferred KeyUpdate must ride out with the next record" + ); + + let _ = std::fs::remove_dir_all(&tmp); +} diff --git a/vendor/monoio-rustls/src/stream.rs b/vendor/monoio-rustls/src/stream.rs index 7115411e8..f9354d3bf 100644 --- a/vendor/monoio-rustls/src/stream.rs +++ b/vendor/monoio-rustls/src/stream.rs @@ -101,16 +101,33 @@ where /// waiting on our write). A partial record in the deframer is fine: /// completing it requires more socket bytes, which fire readability. pub fn task_park_safe(&mut self) -> bool { - if !self.r_buffer.is_drained() || !self.w_buffer.is_drained() || self.session.wants_write() - { + if !self.r_buffer.is_drained() || !self.w_buffer.is_drained() { return false; } - match self.session.process_new_packets() { - Ok(state) => state.plaintext_bytes_to_read() == 0 && !state.peer_has_closed(), + // moon patch (c10k B5): `process_new_packets` must run BEFORE the + // `wants_write` check, not after it. Processing a buffered record can + // itself queue outbound bytes AND still return `Ok`, and sampling + // `wants_write()` first read the state from before those bytes + // existed — so the veto passed and the connection parked owing the + // peer a reply it will never send until someone writes again. + // + // The reachable case in rustls 0.23 is the TLS 1.2 renegotiation + // rebuff (`common_state.rs`: a post-handshake ClientHello → + // `send_warning_alert(NoRenegotiation)` → `Ok`); moon builds rustls + // with `tls12` enabled, so a 1.2 client can reach it. Note the c10k + // review named TLS 1.3 KeyUpdate(update_requested) instead — that one + // is NOT reachable: rustls parks its KeyUpdate reply in + // `queued_key_update_message` and only moves it into `sendable_tls` + // on the next outbound record, so `wants_write()` is false either way + // (see tests/tls_park_keyupdate.rs, which pins that behaviour). + let state = match self.session.process_new_packets() { + Ok(state) => state, // Corrupt/unexpected TLS state: don't park; the next read // surfaces the error and tears down cleanly. - Err(_) => false, - } + Err(_) => return false, + }; + let (plaintext, closed) = (state.plaintext_bytes_to_read(), state.peer_has_closed()); + plaintext == 0 && !closed && !self.session.wants_write() } } From 1a934773ef8bb91b247479dcfed19a04e6c04dc8 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Wed, 5 Aug 2026 10:07:39 +0700 Subject: [PATCH 4/5] test(acl): poke the connection before asserting a killed session is gone (Windows CI) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `CLIENT KILL` does two things: it sets a cooperative `kill_flag` AND it `shutdown(2)`s the fd so a connection parked in `read().await` is torn down at once. Only the first is portable — `force_close_fd` is `#[cfg(unix)]`, so on Windows the session dies when the handler next looks at it, i.e. on the client's next command. `is_closed()` asserted on a bare `read`, which is the unix-only half. It passed on macOS and Linux and failed all four `acl_user_revocation` tests on Windows. It now sends a PING first, exercising the same teardown a real client sees on every platform; on unix the socket is already shut down so the poke fails immediately and costs nothing. Deadline raised 2s -> 5s and a killed connection is re-poked while it drains any reply that was already queued. Worth recording why this reached CI at all: `Check (Windows)`, `Check (macOS)` and `Check (console feature)` are SKIPPED on pull requests — they run only on push-to-main or `workflow_dispatch`. Every PR in this stack shows "green" without ever having been built on Windows. This failure surfaced only because the stack's PRs target non-main bases (so `pull_request: branches: [main]` never fired at all) and the CI had to be dispatched by hand. Gates: acl_user_revocation 4/4 on both runtimes. author: Tin Dang --- tests/acl_user_revocation.rs | 27 +++++++++++++++++++++------ 1 file changed, 21 insertions(+), 6 deletions(-) diff --git a/tests/acl_user_revocation.rs b/tests/acl_user_revocation.rs index 394234369..260b113ab 100644 --- a/tests/acl_user_revocation.rs +++ b/tests/acl_user_revocation.rs @@ -140,17 +140,32 @@ impl Resp { String::from_utf8_lossy(&self.buf).into_owned() } - /// True once the peer has closed (or reset) this socket. Polls for up to - /// 2s so a cooperative teardown has time to land. + /// True once this session is dead — either the peer already tore the + /// socket down, or it refuses to serve another command. + /// + /// The poke matters. `CLIENT KILL` sets a cooperative `kill_flag` AND + /// `shutdown(2)`s the fd, but `force_close_fd` is `#[cfg(unix)]` — on + /// Windows only the flag is set, so the connection is torn down when the + /// handler next looks at it, i.e. on the client's next command. Asserting + /// on a bare `read` therefore passed on unix and failed on Windows, which + /// is exactly what CI caught. Sending a PING first exercises the same + /// teardown Redis clients see on every platform; on unix the socket is + /// already shut down, so the poke fails immediately and costs nothing. fn is_closed(&mut self) -> bool { - let deadline = Instant::now() + Duration::from_secs(2); + let deadline = Instant::now() + Duration::from_secs(5); let mut chunk = [0u8; 4096]; while Instant::now() < deadline { + if self.stream.write_all(b"*1\r\n$4\r\nPING\r\n").is_err() { + return true; // EPIPE / ECONNRESET — already gone + } match self.stream.read(&mut chunk) { Ok(0) => return true, - Ok(_) => {} - Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {} - Err(e) if e.kind() == std::io::ErrorKind::TimedOut => {} + // A killed connection may still drain a reply that was + // already queued; keep poking until it stops answering. + Ok(_) => std::thread::sleep(Duration::from_millis(100)), + Err(e) + if e.kind() == std::io::ErrorKind::WouldBlock + || e.kind() == std::io::ErrorKind::TimedOut => {} Err(_) => return true, // ECONNRESET and friends } } From 5374628b03adde72b220740757ba0a259617d5f4 Mon Sep 17 00:00:00 2001 From: Tin Dang Date: Wed, 5 Aug 2026 11:03:30 +0700 Subject: [PATCH 5/5] fix(acl): let a self-DELUSER answer before it disconnects the caller (#430 review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ACL DELUSER` disconnects the sessions a deleted user still holds (B3). It called `kill_clients(filter, None)` — and `None` means "no self-kill exemption", so when the caller deleted its OWN account, `kill_clients` `shutdown(2)`d the caller's socket while its own `:1` was still in flight. The user saw a connection error instead of a reply. `kill_clients` already carries `self_id` for exactly this: for the matching entry it sets `kill_flag` but skips the fd shutdown, leaving the teardown to the handler loop's next `is_killed()` check. That is reply-then-disconnect, which is what a client expects and what `CLIENT KILL` of oneself already did. `handle_acl` had no way to know who was calling, so the caller's `client_id` is now threaded in (one new parameter; the three production call sites pass their connection's id, the unit tests pass 0). Test: `self_deluser_replies_before_disconnect_{single,multi}_shard`. RED with `None` restored: `a self-DELUSER must answer before disconnecting, got: ""` at shards=1 AND shards=4 — an empty reply, which is the defect exactly. Noted while tracing this, NOT fixed here: `handler_single` never registers connections with the client registry at all, so `CLIENT LIST`, `CLIENT KILL` and this disconnect are all inert on that path. It is the embedded / single-shard tokio handler and the gap predates this branch. Reported by qodo-code-review on #430. Gates: fmt; clippy -D warnings on default and runtime-tokio,jemalloc; lib acl 84; acl_user_revocation 6/6 on both runtimes. author: Tin Dang --- src/command/acl.rs | 59 +++++++++++++--------- src/server/conn/handler_monoio/dispatch.rs | 1 + src/server/conn/handler_sharded/mod.rs | 2 +- src/server/conn/handler_single.rs | 1 + tests/acl_user_revocation.rs | 55 ++++++++++++++++++++ 5 files changed, 92 insertions(+), 26 deletions(-) diff --git a/src/command/acl.rs b/src/command/acl.rs index af6eaef1c..eed3339d8 100644 --- a/src/command/acl.rs +++ b/src/command/acl.rs @@ -27,6 +27,7 @@ pub fn handle_acl( current_user: &str, _client_addr: &str, runtime_config: &Arc>, + caller_client_id: u64, ) -> Frame { let sub = match sub_and_args.first().and_then(|f| extract_str(f)) { Some(s) => s.to_ascii_uppercase(), @@ -196,9 +197,17 @@ pub fn handle_acl( // Released before the kill so a self-kill cannot deadlock on it. drop(table); for name in revoked { + // `Some(caller_client_id)`: if the caller deleted its OWN + // account, self-kill must stay cooperative. `kill_clients` + // sets the flag but skips `shutdown(2)` for `self_id`, so this + // reply still flushes and the connection closes on the loop's + // next `is_killed()` check — reply-then-disconnect, which is + // the semantics a client expects. Passing `None` here shut the + // caller's own socket mid-command, so a self-DELUSER surfaced + // as a connection error instead of its `:1`. crate::client_registry::kill_clients( &crate::client_registry::KillFilter::User(name), - None, + Some(caller_client_id), ); } Frame::Integer(count) @@ -422,7 +431,7 @@ mod tests { let mut log = AclLog::new(128); let rc = make_runtime_config(); let args = vec![Frame::BulkString(Bytes::from_static(b"WHOAMI"))]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); assert_eq!(result, Frame::BulkString(Bytes::from_static(b"default"))); } @@ -441,7 +450,7 @@ mod tests { Frame::BulkString(Bytes::from_static(b"~*")), Frame::BulkString(Bytes::from_static(b"+@all")), ]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); assert_eq!(result, Frame::SimpleString(Bytes::from_static(b"OK"))); // GETUSER alice @@ -449,7 +458,7 @@ mod tests { Frame::BulkString(Bytes::from_static(b"GETUSER")), Frame::BulkString(Bytes::from_static(b"alice")), ]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); match result { Frame::Array(ref fields) => { // Should have username, flags, passwords, keys, channels, commands @@ -473,7 +482,7 @@ mod tests { Frame::BulkString(Bytes::from_static(b"GETUSER")), Frame::BulkString(Bytes::from_static(b"nonexistent")), ]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); assert_eq!(result, Frame::Null); } @@ -489,14 +498,14 @@ mod tests { Frame::BulkString(Bytes::from_static(b"alice")), Frame::BulkString(Bytes::from_static(b"on")), ]; - handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); // DELUSER alice let args = vec![ Frame::BulkString(Bytes::from_static(b"DELUSER")), Frame::BulkString(Bytes::from_static(b"alice")), ]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); assert_eq!(result, Frame::Integer(1)); } @@ -509,7 +518,7 @@ mod tests { Frame::BulkString(Bytes::from_static(b"DELUSER")), Frame::BulkString(Bytes::from_static(b"default")), ]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); assert!(matches!(result, Frame::Error(_))); } @@ -519,7 +528,7 @@ mod tests { let mut log = AclLog::new(128); let rc = make_runtime_config(); let args = vec![Frame::BulkString(Bytes::from_static(b"LIST"))]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); match result { Frame::Array(ref items) => { assert!(!items.is_empty()); @@ -538,7 +547,7 @@ mod tests { let mut log = AclLog::new(128); let rc = make_runtime_config(); let args = vec![Frame::BulkString(Bytes::from_static(b"CAT"))]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); match result { Frame::Array(ref cats) => { assert!(!cats.is_empty()); @@ -556,7 +565,7 @@ mod tests { Frame::BulkString(Bytes::from_static(b"CAT")), Frame::BulkString(Bytes::from_static(b"string")), ]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); match result { Frame::Array(ref cmds) => { assert!(!cmds.is_empty()); @@ -574,7 +583,7 @@ mod tests { Frame::BulkString(Bytes::from_static(b"CAT")), Frame::BulkString(Bytes::from_static(b"nonexistent")), ]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); assert!(matches!(result, Frame::Error(_))); } @@ -594,7 +603,7 @@ mod tests { }); let args = vec![Frame::BulkString(Bytes::from_static(b"LOG"))]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); match result { Frame::Array(ref entries) => assert_eq!(entries.len(), 1), _ => panic!("Expected Array from LOG"), @@ -605,12 +614,12 @@ mod tests { Frame::BulkString(Bytes::from_static(b"LOG")), Frame::BulkString(Bytes::from_static(b"RESET")), ]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); assert_eq!(result, Frame::SimpleString(Bytes::from_static(b"OK"))); // Verify log is empty let args = vec![Frame::BulkString(Bytes::from_static(b"LOG"))]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); match result { Frame::Array(ref entries) => assert_eq!(entries.len(), 0), _ => panic!("Expected Array from LOG after RESET"), @@ -637,7 +646,7 @@ mod tests { Frame::BulkString(Bytes::from_static(b"LOG")), Frame::BulkString(Bytes::from_static(b"5")), ]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); match result { Frame::Array(ref entries) => assert_eq!(entries.len(), 5), _ => panic!("Expected Array from LOG 5"), @@ -650,7 +659,7 @@ mod tests { let mut log = AclLog::new(128); let rc = make_runtime_config(); // no aclfile configured let args = vec![Frame::BulkString(Bytes::from_static(b"SAVE"))]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); assert!(matches!(result, Frame::Error(_))); } @@ -660,7 +669,7 @@ mod tests { let mut log = AclLog::new(128); let rc = make_runtime_config(); // no aclfile configured let args = vec![Frame::BulkString(Bytes::from_static(b"LOAD"))]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); assert!(matches!(result, Frame::Error(_))); } @@ -688,11 +697,11 @@ mod tests { Frame::BulkString(Bytes::from_static(b"~*")), Frame::BulkString(Bytes::from_static(b"+@all")), ]; - handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); // SAVE let args = vec![Frame::BulkString(Bytes::from_static(b"SAVE"))]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); assert_eq!(result, Frame::SimpleString(Bytes::from_static(b"OK"))); // Verify file exists @@ -700,7 +709,7 @@ mod tests { // LOAD into a fresh table let args = vec![Frame::BulkString(Bytes::from_static(b"LOAD"))]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); assert_eq!(result, Frame::SimpleString(Bytes::from_static(b"OK"))); // Verify alice still exists after load @@ -714,7 +723,7 @@ mod tests { let mut log = AclLog::new(128); let rc = make_runtime_config(); let args = vec![Frame::BulkString(Bytes::from_static(b"INVALID"))]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); assert!(matches!(result, Frame::Error(_))); } @@ -724,7 +733,7 @@ mod tests { let mut log = AclLog::new(128); let rc = make_runtime_config(); let args = vec![Frame::BulkString(Bytes::from_static(b"GENPASS"))]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); match result { Frame::BulkString(b) => { assert_eq!(b.len(), 64); // 256 bits = 64 hex chars @@ -743,7 +752,7 @@ mod tests { Frame::BulkString(Bytes::from_static(b"GENPASS")), Frame::BulkString(Bytes::from_static(b"128")), ]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); match result { Frame::BulkString(b) => { assert_eq!(b.len(), 32); // 128 bits = 32 hex chars @@ -761,7 +770,7 @@ mod tests { Frame::BulkString(Bytes::from_static(b"GENPASS")), Frame::BulkString(Bytes::from_static(b"0")), ]; - let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc); + let result = handle_acl(&args, &table, &mut log, "default", "127.0.0.1:1234", &rc, 0); assert!(matches!(result, Frame::Error(_))); } } diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index 8b1ebcee8..336a2a7d2 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -414,6 +414,7 @@ pub(super) fn try_handle_acl( &conn.current_user, peer_addr, &ctx.runtime_config, + conn.client_id, ); responses.push(response); true diff --git a/src/server/conn/handler_sharded/mod.rs b/src/server/conn/handler_sharded/mod.rs index 1fc8621be..9dd0d0045 100644 --- a/src/server/conn/handler_sharded/mod.rs +++ b/src/server/conn/handler_sharded/mod.rs @@ -768,7 +768,7 @@ pub(crate) async fn handle_connection_sharded_inner< // --- ACL --- if cmd.eq_ignore_ascii_case(b"ACL") { let response = crate::command::acl::handle_acl( - cmd_args, &ctx.acl_table, &mut conn.acl_log, &conn.current_user, &peer_addr, &ctx.runtime_config, + cmd_args, &ctx.acl_table, &mut conn.acl_log, &conn.current_user, &peer_addr, &ctx.runtime_config, client_id, ); responses.push(response); continue; diff --git a/src/server/conn/handler_single.rs b/src/server/conn/handler_single.rs index 301635c69..973e96d02 100644 --- a/src/server/conn/handler_single.rs +++ b/src/server/conn/handler_single.rs @@ -544,6 +544,7 @@ pub async fn handle_connection( &conn.current_user, &peer_addr, &runtime_config, + client_id, ); responses.push(response); continue; diff --git a/tests/acl_user_revocation.rs b/tests/acl_user_revocation.rs index 260b113ab..4b842ed5e 100644 --- a/tests/acl_user_revocation.rs +++ b/tests/acl_user_revocation.rs @@ -248,6 +248,61 @@ fn run_deluser_disconnects(shards: &str) { ); } +/// Deleting your OWN account must still answer before it disconnects you. +/// +/// `kill_clients` takes a `self_id` precisely so a self-kill stays +/// cooperative — the flag is set, but the caller's fd is NOT `shutdown(2)` +/// out from under the in-flight reply. `ACL DELUSER` passed `None`, so a user +/// deleting itself had its socket torn down mid-command and saw a connection +/// error instead of the `:1`. Reply-then-disconnect is the contract. +fn run_self_deluser_replies_first(shards: &str) { + let Some(moon) = spawn_moon("self", shards) else { + return; // binary missing — skip + }; + let tag = format!("[shards={shards}]"); + + let mut admin = Resp::connect(moon.port); + // `carol` needs +acl to delete herself, and ~* so the ACL command's key + // check does not deny her first. + let r = admin.cmd(&["ACL", "SETUSER", "carol", "on", ">pw", "~*", "+@all"]); + assert!(r.contains("+OK"), "{tag} ACL SETUSER carol failed: {r:?}"); + + let mut carol = Resp::connect(moon.port); + let r = carol.cmd(&["AUTH", "carol", "pw"]); + assert!(r.contains("+OK"), "{tag} AUTH carol failed: {r:?}"); + let r = carol.cmd(&["PING"]); + assert!(r.contains("+PONG"), "{tag} carol PING failed: {r:?}"); + + // The reply must arrive. Before the fix this came back empty (socket shut + // down mid-command) rather than `:1`. + let r = carol.cmd(&["ACL", "DELUSER", "carol"]); + assert!( + r.starts_with(":1"), + "{tag} a self-DELUSER must answer before disconnecting, got: {r:?}" + ); + + // ...and only then is the session gone. + assert!( + carol.is_closed(), + "{tag} carol's session must still be torn down after the reply" + ); + let r = admin.cmd(&["PING"]); + assert!( + r.contains("+PONG"), + "{tag} other sessions must be undisturbed: {r:?}" + ); +} + +#[test] +fn self_deluser_replies_before_disconnect_single_shard() { + run_self_deluser_replies_first("1"); +} + +#[test] +fn self_deluser_replies_before_disconnect_multi_shard() { + run_self_deluser_replies_first("4"); +} + #[test] fn client_kill_by_user_finds_authenticated_sessions_single_shard() { run_kill_by_user("1");