From 8e2cb56a53bfea3cc9718136f16cc28041f32cea Mon Sep 17 00:00:00 2001 From: rustdesk Date: Fri, 4 Sep 2026 17:59:27 +0800 Subject: [PATCH] port_forward_mux: a window violation drops the channel on the spot Both demultiplexers only queued a `Violation` and left the entry until the channel task woke and exited, so a peer that kept sending past the window queued one more entry per frame in the meantime, bounded by nothing. The entry now goes the moment `accept` fails; later frames for that id are unknown-channel noise. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01EZ49AbZJYfm8NTp5yDPMab --- src/port_forward_mux.rs | 50 ++++++++++++++++++++++++++++------ src/server/port_forward_mux.rs | 41 +++++++++++++++++++++++----- 2 files changed, 75 insertions(+), 16 deletions(-) diff --git a/src/port_forward_mux.rs b/src/port_forward_mux.rs index 1012533eb..ded2508bc 100644 --- a/src/port_forward_mux.rs +++ b/src/port_forward_mux.rs @@ -551,7 +551,7 @@ mod tunnel { Ok(()) } - fn on_frame(&self, ch: PortForwardChannel) -> Option { + pub(super) fn on_frame(&self, ch: PortForwardChannel) -> Option { match ch.union { Some(port_forward_channel::Union::Opened(o)) => { let refused = { @@ -581,14 +581,18 @@ mod tunnel { log::debug!("port forward data for unknown channel {}", d.channel_id); return None; }; - let msg = if e.window.lock().unwrap().accept(d.data.len()) { - Inbound::Data(d.data) - } else { - log::warn!("port forward channel {} overran its window", d.channel_id); - Inbound::Violation - }; - if e.inbound.send(msg).is_err() { - channels.remove(&d.channel_id); + let accepted = e.window.lock().unwrap().accept(d.data.len()); + let delivered = accepted && e.inbound.send(Inbound::Data(d.data)).is_ok(); + if delivered { + return None; + } + // Dropped here and now, so the peer cannot queue anything more + // for this id while the task is still on its way out. + if let Some(e) = channels.remove(&d.channel_id) { + if !accepted { + log::warn!("port forward channel {} overran its window", d.channel_id); + e.inbound.send(Inbound::Violation).ok(); + } } None } @@ -1272,6 +1276,34 @@ mod tests { }); } + #[test] + fn an_over_window_frame_drops_the_channel_at_once() { + rt().block_on(async { + let (ours, mut peer) = stream_pair().await; + let t = Tunnel::new(); + t.claim(); + let h = t.set_muxed(ours, NoUi::default()); + let (_app, sock) = local_pair().await; + h.open("localhost", 1, sock, vec![]).unwrap(); + let id = match recv_frame(&mut peer).await.union { + Some(port_forward_channel::Union::Open(o)) => o.channel_id, + other => panic!("expected open, got {:?}", other), + }; + let mut ch = PortForwardChannel::new(); + ch.set_data(PortForwardData { + channel_id: id, + data: Bytes::from(vec![0u8; CHANNEL_WINDOW as usize + 1]), + ..Default::default() + }); + h.on_frame(ch.clone()); + // Gone on the spot: a peer that keeps sending past the window + // can no longer queue anything for this channel. + assert_eq!(h.live_channels(), 0); + h.on_frame(ch); + assert_eq!(h.live_channels(), 0); + }); + } + #[test] fn tunnel_death_closes_channels_and_resets_state() { rt().block_on(async { diff --git a/src/server/port_forward_mux.rs b/src/server/port_forward_mux.rs index db3ebd672..428d50679 100644 --- a/src/server/port_forward_mux.rs +++ b/src/server/port_forward_mux.rs @@ -61,14 +61,18 @@ impl PortForwardMux { return; }; let accepted = entry.window.lock().unwrap().accept(len); - let msg = if accepted { - Inbound::Data(d.data) - } else { - log::warn!("port forward channel {} overran its window", d.channel_id); - Inbound::Violation + let delivered = accepted && entry.inbound.send(Inbound::Data(d.data)).is_ok(); + if delivered { + return; + } + // Dropped here and now, so the peer cannot queue anything more + // for this id while the task is still on its way out. + let Some(entry) = self.channels.remove(&d.channel_id) else { + return; }; - if entry.inbound.send(msg).is_err() { - self.channels.remove(&d.channel_id); + if !accepted { + log::warn!("port forward channel {} overran its window", d.channel_id); + entry.inbound.send(Inbound::Violation).ok(); } } Some(port_forward_channel::Union::Close(c)) => { @@ -418,6 +422,29 @@ mod tests { }); } + #[test] + fn an_over_window_frame_drops_the_channel_at_once() { + rt().block_on(async { + let port = echo_target().await; + let (tx, mut rx) = mpsc::unbounded_channel(); + let mut mux = PortForwardMux::new(tx, format!("127.0.0.1:{}", port)); + mux.handle(open(1, port), || true); + opened(&next_frame(&mut rx).await); + let too_much = vec![0u8; CHANNEL_WINDOW as usize + 1]; + mux.handle(data(1, &too_much), || true); + // Gone before the channel task has run: whatever the peer keeps + // sending for this id can no longer queue anything. + assert_eq!(mux.live_channels(), 0); + mux.handle(data(1, &too_much), || true); + assert_eq!(mux.live_channels(), 0); + let ch = next_frame(&mut rx).await; + match &ch.union { + Some(port_forward_channel::Union::Close(c)) => assert_eq!(c.channel_id, 1), + other => panic!("expected close, got {:?}", other), + } + }); + } + #[test] fn open_to_a_target_other_than_the_login_target_is_refused() { rt().block_on(async {