1.修改默认任务数

2.修改指纹生成方式

vnt-cli/src/main.rs

@@ -195,7 +195,7 @@ fn main() {
     }
     let tcp_channel = matches.opt_present("tcp");
     let relay = matches.opt_present("relay");
-    let parallel = matches.opt_get::<usize>("par").unwrap().unwrap_or(2);
+    let parallel = matches.opt_get::<usize>("par").unwrap().unwrap_or(1);
     if parallel == 0 {
         println!("--par invalid");
         return;
@@ -213,11 +213,11 @@ fn main() {
                              out_ip, password, simulate_multicast, mtu,
                              tcp_channel, virtual_ip, relay, server_encrypt, parallel);
     let runtime = tokio::runtime::Builder::new_multi_thread().enable_all().worker_threads(thread_num).build().unwrap();
-    runtime.block_on(main0(config,!unused_cmd));
+    runtime.block_on(main0(config, !unused_cmd));
     std::process::exit(0);
 }
 
-async fn main0(config: Config,show_cmd:bool) {
+async fn main0(config: Config, show_cmd: bool) {
     let server_encrypt = config.server_encrypt;
     let mut vnt_util = VntUtil::new(config).await.unwrap();
     let mut conn_count = 0;
@@ -452,7 +452,7 @@ fn print_usage(program: &str, _opts: Options) {
     println!("  --tcp               和服务端使用tcp通信,默认使用udp,遇到udp qos时可指定使用tcp");
     println!("  --ip <ip>           指定虚拟ip,指定的ip不能和其他设备重复,必须有效并且在服务端所属网段下,默认情况由服务端分配");
     println!("  --relay             仅使用服务器转发,不使用p2p,默认情况允许使用p2p");
-    println!("  --par <parallel>    任务并行度(必须为正整数),默认值为2");
+    println!("  --par <parallel>    任务并行度(必须为正整数),默认值为1");
     println!("  --thread <thread>   线程数(必须为正整数),默认为核心数乘2");
     println!();
     println!("  --list              {}", yellow("后台运行时,查看其他设备列表".to_string()));

vnt/src/channel/channel.rs

@@ -444,22 +444,28 @@ impl Channel {
                        relay: bool,
                        parallel: usize,
     ) {
-        let (buf_sender, buf_receiver) = buf_channel_group(parallel);
+        let handler = self.handler.clone();
-        for mut buf_receiver in buf_receiver.0 {
-            let context = self.context.clone();
-            let handler = self.handler.clone();
-            tokio::spawn(async move {
-                while let Some((mut buf, start, end, route_key)) = buf_receiver.recv().await {
-                    handler.handle(&mut buf, start, end, route_key, &context).await;
-                }
-            });
-        }
         let context = self.context;
         let main_channel = context.inner.main_channel.clone();
+        let buf_sender = if parallel > 1 || tcp.is_some() {
+            let (buf_sender, buf_receiver) = buf_channel_group(parallel);
+            for mut buf_receiver in buf_receiver.0 {
+                let context = context.clone();
+                let handler = handler.clone();
+                tokio::spawn(async move {
+                    while let Some((mut buf, start, end, route_key)) = buf_receiver.recv().await {
+                        handler.handle(&mut buf, start, end, route_key, &context).await;
+                    }
+                });
+            }
+            Some(buf_sender)
+        } else {
+            None
+        };
         if let Some((tcp_stream, receiver)) = tcp {
-            tokio::spawn(Self::start_tcp(worker.worker("main_channel_tcp"), tcp_stream, receiver, context.inner.current_device.clone(), buf_sender.clone(), head_reserve));
+            tokio::spawn(Self::start_tcp(worker.worker("main_channel_tcp"), tcp_stream, receiver, context.inner.current_device.clone(), buf_sender.clone().unwrap(), head_reserve));
         }
-        tokio::spawn(Self::start_(worker.worker("main_channel_1"), context.clone(), main_channel.clone(), buf_sender.clone(), head_reserve, true));
+        tokio::spawn(Self::start_(worker.worker("main_channel_1"), context.clone(), main_channel.clone(), handler.clone(), buf_sender.clone(), head_reserve, true));
         if relay {
             worker.stop_wait().await;
             return;
@@ -489,7 +495,7 @@ impl Channel {
                                             Ok(udp) => {
                                                 let udp = Arc::new(udp);
                                                 let context = context.clone();
-                                                tokio::spawn(Self::start_(worker.worker("symmetric_channel"),context, udp,buf_sender.clone(), head_reserve, false));
+                                                tokio::spawn(Self::start_(worker.worker("symmetric_channel"),context, udp,handler.clone(),buf_sender.clone(), head_reserve, false));
                                             }
                                             Err(e) => {
                                                 log::error!("{}",e);
@@ -513,7 +519,8 @@ impl Channel {
     }
     async fn start_(mut worker: VntWorker, context: Context,
                     udp: Arc<UdpSocket>,
-                    mut buf_sender: BufSenderGroup,
+                    handler: ChannelDataHandler,
+                    buf_sender: Option<BufSenderGroup>,
                     head_reserve: usize,
                     is_core: bool) {
         let mut status_receiver = context.inner.status_receiver.clone();
@@ -526,44 +533,88 @@ impl Channel {
         #[cfg(any(unix))]
             let id = 1 + udp.as_raw_fd() as usize;
         context.inner.udp_map.insert(id, udp.clone());
-        loop {
+        match buf_sender {
-            let mut buf = POOL.alloc(4096);
+            None => {
-            tokio::select! {
+                let mut buf = [0; 4096];
-                rs=udp.recv_from(&mut buf[head_reserve..])=>{
+                loop {
-                     match rs {
+                    tokio::select! {
-                        Ok((len, addr)) => {
+                        rs=udp.recv_from(&mut buf[head_reserve..])=>{
-                            if !buf_sender.send((buf,head_reserve,head_reserve+len,RouteKey::new(id, addr))).await{
+                              match rs {
-                                 log::error!("udp buf_sender发送数据失败");
+                                Ok((len, addr)) => {
-                                 break;
+                                    handler.handle(&mut buf, head_reserve, head_reserve + len, RouteKey::new(id, addr), &context).await;
+                                }
+                                Err(e) => {
+                                    log::error!("{:?}",e)
+                                }
                             }
                         }
-                        Err(e) => {
+                        changed=status_receiver.changed()=>{
-                            log::error!("{:?}",e)
+                                match changed {
+                                    Ok(_) => {
+                                        match *status_receiver.borrow() {
+                                            Status::Cone => {
+                                                if !is_core{
+                                                    break;
+                                                }
+                                            }
+                                            Status::Close=>{
+                                                break;
+                                            }
+                                            Status::Symmetric => {}
+                                        }
+                                    }
+                                    Err(_) => {
+                                        break;
+                                    }
+                                }
+                        }
+                        _=worker.stop_wait()=>{
+                            break;
                         }
                     }
                 }
-                changed=status_receiver.changed()=>{
+            }
-                        match changed {
+            Some(mut buf_sender) => {
-                            Ok(_) => {
+                loop {
-                                match *status_receiver.borrow() {
+                    let mut buf = POOL.alloc(4096);
-                                    Status::Cone => {
+                    tokio::select! {
-                                        if !is_core{
+                        rs=udp.recv_from(&mut buf[head_reserve..])=>{
-                                            break;
+                             match rs {
-                                        }
+                                Ok((len, addr)) => {
+                                    if !buf_sender.send((buf,head_reserve,head_reserve+len,RouteKey::new(id, addr))).await{
+                                         log::error!("udp buf_sender发送数据失败");
+                                         break;
                                     }
-                                    Status::Close=>{
+                                }
-                                        break;
+                                Err(e) => {
-                                    }
+                                    log::error!("{:?}",e)
-                                    Status::Symmetric => {}
                                 }
                             }
-                            Err(_) => {
-                                break;
-                            }
                         }
-                }
+                        changed=status_receiver.changed()=>{
-                _=worker.stop_wait()=>{
+                                match changed {
-                    break;
+                                    Ok(_) => {
+                                        match *status_receiver.borrow() {
+                                            Status::Cone => {
+                                                if !is_core{
+                                                    break;
+                                                }
+                                            }
+                                            Status::Close=>{
+                                                break;
+                                            }
+                                            Status::Symmetric => {}
+                                        }
+                                    }
+                                    Err(_) => {
+                                        break;
+                                    }
+                                }
+                        }
+                        _=worker.stop_wait()=>{
+                            break;
+                        }
+                    }
                 }
             }
         }

vnt/src/cipher/aes_c.rs

@@ -3,7 +3,7 @@ use crate::cipher::Finger;
 use crate::protocol::NetPacket;
 use sha2::Digest;
 #[cfg(feature = "ring-cipher")]
-use crate::cipher::ring_cipher::AesGcmCipher;
+use crate::cipher::ring_aes_gcm_cipher::AesGcmCipher;
 #[cfg(not(feature = "ring-cipher"))]
 use crate::cipher::aes_gcm_cipher::AesGcmCipher;
 
@@ -15,7 +15,7 @@ pub enum Cipher {
 
 impl Cipher {
     pub fn new_password(password: Option<String>, token: String) -> Self {
-        let finger = Finger::new(token);
+        let finger = Finger::new(&token);
         if let Some(password) = password {
             let mut hasher = sha2::Sha256::new();
             hasher.update(password.as_bytes());
@@ -32,7 +32,7 @@ impl Cipher {
         }
     }
     pub fn new_key(key: [u8; 32], token: String) -> io::Result<Self> {
-        let finger = Finger::new(token);
+        let finger = Finger::new(&token);
         match key.len() {
             16 => {
                 let aes = AesGcmCipher::new_128(key[..16].try_into().unwrap(), finger);

vnt/src/cipher/aes_gcm_cipher.rs

@@ -58,10 +58,7 @@ impl AesGcmCipher {
 
         let mut secret_body = SecretBody::new(net_packet.payload_mut())?;
         let tag = secret_body.tag();
-        if tag.len() != 16 {
+        let finger = self.finger.calculate_finger(&nonce_raw, secret_body.en_body());
-            return Err(io::Error::new(io::ErrorKind::Other, "tag err"));
-        }
-        let finger = self.finger.calculate_finger(&nonce_raw, &secret_body);
         if &finger != secret_body.finger() {
             return Err(io::Error::new(io::ErrorKind::Other, "finger err"));
         }
@@ -102,7 +99,7 @@ impl AesGcmCipher {
         return match rs {
             Ok(tag) => {
                 secret_body.set_tag(tag.as_slice())?;
-                let finger = self.finger.calculate_finger(&nonce_raw, &secret_body);
+                let finger = self.finger.calculate_finger(&nonce_raw, secret_body.en_body());
                 secret_body.set_finger(&finger)?;
                 net_packet.set_encrypt_flag(true);
                 Ok(())

vnt/src/cipher/finger.rs

@@ -2,23 +2,27 @@ use std::io;
 
 use sha2::Digest;
 
-use crate::protocol::{body::ENCRYPTION_RESERVED, body::SecretBody, NetPacket};
+use crate::protocol::{body::ENCRYPTION_RESERVED, NetPacket};
 
 #[derive(Clone)]
 pub struct Finger {
-    token: String,
+    hash: [u8; 32],
 }
 
 impl Finger {
-    pub fn new(token: String) -> Self {
+    pub fn new(str: &str) -> Self {
-        Finger { token }
+        let mut hasher = sha2::Sha256::new();
+        hasher.update(str.as_bytes());
+        let hash: [u8; 32] = hasher.finalize().into();
+        Finger { hash }
     }
     pub fn check_finger<B: AsRef<[u8]>>(&self, net_packet: &NetPacket<B>) -> io::Result<()> {
         if !net_packet.is_encrypt() {
             //未加密的数据直接丢弃
             return Err(io::Error::new(io::ErrorKind::Other, "not encrypt"));
         }
-        if net_packet.payload().len() < ENCRYPTION_RESERVED {
+        let payload_len = net_packet.payload().len();
+        if payload_len < ENCRYPTION_RESERVED {
             log::error!("数据异常,长度小于{}",ENCRYPTION_RESERVED);
             return Err(io::Error::new(io::ErrorKind::Other, "data err"));
         }
@@ -27,19 +31,20 @@ impl Finger {
         nonce_raw[4..8].copy_from_slice(&net_packet.destination().octets());
         nonce_raw[8] = net_packet.protocol().into();
         nonce_raw[9] = net_packet.transport_protocol();
-        let secret_body = SecretBody::new(net_packet.payload())?;
+        nonce_raw[10] = net_packet.is_gateway() as u8;
-        let finger = self.calculate_finger(&nonce_raw, &secret_body);
+        nonce_raw[11] = net_packet.source_ttl();
-        if &finger != secret_body.finger() {
+        let payload = net_packet.payload();
+        let finger = self.calculate_finger(&nonce_raw, &payload[..payload_len - 12]);
+        if &finger[..] != &payload[payload_len - 12..] {
             return Err(io::Error::new(io::ErrorKind::Other, "finger err"));
         }
         Ok(())
     }
-    pub fn calculate_finger<B: AsRef<[u8]>>(&self, nonce_raw: &[u8; 12], secret_body: &SecretBody<B>) -> [u8; 12] {
+    pub fn calculate_finger(&self, nonce_raw: &[u8; 12], secret_body: &[u8]) -> [u8; 12] {
         let mut hasher = sha2::Sha256::new();
-        hasher.update(secret_body.body());
         hasher.update(nonce_raw);
-        hasher.update(secret_body.tag());
+        hasher.update(secret_body);
-        hasher.update(&self.token);
+        hasher.update(&self.hash);
         let key: [u8; 32] = hasher.finalize().into();
         return key[20..].try_into().unwrap();
     }

vnt/src/cipher/mod.rs

@@ -1,5 +1,5 @@
 #[cfg(feature = "ring-cipher")]
-mod ring_cipher;
+mod ring_aes_gcm_cipher;
 #[cfg(not(feature = "ring-cipher"))]
 mod aes_gcm_cipher;
 mod rsa_cipher;

vnt/src/cipher/ring_aes_gcm_cipher.rs

@@ -67,10 +67,7 @@ impl AesGcmCipher {
         let nonce = aead::Nonce::assume_unique_for_key(nonce_raw);
         let mut secret_body = SecretBody::new(net_packet.payload_mut())?;
         let tag = secret_body.tag();
-        if tag.len() != 16 {
+        let finger = self.finger.calculate_finger(&nonce_raw, secret_body.en_body());
-            return Err(io::Error::new(io::ErrorKind::Other, "tag err"));
-        }
-        let finger = self.finger.calculate_finger(&nonce_raw, &secret_body);
         if &finger != secret_body.finger() {
             return Err(io::Error::new(io::ErrorKind::Other, "finger err"));
         }
@@ -122,7 +119,7 @@ impl AesGcmCipher {
                     return Err(io::Error::new(io::ErrorKind::Other, format!("加密tag长度错误:{}", tag.len())));
                 }
                 secret_body.set_tag(tag)?;
-                let finger = self.finger.calculate_finger(&nonce_raw, &secret_body);
+                let finger = self.finger.calculate_finger(&nonce_raw, secret_body.en_body());
                 secret_body.set_finger(&finger)?;
                 net_packet.set_encrypt_flag(true);
                 Ok(())

vnt/src/ip_proxy/tcp_proxy.rs

@@ -35,12 +35,10 @@ impl TcpProxy {
                                         continue;
                                     }
                                 };
-                                let tcp_proxy_map = tcp_proxy_map.clone();
                                 tokio::spawn(async move {
                                     if let Err(e) = proxy(tcp_stream, peer_tcp_stream).await {
                                         log::warn!("{}->{},{}",sender_addr,dest_addr,e);
                                     }
-                                    tcp_proxy_map.remove(&sender_addr);
                                 });
                             }
                         }

vnt/src/protocol/body.rs

@@ -54,6 +54,11 @@ impl<B: AsRef<[u8]>> SecretBody<B> {
         let end = self.buffer.as_ref().len() - 12;
         &self.buffer.as_ref()[end - 16..end]
     }
+    /// 数据部分+tag部分
+    pub fn en_body(&self) -> &[u8] {
+        let end = self.buffer.as_ref().len() - 12;
+        &self.buffer.as_ref()[..end]
+    }
     pub fn finger(&self) -> &[u8] {
         let end = self.buffer.as_ref().len();
         &self.buffer.as_ref()[end - 12..end]