Enforce an absolute pre-hello deadline against ping slowloris

Fixes #1068

Co-Authored-By: Claude Haiku 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

crates/collab-net/src/hub.rs

@@ -820,8 +820,8 @@ fn refuse(out: &Line, why: &str) {
 /// What the server tells a client when it is too busy to take more from it.
 const BUSY: &str = "the server is busy: try again in a moment";
 
-/// Read the next binary message (skipping pings) and its size, or why the connection should end.
-fn next_message(ws: &mut WebSocket<Duplex>, max: usize) -> Result<Option<(ClientMsg, usize)>, String> {
+/// Read the next binary message (skipping pings, until `deadline`) and its size, or why the connection should end.
+fn next_message(ws: &mut WebSocket<Duplex>, max: usize, deadline: Option<Instant>) -> Result<Option<(ClientMsg, usize)>, String> {
     loop {
         return match ws.read() {
             Ok(Message::Binary(b)) => {
@@ -830,6 +830,10 @@ fn next_message(ws: &mut WebSocket<Duplex>, max: usize) -> Result<Option<(Client
                 message::decode::<ClientMsg>(&b, max).map(|m| Some((m, b.len()))).map_err(|_| "that message couldn't be read".to_string())
             }
             Ok(Message::Text(_)) => Err("only binary messages are spoken here".into()),
+            // Pings keep the per-read timeout alive, but not the overall deadline (slowloris).
+            Ok(Message::Ping(_) | Message::Pong(_) | Message::Frame(_)) if deadline.is_some_and(|d| Instant::now() >= d) => {
+                Err("no hello in time".into())
+            }
             Ok(Message::Ping(_) | Message::Pong(_) | Message::Frame(_)) => continue,
             Ok(Message::Close(_)) => Ok(None),
             Err(tungstenite::Error::Capacity(_)) => Err(too_large(max)),
@@ -851,6 +855,7 @@ fn too_large(max: usize) -> String {
 /// One connection's reader (the writer runs on its own thread).
 fn serve(mut stream: TcpStream, conn: u64, hub: &Sender<HubIn>, limits: NetLimits, open: Open, shared: &Shared) {
     let _ = stream.set_nodelay(true);
+    let hello_deadline = Instant::now() + limits.hello_timeout;
     if stream.set_read_timeout(Some(limits.hello_timeout)).is_err() || stream.set_write_timeout(Some(limits.write_timeout)).is_err() {
         return;
     }
@@ -923,7 +928,7 @@ fn serve(mut stream: TcpStream, conn: u64, hub: &Sender<HubIn>, limits: NetLimit
         log::info!("event=refused ip={ip} reason=connections_per_address");
         end(Some("too many connections from your address: close some and try again"));
     } else {
-        match next_message(&mut ws, limits.max_hello) {
+        match next_message(&mut ws, limits.max_hello, Some(hello_deadline)) {
             Ok(Some((
                 first @ (ClientMsg::Hello(_)
                 | ClientMsg::Manage(_)
@@ -981,7 +986,7 @@ fn read_loop(
     let mut bucket = Bucket::new(limits.rate, limits.burst);
     let mut signals = Bucket::new(SIGNAL_RATE, SIGNAL_BURST);
     loop {
-        match next_message(ws, max) {
+        match next_message(ws, max, None) {
             Ok(Some((msg, bytes))) => {
                 let now = Instant::now();
                 if !bucket.take(now) {

crates/collab-net/src/loopback_tests.rs

@@ -670,3 +670,34 @@ fn a_welcome_over_the_size_limit_says_the_file_is_too_large() {
     assert!(!welcomed(&gp.events));
     host.stop();
 }
+
+#[test]
+fn pings_do_not_keep_a_connection_without_a_hello_alive() {
+    let (mut host, _hp, invite) = start_with(NetLimits { hello_timeout: Duration::from_secs(1), ..fast_limits() }, Limits::default());
+    let addr: SocketAddr = invite.server.host.parse().unwrap();
+    let mut r = Raw::connect(addr);
+    let t0 = Instant::now();
+    let mut closed = None;
+    while t0.elapsed() < Duration::from_secs(6) {
+        if r.0.send(Message::Ping(Bytes::new())).is_err() || r.0.flush().is_err() {
+            closed = Some(t0.elapsed());
+            break;
+        }
+        std::thread::sleep(Duration::from_millis(300));
+        // Drain what the host sent back (pongs, the close); an error means it hung up.
+        r.0.get_mut().set_read_timeout(Some(Duration::from_millis(10))).unwrap();
+        match r.0.read() {
+            Ok(Message::Close(_)) | Err(tungstenite::Error::ConnectionClosed | tungstenite::Error::AlreadyClosed) => {
+                closed = Some(t0.elapsed());
+                break;
+            }
+            Err(tungstenite::Error::Io(e)) if !matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => {
+                closed = Some(t0.elapsed());
+                break;
+            }
+            _ => {}
+        }
+    }
+    assert!(closed.is_some_and(|t| t < Duration::from_secs(3)), "pinged pre-hello connection survived: {closed:?}");
+    host.stop();
+}