port forward: one tunnel per mapping, bound to the authenticated target

The login latches `PortForward.host`/`port` into the session scope and
approval is shown that target, but a window-wide tunnel let any later
`open` name another target with only `enable-tunnel` rechecked. A
tunnel now belongs to one listener and serves the one target its login
authenticated: the controlled side refuses an `open` for any other
target, and a window with several targets uses one connection each,
approved on its own.

With one owner per tunnel the claim needs no waiters: `Establishing`,
`Claim::Wait` and `wait_ready` go, and `try_claim` becomes a plain
read. The CM label that followed a tunnel's targets goes with them; a
row shows its mapping's target, as before.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EZ49AbZJYfm8NTp5yDPMab

flutter/lib/models/model.dart

@@ -436,8 +436,6 @@ class FfiModel with ChangeNotifier {
         parent.target?.chatModel.onVoiceCallIncoming();
       } else if (name == 'update_voice_call_state') {
         parent.target?.serverModel.updateVoiceCallState(evt);
-      } else if (name == 'update_port_forward') {
-        parent.target?.serverModel.updatePortForward(evt);
       } else if (name == 'fingerprint') {
         FingerprintState.find(peerId).value = evt['fingerprint'] ?? '';
       } else if (name == "sync_peer_hash_password_to_personal_ab") {

flutter/lib/models/server_model.dart

@@ -768,16 +768,6 @@ class ServerModel with ChangeNotifier {
     }
   }
 
-  void updatePortForward(Map<String, dynamic> evt) {
-    final id = int.tryParse(evt['id']?.toString() ?? '');
-    final portForward = evt['port_forward']?.toString() ?? '';
-    if (id == null || portForward.isEmpty) return;
-    final index = _clients.indexWhere((c) => c.id == id);
-    if (index < 0) return;
-    _clients[index].portForward = portForward;
-    notifyListeners();
-  }
-
   void androidUpdatekeepScreenOn() async {
     if (!isAndroid) return;
     var floatingWindowDisabled =

src/flutter.rs

@@ -1517,13 +1517,6 @@ pub mod connection_manager {
             );
         }
 
-        fn update_port_forward(&self, id: i32, port_forward: String) {
-            self.push_event(
-                "update_port_forward",
-                &[("id", &id.to_string()), ("port_forward", &port_forward)],
-            );
-        }
-
         fn change_theme(&self, dark: String) {
             self.push_event("theme", &[("dark", &dark)]);
         }

src/ipc.rs

@@ -348,7 +348,6 @@ pub enum Data {
         name: String,
         enabled: bool,
     },
-    UpdatePortForward(String),
     SystemInfo(Option<String>),
     ClickTime(i64),
     #[cfg(not(any(target_os = "android", target_os = "ios")))]

src/port_forward.rs

