diff --git a/Cargo.lock b/Cargo.lock index d1b944a..cf82e75 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2195,6 +2195,7 @@ dependencies = [ "anyhow", "chrono", "common", + "console", "log", "rand", "signal-hook", diff --git a/common/src/callback.rs b/common/src/callback.rs index 4fffa03..dff2f1e 100644 --- a/common/src/callback.rs +++ b/common/src/callback.rs @@ -37,7 +37,8 @@ impl VntCallback for VntHandler { | ErrorType::AddressExhausted | ErrorType::IpAlreadyExists | ErrorType::InvalidIp - | ErrorType::LocalIpExists => { + | ErrorType::LocalIpExists + | ErrorType::FailedToCrateDevice => { self.stop(); } _ => {} diff --git a/common/src/cli.rs b/common/src/cli.rs index 4199606..ceee8d2 100644 --- a/common/src/cli.rs +++ b/common/src/cli.rs @@ -302,7 +302,7 @@ pub fn parse_args_config() -> anyhow::Result, bool)> } else { Compressor::None }; - let config = match Config::new( + let config = Config::new( #[cfg(feature = "integrated_tun")] #[cfg(target_os = "windows")] tap, @@ -337,13 +337,7 @@ pub fn parse_args_config() -> anyhow::Result, bool)> !disable_stats, allow_wire_guard, local_ipv4, - ) { - Ok(config) => config, - Err(e) => { - println!("config error: {}", e); - std::process::exit(1); - } - }; + )?; (config, vnt_mapping_list, cmd) }; println!("version {}", vnt::VNT_VERSION); diff --git a/common/src/config/file_config.rs b/common/src/config/file_config.rs index 97c72a5..847dc8c 100644 --- a/common/src/config/file_config.rs +++ b/common/src/config/file_config.rs @@ -104,8 +104,8 @@ pub fn read_config(file_path: &str) -> anyhow::Result<(Config, Vec, bool let file_conf = match serde_yaml::from_str::(&conf) { Ok(val) => val, Err(e) => { - log::error!("{:?}", e); - return Err(anyhow!("{}", e)); + log::error!("serde_yaml::from_str {:?}", e); + return Err(anyhow!("serde_yaml::from_str {:?}", e)); } }; if file_conf.token.is_empty() { diff --git a/vn-link-cli/src/main.rs b/vn-link-cli/src/main.rs index 85a7a3a..e374033 100644 --- a/vn-link-cli/src/main.rs +++ b/vn-link-cli/src/main.rs @@ -17,7 +17,7 @@ fn main() { e, std::env::args().collect::>() ); - println!("{}", e); + println!("Error {:?}", e); return; } }; diff --git a/vnt-cli/Cargo.toml b/vnt-cli/Cargo.toml index b9495dd..e2239af 100644 --- a/vnt-cli/Cargo.toml +++ b/vnt-cli/Cargo.toml @@ -10,7 +10,7 @@ vnt = { path = "../vnt", package = "vnt", default-features = false, features = [ common = { path = "../common", default-features = false, features = ["integrated_tun"] } log = "0.4.17" anyhow = "1.0.82" - +console = "0.15.2" [target.'cfg(any(target_os = "linux",target_os = "macos"))'.dependencies] sudo = "0.6.0" diff --git a/vnt-cli/README.md b/vnt-cli/README.md index 69da1df..d6b74ee 100644 --- a/vnt-cli/README.md +++ b/vnt-cli/README.md @@ -74,9 +74,11 @@ 设置虚拟网卡的mtu值,大多数情况下使用默认值效率会更高,也可根据实际情况微调这个值,不加密默认为1450,加密默认为1410 -### --tcp +### ~~--tcp~~ -和服务端使用tcp通信。有些网络提供商对UDP限制比较大,这个时候可以选择使用TCP模式,提高稳定性。一般来说udp延迟和消耗更低 +~~和服务端使用tcp通信。有些网络提供商对UDP限制比较大,这个时候可以选择使用TCP模式,提高稳定性。一般来说udp延迟和消耗更低~~ + +新版本使用 `-s tcp://`的形式使用tcp ### --ip `` diff --git a/vnt-cli/src/main.rs b/vnt-cli/src/main.rs index edad9a6..77809b9 100644 --- a/vnt-cli/src/main.rs +++ b/vnt-cli/src/main.rs @@ -1,4 +1,5 @@ use common::callback; +use console::style; use vnt::core::{Config, Vnt}; mod root_check; fn main() { @@ -16,7 +17,7 @@ fn main() { e, std::env::args().collect::>() ); - println!("{}", e); + println!("{}", style(format!("Error {:?}", e)).red()); return; } }; @@ -40,6 +41,7 @@ fn main0(config: Config, _show_cmd: bool) { let vnt_util = match Vnt::new(config, callback::VntHandler {}) { Ok(vnt) => vnt, Err(e) => { + log::error!("vnt create error {:?}", e); println!("error: {:?}", e); std::process::exit(1); } diff --git a/vnt/proto/message.proto b/vnt/proto/message.proto index 197c884..948a123 100644 --- a/vnt/proto/message.proto +++ b/vnt/proto/message.proto @@ -64,11 +64,23 @@ message PunchInfo { uint32 tcp_port = 11; repeated uint32 udp_ports = 12; repeated uint32 public_ports = 13; + uint32 public_tcp_port = 14; + PunchNatModel punch_model = 15; } enum PunchNatType { Symmetric = 0; Cone = 1; } +enum PunchNatModel { + All = 0; + IPv4 = 1; + IPv6 = 2; + IPv4Tcp = 3; + IPv4Udp = 4; + IPv6Tcp = 5; + IPv6Udp = 6; +} + /// 向服务器上报客户端状态信息 message ClientStatusInfo { fixed32 source = 1; diff --git a/vnt/src/channel/context.rs b/vnt/src/channel/context.rs index 505df6f..889c106 100644 --- a/vnt/src/channel/context.rs +++ b/vnt/src/channel/context.rs @@ -60,6 +60,7 @@ impl ChannelContext { up_traffic_meter, down_traffic_meter, default_interface, + default_route_key: AtomicCell::default(), }; Self { inner: Arc::new(inner), @@ -86,7 +87,7 @@ pub struct ContextInner { // 对称网络增加的udp socket sub_udp_socket: RwLock>, // tcp数据发送器 - pub(crate) packet_map: RwLock>, + pub(crate) packet_map: RwLock>, // 路由信息 pub route_table: RouteTable, // 使用什么协议连接服务器 @@ -98,6 +99,7 @@ pub struct ContextInner { pub(crate) up_traffic_meter: Option, pub(crate) down_traffic_meter: Option, default_interface: LocalInterface, + default_route_key: AtomicCell>, } impl ContextInner { @@ -107,6 +109,9 @@ impl ContextInner { pub fn default_interface(&self) -> &LocalInterface { &self.default_interface } + pub fn set_default_route_key(&self, route_key: RouteKey) { + self.default_route_key.store(Some(route_key)); + } /// 通过sub_udp_socket是否为空来判断是否为锥形网络 pub fn is_cone(&self) -> bool { self.sub_udp_socket.read().is_empty() @@ -175,11 +180,14 @@ impl ContextInner { } Ok(ports) } - pub fn send_tcp(&self, buf: &[u8], addr: SocketAddr) -> io::Result<()> { - if let Some(tcp) = self.packet_map.read().get(&addr) { + pub fn send_tcp(&self, buf: &[u8], route_key: &RouteKey) -> io::Result<()> { + if let Some(tcp) = self.packet_map.read().get(route_key) { tcp.try_send(buf) } else { - Err(io::Error::from(io::ErrorKind::NotFound)) + Err(io::Error::new( + io::ErrorKind::NotFound, + format!("dest={:?}", route_key), + )) } } pub fn send_main_udp(&self, index: usize, buf: &[u8], addr: SocketAddr) -> io::Result<()> { @@ -203,7 +211,14 @@ impl ContextInner { self.send_main_udp(self.v4_len, buf.buffer(), addr)? } } else { - self.send_tcp(buf.buffer(), addr)? + if let Some(key) = self.default_route_key.load() { + self.send_tcp(buf.buffer(), &key)? + } else { + return Err(io::Error::new( + io::ErrorKind::NotFound, + format!("dest={:?}", addr), + )); + } } if let Some(up_traffic_meter) = &self.up_traffic_meter { up_traffic_meter.add_traffic(buf.destination(), buf.data_len()); @@ -300,7 +315,7 @@ impl ContextInner { } } ConnectProtocol::TCP | ConnectProtocol::WS | ConnectProtocol::WSS => { - self.send_tcp(buf.buffer(), route_key.addr)? + self.send_tcp(buf.buffer(), &route_key)? } } if let Some(up_traffic_meter) = &self.up_traffic_meter { @@ -376,19 +391,11 @@ impl RouteTable { let key = route.route_key(); if only_if_absent { if let Some((_, list)) = self.route_table.read().get(&id) { - let mut p2p_num = 0; for (x, _) in list { - if x.is_p2p() { - p2p_num += 1; - } if x.route_key() == key { return true; } } - if !self.first_latency && p2p_num >= self.channel_num { - // 非优先延迟的情况下,通道满了则不用再添加 - return false; - } } } let mut route_table = self.route_table.write(); @@ -413,61 +420,42 @@ impl RouteTable { } } if exist { - // 这个排序还有待优化,因为后加入的大概率排最后,被直接淘汰的概率也大,可能导致更好的通道被移除了 list.sort_by_key(|(k, _)| k.rt); - //如果延迟都稳定了,则去除多余通道 - for (route, _) in list.iter() { - if route.rt == DEFAULT_RT { - return true; - } - } - //延迟优先模式需要更多的通道探测延迟最低的路线 - let limit_len = if self.first_latency { - self.channel_num + 2 - } else { - self.channel_num - }; - self.truncate_(list, limit_len); } else { if !self.first_latency { if route.is_p2p() { //非优先延迟的情况下 添加了直连的则排除非直连的 list.retain(|(k, _)| k.is_p2p()); } - if self.channel_num <= list.len() { - return false; - } }; - //增加路由表容量,避免波动 - let limit_len = self.channel_num * 2; list.sort_by_key(|(k, _)| k.rt); - self.truncate_(list, limit_len); list.push((route, AtomicCell::new(Instant::now()))); } return true; } - fn truncate_(&self, list: &mut Vec<(Route, AtomicCell)>, len: usize) { - if list.len() <= len { - return; - } - if self.first_latency { - //找到第一个p2p通道 - if let Some(index) = - list.iter() - .enumerate() - .find_map(|(index, (route, _))| if route.is_p2p() { Some(index) } else { None }) - { - if index >= len { - //保留第一个p2p通道 - let route = list.remove(index); - list.truncate(len - 1); - list.push(route); - return; - } - } - } - list.truncate(len); - } + // 直接移除会导致通道不稳定,所以废弃这个方法,后面改用多余通道不发心跳包,从而让通道自动过期 + // fn truncate_(&self, list: &mut Vec<(Route, AtomicCell)>, len: usize) { + // if list.len() <= len { + // return; + // } + // if self.first_latency { + // //找到第一个p2p通道 + // if let Some(index) = + // list.iter() + // .enumerate() + // .find_map(|(index, (route, _))| if route.is_p2p() { Some(index) } else { None }) + // { + // if index >= len { + // //保留第一个p2p通道 + // let route = list.remove(index); + // list.truncate(len - 1); + // list.push(route); + // return; + // } + // } + // } + // list.truncate(len); + // } pub fn route(&self, id: &Ipv4Addr) -> Option> { if let Some((_, v)) = self.route_table.read().get(id) { Some(v.iter().map(|(i, _)| *i).collect()) diff --git a/vnt/src/channel/mod.rs b/vnt/src/channel/mod.rs index 8cc1f3b..3c62424 100644 --- a/vnt/src/channel/mod.rs +++ b/vnt/src/channel/mod.rs @@ -179,7 +179,7 @@ pub struct RouteKey { } impl RouteKey { - pub(crate) fn new(protocol: ConnectProtocol, index: usize, addr: SocketAddr) -> Self { + pub(crate) const fn new(protocol: ConnectProtocol, index: usize, addr: SocketAddr) -> Self { Self { protocol, index, @@ -262,7 +262,13 @@ pub(crate) fn init_context( let socket = socket2::Socket::new(socket2::Domain::IPV4, socket2::Type::STREAM, None)?; (socket, address) }; - let _ = socket.set_reuse_address(true); + socket + .set_reuse_address(true) + .context("set_reuse_address")?; + #[cfg(unix)] + if let Err(e) = socket.set_reuse_port(true) { + log::warn!("set_reuse_port {:?}", e) + } if let Err(e) = socket.bind(&address.into()) { if ports[0] == 0 { //端口可能冲突,则使用任意端口 diff --git a/vnt/src/channel/punch.rs b/vnt/src/channel/punch.rs index 76e476c..3775134 100644 --- a/vnt/src/channel/punch.rs +++ b/vnt/src/channel/punch.rs @@ -14,16 +14,17 @@ use crate::channel::context::ChannelContext; use crate::channel::sender::ConnectUtil; use crate::handle::CurrentDeviceInfo; use crate::nat::{is_ipv4_global, NatTest}; +use crate::proto::message::{PunchNatModel, PunchNatType}; #[derive(Copy, Clone, Eq, PartialEq, Debug)] pub enum PunchModel { + All, IPv4, IPv6, IPv4Tcp, IPv4Udp, IPv6Tcp, IPv6Udp, - All, } impl PunchModel { @@ -72,6 +73,33 @@ impl Default for PunchModel { PunchModel::All } } +impl From for PunchNatModel { + fn from(value: PunchModel) -> Self { + match value { + PunchModel::All => PunchNatModel::All, + PunchModel::IPv4 => PunchNatModel::IPv4, + PunchModel::IPv6 => PunchNatModel::IPv6, + PunchModel::IPv4Tcp => PunchNatModel::IPv4Tcp, + PunchModel::IPv4Udp => PunchNatModel::IPv4Udp, + PunchModel::IPv6Tcp => PunchNatModel::IPv6Tcp, + PunchModel::IPv6Udp => PunchNatModel::IPv6Udp, + } + } +} + +impl Into for PunchNatModel { + fn into(self) -> PunchModel { + match self { + PunchNatModel::All => PunchModel::All, + PunchNatModel::IPv4 => PunchModel::IPv4, + PunchNatModel::IPv6 => PunchModel::IPv6, + PunchNatModel::IPv4Tcp => PunchModel::IPv4Tcp, + PunchNatModel::IPv4Udp => PunchModel::IPv4Udp, + PunchNatModel::IPv6Tcp => PunchModel::IPv6Tcp, + PunchNatModel::IPv6Udp => PunchModel::IPv6Udp, + } + } +} #[derive(Clone, Debug)] pub struct NatInfo { @@ -83,6 +111,8 @@ pub struct NatInfo { pub(crate) ipv6: Option, pub udp_ports: Vec, pub tcp_port: u16, + pub public_tcp_port: u16, + pub punch_model: PunchModel, } #[derive(Clone, Copy, PartialEq, Eq, Debug, Hash)] @@ -91,6 +121,29 @@ pub enum NatType { Cone, } +impl NatType { + pub fn is_cone(&self) -> bool { + self == &NatType::Cone + } +} +impl From for PunchNatType { + fn from(value: NatType) -> Self { + match value { + NatType::Symmetric => PunchNatType::Symmetric, + NatType::Cone => PunchNatType::Cone, + } + } +} + +impl Into for PunchNatType { + fn into(self) -> NatType { + match self { + PunchNatType::Symmetric => NatType::Symmetric, + PunchNatType::Cone => NatType::Cone, + } + } +} + impl NatInfo { pub fn new( mut public_ips: Vec, @@ -100,7 +153,9 @@ impl NatInfo { mut ipv6: Option, udp_ports: Vec, tcp_port: u16, + public_tcp_port: u16, mut nat_type: NatType, + punch_model: PunchModel, ) -> Self { public_ips.retain(|ip| { !ip.is_multicast() @@ -130,7 +185,9 @@ impl NatInfo { ipv6, udp_ports, tcp_port, + public_tcp_port, nat_type, + punch_model, } } pub fn update_addr(&mut self, index: usize, ip: Ipv4Addr, port: u16) -> bool { @@ -144,7 +201,7 @@ impl NatInfo { *public_port = port; } } - if crate::nat::is_ipv4_global(&ip) { + if is_ipv4_global(&ip) { if !self.public_ips.contains(&ip) { self.public_ips.push(ip); updated = true; @@ -153,6 +210,9 @@ impl NatInfo { } updated } + pub fn update_tcp_port(&mut self, port: u16) { + self.public_tcp_port = port; + } pub fn local_ipv4(&self) -> Option { self.local_ipv4 } @@ -252,7 +312,10 @@ impl Punch { if self.nat_test.is_local_address(true, addr) { return; } - self.connect_util.try_connect_tcp(buf.to_vec(), addr); + if addr.ip().is_unspecified() || addr.port() == 0 { + return; + } + self.connect_util.try_connect_tcp_punch(buf.to_vec(), addr); } pub fn punch( &mut self, @@ -277,44 +340,46 @@ impl Punch { nat_info.local_ipv4 = nat_info .local_ipv4 .filter(|ip| device_info.not_in_network(*ip)); - - if punch_tcp && self.punch_model.use_tcp() && nat_info.tcp_port != 0 { + if punch_tcp && self.punch_model.use_tcp() && nat_info.punch_model.use_tcp() { //向tcp发起连接 - if self.punch_model.use_ipv6() { + if self.punch_model.use_ipv6() && nat_info.punch_model.use_ipv6() { if let Some(ipv6_addr) = nat_info.local_tcp_ipv6addr() { self.connect_tcp(buf, ipv6_addr) } } - if self.punch_model.use_ipv4() { + if self.punch_model.use_ipv4() && nat_info.punch_model.use_ipv4() { if let Some(ipv4_addr) = nat_info.local_tcp_ipv4addr() { self.connect_tcp(buf, ipv4_addr) } for ip in &nat_info.public_ips { let addr = SocketAddr::V4(SocketAddrV4::new(*ip, nat_info.tcp_port)); - self.connect_tcp(buf, addr) + self.connect_tcp(buf, addr); + } + if nat_info.nat_type.is_cone() && nat_info.public_tcp_port != 0 { + for ip in &nat_info.public_ips { + let addr = SocketAddr::V4(SocketAddrV4::new(*ip, nat_info.public_tcp_port)); + self.connect_tcp(buf, addr); + } } } } - if !self.punch_model.use_udp() { + if !self.punch_model.use_udp() || !nat_info.punch_model.use_udp() { return Ok(()); } let channel_num = self.context.channel_num(); let main_len = self.context.main_len(); - if self.punch_model.use_ipv6() { + if self.punch_model.use_ipv6() && nat_info.punch_model.use_ipv6() { for index in channel_num..main_len { if let Some(ipv6_addr) = nat_info.local_udp_ipv6addr(index) { if !self.nat_test.is_local_address(false, ipv6_addr) { let rs = self.context.send_main_udp(index, buf, ipv6_addr); log::info!("发送到ipv6地址:{:?},rs={:?} {}", ipv6_addr, rs, id); - if rs.is_ok() && self.punch_model == PunchModel::IPv6 { - return Ok(()); - } } } } } - if !self.punch_model.use_ipv4() { + if !self.punch_model.use_ipv4() || !nat_info.punch_model.use_ipv4() { return Ok(()); } for index in 0..channel_num { diff --git a/vnt/src/channel/sender.rs b/vnt/src/channel/sender.rs index 5cfa27c..00255c8 100644 --- a/vnt/src/channel/sender.rs +++ b/vnt/src/channel/sender.rs @@ -239,13 +239,13 @@ impl PacketSender { #[derive(Clone)] pub struct ConnectUtil { - connect_tcp: Sender<(Vec, SocketAddr)>, + connect_tcp: Sender<(Vec, Option, SocketAddr)>, connect_ws: Sender<(Vec, String)>, } impl ConnectUtil { pub fn new( - connect_tcp: Sender<(Vec, SocketAddr)>, + connect_tcp: Sender<(Vec, Option, SocketAddr)>, connect_ws: Sender<(Vec, String)>, ) -> Self { Self { @@ -254,7 +254,13 @@ impl ConnectUtil { } } pub fn try_connect_tcp(&self, buf: Vec, addr: SocketAddr) { - if self.connect_tcp.try_send((buf, addr)).is_err() { + if self.connect_tcp.try_send((buf, None, addr)).is_err() { + log::warn!("try_connect_tcp failed {}", addr); + } + } + pub fn try_connect_tcp_punch(&self, buf: Vec, addr: SocketAddr) { + // 打洞的连接可以绑定随机端口 + if self.connect_tcp.try_send((buf, Some(0), addr)).is_err() { log::warn!("try_connect_tcp failed {}", addr); } } diff --git a/vnt/src/channel/socket/mod.rs b/vnt/src/channel/socket/mod.rs index 14052c2..55e495f 100644 --- a/vnt/src/channel/socket/mod.rs +++ b/vnt/src/channel/socket/mod.rs @@ -27,15 +27,22 @@ pub struct LocalInterface { pub async fn connect_tcp( addr: SocketAddr, + bind_port: u16, default_interface: &LocalInterface, ) -> anyhow::Result { - let socket = create_tcp(addr.is_ipv4(), default_interface)?; + let socket = create_tcp0(addr.is_ipv4(), bind_port, default_interface)?; Ok(socket.connect(addr).await?) } - pub fn create_tcp( v4: bool, default_interface: &LocalInterface, +) -> anyhow::Result { + create_tcp0(v4, 0, default_interface) +} +pub fn create_tcp0( + v4: bool, + bind_port: u16, + default_interface: &LocalInterface, ) -> anyhow::Result { let socket = if v4 { socket2::Socket::new( @@ -51,7 +58,26 @@ pub fn create_tcp( )? }; if v4 { - socket.set_ip_unicast_if(default_interface)?; + if let Err(e) = socket.set_ip_unicast_if(default_interface) { + log::warn!("set_ip_unicast_if {:?}", e) + } + } + if bind_port != 0 { + socket + .set_reuse_address(true) + .context("set_reuse_address")?; + #[cfg(unix)] + if let Err(e) = socket.set_reuse_port(true) { + log::warn!("set_reuse_port {:?}", e) + } + if v4 { + let addr: SocketAddr = format!("0.0.0.0:{}", bind_port).parse().unwrap(); + socket.bind(&addr.into())?; + } else { + socket.set_only_v6(true)?; + let addr: SocketAddr = format!("[::]:{}", bind_port).parse().unwrap(); + socket.bind(&addr.into())?; + } } socket.set_nonblocking(true)?; socket.set_nodelay(true)?; @@ -68,7 +94,9 @@ pub fn bind_udp_ops( socket2::Type::DGRAM, Some(Protocol::UDP), )?; - socket.set_ip_unicast_if(default_interface)?; + if let Err(e) = socket.set_ip_unicast_if(default_interface) { + log::warn!("set_ip_unicast_if {:?}", e) + } socket } else { let socket = socket2::Socket::new( diff --git a/vnt/src/channel/socket/unix.rs b/vnt/src/channel/socket/unix.rs index d638cf6..4a7b50a 100644 --- a/vnt/src/channel/socket/unix.rs +++ b/vnt/src/channel/socket/unix.rs @@ -1,8 +1,11 @@ -use crate::channel::socket::{get_interface, LocalInterface, VntSocketTrait}; +#[cfg(any(target_os = "linux", target_os = "macos"))] +use crate::channel::socket::get_interface; +use crate::channel::socket::{LocalInterface, VntSocketTrait}; +#[cfg(any(target_os = "linux", target_os = "macos"))] use anyhow::Context; use std::net::Ipv4Addr; -#[cfg(any(target_os = "linux", target_os = "android"))] +#[cfg(target_os = "linux")] impl VntSocketTrait for socket2::Socket { fn set_ip_unicast_if(&self, interface: &LocalInterface) -> anyhow::Result<()> { if let Some(name) = &interface.name { @@ -22,7 +25,14 @@ impl VntSocketTrait for socket2::Socket { Ok(()) } } +#[cfg(target_os = "android")] +impl VntSocketTrait for socket2::Socket { + fn set_ip_unicast_if(&self, _interface: &LocalInterface) -> anyhow::Result<()> { + Ok(()) + } +} +#[cfg(any(target_os = "linux", target_os = "macos"))] pub fn get_best_interface(dest_ip: Ipv4Addr) -> anyhow::Result { match get_interface(dest_ip) { Ok(iface) => return Ok(iface), @@ -33,3 +43,7 @@ pub fn get_best_interface(dest_ip: Ipv4Addr) -> anyhow::Result { // 应该再查路由表找到默认路由的 Ok(LocalInterface::default()) } +#[cfg(target_os = "android")] +pub fn get_best_interface(_dest_ip: Ipv4Addr) -> anyhow::Result { + Ok(LocalInterface::default()) +} diff --git a/vnt/src/channel/tcp_channel.rs b/vnt/src/channel/tcp_channel.rs index 71aadd6..4f6eb52 100644 --- a/vnt/src/channel/tcp_channel.rs +++ b/vnt/src/channel/tcp_channel.rs @@ -1,5 +1,9 @@ use anyhow::{anyhow, Context}; use std::net::SocketAddr; +#[cfg(unix)] +use std::os::unix::io::AsRawFd; +#[cfg(windows)] +use std::os::windows::io::AsRawSocket; use std::thread; use std::time::Duration; use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt}; @@ -10,13 +14,14 @@ use tokio::sync::mpsc::{channel, Receiver}; use crate::channel::context::ChannelContext; use crate::channel::handler::RecvChannelHandler; use crate::channel::sender::PacketSender; +use crate::channel::socket::create_tcp0; use crate::channel::{ConnectProtocol, RouteKey, BUFFER_SIZE, TCP_MAX_PACKET_SIZE}; use crate::util::StopManager; /// 监听tcp端口,等待客户端连接 pub fn tcp_listen( tcp_server: std::net::TcpListener, - receiver: Receiver<(Vec, SocketAddr)>, + receiver: Receiver<(Vec, Option, SocketAddr)>, recv_handler: H, context: ChannelContext, stop_manager: StopManager, @@ -28,6 +33,7 @@ where let worker = stop_manager.add_listener("tcpChannel".into(), move || { let _ = stop_sender.send(()); })?; + let bind_port = tcp_server.local_addr()?.port(); let runtime = tokio::runtime::Builder::new_multi_thread() .worker_threads(2) .enable_all() @@ -46,9 +52,9 @@ where } }); } - tokio::spawn( - async move { connect_tcp_handle(receiver, recv_handler, context).await }, - ); + tokio::spawn(async move { + connect_tcp_handle(receiver, recv_handler, context, bind_port).await + }); }); runtime.block_on(async { let _ = stop_receiver.await; @@ -61,18 +67,24 @@ where } async fn connect_tcp_handle( - mut receiver: Receiver<(Vec, SocketAddr)>, + mut receiver: Receiver<(Vec, Option, SocketAddr)>, recv_handler: H, context: ChannelContext, + listener_bind_port: u16, ) where H: RecvChannelHandler, { - while let Some((data, addr)) = receiver.recv().await { + while let Some((data, bind_port, addr)) = receiver.recv().await { let recv_handler = recv_handler.clone(); let context = context.clone(); + let bind_port = if let Some(bind_port) = bind_port { + bind_port + } else { + listener_bind_port + }; tokio::spawn(async move { - if let Err(e) = connect_tcp0(data, addr, recv_handler, context).await { - log::warn!("发送失败,链接终止:{:?},{:?}", addr, e); + if let Err(e) = connect_tcp0(data, addr, recv_handler, context, bind_port).await { + log::warn!("连接失败,链接终止:{:?},{:?}", addr, e); } }); } @@ -83,15 +95,23 @@ async fn connect_tcp0( addr: SocketAddr, recv_handler: H, context: ChannelContext, + bind_port: u16, ) -> anyhow::Result<()> where H: RecvChannelHandler, { - let mut stream = tokio::time::timeout( - Duration::from_secs(3), - crate::channel::socket::connect_tcp(addr, context.default_interface()), - ) - .await??; + let socket = if bind_port != 0 { + match create_tcp0(addr.is_ipv4(), bind_port, context.default_interface()) { + Ok(socket) => socket, + Err(e) => { + log::warn!("{:?}", e); + create_tcp0(addr.is_ipv4(), 0, context.default_interface())? + } + } + } else { + create_tcp0(addr.is_ipv4(), 0, context.default_interface())? + }; + let mut stream = tokio::time::timeout(Duration::from_secs(3), socket.connect(addr)).await??; tcp_write(&mut stream, &data).await?; tcp_stream_handle(stream, addr, recv_handler, context).await; @@ -110,6 +130,7 @@ where loop { let (stream, addr) = tcp_server.accept().await?; + tcp_stream_handle(stream, addr, recv_handler.clone(), context.clone()).await; } } @@ -123,12 +144,18 @@ pub async fn tcp_stream_handle( H: RecvChannelHandler, { let _ = stream.set_nodelay(true); + let local = stream.local_addr(); + #[cfg(windows)] + let index = stream.as_raw_socket() as usize; + #[cfg(unix)] + let index = stream.as_raw_fd() as usize; + let route_key = RouteKey::new(ConnectProtocol::TCP, index, addr); let (r, mut w) = stream.into_split(); let (sender, mut receiver) = channel::>(100); context .packet_map .write() - .insert(addr, PacketSender::new(sender)); + .insert(route_key, PacketSender::new(sender)); tokio::spawn(async move { while let Some(data) = receiver.recv().await { if let Err(e) = tcp_write(&mut w, &data).await { @@ -139,10 +166,10 @@ pub async fn tcp_stream_handle( 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) + if let Err(e) = tcp_read(r, addr, &context, recv_handler, route_key).await { + log::warn!("tcp_read {:?} {local:?}-{addr}", e) } - context.packet_map.write().remove(&addr); + context.packet_map.write().remove(&route_key); }); } @@ -162,6 +189,7 @@ async fn tcp_read( addr: SocketAddr, context: &ChannelContext, recv_handler: H, + route_key: RouteKey, ) -> anyhow::Result<()> where H: RecvChannelHandler, @@ -179,11 +207,6 @@ where 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, - ); + recv_handler.handle(&mut buf[..len], &mut extend, route_key, context); } } diff --git a/vnt/src/channel/ws_channel.rs b/vnt/src/channel/ws_channel.rs index f786262..bb7409a 100644 --- a/vnt/src/channel/ws_channel.rs +++ b/vnt/src/channel/ws_channel.rs @@ -58,14 +58,16 @@ async fn connect_ws_handle( ) where H: RecvChannelHandler, { + let mut index = 0; 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 { + if let Err(e) = connect_ws(data, url, recv_handler, context, index).await { log::warn!("发送失败,ws链接终止:{:?}", e); } }); + index += 1; } } const WS_ADDR: SocketAddr = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0)); @@ -75,6 +77,7 @@ async fn connect_ws( mut url: String, recv_handler: H, context: ChannelContext, + index: usize, ) -> anyhow::Result<()> where H: RecvChannelHandler, @@ -114,10 +117,12 @@ where ws.send(Message::Binary(data)).await?; let (mut ws_write, ws_read) = ws.split(); let (sender, mut receiver) = channel::>(100); + let route_key = RouteKey::new(ConnectProtocol::WS, index, WS_ADDR); + context .packet_map .write() - .insert(WS_ADDR, PacketSender::new(sender)); + .insert(route_key, 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 { @@ -127,22 +132,22 @@ where } let _ = ws_write.close().await; }); - if let Err(e) = ws_read_handle(ws_read, recv_handler, &context).await { + if let Err(e) = ws_read_handle(ws_read, recv_handler, &context, route_key).await { log::warn!("{:?}", e); } - context.packet_map.write().remove(&WS_ADDR); + context.packet_map.write().remove(&route_key); Ok(()) } async fn ws_read_handle( mut ws_read: SplitStream>>, recv_handler: H, context: &ChannelContext, + route_key: RouteKey, ) -> 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 { diff --git a/vnt/src/core/conn.rs b/vnt/src/core/conn.rs index 248f8a8..dfad263 100644 --- a/vnt/src/core/conn.rs +++ b/vnt/src/core/conn.rs @@ -12,7 +12,6 @@ use crate::channel::context::ChannelContext; use crate::channel::idle::Idle; use crate::channel::punch::{NatInfo, Punch}; use crate::channel::sender::IpPacketSender; -use crate::channel::socket::LocalInterface; use crate::channel::{init_channel, init_context, Route, RouteKey}; use crate::cipher::Cipher; #[cfg(feature = "server_encrypt")] @@ -30,7 +29,7 @@ use crate::tun_tap_device::tun_create_helper::{DeviceAdapter, TunDeviceHelper}; use crate::tun_tap_device::vnt_device::DeviceWrite; use crate::util::limit::TrafficMeterMultiAddress; use crate::util::{Scheduler, StopManager}; -use crate::{channel, nat, VntCallback}; +use crate::{nat, VntCallback}; #[derive(Clone)] pub struct Vnt { @@ -138,17 +137,7 @@ impl VntInner { } else { nat::local_ipv4() }; - - let default_interface = if config.in_ips.is_empty() { - //没有改变路由,不需要绑定网卡 - LocalInterface::default() - } else { - //vnt的流量都走这个接口 - let default_interface = - channel::socket::get_best_interface(local_ipv4.unwrap_or(Ipv4Addr::UNSPECIFIED))?; - log::info!("default_interface = {:?}", default_interface); - default_interface - }; + let default_interface = config.local_interface.clone(); //基础信息 let config_info = BaseConfigInfo::new( @@ -215,6 +204,7 @@ impl VntInner { udp_ports, tcp_port, config.local_ipv4.is_none(), + config.punch_model, ); // 定时器 let scheduler = Scheduler::new(stop_manager.clone())?; diff --git a/vnt/src/core/mod.rs b/vnt/src/core/mod.rs index ac2d54b..ab8b8b7 100644 --- a/vnt/src/core/mod.rs +++ b/vnt/src/core/mod.rs @@ -5,6 +5,7 @@ use std::str::FromStr; pub use conn::Vnt; use crate::channel::punch::PunchModel; +use crate::channel::socket::LocalInterface; use crate::channel::{ConnectProtocol, UseChannelType}; use crate::cipher::CipherModel; use crate::compression::Compressor; @@ -53,6 +54,7 @@ pub struct Config { pub enable_traffic: bool, pub allow_wire_guard: bool, pub local_ipv4: Option, + pub local_interface: LocalInterface, } impl Config { @@ -94,6 +96,15 @@ impl Config { allow_wire_guard: bool, local_ipv4: Option, ) -> anyhow::Result { + #[cfg(windows)] + #[cfg(feature = "integrated_tun")] + if !tap { + if let Err(e) = tun::Device::check_tun_dll() { + log::warn!("校验平台dll {:?}", e); + Err(e)?; + } + } + for x in stun_server.iter_mut() { if !x.contains(":") { x.push_str(":3478"); @@ -139,8 +150,11 @@ impl Config { server_address_str = s.to_string(); protocol = ConnectProtocol::TCP; } - server_address = - address_choose(dns_query_all(&server_address_str, name_servers.clone())?)?; + server_address = address_choose(dns_query_all( + &server_address_str, + name_servers.clone(), + &LocalInterface::default(), + )?)?; } #[cfg(feature = "port_mapping")] let port_mapping_list = crate::port_mapping::convert(port_mapping_list)?; @@ -149,9 +163,13 @@ impl Config { *dest = *mask & *dest; } in_ips.sort_by(|(dest1, _, _), (dest2, _, _)| dest2.cmp(dest1)); - if let Some(local_ip) = local_ipv4 { - let _ = crate::channel::socket::get_interface(local_ip)?; - } + let local_interface = if let Some(local_ip) = local_ipv4 { + let default_interface = crate::channel::socket::get_interface(local_ip)?; + log::info!("default_interface = {:?}", default_interface); + default_interface + } else { + LocalInterface::default() + }; Ok(Self { #[cfg(feature = "integrated_tun")] #[cfg(target_os = "windows")] @@ -190,6 +208,7 @@ impl Config { enable_traffic, allow_wire_guard, local_ipv4, + local_interface, }) } } diff --git a/vnt/src/handle/callback.rs b/vnt/src/handle/callback.rs index 7d33f61..7f32eb4 100644 --- a/vnt/src/handle/callback.rs +++ b/vnt/src/handle/callback.rs @@ -172,6 +172,8 @@ pub enum ErrorType { IpAlreadyExists, InvalidIp, LocalIpExists, + FailedToCrateDevice, + Warn, Unknown, } @@ -184,6 +186,8 @@ impl Into for ErrorType { ErrorType::IpAlreadyExists => 4, ErrorType::InvalidIp => 5, ErrorType::LocalIpExists => 6, + ErrorType::FailedToCrateDevice => 101, + ErrorType::Warn => 102, ErrorType::Unknown => 255, } } diff --git a/vnt/src/handle/maintain/heartbeat.rs b/vnt/src/handle/maintain/heartbeat.rs index d5e21ee..5327a11 100644 --- a/vnt/src/handle/maintain/heartbeat.rs +++ b/vnt/src/handle/maintain/heartbeat.rs @@ -56,6 +56,7 @@ fn heartbeat0( ) { let gateway_ip = current_device.virtual_gateway; let src_ip = current_device.virtual_ip; + let channel_num = context.channel_num(); // 可能服务器ip发生变化,导致发送失败 let mut is_send_gateway = false; match heartbeat_packet_server(device_map, server_cipher, src_ip, gateway_ip) { @@ -87,7 +88,16 @@ fn heartbeat0( continue; } }; - for route in routes { + for (index, route) in routes.iter().enumerate() { + let limit = if context.first_latency() { + channel_num + 1 + } else { + channel_num + }; + if index >= limit { + // 多余的通道不再发送心跳包,让它自动过期 + break; + } if let Err(e) = context.send_by_key(&net_packet, route.route_key()) { log::warn!("heartbeat err={:?}", e) } diff --git a/vnt/src/handle/maintain/idle.rs b/vnt/src/handle/maintain/idle.rs index d610018..f36a26a 100644 --- a/vnt/src/handle/maintain/idle.rs +++ b/vnt/src/handle/maintain/idle.rs @@ -7,6 +7,7 @@ use crossbeam_utils::atomic::AtomicCell; use crate::channel::context::ChannelContext; use crate::channel::idle::{Idle, IdleType}; use crate::channel::sender::ConnectUtil; +use crate::channel::socket::LocalInterface; use crate::channel::ConnectProtocol; use crate::handle::callback::{ConnectInfo, ErrorType}; use crate::handle::handshaker::Handshake; @@ -130,7 +131,8 @@ fn check_gateway_channel( let connect_protocol = context.main_protocol(); if connect_protocol.is_transport() { // 传输层的协议需要探测服务器地址 - current_device = domain_request0(current_device_info, config); + current_device = + domain_request0(current_device_info, config, context.default_interface()); } //需要重连 call.connect(ConnectInfo::new(*count, current_device.connect_server)); @@ -160,11 +162,16 @@ fn check_gateway_channel( pub fn domain_request0( current_device: &AtomicCell, config: &BaseConfigInfo, + default_interface: &LocalInterface, ) -> CurrentDeviceInfo { let mut current_dev = current_device.load(); // 探测服务端地址变化 - match dns_query_all(&config.server_addr, config.name_servers.clone()) { + match dns_query_all( + &config.server_addr, + config.name_servers.clone(), + default_interface, + ) { Ok(addrs) => { log::info!( "domain {} dns {:?} addr {:?}", diff --git a/vnt/src/handle/maintain/punch.rs b/vnt/src/handle/maintain/punch.rs index 98ded7a..cef4039 100644 --- a/vnt/src/handle/maintain/punch.rs +++ b/vnt/src/handle/maintain/punch.rs @@ -149,7 +149,7 @@ fn punch_start( *v += 1; *v } else { - guard.insert(peer_ip, 1); + guard.insert(peer_ip, 0); 0 } }; @@ -326,6 +326,7 @@ fn punch_packet( punch_reply.public_port = nat_info.public_ports.get(0).map_or(0, |v| *v as u32); punch_reply.public_ports = nat_info.public_ports.iter().map(|e| *e as u32).collect(); punch_reply.public_port_range = nat_info.public_port_range as u32; + punch_reply.public_tcp_port = nat_info.public_tcp_port as u32; punch_reply.local_ip = u32::from(nat_info.local_ipv4().unwrap_or(Ipv4Addr::UNSPECIFIED)); punch_reply.local_port = nat_info.udp_ports[0] as u32; punch_reply.tcp_port = nat_info.tcp_port as u32; @@ -335,6 +336,7 @@ fn punch_packet( punch_reply.ipv6 = ipv6.octets().to_vec(); } punch_reply.nat_type = protobuf::EnumOrUnknown::new(PunchNatType::from(nat_info.nat_type)); + punch_reply.punch_model = protobuf::EnumOrUnknown::new(nat_info.punch_model.into()); log::info!("请求打洞={:?}", punch_reply); let bytes = punch_reply .write_to_bytes() diff --git a/vnt/src/handle/mod.rs b/vnt/src/handle/mod.rs index ea51732..2fec003 100644 --- a/vnt/src/handle/mod.rs +++ b/vnt/src/handle/mod.rs @@ -223,10 +223,10 @@ impl CurrentDeviceInfo { virtual_gateway: Ipv4Addr, ) { let broadcast_ip = (!u32::from_be_bytes(virtual_netmask.octets())) - | u32::from_be_bytes(virtual_gateway.octets()); + | u32::from_be_bytes(virtual_ip.octets()); let broadcast_ip = Ipv4Addr::from(broadcast_ip); - let virtual_network = u32::from_be_bytes(virtual_netmask.octets()) - & u32::from_be_bytes(virtual_gateway.octets()); + let virtual_network = + u32::from_be_bytes(virtual_netmask.octets()) & u32::from_be_bytes(virtual_ip.octets()); let virtual_network = Ipv4Addr::from(virtual_network); self.virtual_ip = virtual_ip; self.virtual_netmask = virtual_netmask; diff --git a/vnt/src/handle/recv_data/client.rs b/vnt/src/handle/recv_data/client.rs index fc61721..e658ad7 100644 --- a/vnt/src/handle/recv_data/client.rs +++ b/vnt/src/handle/recv_data/client.rs @@ -216,17 +216,13 @@ impl ClientPacketHandler { match ControlPacket::new(net_packet.transport_protocol(), net_packet.payload())? { ControlPacket::PingPacket(_) => { let route = Route::from_default_rt(route_key, metric); - if context.route_table.add_route_if_absent(source, route) - || net_packet.source() < current_device.virtual_ip - { - //在路由表中,或者来源比自己小,就需要回复,注意不能调换顺序 - net_packet.set_transport_protocol(control_packet::Protocol::Pong.into()); - net_packet.set_source(current_device.virtual_ip); - net_packet.set_destination(source); - net_packet.first_set_ttl(MAX_TTL); - self.client_cipher.encrypt_ipv4(&mut net_packet)?; - context.send_by_key(&net_packet, route_key)?; - } + context.route_table.add_route_if_absent(source, route); + net_packet.set_transport_protocol(control_packet::Protocol::Pong.into()); + net_packet.set_source(current_device.virtual_ip); + net_packet.set_destination(source); + net_packet.first_set_ttl(MAX_TTL); + self.client_cipher.encrypt_ipv4(&mut net_packet)?; + context.send_by_key(&net_packet, route_key)?; } ControlPacket::PongPacket(pong_packet) => { let current_time = crate::handle::now_time() as u16; @@ -272,7 +268,7 @@ impl ClientPacketHandler { { return Ok(()); } - let route = Route::from_default_rt(route_key, 1); + let route = Route::from_default_rt(route_key, metric); context.route_table.add_route_if_absent(source, route); } ControlPacket::AddrRequest => match route_key.addr.ip() { @@ -318,6 +314,7 @@ impl ClientPacketHandler { .collect(); let local_ipv4 = Some(Ipv4Addr::from(punch_info.local_ip.to_be_bytes())); let tcp_port = punch_info.tcp_port as u16; + let public_tcp_port = punch_info.public_tcp_port as u16; let ipv6 = if punch_info.ipv6.len() == 16 { let ipv6: [u8; 16] = punch_info.ipv6.try_into().unwrap(); Some(Ipv6Addr::from(ipv6)) @@ -340,7 +337,9 @@ impl ClientPacketHandler { ipv6, punch_info.udp_ports.iter().map(|e| *e as u16).collect(), tcp_port, + public_tcp_port, punch_info.nat_type.enum_value_or_default().into(), + punch_info.punch_model.enum_value_or_default().into(), ); { let peer_nat_info = peer_nat_info.clone(); @@ -360,8 +359,11 @@ impl ClientPacketHandler { nat_info.public_ports.iter().map(|e| *e as u32).collect(); punch_reply.public_port_range = nat_info.public_port_range as u32; punch_reply.tcp_port = nat_info.tcp_port as u32; + punch_reply.public_tcp_port = nat_info.public_tcp_port as u32; punch_reply.nat_type = protobuf::EnumOrUnknown::new(PunchNatType::from(nat_info.nat_type)); + punch_reply.punch_model = + protobuf::EnumOrUnknown::new(nat_info.punch_model.into()); punch_reply.local_ip = u32::from(nat_info.local_ipv4().unwrap_or(Ipv4Addr::UNSPECIFIED)); punch_reply.local_port = nat_info.udp_ports[0] as u32; diff --git a/vnt/src/handle/recv_data/server.rs b/vnt/src/handle/recv_data/server.rs index a9a4380..c3adf7b 100644 --- a/vnt/src/handle/recv_data/server.rs +++ b/vnt/src/handle/recv_data/server.rs @@ -151,6 +151,8 @@ impl PacketHandler for ServerPacketHandl let response = HandshakeResponse::parse_from_bytes(net_packet.payload()) .map_err(|e| anyhow!("HandshakeResponse {:?}", e))?; log::info!("握手响应:{:?},{}", route_key, response); + //设置为默认通道 + context.set_default_route_key(route_key); //如果开启了加密,则发送加密握手请求 #[cfg(feature = "server_encrypt")] if let Some(key) = self.server_cipher.key() { @@ -211,7 +213,7 @@ impl PacketHandler for ServerPacketHandl let handshake_info = HandshakeInfo::new_no_secret(response.version); if self.callback.handshake(handshake_info) { //没有加密,则发送注册请求 - self.register(current_device, context)?; + self.register(current_device, context, route_key)?; } return Ok(()); @@ -295,6 +297,10 @@ impl ServerPacketHandler { let public_port = response.public_port as u16; self.nat_test .update_addr(route_key.index(), public_ip, public_port); + if route_key.protocol().is_tcp() { + log::info!("更新公网tcp端口 {public_port}"); + self.nat_test.update_tcp_port(public_port); + } let old = current_device; let mut cur = *current_device; loop { @@ -349,7 +355,10 @@ impl ServerPacketHandler { target_os = "linux", target_os = "macos" ))] - match crate::tun_tap_device::create_device(device_config) { + match crate::tun_tap_device::create_device( + device_config, + &self.callback, + ) { Ok(device) => { use tun::device::IFace; let tun_info = crate::handle::callback::DeviceInfo::new( @@ -379,7 +388,7 @@ impl ServerPacketHandler { let device_fd = self.callback.generate_tun(device_config); if device_fd == 0 { self.callback.error(ErrorInfo::new_msg( - ErrorType::Unknown, + ErrorType::FailedToCrateDevice, "device_fd == 0".into(), )); } else { @@ -390,14 +399,14 @@ impl ServerPacketHandler { self.config_info.allow_wire_guard, ) { self.callback.error(ErrorInfo::new_msg( - ErrorType::Unknown, + ErrorType::FailedToCrateDevice, format!("{:?}", e), )); } } Err(e) => { self.callback.error(ErrorInfo::new_msg( - ErrorType::Unknown, + ErrorType::FailedToCrateDevice, format!("{:?}", e), )); } @@ -421,7 +430,7 @@ impl ServerPacketHandler { service_packet::Protocol::SecretHandshakeResponse => { log::info!("SecretHandshakeResponse"); //加密握手结束,发送注册数据 - self.register(current_device, context)?; + self.register(current_device, context, route_key)?; } _ => { log::warn!( @@ -466,11 +475,14 @@ impl ServerPacketHandler { &self, current_device: &CurrentDeviceInfo, context: &ChannelContext, + route_key: RouteKey, ) -> anyhow::Result<()> { if current_device.status.online() { log::info!("已连接的不需要注册,{:?}", self.config_info); return Ok(()); } + //设置为默认通道 + context.set_default_route_key(route_key); let token = self.config_info.token.clone(); let device_id = self.config_info.device_id.clone(); let name = self.config_info.name.clone(); diff --git a/vnt/src/ip_proxy/icmp_proxy.rs b/vnt/src/ip_proxy/icmp_proxy.rs index 60353df..9afca4b 100644 --- a/vnt/src/ip_proxy/icmp_proxy.rs +++ b/vnt/src/ip_proxy/icmp_proxy.rs @@ -52,7 +52,9 @@ impl IcmpProxy { .bind(&socket2::SockAddr::from(addr)) .context("bind Socket ICMPV4 failed")?; icmp_socket.set_nonblocking(true)?; - icmp_socket.set_ip_unicast_if(default_interface)?; + if let Err(e) = icmp_socket.set_ip_unicast_if(default_interface) { + log::warn!("set_ip_unicast_if {:?}", e) + } let std_socket: std::net::UdpSocket = icmp_socket.into(); let tokio_icmp_socket = UdpSocket::from_std(std_socket.try_clone()?)?; diff --git a/vnt/src/nat/mod.rs b/vnt/src/nat/mod.rs index 1ef7b2b..628eddf 100644 --- a/vnt/src/nat/mod.rs +++ b/vnt/src/nat/mod.rs @@ -10,9 +10,8 @@ use parking_lot::Mutex; use rand::prelude::SliceRandom; use rand::Rng; -use crate::channel::punch::{NatInfo, NatType}; +use crate::channel::punch::{NatInfo, NatType, PunchModel}; use crate::channel::socket::LocalInterface; -use crate::proto::message::PunchNatType; #[cfg(feature = "upnp")] use crate::util::UPnP; @@ -120,24 +119,6 @@ pub struct NatTest { pub(crate) update_local_ipv4: bool, } -impl From for PunchNatType { - fn from(value: NatType) -> Self { - match value { - NatType::Symmetric => PunchNatType::Symmetric, - NatType::Cone => PunchNatType::Cone, - } - } -} - -impl Into for PunchNatType { - fn into(self) -> NatType { - match self { - PunchNatType::Symmetric => NatType::Symmetric, - PunchNatType::Cone => NatType::Cone, - } - } -} - impl NatTest { pub fn new( _channel_num: usize, @@ -147,6 +128,7 @@ impl NatTest { udp_ports: Vec, tcp_port: u16, update_local_ipv4: bool, + punch_model: PunchModel, ) -> NatTest { let ports = vec![0; udp_ports.len()]; let nat_info = NatInfo::new( @@ -157,7 +139,9 @@ impl NatTest { ipv6, udp_ports.clone(), tcp_port, + 0, NatType::Cone, + punch_model, ); let info = Arc::new(Mutex::new(nat_info)); #[cfg(feature = "upnp")] @@ -257,6 +241,10 @@ impl NatTest { let mut guard = self.info.lock(); guard.update_addr(index, ip, port) } + pub fn update_tcp_port(&self, port: u16) { + let mut guard = self.info.lock(); + guard.update_tcp_port(port) + } pub fn re_test( &self, local_ipv4: Option, diff --git a/vnt/src/tun_tap_device/create_device.rs b/vnt/src/tun_tap_device/create_device.rs index 7485cde..2514c47 100644 --- a/vnt/src/tun_tap_device/create_device.rs +++ b/vnt/src/tun_tap_device/create_device.rs @@ -1,4 +1,4 @@ -use crate::{DeviceConfig, ErrorInfo, ErrorType}; +use crate::{DeviceConfig, ErrorInfo, ErrorType, VntCallback}; use std::io; use std::net::Ipv4Addr; use std::sync::Arc; @@ -10,12 +10,15 @@ const DEFAULT_TUN_NAME: &str = "vnt-tun"; #[cfg(target_os = "windows")] const DEFAULT_TAP_NAME: &str = "vnt-tap"; -pub fn create_device(config: DeviceConfig) -> Result, ErrorInfo> { +pub fn create_device( + config: DeviceConfig, + call: &Call, +) -> Result, ErrorInfo> { let device = match create_device0(&config) { Ok(device) => device, Err(e) => { return Err(ErrorInfo::new_msg( - ErrorType::Unknown, + ErrorType::FailedToCrateDevice, format!("create device {:?}", e), )); } @@ -44,7 +47,14 @@ pub fn create_device(config: DeviceConfig) -> Result, ErrorInfo> { for (dest, mask) in config.external_route { if let Err(e) = device.add_route(dest, mask, 1) { - log::warn!("添加路由失败 ={:?}", e); + log::warn!("添加路由失败,请检查-i参数是否和现有路由冲突 ={:?}", e); + call.error(ErrorInfo::new_msg( + ErrorType::Warn, + format!( + "警告! 添加路由失败,请检查-i参数是否和现有路由冲突 ={:?}", + e + ), + )) } } Ok(device) diff --git a/vnt/src/util/dns_query.rs b/vnt/src/util/dns_query.rs index 4b7300c..04314f3 100644 --- a/vnt/src/util/dns_query.rs +++ b/vnt/src/util/dns_query.rs @@ -5,6 +5,7 @@ use std::str::FromStr; use std::time::Duration; use std::{io, thread}; +use crate::channel::socket::LocalInterface; use anyhow::Context; use dns_parser::{Builder, Packet, QueryClass, QueryType, RData, ResponseCode}; @@ -79,6 +80,7 @@ fn address_choose0(addrs: Vec) -> anyhow::Result { pub fn dns_query_all( domain: &str, mut name_servers: Vec, + default_interface: &LocalInterface, ) -> anyhow::Result> { match SocketAddr::from_str(domain) { Ok(addr) => Ok(vec![addr]), @@ -102,7 +104,7 @@ pub fn dns_query_all( let mut err: Option = None; for name_server in name_servers { if let Some(domain) = txt_domain.as_ref() { - match txt_dns(domain, name_server) { + match txt_dns(domain, name_server, default_interface) { Ok(addr) => { if !addr.is_empty() { return Ok(addr); @@ -127,12 +129,14 @@ pub fn dns_query_all( let th1 = { let host = host.to_string(); let name_server = name_server.clone(); - thread::spawn(move || a_dns(host, name_server)) + let default_interface = default_interface.clone(); + thread::spawn(move || a_dns(host, name_server, &default_interface)) }; let th2 = { let host = host.to_string(); let name_server = name_server.clone(); - thread::spawn(move || aaaa_dns(host, name_server)) + let default_interface = default_interface.clone(); + thread::spawn(move || aaaa_dns(host, name_server, &default_interface)) }; let mut addr = Vec::new(); match th1.join().unwrap() { @@ -230,9 +234,13 @@ fn query<'a>( Ok(pkt) } -pub fn txt_dns(domain: &str, name_server: String) -> anyhow::Result> { +pub fn txt_dns( + domain: &str, + name_server: String, + default_interface: &LocalInterface, +) -> anyhow::Result> { let name_server: SocketAddr = name_server.parse()?; - let udp = bind_udp(name_server)?; + let udp = bind_udp(name_server, default_interface)?; let mut buf = [0; 65536]; let message = query(&udp, domain, name_server, QueryType::TXT, &mut buf)?; let mut rs = Vec::new(); @@ -249,19 +257,28 @@ pub fn txt_dns(domain: &str, name_server: String) -> anyhow::Result anyhow::Result { - let udp = if name_server.is_ipv4() { - UdpSocket::bind("0.0.0.0:0")? +fn bind_udp( + name_server: SocketAddr, + default_interface: &LocalInterface, +) -> anyhow::Result { + let addr: SocketAddr = if name_server.is_ipv4() { + "0.0.0.0:0".parse().unwrap() } else { - UdpSocket::bind("[::]:0")? + "[::]:0".parse().unwrap() }; - udp.set_read_timeout(Some(Duration::from_millis(800)))?; - Ok(udp) + let socket = crate::channel::socket::bind_udp(addr, default_interface)?; + socket.set_nonblocking(false)?; + socket.set_read_timeout(Some(Duration::from_millis(800)))?; + Ok(socket.into()) } -pub fn a_dns(domain: String, name_server: String) -> anyhow::Result> { +pub fn a_dns( + domain: String, + name_server: String, + default_interface: &LocalInterface, +) -> anyhow::Result> { let name_server: SocketAddr = name_server.parse()?; - let udp = bind_udp(name_server)?; + let udp = bind_udp(name_server, default_interface)?; let mut buf = [0; 65536]; let message = query(&udp, &domain, name_server, QueryType::A, &mut buf)?; let mut rs = Vec::new(); @@ -273,9 +290,13 @@ pub fn a_dns(domain: String, name_server: String) -> anyhow::Result anyhow::Result> { +pub fn aaaa_dns( + domain: String, + name_server: String, + default_interface: &LocalInterface, +) -> anyhow::Result> { let name_server: SocketAddr = name_server.parse()?; - let udp = bind_udp(name_server)?; + let udp = bind_udp(name_server, default_interface)?; let mut buf = [0; 65536]; let message = query(&udp, &domain, name_server, QueryType::AAAA, &mut buf)?; let mut rs = Vec::new(); diff --git a/vnt/tun/Cargo.toml b/vnt/tun/Cargo.toml index 08b8d16..0e56282 100644 --- a/vnt/tun/Cargo.toml +++ b/vnt/tun/Cargo.toml @@ -19,15 +19,16 @@ ioctl = { version = "0.8", package = "ioctl-sys" } [target.'cfg(target_os = "windows")'.dependencies] libloading = "0.8.0" widestring = "1.0.2" -winapi = {version = "0.3",features = [ +winapi = { version = "0.3", features = [ "errhandlingapi", + "libloaderapi", "combaseapi", "ioapiset", "winioctl", "setupapi", "synchapi", "netioapi", - "fileapi","handleapi","winerror","minwindef","ifdef","basetsd","winnt","winreg","winbase","minwinbase", + "fileapi", "handleapi", "winerror", "minwindef", "ifdef", "basetsd", "winnt", "winreg", "winbase", "minwinbase", "impl-default" -]} +] } diff --git a/vnt/tun/src/windows/check.rs b/vnt/tun/src/windows/check.rs new file mode 100644 index 0000000..2e34779 --- /dev/null +++ b/vnt/tun/src/windows/check.rs @@ -0,0 +1,167 @@ +use libloading::Library; +use std::ffi::{c_char, CStr, CString}; +use std::fs::File; +use std::io::{self, Read, Seek}; +use std::path::PathBuf; +use winapi::shared::minwindef::HINSTANCE; +use winapi::um::libloaderapi::{GetModuleFileNameA, GetModuleHandleA}; + +#[repr(C)] +#[derive(Debug)] +struct DosHeader { + e_magic: u16, + e_cblp: u16, + e_cp: u16, + e_crlc: u16, + e_cparhdr: u16, + e_minalloc: u16, + e_maxalloc: u16, + e_ss: u16, + e_sp: u16, + e_csum: u16, + e_ip: u16, + e_cs: u16, + e_lfarlc: u16, + e_ovno: u16, + e_res: [u16; 4], + e_oemid: u16, + e_oeminfo: u16, + e_res2: [u16; 10], + e_lfanew: i32, +} + +#[repr(C)] +#[derive(Debug)] +struct FileHeader { + machine: u16, + number_of_sections: u16, + time_date_stamp: u32, + pointer_to_symbol_table: u32, + number_of_symbols: u32, + size_of_optional_header: u16, + characteristics: u16, +} + +const IMAGE_FILE_MACHINE_I386: u16 = 0x014C; +const IMAGE_FILE_MACHINE_AMD64: u16 = 0x8664; +const IMAGE_FILE_MACHINE_ARM: u16 = 0x01C4; +const IMAGE_FILE_MACHINE_ARM64: u16 = 0xAA64; + +fn get_dll_path(dll_name: &str) -> Result { + unsafe { + // 使用libloading加载DLL + + // 转换DLL名称为C字符串 + let dll_name_c = + CString::new(dll_name).map_err(|e| format!("Failed to convert to CString: {}", e))?; + + // 获取DLL的模块句柄 + let h_instance: HINSTANCE = GetModuleHandleA(dll_name_c.as_ptr() as *const c_char); + + if h_instance.is_null() { + return Err("Failed to get module handle".to_string()); + } + + // 获取DLL文件路径 + let mut buffer: [c_char; 260] = [0; 260]; + let length = GetModuleFileNameA(h_instance, buffer.as_mut_ptr(), buffer.len() as u32); + + if length == 0 { + return Err("Failed to get module file name".to_string()); + } + + let path = CStr::from_ptr(buffer.as_ptr()); + let path_str = path + .to_str() + .map_err(|e| format!("Failed to convert to &str: {}", e))?; + Ok(PathBuf::from(path_str)) + } +} + +pub fn check_win_tun_dll() -> io::Result<()> { + let _lib = unsafe { + Library::new("wintun.dll").map_err(|_| { + io::Error::new( + io::ErrorKind::NotFound, + "wintun.dll not found,Please download https://www.wintun.net", + ) + }) + }; + match get_dll_path("wintun.dll") { + Ok(path) => match_platform(path), + Err(e) => { + // 能加载说明存在wintun,这里获取不到路径是代码的问题 + log::info!("{:?}", e); + Ok(()) + } + } +} + +fn match_platform(path: PathBuf) -> io::Result<()> { + let current_arch = if cfg!(target_arch = "x86") { + "x86" + } else if cfg!(target_arch = "x86_64") { + "AMD64" + } else if cfg!(target_arch = "arm") { + "ARM" + } else if cfg!(target_arch = "aarch64") { + "ARM64" + } else { + return Ok(()); + }; + + let mut file = File::open(&path)?; + + // 读取 DOS 头部 + let mut dos_header = [0u8; std::mem::size_of::()]; + file.read_exact(&mut dos_header)?; + let dos_header: DosHeader = unsafe { std::ptr::read(dos_header.as_ptr() as *const _) }; + + if dos_header.e_magic != 0x5A4D { + return Err(io::Error::new( + io::ErrorKind::Other, + format!("Not a valid PE file {:?}", path), + )); + } + + // 跳转到 PE 头部 + file.seek(io::SeekFrom::Start(dos_header.e_lfanew as u64))?; + + // 读取 PE 头部 + let mut pe_signature = [0u8; 4]; + file.read_exact(&mut pe_signature)?; + if &pe_signature != b"PE\0\0" { + return Err(io::Error::new( + io::ErrorKind::Other, + format!("Not a valid PE file {:?}", path), + )); + } + + // 读取文件头部 + let mut file_header = [0u8; std::mem::size_of::()]; + file.read_exact(&mut file_header)?; + let file_header: FileHeader = unsafe { std::ptr::read(file_header.as_ptr() as *const _) }; + let dll_arch = match file_header.machine { + IMAGE_FILE_MACHINE_I386 => "x86", + IMAGE_FILE_MACHINE_AMD64 => "AMD64", + IMAGE_FILE_MACHINE_ARM => "ARM", + IMAGE_FILE_MACHINE_ARM64 => "ARM64", + _ => { + return Err(io::Error::new( + io::ErrorKind::Other, + format!("Unknown machine type: {}", file_header.machine), + )) + } + }; + + if dll_arch != current_arch { + return Err(io::Error::new( + io::ErrorKind::Other, + format!( + "wintun.dll architecture ({}) does not match the current platform architecture ({}).", + dll_arch, current_arch + ), + )); + } + Ok(()) +} diff --git a/vnt/tun/src/windows/device.rs b/vnt/tun/src/windows/device.rs index a557125..d0f81e8 100644 --- a/vnt/tun/src/windows/device.rs +++ b/vnt/tun/src/windows/device.rs @@ -16,6 +16,9 @@ impl Device { Ok(Device::Tun(tun::Device::new(name)?)) } } + pub fn check_tun_dll() -> io::Result<()> { + crate::windows::check::check_win_tun_dll() + } } impl IFace for Device { diff --git a/vnt/tun/src/windows/mod.rs b/vnt/tun/src/windows/mod.rs index 238b073..cc3a17a 100644 --- a/vnt/tun/src/windows/mod.rs +++ b/vnt/tun/src/windows/mod.rs @@ -3,6 +3,7 @@ use std::os::windows::process::CommandExt; use winapi::shared::minwindef::DWORD; use winapi::um::winbase::CREATE_NO_WINDOW; +mod check; mod device; mod ffi; mod netsh; diff --git a/vnt/tun/src/windows/tun/mod.rs b/vnt/tun/src/windows/tun/mod.rs index 8c826eb..27438bd 100644 --- a/vnt/tun/src/windows/tun/mod.rs +++ b/vnt/tun/src/windows/tun/mod.rs @@ -154,6 +154,7 @@ impl Device { } fn hash_guid(input: &str) -> [u8; 16] { let mut hasher = sha2::Sha256::new(); + hasher.update(input.as_bytes()); hasher.update(b"VNT"); hasher.update(input.as_bytes()); hasher.update(b"2024");