commit:

1.去除缓冲池
2.数据处理改为同步方法
3.fmt

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)

common/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "common"
-version = "1.2.3"
+version = "1.2.4"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

vnt-cli/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "vnt-cli"
-version = "1.2.3"
+version = "1.2.4"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

vnt-cli/README.md

@@ -36,6 +36,11 @@
 
 ### -W
 开启和服务端通信的数据加密，采用rsa+aes256gcm加密客户端和服务端之间通信的数据，可以避免token泄漏、中间人攻击
+
+注意：
+1. -w `<password>`是用于客户端-客户端之间的加密，password不会传递到服务端，只添加这个参数不会加密客户端-服务端通信的数据
+2. -W 用于开启客户端-服务端之间的加密
+
 ### -m
 模拟组播，高频使用组播通信时，可以尝试开启此参数，默认情况下会把组播当作广播发给所有节点
 
@@ -65,8 +70,11 @@
 | 1~8位  | aes_ecb | AES128-ECB |
 | `>=`8 | aes_ecb | AES256-ECB |
 ### --finger 
+
 开启数据指纹校验，可增加安全性，如果服务端开启指纹校验，则客户端也必须开启，开启会损耗一部分性能
 
+注意：默认情况下服务端不会对中转的数据做校验，如果要对中转的数据做校验，则需要客户端、服务端都开启此参数
+
 ### --relay
 禁用p2p,在网络环境很差时，只使用服务器中转效果可能更好（可以配合--tcp参数一起使用）
 ### --list

vnt-cli/src/main.rs

@@ -214,10 +214,13 @@ fn main() {
         return;
     }
 
-    let cipher_model = matches
-        .opt_get::<CipherModel>("model")
-        .unwrap()
-        .unwrap_or(CipherModel::AesGcm);
+    let cipher_model = match matches.opt_get::<CipherModel>("model") {
+        Ok(model) => model.unwrap_or(CipherModel::AesGcm),
+        Err(e) => {
+            println!("'--model ' invalid,{}", e);
+            return;
+        }
+    };
 
     let finger = matches.opt_present("finger");
     let punch_model = matches
@@ -250,6 +253,7 @@ fn main() {
     main0(config, !unused_cmd);
     std::process::exit(0);
 }
+
 #[tokio::main]
 async fn main0(config: Config, show_cmd: bool) {
     let server_encrypt = config.server_encrypt;

vnt-jni/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "vnt-jni"
-version = "1.2.3"
+version = "1.2.4"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

vnt/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "vnt"
-version = "1.2.3"
+version = "1.2.4"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -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"

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<StdUdpSocket>,
@@ -526,6 +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());
         }
     }
 }
@@ -544,13 +544,13 @@ impl Channel {
 #[derive(Clone)]
 struct BufSenderGroup(
     usize,
-    Vec<std::sync::mpsc::SyncSender<(Block<'static>, usize, usize, RouteKey)>>,
+    Vec<std::sync::mpsc::SyncSender<(Vec<u8>, usize, usize, RouteKey)>>,
 );
 
-struct BufReceiverGroup(Vec<std::sync::mpsc::Receiver<(Block<'static>, usize, usize, RouteKey)>>);
+struct BufReceiverGroup(Vec<std::sync::mpsc::Receiver<(Vec<u8>, usize, usize, RouteKey)>>);
 
 impl BufSenderGroup {
-    pub fn send(&mut self, val: (Block<'static>, usize, usize, RouteKey)) -> bool {
+    pub fn send(&mut self, val: (Vec<u8>, 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()
@@ -562,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<u8>>, usize, usize, RouteKey)>(1);
+            std::sync::mpsc::sync_channel::<(Vec<u8>, usize, usize, RouteKey)>(1);
         buf_sender_group.push(buf_sender);
         buf_receiver_group.push(buf_receiver);
     }
@@ -596,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(
@@ -684,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)
@@ -721,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,
@@ -734,7 +721,7 @@ impl Channel {
                     handler,
                     buf_sender,
                     head_reserve,
-                ));
+                )
             });
         }
         {
@@ -744,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,
@@ -757,7 +740,7 @@ impl Channel {
                     handler,
                     buf_sender,
                     head_reserve,
-                ));
+                )
             });
         }
         if relay {
@@ -811,7 +794,7 @@ impl Channel {
         }
         worker.stop_all();
     }
