From bb0ed18133fcbe560c6671841bbe3912568d70ce Mon Sep 17 00:00:00 2001 From: skylar taylor-barrick Date: Sun, 18 Oct 2026 00:21:38 -0600 Subject: [PATCH] Control channel: refused connections linger outside the auth places A connection refused before auth used to keep its place among max_unauthenticated while it lingered (up to 1 s), so refusals could keep real clients out. Now `refuse` gives that place back and takes one of a separate, smaller set of lingering places (Limits::max_lingering, 8) before the reply is written. When none is free the connection is closed right after the FIN instead of lingering, so threads stay bounded as #1806 intended. Tests: a_lingering_connection_holds_no_place_to_authenticate and only_so_many_refused_connections_linger. --- crates/ui-egui/src/control.rs | 105 ++++++++++++++++++++++++++++++++-- 1 file changed, 99 insertions(+), 6 deletions(-) diff --git a/crates/ui-egui/src/control.rs b/crates/ui-egui/src/control.rs index 3f63f375c6..363b77a27a 100644 --- a/crates/ui-egui/src/control.rs +++ b/crates/ui-egui/src/control.rs @@ -796,6 +796,10 @@ struct Limits { /// How long a connection refused after a reply is kept half-open, discarding input, so the /// reply is not lost to a reset (see [`close_after_reply`]). linger: std::time::Duration, + /// How many refused connections may linger at once. A lingering connection has given back + /// its place among [`Self::max_unauthenticated`], so this is what bounds their threads; + /// beyond it a refused connection is closed at once. + max_lingering: usize, } #[cfg(not(target_arch = "wasm32"))] @@ -804,10 +808,12 @@ const LIMITS: Limits = Limits { max_unauthenticated: 8, write_timeout: std::time::Duration::from_secs(30), linger: std::time::Duration::from_secs(1), + max_lingering: 8, }; -/// One of the [`Limits::max_unauthenticated`] places, given back when dropped: when the -/// connection authenticates, and on every way its thread can end. +/// One of a limited number of places, given back when dropped: a place among +/// [`Limits::max_unauthenticated`] (given back when the connection authenticates or is refused, +/// and on every way its thread can end), or among [`Limits::max_lingering`]. #[cfg(not(target_arch = "wasm32"))] struct Slot(std::sync::Arc); @@ -863,16 +869,32 @@ impl std::io::Read for Reading { /// the reset can discard the reply before the client reads it (Linux does): a client still /// sending when it is refused would never learn why. So send FIN first, then read and discard /// whatever arrives until the client closes its side or `linger` passes, and only then close. +/// +/// Only [`Limits::max_lingering`] connections linger at once: `place` is one of those places, +/// taken by [`refuse`] before the reply was written, and held for the drain only. Without one +/// the connection is closed at once after the FIN, accepting the risk of a reset rather than +/// holding another thread. #[cfg(not(target_arch = "wasm32"))] -fn close_after_reply(reading: &mut Reading, linger: std::time::Duration) { +fn close_after_reply(reading: &mut Reading, place: Option, linger: std::time::Duration) { use std::io::Read; let _ = reading.stream.shutdown(std::net::Shutdown::Write); + let Some(_place) = place else { return }; reading.until = Some(std::time::Instant::now() + linger); let mut sink = [0u8; 1024]; // Each read is bounded by the time left, and none is made once `linger` has passed. while matches!(reading.read(&mut sink), Ok(n) if n > 0) {} } +/// A connection is being refused before `auth`: give back its place among those waiting to +/// authenticate, so it does not keep a real client out while it lingers, and take a place to +/// linger in if one is free. Done before the reply is written, so a client that has read the +/// reply sees the places as they will be. +#[cfg(not(target_arch = "wasm32"))] +fn refuse(slot: &mut Option, lingering: &std::sync::Arc, limits: Limits) -> Option { + drop(slot.take()); + Slot::take(lingering, limits.max_lingering) +} + /// Listen on `127.0.0.1` (random port) and forward authenticated requests to `client`. /// /// Protocol: newline-delimited JSON-RPC 2.0. The first request on a connection must be @@ -891,6 +913,7 @@ fn serve_with(client: ControlClient, limits: Limits) -> std::io::Result std::io::Result std::io::Result { let secs = limits.auth_deadline.as_secs_f64(); let e = rpc_error(Value::Null, -32004, &format!("not authenticated within {secs} s")); + let place = refuse(&mut slot, &lingering, limits); if writeln!(write, "{e}").and_then(|_| write.flush()).is_ok() { - close_after_reply(reader.get_mut(), limits.linger); + close_after_reply(reader.get_mut(), place, limits.linger); } break; } @@ -931,8 +956,9 @@ fn serve_with(client: ControlClient, limits: Limits) -> std::io::Result break, Ok(LineRead::TooLong) => { let e = rpc_error(Value::Null, -32600, &format!("request line longer than {max} bytes before auth")); + let place = refuse(&mut slot, &lingering, limits); if writeln!(write, "{e}").and_then(|_| write.flush()).is_ok() { - close_after_reply(reader.get_mut(), limits.linger); + close_after_reply(reader.get_mut(), place, limits.linger); } break; } @@ -972,11 +998,13 @@ fn serve_with(client: ControlClient, limits: Limits) -> std::io::Result = (0..TEST_LIMITS.max_unauthenticated).map(|_| connect(ep.port)).collect(); + for s in &others { + // A refused one has already been closed, so its write may fail; its reply says why. + let _ = writeln!(s.try_clone().unwrap(), "{auth}"); + let reply = next_line(&mut BufReader::new(s.try_clone().unwrap())).expect("a reply"); + assert_eq!(reply["result"]["ok"], true, "admitted while the refused one lingers: {reply}"); + } + assert!(started.elapsed() < TEST_LIMITS.linger, "checked while it lingered: {:?}", started.elapsed()); + drop(refused); + } + + /// Only [`super::Limits::max_lingering`] refused connections linger at once: one more still + /// gets its reply, and is then closed at once instead of lingering. + #[cfg(not(target_arch = "wasm32"))] + #[test] + fn only_so_many_refused_connections_linger() { + use std::io::{BufReader, Write}; + let limits = super::Limits { max_lingering: 1, ..TEST_LIMITS }; + let (_control, client) = super::attach(&egui::Context::default()); + let ep = super::serve_with(client, limits).unwrap(); + let bad = r#"{"jsonrpc": "2.0", "id": 1, "method": "page_count"}"#; + // The first refused connection takes the only lingering place and keeps it. + let first = connect(ep.port); + writeln!(first.try_clone().unwrap(), "{bad}").unwrap(); + let mut r1 = BufReader::new(first.try_clone().unwrap()); + assert_eq!(next_line(&mut r1).expect("refused with a reason")["error"]["code"], -32001); + // The second still gets its reason (it has nothing unread, so no reset)... + let second = connect(ep.port); + let mut w2 = second.try_clone().unwrap(); + writeln!(w2, "{bad}").unwrap(); + let mut r2 = BufReader::new(second.try_clone().unwrap()); + assert_eq!(next_line(&mut r2).expect("refused with a reason")["error"]["code"], -32001); + // ...and is closed outright, not lingering: writing to it soon fails, where a lingering + // connection would read and discard for the whole of `linger`. + let started = std::time::Instant::now(); + let closed = loop { + if w2.write_all(b"x").is_err() { + break true; + } + if started.elapsed() > limits.linger { + break false; + } + std::thread::sleep(std::time::Duration::from_millis(20)); + }; + assert!(closed && started.elapsed() < limits.linger / 2, "it lingered: {:?}", started.elapsed()); + drop(first); + } + /// #1806: only so many connections may wait to authenticate at once. One more is refused /// on the spot; once a waiting one authenticates or times out, its place is free again. #[cfg(not(target_arch = "wasm32"))]