TheLanc3/rustdesk
1
0
Fork 0
mirror of https://github.com/rustdesk/rustdesk.git synced 2026-09-07 21:11:05 +03:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
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
This commit is contained in:
rustdesk
2026-09-04 16:48:53 +08:00
parent babac46999
commit c109145c54
12 changed files with 79 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
flutter/lib/models/model.dart
View File

@@ -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") {

10
flutter/lib/models/server_model.dart
View File

@@ -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 =

7
src/flutter.rs
View File

@@ -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)]);
}

1
src/ipc.rs
View File

@@ -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")))]

34
src/port_forward.rs
View File

@@ -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() {
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 {
match tunnel.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) {

95
src/port_forward_mux.rs
View File

@@ -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
/// atomic claim; nothing is ever awaited while it runs.
/// One per listener. The accept loop owns it and reads it between
/// 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 {
let mut outcome = Claim::Wait;
self.state.send_if_modified(|s| match s {
TunnelState::Unset | TunnelState::Failed => {
*s = TunnelState::Establishing;
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;
}
pub fn claim(&self) -> Claim {
match &*self.state.borrow() {
TunnelState::Unset | TunnelState::Failed => Claim::Claimed,
TunnelState::Muxed(h) => Claim::Muxed(h.clone()),
TunnelState::Legacy => Claim::Legacy,
}
}
@@ -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() {
rt().block_on(async {
let t = Arc::new(Tunnel::new());
assert!(matches!(t.try_claim(), Claim::Claimed));
assert!(matches!(t.try_claim(), Claim::Wait));
let w = { let t = t.clone(); tokio::spawn(async move { t.wait_ready().await }) };
t.set_failed();
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)));
});
fn claim_follows_the_tunnel_state() {
let t = Tunnel::new();
assert!(matches!(t.claim(), Claim::Claimed));
t.set_failed();
assert!(matches!(t.claim(), Claim::Claimed));
t.set_legacy();
assert!(matches!(t.claim(), 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)
}

10
src/server/connection.rs
View File

@@ -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");

111
src/server/port_forward_mux.rs
View File

@@ -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)));
});
}
}

4
src/ui/cm.rs
View File

@@ -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) {}
}

7
src/ui/cm.tis
View File

@@ -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) {

24
src/ui_cm_interface.rs
View File

@@ -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);
}

8
src/ui_session_interface.rs
View File

@@ -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
{