From 3fdbaf50061136d9d2e531bf9a55e83095d9e648 Mon Sep 17 00:00:00 2001 From: rustdesk Date: Fri, 4 Sep 2026 00:50:47 +0800 Subject: [PATCH] server: multiplexed port-forward connections stay in the protobuf loop Wire PortForwardMux into Connection: take the multiplexed path at login when the controller sets PortForward.multiplex, route PortForwardChannel frames to it from on_message, sweep the channel table's targets after open/close, and clean it up on connection close. Introduce is_port_forward() (socket-based or multiplexed) and use it at the four sites that classify the connection, so a multiplexed connection stays in the message loop, gets TestDelay keepalives, and reports features.port_forward_mux in PeerInfo. The three sites that break into the raw pipe loop or gate the keepalive still check port_forward_socket specifically, since a multiplexed connection must not take that path. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01EZ49AbZJYfm8NTp5yDPMab --- src/server/connection.rs | 77 +++++++++++++++++++++++++++++++++++++--- 1 file changed, 73 insertions(+), 4 deletions(-) diff --git a/src/server/connection.rs b/src/server/connection.rs index ec7fef120..314a6489f 100644 --- a/src/server/connection.rs +++ b/src/server/connection.rs @@ -257,6 +257,7 @@ pub struct Connection { view_camera: bool, terminal: bool, port_forward_socket: Option>, + port_forward_mux: Option, port_forward_address: String, tx_to_cm: mpsc::UnboundedSender, authorized: bool, @@ -469,6 +470,7 @@ impl Connection { view_camera: false, terminal: false, port_forward_socket: None, + port_forward_mux: None, port_forward_address: "".to_owned(), tx_to_cm, authorized: false, @@ -1044,6 +1046,7 @@ impl Connection { conn.on_close("Timeout", true).await; break; } + conn.push_port_forward_label(); // The control end will jump out of the loop after receiving LoginResponse and will not reply to the TestDelay if conn.last_test_delay.is_none() && !(conn.port_forward_socket.is_some() && conn.authorized) { conn.last_test_delay = Some(Instant::now()); @@ -1665,6 +1668,18 @@ impl Connection { let Some(login_request::Union::PortForward(pf)) = self.lr.union.as_ref() else { return true; }; + if pf.multiplex { + if let Some(tx) = self.inner.tx.clone() { + self.port_forward_mux = Some(super::port_forward_mux::PortForwardMux::new( + tx, + self.port_forward_address.clone(), + )); + return true; + } + // Without a sender there is no way to answer channel frames: fall + // through to the raw pipe below rather than return with neither + // flag set, which would make the connection look like remote desktop. + } let mut pf = pf.clone(); let (mut addr, is_rdp) = Self::normalize_port_forward_target(&mut pf); self.port_forward_address = addr.clone(); @@ -1752,7 +1767,7 @@ impl Connection { self.clear_id_whitelist_failures(); let (conn_type, auth_conn_type) = if self.file_transfer.is_some() { (1, AuthConnType::FileTransfer) - } else if self.port_forward_socket.is_some() { + } else if self.is_port_forward() { (2, AuthConnType::PortForward) } else if self.view_camera { (3, AuthConnType::ViewCamera) @@ -1865,7 +1880,12 @@ impl Connection { pi.platform_additions = serde_json::to_string(&platform_additions).unwrap_or("".into()); } - if self.port_forward_socket.is_some() { + if self.is_port_forward() { + pi.features = Some(Features { + port_forward_mux: self.port_forward_mux.is_some(), + ..Default::default() + }) + .into(); let mut msg_out = Message::new(); res.set_peer_info(pi); msg_out.set_login_response(res); @@ -2063,11 +2083,16 @@ impl Connection { #[inline] fn is_remote(&self) -> bool { self.file_transfer.is_none() - && self.port_forward_socket.is_none() + && !self.is_port_forward() && !self.view_camera && !self.terminal } + #[inline] + fn is_port_forward(&self) -> bool { + self.port_forward_socket.is_some() || self.port_forward_mux.is_some() + } + fn try_sub_monitor_services(&mut self) { let is_remote = self.is_remote(); if is_remote && !self.services_subed { @@ -2211,6 +2236,14 @@ impl Connection { self.tx_to_cm.send(data).ok(); } + fn push_port_forward_label(&mut self) { + let Some(label) = self.port_forward_mux.as_mut().and_then(|m| m.sweep()) else { + return; + }; + self.port_forward_address = label.clone(); + log::debug!("port forward targets now {}", label); + } + #[inline] fn send_fs(&mut self, data: ipc::FS) { self.send_to_cm(ipc::Data::FS(data)); @@ -3867,6 +3900,25 @@ impl Connection { self.refresh_video_display(Some(request.display as usize)); } } + Some(message::Union::PortForwardChannel(ch)) => { + let permitted = + Self::permission(keys::OPTION_ENABLE_TUNNEL, &self.control_permissions); + // Only open/close can change the target set; sweeping after every + // data frame would walk the whole table per 64 KiB. + let may_change_targets = matches!( + ch.union, + Some(port_forward_channel::Union::Open(_)) + | Some(port_forward_channel::Union::Close(_)) + ); + if let Some(mux) = self.port_forward_mux.as_mut() { + mux.handle(ch, permitted); + if may_change_targets { + self.push_port_forward_label(); + } + } else { + log::debug!("port forward channel frame on a non-multiplexed connection"); + } + } Some(message::Union::TerminalAction(action)) => { #[cfg(not(any(target_os = "android", target_os = "ios")))] allow_err!(self.handle_terminal_action(action).await); @@ -5076,6 +5128,9 @@ impl Connection { let data = ipc::Data::Close; self.tx_to_cm.send(data).ok(); self.port_forward_socket.take(); + if let Some(mut mux) = self.port_forward_mux.take() { + mux.close_all(); + } } // The `reason` should be consistent with `check_if_retry` if not empty @@ -5677,7 +5732,7 @@ impl Connection { let allowed = match conn_type { AuthConnType::Remote => true, AuthConnType::FileTransfer => Self::is_file_transfer_scoped_message(msg), - AuthConnType::PortForward => false, + AuthConnType::PortForward => Self::is_port_forward_scoped_message(msg), AuthConnType::ViewCamera => Self::is_view_camera_scoped_message(msg), AuthConnType::Terminal => Self::is_terminal_scoped_message(msg), }; @@ -5751,6 +5806,10 @@ impl Connection { false } + fn is_port_forward_scoped_message(msg: &Message) -> bool { + matches!(msg.union.as_ref(), Some(message::Union::PortForwardChannel(_))) + } + fn is_terminal_scoped_message(msg: &Message) -> bool { match msg.union.as_ref() { Some(message::Union::TerminalAction(_)) => true, @@ -5901,6 +5960,7 @@ impl Connection { Some(message::Union::ScreenshotResponse(_)) => "screenshot_response", Some(message::Union::TerminalAction(_)) => "terminal_action", Some(message::Union::TerminalResponse(_)) => "terminal_response", + Some(message::Union::PortForwardChannel(_)) => "port_forward_channel", Some(message::Union::Misc(misc)) => Self::misc_message_family(misc), Some(_) => "message.other", None => "empty", @@ -7230,6 +7290,10 @@ mod test { }), Some("misc.option"), ), + ( + msg(|m| m.set_port_forward_channel(PortForwardChannel::new())), + Some("port_forward_channel"), + ), ], ), ( @@ -7291,6 +7355,10 @@ mod test { }), Some("misc.option"), ), + ( + msg(|m| m.set_port_forward_channel(PortForwardChannel::new())), + Some("port_forward_channel"), + ), ], ), ( @@ -7391,6 +7459,7 @@ mod test { }), None, ), + (msg(|m| m.set_port_forward_channel(PortForwardChannel::new())), None), ], ), ];