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..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(), @@ -170,6 +171,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 +184,32 @@ 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 { + // `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), + Some(caller_client_id), + ); + } Frame::Integer(count) } @@ -318,6 +343,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")); }; @@ -402,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"))); } @@ -421,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 @@ -429,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 @@ -453,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); } @@ -469,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)); } @@ -489,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(_))); } @@ -499,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()); @@ -518,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()); @@ -536,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()); @@ -554,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(_))); } @@ -574,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"), @@ -585,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"), @@ -617,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"), @@ -630,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(_))); } @@ -640,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(_))); } @@ -668,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 @@ -680,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 @@ -694,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(_))); } @@ -704,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 @@ -723,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 @@ -741,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/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..336a2a7d2 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::() { @@ -418,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_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..9dd0d0045 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,50 +761,14 @@ 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") { 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; @@ -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..973e96d02 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; @@ -549,6 +544,7 @@ pub async fn handle_connection( &conn.current_user, &peer_addr, &runtime_config, + client_id, ); responses.push(response); continue; 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" + ); + } +} 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..4b842ed5e --- /dev/null +++ b/tests/acl_user_revocation.rs @@ -0,0 +1,324 @@ +//! 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 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(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, + // 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 + } + } + 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:?}" + ); +} + +/// 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"); +} + +#[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"); +} 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() } }