From 284cf38f0b32b3676a86f3c7be06586339faddf2 Mon Sep 17 00:00:00 2001 From: lubeilin <1791778603@qq.com> Date: Sat, 30 Mar 2024 00:24:19 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E5=B9=BF=E6=92=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- vnt/src/core/conn.rs | 1 + vnt/src/handle/tun_tap/mod.rs | 118 +++++++++++++++++--------- vnt/src/handle/tun_tap/tun_handler.rs | 11 ++- 3 files changed, 87 insertions(+), 43 deletions(-) diff --git a/vnt/src/core/conn.rs b/vnt/src/core/conn.rs index 42a13e5..48b2ad4 100644 --- a/vnt/src/core/conn.rs +++ b/vnt/src/core/conn.rs @@ -190,6 +190,7 @@ impl Vnt { server_cipher.clone(), config.parallel, up_counter, + device_list.clone(), )?; maintain::idle_gateway( &scheduler, diff --git a/vnt/src/handle/tun_tap/mod.rs b/vnt/src/handle/tun_tap/mod.rs index 4c4e190..e26f788 100644 --- a/vnt/src/handle/tun_tap/mod.rs +++ b/vnt/src/handle/tun_tap/mod.rs @@ -1,13 +1,15 @@ use std::io; use std::net::Ipv4Addr; -use crate::channel::context::Context; +use parking_lot::Mutex; + use packet::ip::ipv4::packet::IpV4Packet; use packet::ip::ipv4::protocol::Protocol; +use crate::channel::context::Context; use crate::cipher::Cipher; use crate::external_route::ExternalRoute; -use crate::handle::{check_dest, CurrentDeviceInfo}; +use crate::handle::{check_dest, CurrentDeviceInfo, PeerDeviceInfo}; #[cfg(feature = "ip_proxy")] use crate::ip_proxy::{IpProxyMap, ProxyHandler}; use crate::protocol; @@ -23,57 +25,82 @@ fn broadcast( sender: &Context, net_packet: &mut NetPacket<&mut [u8]>, current_device: &CurrentDeviceInfo, + device_list: &Mutex<(u16, Vec)>, ) -> io::Result<()> { - let mut peer_ips = Vec::with_capacity(8); - let vec = sender.route_table.route_table_one(); - let mut relay_count = 0; + let list: Vec = device_list + .lock() + .1 + .iter() + .filter(|info| info.status.is_online()) + .map(|info| info.virtual_ip) + .collect(); const MAX_COUNT: usize = 8; - for (peer_ip, route) in vec { - if peer_ip == current_device.virtual_gateway { - continue; - } - if peer_ips.len() == MAX_COUNT { - relay_count += 1; + let mut p2p_ips = Vec::with_capacity(8); + let mut relay_ips = Vec::with_capacity(8); + let mut overflow = false; + for (index, peer_ip) in list.into_iter().enumerate() { + if index > MAX_COUNT { + overflow = true; break; } - if route.is_p2p() - && sender + if let Some(route) = sender.route_table.route_one_p2p(&peer_ip) { + if sender .send_by_key(net_packet.buffer(), route.route_key()) .is_ok() - { - peer_ips.push(peer_ip); - } else { - relay_count += 1; + { + p2p_ips.push(peer_ip); + continue; + } } + relay_ips.push(peer_ip); } - if (relay_count == 0 && !peer_ips.is_empty()) || current_device.status.offline() { - //不需要转发 + if !overflow && relay_ips.is_empty() { + //全部p2p,不需要服务器中转 return Ok(()); } - //转发到服务端的可选择广播，还要进行服务端加密 - if peer_ips.is_empty() { - sender.send_default(net_packet.buffer(), current_device.connect_server)?; - } else { - let buf = - vec![0u8; 12 + 1 + peer_ips.len() * 4 + net_packet.data_len() + ENCRYPTION_RESERVED]; - //剩余的发送到服务端，需要告知哪些已发送过 - let mut server_packet = NetPacket::new_encrypt(buf)?; - server_packet.set_version(Version::V1); - server_packet.set_gateway_flag(true); - server_packet.first_set_ttl(MAX_TTL); - server_packet.set_source(net_packet.source()); - //使用对应的目的地址 - server_packet.set_destination(net_packet.destination()); - server_packet.set_protocol(protocol::Protocol::IpTurn); - server_packet.set_transport_protocol(ip_turn_packet::Protocol::Ipv4Broadcast.into()); - let mut broadcast = BroadcastPacket::unchecked(server_packet.payload_mut()); - broadcast.set_address(&peer_ips)?; - broadcast.set_data(net_packet.buffer())?; - server_cipher.encrypt_ipv4(&mut server_packet)?; - sender.send_default(server_packet.buffer(), current_device.connect_server)?; + if p2p_ips.is_empty() { + //都没有p2p则直接由服务器转发 + if current_device.status.online() { + sender.send_default(net_packet.buffer(), current_device.connect_server)?; + } + return Ok(()); } - Ok(()) + if !overflow && relay_ips.len() == 2 { + // 如果转发的ip数不多就直接发 + for peer_ip in relay_ips { + //非直连的广播要改变目的地址，不然服务端收到了会再次广播 + net_packet.set_destination(peer_ip); + sender.send_ipv4_by_id( + net_packet.buffer(), + &peer_ip, + current_device.connect_server, + current_device.status.online(), + )?; + } + return Ok(()); + } + if current_device.status.offline() { + //离线的不再转发 + return Ok(()); + } + let buf = vec![0u8; 12 + 1 + p2p_ips.len() * 4 + net_packet.data_len() + ENCRYPTION_RESERVED]; + //剩余的发送到服务端，需要告知哪些已发送过 + let mut server_packet = NetPacket::new_encrypt(buf)?; + server_packet.set_version(Version::V1); + server_packet.set_gateway_flag(true); + server_packet.first_set_ttl(MAX_TTL); + server_packet.set_source(net_packet.source()); + //使用对应的目的地址 + server_packet.set_destination(net_packet.destination()); + server_packet.set_protocol(protocol::Protocol::IpTurn); + server_packet.set_transport_protocol(ip_turn_packet::Protocol::Ipv4Broadcast.into()); + + let mut broadcast = BroadcastPacket::unchecked(server_packet.payload_mut()); + broadcast.set_address(&p2p_ips)?; + broadcast.set_data(net_packet.buffer())?; + server_cipher.encrypt_ipv4(&mut server_packet)?; + sender.send_default(server_packet.buffer(), current_device.connect_server) } /// 实现一个原地发送，必须保证是如下结构 @@ -89,6 +116,7 @@ pub fn base_handle( #[cfg(feature = "ip_proxy")] proxy_map: &Option, client_cipher: &Cipher, server_cipher: &Cipher, + device_list: &Mutex<(u16, Vec)>, ) -> io::Result<()> { let ipv4_packet = IpV4Packet::new(&buf[12..data_len])?; let protocol = ipv4_packet.protocol(); @@ -118,7 +146,13 @@ pub fn base_handle( if dest_ip.is_broadcast() || current_device.broadcast_ip == dest_ip { // 广播 发送到直连目标 client_cipher.encrypt_ipv4(&mut net_packet)?; - broadcast(server_cipher, context, &mut net_packet, ¤t_device)?; + broadcast( + server_cipher, + context, + &mut net_packet, + ¤t_device, + device_list, + )?; return Ok(()); } if !check_dest( diff --git a/vnt/src/handle/tun_tap/tun_handler.rs b/vnt/src/handle/tun_tap/tun_handler.rs index 7ebcfe5..2803915 100644 --- a/vnt/src/handle/tun_tap/tun_handler.rs +++ b/vnt/src/handle/tun_tap/tun_handler.rs @@ -2,6 +2,7 @@ use std::sync::Arc; use std::{io, thread}; use crossbeam_utils::atomic::AtomicCell; +use parking_lot::Mutex; use packet::icmp::icmp::IcmpPacket; use packet::icmp::Kind; @@ -14,7 +15,7 @@ use crate::channel::context::Context; use crate::cipher::Cipher; use crate::external_route::ExternalRoute; use crate::handle::tun_tap::channel_group::{channel_group, GroupSyncSender}; -use crate::handle::CurrentDeviceInfo; +use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo}; #[cfg(feature = "ip_proxy")] use crate::ip_proxy::IpProxyMap; use crate::util::{SingleU64Adder, StopManager}; @@ -46,6 +47,7 @@ fn handle( #[cfg(feature = "ip_proxy")] proxy_map: &Option, client_cipher: &Cipher, server_cipher: &Cipher, + device_list: &Mutex<(u16, Vec)>, ) -> io::Result<()> { //忽略掉结构不对的情况（ipv6数据、win tap会读到空数据），不然日志打印太多了 let ipv4_packet = match IpV4Packet::new(&mut data[12..len]) { @@ -67,6 +69,7 @@ fn handle( proxy_map, client_cipher, server_cipher, + device_list, ); } @@ -81,6 +84,7 @@ pub fn start( server_cipher: Cipher, parallel: usize, mut up_counter: SingleU64Adder, + device_list: Arc)>>, ) -> io::Result<()> { let worker = { #[cfg(target_os = "macos")] @@ -110,6 +114,7 @@ pub fn start( let ip_proxy_map = ip_proxy_map.clone(); let client_cipher = client_cipher.clone(); let server_cipher = server_cipher.clone(); + let device_list = device_list.clone(); thread::Builder::new() .name(format!("tunHandler-{}", index)) .spawn(move || { @@ -129,6 +134,7 @@ pub fn start( &ip_proxy_map, &client_cipher, &server_cipher, + &device_list, ) { Ok(_) => {} Err(e) => { @@ -161,6 +167,7 @@ pub fn start( client_cipher, server_cipher, &mut up_counter, + device_list, ) { log::warn!("stop:{}", e); } @@ -180,6 +187,7 @@ fn start_simple( client_cipher: Cipher, server_cipher: Cipher, up_counter: &mut SingleU64Adder, + device_list: Arc)>>, ) -> io::Result<()> { let mut buf = [0; 1024 * 16]; loop { @@ -204,6 +212,7 @@ fn start_simple( &ip_proxy_map, &client_cipher, &server_cipher, + &device_list, ) { Ok(_) => {} Err(e) => {