Merge pull request #1127 from 21pages/fix_10054

fix 10054: change direct to relay when RST
2022-08-24 17:05:57 +08:00 · 2022-08-24 17:05:57 +08:00 · 3b3d34fabf
commit 3b3d34fabf
parent 3394975aa2 a7c87a5f57
7 changed files with 166 additions and 35 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@ -1470,6 +1470,27 @@ dependencies = [
 "synstructure",
 ]

+[[package]]
+name = "errno"
+version = "0.2.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f639046355ee4f37944e44f60642c6f3a7efa3cf6b78c78a0d989a8ce6c396a1"
+dependencies = [
+ "errno-dragonfly",
+ "libc",
+ "winapi 0.3.9",
+]
+
+[[package]]
+name = "errno-dragonfly"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "aa68f1b12764fab894d2755d2518754e71b4fd80ecfb822714a1206c2aab39bf"
+dependencies = [
+ "cc",
+ "libc",
+]
+
 [[package]]
 name = "error-code"
 version = "2.3.1"
@ -4168,6 +4189,7 @@ dependencies = [
 "default-net",
 "dispatch",
 "enigo",
+ "errno",
 "evdev",
 "flexi_logger",
 "flutter_rust_bridge",
--- a/Cargo.toml
+++ b/Cargo.toml
@ -58,6 +58,7 @@ num_cpus = "1.13"
 bytes = { version = "1.2", features = ["serde"] }
 default-net = "0.11.0"
 wol-rs = "0.9.1"
+errno = "0.2.8"

 [target.'cfg(not(target_os = "linux"))'.dependencies]
 reqwest = { version = "0.11", features = ["json", "rustls-tls"], default-features=false }
--- a/libs/hbb_common/src/config.rs
+++ b/libs/hbb_common/src/config.rs
@ -21,6 +21,7 @@ use std::{

 pub const RENDEZVOUS_TIMEOUT: u64 = 12_000;
 pub const CONNECT_TIMEOUT: u64 = 18_000;
+pub const READ_TIMEOUT: u64 = 30_000;
 pub const REG_INTERVAL: i64 = 12_000;
 pub const COMPRESS_LEVEL: i32 = 3;
 const SERIAL: i32 = 3;
--- a/src/client.rs
+++ b/src/client.rs
@ -24,7 +24,10 @@ use hbb_common::{
    allow_err,
    anyhow::{anyhow, Context},
    bail,
-    config::{Config, PeerConfig, PeerInfoSerde, CONNECT_TIMEOUT, RELAY_PORT, RENDEZVOUS_TIMEOUT},
+    config::{
+        Config, PeerConfig, PeerInfoSerde, CONNECT_TIMEOUT, READ_TIMEOUT, RELAY_PORT,
+        RENDEZVOUS_TIMEOUT,
+    },
    log,
    message_proto::{option_message::BoolOption, *},
    protobuf::Message as _,
@ -116,8 +119,9 @@ impl Client {
        key: &str,
        token: &str,
        conn_type: ConnType,
+        interface: impl Interface,
    ) -> ResultType<(Stream, bool)> {
-        match Self::_start(peer, key, token, conn_type).await {
+        match Self::_start(peer, key, token, conn_type, interface).await {
            Err(err) => {
                let err_str = err.to_string();
                if err_str.starts_with("Failed") {
@ -135,6 +139,7 @@ impl Client {
        key: &str,
        token: &str,
        conn_type: ConnType,
+        interface: impl Interface,
    ) -> ResultType<(Stream, bool)> {
        // to-do: remember the port for each peer, so that we can retry easier
        let any_addr = Config::get_any_listen_addr();
@ -181,7 +186,11 @@ impl Client {
            log::info!("#{} punch attempt with {}, id: {}", i, my_addr, peer);
            let mut msg_out = RendezvousMessage::new();
            use hbb_common::protobuf::Enum;
-            let nat_type = NatType::from_i32(my_nat_type).unwrap_or(NatType::UNKNOWN_NAT);
+            let nat_type = if interface.is_force_relay() {
+                NatType::SYMMETRIC
+            } else {
+                NatType::from_i32(my_nat_type).unwrap_or(NatType::UNKNOWN_NAT)
+            };
            msg_out.set_punch_hole_request(PunchHoleRequest {
                id: peer.to_owned(),
                token: token.to_owned(),
@ -233,7 +242,15 @@ impl Client {
                            let mut conn =
                                Self::create_relay(peer, rr.uuid, rr.relay_server, key, conn_type)
                                    .await?;
-                            Self::secure_connection(peer, signed_id_pk, key, &mut conn).await?;
+                            Self::secure_connection(
+                                peer,
+                                signed_id_pk,
+                                key,
+                                &mut conn,
+                                false,
+                                interface,
+                            )
+                            .await?;
                            return Ok((conn, false));
                        }
                        _ => {
@ -274,6 +291,7 @@ impl Client {
            key,
            token,
            conn_type,
+            interface,
        )
        .await
    }
@ -292,6 +310,7 @@ impl Client {
        key: &str,
        token: &str,
        conn_type: ConnType,
+        interface: impl Interface,
    ) -> ResultType<(Stream, bool)> {
        let direct_failures = PeerConfig::load(peer_id).direct_failures;
        let mut connect_timeout = 0;
@ -329,8 +348,8 @@ impl Client {
        let start = std::time::Instant::now();
        // NOTICE: Socks5 is be used event in intranet. Which may be not a good way.
        let mut conn = socket_client::connect_tcp(peer, local_addr, connect_timeout).await;
-        let direct = !conn.is_err();
-        if conn.is_err() {
+        let mut direct = !conn.is_err();
+        if interface.is_force_relay() || conn.is_err() {
            if !relay_server.is_empty() {
                conn = Self::request_relay(
                    peer_id,
@ -348,6 +367,7 @@ impl Client {
                        conn.err().unwrap()
                    );
                }
+                direct = false;
            } else {
                bail!("Failed to make direct connection to remote desktop");
            }
@ -360,7 +380,7 @@ impl Client {
        }
        let mut conn = conn?;
        log::info!("{:?} used to establish connection", start.elapsed());
-        Self::secure_connection(peer_id, signed_id_pk, key, &mut conn).await?;
+        Self::secure_connection(peer_id, signed_id_pk, key, &mut conn, direct, interface).await?;
        Ok((conn, direct))
    }

@ -369,6 +389,8 @@ impl Client {
        signed_id_pk: Vec<u8>,
        key: &str,
        conn: &mut Stream,
+        direct: bool,
+        mut interface: impl Interface,
    ) -> ResultType<()> {
        let rs_pk = get_rs_pk(if key.is_empty() {
            hbb_common::config::RS_PUB_KEY
@ -394,9 +416,15 @@ impl Client {
                return Ok(());
            }
        };
-        match timeout(CONNECT_TIMEOUT, conn.next()).await? {
+        match timeout(READ_TIMEOUT, conn.next()).await? {
            Some(res) => {
-                let bytes = res?;
+                let bytes = match res {
+                    Ok(bytes) => bytes,
+                    Err(err) => {
+                        interface.set_force_relay(direct, false);
+                        bail!("{}", err);
+                    }
+                };
                if let Ok(msg_in) = Message::parse_from_bytes(&bytes) {
                    if let Some(message::Union::SignedId(si)) = msg_in.union {
                        if let Ok((id, their_pk_b)) = decode_id_pk(&si.id, &sign_pk) {
@ -786,6 +814,7 @@ pub struct LoginConfigHandler {
    session_id: u64,
    pub supported_encoding: Option<(bool, bool)>,
    pub restarting_remote_device: bool,
+    pub force_relay: bool,
 }

 impl Deref for LoginConfigHandler {
@ -812,6 +841,7 @@ impl LoginConfigHandler {
        self.session_id = rand::random();
        self.supported_encoding = None;
        self.restarting_remote_device = false;
+        self.force_relay = !self.get_option("force-always-relay").is_empty();
    }

    pub fn should_auto_login(&self) -> String {
@ -1418,6 +1448,8 @@ pub trait Interface: Send + Clone + 'static + Sized {
    fn msgbox(&self, msgtype: &str, title: &str, text: &str);
    fn handle_login_error(&mut self, err: &str) -> bool;
    fn handle_peer_info(&mut self, pi: PeerInfo);
+    fn set_force_relay(&mut self, direct: bool, received: bool);
+    fn is_force_relay(&self) -> bool;
    async fn handle_hash(&mut self, pass: &str, hash: Hash, peer: &mut Stream);
    async fn handle_login_from_ui(&mut self, password: String, remember: bool, peer: &mut Stream);
    async fn handle_test_delay(&mut self, t: TestDelay, peer: &mut Stream);
@ -1579,14 +1611,16 @@ lazy_static::lazy_static! {
 pub fn check_if_retry(msgtype: &str, title: &str, text: &str) -> bool {
    msgtype == "error"
        && title == "Connection Error"
-        && !text.to_lowercase().contains("offline")
-        && !text.to_lowercase().contains("exist")
-        && !text.to_lowercase().contains("handshake")
-        && !text.to_lowercase().contains("failed")
-        && !text.to_lowercase().contains("resolve")
-        && !text.to_lowercase().contains("mismatch")
-        && !text.to_lowercase().contains("manually")
-        && !text.to_lowercase().contains("not allowed")
+        && (text.contains("10054")
+            || text.contains("104")
+            || (!text.to_lowercase().contains("offline")
+                && !text.to_lowercase().contains("exist")
+                && !text.to_lowercase().contains("handshake")
+                && !text.to_lowercase().contains("failed")
+                && !text.to_lowercase().contains("resolve")
+                && !text.to_lowercase().contains("mismatch")
+                && !text.to_lowercase().contains("manually")
+                && !text.to_lowercase().contains("not allowed")))
 }

 #[inline]
--- a/src/port_forward.rs
+++ b/src/port_forward.rs
@ -1,7 +1,7 @@
 use crate::client::*;
 use hbb_common::{
    allow_err, bail,
-    config::CONNECT_TIMEOUT,
+    config::READ_TIMEOUT,
    futures::{SinkExt, StreamExt},
    log,
    message_proto::*,
@ -105,22 +105,61 @@ async fn connect_and_login(
    key: &str,
    token: &str,
    is_rdp: bool,
+) -> ResultType<Option<Stream>> {
+    let mut res = connect_and_login_2(
+        id,
+        password,
+        ui_receiver,
+        interface.clone(),
+        forward,
+        key,
+        token,
+        is_rdp,
+    )
+    .await;
+    if res.is_err() && interface.is_force_relay() {
+        res = connect_and_login_2(
+            id,
+            password,
+            ui_receiver,
+            interface,
+            forward,
+            key,
+            token,
+            is_rdp,
+        )
+        .await;
+    }
+    res
+}
+
+async fn connect_and_login_2(
+    id: &str,
+    password: &str,
+    ui_receiver: &mut mpsc::UnboundedReceiver<Data>,
+    interface: impl Interface,
+    forward: &mut Framed<TcpStream, BytesCodec>,
+    key: &str,
+    token: &str,
+    is_rdp: bool,
 ) -> ResultType<Option<Stream>> {
    let conn_type = if is_rdp {
        ConnType::RDP
    } else {
        ConnType::PORT_FORWARD
    };
-    let (mut stream, _) = Client::start(id, key, token, conn_type).await?;
+    let (mut stream, direct) = Client::start(id, key, token, conn_type, interface.clone()).await?;
    let mut interface = interface;
    let mut buffer = Vec::new();
+    let mut received = false;
    loop {
        tokio::select! {
-            res = timeout(CONNECT_TIMEOUT, stream.next()) => match res {
+            res = timeout(READ_TIMEOUT, stream.next()) => match res {
                Err(_) => {
                    bail!("Timeout");
                }
                Ok(Some(Ok(bytes))) => {
+                    received = true;
                    let msg_in = Message::parse_from_bytes(&bytes)?;
                    match msg_in.union {
                        Some(message::Union::Hash(hash)) => {
@ -143,6 +182,11 @@ async fn connect_and_login(
                        _ => {}
                    }
                }
+                Ok(Some(Err(err))) => {
+                    log::error!("Connection closed: {}", err);
+                    interface.set_force_relay(direct, received);
+                    bail!("Connection closed: {}", err);
+                }
                _ => {
                    bail!("Reset by the peer");
                }
--- a/src/ui/ab.tis
+++ b/src/ui/ab.tis
@ -316,7 +316,7 @@ class SessionList: Reactor.Component {
                    <li #connect>{translate('Connect')}</li>
                    <li #transfer>{translate('Transfer File')}</li>
                    <li #tunnel>{translate('TCP Tunneling')}</li>
-                    {false && !handler.using_public_server() && <li #force-always-relay><span>{svg_checkmark}</span>{translate('Always connect via relay')}</li>}
+                    <li #force-always-relay><span>{svg_checkmark}</span>{translate('Always connect via relay')}</li>
                    <li #rdp>RDP<EditRdpPort /></li>
                    <li #wol>{translate('WOL')}</li>
                    <div .separator />
@ -396,7 +396,6 @@ class SessionList: Reactor.Component {
        if (el) {
          var force = handler.get_peer_option(id, "force-always-relay");
          el.attributes.toggleClass("selected", force == "Y");
-          el.attributes.toggleClass("line-through", force != "Y");
        }
        var conn = this.$(menu #connect);
        if (conn) {
--- a/src/ui/remote.rs
+++ b/src/ui/remote.rs
@ -53,6 +53,7 @@ use crate::{
    client::*,
    common::{self, check_clipboard, update_clipboard, ClipboardContext, CLIPBOARD_INTERVAL},
 };
+use errno;

 type Video = AssetPtr<video_destination>;

@ -1456,12 +1457,21 @@ impl Remote {
    async fn io_loop(&mut self, key: &str, token: &str) {
        let stop_clipboard = self.start_clipboard();
        let mut last_recv_time = Instant::now();
+        let mut received = false;
        let conn_type = if self.handler.is_file_transfer() {
            ConnType::FILE_TRANSFER
        } else {
            ConnType::default()
        };
-        match Client::start(&self.handler.id, key, token, conn_type).await {
+        match Client::start(
+            &self.handler.id,
+            key,
+            token,
+            conn_type,
+            self.handler.clone(),
+        )
+        .await
+        {
            Ok((mut peer, direct)) => {
                SERVER_KEYBOARD_ENABLED.store(true, Ordering::SeqCst);
                SERVER_CLIPBOARD_ENABLED.store(true, Ordering::SeqCst);
@ -1484,11 +1494,13 @@ impl Remote {
                                match res {
                                    Err(err) => {
                                        log::error!("Connection closed: {}", err);
+                                        self.handler.set_force_relay(direct, received);
                                        self.handler.msgbox("error", "Connection Error", &err.to_string());
                                        break;
                                    }
                                    Ok(ref bytes) => {
                                        last_recv_time = Instant::now();
+                                        received = true;
                                        self.data_count.fetch_add(bytes.len(), Ordering::Relaxed);
                                        if !self.handle_msg_from_peer(bytes, &mut peer).await {
                                            break
@ -2073,18 +2085,18 @@ impl Remote {

    async fn send_opts_after_login(&self, peer: &mut Stream) {
        if let Some(opts) = self
-        .handler
-        .lc
-        .read()
-        .unwrap()
-        .get_option_message_after_login()
-    {
-        let mut misc = Misc::new();
-        misc.set_option(opts);
-        let mut msg_out = Message::new();
-        msg_out.set_misc(misc);
-        allow_err!(peer.send(&msg_out).await);
-    }
+            .handler
+            .lc
+            .read()
+            .unwrap()
+            .get_option_message_after_login()
+        {
+            let mut misc = Misc::new();
+            misc.set_option(opts);
+            let mut msg_out = Message::new();
+            msg_out.set_misc(misc);
+            allow_err!(peer.send(&msg_out).await);
+        }
    }

    async fn handle_msg_from_peer(&mut self, data: &[u8], peer: &mut Stream) -> bool {
@ -2695,6 +2707,24 @@ impl Interface for Handler {
            handle_test_delay(t, peer).await;
        }
    }
+
+    fn set_force_relay(&mut self, direct: bool, received: bool) {
+        let mut lc = self.lc.write().unwrap();
+        lc.force_relay = false;
+        if direct && !received {
+            let errno = errno::errno().0;
+            log::info!("errno is {}", errno);
+            // TODO: check mac and ios
+            if cfg!(windows) && errno == 10054 || !cfg!(windows) && errno == 104 {
+                lc.force_relay = true;
+                lc.set_option("force-always-relay".to_owned(), "Y".to_owned());
+            }
+        }
+    }
+
+    fn is_force_relay(&self) -> bool {
+        self.lc.read().unwrap().force_relay
+    }
 }

 impl Handler {