-    async fn main_start_(
+    fn main_start_(
         worker: VntWorker,
         context: Context,
         id: usize,
@@ -839,8 +822,7 @@ impl Channel {
                                     end,
                                     RouteKey::new(id, addr),
                                     &context,
-                                )
-                                .await;
+                                );
                         }
                         Err(e) => {
                             log::error!("udp :{:?}", e);
@@ -849,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;
@@ -882,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 {
@@ -897,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)
@@ -931,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 {

vnt/src/channel/punch.rs

@@ -50,9 +50,12 @@ impl NatInfo {
         public_port_range: u16,
         local_ipv4_addr: SocketAddrV4,
         ipv6_addr: SocketAddrV6,
-        nat_type: NatType,
+        mut nat_type: NatType,
     ) -> Self {
-        public_ips.retain(|ip| !ip.is_loopback() && !ip.is_private());
+        public_ips.retain(|ip| !ip.is_loopback() && !ip.is_private() && !ip.is_unspecified());
+        if public_ips.len() > 1 {
+            nat_type = NatType::Symmetric;
+        }
         Self {
             public_ips,
             public_port,

vnt/src/cipher/cipher.rs

@@ -28,7 +28,7 @@ impl FromStr for CipherModel {
             "aes_gcm" => Ok(CipherModel::AesGcm),
             "aes_cbc" => Ok(CipherModel::AesCbc),
             "aes_ecb" => Ok(CipherModel::AesEcb),
-            _ => Err(format!("not match '{}'", s)),
+            _ => Err(format!("not match '{}', enum:aes_gcm/aes_cbc/aes_ecb", s)),
         }
     }
 }

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 {
@@ -431,22 +427,13 @@ impl ChannelDataHandler {
                 }
                 std::net::IpAddr::V6(_) => {}
             },
-            ControlPacket::AddrResponse(addr_packet) => {
-                if !addr_packet.ipv4().is_multicast()
-                    && !addr_packet.ipv4().is_broadcast()
-                    && !addr_packet.ipv4().is_unspecified()
-                    && !addr_packet.ipv4().is_loopback()
-                    && !addr_packet.ipv4().is_private()
-                    && addr_packet.port() != 0
-                {
-                    self.nat_test
-                        .update_addr(addr_packet.ipv4(), addr_packet.port())
-                }
-            }
+            ControlPacket::AddrResponse(addr_packet) => self
+                .nat_test
+                .update_addr(addr_packet.ipv4(), addr_packet.port()),
         }
         Ok(())
     }