@@ -80,9 +80,11 @@ pub async fn listen(
     lc: Arc<RwLock<LoginConfigHandler>>,
     remote_host: String,
     remote_port: i32,
-    tunnel: Arc<Tunnel>,
 ) -> ResultType<()> {
     let listener = tcp::new_listener(format!("127.0.0.1:{}", port), true).await?;
+    // One tunnel per mapping: every accept here goes to the one target the
+    // peer authenticated, and dropping it at the end closes that tunnel.
+    let tunnel = Tunnel::new();
     let addr = listener.local_addr()?;
     log::info!("listening on port {:?}", addr);
     let is_rdp = port == 0;
@@ -97,36 +99,14 @@ pub async fn listen(
             // never shadow it.
             Ok((forward, peer_addr)) = listener.accept() => {
                 log::debug!("new connection from {:?}", peer_addr);
-                let claim = match tunnel.try_claim() {
+                match tunnel.claim() {
-                    Claim::Wait => {
-                        // Keep servicing the UI while the establishing accept holds the
-                        // prompt. `None` means establishment failed: this accept is dropped.
-                        let resolved = loop {
-                            tokio::select! {
-                                r = tunnel.wait_ready() => break r,
-                                d = ui_receiver.recv() => if on_ui_command(d, addr.port(), &lc, &id) {
-                                    return Ok(());
-                                },
-                            }
-                        };
-                        match resolved {
-                            Some(c) => c,
-                            None => {
-                                log::debug!("tunnel failed while {:?} waited; dropping it", peer_addr);
-                                continue;
-                            }
-                        }
-                    }
-                    other => other,
-                };
-                match claim {
                     Claim::Muxed(handle) => {
                         if let Err(e) = handle.open(&remote_host, remote_port, forward, Vec::new()) {
                             log::debug!("cannot open channel for {:?}: {}", peer_addr, e);
                         }
                     }
                     // The claiming accept negotiates: it asks for the tunnel, and
-                    // the peer's answer fixes the window's mode until it closes.
+                    // the peer's answer fixes this listener's mode until it closes.
                     Claim::Claimed => {
                         {
                             let mut lc = lc.write().unwrap();
@@ -164,7 +144,7 @@ pub async fn listen(
                             _ => tunnel.set_failed(),
                         }
                     }
-                    // A `Legacy` window stays legacy until it closes: every accept
+                    // A `Legacy` listener stays legacy until it closes: every accept
                     // logs in on its own, asks for no tunnel, and takes the raw pipe
                     // whatever the peer reports. Reopening the window is how a user
                     // picks up an upgraded peer; nothing switches modes underneath
@@ -187,8 +167,6 @@ pub async fn listen(
                             _ => {}
                         }
                     }
-                    // Resolved above; a stray `Wait` just drops this accept.
-                    Claim::Wait => continue,
                 }
             }
             d = ui_receiver.recv() => if on_ui_command(d, addr.port(), &lc, &id) {

src/port_forward_mux.rs

@@ -419,7 +419,6 @@ mod tunnel {
     // Internal state only; `Claim` is the API listeners see.
     enum TunnelState {
         Unset,
-        Establishing,
         Muxed(Arc<TunnelHandle>),
         Legacy,
         Failed,
@@ -427,16 +426,16 @@ mod tunnel {
 
     pub enum Claim {
         Claimed,
-        Wait,
         Muxed(Arc<TunnelHandle>),
         Legacy,
     }
 
-    /// One per port-forward window. `watch::Sender::send_if_modified` is the
+    /// One per listener. The accept loop owns it and reads it between
-    /// atomic claim; nothing is ever awaited while it runs.
+    /// accepts; the tunnel loop resets it when it ends, so the next accept
+    /// establishes again.
     pub struct Tunnel {
         state: watch::Sender<TunnelState>,
-        /// Never sent on. The loop's receiver errors when the window drops
+        /// Never sent on. The loop's receiver errors when the listener drops
         /// this `Tunnel`, and that is what ends a tunnel nothing else ends.
         lifetime: watch::Sender<()>,
     }
@@ -448,42 +447,11 @@ mod tunnel {
             Self { state, lifetime }
         }
 
-        pub fn try_claim(&self) -> Claim {
+        pub fn claim(&self) -> Claim {
-            let mut outcome = Claim::Wait;
+            match &*self.state.borrow() {
-            self.state.send_if_modified(|s| match s {
+                TunnelState::Unset | TunnelState::Failed => Claim::Claimed,
-                TunnelState::Unset | TunnelState::Failed => {
+                TunnelState::Muxed(h) => Claim::Muxed(h.clone()),
-                    *s = TunnelState::Establishing;
+                TunnelState::Legacy => Claim::Legacy,
-                    outcome = Claim::Claimed;
-                    true
-                }
-                TunnelState::Establishing => false,
-                TunnelState::Muxed(h) => {
-                    outcome = Claim::Muxed(h.clone());
-                    false
-                }
-                TunnelState::Legacy => {
-                    outcome = Claim::Legacy;
-                    false
-                }
-            });
-            outcome
-        }
-
-        /// What the establishing accept ended up with; `None` when it failed.
-        pub async fn wait_ready(&self) -> Option<Claim> {
-            let mut rx = self.state.subscribe();
-            loop {
-                match &*rx.borrow_and_update() {
-                    TunnelState::Muxed(h) => return Some(Claim::Muxed(h.clone())),
-                    TunnelState::Legacy => return Some(Claim::Legacy),
-                    // `Unset` here means the tunnel died between the claim
-                    // and this wait; the waiter treats it as a failure.
-                    TunnelState::Failed | TunnelState::Unset => return None,
-                    TunnelState::Establishing => {}
-                }
-                if rx.changed().await.is_err() {
-                    return None;
-                }
             }
         }
 
@@ -499,7 +467,7 @@ mod tunnel {
             let state = self.state.clone();
             // Publish before spawning: if the loop exits first and resets the
             // state, a later publish here would pin it at Muxed with a dead
-            // handle and the window could never re-establish.
+            // handle and the listener could never re-establish.
             self.state.send_replace(TunnelState::Muxed(handle.clone()));
             tokio::spawn(tunnel_loop(
                 stream,
@@ -1119,19 +1087,13 @@ mod tests {
         }
 
         #[test]
-        fn claim_is_exclusive_and_waiters_see_the_outcome() {
+        fn claim_follows_the_tunnel_state() {
-            rt().block_on(async {
+            let t = Tunnel::new();
-                let t = Arc::new(Tunnel::new());
+            assert!(matches!(t.claim(), Claim::Claimed));
-                assert!(matches!(t.try_claim(), Claim::Claimed));
+            t.set_failed();
-                assert!(matches!(t.try_claim(), Claim::Wait));
+            assert!(matches!(t.claim(), Claim::Claimed));
-                let w = { let t = t.clone(); tokio::spawn(async move { t.wait_ready().await }) };
+            t.set_legacy();
-                t.set_failed();
+            assert!(matches!(t.claim(), Claim::Legacy));
-                assert!(w.await.unwrap().is_none());
-                assert!(matches!(t.try_claim(), Claim::Claimed));
-                t.set_legacy();
-                assert!(matches!(t.try_claim(), Claim::Legacy));
-                assert!(matches!(t.wait_ready().await, Some(Claim::Legacy)));
-            });
         }
 
         #[test]
@@ -1139,7 +1101,7 @@ mod tests {
             rt().block_on(async {
                 let (ours, mut peer) = stream_pair().await;
                 let t = Tunnel::new();
-                assert!(matches!(t.try_claim(), Claim::Claimed));
+                assert!(matches!(t.claim(), Claim::Claimed));
                 let h = t.set_muxed(ours, NoUi::default());
                 let (mut app, sock) = local_pair().await;
                 h.open("localhost", 80, sock, b"GET / HTTP/1.0\r\n\r\n".to_vec()).unwrap();
@@ -1171,7 +1133,7 @@ mod tests {
             rt().block_on(async {
                 let (ours, mut peer) = stream_pair().await;
                 let t = Tunnel::new();
-                assert!(matches!(t.try_claim(), Claim::Claimed));
+                assert!(matches!(t.claim(), Claim::Claimed));
                 let h = t.set_muxed(ours, NoUi::default());
                 let (app, sock) = local_pair().await;
                 h.open("localhost", 80, sock, Vec::new()).unwrap();
@@ -1182,7 +1144,7 @@ mod tests {
                 peer.send(&close_msg(id)).await.unwrap();
                 drop(app);
                 // The loop holds a handle of its own, so dropping ours proves
-                // nothing; the window's `Tunnel` is what must end the peer.
+                // nothing; the listener's `Tunnel` is what must end the peer.
                 drop(h);
                 drop(t);
                 let end = hbb_common::timeout(2000, peer.next()).await;
@@ -1199,7 +1161,7 @@ mod tests {
             rt().block_on(async {
                 let (ours, mut peer) = stream_pair().await;
                 let t = Tunnel::new();
-                assert!(matches!(t.try_claim(), Claim::Claimed));
+                assert!(matches!(t.claim(), Claim::Claimed));
                 let h = t.set_muxed(ours, NoUi::default());
                 // Twenty channels, each with one byte of pipelined data behind
                 // its open. An open on the control queue can lose the loop's
@@ -1239,7 +1201,7 @@ mod tests {
             rt().block_on(async {
                 let (ours, mut peer) = stream_pair().await;
                 let t = Tunnel::new();
-                t.try_claim();
+                t.claim();
                 let h = t.set_muxed(ours, NoUi::default());
                 let (mut app, sock) = local_pair().await;
                 h.open("localhost", 1, sock, vec![]).unwrap();
@@ -1258,7 +1220,7 @@ mod tests {
             rt().block_on(async {
                 let (ours, mut peer) = stream_pair().await;
                 let t = Tunnel::new();
-                assert!(matches!(t.try_claim(), Claim::Claimed));
+                assert!(matches!(t.claim(), Claim::Claimed));
                 let ui = NoUi::default();
                 let h = t.set_muxed(ours, ui.clone());
                 for reason in ["unreachable", "unreachable", "no permission"] {
@@ -1286,7 +1248,7 @@ mod tests {
             rt().block_on(async {
                 let (ours, _peer) = stream_pair().await;
                 let t = Tunnel::new();
-                t.try_claim();
+                t.claim();
                 let h = t.set_muxed(ours, NoUi::default());
                 let t0 = Instant::now();
                 let at = |secs: u64| t0 + std::time::Duration::from_secs(secs);
@@ -1315,7 +1277,7 @@ mod tests {
             rt().block_on(async {
                 let (ours, peer) = stream_pair().await;
                 let t = Tunnel::new();
-                t.try_claim();
+                t.claim();
                 let h = t.set_muxed(ours, NoUi::default());
                 let (mut app, sock) = local_pair().await;
                 h.open("localhost", 1, sock, vec![]).unwrap();
@@ -1323,7 +1285,7 @@ mod tests {
                 let mut buf = [0u8; 1];
                 assert_eq!(app.read(&mut buf).await.unwrap(), 0);
                 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
-                assert!(matches!(t.try_claim(), Claim::Claimed));
+                assert!(matches!(t.claim(), Claim::Claimed));
                 assert!(h.open("localhost", 1, local_pair().await.1, vec![]).is_err());
             });
         }
@@ -1381,7 +1343,6 @@ mod tests {
             tokio::spawn(async move {
                 let (tx, mut rx) = mpsc::unbounded_channel::<(Instant, Arc<Message>)>();
                 let mut mux = PortForwardMux::new(tx, login_target);
-                let mut tick = tokio::time::interval(std::time::Duration::from_millis(100));
                 loop {
                     tokio::select! {
                         Some((_, m)) = rx.recv() => {
@@ -1392,12 +1353,10 @@ mod tests {
                                 let Ok(m) = Message::parse_from_bytes(&bytes) else { continue };
                                 if let Some(message::Union::PortForwardChannel(ch)) = m.union {
                                     mux.handle(ch, || true);
-                                    mux.sweep();
                                 }
                             }
                             _ => return,
                         },
-                        _ = tick.tick() => { mux.sweep(); }
                     }
                 }
             });
@@ -1428,7 +1387,7 @@ mod tests {
             let port = echo_target().await;
             fake_controlled(theirs, format!("127.0.0.1:{}", port));
             let t = Tunnel::new();
-            t.try_claim();
+            t.claim();
             let h = t.set_muxed(ours, NoUi::default());
             (t, h, port)
         }

src/server/connection.rs

@@ -1046,7 +1046,6 @@ 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());
@@ -2232,15 +2231,6 @@ 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::info!("port forward targets now {}", label);
-        self.send_to_cm(ipc::Data::UpdatePortForward(label));
-    }
-
     fn handle_port_forward_channel(&mut self, ch: PortForwardChannel) {
         let Some(mux) = self.port_forward_mux.as_mut() else {
             log::debug!("port forward channel frame on a non-multiplexed connection");

src/server/port_forward_mux.rs

@@ -11,7 +11,7 @@ use hbb_common::{
     tokio::{self, net::TcpStream, sync::mpsc},
 };
 use std::{
-    collections::{BTreeSet, HashMap},
+    collections::HashMap,
     sync::{Arc, Mutex},
 };
 
@@ -27,7 +27,6 @@ struct Entry {
     inbound: mpsc::UnboundedSender<Inbound>,
     credit: Arc<SendCredit>,
     window: Arc<Mutex<RecvWindow>>,
-    target: String,
 }
 
 /// The controlled side of one multiplexed tunnel. The main loop owns it and
@@ -36,17 +35,14 @@ pub struct PortForwardMux {
     channels: HashMap<i32, Entry>,
     tx: Sender,
     login_target: String,
-    last_label: String,
 }
 
 impl PortForwardMux {
     pub fn new(tx: Sender, login_target: String) -> Self {
-        let last_label = login_target.clone();
         Self {
             channels: HashMap::new(),
             tx,
             login_target,
-            last_label,
         }
     }
 
@@ -119,6 +115,18 @@ impl PortForwardMux {
             ..Default::default()
         };
         let (addr, is_rdp) = Connection::normalize_port_forward_target(&mut pf);
+        // Approval and permission checks saw the login's target; a tunnel
+        // serves that one target and nothing else.
+        if addr != self.login_target {
+            log::warn!(
+                "port forward channel {} asked for {} on a tunnel logged in for {}",
+                id,
+                addr,
+                self.login_target
+            );
+            self.reply(opened_msg(id, false, "Port forward target not authorized", 0));
+            return;
+        }
         let (inbound_tx, inbound_rx) = mpsc::unbounded_channel();
         let credit = Arc::new(SendCredit::new(effective_window(open.window)));
         let window = Arc::new(Mutex::new(RecvWindow::new(INITIAL_WINDOW)));
@@ -128,7 +136,6 @@ impl PortForwardMux {
                 inbound: inbound_tx,
                 credit: credit.clone(),
                 window: window.clone(),
-                target: addr.clone(),
             },
         );
         tokio::spawn(run_controlled_channel(
@@ -148,31 +155,6 @@ impl PortForwardMux {
             .ok();
     }
 
-    /// Drops dead entries and recomputes the CM label from the survivors.
-    /// Returns the label only when it differs from the last one returned.
-    pub fn sweep(&mut self) -> Option<String> {
-        self.channels.retain(|_, e| !e.inbound.is_closed());
-        let targets: BTreeSet<&str> = self.channels.values().map(|e| e.target.as_str()).collect();
-        let label = if targets.is_empty() {
-            self.login_target.clone()
-        } else {
-            let first = if targets.contains(self.login_target.as_str()) {
-                self.login_target.as_str()
-            } else {
-                targets.iter().next().copied().unwrap_or(self.login_target.as_str())
-            };
-            match targets.len() - 1 {
-                0 => first.to_owned(),
-                n => format!("{} +{}", first, n),
-            }
-        };
-        if label == self.last_label {
-            return None;
-        }
-        self.last_label = label.clone();
-        Some(label)
-    }
-
     #[cfg(test)]
     pub fn live_channels(&self) -> usize {
         self.channels.len()
@@ -359,7 +341,7 @@ mod tests {
         rt().block_on(async {
             let port = echo_target().await;
             let (tx, mut rx) = mpsc::unbounded_channel();
-            let mut mux = PortForwardMux::new(tx, "localhost:1".to_owned());
+            let mut mux = PortForwardMux::new(tx, format!("127.0.0.1:{}", port));
             mux.handle(open(1, port), || true);
             mux.handle(data(1, b"ping"), || true);
             assert_eq!(opened(&next_frame(&mut rx).await), (1, true));
@@ -375,7 +357,7 @@ mod tests {
             let port = l.local_addr().unwrap().port();
             drop(l);
             let (tx, mut rx) = mpsc::unbounded_channel();
-            let mut mux = PortForwardMux::new(tx, "localhost:1".to_owned());
+            let mut mux = PortForwardMux::new(tx, format!("127.0.0.1:{}", port));
             mux.handle(open(1, port), || true);
             mux.handle(data(1, b"lost"), || true);
             assert_eq!(opened(&next_frame(&mut rx).await), (1, false));
@@ -388,7 +370,7 @@ mod tests {
         rt().block_on(async {
             let port = echo_target().await;
             let (tx, mut rx) = mpsc::unbounded_channel();
-            let mut mux = PortForwardMux::new(tx, "localhost:1".to_owned());
+            let mut mux = PortForwardMux::new(tx, format!("127.0.0.1:{}", port));
             mux.handle(open(1, port), || false);
             assert_eq!(opened(&next_frame(&mut rx).await), (1, false));
             assert_eq!(mux.live_channels(), 0);
@@ -403,7 +385,7 @@ mod tests {
             // that same poll loses: no `opened` may ever be sent.
             let port = echo_target().await;
             let (tx, mut rx) = mpsc::unbounded_channel();
-            let mut mux = PortForwardMux::new(tx, "localhost:1".to_owned());
+            let mut mux = PortForwardMux::new(tx, format!("127.0.0.1:{}", port));
             mux.handle(open(1, port), || true);
             mux.handle(close(1), || true);
             assert!(tokio::time::timeout(std::time::Duration::from_millis(200), rx.recv()).await.is_err());
@@ -416,7 +398,7 @@ mod tests {
         rt().block_on(async {
             let port = echo_target().await;
             let (tx, mut rx) = mpsc::unbounded_channel();
-            let mut mux = PortForwardMux::new(tx, "localhost:1".to_owned());
+            let mut mux = PortForwardMux::new(tx, format!("127.0.0.1:{}", port));
             mux.handle(open(1, port), || true);
             mux.handle(open(2, port), || true);
             let mut seen = 0;
@@ -436,6 +418,29 @@ mod tests {
         });
     }
 
+    #[test]
+    fn open_to_a_target_other_than_the_login_target_is_refused() {
+        rt().block_on(async {
+            let a = echo_target().await;
+            let b = echo_target().await;
+            let (tx, mut rx) = mpsc::unbounded_channel();
+            let mut mux = PortForwardMux::new(tx, format!("127.0.0.1:{}", a));
+            mux.handle(open(1, a), || true);
+            assert_eq!(opened(&next_frame(&mut rx).await), (1, true));
+            // Approval was for target a; b needs a login of its own.
+            mux.handle(open(2, b), || true);
+            let ch = next_frame(&mut rx).await;
+            match &ch.union {
+                Some(port_forward_channel::Union::Opened(o)) => {
+                    assert_eq!((o.channel_id, o.success), (2, false));
+                    assert!(!o.message.is_empty());
+                }
+                other => panic!("expected opened, got {:?}", other),
+            }
+            assert_eq!(mux.live_channels(), 1);
+        });
+    }
+
     #[test]
     fn demux_admits_only_the_initial_window_before_opened() {
         rt().block_on(async {
@@ -469,7 +474,7 @@ mod tests {
     }
 
     /// A target that accepts and hangs up at once, so every channel ends on
-    /// the target's EOF — the case where only the sweep can free the entry.
+    /// the target's EOF — the case where only the next `open` frees the entry.
     async fn drop_target() -> u16 {
         let l = TcpListener::bind("127.0.0.1:0").await.unwrap();
         let port = l.local_addr().unwrap().port();
@@ -483,16 +488,16 @@ mod tests {
     }
 
     #[test]
-    fn sweep_frees_dead_entries_so_the_cap_counts_live_channels() {
+    fn open_frees_dead_entries_so_the_cap_counts_live_channels() {
         rt().block_on(async {
             let port = drop_target().await;
             let (tx, mut rx) = mpsc::unbounded_channel();
-            let mut mux = PortForwardMux::new(tx, "localhost:1".to_owned());
+            let mut mux = PortForwardMux::new(tx, format!("127.0.0.1:{}", port));
             for id in 1..=(MAX_CHANNELS as i32 * 2) {
                 mux.handle(open(id, port), || true);
                 assert_eq!(opened(&next_frame(&mut rx).await), (id, true));
                 // The task sends `close` on the target's EOF and exits; the
-                // entry is dead until the next `open` sweeps it.
+                // entry is dead until the next `open` drops it.
                 let ch = next_frame(&mut rx).await;
                 match &ch.union {
                     Some(port_forward_channel::Union::Close(c)) => assert_eq!(c.channel_id, id),
@@ -500,32 +505,6 @@ mod tests {
                 }
                 tokio::task::yield_now().await;
             }
-            // A task has exited by the time its `close` is read, so one sweep
-            // must leave nothing behind.
-            mux.sweep();
-            assert_eq!(mux.live_channels(), 0);
-        });
-    }
-
-    #[test]
-    fn label_counts_distinct_targets_only() {
-        rt().block_on(async {
-            let a = echo_target().await;
-            let b = echo_target().await;
-            let (tx, mut rx) = mpsc::unbounded_channel();
-            let mut mux = PortForwardMux::new(tx, format!("127.0.0.1:{}", a));
-            assert_eq!(mux.sweep(), None);
-            mux.handle(open(1, a), || true);
-            mux.handle(open(2, a), || true);
-            opened(&next_frame(&mut rx).await);
-            opened(&next_frame(&mut rx).await);
-            assert_eq!(mux.sweep(), None);
-            mux.handle(open(3, b), || true);
-            opened(&next_frame(&mut rx).await);
-            assert_eq!(mux.sweep(), Some(format!("127.0.0.1:{} +1", a)));
-            mux.handle(close(3), || true);
-            tokio::task::yield_now().await;
-            assert_eq!(mux.sweep(), Some(format!("127.0.0.1:{}", a)));
         });
     }
 }

src/ui/cm.rs

@@ -72,10 +72,6 @@ impl InvokeUiCM for SciterHandler {
         );
     }
 
-    fn update_port_forward(&self, id: i32, port_forward: String) {
-        self.call("updatePortForward", &make_args!(id, port_forward));
-    }
-
     fn file_transfer_log(&self, _action: &str, _log: &str) {}
 }
 

src/ui/cm.tis

@@ -434,13 +434,6 @@ handler.addConnection = function(id, is_file_transfer, is_view_camera, is_termin
     }
 }
 
-handler.updatePortForward = function(id, port_forward) {
-    connections.map(function(c) {
-        if (c.id == id) c.port_forward = port_forward;
-    });
-    update();
-}
-
 handler.removeConnection = function(id, close) {
     var i = -1;
     connections.map(function(c, idx) {

src/ui_cm_interface.rs

@@ -196,8 +196,6 @@ pub trait InvokeUiCM: Send + Clone + 'static + Sized {
     fn update_voice_call_state(&self, client: &Client);
 
     fn file_transfer_log(&self, action: &str, log: &str);
-
-    fn update_port_forward(&self, id: i32, port_forward: String);
 }
 
 impl<T: InvokeUiCM> Deref for ConnectionManager<T> {
@@ -604,17 +602,6 @@ impl<T: InvokeUiCM> IpcTaskRunner<T> {
                                         }
                                     }
                                 }
-                                Data::UpdatePortForward(port_forward) => {
-                                    let updated = {
-                                        let mut clients = CLIENTS.write().unwrap();
-                                        clients.get_mut(&self.conn_id).map(|c| {
-                                            c.port_forward = port_forward.clone();
-                                        })
-                                    };
-                                    if updated.is_some() {
-                                        self.cm.ui_handler.update_port_forward(self.conn_id, port_forward);
-                                    }
-                                }
                                 Data::FS(mut fs) => {
                                     if let ipc::FS::WriteBlock { id, file_num, data: _, compressed } = fs {
                                         if let Ok(bytes) = self.stream.next_raw().await {
@@ -963,17 +950,6 @@ pub async fn start_listen<T: InvokeUiCM>(
             Some(Data::Close) => {
                 break;
             }
-            Some(Data::UpdatePortForward(port_forward)) => {
-                let updated = {
-                    let mut clients = CLIENTS.write().unwrap();
-                    clients.get_mut(&current_id).map(|c| {
-                        c.port_forward = port_forward.clone();
-                    })
-                };
-                if updated.is_some() {
-                    cm.ui_handler.update_port_forward(current_id, port_forward);
-                }
-            }
             Some(Data::StartVoiceCall) => {
                 cm.voice_call_started(current_id);
             }

src/ui_session_interface.rs

@@ -1944,7 +1944,6 @@ pub async fn io_loop<T: InvokeUiSession>(handler: Session<T>, round: u32) {
     let key = crate::get_key(false).await;
     #[cfg(not(any(target_os = "android", target_os = "ios")))]
     if handler.is_port_forward() {
-        let tunnel = std::sync::Arc::new(crate::port_forward_mux::Tunnel::new());
         if handler.is_rdp() {
             let port = handler
                 .get_option("rdp_port".to_owned())
@@ -1959,7 +1958,7 @@ pub async fn io_loop<T: InvokeUiSession>(handler: Session<T>, round: u32) {
                 handler.get_option("rdp_password".to_owned()),
             );
             log::info!("Remote rdp port: {}", port);
-            start_one_port_forward(handler, 0, "".to_owned(), port, receiver, &key, &token, tunnel).await;
+            start_one_port_forward(handler, 0, "".to_owned(), port, receiver, &key, &token).await;
         } else if handler.args.len() == 0 {
             let pfs = handler.lc.read().unwrap().port_forwards.clone();
             let mut queues = HashMap::<i32, mpsc::UnboundedSender<Data>>::new();
@@ -1977,7 +1976,6 @@ pub async fn io_loop<T: InvokeUiSession>(handler: Session<T>, round: u32) {
                         let handler = handler.clone();
                         let key = key.clone();
                         let token = token.clone();
-                        let tunnel = tunnel.clone();
                         tokio::spawn(async move {
                             start_one_port_forward(
                                 handler,
@@ -1987,7 +1985,6 @@ pub async fn io_loop<T: InvokeUiSession>(handler: Session<T>, round: u32) {
                                 receiver,
                                 &key,
                                 &token,
-                                tunnel,
                             )
                             .await;
                         });
@@ -2026,7 +2023,6 @@ pub async fn io_loop<T: InvokeUiSession>(handler: Session<T>, round: u32) {
                 receiver,
                 &key,
                 &token,
-                tunnel,
             )
             .await;
         }
@@ -2046,7 +2042,6 @@ async fn start_one_port_forward<T: InvokeUiSession>(
     receiver: mpsc::UnboundedReceiver<Data>,
     key: &str,
     token: &str,
-    tunnel: std::sync::Arc<crate::port_forward_mux::Tunnel>,
 ) {
     if let Err(err) = crate::port_forward::listen(
         handler.get_id(),
@@ -2059,7 +2054,6 @@ async fn start_one_port_forward<T: InvokeUiSession>(
         handler.lc.clone(),
         remote_host,
         remote_port,
-        tunnel,
     )
     .await
     {