From d7c121a756d8276ee223483e660508084a7ec613 Mon Sep 17 00:00:00 2001 From: lubeilin <1791778603@qq.com> Date: Wed, 20 Sep 2023 19:54:49 +0800 Subject: [PATCH] =?UTF-8?q?commit:=201.=E5=8E=BB=E9=99=A4=E7=BC=93?= =?UTF-8?q?=E5=86=B2=E6=B1=A0=202.=E6=95=B0=E6=8D=AE=E5=A4=84=E7=90=86?= =?UTF-8?q?=E6=94=B9=E4=B8=BA=E5=90=8C=E6=AD=A5=E6=96=B9=E6=B3=95=203.fmt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 2 +- vnt-cli/src/main.rs | 7 +-- vnt/Cargo.toml | 2 - vnt/src/channel/channel.rs | 70 ++++++++------------- vnt/src/handle/recv_handler.rs | 82 ++++++++++++------------- vnt/src/handle/tun_tap/channel_group.rs | 10 ++- vnt/src/handle/tun_tap/tap_handler.rs | 7 +-- vnt/src/handle/tun_tap/tun_handler.rs | 6 +- vnt/src/ip_proxy/tcp_proxy.rs | 6 +- 9 files changed, 76 insertions(+), 116 deletions(-) diff --git a/README.md b/README.md index 1fb1266..9558e4e 100644 --- a/README.md +++ b/README.md @@ -68,7 +68,7 @@ A virtual network tool (VPN) - Mac - Linux - Windows - - 使用tun网卡 依赖wintun.dll([win-tun](https://www.wintun.net/))(将dll放到同目录下，建议使用版本0.14.1) + - 默认使用tun网卡 依赖wintun.dll([win-tun](https://www.wintun.net/))(将dll放到同目录下，建议使用版本0.14.1) - 使用tap网卡 依赖tap-windows([win-tap](https://build.openvpn.net/downloads/releases/))(建议使用版本9.24.7) - Android - [VntApp](https://github.com/lbl8603/VntApp) diff --git a/vnt-cli/src/main.rs b/vnt-cli/src/main.rs index ea7adba..4049c0d 100644 --- a/vnt-cli/src/main.rs +++ b/vnt-cli/src/main.rs @@ -214,11 +214,8 @@ fn main() { return; } - let cipher_model = match matches - .opt_get::("model") { - Ok(model) => { - model.unwrap_or(CipherModel::AesGcm) - } + let cipher_model = match matches.opt_get::("model") { + Ok(model) => model.unwrap_or(CipherModel::AesGcm), Err(e) => { println!("'--model ' invalid,{}", e); return; diff --git a/vnt/Cargo.toml b/vnt/Cargo.toml index 7cc821b..9a02adc 100644 --- a/vnt/Cargo.toml +++ b/vnt/Cargo.toml @@ -14,8 +14,6 @@ crossbeam-utils = "0.8" crossbeam-epoch = "0.9.15" dashmap = "5.5.1" parking_lot = "0.12.1" -byte-pool = "0.2.4" -lazy_static = "1.4.0" rand = "0.8.5" sha2 = { version = "0.10.6", features = ["oid"] } thiserror = "1.0.37" diff --git a/vnt/src/channel/channel.rs b/vnt/src/channel/channel.rs index 8c6bbfb..d8957b3 100644 --- a/vnt/src/channel/channel.rs +++ b/vnt/src/channel/channel.rs @@ -6,7 +6,6 @@ use std::sync::atomic::Ordering; use std::sync::Arc; use std::time::{Duration, Instant}; -use byte_pool::{Block, BytePool}; use crossbeam_epoch::{Atomic, Owned}; use crossbeam_utils::atomic::AtomicCell; use dashmap::DashMap; @@ -23,9 +22,6 @@ use crate::handle::recv_handler::ChannelDataHandler; use crate::handle::CurrentDeviceInfo; use crate::ip_proxy::DashMapNew; -lazy_static::lazy_static! { - static ref POOL:BytePool = BytePool::new(); -} pub struct ContextInner { //udp用于打洞、服务端通信(可选) pub(crate) main_channel: Arc, @@ -526,8 +522,10 @@ impl Context { pub fn update_read_time(&self, id: &Ipv4Addr, route_key: &RouteKey) { if let Some(mut time) = self.inner.route_table_time.get_mut(&(*route_key, *id)) { *time.value_mut() = Instant::now(); - }else{ - self.inner.route_table_time.insert((*route_key,*id),Instant::now()); + } else { + self.inner + .route_table_time + .insert((*route_key, *id), Instant::now()); } } } @@ -546,13 +544,13 @@ impl Channel { #[derive(Clone)] struct BufSenderGroup( usize, - Vec, usize, usize, RouteKey)>>, + Vec, usize, usize, RouteKey)>>, ); -struct BufReceiverGroup(Vec, usize, usize, RouteKey)>>); +struct BufReceiverGroup(Vec, usize, usize, RouteKey)>>); impl BufSenderGroup { - pub fn send(&mut self, val: (Block<'static>, usize, usize, RouteKey)) -> bool { + pub fn send(&mut self, val: (Vec, usize, usize, RouteKey)) -> bool { let index = self.0 % self.1.len(); self.0 = self.0.wrapping_add(1); self.1[index].send(val).is_ok() @@ -564,7 +562,7 @@ fn buf_channel_group(size: usize) -> (BufSenderGroup, BufReceiverGroup) { let mut buf_receiver_group = Vec::with_capacity(size); for _ in 0..size { let (buf_sender, buf_receiver) = - std::sync::mpsc::sync_channel::<(Block<'static, Vec>, usize, usize, RouteKey)>(1); + std::sync::mpsc::sync_channel::<(Vec, usize, usize, RouteKey)>(1); buf_sender_group.push(buf_sender); buf_receiver_group.push(buf_receiver); } @@ -598,8 +596,7 @@ impl Channel { .read_exact(&mut buf[head_reserve..head_reserve + len]) .await?; handler - .handle(&mut buf, head_reserve, head_reserve + len, key, &context) - .await; + .handle(&mut buf, head_reserve, head_reserve + len, key, &context); } } async fn start_tcp( @@ -686,19 +683,11 @@ impl Channel { let context = context.clone(); let handler = handler.clone(); std::thread::spawn(move || { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .unwrap(); - log::info!("启动异步处理"); - runtime.block_on(async move { - while let Ok((mut buf, start, end, route_key)) = buf_receiver.recv() { - handler - .handle(&mut buf, start, end, route_key, &context) - .await; - } - log::warn!("异步处理停止"); - }); + while let Ok((mut buf, start, end, route_key)) = buf_receiver.recv() { + handler + .handle(&mut buf, start, end, route_key, &context); + } + log::warn!("异步处理停止"); }); } Some(buf_sender) @@ -723,12 +712,8 @@ impl Channel { let handler = handler.clone(); let buf_sender = buf_sender.clone(); std::thread::spawn(move || { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .unwrap(); log::info!("启动udp v6"); - runtime.block_on(Self::main_start_( + Self::main_start_( worker, context, UDP_V6_ID, @@ -736,7 +721,7 @@ impl Channel { handler, buf_sender, head_reserve, - )); + ) }); } { @@ -746,12 +731,8 @@ impl Channel { let handler = handler.clone(); let buf_sender = buf_sender.clone(); std::thread::spawn(move || { - let runtime = tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .unwrap(); log::info!("启动udp v4"); - runtime.block_on(Self::main_start_( + Self::main_start_( worker, context, UDP_ID, @@ -759,7 +740,7 @@ impl Channel { handler, buf_sender, head_reserve, - )); + ) }); } if relay { @@ -813,7 +794,7 @@ impl Channel { } worker.stop_all(); } - async fn main_start_( + fn main_start_( worker: VntWorker, context: Context, id: usize, @@ -841,8 +822,7 @@ impl Channel { end, RouteKey::new(id, addr), &context, - ) - .await; + ); } Err(e) => { log::error!("udp :{:?}", e); @@ -851,7 +831,7 @@ impl Channel { } } Some(mut buf_sender) => loop { - let mut buf = POOL.alloc(4096); + let mut buf = vec![0; 4096]; match udp.recv_from(&mut buf[head_reserve..]) { Ok((len, addr)) => { let end = head_reserve + len; @@ -884,11 +864,11 @@ impl Channel { #[cfg(target_os = "windows")] use std::os::windows::io::AsRawSocket; #[cfg(target_os = "windows")] - let id = 3 + udp.as_raw_socket() as usize; + let id = 3 + udp.as_raw_socket() as usize; #[cfg(any(unix))] use std::os::fd::AsRawFd; #[cfg(any(unix))] - let id = 3 + udp.as_raw_fd() as usize; + let id = 3 + udp.as_raw_fd() as usize; context.insert_udp(id, udp.clone()); match buf_sender { @@ -899,7 +879,7 @@ impl Channel { rs=udp.recv_from(&mut buf[head_reserve..])=>{ match rs { Ok((len, addr)) => { - handler.handle(&mut buf, head_reserve, head_reserve + len, RouteKey::new(id, addr), &context).await; + handler.handle(&mut buf, head_reserve, head_reserve + len, RouteKey::new(id, addr), &context); } Err(e) => { log::error!("{:?}",e) @@ -933,7 +913,7 @@ impl Channel { } } Some(mut buf_sender) => loop { - let mut buf = POOL.alloc(4096); + let mut buf = vec![0; 4096]; tokio::select! { rs=udp.recv_from(&mut buf[head_reserve..])=>{ match rs { diff --git a/vnt/src/handle/recv_handler.rs b/vnt/src/handle/recv_handler.rs index f133a2a..71c2665 100644 --- a/vnt/src/handle/recv_handler.rs +++ b/vnt/src/handle/recv_handler.rs @@ -98,7 +98,7 @@ impl ChannelDataHandler { } impl ChannelDataHandler { - pub async fn handle( + pub fn handle( &self, buf: &mut [u8], start: usize, @@ -107,14 +107,14 @@ impl ChannelDataHandler { context: &Context, ) { assert_eq!(start, 14); - match self.handle0(&mut buf[..end], &route_key, context).await { + match self.handle0(&mut buf[..end], &route_key, context) { Ok(_) => {} Err(e) => { log::warn!("{:?}", e); } } } - async fn handle0( + fn handle0( &self, buf: &mut [u8], route_key: &RouteKey, @@ -173,8 +173,7 @@ impl ChannelDataHandler { //服务端解密 self.server_cipher.decrypt_ipv4(&mut net_packet)?; let data_len = net_packet.data_len(); - self.server_packet_handle(context, current_device, buf, data_len, route_key) - .await?; + self.server_packet_handle(context, current_device, buf, data_len, route_key)?; } return Ok(()); } @@ -313,12 +312,10 @@ impl ChannelDataHandler { Protocol::Service => {} Protocol::Error => {} Protocol::Control => { - self.control(context, current_device, source, net_packet, route_key) - .await?; + self.control(context, current_device, source, net_packet, route_key)?; } Protocol::OtherTurn => { - self.other_turn(context, current_device, source, net_packet, route_key) - .await?; + self.other_turn(context, current_device, source, net_packet, route_key)?; } Protocol::UnKnow(e) => { log::info!("不支持的协议:{}", e); @@ -327,7 +324,7 @@ impl ChannelDataHandler { Ok(()) } - async fn pong_packet( + fn pong_packet( &self, gateway: bool, metric: u8, @@ -361,7 +358,7 @@ impl ChannelDataHandler { } Ok(()) } - async fn control( + fn control( &self, context: &Context, current_device: CurrentDeviceInfo, @@ -390,8 +387,7 @@ impl ChannelDataHandler { source, pong_packet, route_key, - ) - .await?; + )?; } ControlPacket::PunchRequest => { if self.relay { @@ -437,7 +433,7 @@ impl ChannelDataHandler { } Ok(()) } - async fn other_turn( + fn other_turn( &self, context: &Context, current_device: CurrentDeviceInfo, @@ -518,12 +514,12 @@ impl ChannelDataHandler { // let _ = context.try_send_main_udp(packet.buffer(), // SocketAddr::V4(SocketAddrV4::new(peer_nat_info.local_ip, peer_nat_info.local_port))); // } - if self.punch(source, peer_nat_info).await { + if self.punch(source, peer_nat_info) { self.client_cipher.encrypt_ipv4(&mut punch_packet)?; context.try_send_by_key(punch_packet.buffer(), route_key)?; } } else { - self.punch(source, peer_nat_info).await; + self.punch(source, peer_nat_info); } } other_turn_packet::Protocol::Unknown(e) => { @@ -532,7 +528,7 @@ impl ChannelDataHandler { } Ok(()) } - async fn punch(&self, peer_ip: Ipv4Addr, peer_nat_info: NatInfo) -> bool { + fn punch(&self, peer_ip: Ipv4Addr, peer_nat_info: NatInfo) -> bool { match peer_nat_info.nat_type { NatType::Symmetric => self .symmetric_sender @@ -545,7 +541,7 @@ impl ChannelDataHandler { /// 处理服务端数据 impl ChannelDataHandler { - async fn server_packet_handle( + fn server_packet_handle( &self, context: &Context, current_device: CurrentDeviceInfo, @@ -557,16 +553,13 @@ impl ChannelDataHandler { let source = net_packet.source(); match net_packet.protocol() { Protocol::Service => { - self.service(context, current_device, net_packet, route_key) - .await?; + self.service(context, current_device, net_packet, route_key)?; } Protocol::Error => { - self.error(context, current_device, source, net_packet, route_key) - .await?; + self.error(context, current_device, source, net_packet, route_key)?; } Protocol::Control => { - self.control_gateway(context, current_device, net_packet, route_key) - .await?; + self.control_gateway(context, current_device, net_packet, route_key)?; } Protocol::IpTurn => { match ip_turn_packet::Protocol::from(net_packet.transport_protocol()) { @@ -600,7 +593,7 @@ impl ChannelDataHandler { } return Ok(()); } - async fn control_gateway( + fn control_gateway( &self, context: &Context, current_device: CurrentDeviceInfo, @@ -618,8 +611,7 @@ impl ChannelDataHandler { net_packet.source(), pong_packet, route_key, - ) - .await?; + )?; } ControlPacket::AddrResponse(addr_packet) => self .nat_test @@ -628,7 +620,7 @@ impl ChannelDataHandler { } Ok(()) } - async fn service( + fn service( &self, context: &Context, current_device: CurrentDeviceInfo, @@ -643,20 +635,24 @@ impl ChannelDataHandler { { let context = context.clone(); let nat_test = self.nat_test.clone(); - tokio::spawn(async move { - let local_port = context.main_local_ipv4_port().unwrap_or(0); - let local_ipv4_addr = nat::local_ipv4_addr(local_port); - let local_port = context.main_local_ipv6_port().unwrap_or(0); - let ipv6_addr = nat::local_ipv6_addr(local_port); - let nat_info = nat_test - .re_test( - Ipv4Addr::from(response.public_ip), - response.public_port as u16, - local_ipv4_addr, - ipv6_addr, - ) - .await; - context.switch(nat_info.nat_type); + std::thread::spawn(move ||{ + tokio::runtime::Builder::new_current_thread() + .enable_all().build().unwrap() + .block_on(async move { + let local_port = context.main_local_ipv4_port().unwrap_or(0); + let local_ipv4_addr = nat::local_ipv4_addr(local_port); + let local_port = context.main_local_ipv6_port().unwrap_or(0); + let ipv6_addr = nat::local_ipv6_addr(local_port); + let nat_info = nat_test + .re_test( + Ipv4Addr::from(response.public_ip), + response.public_port as u16, + local_ipv4_addr, + ipv6_addr, + ) + .await; + context.switch(nat_info.nat_type); + }) }); } let new_ip = Ipv4Addr::from(response.virtual_ip); @@ -728,7 +724,7 @@ impl ChannelDataHandler { } Ok(()) } - async fn error( + fn error( &self, _context: &Context, current_device: CurrentDeviceInfo, diff --git a/vnt/src/handle/tun_tap/channel_group.rs b/vnt/src/handle/tun_tap/channel_group.rs index 2974c24..10160e0 100644 --- a/vnt/src/handle/tun_tap/channel_group.rs +++ b/vnt/src/handle/tun_tap/channel_group.rs @@ -1,15 +1,13 @@ -use byte_pool::Block; - #[derive(Clone)] pub struct BufSenderGroup( usize, - Vec, usize, usize)>>, + Vec, usize, usize)>>, ); -pub struct BufReceiverGroup(pub Vec, usize, usize)>>); +pub struct BufReceiverGroup(pub Vec, usize, usize)>>); impl BufSenderGroup { - pub fn send(&mut self, val: (Block<'static>, usize, usize)) -> bool { + pub fn send(&mut self, val: (Vec, usize, usize)) -> bool { let index = self.0 % self.1.len(); self.0 = self.0.wrapping_add(1); self.1[index].send(val).is_ok() @@ -21,7 +19,7 @@ pub fn buf_channel_group(size: usize) -> (BufSenderGroup, BufReceiverGroup) { let mut buf_receiver_group = Vec::with_capacity(size); for _ in 0..size { let (buf_sender, buf_receiver) = - std::sync::mpsc::sync_channel::<(Block<'static>, usize, usize)>(1); + std::sync::mpsc::sync_channel::<(Vec, usize, usize)>(1); buf_sender_group.push(buf_sender); buf_receiver_group.push(buf_receiver); } diff --git a/vnt/src/handle/tun_tap/tap_handler.rs b/vnt/src/handle/tun_tap/tap_handler.rs index fd50d67..9dd6d7f 100644 --- a/vnt/src/handle/tun_tap/tap_handler.rs +++ b/vnt/src/handle/tun_tap/tap_handler.rs @@ -1,9 +1,7 @@ -use byte_pool::BytePool; use std::sync::Arc; use std::{io, thread}; use crossbeam_utils::atomic::AtomicCell; -use lazy_static::lazy_static; use packet::arp::arp::ArpPacket; use packet::ethernet; @@ -22,9 +20,6 @@ use crate::handle::CurrentDeviceInfo; use crate::igmp_server::IgmpServer; use crate::ip_proxy::IpProxyMap; use crate::tun_tap_device::{DeviceReader, DeviceWriter}; -lazy_static! { - static ref POOL: BytePool> = BytePool::>::new(); -} pub fn start( worker: VntWorker, @@ -116,7 +111,7 @@ fn start_( mut buf_sender: BufSenderGroup, ) -> io::Result<()> { loop { - let mut buf = POOL.alloc(4096); + let mut buf = vec![0; 4096]; if sender.is_close() { return Ok(()); } diff --git a/vnt/src/handle/tun_tap/tun_handler.rs b/vnt/src/handle/tun_tap/tun_handler.rs index 177fe80..60681a5 100644 --- a/vnt/src/handle/tun_tap/tun_handler.rs +++ b/vnt/src/handle/tun_tap/tun_handler.rs @@ -1,4 +1,3 @@ -use byte_pool::BytePool; use std::sync::Arc; use std::{io, thread}; @@ -19,9 +18,6 @@ use crate::handle::CurrentDeviceInfo; use crate::igmp_server::IgmpServer; use crate::ip_proxy::IpProxyMap; use crate::tun_tap_device::{DeviceReader, DeviceWriter}; -lazy_static::lazy_static! { - static ref POOL:BytePool> = BytePool::>::new(); -} fn icmp(device_writer: &DeviceWriter, mut ipv4_packet: IpV4Packet<&mut [u8]>) -> Result<()> { if ipv4_packet.protocol() == ipv4::protocol::Protocol::Icmp { let mut icmp = IcmpPacket::new(ipv4_packet.payload_mut())?; @@ -169,7 +165,7 @@ fn start_( mut buf_sender: BufSenderGroup, ) -> io::Result<()> { loop { - let mut buf = POOL.alloc(4096); + let mut buf = vec![0; 4096]; buf[..12].fill(0); if sender.is_close() { return Ok(()); diff --git a/vnt/src/ip_proxy/tcp_proxy.rs b/vnt/src/ip_proxy/tcp_proxy.rs index f9dc750..787d419 100644 --- a/vnt/src/ip_proxy/tcp_proxy.rs +++ b/vnt/src/ip_proxy/tcp_proxy.rs @@ -5,8 +5,8 @@ use std::sync::Arc; use std::time::Duration; use tokio::io::AsyncReadExt; use tokio::io::AsyncWriteExt; -use tokio::net::{TcpListener, TcpStream}; use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; +use tokio::net::{TcpListener, TcpStream}; pub struct TcpProxy { tcp_listener: TcpListener, @@ -39,7 +39,7 @@ impl TcpProxy { Duration::from_secs(5), TcpStream::connect(dest_addr), ) - .await + .await { Ok(peer_tcp_stream) => match peer_tcp_stream { Ok(peer_tcp_stream) => peer_tcp_stream, @@ -86,7 +86,7 @@ async fn proxy(client: TcpStream, server: TcpStream) -> io::Result<()> { let (server_read, server_write) = server.into_split(); tokio::spawn(async move { if let Err(e) = copy(client_read, server_write).await { - log::warn!("{:?}",e); + log::warn!("{:?}", e); } }); copy(server_read, client_write).await