-    async fn other_turn(
+    fn other_turn(
         &self,
         context: &Context,
         current_device: CurrentDeviceInfo,
@@ -527,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) => {
@@ -541,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
@@ -554,7 +541,7 @@ impl ChannelDataHandler {
 
 /// 处理服务端数据
 impl ChannelDataHandler {
-    async fn server_packet_handle(
+    fn server_packet_handle(
         &self,
         context: &Context,
         current_device: CurrentDeviceInfo,
@@ -566,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()) {
@@ -609,7 +593,7 @@ impl ChannelDataHandler {
         }
         return Ok(());
     }
-    async fn control_gateway(
+    fn control_gateway(
         &self,
         context: &Context,
         current_device: CurrentDeviceInfo,
@@ -627,26 +611,16 @@ impl ChannelDataHandler {
                     net_packet.source(),
                     pong_packet,
                     route_key,
-                )
-                .await?;
-            }
-            ControlPacket::AddrResponse(addr_packet) => {
-                if addr_packet.port() != 0
-                    && !addr_packet.ipv4().is_multicast()
-                    && !addr_packet.ipv4().is_broadcast()
-                    && !addr_packet.ipv4().is_unspecified()
-                    && !addr_packet.ipv4().is_loopback()
-                    && !addr_packet.ipv4().is_private()
-                {
-                    self.nat_test
-                        .update_addr(addr_packet.ipv4(), addr_packet.port())
-                }
+                )?;
             }
+            ControlPacket::AddrResponse(addr_packet) => self
+                .nat_test
+                .update_addr(addr_packet.ipv4(), addr_packet.port()),
             _ => {}
         }
         Ok(())
     }
-    async fn service(
+    fn service(
         &self,
         context: &Context,
         current_device: CurrentDeviceInfo,
@@ -661,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);
@@ -746,7 +724,7 @@ impl ChannelDataHandler {
         }
         Ok(())
     }
-    async fn error(
+    fn error(
         &self,
         _context: &Context,
         current_device: CurrentDeviceInfo,

vnt/src/handle/tun_tap/channel_group.rs

@@ -1,15 +1,13 @@
-use byte_pool::Block;
-
 #[derive(Clone)]
 pub struct BufSenderGroup(
     usize,
-    Vec<std::sync::mpsc::SyncSender<(Block<'static>, usize, usize)>>,
+    Vec<std::sync::mpsc::SyncSender<(Vec<u8>, usize, usize)>>,
 );
 
-pub struct BufReceiverGroup(pub Vec<std::sync::mpsc::Receiver<(Block<'static>, usize, usize)>>);
+pub struct BufReceiverGroup(pub Vec<std::sync::mpsc::Receiver<(Vec<u8>, usize, usize)>>);
 
 impl BufSenderGroup {
-    pub fn send(&mut self, val: (Block<'static>, usize, usize)) -> bool {
+    pub fn send(&mut self, val: (Vec<u8>, 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<u8>, usize, usize)>(1);
         buf_sender_group.push(buf_sender);
         buf_receiver_group.push(buf_receiver);
     }

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<Vec<u8>> = BytePool::<Vec<u8>>::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(());
         }

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<Vec<u8>> = BytePool::<Vec<u8>>::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(());

vnt/src/ip_proxy/tcp_proxy.rs

@@ -3,7 +3,9 @@ use std::io;
 use std::net::{SocketAddr, SocketAddrV4};
 use std::sync::Arc;
 use std::time::Duration;
-
+use tokio::io::AsyncReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
 use tokio::net::{TcpListener, TcpStream};
 
 pub struct TcpProxy {
@@ -37,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,
@@ -79,15 +81,32 @@ impl TcpProxy {
     }
 }
 
-async fn proxy(mut client: TcpStream, mut server: TcpStream) -> io::Result<()> {
-    let (mut client_reader, mut client_writer) = client.split();
-    let (mut server_reader, mut server_writer) = server.split();
+async fn proxy(client: TcpStream, server: TcpStream) -> io::Result<()> {
+    let (client_read, client_write) = client.into_split();
+    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);
+        }
+    });
+    copy(server_read, client_write).await
+}
 
-    let client_to_server = tokio::io::copy(&mut client_reader, &mut server_writer);
-    let server_to_client = tokio::io::copy(&mut server_reader, &mut client_writer);
-    tokio::select! {
-        _ = tokio::time::timeout(Duration::from_secs(10), client_to_server) =>{},
-        _ = tokio::time::timeout(Duration::from_secs(10), server_to_client) =>{},
+async fn copy(mut read: OwnedReadHalf, mut write: OwnedWriteHalf) -> io::Result<()> {
+    let mut buf = [0; 10240];
+    loop {
+        tokio::select! {
+            result = read.read(&mut buf) =>{
+                let len = result?;
+                if len==0{
+                    break;
+                }
+                write.write_all(&buf[..len]).await?;
+            }
+            _ = tokio::time::sleep(Duration::from_secs(300)) =>{
+                break;
+            }
+        }
     }
     Ok(())
 }

vnt/src/lib.rs

@@ -1,5 +1,5 @@
 use crate::error::Error;
-pub const VNT_VERSION: &'static str = "1.2.3";
+pub const VNT_VERSION: &'static str = "1.2.4";
 pub type Result<T> = std::result::Result<T, Error>;
 
 pub mod channel;

vnt/src/nat/mod.rs

@@ -99,10 +99,18 @@ impl NatTest {
         self.info.lock().clone()
     }
     pub fn update_addr(&self, ip: Ipv4Addr, port: u16) {
-        let mut guard = self.info.lock();
-        guard.public_port = port;
-        if !guard.public_ips.contains(&ip) {
-            guard.public_ips.push(ip);
+        if !ip.is_multicast()
+            && !ip.is_broadcast()
+            && !ip.is_unspecified()
+            && !ip.is_loopback()
+            && !ip.is_private()
+            && port != 0
+        {
+            let mut guard = self.info.lock();
+            guard.public_port = port;
+            if !guard.public_ips.contains(&ip) {
+                guard.public_ips.push(ip);
+            }
         }
     }
     pub async fn re_test(
@@ -131,13 +139,9 @@ impl NatTest {
         ipv6_addr: SocketAddrV6,
     ) -> NatInfo {
         return match stun_test::stun_test_nat(stun_server.clone()).await {
-            Ok((nat_type, ips, port_range)) => {
-                let mut public_ips = Vec::new();
-                public_ips.push(Ipv4Addr::from(public_ip));
-                for ip in ips {
-                    if ip != public_ip {
-                        public_ips.push(ip);
-                    }
+            Ok((nat_type, mut public_ips, port_range)) => {
+                if !public_ips.contains(&public_ip) {
+                    public_ips.push(public_ip)
                 }
                 NatInfo::new(
                     public_ips,