diff --git a/src/client/io_loop.rs b/src/client/io_loop.rs index 880a0260f..b4d0a6c64 100644 --- a/src/client/io_loop.rs +++ b/src/client/io_loop.rs @@ -641,6 +641,19 @@ impl Remote { self.check_clipboard_file_context(); } Data::Message(msg) => { + // The Flutter clipboard broadcast is process-wide, so a clipboard can reach this + // round's queue before the round has logged in; it is dropped here, on the round + // itself. + #[cfg(feature = "flutter")] + if !self.is_connected + && matches!( + msg.union.as_ref(), + Some(message::Union::Clipboard(_)) + | Some(message::Union::MultiClipboards(_)) + ) + { + return true; + } match &msg.union { Some(message::Union::Misc(misc)) => match misc.union { Some(misc::Union::RefreshVideo(_)) => { @@ -2643,3 +2656,72 @@ impl Drop for VideoThread { *self.discard_queue.write().unwrap() = true; } } + +#[cfg(test)] +#[cfg(feature = "flutter")] +mod tests { + use super::*; + use crate::flutter::FlutterHandler; + + /// A round's `Remote` over a loopback pair, before any login: what it sends to `peer` + /// arrives at `far`. + async fn remote_and_peer() -> (Remote, Stream, Stream) { + let listener = hbb_common::tcp::new_listener("127.0.0.1:0", false) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let (peer, accepted) = tokio::join!( + hbb_common::socket_client::connect_tcp(addr.to_string(), 3000), + listener.accept() + ); + let (accepted, far_addr) = accepted.unwrap(); + let far = Stream::Tcp(hbb_common::tcp::FramedStream::from(accepted, far_addr)); + let (sender, receiver) = mpsc::unbounded_channel::(); + let remote = Remote::new(Session::::default(), receiver, sender); + (remote, peer.unwrap(), far) + } + + async fn arrives(far: &mut Stream) -> bool { + matches!(hbb_common::timeout(300, far.next()).await, Ok(Some(Ok(_)))) + } + + fn clipboard() -> Data { + let mut msg = Message::new(); + msg.set_clipboard(Clipboard { + content: b"copied while this login was pending".to_vec().into(), + ..Default::default() + }); + Data::Message(msg) + } + + fn auth_2fa() -> Data { + let mut msg = Message::new(); + msg.set_auth_2fa(Auth2FA { + code: "123456".to_owned(), + ..Default::default() + }); + Data::Message(msg) + } + + // A clipboard queued before this round's login stays here; what the login itself sends + // through the same queue does not. + #[tokio::test] + async fn a_clipboard_queued_before_this_rounds_login_is_dropped() { + let (mut remote, mut peer, mut far) = remote_and_peer().await; + assert!(!remote.is_connected); + assert!(remote.handle_msg_from_ui(clipboard(), &mut peer).await); + assert!( + !arrives(&mut far).await, + "a clipboard went out before the login" + ); + assert!(remote.handle_msg_from_ui(auth_2fa(), &mut peer).await); + assert!(arrives(&mut far).await, "the 2FA code was held back"); + + remote.is_connected = true; + assert!(remote.handle_msg_from_ui(clipboard(), &mut peer).await); + assert!( + arrives(&mut far).await, + "a clipboard after the login was held back" + ); + } +} diff --git a/src/server.rs b/src/server.rs index 47be1d359..7b2cbe792 100644 --- a/src/server.rs +++ b/src/server.rs @@ -224,6 +224,8 @@ pub async fn create_tcp_connection( let Some(unauthorized) = admit_unauthorized(id, addr.ip()) else { bail!("too many unauthenticated connections from {}", addr.ip()); }; + // Before the handshake, so its read is bounded too; lifted again at authorization. + stream.set_max_packet_length(MAX_UNAUTHORIZED_MESSAGE); tokio::select! { handshake = identity_handshake(&mut stream, secure) => handshake?, _ = unauthorized.evicted() => { diff --git a/src/server/connection.rs b/src/server/connection.rs index cd180ef97..7ed59b40c 100644 --- a/src/server/connection.rs +++ b/src/server/connection.rs @@ -93,6 +93,14 @@ const MAX_UNAUTHORIZED_CONNS: usize = 64; /// of addresses passes it, and the bound above is what holds. Meaningful only while the /// address is the controller's own, which punch and relay messages carry today. const MAX_UNAUTHORIZED_CONNS_PER_ADDR: usize = 16; +/// The largest message a connection may send before it authorizes. Until then a peer sends only +/// a public key, a login request, a test delay and a close reason, none of which carries an +/// unbounded field - a server hands the login request's avatar out as a URL, and only a custom +/// client that inlines an image into the avatar option instead reaches this. Sized to the read +/// buffer tungstenite allocates per WebSocket connection regardless, so there the cap costs +/// nothing beyond a floor already paid; with MAX_UNAUTHORIZED_CONNS it holds them to 8 MiB in +/// all, against the 1 GiB a single one could make us hold before. +pub const MAX_UNAUTHORIZED_MESSAGE: usize = 128 * 1024; /// A place among the unauthorized connections, taken before the identity handshake and given /// back on drop: at authorization, or when the connection ends first. The count of live @@ -1868,6 +1876,10 @@ impl Connection { if let Some(keep_alive) = self.prepare_terminal_login_for_authorization().await { return keep_alive; } + // Lifted here rather than below with the rest of authorization: a multiplexed tunnel + // narrows it again for its own framing (`port_forward_mux::cap_packet_size`), so that + // call has to come after this one, not before. + self.stream.set_max_packet_length(usize::MAX); if !self.connect_port_forward_if_needed().await { return false; } @@ -7240,6 +7252,41 @@ mod test { assert_eq!(unauthorized_count(), 0, "no handshake outlives the test"); } + // The cap is on before the identity handshake reads. A header declaring one byte over it, + // written to the wire as the codec would read it, ends the handshake on the header alone - + // the payload is neither waited for nor read - and releases the place the connection held. + #[tokio::test] + async fn test_unauthorized_frame_is_refused_on_its_header() { + use hbb_common::tokio::io::AsyncWriteExt; + let _serial = UNAUTHORIZED_TESTS.lock().unwrap_or_else(|e| e.into_inner()); + let server = crate::server::new_for_test(); + let listener = hbb_common::tcp::new_listener("127.0.0.1:0", false) + .await + .unwrap(); + let (controller, accepted) = tokio::join!( + tokio::net::TcpStream::connect(listener.local_addr().unwrap()), + listener.accept() + ); + let (mut controller, (accepted, addr)) = (controller.unwrap(), accepted.unwrap()); + let served = Stream::Tcp(hbb_common::tcp::FramedStream::from(accepted, addr)); + let handshake = tokio::spawn(async move { + crate::server::create_tcp_connection(server, served, addr, true, Default::default()) + .await + }); + let n = MAX_UNAUTHORIZED_MESSAGE + 1; + controller + .write_all(&(((n << 2) | 0x3) as u32).to_le_bytes()) + .await + .unwrap(); + match hbb_common::timeout(2000, handshake).await { + Ok(Ok(Err(e))) => assert!(e.to_string().contains("Too big packet"), "{}", e), + Ok(Ok(Ok(_))) => panic!("a frame over the cap was accepted"), + Ok(Err(e)) => panic!("the handshake task panicked: {}", e), + Err(_) => panic!("the handshake waited for a payload the header should have refused"), + } + assert_eq!(unauthorized_count(), 0); + } + #[cfg(feature = "flutter")] #[cfg(not(any(target_os = "android", target_os = "ios")))] #[test]