diff --git a/Cargo.lock b/Cargo.lock index 2ab6bf1..7d52d24 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -94,7 +94,7 @@ version = "0.16.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fdb8867f378f33f78a811a8eb9bf108ad99430d7aad43315dd9319c827ef6247" dependencies = [ - "http", + "http 0.2.12", "log", "url", "wildmatch", @@ -301,7 +301,7 @@ dependencies = [ [[package]] name = "common" -version = "1.2.10" +version = "1.2.11" dependencies = [ "anyhow", "chrono", @@ -427,6 +427,12 @@ dependencies = [ "cipher", ] +[[package]] +name = "data-encoding" +version = "2.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8566979429cf69b49a5c740c60791108e86440e8be149bbea4fe54d2c32d6e2" + [[package]] name = "der" version = "0.7.9" @@ -534,6 +540,50 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "futures-core" +version = "0.3.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfc6580bb841c5a68e9ef15c77ccc837b40a7504914d52e47b8b0e9bbda25a1d" + +[[package]] +name = "futures-macro" +version = "0.3.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87750cf4b7a4c0625b1529e4c543c2182106e4dedc60a2a6455e00d212c489ac" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.60", +] + +[[package]] +name = "futures-sink" +version = "0.3.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fb8e00e87438d937621c1c6269e53f536c14d3fbd6a042bb24879e57d474fb5" + +[[package]] +name = "futures-task" +version = "0.3.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "38d84fa142264698cdce1a9f9172cf383a0c82de1bddcf3092901442c4097004" + +[[package]] +name = "futures-util" +version = "0.3.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d6401deb83407ab3da39eba7e33987a73c3df0c82b4bb5813ee871c19c41d48" +dependencies = [ + "futures-core", + "futures-macro", + "futures-sink", + "futures-task", + "pin-project-lite", + "pin-utils", + "slab", +] + [[package]] name = "generic-array" version = "0.14.7" @@ -636,6 +686,23 @@ dependencies = [ "itoa", ] +[[package]] +name = "http" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "21b9ddb458710bc376481b842f5da65cdf31522de232c1ca8146abce2a358258" +dependencies = [ + "bytes", + "fnv", + "itoa", +] + +[[package]] +name = "httparse" +version = "1.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fcc0b4a115bf80b728eb8ea024ad5bd707b615bfed49e0665b6e0f86fd082d9" + [[package]] name = "humantime" version = "2.1.0" @@ -1112,6 +1179,12 @@ version = "0.2.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bda66fc9667c18cb2758a2ac84d1167245054bcf85d5d1aaa6923f45801bdd02" +[[package]] +name = "pin-utils" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" + [[package]] name = "pkcs1" version = "0.7.5" @@ -1503,6 +1576,17 @@ dependencies = [ "unsafe-libyaml", ] +[[package]] +name = "sha1" +version = "0.10.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "sha2" version = "0.10.8" @@ -1549,6 +1633,15 @@ dependencies = [ "rand_core", ] +[[package]] +name = "slab" +version = "0.4.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f92a496fb766b417c996b9c5e57daf2f7ad3b0bebe1ccfca4856390e3d3bb67" +dependencies = [ + "autocfg", +] + [[package]] name = "smallvec" version = "1.13.2" @@ -1732,6 +1825,18 @@ dependencies = [ "syn 2.0.60", ] +[[package]] +name = "tokio-tungstenite" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6989540ced10490aaf14e6bad2e3d33728a2813310a0c71d1574304c49631cd" +dependencies = [ + "futures-util", + "log", + "tokio", + "tungstenite", +] + [[package]] name = "tun" version = "0.1.0" @@ -1745,6 +1850,24 @@ dependencies = [ "winapi", ] +[[package]] +name = "tungstenite" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e2e2ce1e47ed2994fd43b04c8f618008d4cabdd5ee34027cf14f9d918edd9c8" +dependencies = [ + "byteorder", + "bytes", + "data-encoding", + "http 1.1.0", + "httparse", + "log", + "rand", + "sha1", + "thiserror", + "utf-8", +] + [[package]] name = "typemap-ors" version = "1.0.0" @@ -1829,6 +1952,12 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "utf-8" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" + [[package]] name = "uuid" version = "1.8.0" @@ -1852,7 +1981,7 @@ checksum = "49874b5167b65d7193b8aba1567f5c7d93d001cafc34600cee003eda787e483f" [[package]] name = "vn-link" -version = "1.2.10" +version = "1.2.11" dependencies = [ "anyhow", "crossbeam-utils", @@ -1865,7 +1994,7 @@ dependencies = [ [[package]] name = "vn-link-cli" -version = "1.2.10" +version = "1.2.11" dependencies = [ "common", "log", @@ -1875,7 +2004,7 @@ dependencies = [ [[package]] name = "vnt" -version = "1.2.10" +version = "1.2.11" dependencies = [ "aes", "aes-gcm", @@ -1891,6 +2020,7 @@ dependencies = [ "dns-parser", "ecb", "fnv", + "futures-util", "igd", "libc", "libloading", @@ -1913,13 +2043,14 @@ dependencies = [ "stun-format", "thiserror", "tokio", + "tokio-tungstenite", "tun", "zstd", ] [[package]] name = "vnt-cli" -version = "1.2.10" +version = "1.2.11" dependencies = [ "anyhow", "chrono", diff --git a/common/Cargo.toml b/common/Cargo.toml index 60a600d..66d2379 100644 --- a/common/Cargo.toml +++ b/common/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "common" -version = "1.2.10" +version = "1.2.11" edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html @@ -33,6 +33,7 @@ port_mapping = ["vnt/port_mapping"] lz4 = ["vnt/lz4_compress"] zstd = ["vnt/zstd_compress"] upnp = ["vnt/upnp"] +websocket = ["vnt/websocket"] command = [] file_config = [] log = ["log4rs"] diff --git a/common/src/cli.rs b/common/src/cli.rs index e137104..6e81559 100644 --- a/common/src/cli.rs +++ b/common/src/cli.rs @@ -56,7 +56,6 @@ pub fn parse_args_config() -> anyhow::Result, bool)> opts.optopt("w", "", "客户端加密", ""); opts.optflag("W", "", "服务端加密"); opts.optopt("u", "", "自定义mtu(默认为1430)", ""); - opts.optflag("", "tcp", "tcp"); opts.optopt("", "ip", "指定虚拟ip", ""); opts.optflag("", "relay", "仅使用服务器转发"); opts.optopt("", "par", "任务并行度(必须为正整数)", ""); @@ -213,7 +212,6 @@ pub fn parse_args_config() -> anyhow::Result, bool)> return Err(anyhow::anyhow!("'--ip {}' invalid", virtual_ip)); } } - let tcp_channel = matches.opt_present("tcp"); let relay = matches.opt_present("relay"); let cipher_model = match matches.opt_get::("model") { @@ -290,7 +288,6 @@ pub fn parse_args_config() -> anyhow::Result, bool)> out_ip, password, mtu, - tcp_channel, virtual_ip, #[cfg(feature = "integrated_tun")] #[cfg(feature = "ip_proxy")] @@ -339,7 +336,9 @@ fn print_usage(program: &str, _opts: Options) { ); println!(" -n 给设备一个名字,便于区分不同设备,默认使用系统版本"); println!(" -d 设备唯一标识符,不使用--ip参数时,服务端凭此参数分配虚拟ip,注意不能重复"); - println!(" -s 注册和中继服务器地址,以'TXT:'开头表示解析TXT记录"); + println!( + " -s 注册和中继服务器地址,协议支持使用tcp://和ws://和wss://,默认为udp://" + ); println!(" -e stun服务器,用于探测NAT类型,可使用多个地址,如-e stun.miwifi.com -e turn.cloudflare.com"); #[cfg(target_os = "windows")] #[cfg(feature = "integrated_tun")] @@ -357,7 +356,6 @@ fn print_usage(program: &str, _opts: Options) { #[cfg(feature = "file_config")] println!(" -f 读取配置文件中的配置"); - println!(" --tcp 和服务端使用tcp通信,默认使用udp,遇到udp qos时可指定使用tcp"); println!(" --ip 指定虚拟ip,指定的ip不能和其他设备重复,必须有效并且在服务端所属网段下,默认情况由服务端分配"); let mut enums = String::new(); #[cfg(any(feature = "aes_gcm", feature = "server_encrypt"))] diff --git a/common/src/command/entity.rs b/common/src/command/entity.rs index 5c06d18..b741488 100644 --- a/common/src/command/entity.rs +++ b/common/src/command/entity.rs @@ -17,6 +17,8 @@ pub struct Info { pub port_mapping_list: Vec<(bool, SocketAddr, String)>, pub in_ips: Vec<(u32, u32, Ipv4Addr)>, pub out_ips: Vec<(u32, u32)>, + pub udp_listen_addr: Vec, + pub tcp_listen_addr: String, } #[derive(Serialize, Deserialize, Debug)] diff --git a/common/src/command/mod.rs b/common/src/command/mod.rs index 069e622..0f2f1d1 100644 --- a/common/src/command/mod.rs +++ b/common/src/command/mod.rs @@ -1,4 +1,5 @@ use std::io; +use vnt::channel::ConnectProtocol; use vnt::core::Vnt; use crate::command::entity::{DeviceItem, Info, RouteItem}; @@ -81,6 +82,7 @@ fn command_(cmd: CommandEnum) -> io::Result<()> { pub fn command_route(vnt: &Vnt) -> Vec { let route_table = vnt.route_table(); + let server_addr = vnt.config().server_address_str.clone(); let mut route_list = Vec::with_capacity(route_table.len()); for (destination, routes) in route_table { for route in routes { @@ -93,11 +95,14 @@ pub fn command_route(vnt: &Vnt) -> Vec { } else { route.rt.to_string() }; - let interface = if route.is_tcp { - format!("tcp@{}", route.addr) - } else { - route.addr.to_string() + let interface = match route.protocol { + ConnectProtocol::UDP => route.addr.to_string(), + ConnectProtocol::TCP => { + format!("tcp@{}", route.addr) + } + ConnectProtocol::WS | ConnectProtocol::WSS => server_addr.clone(), }; + let item = RouteItem { destination: destination.to_string(), next_hop, @@ -145,7 +150,7 @@ pub fn command_list(vnt: &Vnt) -> Vec { }; let (nat_traversal_type, rt) = if let Some(route) = vnt.route(&peer.virtual_ip) { let nat_traversal_type = if route.metric == 1 { - if route.is_tcp { + if route.protocol.is_base_tcp() { "tcp-p2p" } else { "p2p" @@ -195,6 +200,7 @@ pub fn command_list(vnt: &Vnt) -> Vec { } pub fn command_info(vnt: &Vnt) -> Info { + let config = vnt.config(); let current_device = vnt.current_device(); let nat_info = vnt.nat_info(); let name = vnt.name().to_string(); @@ -202,7 +208,11 @@ pub fn command_info(vnt: &Vnt) -> Info { let virtual_gateway = current_device.virtual_gateway().to_string(); let virtual_netmask = current_device.virtual_netmask.to_string(); let connect_status = format!("{:?}", vnt.connection_status()); - let relay_server = current_device.connect_server.to_string(); + let relay_server = if current_device.connect_server.port() == 0 { + config.server_address_str.clone() + } else { + current_device.connect_server.to_string() + }; let nat_type = format!("{:?}", nat_info.nat_type); let public_ips: Vec = nat_info.public_ips.iter().map(|v| v.to_string()).collect(); let public_ips = public_ips.join(","); @@ -222,6 +232,12 @@ pub fn command_info(vnt: &Vnt) -> Info { let port_mapping_list = vec![]; let in_ips = vnt.config().in_ips.clone(); let out_ips = vnt.config().out_ips.clone(); + let udp_listen_addr = nat_info + .udp_ports + .iter() + .map(|port| format!("0.0.0.0:{}", port)) + .collect(); + let tcp_listen_addr = format!("0.0.0.0:{}", nat_info.tcp_port); Info { name, virtual_ip, @@ -238,5 +254,7 @@ pub fn command_info(vnt: &Vnt) -> Info { port_mapping_list, in_ips, out_ips, + udp_listen_addr, + tcp_listen_addr, } } diff --git a/common/src/config/file_config.rs b/common/src/config/file_config.rs index 6d177c7..4af6075 100644 --- a/common/src/config/file_config.rs +++ b/common/src/config/file_config.rs @@ -156,7 +156,6 @@ pub fn read_config(file_path: &str) -> anyhow::Result<(Config, Vec, bool out_ips, file_conf.password, file_conf.mtu, - file_conf.tcp, virtual_ip, #[cfg(feature = "integrated_tun")] #[cfg(feature = "ip_proxy")] diff --git a/common/src/console_out/mod.rs b/common/src/console_out/mod.rs index 44bde16..641767f 100644 --- a/common/src/console_out/mod.rs +++ b/common/src/console_out/mod.rs @@ -21,6 +21,11 @@ pub fn console_info(status: Info) { println!("NAT type: {}", style(status.nat_type).green()); println!("Relay server: {}", style(status.relay_server).green()); + println!( + "Udp listen: {}", + style(status.udp_listen_addr.join(", ")).green() + ); + println!("Tcp listen: {}", style(status.tcp_listen_addr).green()); println!("Public ips: {}", style(status.public_ips).green()); println!("Local addr: {}", style(status.local_addr).green()); println!("IPv6: {}", style(status.ipv6_addr).green()); diff --git a/vn-link-cli/Cargo.toml b/vn-link-cli/Cargo.toml index 23a1a9a..8b471b8 100644 --- a/vn-link-cli/Cargo.toml +++ b/vn-link-cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "vn-link-cli" -version = "1.2.10" +version = "1.2.11" edition = "2021" [dependencies] @@ -11,18 +11,20 @@ log = "0.4.17" [features] default = ["default-feature"] -openssl = ["vn-link/openssl","common/openssl"] -openssl-vendored = ["vn-link/openssl-vendored","common/openssl-vendored"] -ring-cipher = ["vn-link/ring-cipher","common/ring-cipher"] -aes_cbc = ["vn-link/aes_cbc","common/aes_cbc"] -aes_ecb = ["vn-link/aes_ecb","common/aes_ecb"] -sm4_cbc = ["vn-link/sm4_cbc","common/sm4_cbc"] -aes_gcm = ["vn-link/aes_gcm","common/aes_gcm"] -chacha20_poly1305 = ["vn-link/chacha20_poly1305","common/chacha20_poly1305"] -server_encrypt = ["vn-link/server_encrypt","common/server_encrypt"] -port_mapping = ["vn-link/port_mapping","common/port_mapping"] -lz4 = ["vn-link/lz4_compress","common/lz4"] -zstd = ["vn-link/zstd_compress","common/zstd"] +openssl = ["vn-link/openssl", "common/openssl"] +openssl-vendored = ["vn-link/openssl-vendored", "common/openssl-vendored"] +ring-cipher = ["vn-link/ring-cipher", "common/ring-cipher"] +aes_cbc = ["vn-link/aes_cbc", "common/aes_cbc"] +aes_ecb = ["vn-link/aes_ecb", "common/aes_ecb"] +sm4_cbc = ["vn-link/sm4_cbc", "common/sm4_cbc"] +aes_gcm = ["vn-link/aes_gcm", "common/aes_gcm"] +chacha20_poly1305 = ["vn-link/chacha20_poly1305", "common/chacha20_poly1305"] +server_encrypt = ["vn-link/server_encrypt", "common/server_encrypt"] +port_mapping = ["vn-link/port_mapping", "common/port_mapping"] +lz4 = ["vn-link/lz4_compress", "common/lz4"] +zstd = ["vn-link/zstd_compress", "common/zstd"] +upnp = ["vn-link/upnp", "common/upnp"] +websocket = ["vn-link/websocket", "common/websocket"] log = ["common/log"] command = ["common/command"] file_config = ["common/file_config"] diff --git a/vn-link/Cargo.toml b/vn-link/Cargo.toml index b9429a5..7090458 100644 --- a/vn-link/Cargo.toml +++ b/vn-link/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "vn-link" -version = "1.2.10" +version = "1.2.11" edition = "2021" [dependencies] @@ -27,3 +27,5 @@ server_encrypt = ["vnt/server_encrypt"] port_mapping = ["vnt/port_mapping"] lz4_compress = ["vnt/lz4_compress"] zstd_compress = ["vnt/zstd_compress"] +upnp = ["vnt/upnp"] +websocket = ["vnt/websocket"] \ No newline at end of file diff --git a/vnt-cli/Cargo.toml b/vnt-cli/Cargo.toml index f4dfaa4..397a2b2 100644 --- a/vnt-cli/Cargo.toml +++ b/vnt-cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "vnt-cli" -version = "1.2.10" +version = "1.2.11" edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html @@ -40,6 +40,7 @@ log = ["common/log"] command = ["common/command"] file_config = ["common/file_config"] upnp = ["common/upnp"] +websocket = ["common/websocket"] [build-dependencies] rand = "0.8.5" diff --git a/vnt/Cargo.toml b/vnt/Cargo.toml index aa2c7c1..d6108d1 100644 --- a/vnt/Cargo.toml +++ b/vnt/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "vnt" -version = "1.2.10" +version = "1.2.11" edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html @@ -37,13 +37,15 @@ crossbeam-queue = "0.3.11" anyhow = "1.0.82" dns-parser = "0.8.0" -tokio = { version = "1.37.0", features = ["full"], optional = true } +tokio = { version = "1.37.0", features = ["full"] } lz4_flex = { version = "0.11", default-features = false, optional = true } zstd = { version = "0.13.1", optional = true } fnv = "1.0.7" igd = { version = "0.12.1", optional = true } +tokio-tungstenite = { version = "0.23.1", optional = true } +futures-util = "0.3.30" [target.'cfg(target_os = "windows")'.dependencies] libloading = "0.8.0" @@ -54,7 +56,7 @@ protoc-bin-vendored = "3.0.0" cfg_aliases = "0.2.1" [features] -default = ["server_encrypt", "aes_gcm", "aes_cbc", "aes_ecb", "sm4_cbc", "chacha20_poly1305", "ip_proxy", "port_mapping", "lz4_compress", "zstd_compress", "integrated_tun"] +default = ["websocket", "server_encrypt", "aes_gcm", "aes_cbc", "aes_ecb", "sm4_cbc", "chacha20_poly1305", "ip_proxy", "port_mapping", "lz4_compress", "zstd_compress", "integrated_tun"] openssl = ["openssl-sys"] # 从源码编译 openssl-vendored = ["openssl-sys/vendored"] @@ -65,9 +67,10 @@ sm4_cbc = ["libsm"] aes_gcm = ["aes-gcm"] chacha20_poly1305 = ["chacha20poly1305", "chacha20"] server_encrypt = ["aes-gcm", "rsa", "spki"] -ip_proxy = ["tokio"] -port_mapping = ["tokio"] +ip_proxy = [] +port_mapping = [] lz4_compress = ["lz4_flex"] zstd_compress = ["zstd"] integrated_tun = ["tun"] upnp = ["igd"] +websocket = ["tokio-tungstenite"] \ No newline at end of file diff --git a/vnt/src/channel/context.rs b/vnt/src/channel/context.rs index 203a218..6c5da52 100644 --- a/vnt/src/channel/context.rs +++ b/vnt/src/channel/context.rs @@ -12,7 +12,7 @@ use rand::Rng; use crate::channel::punch::NatType; use crate::channel::sender::{AcceptSocketSender, PacketSender}; -use crate::channel::{Route, RouteKey, UseChannelType, DEFAULT_RT}; +use crate::channel::{ConnectProtocol, Route, RouteKey, UseChannelType, DEFAULT_RT}; /// 传输通道上下文，持有udp socket、tcp socket和路由信息 #[derive(Clone)] @@ -25,7 +25,7 @@ impl ChannelContext { main_udp_socket: Vec, use_channel_type: UseChannelType, first_latency: bool, - is_tcp: bool, + protocol: ConnectProtocol, packet_loss_rate: Option, packet_delay: u32, use_ipv6: bool, @@ -45,9 +45,9 @@ impl ChannelContext { let inner = ContextInner { main_udp_socket, sub_udp_socket: RwLock::new(Vec::new()), - tcp_map: RwLock::new(FnvHashMap::default()), + packet_map: RwLock::new(FnvHashMap::default()), route_table: RouteTable::new(use_channel_type, first_latency, channel_num), - is_tcp, + protocol, packet_loss_rate, packet_delay, main_index: AtomicUsize::new(0), @@ -77,11 +77,11 @@ pub struct ContextInner { // 对称网络增加的udp socket sub_udp_socket: RwLock>, // tcp数据发送器 - pub(crate) tcp_map: RwLock>, + pub(crate) packet_map: RwLock>, // 路由信息 pub route_table: RouteTable, - // 是否使用tcp连接服务器 - is_tcp: bool, + // 使用什么协议连接服务器 + protocol: ConnectProtocol, //控制丢包率，取值v=[0,100_0000] 丢包率r=v/100_0000 packet_loss_rate: u32, //控制延迟 @@ -98,11 +98,11 @@ impl ContextInner { pub fn is_cone(&self) -> bool { self.sub_udp_socket.read().is_empty() } - pub fn is_main_tcp(&self) -> bool { - self.is_tcp + pub fn main_protocol(&self) -> ConnectProtocol { + self.protocol } pub fn is_udp_main(&self, route_key: &RouteKey) -> bool { - !route_key.is_tcp() && route_key.index < self.main_udp_socket.len() + route_key.protocol().is_udp() && route_key.index < self.main_udp_socket.len() } pub fn first_latency(&self) -> bool { self.route_table.first_latency @@ -157,7 +157,7 @@ impl ContextInner { Ok(ports) } pub fn send_tcp(&self, buf: &[u8], addr: SocketAddr) -> io::Result<()> { - if let Some(tcp) = self.tcp_map.read().get(&addr) { + if let Some(tcp) = self.packet_map.read().get(&addr) { tcp.try_send(buf) } else { Err(io::Error::from(io::ErrorKind::NotFound)) @@ -180,11 +180,10 @@ impl ContextInner { } /// 将数据发送到默认通道，一般发往服务器才用此方法 pub fn send_default(&self, buf: &[u8], addr: SocketAddr) -> io::Result<()> { - if self.is_tcp { - //服务端地址只在重连时检测变化 - self.send_tcp(buf, addr) - } else { + if self.protocol.is_udp() { self.send_main_udp(self.main_index.load(Ordering::Relaxed), buf, addr) + } else { + self.send_tcp(buf, addr) } } @@ -259,23 +258,26 @@ impl ContextInner { } /// 将数据发到指定路由 pub fn send_by_key(&self, buf: &[u8], route_key: RouteKey) -> io::Result<()> { - if route_key.is_tcp { - self.send_tcp(buf, route_key.addr) - } else { - if let Some(main_udp) = self.main_udp_socket.get(route_key.index) { - main_udp.send_to(buf, route_key.addr)?; - } else { - if let Some(udp) = self - .sub_udp_socket - .read() - .get(route_key.index - self.main_udp_socket.len()) - { - udp.send_to(buf, route_key.addr)?; + match route_key.protocol() { + ConnectProtocol::UDP => { + if let Some(main_udp) = self.main_udp_socket.get(route_key.index) { + main_udp.send_to(buf, route_key.addr)?; } else { - Err(io::Error::from(io::ErrorKind::NotFound))? + if let Some(udp) = self + .sub_udp_socket + .read() + .get(route_key.index - self.main_udp_socket.len()) + { + udp.send_to(buf, route_key.addr)?; + } else { + Err(io::Error::from(io::ErrorKind::NotFound))? + } } + Ok(()) + } + ConnectProtocol::TCP | ConnectProtocol::WS | ConnectProtocol::WSS => { + self.send_tcp(buf, route_key.addr) } - Ok(()) } } pub fn remove_route(&self, ip: &Ipv4Addr, route_key: RouteKey) { diff --git a/vnt/src/channel/handler.rs b/vnt/src/channel/handler.rs index 4715fb9..b5e4b41 100644 --- a/vnt/src/channel/handler.rs +++ b/vnt/src/channel/handler.rs @@ -3,7 +3,7 @@ use crate::channel::RouteKey; pub trait RecvChannelHandler: Clone + Send + 'static { fn handle( - &mut self, + &self, buf: &mut [u8], extend: &mut [u8], route_key: RouteKey, diff --git a/vnt/src/channel/mod.rs b/vnt/src/channel/mod.rs index e8bb276..c365565 100644 --- a/vnt/src/channel/mod.rs +++ b/vnt/src/channel/mod.rs @@ -1,12 +1,15 @@ use anyhow::Context; use std::net::{SocketAddr, UdpSocket}; use std::str::FromStr; +use tokio::sync::mpsc::channel; use crate::channel::context::ChannelContext; use crate::channel::handler::RecvChannelHandler; -use crate::channel::sender::AcceptSocketSender; +use crate::channel::sender::{AcceptSocketSender, ConnectUtil}; use crate::channel::tcp_channel::tcp_listen; use crate::channel::udp_channel::udp_listen; +#[cfg(feature = "websocket")] +use crate::channel::ws_channel::ws_connect_accept; use crate::util::StopManager; pub mod context; @@ -17,14 +20,21 @@ pub mod punch; pub mod sender; pub mod tcp_channel; pub mod udp_channel; +#[cfg(feature = "websocket")] +pub mod ws_channel; + +pub const BUFFER_SIZE: usize = 1024 * 64; +// 这里留个坑，tcp是支持_TCP_MAX_PACKET_SIZE长度的， +// 但是缓存只用BUFFER_SIZE，会导致多余的数据接收不了 +const TCP_MAX_PACKET_SIZE: usize = (1 << 24) - 1; -pub const BUFFER_SIZE: usize = 1024 * 16; #[derive(Debug, Copy, Clone, Eq, PartialEq)] pub enum UseChannelType { Relay, P2p, All, } + impl UseChannelType { pub fn is_only_relay(&self) -> bool { self == &UseChannelType::Relay @@ -36,6 +46,7 @@ impl UseChannelType { self == &UseChannelType::All } } + impl FromStr for UseChannelType { type Err = String; @@ -48,15 +59,49 @@ impl FromStr for UseChannelType { } } } + impl Default for UseChannelType { fn default() -> Self { UseChannelType::All } } +#[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)] +pub enum ConnectProtocol { + UDP, + TCP, + WS, + WSS, +} + +impl ConnectProtocol { + #[inline] + pub fn is_tcp(&self) -> bool { + self == &ConnectProtocol::TCP + } + #[inline] + pub fn is_udp(&self) -> bool { + self == &ConnectProtocol::UDP + } + #[inline] + pub fn is_ws(&self) -> bool { + self == &ConnectProtocol::WS + } + #[inline] + pub fn is_wss(&self) -> bool { + self == &ConnectProtocol::WSS + } + pub fn is_transport(&self) -> bool { + self.is_tcp() || self.is_udp() + } + pub fn is_base_tcp(&self) -> bool { + self.is_tcp() || self.is_ws() || self.is_wss() + } +} + #[derive(Copy, Clone, Debug)] pub struct Route { - pub is_tcp: bool, + pub protocol: ConnectProtocol, index: usize, pub addr: SocketAddr, pub metric: u8, @@ -68,11 +113,19 @@ pub struct RouteSortKey { pub metric: u8, pub rt: i64, } + const DEFAULT_RT: i64 = 9999; + impl Route { - pub fn new(is_tcp: bool, index: usize, addr: SocketAddr, metric: u8, rt: i64) -> Self { + pub fn new( + protocol: ConnectProtocol, + index: usize, + addr: SocketAddr, + metric: u8, + rt: i64, + ) -> Self { Self { - is_tcp, + protocol, index, addr, metric, @@ -81,7 +134,7 @@ impl Route { } pub fn from(route_key: RouteKey, metric: u8, rt: i64) -> Self { Self { - is_tcp: route_key.is_tcp, + protocol: route_key.protocol, index: route_key.index, addr: route_key.addr, metric, @@ -90,7 +143,7 @@ impl Route { } pub fn from_default_rt(route_key: RouteKey, metric: u8) -> Self { Self { - is_tcp: route_key.is_tcp, + protocol: route_key.protocol, index: route_key.index, addr: route_key.addr, metric, @@ -99,7 +152,7 @@ impl Route { } pub fn route_key(&self) -> RouteKey { RouteKey { - is_tcp: self.is_tcp, + protocol: self.protocol, index: self.index, addr: self.addr, } @@ -117,22 +170,24 @@ impl Route { #[derive(Copy, Clone, Ord, PartialOrd, Eq, PartialEq, Hash, Debug)] pub struct RouteKey { - is_tcp: bool, + protocol: ConnectProtocol, index: usize, pub addr: SocketAddr, } impl RouteKey { - pub(crate) fn new(is_tcp: bool, index: usize, addr: SocketAddr) -> Self { + pub(crate) fn new(protocol: ConnectProtocol, index: usize, addr: SocketAddr) -> Self { Self { - is_tcp, + protocol, index, addr, } } - pub fn is_tcp(&self) -> bool { - self.is_tcp + #[inline] + pub fn protocol(&self) -> ConnectProtocol { + self.protocol } + #[inline] pub fn index(&self) -> usize { self.index } @@ -142,10 +197,10 @@ pub(crate) fn init_context( ports: Vec, use_channel_type: UseChannelType, first_latency: bool, - is_tcp: bool, + protocol: ConnectProtocol, packet_loss_rate: Option, packet_delay: u32, -) -> anyhow::Result<(ChannelContext, mio::net::TcpListener)> { +) -> anyhow::Result<(ChannelContext, std::net::TcpListener)> { assert!(!ports.is_empty(), "not channel"); let mut udps = Vec::with_capacity(ports.len()); //检查系统是否支持ipv6 @@ -188,7 +243,7 @@ pub(crate) fn init_context( udps, use_channel_type, first_latency, - is_tcp, + protocol, packet_loss_rate, packet_delay, use_ipv6, @@ -229,32 +284,37 @@ pub(crate) fn init_context( socket.listen(128)?; socket.set_nonblocking(true)?; socket.set_nodelay(false)?; - let tcp_listener = mio::net::TcpListener::from_std(socket.into()); - Ok((context, tcp_listener)) + Ok((context, socket.into())) } pub(crate) fn init_channel( - tcp_listener: mio::net::TcpListener, + tcp_listener: std::net::TcpListener, context: ChannelContext, stop_manager: StopManager, recv_handler: H, ) -> anyhow::Result<( AcceptSocketSender>>, - AcceptSocketSender<(mio::net::TcpStream, SocketAddr, Option>)>, + ConnectUtil, )> where H: RecvChannelHandler, { + let (tcp_connect_s, tcp_connect_r) = channel(16); + let (ws_connect_s, _ws_connect_r) = channel(16); + let connect_util = ConnectUtil::new(tcp_connect_s, ws_connect_s); // udp监听，udp_socket_sender 用于NAT类型切换 let udp_socket_sender = udp_listen(stop_manager.clone(), recv_handler.clone(), context.clone())?; // 建立tcp监听，tcp_socket_sender 用于tcp 直连 - let tcp_socket_sender = tcp_listen( + tcp_listen( tcp_listener, - stop_manager.clone(), + tcp_connect_r, recv_handler.clone(), context.clone(), + stop_manager.clone(), )?; + #[cfg(feature = "websocket")] + ws_connect_accept(_ws_connect_r, recv_handler, context.clone(), stop_manager)?; - Ok((udp_socket_sender, tcp_socket_sender)) + Ok((udp_socket_sender, connect_util)) } diff --git a/vnt/src/channel/punch.rs b/vnt/src/channel/punch.rs index 225e5dd..5b2a760 100644 --- a/vnt/src/channel/punch.rs +++ b/vnt/src/channel/punch.rs @@ -7,12 +7,11 @@ use std::sync::Arc; use std::time::Duration; use std::{io, thread}; -use mio::net::TcpStream; use rand::prelude::SliceRandom; use rand::Rng; use crate::channel::context::ChannelContext; -use crate::channel::sender::AcceptSocketSender; +use crate::channel::sender::ConnectUtil; use crate::external_route::ExternalRoute; use crate::handle::CurrentDeviceInfo; use crate::nat::NatTest; @@ -51,7 +50,7 @@ pub struct NatInfo { pub nat_type: NatType, pub(crate) local_ipv4: Option, pub(crate) ipv6: Option, - pub(crate) udp_ports: Vec, + pub udp_ports: Vec, pub tcp_port: u16, } @@ -189,7 +188,7 @@ pub struct Punch { port_index: HashMap, punch_model: PunchModel, is_tcp: bool, - tcp_socket_sender: AcceptSocketSender<(TcpStream, SocketAddr, Option>)>, + connect_util: ConnectUtil, external_route: ExternalRoute, nat_test: NatTest, current_device: Arc>, @@ -200,7 +199,7 @@ impl Punch { context: ChannelContext, punch_model: PunchModel, is_tcp: bool, - tcp_socket_sender: AcceptSocketSender<(TcpStream, SocketAddr, Option>)>, + connect_util: ConnectUtil, external_route: ExternalRoute, nat_test: NatTest, current_device: Arc>, @@ -215,7 +214,7 @@ impl Punch { port_index: HashMap::new(), punch_model, is_tcp, - tcp_socket_sender, + connect_util, external_route, nat_test, current_device, @@ -224,26 +223,11 @@ impl Punch { } impl Punch { - fn connect_tcp(&self, buf: &[u8], addr: SocketAddr) -> bool { + fn connect_tcp(&self, buf: &[u8], addr: SocketAddr) { if self.nat_test.is_local_address(true, addr) { - return false; + return; } - // mio是非阻塞的，不能立马判断是否能连接成功，所以用标准库的tcp - match std::net::TcpStream::connect_timeout(&addr, Duration::from_millis(100)) { - Ok(tcp_stream) => { - if tcp_stream.set_nonblocking(true).is_err() { - return false; - } - return self - .tcp_socket_sender - .try_add_socket((TcpStream::from_std(tcp_stream), addr, Some(buf.to_vec()))) - .is_ok(); - } - Err(e) => { - log::warn!("连接到tcp失败,addr={},err={}", addr, e); - } - } - false + self.connect_util.try_connect_tcp(buf.to_vec(), addr); } pub fn punch( &mut self, @@ -274,22 +258,16 @@ impl Punch { if punch_tcp && self.is_tcp && nat_info.tcp_port != 0 { //向tcp发起连接 if let Some(ipv6_addr) = nat_info.local_tcp_ipv6addr() { - if self.connect_tcp(buf, ipv6_addr) { - // return Ok(()); - } + self.connect_tcp(buf, ipv6_addr) } //向tcp发起连接 if let Some(ipv4_addr) = nat_info.local_tcp_ipv4addr() { - if self.connect_tcp(buf, ipv4_addr) { - // return Ok(()); - } + self.connect_tcp(buf, ipv4_addr) } if nat_info.nat_type == NatType::Cone && nat_info.public_ips.len() == 1 { let addr = SocketAddr::V4(SocketAddrV4::new(nat_info.public_ips[0], nat_info.tcp_port)); - if self.connect_tcp(buf, addr) { - // return Ok(()); - } + self.connect_tcp(buf, addr) } } let channel_num = self.context.channel_num(); diff --git a/vnt/src/channel/sender.rs b/vnt/src/channel/sender.rs index 9574083..e5fbe39 100644 --- a/vnt/src/channel/sender.rs +++ b/vnt/src/channel/sender.rs @@ -1,13 +1,13 @@ use std::io; -use std::net::Ipv4Addr; +use std::net::{Ipv4Addr, SocketAddr}; use std::sync::mpsc::{SyncSender, TrySendError}; use std::sync::Arc; use crossbeam_utils::atomic::AtomicCell; -use mio::Token; +use tokio::sync::mpsc::Sender; use crate::channel::context::ChannelContext; -use crate::channel::notify::{AcceptNotify, WritableNotify}; +use crate::channel::notify::AcceptNotify; use crate::cipher::Cipher; use crate::compression::Compressor; use crate::external_route::ExternalRoute; @@ -133,58 +133,54 @@ impl AcceptSocketSender { } } } - #[derive(Clone)] pub struct PacketSender { - inner: Arc, + sender: Sender>, } impl PacketSender { - pub fn new(notify: WritableNotify, buffer: SyncSender>, token: Token) -> Self { - Self { - inner: Arc::new(PacketSenderInner { - token, - notify, - buffer, - }), - } + pub fn new(sender: Sender>) -> Self { + Self { sender } } - #[inline] pub fn try_send(&self, buf: &[u8]) -> io::Result<()> { - self.inner.try_send(buf) - } - pub fn shutdown(&self) -> io::Result<()> { - self.inner.shutdown() - } -} - -pub struct PacketSenderInner { - token: Token, - notify: WritableNotify, - buffer: SyncSender>, -} - -impl PacketSenderInner { - #[inline] - fn try_send(&self, buf: &[u8]) -> io::Result<()> { - let len = buf.len(); - let mut buf_vec = Vec::with_capacity(buf.len() + 4); - buf_vec.extend_from_slice(&[ - (len >> 24) as u8, - (len >> 16) as u8, - (len >> 8) as u8, - len as u8, - ]); - buf_vec.extend_from_slice(buf); - match self.buffer.try_send(buf_vec) { - Ok(_) => self.notify.notify(self.token, true), - Err(e) => match e { - TrySendError::Disconnected(_) => Err(io::Error::from(io::ErrorKind::WriteZero)), - TrySendError::Full(_) => Err(io::Error::from(io::ErrorKind::WouldBlock)), - }, + match self.sender.try_send(buf.to_vec()) { + Ok(_) => Ok(()), + Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => Err(io::Error::new( + io::ErrorKind::WouldBlock, + "通道已满，发生丢包", + )), + Err(_) => Err(io::Error::new( + io::ErrorKind::ConnectionRefused, + "通道关闭，发生丢包", + )), } } - fn shutdown(&self) -> io::Result<()> { - self.notify.notify(self.token, false) +} + +#[derive(Clone)] +pub struct ConnectUtil { + connect_tcp: Sender<(Vec, SocketAddr)>, + connect_ws: Sender<(Vec, String)>, +} + +impl ConnectUtil { + pub fn new( + connect_tcp: Sender<(Vec, SocketAddr)>, + connect_ws: Sender<(Vec, String)>, + ) -> Self { + Self { + connect_tcp, + connect_ws, + } + } + pub fn try_connect_tcp(&self, buf: Vec, addr: SocketAddr) { + if self.connect_tcp.try_send((buf, addr)).is_err() { + log::warn!("try_connect_tcp failed {}", addr); + } + } + pub fn try_connect_ws(&self, buf: Vec, addr: String) { + if self.connect_ws.try_send((buf, addr)).is_err() { + log::warn!("try_connect_ws failed"); + } } } diff --git a/vnt/src/channel/tcp_channel.rs b/vnt/src/channel/tcp_channel.rs index 097c71e..97debc9 100644 --- a/vnt/src/channel/tcp_channel.rs +++ b/vnt/src/channel/tcp_channel.rs @@ -1,472 +1,186 @@ -use std::collections::HashMap; -use std::io::{Read, Write}; -use std::net::{Shutdown, SocketAddr}; -#[cfg(any(unix))] -use std::os::fd::FromRawFd; -#[cfg(any(unix))] -use std::os::fd::IntoRawFd; -#[cfg(windows)] -use std::os::windows::io::FromRawSocket; -#[cfg(windows)] -use std::os::windows::io::IntoRawSocket; -use std::sync::mpsc::{sync_channel, Receiver, SyncSender, TryRecvError, TrySendError}; -use std::{io, thread}; - -use mio::net::{TcpListener, TcpStream}; -use mio::{Events, Interest, Poll, Registry, Token, Waker}; +use anyhow::{anyhow, Context}; +use std::net::SocketAddr; +use std::thread; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt}; +use tokio::net::tcp::OwnedReadHalf; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::mpsc::{channel, Receiver}; use crate::channel::context::ChannelContext; use crate::channel::handler::RecvChannelHandler; -use crate::channel::notify::{AcceptNotify, WritableNotify}; -use crate::channel::sender::{AcceptSocketSender, PacketSender}; -use crate::channel::{RouteKey, BUFFER_SIZE}; +use crate::channel::sender::PacketSender; +use crate::channel::{ConnectProtocol, RouteKey, BUFFER_SIZE, TCP_MAX_PACKET_SIZE}; use crate::util::StopManager; -const SERVER: Token = Token(0); -const NOTIFY: Token = Token(1); - /// 监听tcp端口，等待客户端连接 pub fn tcp_listen( - tcp_server: TcpListener, - stop_manager: StopManager, + tcp_server: std::net::TcpListener, + receiver: Receiver<(Vec, SocketAddr)>, recv_handler: H, context: ChannelContext, -) -> anyhow::Result>)>> + stop_manager: StopManager, +) -> anyhow::Result<()> where H: RecvChannelHandler, { - let (tcp_sender, tcp_receiver) = sync_channel(64); - let poll = Poll::new()?; - let waker = AcceptNotify::new(Waker::new(poll.registry(), NOTIFY)?); - let accept = AcceptSocketSender::new(waker.clone(), tcp_sender); - let worker = { - let waker = waker.clone(); - stop_manager.add_listener("tcp_listen".into(), move || { - if let Err(e) = waker.stop() { - log::error!("{:?}", e); - } - })? - }; - + let (stop_sender, stop_receiver) = tokio::sync::oneshot::channel::<()>(); + let worker = stop_manager.add_listener("tcpChannel".into(), move || { + let _ = stop_sender.send(()); + })?; + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .context("tcp tokio runtime build failed")?; thread::Builder::new() - .name("tcpRead".into()) + .name("tcpChannel".into()) .spawn(move || { - if let Err(e) = tcp_listen0( - poll, - tcp_server, - &stop_manager, - waker, - tcp_receiver, - recv_handler, - context, - ) { - log::error!("{:?}", e); - } + runtime.spawn(async move { + { + let recv_handler = recv_handler.clone(); + let context = context.clone(); + tokio::spawn(async move { + if let Err(e) = tcp_accept(tcp_server, recv_handler, context).await { + log::warn!("tcp_listen {:?}", e); + } + }); + } + tokio::spawn( + async move { connect_tcp_handle(receiver, recv_handler, context).await }, + ); + }); + runtime.block_on(async { + let _ = stop_receiver.await; + }); + runtime.shutdown_background(); worker.stop_all(); - })?; - Ok(accept) + }) + .context("tcp thread build failed")?; + Ok(()) } -fn tcp_listen0( - mut poll: Poll, - mut tcp_server: TcpListener, - stop_manager: &StopManager, - accept_notify: AcceptNotify, - accept_tcp_receiver: Receiver<(TcpStream, SocketAddr, Option>)>, - mut recv_handler: H, +async fn connect_tcp_handle( + mut receiver: Receiver<(Vec, SocketAddr)>, + recv_handler: H, + context: ChannelContext, +) where + H: RecvChannelHandler, +{ + while let Some((data, addr)) = receiver.recv().await { + let recv_handler = recv_handler.clone(); + let context = context.clone(); + tokio::spawn(async move { + if let Err(e) = connect_tcp0(data, addr, recv_handler, context).await { + log::warn!("发送失败,链接终止:{:?},{:?}", addr, e); + } + }); + } +} + +async fn connect_tcp0( + data: Vec, + addr: SocketAddr, + recv_handler: H, context: ChannelContext, ) -> anyhow::Result<()> where H: RecvChannelHandler, { - let (tcp_sender, tcp_receiver) = sync_channel(64); - let write_waker = init_writable_handler(tcp_receiver, stop_manager.clone(), context.clone())?; - poll.registry() - .register(&mut tcp_server, SERVER, Interest::READABLE)?; - let mut events = Events::with_capacity(1024); + let mut stream = + tokio::time::timeout(Duration::from_secs(3), TcpStream::connect(addr)).await??; + tcp_write(&mut stream, &data).await?; - let mut read_map: HashMap, usize)> = - HashMap::with_capacity(32); - let mut extend = [0; BUFFER_SIZE]; - loop { - if let Err(e) = poll.poll(&mut events, None) { - crate::ignore_io_interrupted(e)?; - continue; - } - for event in events.iter() { - match event.token() { - SERVER => loop { - match tcp_server.accept() { - Ok((stream, addr)) => { - accept_handle( - stream, - addr, - None, - &write_waker, - &mut read_map, - &tcp_sender, - poll.registry(), - )?; - } - Err(e) => { - if e.kind() == io::ErrorKind::WouldBlock { - break; - } - return Err(e)?; - } - } - }, - NOTIFY => { - if accept_notify.is_stop() { - return Ok(()); - } - if accept_notify.is_add_socket() { - while let Ok((stream, addr, init_buf)) = accept_tcp_receiver.try_recv() { - accept_handle( - stream, - addr, - init_buf, - &write_waker, - &mut read_map, - &tcp_sender, - poll.registry(), - )?; - } - } - } - token => { - if event.is_readable() { - if let Err(e) = readable_handle( - &token, - &mut read_map, - &mut recv_handler, - &context, - &mut extend, - ) { - closed_handle_r(&token, &mut read_map); - log::warn!("{:?}", e); - if let Err(e) = write_waker.notify(token, false) { - log::warn!("{:?}", e); - } - } - } else { - closed_handle_r(&token, &mut read_map); - if let Err(e) = write_waker.notify(token, false) { - log::warn!("{:?}", e); - } - } - } - } - } - } -} - -/// 处理写事件 - -fn init_writable_handler( - receiver: Receiver<(TcpStream, Token, SocketAddr, Option>)>, - stop_manager: StopManager, - context: ChannelContext, -) -> anyhow::Result { - let poll = Poll::new()?; - let writable_notify = WritableNotify::new(Waker::new(poll.registry(), NOTIFY)?); - let worker = { - let writable_notify = writable_notify.clone(); - stop_manager.add_listener("tcp_writable_handler".into(), move || { - if let Err(e) = writable_notify.stop() { - log::error!("{:?}", e); - } - })? - }; - { - let writable_notify = writable_notify.clone(); - thread::Builder::new() - .name("tcpWriteableListen".into()) - .spawn(move || { - if let Err(e) = tcp_writable_listen(receiver, poll, writable_notify, &context) { - log::error!("{:?}", e); - } - worker.stop_all(); - })?; - } - - Ok(writable_notify) -} - -/// 处理写事件 -fn tcp_writable_listen( - receiver: Receiver<(TcpStream, Token, SocketAddr, Option>)>, - mut poll: Poll, - writable_notify: WritableNotify, - context: &ChannelContext, -) -> io::Result<()> { - let mut events = Events::with_capacity(1024); - let mut write_map: HashMap< - Token, - ( - TcpStream, - SocketAddr, - Receiver>, - Option<(Vec, usize)>, - ), - > = HashMap::with_capacity(32); - loop { - if let Err(e) = poll.poll(&mut events, None) { - crate::ignore_io_interrupted(e)?; - continue; - } - for event in events.iter() { - match event.token() { - NOTIFY => { - if writable_notify.is_stop() { - //服务停止 - return Ok(()); - } - if writable_notify.is_need_write() { - // 需要写入数据 - if let Some(tokens) = writable_notify.take_all() { - for (token, state) in tokens { - if !state { - closed_handle_w(&token, &mut write_map, &context); - continue; - } - if let Err(e) = writable_handle(&token, &mut write_map) { - closed_handle_w(&token, &mut write_map, &context); - log::warn!("{:?}", e); - } - } - } - } - if writable_notify.is_add_socket() { - //添加tcp连接，并监听写事件 - while let Ok((mut stream, token, addr, init_buf)) = receiver.try_recv() { - if let Err(e) = stream.set_nodelay(true) { - log::warn!("set_nodelay err={:?}", e); - } - if let Err(e) = - poll.registry() - .register(&mut stream, token, Interest::WRITABLE) - { - log::warn!("registry err={:?}", e); - continue; - } - let (sender, receiver) = sync_channel(128); - let packet_sender = - PacketSender::new(writable_notify.clone(), sender, token); - if let Some(init_buf) = init_buf { - packet_sender.try_send(&init_buf)?; - } - - context.tcp_map.write().insert(addr, packet_sender); - write_map.insert(token, (stream, addr, receiver, None)); - } - } - } - token => { - if event.is_writable() { - if let Err(e) = writable_handle(&token, &mut write_map) { - closed_handle_w(&token, &mut write_map, &context); - log::warn!("{:?}", e); - } - } else { - closed_handle_w(&token, &mut write_map, &context); - } - } - } - } - } -} - -fn accept_handle( - stream: TcpStream, - addr: SocketAddr, - init_buf: Option>, - write_waker: &WritableNotify, - read_map: &mut HashMap, usize)>, - tcp_sender: &SyncSender<(TcpStream, Token, SocketAddr, Option>)>, - registry: &Registry, -) -> io::Result<()> { - #[cfg(windows)] - let (tcp_stream, index) = unsafe { - let fd = stream.into_raw_socket(); - (std::net::TcpStream::from_raw_socket(fd), fd as usize) - }; - #[cfg(any(unix))] - let (tcp_stream, index) = unsafe { - let fd = stream.into_raw_fd(); - (std::net::TcpStream::from_raw_fd(fd), fd as usize) - }; - if index == 0 || index == 1 { - log::error!("index err={:?}", addr); - return Ok(()); - } - let token = Token(index); - match tcp_stream.try_clone() { - Ok(tcp_writer) => { - match tcp_sender.try_send((TcpStream::from_std(tcp_writer), token, addr, init_buf)) { - Ok(_) => { - if let Err(e) = write_waker.add_socket() { - log::error!("write_waker,err={:?},addr={:?}", e, addr); - return Ok(()); - } - } - Err(e) => { - return match e { - TrySendError::Full(_) => { - log::error!("Full,addr={:?}", addr); - Ok(()) - } - TrySendError::Disconnected(_) => { - Err(io::Error::new(io::ErrorKind::Other, "write thread exit")) - } - }; - } - } - } - Err(e) => { - log::error!("try_clone err={:?},addr={:?}", e, addr); - return Ok(()); - } - } - let mut stream = TcpStream::from_std(tcp_stream); - if let Err(e) = registry.register(&mut stream, token, Interest::READABLE) { - log::error!("registry err={:?},addr={:?}", e, addr); - return Ok(()); - } - read_map.insert( - token, - ( - RouteKey::new(true, index, addr), - stream, - Box::new([0; BUFFER_SIZE]), - 0, - ), - ); + tcp_stream_handle(stream, addr, recv_handler, context).await; Ok(()) } -fn readable_handle( - token: &Token, - map: &mut HashMap, usize)>, - recv_handler: &mut H, - context: &ChannelContext, - extend: &mut [u8], -) -> io::Result<()> +async fn tcp_accept( + tcp_server: std::net::TcpListener, + recv_handler: H, + context: ChannelContext, +) -> anyhow::Result<()> where H: RecvChannelHandler, { - if let Some((route_key, stream, buf, begin)) = map.get_mut(token) { - loop { - let end = if *begin >= 4 { - let len = ((buf[0] as usize) << 24) - | ((buf[1] as usize) << 16) - | ((buf[2] as usize) << 8) - | buf[3] as usize; - 4 + len - } else { - 4 - }; - if end > BUFFER_SIZE { - return Err(io::Error::from(io::ErrorKind::InvalidData)); - } - match stream.read(&mut buf[*begin..end]) { - Ok(len) => { - if len == 0 { - return Err(io::Error::from(io::ErrorKind::UnexpectedEof)); - } - *begin += len; - if end > 4 && *begin == end { - recv_handler.handle(&mut buf[4..end], extend, *route_key, context); - *begin = 0; - } - } - Err(e) => { - if e.kind() == io::ErrorKind::WouldBlock { - break; - } - return Err(e); - } + let tcp_server = TcpListener::from_std(tcp_server)?; + + loop { + let (stream, addr) = tcp_server.accept().await?; + tcp_stream_handle(stream, addr, recv_handler.clone(), context.clone()).await; + } +} + +pub async fn tcp_stream_handle( + stream: TcpStream, + addr: SocketAddr, + recv_handler: H, + context: ChannelContext, +) where + H: RecvChannelHandler, +{ + let _ = stream.set_nodelay(true); + let (r, mut w) = stream.into_split(); + let (sender, mut receiver) = channel::>(100); + context + .packet_map + .write() + .insert(addr, PacketSender::new(sender)); + tokio::spawn(async move { + while let Some(data) = receiver.recv().await { + if let Err(e) = tcp_write(&mut w, &data).await { + log::info!("发送失败,tcp链接终止:{:?},{:?}", addr, e); + break; } } + let _ = w.shutdown().await; + }); + tokio::spawn(async move { + if let Err(e) = tcp_read(r, addr, &context, recv_handler).await { + log::warn!("tcp_read {:?}", e) + } + context.packet_map.write().remove(&addr); + }); +} + +async fn tcp_write(w: &mut W, buf: &[u8]) -> anyhow::Result<()> { + let len = buf.len(); + if len > TCP_MAX_PACKET_SIZE { + return Err(anyhow!("超过了tcp的最大长度传输")); } + w.write_all(&[0, (len >> 16) as u8, (len >> 8) as u8, len as u8]) + .await?; + w.write_all(&buf).await?; Ok(()) } -fn writable_handle( - token: &Token, - map: &mut HashMap< - Token, - ( - TcpStream, - SocketAddr, - Receiver>, - Option<(Vec, usize)>, - ), - >, -) -> io::Result<()> { - if let Some((stream, _, receiver, last)) = map.get_mut(token) { - loop { - if let Some((buf, begin)) = last { - match stream.write(&buf[*begin..]) { - Ok(len) => { - if len == 0 { - return Err(io::Error::from(io::ErrorKind::WriteZero)); - } - if len + *begin == buf.len() { - *last = None; - } else { - *begin += len; - continue; - } - } - Err(e) => { - if e.kind() == io::ErrorKind::WouldBlock { - break; - } - return Err(e); - } - } - } - match receiver.try_recv() { - Ok(buf) => *last = Some((buf, 0)), - Err(e) => match e { - TryRecvError::Empty => { - break; - } - TryRecvError::Disconnected => { - return Err(io::Error::from(io::ErrorKind::Other)); - } - }, - } - } - } - Ok(()) -} - -fn closed_handle_r( - token: &Token, - map: &mut HashMap, usize)>, -) { - if let Some((_, tcp, _, _)) = map.remove(token) { - let _ = tcp.shutdown(Shutdown::Both); - } -} - -fn closed_handle_w( - token: &Token, - map: &mut HashMap< - Token, - ( - TcpStream, - SocketAddr, - Receiver>, - Option<(Vec, usize)>, - ), - >, +async fn tcp_read( + mut read: OwnedReadHalf, + addr: SocketAddr, context: &ChannelContext, -) { - if let Some((tcp, addr, _, _)) = map.remove(token) { - context.tcp_map.write().remove(&addr); - let _ = tcp.shutdown(Shutdown::Both); + recv_handler: H, +) -> anyhow::Result<()> +where + H: RecvChannelHandler, +{ + let mut head = [0; 4]; + let mut buf = [0; BUFFER_SIZE]; + let mut extend = [0; BUFFER_SIZE]; + loop { + read.read_exact(&mut head).await?; + if head[0] != 0 { + return Err(anyhow!("tcp数据流错误 {}", addr)); + } + let len = ((head[1] as usize) << 16) | ((head[2] as usize) << 8) | head[3] as usize; + if len < 12 || len > buf.len() { + return Err(anyhow!("tcp数据长度无效 {}", addr)); + } + read.read_exact(&mut buf[..len]).await?; + recv_handler.handle( + &mut buf[..len], + &mut extend, + RouteKey::new(ConnectProtocol::TCP, 0, addr), + context, + ); } } diff --git a/vnt/src/channel/udp_channel.rs b/vnt/src/channel/udp_channel.rs index b7f5a46..a895e6f 100644 --- a/vnt/src/channel/udp_channel.rs +++ b/vnt/src/channel/udp_channel.rs @@ -10,7 +10,7 @@ use crate::channel::context::ChannelContext; use crate::channel::handler::RecvChannelHandler; use crate::channel::notify::AcceptNotify; use crate::channel::sender::AcceptSocketSender; -use crate::channel::{RouteKey, BUFFER_SIZE}; +use crate::channel::{ConnectProtocol, RouteKey, BUFFER_SIZE}; use crate::util::StopManager; pub fn udp_listen( @@ -60,7 +60,7 @@ where fn sub_udp_listen0( mut poll: Poll, - mut recv_handler: H, + recv_handler: H, context: ChannelContext, accept_notify: AcceptNotify, accept_receiver: Receiver>>, @@ -120,7 +120,7 @@ where recv_handler.handle( &mut buf[..len], &mut extend, - RouteKey::new(false, token.0, addr), + RouteKey::new(ConnectProtocol::UDP, token.0, addr), &context, ); } @@ -238,7 +238,7 @@ where pub fn main_udp_listen0( mut poll: Poll, - mut recv_handler: H, + recv_handler: H, context: ChannelContext, ) -> io::Result<()> where @@ -280,7 +280,7 @@ where recv_handler.handle( &mut buf[..len], &mut extend, - RouteKey::new(false, index, addr), + RouteKey::new(ConnectProtocol::UDP, index, addr), &context, ); } diff --git a/vnt/src/channel/ws_channel.rs b/vnt/src/channel/ws_channel.rs new file mode 100644 index 0000000..0e79dc3 --- /dev/null +++ b/vnt/src/channel/ws_channel.rs @@ -0,0 +1,130 @@ +use crate::channel::{ConnectProtocol, RouteKey, BUFFER_SIZE}; +use anyhow::Context; +use futures_util::stream::SplitStream; +use futures_util::{SinkExt, StreamExt}; +use std::convert::Into; +use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4}; +use std::thread; +use std::time::Duration; +use tokio::net::TcpStream; +use tokio::sync::mpsc::{channel, Receiver}; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream}; + +use crate::channel::context::ChannelContext; +use crate::channel::handler::RecvChannelHandler; +use crate::channel::sender::PacketSender; +use crate::util::StopManager; + +/// ws协议， +/// 暂时只允许用ws连服务端，不能用ws打洞/连客户端 +pub fn ws_connect_accept( + receiver: Receiver<(Vec, String)>, + recv_handler: H, + context: ChannelContext, + stop_manager: StopManager, +) -> anyhow::Result<()> +where + H: RecvChannelHandler, +{ + let (stop_sender, stop_receiver) = tokio::sync::oneshot::channel::<()>(); + let worker = stop_manager.add_listener("wsChannel".into(), move || { + let _ = stop_sender.send(()); + })?; + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .context("ws tokio runtime build failed")?; + thread::Builder::new() + .name("wsChannel".into()) + .spawn(move || { + runtime.spawn(async move { connect_ws_handle(receiver, recv_handler, context).await }); + runtime.block_on(async { + let _ = stop_receiver.await; + }); + runtime.shutdown_background(); + worker.stop_all(); + }) + .context("ws thread build failed")?; + Ok(()) +} + +async fn connect_ws_handle( + mut receiver: Receiver<(Vec, String)>, + recv_handler: H, + context: ChannelContext, +) where + H: RecvChannelHandler, +{ + while let Some((data, url)) = receiver.recv().await { + let recv_handler = recv_handler.clone(); + let context = context.clone(); + tokio::spawn(async move { + if let Err(e) = connect_ws(data, url, recv_handler, context).await { + log::warn!("发送失败,ws链接终止:{:?}", e); + } + }); + } +} +const WS_ADDR: SocketAddr = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0)); + +async fn connect_ws( + data: Vec, + url: String, + recv_handler: H, + context: ChannelContext, +) -> anyhow::Result<()> +where + H: RecvChannelHandler, +{ + println!("ws协议 {}", url); + let (mut ws, response) = + tokio::time::timeout(Duration::from_secs(3), connect_async(url)).await??; + println!("ws协议 {:?}", response); + ws.send(Message::Binary(data)).await?; + let (mut ws_write, ws_read) = ws.split(); + let (sender, mut receiver) = channel::>(100); + context + .packet_map + .write() + .insert(WS_ADDR, PacketSender::new(sender)); + tokio::spawn(async move { + while let Some(data) = receiver.recv().await { + if let Err(e) = ws_write.send(Message::Binary(data)).await { + log::warn!("websocket err {:?}", e); + break; + } + } + let _ = ws_write.close().await; + }); + if let Err(e) = ws_read_handle(ws_read, recv_handler, &context).await { + log::warn!("{:?}", e); + } + context.packet_map.write().remove(&WS_ADDR); + Ok(()) +} +async fn ws_read_handle( + mut ws_read: SplitStream>>, + recv_handler: H, + context: &ChannelContext, +) -> anyhow::Result<()> +where + H: RecvChannelHandler, +{ + let mut extend = [0; BUFFER_SIZE]; + let route_key = RouteKey::new(ConnectProtocol::WS, 0, WS_ADDR); + while let Some(msg) = ws_read.next().await { + let msg = msg.context("Error during WebSocket ")?; + match msg { + Message::Text(txt) => log::info!("Received text message: {}", txt), + Message::Binary(mut data) => { + recv_handler.handle(&mut data, &mut extend, route_key, context); + } + Message::Ping(_) | Message::Pong(_) => (), + Message::Close(_) => break, + _ => {} + } + } + Ok(()) +} diff --git a/vnt/src/core/conn.rs b/vnt/src/core/conn.rs index b63c55f..97b03fa 100644 --- a/vnt/src/core/conn.rs +++ b/vnt/src/core/conn.rs @@ -164,7 +164,7 @@ impl VntInner { ports, config.use_channel_type, config.first_latency, - config.tcp, + config.protocol, config.packet_loss_rate, config.packet_delay, )?; @@ -253,14 +253,14 @@ impl VntInner { ); //初始化网络数据通道 - let (udp_socket_sender, tcp_socket_sender) = + let (udp_socket_sender, connect_util) = init_channel(tcp_listener, context.clone(), stop_manager.clone(), handler)?; // 打洞逻辑 let punch = Punch::new( context.clone(), config.punch_model, - config.tcp, - tcp_socket_sender.clone(), + config.protocol.is_base_tcp(), + connect_util.clone(), external_route.clone(), nat_test.clone(), current_device.clone(), @@ -274,7 +274,7 @@ impl VntInner { context.clone(), current_device.clone(), config_info.clone(), - tcp_socket_sender.clone(), + connect_util.clone(), callback.clone(), 0, handshake, diff --git a/vnt/src/core/mod.rs b/vnt/src/core/mod.rs index f674cd5..74720d2 100644 --- a/vnt/src/core/mod.rs +++ b/vnt/src/core/mod.rs @@ -5,7 +5,7 @@ use std::str::FromStr; pub use conn::Vnt; use crate::channel::punch::PunchModel; -use crate::channel::UseChannelType; +use crate::channel::{ConnectProtocol, UseChannelType}; use crate::cipher::CipherModel; use crate::compression::Compressor; use crate::util::{address_choose, dns_query_all}; @@ -28,7 +28,7 @@ pub struct Config { pub out_ips: Vec<(u32, u32)>, pub password: Option, pub mtu: Option, - pub tcp: bool, + pub protocol: ConnectProtocol, pub ip: Option, #[cfg(feature = "ip_proxy")] #[cfg(feature = "integrated_tun")] @@ -67,7 +67,6 @@ impl Config { out_ips: Vec<(u32, u32)>, password: Option, mtu: Option, - tcp: bool, ip: Option, #[cfg(feature = "integrated_tun")] #[cfg(feature = "ip_proxy")] @@ -109,8 +108,32 @@ impl Config { if name.is_empty() || name.len() > 128 { return Err(anyhow!("name too long")); } - let server_address = - address_choose(dns_query_all(&server_address_str, name_servers.clone())?)?; + let mut server_address_str = server_address_str.to_lowercase(); + let mut _query_dns = true; + let mut protocol = ConnectProtocol::UDP; + #[cfg(feature = "websocket")] + { + if server_address_str.starts_with("ws://") { + protocol = ConnectProtocol::WS; + _query_dns = false; + } + if server_address_str.starts_with("wss://") { + protocol = ConnectProtocol::WSS; + _query_dns = false; + } + } + + let mut server_address = "0.0.0.0:0".parse().unwrap(); + if _query_dns { + if let Some(s) = server_address_str.strip_prefix("udp://") { + server_address_str = s.to_string(); + } else if let Some(s) = server_address_str.strip_prefix("tcp://") { + server_address_str = s.to_string(); + protocol = ConnectProtocol::TCP; + } + server_address = + address_choose(dns_query_all(&server_address_str, name_servers.clone())?)?; + } #[cfg(feature = "port_mapping")] let port_mapping_list = crate::port_mapping::convert(port_mapping_list)?; @@ -133,7 +156,7 @@ impl Config { out_ips, password, mtu, - tcp, + protocol, ip, #[cfg(feature = "ip_proxy")] #[cfg(feature = "integrated_tun")] diff --git a/vnt/src/handle/maintain/addr_request.rs b/vnt/src/handle/maintain/addr_request.rs index 638ba36..1030e5c 100644 --- a/vnt/src/handle/maintain/addr_request.rs +++ b/vnt/src/handle/maintain/addr_request.rs @@ -92,7 +92,7 @@ fn addr_request0( return Ok(()); } - if current_dev.connect_server.is_ipv4() && !context.is_main_tcp() { + if current_dev.connect_server.is_ipv4() && !context.main_protocol().is_base_tcp() { // 如果连接的是ipv4服务，则探测公网端口 let gateway_ip = current_dev.virtual_gateway; let src_ip = current_dev.virtual_ip; diff --git a/vnt/src/handle/maintain/idle.rs b/vnt/src/handle/maintain/idle.rs index 2589f78..d610018 100644 --- a/vnt/src/handle/maintain/idle.rs +++ b/vnt/src/handle/maintain/idle.rs @@ -1,14 +1,13 @@ use std::io; -use std::net::SocketAddr; use std::sync::Arc; use std::time::Duration; use crossbeam_utils::atomic::AtomicCell; -use mio::net::TcpStream; use crate::channel::context::ChannelContext; use crate::channel::idle::{Idle, IdleType}; -use crate::channel::sender::AcceptSocketSender; +use crate::channel::sender::ConnectUtil; +use crate::channel::ConnectProtocol; use crate::handle::callback::{ConnectInfo, ErrorType}; use crate::handle::handshaker::Handshake; use crate::handle::{BaseConfigInfo, ConnectStatus, CurrentDeviceInfo}; @@ -36,7 +35,7 @@ pub fn idle_gateway( context: ChannelContext, current_device_info: Arc>, config: BaseConfigInfo, - tcp_socket_sender: AcceptSocketSender<(TcpStream, SocketAddr, Option>)>, + connect_util: ConnectUtil, call: Call, mut connect_count: usize, handshake: Handshake, @@ -45,7 +44,7 @@ pub fn idle_gateway( &context, ¤t_device_info, &config, - &tcp_socket_sender, + &connect_util, &call, &mut connect_count, &handshake, @@ -56,7 +55,7 @@ pub fn idle_gateway( context, current_device_info, config, - tcp_socket_sender, + connect_util, call, connect_count, handshake, @@ -71,7 +70,7 @@ fn idle_gateway0( context: &ChannelContext, current_device: &AtomicCell, config: &BaseConfigInfo, - tcp_socket_sender: &AcceptSocketSender<(TcpStream, SocketAddr, Option>)>, + connect_util: &ConnectUtil, call: &Call, connect_count: &mut usize, handshake: &Handshake, @@ -80,7 +79,7 @@ fn idle_gateway0( context, current_device, config, - tcp_socket_sender, + connect_util, call, connect_count, handshake, @@ -120,7 +119,7 @@ fn check_gateway_channel( context: &ChannelContext, current_device_info: &AtomicCell, config: &BaseConfigInfo, - tcp_socket_sender: &AcceptSocketSender<(TcpStream, SocketAddr, Option>)>, + connect_util: &ConnectUtil, call: &Call, count: &mut usize, handshake: &Handshake, @@ -128,28 +127,29 @@ fn check_gateway_channel( let mut current_device = current_device_info.load(); if current_device.status.offline() { *count += 1; - // 探测服务器地址 - current_device = domain_request0(current_device_info, config); + let connect_protocol = context.main_protocol(); + if connect_protocol.is_transport() { + // 传输层的协议需要探测服务器地址 + current_device = domain_request0(current_device_info, config); + } //需要重连 call.connect(ConnectInfo::new(*count, current_device.connect_server)); log::info!("发送握手请求,{:?}", config); if let Err(e) = handshake.send(context, config.server_secret, current_device.connect_server) { log::warn!("{:?}", e); - if context.is_main_tcp() { - let request_packet = handshake.handshake_request_packet(config.server_secret)?; - //tcp需要重连 - let tcp_stream = std::net::TcpStream::connect_timeout( - ¤t_device.connect_server, - Duration::from_secs(5), - )?; - tcp_stream.set_nonblocking(true)?; - if let Err(e) = tcp_socket_sender.try_add_socket(( - TcpStream::from_std(tcp_stream), - current_device.connect_server, - Some(request_packet.into_buffer()), - )) { - log::warn!("{:?}", e) + let request_packet = handshake.handshake_request_packet(config.server_secret)?; + match connect_protocol { + ConnectProtocol::UDP => {} + ConnectProtocol::TCP => { + connect_util.try_connect_tcp( + request_packet.into_buffer(), + current_device.connect_server, + ); + } + ConnectProtocol::WS | ConnectProtocol::WSS => { + connect_util + .try_connect_ws(request_packet.into_buffer(), config.server_addr.clone()); } } } diff --git a/vnt/src/handle/recv_data/client.rs b/vnt/src/handle/recv_data/client.rs index 4e54610..5970867 100644 --- a/vnt/src/handle/recv_data/client.rs +++ b/vnt/src/handle/recv_data/client.rs @@ -238,7 +238,7 @@ impl ClientPacketHandler { //忽略掉来源于自己的包 if self .nat_test - .is_local_address(route_key.is_tcp(), route_key.addr) + .is_local_address(route_key.protocol().is_base_tcp(), route_key.addr) { return Ok(()); } @@ -261,7 +261,7 @@ impl ClientPacketHandler { } if self .nat_test - .is_local_address(route_key.is_tcp(), route_key.addr) + .is_local_address(route_key.protocol().is_base_tcp(), route_key.addr) { return Ok(()); } diff --git a/vnt/src/handle/recv_data/mod.rs b/vnt/src/handle/recv_data/mod.rs index 0883df4..a23b185 100644 --- a/vnt/src/handle/recv_data/mod.rs +++ b/vnt/src/handle/recv_data/mod.rs @@ -44,7 +44,7 @@ pub struct RecvDataHandler { impl RecvChannelHandler for RecvDataHandler { fn handle( - &mut self, + &self, buf: &mut [u8], extend: &mut [u8], route_key: RouteKey, @@ -54,7 +54,7 @@ impl RecvChannelHandler for RecvDataHand return; } //判断stun响应包 - if !route_key.is_tcp() { + if route_key.protocol().is_udp() { if let Ok(rs) = self .nat_test .recv_data(route_key.index(), route_key.addr, buf) @@ -135,7 +135,7 @@ impl RecvDataHandler { } } fn handle0( - &mut self, + &self, buf: &mut [u8], extend: &mut [u8], route_key: RouteKey, diff --git a/vnt/src/handle/recv_data/server.rs b/vnt/src/handle/recv_data/server.rs index c3a018e..c28cb15 100644 --- a/vnt/src/handle/recv_data/server.rs +++ b/vnt/src/handle/recv_data/server.rs @@ -24,9 +24,7 @@ use crate::handle::callback::{ErrorInfo, ErrorType, HandshakeInfo, RegisterInfo, use crate::handle::handshaker; use crate::handle::handshaker::Handshake; use crate::handle::recv_data::PacketHandler; -use crate::handle::{ - registrar, BaseConfigInfo, ConnectStatus, CurrentDeviceInfo, PeerDeviceInfo, GATEWAY_IP, -}; +use crate::handle::{registrar, BaseConfigInfo, ConnectStatus, CurrentDeviceInfo, PeerDeviceInfo}; use crate::nat::NatTest; use crate::proto::message::{DeviceList, HandshakeResponse, RegistrationResponse}; use crate::protocol::body::ENCRYPTION_RESERVED; diff --git a/vnt/src/util/counter/adder.rs b/vnt/src/util/counter/adder.rs index 0dc3701..f8268ea 100644 --- a/vnt/src/util/counter/adder.rs +++ b/vnt/src/util/counter/adder.rs @@ -90,7 +90,7 @@ impl U64Adder { index: 0, } } - pub fn add(&mut self, num: u64) { + pub fn add(&self, num: u64) { self.inner.base[self.index].add(num); } pub fn get(&self) -> u64 {