From 1f9ed7b314b35b1cf882e3d79bbf0936a06a5516 Mon Sep 17 00:00:00 2001 From: lbl8603 <49143209+lbl8603@users.noreply.github.com> Date: Sat, 13 Jul 2024 12:01:41 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=E4=BF=AE=E6=94=B9features?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- common/src/cli.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/common/src/cli.rs b/common/src/cli.rs index f5c5c8b..e66bab8 100644 --- a/common/src/cli.rs +++ b/common/src/cli.rs @@ -113,10 +113,13 @@ pub fn parse_args_config() -> anyhow::Result, bool)> } else if matches.opt_present("all") { command::command(command::CommandEnum::All); return Ok(None); - } else if matches.opt_present("chart_a") { + } + #[cfg(feature = "command")] + if matches.opt_present("chart_a") { command::command(command::CommandEnum::ChartA); return Ok(None); } + #[cfg(feature = "command")] if let Some(v) = matches.opt_str("chart_b") { command::command(command::CommandEnum::ChartB(v)); return Ok(None); From ddfa89167d8b38331f3be5d5409d50a39cd7cd8e Mon Sep 17 00:00:00 2001 From: lbl8603 <49143209+lbl8603@users.noreply.github.com> Date: Sat, 13 Jul 2024 12:09:15 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E5=9C=B0=E5=9D=80?= =?UTF-8?q?=E6=8E=A2=E6=B5=8B=E5=92=8C=E8=BD=AC=E5=8F=91=E8=B7=AF=E5=BE=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- vnt/src/channel/context.rs | 30 ++++++++++---- vnt/src/core/conn.rs | 1 - vnt/src/handle/maintain/addr_request.rs | 55 +++---------------------- vnt/src/handle/maintain/punch.rs | 2 +- vnt/src/handle/maintain/re_nat_type.rs | 2 +- vnt/src/handle/recv_data/client.rs | 18 ++++---- vnt/src/nat/mod.rs | 22 ++++++---- 7 files changed, 53 insertions(+), 77 deletions(-) diff --git a/vnt/src/channel/context.rs b/vnt/src/channel/context.rs index a78c999..0bdce08 100644 --- a/vnt/src/channel/context.rs +++ b/vnt/src/channel/context.rs @@ -353,18 +353,18 @@ impl RouteTable { } Err(io::Error::new(io::ErrorKind::NotFound, "route not found")) } - pub fn add_route_if_absent(&self, id: Ipv4Addr, route: Route) { + pub fn add_route_if_absent(&self, id: Ipv4Addr, route: Route) -> bool { self.add_route_(id, route, true) } - pub fn add_route(&self, id: Ipv4Addr, route: Route) { + pub fn add_route(&self, id: Ipv4Addr, route: Route) -> bool { self.add_route_(id, route, false) } - fn add_route_(&self, id: Ipv4Addr, route: Route, only_if_absent: bool) { + fn add_route_(&self, id: Ipv4Addr, route: Route, only_if_absent: bool) -> bool { // 限制通道类型 match self.use_channel_type { UseChannelType::P2p => { if !route.is_p2p() { - return; + return false; } } _ => {} @@ -372,10 +372,18 @@ 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.route_key() == key { - return; + 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; } } } @@ -387,11 +395,11 @@ impl RouteTable { for (x, time) in list.iter_mut() { if x.metric < route.metric && !self.first_latency { //非优先延迟的情况下 不能比当前的路径更长 - return; + return false; } if x.route_key() == key { if only_if_absent { - return; + return true; } x.metric = route.metric; x.rt = route.rt; @@ -406,7 +414,7 @@ impl RouteTable { //如果延迟都稳定了，则去除多余通道 for (route, _) in list.iter() { if route.rt == DEFAULT_RT { - return; + return true; } } //延迟优先模式需要更多的通道探测延迟最低的路线 @@ -422,6 +430,9 @@ impl RouteTable { //非优先延迟的情况下 添加了直连的则排除非直连的 list.retain(|(k, _)| k.is_p2p()); } + if self.channel_num <= list.len() { + return false; + } }; //增加路由表容量，避免波动 let limit_len = self.channel_num * 2; @@ -429,6 +440,7 @@ impl RouteTable { 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 { diff --git a/vnt/src/core/conn.rs b/vnt/src/core/conn.rs index dfd35ad..6aab462 100644 --- a/vnt/src/core/conn.rs +++ b/vnt/src/core/conn.rs @@ -385,7 +385,6 @@ pub fn start( &scheduler, context.clone(), current_device.clone(), - server_cipher.clone(), nat_test.clone(), config_info.clone(), ); diff --git a/vnt/src/handle/maintain/addr_request.rs b/vnt/src/handle/maintain/addr_request.rs index 1030e5c..fff00e8 100644 --- a/vnt/src/handle/maintain/addr_request.rs +++ b/vnt/src/handle/maintain/addr_request.rs @@ -5,48 +5,30 @@ use crossbeam_utils::atomic::AtomicCell; use crate::channel::context::ChannelContext; use crate::channel::punch::NatType; -use crate::cipher::Cipher; use crate::handle::{BaseConfigInfo, CurrentDeviceInfo}; use crate::nat::NatTest; -use crate::protocol::body::ENCRYPTION_RESERVED; -use crate::protocol::{control_packet, NetPacket, Protocol, MAX_TTL}; use crate::util::Scheduler; pub fn addr_request( scheduler: &Scheduler, context: ChannelContext, current_device_info: Arc>, - server_cipher: Cipher, nat_test: NatTest, _config: BaseConfigInfo, ) { - pub_address_request( - scheduler, - context, - current_device_info.clone(), - server_cipher, - nat_test, - 0, - ); + pub_address_request(scheduler, context, current_device_info.clone(), nat_test, 0); } fn pub_address_request( scheduler: &Scheduler, context: ChannelContext, current_device_info: Arc>, - server_cipher: Cipher, nat_test: NatTest, count: usize, ) { let channel_num = context.channel_num(); let index = count % channel_num; - if let Err(e) = addr_request0( - &context, - ¤t_device_info, - &server_cipher, - &nat_test, - index, - ) { + if let Err(e) = addr_request0(&context, ¤t_device_info, &nat_test, index) { log::warn!("{:?}", e); } let nat_info = nat_test.nat_info(); @@ -58,7 +40,7 @@ fn pub_address_request( if index == channel_num - 1 { 19 } else { - 7 + 9 } } } else { @@ -66,14 +48,7 @@ fn pub_address_request( }; let rs = scheduler.timeout(Duration::from_secs(time), move |s| { - pub_address_request( - s, - context, - current_device_info, - server_cipher, - nat_test, - index + 1, - ) + pub_address_request(s, context, current_device_info, nat_test, index + 1) }); if !rs { log::info!("定时任务停止"); @@ -83,7 +58,6 @@ fn pub_address_request( fn addr_request0( context: &ChannelContext, current_device: &AtomicCell, - server_cipher: &Cipher, nat_test: &NatTest, index: usize, ) -> anyhow::Result<()> { @@ -91,24 +65,7 @@ fn addr_request0( if current_dev.status.offline() { return Ok(()); } - - 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; - let mut packet = NetPacket::new_encrypt([0; 12 + ENCRYPTION_RESERVED]).unwrap(); - packet.set_default_version(); - packet.set_gateway_flag(true); - packet.set_protocol(Protocol::Control); - packet.set_transport_protocol(control_packet::Protocol::AddrRequest.into()); - packet.first_set_ttl(MAX_TTL); - packet.set_source(src_ip); - packet.set_destination(gateway_ip); - server_cipher.encrypt_ipv4(&mut packet)?; - context.send_main_udp(index, packet.buffer(), current_dev.connect_server)?; - } else { - let (data, addr) = nat_test.send_data()?; - context.send_main_udp(index, &data, addr)?; - } + let (data, addr) = nat_test.send_data()?; + context.send_main_udp(index, &data, addr)?; Ok(()) } diff --git a/vnt/src/handle/maintain/punch.rs b/vnt/src/handle/maintain/punch.rs index 69c5ff8..51a21e9 100644 --- a/vnt/src/handle/maintain/punch.rs +++ b/vnt/src/handle/maintain/punch.rs @@ -233,7 +233,7 @@ fn punch0( || nat_info.public_ports.iter().filter(|&&v| v == 0).count() > nat_info.public_ports.len() / 2) { - log::info!("公网地址为空，暂时放弃打洞,第{}轮", total_count); + log::info!("未获取到公网地址，暂时放弃打洞,第{}轮", total_count); return Ok(()); } let current_ip = current_device.virtual_ip; diff --git a/vnt/src/handle/maintain/re_nat_type.rs b/vnt/src/handle/maintain/re_nat_type.rs index e8eca7d..0c784fc 100644 --- a/vnt/src/handle/maintain/re_nat_type.rs +++ b/vnt/src/handle/maintain/re_nat_type.rs @@ -44,7 +44,7 @@ fn retrieve_nat_type0( }; #[cfg(feature = "upnp")] nat_test.reset_upnp(); - log::info!("刷新nat成功") + log::info!("刷新nat结束") } }) .expect("natTest"); diff --git a/vnt/src/handle/recv_data/client.rs b/vnt/src/handle/recv_data/client.rs index 54a7b93..13879d3 100644 --- a/vnt/src/handle/recv_data/client.rs +++ b/vnt/src/handle/recv_data/client.rs @@ -212,14 +212,18 @@ impl ClientPacketHandler { let source = net_packet.source(); match ControlPacket::new(net_packet.transport_protocol(), net_packet.payload())? { ControlPacket::PingPacket(_) => { - 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)?; let route = Route::from_default_rt(route_key, metric); - context.route_table.add_route_if_absent(source, route); + 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)?; + } } ControlPacket::PongPacket(pong_packet) => { let current_time = crate::handle::now_time() as u16; diff --git a/vnt/src/nat/mod.rs b/vnt/src/nat/mod.rs index 6c3c4be..4f34810 100644 --- a/vnt/src/nat/mod.rs +++ b/vnt/src/nat/mod.rs @@ -1,4 +1,4 @@ -use anyhow::Context; +use anyhow::{anyhow, Context}; use std::io; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, ToSocketAddrs}; use std::net::{SocketAddr, UdpSocket}; @@ -139,17 +139,12 @@ impl Into for PunchNatType { impl NatTest { pub fn new( _channel_num: usize, - mut stun_server: Vec, + stun_server: Vec, local_ipv4: Option, ipv6: Option, udp_ports: Vec, tcp_port: u16, ) -> NatTest { - if stun_server.len() > 5 { - stun_server.shuffle(&mut rand::thread_rng()); - stun_server.truncate(5); - log::info!("stun_server truncate {:?}", stun_server); - } let ports = vec![0; udp_ports.len()]; let nat_info = NatInfo::new( Vec::new(), @@ -262,8 +257,17 @@ impl NatTest { &self, local_ipv4: Option, ipv6: Option, - ) -> io::Result { - let (nat_type, public_ips, port_range) = stun::stun_test_nat(self.stun_server.clone())?; + ) -> anyhow::Result { + let mut stun_server = self.stun_server.clone(); + if stun_server.len() > 5 { + stun_server.shuffle(&mut rand::thread_rng()); + stun_server.truncate(5); + log::info!("stun_server truncate {:?}", stun_server); + } + let (nat_type, public_ips, port_range) = stun::stun_test_nat(stun_server)?; + if public_ips.is_empty() { + Err(anyhow!("public_ips.is_empty"))? + } let mut guard = self.info.lock(); guard.nat_type = nat_type; guard.public_ips = public_ips;