1.2.3

1.同步处理UDP，提升性能
2.加密引入openssl，提升性能
3.安卓支持ip代理

.github/workflows/rust.yml

@@ -37,49 +37,41 @@ jobs:
       fail-fast: false
       matrix:
         # a list of all the targets
+        # 选择使用openssl或者ring，并不是所有平台都支持
         include:
           - TARGET: i686-unknown-linux-musl # test in an alpine container on a mac
             OS: ubuntu-latest
-            FEATURES: normal
-          - TARGET: x86_64-unknown-linux-gnu # tested in a debian container on a mac
-            OS: ubuntu-latest
-            FEATURES: ring-cipher
+            FEATURES: default
           - TARGET: x86_64-unknown-linux-musl # test in an alpine container on a mac
             OS: ubuntu-latest
-            FEATURES: ring-cipher
-          - TARGET: aarch64-unknown-linux-gnu # tested on aws t4g.nano
-            OS: ubuntu-latest
-            FEATURES: ring-cipher
+            FEATURES: ring-cipher,openssl-vendored
           - TARGET: aarch64-unknown-linux-musl # tested on aws t4g.nano in alpine container
             OS: ubuntu-latest
-            FEATURES: normal
-          - TARGET: armv7-unknown-linux-gnueabihf # raspberry pi 2-3-4, not tested
-            OS: ubuntu-latest
-            FEATURES: ring-cipher
+            FEATURES: default
           - TARGET: armv7-unknown-linux-musleabihf # raspberry pi 2-3-4, not tested
             OS: ubuntu-latest
-            FEATURES: normal
-          - TARGET: arm-unknown-linux-gnueabihf # raspberry pi 0-1, not tested
-            OS: ubuntu-latest
-            FEATURES: ring-cipher
+            FEATURES: openssl-vendored
           - TARGET: arm-unknown-linux-musleabihf # raspberry pi 0-1, not tested
             OS: ubuntu-latest
-            FEATURES: normal
+            FEATURES: openssl-vendored
           - TARGET: x86_64-apple-darwin # tested on a mac, is not properly signed so there are security warnings
             OS: macos-latest
-            FEATURES: ring-cipher
+            FEATURES: ring-cipher,openssl-vendored
           - TARGET: aarch64-apple-darwin # tested on a mac, is not properly signed so there are security warnings
             OS: macos-latest
-            FEATURES: ring-cipher
+            FEATURES: ring-cipher,openssl-vendored
           - TARGET: i686-pc-windows-msvc # tested on a windows machine
-            OS: windows-latest
-            FEATURES: ring-cipher
+            OS: windows-2019
+            FEATURES: ring-cipher,openssl-vendored
           - TARGET: x86_64-pc-windows-msvc # tested on a windows machine
             OS: windows-latest
-            FEATURES: ring-cipher
+            FEATURES: ring-cipher,openssl-vendored
           - TARGET: mipsel-unknown-linux-musl # openwrt
             OS: ubuntu-latest
-            FEATURES: normal
+            FEATURES: openssl-vendored
+          - TARGET: mips-unknown-linux-musl # openwrt
+            OS: ubuntu-latest
+            FEATURES: openssl-vendored
     # needs: test
     runs-on: ${{ matrix.OS }}
     env:
@@ -98,6 +90,9 @@ jobs:
             ~/.cargo/registry
             ./target
           key: build-cargo-registry-${{matrix.TARGET}}
+      - name: Use strawberry perl
+        if: startsWith(matrix.os, 'windows')
+        run: echo OPENSSL_SRC_PERL=C:/Strawberry/perl/bin/perl >> $GITHUB_ENV
       - name: List
         run: find ./
       - name: Install and configure dependencies
@@ -105,45 +100,64 @@ jobs:
           # dependencies are only needed on ubuntu as that's the only place where
           # we make cross-compilation
           if [[ $OS =~ ^ubuntu.*$ ]]; then
-            sudo apt-get update && sudo apt-get install -qq crossbuild-essential-arm64 crossbuild-essential-armhf musl-tools gcc-mipsel-linux-gnu
+            sudo apt-get update && sudo apt-get install -qq crossbuild-essential-arm64 crossbuild-essential-armhf musl-tools
+            #  curl -s musl.cc | grep mipsel
+            case $TARGET in 
+              mipsel-unknown-linux-musl)
+                MUSL_URI=mipsel-linux-musl-cross
+                ;;
+              aarch64-unknown-linux-musl)
+                MUSL_URI=aarch64-linux-musl-cross
+                ;;
+              armv7-unknown-linux-musleabihf)
+                MUSL_URI=armv7l-linux-musleabihf-cross
+                ;;   
+              arm-unknown-linux-musleabihf)
+                MUSL_URI=arm-linux-musleabihf-cross
+                ;;        
+              mips-unknown-linux-musl)
+                MUSL_URI=mips-linux-musl-cross
+                ;;
+            esac
+          
+            if [  -n "$MUSL_URI" ]; then
+              mkdir -p ./musl_gcc 
+              wget -c https://musl.cc/$MUSL_URI.tgz -P ./musl_gcc/
+              tar zxf ./musl_gcc/$MUSL_URI.tgz -C ./musl_gcc/
+              sudo ln -s $(pwd)/musl_gcc/$MUSL_URI/bin/*gcc /usr/bin/
+            fi
           fi
+          
+          
           # some additional configuration for cross-compilation on linux
           cat >>~/.cargo/config <<EOF
           [target.x86_64-unknown-linux-musl]
-          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols","--cfg","aes_armv8"]
-          [target.aarch64-unknown-linux-gnu]
-          linker = "aarch64-linux-gnu-gcc"
-          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols","--cfg","aes_armv8"]
+          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols"]
           [target.aarch64-unknown-linux-musl]
-          linker = "aarch64-linux-gnu-gcc"
-          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols","--cfg","aes_armv8"]
-          [target.armv7-unknown-linux-gnueabihf]
-          linker = "arm-linux-gnueabihf-gcc"
-          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols","--cfg","aes_armv8"]
+          linker = "aarch64-linux-musl-gcc"
+          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols"]
           [target.armv7-unknown-linux-musleabihf]
-          linker = "arm-linux-gnueabihf-gcc"
-          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols","--cfg","aes_armv8"]
-          [target.arm-unknown-linux-gnueabihf]
-          linker = "arm-linux-gnueabihf-gcc"
-          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols","--cfg","aes_armv8"]
+          linker = "armv7l-linux-musleabihf-gcc"
+          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols"]
           [target.arm-unknown-linux-musleabihf]
-          linker = "arm-linux-gnueabihf-gcc"
-          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols","--cfg","aes_armv8"]
+          linker = "arm-linux-musleabihf-gcc"
+          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols"]
           [target.mipsel-unknown-linux-musl]
-          linker = "mipsel-linux-gnu-gcc"
-          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols","--cfg","aes_armv8"]
+          linker = "mipsel-linux-musl-gcc"
+          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols"]
+          [target.mips-unknown-linux-musl]
+          linker = "mips-linux-musl-gcc"
+          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols"]
           [target.x86_64-pc-windows-msvc]
-          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols","--cfg","aes_armv8"]    
+          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols"]    
           [target.i686-pc-windows-msvc]
-          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols","--cfg","aes_armv8"]      
+          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols"]      
           [target.x86_64-apple-darwin]
-          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols","--cfg","aes_armv8"] 
+          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols"] 
           [target.aarch64-apple-darwin]
-          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols","--cfg","aes_armv8"]     
+          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols"]     
           [target.i686-unknown-linux-musl]
-          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols","--cfg","aes_armv8"]    
-          [target.x86_64-unknown-linux-gnu]
-          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols","--cfg","aes_armv8"]
+          rustflags = ["-C", "target-feature=+crt-static","-C", "strip=symbols"]    
           EOF
       - name: Install rust target
         run: rustup target add $TARGET

common/Cargo.toml

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

common/src/args_parse.rs

@@ -68,6 +68,9 @@ pub fn out_ips_parse(ips: &Vec<String>) -> Result<Vec<(u32, u32)>, String> {
 
 pub fn to_ip(mask: &str) -> Result<u32, String> {
     if let Ok(m) = mask.parse::<u32>() {
+        if m >= 32 {
+            return Err("not netmask".to_string());
+        }
         let mut mask = 0 as u32;
         for i in 0..m {
             mask = mask | (1 << (31 - i));

common/src/identifier.rs

@@ -1,8 +1,7 @@
-use std::process::Command;
-
 #[cfg(target_os = "windows")]
 pub fn get_unique_identifier() -> Option<String> {
     use std::os::windows::process::CommandExt;
+    use std::process::Command;
     let output = match Command::new("wmic")
         .creation_flags(0x08000000)
         .args(&["csproduct", "get", "UUID"])
@@ -25,6 +24,7 @@ pub fn get_unique_identifier() -> Option<String> {
 
 #[cfg(target_os = "macos")]
 pub fn get_unique_identifier() -> Option<String> {
+    use std::process::Command;
     let output = match Command::new("ioreg")
         .args(&["-rd1", "-c", "IOPlatformExpertDevice"])
         .output()
@@ -50,6 +50,7 @@ pub fn get_unique_identifier() -> Option<String> {
 
 #[cfg(target_os = "linux")]
 pub fn get_unique_identifier() -> Option<String> {
+    use std::process::Command;
     let output = match Command::new("dmidecode")
         .arg("-s")
         .arg("system-uuid")

vnt-cli/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "vnt-cli"
-version = "1.2.2"
+version = "1.2.3"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -8,7 +8,7 @@ edition = "2021"
 [dependencies]
 vnt = { path = "../vnt", package = "vnt", optional = true }
 common = { path = "../common" }
-tokio = { version = "1.28.1", features = ["full"] }
+tokio = { version = "1.32.0", features = ["full"] }
 getopts = "0.2.21"
 console = "0.15.2"
 os_info = "3.7.0"
@@ -30,8 +30,9 @@ sudo = "0.6.0"
 winapi = { version = "0.3.9", features = ["handleapi", "processthreadsapi", "winnt", "securitybaseapi", "impl-default"] }
 
 [features]
-default = ["normal"]
-normal = ["vnt"]
+default = ["vnt"]
+openssl = ["vnt/openssl"]
+openssl-vendored = ["vnt/openssl-vendored"]
 ring-cipher = ["vnt/ring-cipher"]
 
 

vnt-cli/README.md

@@ -43,7 +43,7 @@
 
 ### -u `<mtu>`
 
-设置虚拟网卡的mtu值，大多数情况下使用默认值效率会更高，也可根据实际情况微调这个值，不加密默认为1430，加密默认为1410
+设置虚拟网卡的mtu值，大多数情况下使用默认值效率会更高，也可根据实际情况微调这个值，不加密默认为1450，加密默认为1410
 
 ###  --tcp
 和服务端使用tcp通信。有些网络提供商对UDP限制比较大，这个时候可以选择使用TCP模式，提高稳定性。一般来说udp延迟和消耗更低
@@ -51,17 +51,17 @@
 指定虚拟ip,指定的ip不能和其他设备重复,必须有效并且在服务端所属网段下,默认情况由服务端分配
 ### --par `<parallel>`
 任务并行度(必须为正整数),默认值为1,该值表示处理网卡读写的任务数,组网设备数较多、处理延迟较大时可适当调大此值
-### --thread `<thread>`
-线程数(必须为正整数),默认为核心数乘2,该值表示处理网络读写、ip代理、打洞等用到的线程数,组网设备数较多、处理延迟较大时可适当调大此值
 ### --model `<model>`
-加密模式，可选值 aes_gcm/aes_cbc，默认使用aes_gcm，通常情况使用aes_cbc性能更好
+加密模式，可选值 aes_gcm/aes_cbc/aes_ecb，默认使用aes_gcm，通常情况aes_gcm安全性高、aes_ecb性能更好
 
-| 密码位数  | model  | 加密算法       |  
-|-------|--------|------------|
-| 1~8位  |   aes_gcm     | AES128-GCM |
-| `>=`8 |   aes_gcm     | AES256-GCM |
-| 1~8位  |   aes_cbc     | AES128-CBC |
-| `>=`8 |   aes_cbc     | AES256-CBC |
+| 密码位数  | model   | 加密算法       |  
+|-------|---------|------------|
+| 1~8位  | aes_gcm | AES128-GCM |
+| `>=`8 | aes_gcm | AES256-GCM |
+| 1~8位  | aes_cbc | AES128-CBC |
+| `>=`8 | aes_cbc | AES256-CBC |
+| 1~8位  | aes_ecb | AES128-ECB |
+| `>=`8 | aes_ecb | AES256-ECB |
 ### --finger 
 开启数据指纹校验，可增加安全性，如果服务端开启指纹校验，则客户端也必须开启，开启会损耗一部分性能
 

vnt-cli/src/main.rs

@@ -7,8 +7,6 @@ use console::style;
 use getopts::Options;
 use tokio::io::{AsyncBufReadExt, BufReader};
 use tokio::signal;
-#[cfg(unix)]
-use tokio::signal::unix::{signal, SignalKind};
 
 use common::args_parse::{ips_parse, out_ips_parse};
 use vnt::channel::punch::PunchModel;
@@ -143,12 +141,12 @@ fn main() {
             if let Some(addr) = addr.next() {
                 addr
             } else {
-                println!("parameter -s error .");
+                println!("parameter '-s {}' error .", server_address_str);
                 return;
             }
         }
         Err(e) => {
-            println!("parameter -s error {}.", e);
+            println!("parameter '-s {}' error {}.", server_address_str, e);
             return;
         }
     };
@@ -165,7 +163,7 @@ fn main() {
         Err(e) => {
             print_usage(&program, opts);
             println!();
-            println!("-i {}", e);
+            println!("-i: {:?} {}", in_ip, e);
             println!("example: -i 192.168.0.0/24,10.26.0.3");
             return;
         }
@@ -176,7 +174,7 @@ fn main() {
         Err(e) => {
             print_usage(&program, opts);
             println!();
-            println!("-o {}", e);
+            println!("-o: {:?} {}", out_ip, e);
             println!("example: -o 0.0.0.0/0");
             return;
         }
@@ -192,7 +190,7 @@ fn main() {
             Err(e) => {
                 print_usage(&program, opts);
                 println!();
-                println!("-u {}", e);
+                println!("'-u {}' {}", mtu, e);
                 return;
             }
         }
@@ -200,10 +198,11 @@ fn main() {
         None
     };
     let virtual_ip: Option<String> = matches.opt_get("ip").unwrap();
-    let virtual_ip = virtual_ip.map(|v| Ipv4Addr::from_str(&v).expect("--ip error"));
+    let virtual_ip =
+        virtual_ip.map(|v| Ipv4Addr::from_str(&v).expect(&format!("'--ip {}' error", v)));
     if let Some(virtual_ip) = virtual_ip {
         if virtual_ip.is_unspecified() || virtual_ip.is_broadcast() || virtual_ip.is_multicast() {
-            println!("--ip invalid");
+            println!("'--ip {}' invalid", virtual_ip);
             return;
         }
     }
@@ -211,21 +210,15 @@ fn main() {
     let relay = matches.opt_present("relay");
     let parallel = matches.opt_get::<usize>("par").unwrap().unwrap_or(1);
     if parallel == 0 {
-        println!("--par invalid");
+        println!("'--par {}' invalid", parallel);
         return;
     }
-    let thread_num = matches
-        .opt_get::<usize>("thread")
-        .unwrap()
-        .unwrap_or(std::thread::available_parallelism().unwrap().get() * 2);
+
     let cipher_model = matches
         .opt_get::<CipherModel>("model")
         .unwrap()
         .unwrap_or(CipherModel::AesGcm);
-    if thread_num == 0 {
-        println!("--thread invalid");
-        return;
-    }
+
     let finger = matches.opt_present("finger");
     let punch_model = matches
         .opt_get::<PunchModel>("punch")
@@ -254,15 +247,10 @@ fn main() {
         finger,
         punch_model,
     );
-    let runtime = tokio::runtime::Builder::new_multi_thread()
-        .enable_all()
-        .worker_threads(thread_num)
-        .build()
-        .unwrap();
-    runtime.block_on(main0(config, !unused_cmd));
+    main0(config, !unused_cmd);
     std::process::exit(0);
 }
-
+#[tokio::main]
 async fn main0(config: Config, show_cmd: bool) {
     let server_encrypt = config.server_encrypt;
     let mut vnt_util = VntUtil::new(config).await.unwrap();
@@ -373,8 +361,6 @@ async fn main0(config: Config, show_cmd: bool) {
             println!("command error :{}", e);
         }
     });
-    #[cfg(unix)]
-    let mut sigterm = signal(SignalKind::terminate()).expect("Error setting SIGTERM handler");
     if show_cmd {
         let stdin = tokio::io::stdin();
         let mut cmd = String::new();
@@ -382,7 +368,6 @@ async fn main0(config: Config, show_cmd: bool) {
         loop {
             cmd.clear();
             println!("input:list,info,route,all,stop");
-            #[cfg(unix)]
             tokio::select! {
                 _ = vnt.wait_stop()=>{
                     return;
@@ -390,36 +375,7 @@ async fn main0(config: Config, show_cmd: bool) {
                 _ = signal::ctrl_c()=>{
                     let _ = vnt.stop();
                     vnt.wait_stop_ms(std::time::Duration::from_secs(3)).await;
-                    return;
-                }
-                _ = sigterm.recv()=>{
-                    let _ = vnt.stop();
-                    vnt.wait_stop_ms(std::time::Duration::from_secs(3)).await;
-                    return;
-                }
-                rs = reader.read_line(&mut cmd)=>{
-                     match rs {
-                        Ok(len) => {
-                            if !command(&cmd[..len],&vnt){
-                                break;
-                            }
-                        }
-                        Err(e) => {
-                            println!("input err:{}",e);
-                            break;
-                        }
-                    }
-                }
-            }
-            #[cfg(windows)]
-            tokio::select! {
-                _ = vnt.wait_stop()=>{
-                    return;
-                }
-                _ = signal::ctrl_c()=>{
-                    let _ = vnt.stop();
-                    vnt.wait_stop_ms(std::time::Duration::from_secs(3)).await;
-                    return;
+                    std::process::exit(0);
                 }
                 rs = reader.read_line(&mut cmd)=>{
                      match rs {
@@ -437,23 +393,6 @@ async fn main0(config: Config, show_cmd: bool) {
             }
         }
     }
-    #[cfg(unix)]
-    tokio::select! {
-                 _ = vnt.wait_stop()=>{
-                     return;
-                 }
-                 _ = signal::ctrl_c()=>{
-                     let _ = vnt.stop();
-                     vnt.wait_stop_ms(std::time::Duration::from_secs(3)).await;
-                     return;
-                 }
-                 _ = sigterm.recv()=>{
-                     let _ = vnt.stop();
-                     vnt.wait_stop_ms(std::time::Duration::from_secs(3)).await;
-                     return;
-                 }
-    }
-    #[cfg(windows)]
     vnt.wait_stop().await;
 }
 
@@ -508,13 +447,12 @@ fn print_usage(program: &str, _opts: Options) {
     println!("  -w <password>       使用该密码生成的密钥对客户端数据进行加密,并且服务端无法解密,使用相同密码的客户端才能通信");
     println!("  -W                  加密当前客户端和服务端通信的数据,请留意服务端指纹是否正确");
     println!("  -m                  模拟组播,默认情况下组播数据会被当作广播发送,开启后会模拟真实组播的数据发送");
-    println!("  -u <mtu>            自定义mtu(不加密默认为1430，加密默认为1410)");
+    println!("  -u <mtu>            自定义mtu(不加密默认为1450，加密默认为1410)");
     println!("  --tcp               和服务端使用tcp通信,默认使用udp,遇到udp qos时可指定使用tcp");
     println!("  --ip <ip>           指定虚拟ip,指定的ip不能和其他设备重复,必须有效并且在服务端所属网段下,默认情况由服务端分配");
     println!("  --relay             仅使用服务器转发,不使用p2p,默认情况允许使用p2p");
     println!("  --par <parallel>    任务并行度(必须为正整数),默认值为1");
-    println!("  --thread <thread>   线程数(必须为正整数),默认为核心数乘2");
-    println!("  --model <model>     加密模式(默认aes_gcm)，可选值aes_gcm/aes_cbc/aes_ecb，通常性能aes_ecb>aes_cbc>aes_gcm,安全性则相反");
+    println!("  --model <model>     加密模式(默认aes_gcm)，可选值aes_gcm/aes_cbc/aes_ecb，一般来说性能:aes_ecb>aes_cbc>aes_gcm");
     println!("  --finger            增加数据指纹校验，可增加安全性，如果服务端开启指纹校验，则客户端也必须开启");
     println!("  --punch <punch>     取值ipv4/ipv6，ipv4表示仅使用ipv4打洞");
 

vnt-jni/Cargo.toml

@@ -1,11 +1,12 @@
 [package]
 name = "vnt-jni"
-version = "1.2.2"
+version = "1.2.3"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+common = { path = "../common" }
 vnt = {path="../vnt"}
 
 jni = { version = "0.21.1", default-features = false }

vnt-jni/src/vnt_util.rs

@@ -66,6 +66,36 @@ fn new_sync(env: &mut JNIEnv, config: JObject) -> Result<VntUtilSync, Error> {
     let cipher_model = to_string_not_null(env, &config, "cipherModel")?;
     let tcp = env.get_field(&config, "tcp", "Z")?.z()?;
     let finger = env.get_field(&config, "finger", "Z")?.z()?;
+    let in_ips = to_string(env, &config, "inIps")?;
+    let out_ips = to_string(env, &config, "outIps")?;
+    let in_ips = if let Some(in_ips) = in_ips {
+        let in_ips: Vec<&str> = in_ips.split("\n").collect();
+        let in_ips = in_ips.iter().map(|v| v.to_string()).collect();
+        match common::args_parse::ips_parse(&in_ips) {
+            Ok(in_ips) => in_ips,
+            Err(e) => {
+                env.throw_new("java/lang/RuntimeException", format!("in_ips {}", e))
+                    .expect("throw");
+                return Err(Error::JavaException);
+            }
+        }
+    } else {
+        vec![]
+    };
+    let out_ips = if let Some(out_ips) = out_ips {
+        let out_ips: Vec<&str> = out_ips.split("\n").collect();
+        let out_ips = out_ips.iter().map(|v| v.to_string()).collect();
+        match common::args_parse::out_ips_parse(&out_ips) {
+            Ok(out_ips) => out_ips,
+            Err(e) => {
+                env.throw_new("java/lang/RuntimeException", format!("out_ips {}", e))
+                    .expect("throw");
+                return Err(Error::JavaException);
+            }
+        }
+    } else {
+        vec![]
+    };
 
     let server_address = match server_address_str.to_socket_addrs() {
         Ok(mut rs) => {
@@ -106,8 +136,8 @@ fn new_sync(env: &mut JNIEnv, config: JObject) -> Result<VntUtilSync, Error> {
         server_address,
         server_address_str,
         stun_server,
-        vec![],
-        vec![],
+        in_ips,
+        out_ips,
         password,
         false,
         None,

vnt/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "vnt"
-version = "1.2.2"
+version = "1.2.3"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -11,6 +11,7 @@ bytes = "1.3.0"
 log = "0.4.17"
 libc = "0.2.137"
 crossbeam-utils = "0.8"
+crossbeam-epoch = "0.9.15"
 dashmap = "5.5.1"
 parking_lot = "0.12.1"
 byte-pool = "0.2.4"
@@ -19,31 +20,36 @@ rand = "0.8.5"
 sha2 = { version = "0.10.6", features = ["oid"] }
 thiserror = "1.0.37"
 protobuf = "3.2.0"
-socket2 ={ version = "0.5.2", features = ["all"] }
-tokio = { version = "1.28.1", features = ["full"] }
-aes-gcm = {version="0.10.2", optional = true}
-ring = {version="0.16.20", optional = true}
+socket2 = { version = "0.5.2", features = ["all"] }
+tokio = { version = "1.32.0", features = ["full"] }
+aes-gcm = { version = "0.10.2" }
+ring = { version = "0.16.20", optional = true }
 cbc = "0.1.2"
 ecb = "0.1.2"
 aes = "0.8.3"
-stun-format = {version="1.0.1",features=["fmt","rfc3489"]}
-rsa = {version="0.7.2", features = [] }
-spki = {version="0.6.0",features=["fingerprint","alloc"]}
+stun-format = { version = "1.0.1", features = ["fmt", "rfc3489"] }
+rsa = { version = "0.7.2", features = [] }
+spki = { version = "0.6.0", features = ["fingerprint", "alloc"] }
+openssl-sys = { git = "https://github.com/lbl8603/rust-openssl" ,optional = true}
+
 
 [target.'cfg(any(target_os = "linux",target_os = "macos"))'.dependencies]
 tun = { path = "./rust-tun" }
 
 [target.'cfg(target_os = "windows")'.dependencies]
-
-win-tun-tap = {path = "./win-tun-tap"}
+win-tun-tap = { path = "./win-tun-tap" }
 libloading = "0.7.4"
 
+
 [build-dependencies]
 protobuf-codegen = "3.2.0"
 protoc-bin-vendored = "3.0.0"
 
 [features]
-default=["aes-gcm"]
-ring-cipher=["ring"]
+default = []
+openssl = ["openssl-sys"]
+# 从源码编译
+openssl-vendored = ["openssl-sys/vendored"]
+ring-cipher = ["ring"]
 
 

vnt/src/channel/channel.rs

@@ -1,36 +1,42 @@
+use std::collections::HashMap;
 use std::io;
-use std::net::{Ipv4Addr, SocketAddr};
+use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
+use std::ops::Sub;
+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;
+use std::net::UdpSocket as StdUdpSocket;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::tcp::OwnedReadHalf;
 use tokio::net::{TcpStream, UdpSocket};
 use tokio::sync::watch::{channel, Receiver, Sender};
 
 use crate::channel::punch::NatType;
-use crate::channel::{Route, RouteKey, Status};
+use crate::channel::{Route, RouteKey, Status, TCP_ID, UDP_ID, UDP_V6_ID};
 use crate::core::status::VntWorker;
 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<UdpSocket>,
-    pub(crate) main_channel_ipv6: Option<Arc<UdpSocket>>,
+    pub(crate) main_channel: Arc<StdUdpSocket>,
+    pub(crate) main_channel_ipv6: Option<Arc<StdUdpSocket>>,
     //在udp的基础上，可以选择使用tcp和服务端通信
     pub(crate) main_tcp_channel: Option<tokio::sync::mpsc::Sender<Vec<u8>>>,
-    pub(crate) route_table: DashMap<Ipv4Addr, Vec<Route>>,
+    pub(crate) route_table: Atomic<HashMap<Ipv4Addr, Vec<Route>>>,
     pub(crate) route_table_time: DashMap<(RouteKey, Ipv4Addr), Instant>,
     pub(crate) status_receiver: Receiver<Status>,
     pub(crate) status_sender: Sender<Status>,
-    pub(crate) udp_map: DashMap<usize, Arc<UdpSocket>>,
+    pub(crate) udp_map: Atomic<HashMap<usize, Arc<UdpSocket>>>,
     pub(crate) channel_num: usize,
     current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
 }
@@ -42,8 +48,8 @@ pub struct Context {
 
 impl Context {
     pub fn new(
-        main_channel: Arc<UdpSocket>,
-        main_channel_ipv6: Option<Arc<UdpSocket>>,
+        main_channel: Arc<StdUdpSocket>,
+        main_channel_ipv6: Option<Arc<StdUdpSocket>>,
         main_tcp_channel: Option<tokio::sync::mpsc::Sender<Vec<u8>>>,
         current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
         _channel_num: usize,
@@ -55,11 +61,11 @@ impl Context {
             main_channel,
             main_channel_ipv6,
             main_tcp_channel,
-            route_table: DashMap::with_capacity(16),
-            route_table_time: DashMap::with_capacity(16),
+            route_table: Atomic::new(HashMap::with_capacity(16)),
+            route_table_time: DashMap::new_cap(16),
             status_receiver,
             status_sender,
-            udp_map: DashMap::new(),
+            udp_map: Atomic::new(HashMap::with_capacity(16)),
             channel_num,
             current_device,
         });
@@ -74,8 +80,21 @@ impl Context {
     pub fn is_cone(&self) -> bool {
         *self.inner.status_receiver.borrow() == Status::Cone
     }
-    pub fn close(&self) {
+    pub fn close(&self) -> io::Result<()> {
         let _ = self.inner.status_sender.send(Status::Close);
+        if let Ok(port) = self.main_local_ipv4_port() {
+            let _ = StdUdpSocket::bind("127.0.0.1:0")?.send_to(
+                b"stop",
+                SocketAddr::V4(std::net::SocketAddrV4::new(Ipv4Addr::LOCALHOST, port)),
+            );
+        }
+        if let Ok(port) = self.main_local_ipv6_port() {
+            let _ = StdUdpSocket::bind("[::]:0")?.send_to(
+                b"stop",
+                SocketAddr::V6(std::net::SocketAddrV6::new(Ipv6Addr::LOCALHOST, port, 0, 0)),
+            );
+        }
+        Ok(())
     }
     pub fn is_main_tcp(&self) -> bool {
         self.inner.main_tcp_channel.is_some()
@@ -106,68 +125,112 @@ impl Context {
             Err(io::Error::new(io::ErrorKind::Other, "not ipv6"))
         }
     }
-    pub async fn send_main_udp(&self, buf: &[u8], addr: SocketAddr) -> io::Result<usize> {
+    fn insert_udp(&self, id: usize, udp: Arc<UdpSocket>) {
+        self.insert_udp_(id, Some(udp))
+    }
+    fn remove_udp(&self, id: usize) {
+        self.insert_udp_(id, None)
+    }
+    fn insert_udp_(&self, id: usize, udp: Option<Arc<UdpSocket>>) {
+        let guard = &crossbeam_epoch::pin();
+        let udp_map = &self.inner.udp_map;
+        let mut udp_map_shared = self.inner.udp_map.load(Ordering::Relaxed, guard);
+        loop {
+            let mut map = unsafe { udp_map_shared.as_ref().unwrap().clone() };
+            match udp.clone() {
+                None => {
+                    map.remove(&id);
+                }
+                Some(udp) => {
+                    map.insert(id, udp);
+                }
+            }
+            match udp_map.compare_exchange(
+                udp_map_shared,
+                Owned::new(map),
+                Ordering::Relaxed,
+                Ordering::Relaxed,
+                guard,
+            ) {
+                Ok(p) => unsafe {
+                    guard.defer_destroy(p);
+                    return;
+                },
+                Err(e) => {
+                    udp_map_shared = e.current;
+                }
+            }
+        }
+    }
+    pub fn send_main_udp(&self, buf: &[u8], addr: SocketAddr) -> io::Result<usize> {
         if addr.is_ipv6() {
             if let Some(udp_ipv6) = &self.inner.main_channel_ipv6 {
-                udp_ipv6.send_to(buf, addr).await
+                udp_ipv6.send_to(buf, addr)
             } else {
                 Err(io::Error::new(io::ErrorKind::Other, "not ipv6"))
             }
         } else {
-            self.inner.main_channel.send_to(buf, addr).await
+            self.inner.main_channel.send_to(buf, addr)
         }
     }
-    pub fn try_send_main_udp(&self, buf: &[u8], addr: SocketAddr) -> io::Result<usize> {
-        if addr.is_ipv6() {
-            if let Some(udp_ipv6) = &self.inner.main_channel_ipv6 {
-                udp_ipv6.try_send_to(buf, addr)
-            } else {
-                Err(io::Error::new(io::ErrorKind::Other, "not ipv6"))
-            }
-        } else {
-            self.inner.main_channel.try_send_to(buf, addr)
-        }
-    }
-    pub async fn send_main(&self, buf: &[u8], addr: SocketAddr) -> io::Result<usize> {
+
+    pub fn send_main(&self, buf: &[u8], addr: SocketAddr) -> io::Result<usize> {
         if let Some(sender) = &self.inner.main_tcp_channel {
-            if sender.send(buf.to_vec()).await.is_ok() {
+            if sender.try_send(buf.to_vec()).is_ok() {
                 Ok(buf.len())
             } else {
                 Err(io::Error::new(io::ErrorKind::Other, "send_main err"))
             }
         } else {
-            self.send_main_udp(buf, addr).await
-        }
-    }
-    pub fn try_send_main(&self, buf: &[u8], addr: SocketAddr) -> io::Result<usize> {
-        if let Some(sender) = &self.inner.main_tcp_channel {
-            if sender.try_send(buf.to_vec()).is_ok() {
-                Ok(buf.len())
-            } else {
-                Err(io::Error::new(io::ErrorKind::Other, "try_send_main err"))
-            }
-        } else {
-            self.try_send_main_udp(buf, addr)
+            self.send_main_udp(buf, addr)
         }
     }
 
-    pub(crate) async fn send_all(&self, buf: &[u8], addr: SocketAddr) -> io::Result<()> {
-        for udp_ref in self.inner.udp_map.iter() {
-            let udp = udp_ref.clone();
-            drop(udp_ref);
+    pub(crate) fn try_send_all(&self, buf: &[u8], addr: SocketAddr) -> io::Result<()> {
+        let table = unsafe {
+            let guard = &crossbeam_epoch::pin();
+            self.inner
+                .udp_map
+                .load(Ordering::Relaxed, guard)
+                .as_ref()
+                .unwrap()
+                .clone()
+        };
+        if table.is_empty() {
+            log::error!("udp列表为空,addr={}", addr);
+            return Ok(());
+        }
+        for (_, udp) in table {
             //使用ipv6的udp发送ipv4报文会出错
-            let _ = udp.send_to(buf, addr).await;
+            if let Err(e) = udp.try_send_to(buf, addr) {
+                log::error!("{:?}", e);
+            }
         }
         Ok(())
     }
 
     pub async fn send_by_id(&self, buf: &[u8], id: &Ipv4Addr) -> io::Result<usize> {
-        if let Some(v) = self.inner.route_table.get(id) {
-            if v.value().is_empty() {
+        let route = self.get_route_by_id(id)?;
+        self.send_by_key(buf, &route.route_key()).await
+    }
+    pub fn try_send_by_id(&self, buf: &[u8], id: &Ipv4Addr) -> io::Result<usize> {
+        let route = self.get_route_by_id(id)?;
+        self.try_send_by_key(buf, &route.route_key())
+    }
+    fn get_route_by_id(&self, id: &Ipv4Addr) -> io::Result<Route> {
+        let guard = &crossbeam_epoch::pin();
+        let table = unsafe {
+            self.inner
+                .route_table
+                .load(Ordering::Relaxed, guard)
+                .as_ref()
+                .unwrap()
+        };
+        if let Some(v) = table.get(id) {
+            if v.is_empty() {
                 return Err(io::Error::new(io::ErrorKind::NotFound, "route not found"));
             }
-            let route = v.value()[0];
-            drop(v);
+            let route = v[0];
             if route.rt == 199 {
                 //这通常是刚加入路由，直接放弃使用,避免抖动
                 return Err(io::Error::new(io::ErrorKind::NotFound, "route not found"));
@@ -180,60 +243,81 @@ impl Context {
                     }
                 }
             }
-            return self.send_by_key(buf, &route.route_key()).await;
+            return Ok(route);
         }
         Err(io::Error::new(io::ErrorKind::NotFound, "route not found"))
     }
 
-    pub fn try_send_by_id(&self, buf: &[u8], id: &Ipv4Addr) -> io::Result<usize> {
-        if let Some(v) = self.inner.route_table.get(id) {
-            if v.value().is_empty() {
-                return Err(io::Error::new(io::ErrorKind::NotFound, "route not found"));
-            }
-            let route = v.value()[0];
-            drop(v);
-            if let Some(udp) = self.inner.udp_map.get(&route.index) {
-                return udp.value().try_send_to(buf, route.addr);
-            }
-        }
-        Err(io::Error::new(io::ErrorKind::NotFound, "route not found"))
-    }
     pub async fn send_by_key(&self, buf: &[u8], route_key: &RouteKey) -> io::Result<usize> {
-        if route_key.index == 0 {
-            if let Some(sender) = &self.inner.main_tcp_channel {
-                let mut vec = vec![0; 4 + buf.len()];
-                vec[4..].copy_from_slice(buf);
-                return if sender.send(vec).await.is_ok() {
-                    Ok(buf.len())
+        match route_key.index {
+            TCP_ID => {
+                if let Some(sender) = &self.inner.main_tcp_channel {
+                    if sender.send(buf.to_vec()).await.is_ok() {
+                        Ok(buf.len())
+                    } else {
+                        Err(io::Error::new(io::ErrorKind::Other, "send_by_key err"))
+                    }
                 } else {
                     Err(io::Error::new(io::ErrorKind::Other, "send_by_key err"))
-                };
+                }
+            }
+            UDP_ID => self.inner.main_channel.send_to(buf, route_key.addr),
+            UDP_V6_ID => {
+                if let Some(udp_ipv6) = &self.inner.main_channel_ipv6 {
+                    udp_ipv6.send_to(buf, route_key.addr)
+                } else {
+                    Err(io::Error::new(io::ErrorKind::Other, "not ipv6 udp"))
+                }
+            }
+            _ => {
+                if let Some(udp) = self.get_udp_by_route(route_key) {
+                    return udp.send_to(buf, route_key.addr).await;
+                }
+                Err(io::Error::new(io::ErrorKind::NotFound, "route not found"))
             }
         }
-        if let Some(udp_ref) = self.inner.udp_map.get(&route_key.index) {
-            let udp = udp_ref.value().clone();
-            drop(udp_ref);
-            return udp.send_to(buf, route_key.addr).await;
-        }
-        Err(io::Error::new(io::ErrorKind::NotFound, "route not found"))
     }
     pub fn try_send_by_key(&self, buf: &[u8], route_key: &RouteKey) -> io::Result<usize> {
-        if route_key.index == 0 {
-            if let Some(sender) = &self.inner.main_tcp_channel {
-                let mut vec = vec![0; 4 + buf.len()];
-                vec[4..].copy_from_slice(buf);
-                return if sender.try_send(vec).is_ok() {
-                    Ok(buf.len())
+        match route_key.index {
+            TCP_ID => {
+                if let Some(sender) = &self.inner.main_tcp_channel {
+                    if sender.try_send(buf.to_vec()).is_ok() {
+                        Ok(buf.len())
+                    } else {
+                        Err(io::Error::new(io::ErrorKind::Other, "send_by_key err"))
+                    }
                 } else {
-                    Err(io::Error::new(io::ErrorKind::Other, "try_send_by_key err"))
-                };
+                    Err(io::Error::new(io::ErrorKind::Other, "send_by_key err"))
+                }
+            }
+            UDP_ID => self.inner.main_channel.send_to(buf, route_key.addr),
+            UDP_V6_ID => {
+                if let Some(udp_ipv6) = &self.inner.main_channel_ipv6 {
+                    udp_ipv6.send_to(buf, route_key.addr)
+                } else {
+                    Err(io::Error::new(io::ErrorKind::Other, "not ipv6 udp"))
+                }
+            }
+            _ => {
+                if let Some(udp) = self.get_udp_by_route(route_key) {
+                    return udp.try_send_to(buf, route_key.addr);
+                }
+                Err(io::Error::new(io::ErrorKind::NotFound, "route not found"))
             }
         }
-        if let Some(udp) = self.inner.udp_map.get(&route_key.index) {
-            return udp.value().try_send_to(buf, route_key.addr);
-        }
-        Err(io::Error::new(io::ErrorKind::NotFound, "route not found"))
     }
+    fn get_udp_by_route(&self, route_key: &RouteKey) -> Option<Arc<UdpSocket>> {
+        let guard = &crossbeam_epoch::pin();
+        let udp_map = unsafe {
+            self.inner
+                .udp_map
+                .load(Ordering::Relaxed, guard)
+                .as_ref()
+                .unwrap()
+        };
+        udp_map.get(&route_key.index).cloned()
+    }
+
     pub fn add_route_if_absent(&self, id: Ipv4Addr, route: Route) {
         self.add_route_(id, route, true)
     }
@@ -242,120 +326,203 @@ impl Context {
     }
     fn add_route_(&self, id: Ipv4Addr, route: Route, only_if_absent: bool) {
         let key = route.route_key();
-        let mut list = self
-            .inner
-            .route_table
-            .entry(id)
-            .or_insert_with(|| Vec::with_capacity(4));
-        let mut exist = false;
-        for x in list.iter_mut() {
-            if x.metric < route.metric {
-                //不能比当前的路径更长
-                return;
-            }
-            if x.route_key() == key {
-                if only_if_absent {
+        let guard = &crossbeam_epoch::pin();
+        let route_table = &self.inner.route_table;
+        let mut table_share = route_table.load(Ordering::Relaxed, guard);
+        loop {
+            let mut table = unsafe { table_share.as_ref().unwrap().clone() };
+
+            let list = table.entry(id).or_insert_with(|| Vec::with_capacity(4));
+            let mut exist = false;
+            for x in list.iter_mut() {
+                if x.metric < route.metric {
+                    //不能比当前的路径更长
                     return;
                 }
-                x.metric = route.metric;
-                x.rt = route.rt;
-                exist = true;
-                break;
+                if x.route_key() == key {
+                    if only_if_absent {
+                        return;
+                    }
+                    x.metric = route.metric;
+                    x.rt = route.rt;
+                    exist = true;
+                    break;
+                }
             }
-        }
-        if exist {
-            list.sort_by_key(|k| k.sort_key());
-        } else {
-            if route.metric == 1 {
-                //添加了直连的则排除非直连的
-                list.retain(|k| k.metric == 1);
+            if exist {
+                list.sort_by_key(|k| k.sort_key());
+            } else {
+                if route.metric == 1 {
+                    //添加了直连的则排除非直连的
+                    list.retain(|k| k.metric == 1);
+                }
+                list.push(route);
+                list.sort_by_key(|k| k.sort_key());
+                let max_len = self.inner.channel_num + 1;
+                if list.len() > max_len {
+                    list.truncate(max_len);
+                }
             }
-            list.push(route);
-            list.sort_by_key(|k| k.sort_key());
-            let max_len = self.inner.channel_num + 1;
-            if list.len() > max_len {
-                list.truncate(max_len);
+            match route_table.compare_exchange(
+                table_share,
+                Owned::new(table),
+                Ordering::Relaxed,
+                Ordering::Relaxed,
+                guard,
+            ) {
+                Ok(p) => unsafe {
+                    guard.defer_destroy(p);
+                    break;
+                },
+                Err(e) => {
+                    table_share = e.current;
+                }
             }
         }
+
         self.inner
             .route_table_time
-            .insert((key, id), Instant::now());
+            .insert((key, id), Instant::now().sub(Duration::from_secs(10)));
     }
     pub fn route(&self, id: &Ipv4Addr) -> Option<Vec<Route>> {
-        if let Some(v) = self.inner.route_table.get(id) {
-            Some(v.value().clone())
+        let guard = &crossbeam_epoch::pin();
+        let table = unsafe {
+            self.inner
+                .route_table
+                .load(Ordering::Relaxed, guard)
+                .as_ref()
+                .unwrap()
+        };
+        if let Some(v) = table.get(id) {
+            Some(v.clone())
         } else {
             None
         }
     }
     pub fn route_one(&self, id: &Ipv4Addr) -> Option<Route> {
-        if let Some(v) = self.inner.route_table.get(id) {
-            v.value().first().map(|v| *v)
+        let guard = &crossbeam_epoch::pin();
+        let table = unsafe {
+            self.inner
+                .route_table
+                .load(Ordering::Relaxed, guard)
+                .as_ref()
+                .unwrap()
+        };
+        if let Some(v) = table.get(id) {
+            v.first().map(|v| *v)
         } else {
             None
         }
     }
     pub fn route_to_id(&self, route_key: &RouteKey) -> Option<Ipv4Addr> {
-        for x in self.inner.route_table.iter() {
-            for route in x.value() {
+        let guard = &crossbeam_epoch::pin();
+        let table = unsafe {
+            self.inner
+                .route_table
+                .load(Ordering::Relaxed, guard)
+                .as_ref()
+                .unwrap()
+        };
+        for (k, v) in table.iter() {
+            for route in v {
                 if &route.route_key() == route_key && route.is_p2p() {
-                    return Some(*x.key());
+                    return Some(*k);
                 }
             }
         }
         None
     }
     pub fn need_punch(&self, id: &Ipv4Addr) -> bool {
-        if let Some(v) = self.inner.route_table.get(id) {
-            if v.value().iter().filter(|k| k.is_p2p()).count() >= self.inner.channel_num {
+        let guard = &crossbeam_epoch::pin();
+        let table = unsafe {
+            self.inner
+                .route_table
+                .load(Ordering::Relaxed, guard)
+                .as_ref()
+                .unwrap()
+        };
+        if let Some(v) = table.get(id) {
+            if v.iter().filter(|k| k.is_p2p()).count() >= self.inner.channel_num {
                 return false;
             }
         }
         true
     }
     pub fn route_table(&self) -> Vec<(Ipv4Addr, Vec<Route>)> {
-        self.inner
-            .route_table
-            .iter()
-            .map(|k| (k.key().clone(), k.value().clone()))
-            .collect()
+        let guard = &crossbeam_epoch::pin();
+        let table = unsafe {
+            self.inner
+                .route_table
+                .load(Ordering::Relaxed, guard)
+                .as_ref()
+                .unwrap()
+        };
+        table.iter().map(|(k, v)| (k.clone(), v.clone())).collect()
     }
     pub fn route_table_one(&self) -> Vec<(Ipv4Addr, Route)> {
-        let mut v = Vec::with_capacity(8);
-        for x in self.inner.route_table.iter() {
-            if let Some(route) = x.value().first() {
-                v.push((*x.key(), *route));
+        let mut list = Vec::with_capacity(8);
+        let guard = &crossbeam_epoch::pin();
+        let table = unsafe {
+            self.inner
+                .route_table
+                .load(Ordering::Relaxed, guard)
+                .as_ref()
+                .unwrap()
+        };
+        for (k, v) in table {
+            if let Some(route) = v.first() {
+                list.push((*k, *route));
             }
         }
-        v
+        list
     }
     pub fn direct_route_table_one(&self) -> Vec<(Ipv4Addr, Route)> {
-        let mut v = Vec::with_capacity(8);
-        for x in self.inner.route_table.iter() {
-            if let Some(route) = x.value().first() {
+        let mut list = Vec::with_capacity(8);
+        let guard = &crossbeam_epoch::pin();
+        let table = unsafe {
+            self.inner
+                .route_table
+                .load(Ordering::Relaxed, guard)
+                .as_ref()
+                .unwrap()
+        };
+        for (k, v) in table {
+            if let Some(route) = v.first() {
                 if route.metric == 1 {
-                    v.push((*x.key(), *route));
+                    list.push((*k, *route));
                 }
             }
         }
-        v
+        list
     }
-    pub fn remove_route_all(&self, id: &Ipv4Addr) {
-        if let Some((_, routes)) = self.inner.route_table.remove(id) {
-            for x in routes {
-                self.inner.route_table_time.remove(&(x.route_key(), *id));
+
+    pub fn remove_route(&self, id: &Ipv4Addr, route_key: RouteKey) {
+        let guard = &crossbeam_epoch::pin();
+        let route_table = &self.inner.route_table;
+        let mut table_share = route_table.load(Ordering::Relaxed, guard);
+        loop {
+            let mut table = unsafe { table_share.as_ref().unwrap().clone() };
+            if let Some(routes) = table.get_mut(id) {
+                routes.retain(|x| x.route_key() != route_key);
+                match route_table.compare_exchange(
+                    table_share,
+                    Owned::new(table),
+                    Ordering::Relaxed,
+                    Ordering::Relaxed,
+                    guard,
+                ) {
+                    Ok(p) => unsafe {
+                        guard.defer_destroy(p);
+                        self.inner.route_table_time.remove(&(route_key, *id));
+                        return;
+                    },
+                    Err(e) => {
+                        table_share = e.current;
+                    }
+                }
             }
         }
     }
-    pub fn remove_route(&self, id: &Ipv4Addr, route_key: RouteKey) {
-        if let Some(v) = self.inner.route_table.get(id) {
-            let mut routes = v.value().clone();
-            drop(v);
-            routes.retain(|x| x.route_key() != route_key);
-            self.inner.route_table.insert(*id, routes);
-        }
-        self.inner.route_table_time.remove(&(route_key, *id));
-    }
     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();
@@ -377,16 +544,16 @@ impl Channel {
 #[derive(Clone)]
 struct BufSenderGroup(
     usize,
-    Vec<tokio::sync::mpsc::Sender<(Block<'static>, usize, usize, RouteKey)>>,
+    Vec<std::sync::mpsc::SyncSender<(Block<'static>, usize, usize, RouteKey)>>,
 );
 
-struct BufReceiverGroup(Vec<tokio::sync::mpsc::Receiver<(Block<'static>, usize, usize, RouteKey)>>);
+struct BufReceiverGroup(Vec<std::sync::mpsc::Receiver<(Block<'static>, usize, usize, RouteKey)>>);
 
 impl BufSenderGroup {
-    pub async fn send(&mut self, val: (Block<'static>, usize, usize, RouteKey)) -> bool {
+    pub fn send(&mut self, val: (Block<'static>, usize, usize, RouteKey)) -> bool {
         let index = self.0 % self.1.len();
         self.0 = self.0.wrapping_add(1);
-        self.1[index].send(val).await.is_ok()
+        self.1[index].send(val).is_ok()
     }
 }
 
@@ -395,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) =
-            tokio::sync::mpsc::channel::<(Block<'static, Vec<u8>>, usize, usize, RouteKey)>(10);
+            std::sync::mpsc::sync_channel::<(Block<'static, Vec<u8>>, usize, usize, RouteKey)>(1);
         buf_sender_group.push(buf_sender);
         buf_receiver_group.push(buf_receiver);
     }
@@ -408,14 +575,15 @@ fn buf_channel_group(size: usize) -> (BufSenderGroup, BufReceiverGroup) {
 impl Channel {
     async fn tcp_handle(
         mut tcp_r: OwnedReadHalf,
-        mut buf_sender: BufSenderGroup,
+        context: Context,
+        handler: ChannelDataHandler,
         head_reserve: usize,
     ) -> io::Result<()> {
         let mut head = [0; 4];
         let addr = tcp_r.peer_addr()?;
-        let key = RouteKey::new(0, addr);
+        let key = RouteKey::new(TCP_ID, addr);
         loop {
-            let mut buf = POOL.alloc(4096);
+            let mut buf = [0; 4096];
             tcp_r.read_exact(&mut head).await?;
             let len = (((head[2] as u16) << 8) | head[3] as u16) as usize;
             if len < 12 || len > buf.len() {
@@ -427,15 +595,9 @@ impl Channel {
             tcp_r
                 .read_exact(&mut buf[head_reserve..head_reserve + len])
                 .await?;
-            if !buf_sender
-                .send((buf, head_reserve, head_reserve + len, key))
-                .await
-            {
-                return Err(io::Error::new(
-                    io::ErrorKind::Other,
-                    "buf_sender发送数据失败",
-                ));
-            }
+            handler
+                .handle(&mut buf, head_reserve, head_reserve + len, key, &context)
+                .await;
         }
     }
     async fn start_tcp(
@@ -443,14 +605,16 @@ impl Channel {
         tcp_stream: TcpStream,
         mut receiver: tokio::sync::mpsc::Receiver<Vec<u8>>,
         current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
-        buf_sender: BufSenderGroup,
+        context: Context,
+        handler: ChannelDataHandler,
         head_reserve: usize,
     ) {
         let (tcp_r, mut tcp_w) = tcp_stream.into_split();
         {
-            let buf_sender = buf_sender.clone();
+            let context = context.clone();
+            let handler = handler.clone();
             tokio::spawn(async move {
-                if let Err(e) = Self::tcp_handle(tcp_r, buf_sender, head_reserve).await {
+                if let Err(e) = Self::tcp_handle(tcp_r, context, handler, head_reserve).await {
                     log::info!("tcp链接断开:{:?}", e);
                 }
             });
@@ -480,9 +644,10 @@ impl Channel {
                                 Ok(tcp_stream) => {
                                     let (r, w) = tcp_stream.into_split();
                                     tcp_w = w;
-                                    let buf_sender = buf_sender.clone();
+                                    let context = context.clone();
+                                    let handler = handler.clone();
                                     tokio::spawn(async move {
-                                        if let Err(e) = Self::tcp_handle(r, buf_sender, head_reserve).await {
+                                        if let Err(e) = Self::tcp_handle(r, context,handler, head_reserve).await {
                                             log::info!("tcp 链接断开:{:?}",e);
                                         }
                                     });
@@ -513,17 +678,25 @@ impl Channel {
         let handler = self.handler.clone();
         let context = self.context;
         let main_channel = context.inner.main_channel.clone();
-        let buf_sender = if parallel > 1 || tcp.is_some() {
+        let buf_sender = if parallel > 1 {
             let (buf_sender, buf_receiver) = buf_channel_group(parallel);
-            for mut buf_receiver in buf_receiver.0 {
+            for 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;
-                    }
+                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!("异步处理停止");
+                    });
                 });
             }
             Some(buf_sender)
@@ -536,30 +709,57 @@ impl Channel {
                 tcp_stream,
                 receiver,
                 context.inner.current_device.clone(),
-                buf_sender.clone().unwrap(),
+                context.clone(),
+                handler.clone(),
                 head_reserve,
             ));
         }
         if let Some(main_channel_ipv6) = &context.inner.main_channel_ipv6 {
-            tokio::spawn(Self::start_(
-                worker.worker("main_channel_ipv6"),
-                context.clone(),
-                main_channel_ipv6.clone(),
-                handler.clone(),
-                buf_sender.clone(),
-                head_reserve,
-                true,
-            ));
+            let worker = worker.worker("main_channel_ipv6");
+            let context = context.clone();
+            let main_channel_ipv6 = main_channel_ipv6.clone();
+            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_(
+                    worker,
+                    context,
+                    UDP_V6_ID,
+                    main_channel_ipv6,
+                    handler,
+                    buf_sender,
+                    head_reserve,
+                ));
+            });
+        }
+        {
+            let worker = worker.worker("main_channel_1");
+            let context = context.clone();
+            let main_channel = main_channel.clone();
+            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_(
+                    worker,
+                    context,
+                    UDP_ID,
+                    main_channel,
+                    handler,
+                    buf_sender,
+                    head_reserve,
+                ));
+            });
         }
-        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;
@@ -611,6 +811,64 @@ impl Channel {
         }
         worker.stop_all();
     }
+    async fn main_start_(
+        worker: VntWorker,
+        context: Context,
+        id: usize,
+        udp: Arc<StdUdpSocket>,
+        handler: ChannelDataHandler,
+        buf_sender: Option<BufSenderGroup>,
+        head_reserve: usize,
+    ) {
+        match buf_sender {
+            None => {
+                let mut buf = [0; 4096];
+                loop {
+                    match udp.recv_from(&mut buf[head_reserve..]) {
+                        Ok((len, addr)) => {
+                            let end = head_reserve + len;
+                            if &buf[head_reserve..end] == b"stop" {
+                                if context.is_close() {
+                                    break;
+                                }
+                            }
+                            handler
+                                .handle(
+                                    &mut buf,
+                                    head_reserve,
+                                    end,
+                                    RouteKey::new(id, addr),
+                                    &context,
+                                )
+                                .await;
+                        }
+                        Err(e) => {
+                            log::error!("udp :{:?}", e);
+                        }
+                    }
+                }
+            }
+            Some(mut buf_sender) => loop {
+                let mut buf = POOL.alloc(4096);
+                match udp.recv_from(&mut buf[head_reserve..]) {
+                    Ok((len, addr)) => {
+                        let end = head_reserve + len;
+                        if &buf[head_reserve..end] == b"stop" {
+                            if context.is_close() {
+                                break;
+                            }
+                        }
+                        buf_sender.send((buf, head_reserve, end, RouteKey::new(id, addr)));
+                    }
+                    Err(e) => {
+                        log::error!("udp :{:?}", e);
+                    }
+                }
+            },
+        }
+
+        worker.stop_all();
+    }
     async fn start_(
         mut worker: VntWorker,
         context: Context,
@@ -624,12 +882,13 @@ impl Channel {
         #[cfg(target_os = "windows")]
         use std::os::windows::io::AsRawSocket;
         #[cfg(target_os = "windows")]
-        let id = 1 + 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 = 1 + udp.as_raw_fd() as usize;
-        context.inner.udp_map.insert(id, udp.clone());
+        let id = 3 + udp.as_raw_fd() as usize;
+
+        context.insert_udp(id, udp.clone());
         match buf_sender {
             None => {
                 let mut buf = [0; 4096];
@@ -677,7 +936,7 @@ impl Channel {
                     rs=udp.recv_from(&mut buf[head_reserve..])=>{
                          match rs {
                             Ok((len, addr)) => {
-                                if !buf_sender.send((buf,head_reserve,head_reserve+len,RouteKey::new(id, addr))).await{
+                                if !buf_sender.send((buf,head_reserve,head_reserve+len,RouteKey::new(id, addr))){
                                      log::error!("udp buf_sender发送数据失败");
                                      break;
                                 }
@@ -713,7 +972,7 @@ impl Channel {
                 }
             },
         }
-        context.inner.udp_map.remove(&id);
+        context.remove_udp(id);
         if is_core {
             worker.stop_all();
         }

vnt/src/channel/mod.rs

@@ -5,6 +5,10 @@ pub mod idle;
 pub mod punch;
 pub mod sender;
 
+const TCP_ID: usize = 0;
+const UDP_ID: usize = 1;
+const UDP_V6_ID: usize = 2;
+
 #[derive(Copy, Clone, Eq, PartialEq)]
 pub enum Status {
     Cone,
@@ -70,4 +74,7 @@ impl RouteKey {
     pub(crate) fn new(index: usize, addr: SocketAddr) -> Self {
         Self { index, addr }
     }
+    pub fn is_tcp(&self) -> bool {
+        self.index == TCP_ID
+    }
 }

vnt/src/channel/punch.rs

@@ -95,8 +95,7 @@ impl Punch {
         if !nat_info.local_ipv4_addr.ip().is_unspecified() && nat_info.local_ipv4_addr.port() != 0 {
             let _ = self
                 .context
-                .send_main_udp(buf, SocketAddr::V4(nat_info.local_ipv4_addr))
-                .await;
+                .send_main_udp(buf, SocketAddr::V4(nat_info.local_ipv4_addr));
         }
         if self.punch_model != PunchModel::IPv4
             && !nat_info.ipv6_addr.ip().is_unspecified()
@@ -104,8 +103,7 @@ impl Punch {
         {
             let rs = self
                 .context
-                .send_main_udp(buf, SocketAddr::V6(nat_info.ipv6_addr))
-                .await;
+                .send_main_udp(buf, SocketAddr::V6(nat_info.ipv6_addr));
             log::info!("发送到ipv6地址:{:?},rs={:?}", nat_info.ipv6_addr, rs);
             if rs.is_ok() && self.punch_model == PunchModel::IPv6 {
                 return Ok(());
@@ -167,11 +165,10 @@ impl Punch {
                 let is_cone = self.context.is_cone();
                 for ip in nat_info.public_ips {
                     let addr = SocketAddr::V4(SocketAddrV4::new(ip, nat_info.public_port));
-                    if is_cone {
-                        self.context.send_main_udp(buf, addr).await?;
-                    } else {
+                    self.context.send_main_udp(buf, addr)?;
+                    if !is_cone {
                         //只有一方是对称，则对称方要使用全部端口发送数据，符合上述计算的概率
-                        self.context.send_all(buf, addr).await?;
+                        self.context.try_send_all(buf, addr)?;
                     }
                     tokio::time::sleep(Duration::from_millis(2)).await;
                 }
@@ -195,7 +192,7 @@ impl Punch {
                     return Ok(());
                 }
                 let addr = SocketAddr::V4(SocketAddrV4::new(*pub_ip, *port));
-                self.context.send_main_udp(buf, addr).await?;
+                self.context.send_main_udp(buf, addr)?;
                 tokio::time::sleep(Duration::from_millis(2)).await;
             }
         }

vnt/src/cipher/aes_ecb.rs

@@ -1,8 +1,6 @@
 use crate::cipher::Finger;
-use crate::protocol::body::AesCbcSecretBody;
 use crate::protocol::{NetPacket, HEAD_LEN};
 use aes::cipher::{block_padding::Pkcs7, BlockDecryptMut, BlockEncryptMut, KeyInit};
-use rand::RngCore;
 use std::io;
 
 type Aes128EcbEnc = ecb::Encryptor<aes::Aes128>;
@@ -12,11 +10,11 @@ type Aes256EcbDec = ecb::Decryptor<aes::Aes256>;
 
 #[derive(Clone)]
 pub struct AesEcbCipher {
-    pub(crate) cipher: AesEcbEnum,
+    key: AesEcbEnum,
     pub(crate) finger: Option<Finger>,
 }
 
-#[derive(Clone)]
+#[derive(Clone, Copy)]
 pub enum AesEcbEnum {
     AES128ECB([u8; 16]),
     AES256ECB([u8; 32]),
@@ -24,7 +22,7 @@ pub enum AesEcbEnum {
 
 impl AesEcbCipher {
     pub fn key(&self) -> &[u8] {
-        match &self.cipher {
+        match &self.key {
             AesEcbEnum::AES128ECB(key) => key,
             AesEcbEnum::AES256ECB(key) => key,
         }
@@ -34,13 +32,13 @@ impl AesEcbCipher {
 impl AesEcbCipher {
     pub fn new_128(key: [u8; 16], finger: Option<Finger>) -> Self {
         Self {
-            cipher: AesEcbEnum::AES128ECB(key),
+            key: AesEcbEnum::AES128ECB(key),
             finger,
         }
     }
     pub fn new_256(key: [u8; 32], finger: Option<Finger>) -> Self {
         Self {
-            cipher: AesEcbEnum::AES256ECB(key),
+            key: AesEcbEnum::AES256ECB(key),
             finger,
         }
     }
@@ -57,37 +55,58 @@ impl AesEcbCipher {
             log::error!("数据异常,长度{}小于{}", net_packet.payload().len(), 16);
             return Err(io::Error::new(io::ErrorKind::Other, "data err"));
         }
-        let mut iv = [0; 16];
-        iv[0..4].copy_from_slice(&net_packet.source().octets());
-        iv[4..8].copy_from_slice(&net_packet.destination().octets());
-        iv[8] = net_packet.protocol().into();
-        iv[9] = net_packet.transport_protocol();
-        iv[10] = net_packet.is_gateway() as u8;
-        iv[11] = net_packet.source_ttl();
-        if let Some(finger) = &self.finger {
-            iv[12..16].copy_from_slice(&finger.hash[0..4]);
-        }
 
-        let mut secret_body =
-            AesCbcSecretBody::new(net_packet.payload_mut(), self.finger.is_some())?;
         if let Some(finger) = &self.finger {
-            let finger = finger.calculate_finger(&iv[..12], secret_body.en_body());
-            if &finger != secret_body.finger() {
+            let mut nonce_raw = [0; 12];
+            nonce_raw[0..4].copy_from_slice(&net_packet.source().octets());
+            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();
+            nonce_raw[10] = net_packet.is_gateway() as u8;
+            nonce_raw[11] = net_packet.source_ttl();
+            let len = net_packet.payload().len();
+            if len < 12 {
+                return Err(io::Error::new(io::ErrorKind::Other, "payload len <12"));
+            }
+            let secret_body = &net_packet.payload()[..len - 12];
+            let finger = finger.calculate_finger(&nonce_raw, secret_body);
+            if &finger != &net_packet.payload()[len - 12..] {
                 return Err(io::Error::new(io::ErrorKind::Other, "finger err"));
             }
+            net_packet.set_data_len(net_packet.data_len() - finger.len())?;
         }
-        let rs = match &self.cipher {
-            AesEcbEnum::AES128ECB(key) => Aes128EcbDec::new(&(*key).into())
-                .decrypt_padded_mut::<Pkcs7>(secret_body.en_body_mut()),
-            AesEcbEnum::AES256ECB(key) => Aes256EcbDec::new(&(*key).into())
-                .decrypt_padded_mut::<Pkcs7>(secret_body.en_body_mut()),
+        let mut out = [0u8; 1024 * 5];
+        let rs = match self.key {
+            AesEcbEnum::AES128ECB(key) => Aes128EcbDec::new(&key.into())
+                .decrypt_padded_b2b_mut::<Pkcs7>(net_packet.payload(), &mut out),
+            AesEcbEnum::AES256ECB(key) => Aes256EcbDec::new(&key.into())
+                .decrypt_padded_b2b_mut::<Pkcs7>(net_packet.payload(), &mut out),
         };
         match rs {
             Ok(buf) => {
-                let len = buf.len();
+                //校验头部
+                let src_net_packet = NetPacket::new(buf)?;
+                if src_net_packet.source() != net_packet.source() {
+                    return Err(io::Error::new(io::ErrorKind::Other, "data err"));
+                }
+                if src_net_packet.destination() != net_packet.destination() {
+                    return Err(io::Error::new(io::ErrorKind::Other, "data err"));
+                }
+                if src_net_packet.protocol() != net_packet.protocol() {
+                    return Err(io::Error::new(io::ErrorKind::Other, "data err"));
+                }
+                if src_net_packet.transport_protocol() != net_packet.transport_protocol() {
+                    return Err(io::Error::new(io::ErrorKind::Other, "data err"));
+                }
+                if src_net_packet.is_gateway() != net_packet.is_gateway() {
+                    return Err(io::Error::new(io::ErrorKind::Other, "data err"));
+                }
+                if src_net_packet.source_ttl() != net_packet.source_ttl() {
+                    return Err(io::Error::new(io::ErrorKind::Other, "data err"));
+                }
+                net_packet.set_data_len(buf.len())?;
+                net_packet.set_payload(src_net_packet.payload())?;
                 net_packet.set_encrypt_flag(false);
-                //减去末尾的随机数
-                net_packet.set_data_len(HEAD_LEN + len - 4)?;
                 Ok(())
             }
             Err(e) => Err(io::Error::new(
@@ -102,47 +121,35 @@ impl AesEcbCipher {
         &self,
         net_packet: &mut NetPacket<B>,
     ) -> io::Result<()> {
-        let data_len = net_packet.data_len();
-        let mut iv = [0; 16];
-        iv[0..4].copy_from_slice(&net_packet.source().octets());
-        iv[4..8].copy_from_slice(&net_packet.destination().octets());
-        iv[8] = net_packet.protocol().into();
-        iv[9] = net_packet.transport_protocol();
-        iv[10] = net_packet.is_gateway() as u8;
-        iv[11] = net_packet.source_ttl();
-        if let Some(finger) = &self.finger {
-            iv[12..16].copy_from_slice(&finger.hash[0..4]);
-            net_packet.set_data_len(data_len + 16)?;
-        } else {
-            net_packet.set_data_len(data_len + 4)?;
-        }
-        //先扩充随机数
-
-        let mut secret_body =
-            AesCbcSecretBody::new(net_packet.payload_mut(), self.finger.is_some())?;
-        secret_body.set_random(rand::thread_rng().next_u32());
-        let p_len = secret_body.en_body().len();
-        net_packet.set_data_len_max();
-        let rs = match &self.cipher {
-            AesEcbEnum::AES128ECB(key) => Aes128EcbEnc::new(&(*key).into())
-                .encrypt_padded_mut::<Pkcs7>(net_packet.payload_mut(), p_len),
-            AesEcbEnum::AES256ECB(key) => Aes256EcbEnc::new(&(*key).into())
-                .encrypt_padded_mut::<Pkcs7>(net_packet.payload_mut(), p_len),
+        let mut out = [0u8; 1024 * 5];
+        let rs = match self.key {
+            AesEcbEnum::AES128ECB(key) => Aes128EcbEnc::new(&key.into())
+                .encrypt_padded_b2b_mut::<Pkcs7>(net_packet.buffer(), &mut out),
+            AesEcbEnum::AES256ECB(key) => Aes256EcbEnc::new(&key.into())
+                .encrypt_padded_b2b_mut::<Pkcs7>(net_packet.buffer(), &mut out),
         };
+
         return match rs {
             Ok(buf) => {
-                let len = buf.len();
-                if let Some(finger) = &self.finger {
-                    let finger = finger.calculate_finger(&iv[..12], buf);
-                    //设置实际长度
-                    net_packet.set_data_len(HEAD_LEN + len + finger.len())?;
-                    let mut secret_body = AesCbcSecretBody::new(net_packet.payload_mut(), true)?;
-                    secret_body.set_finger(&finger)?;
-                } else {
-                    net_packet.set_data_len(HEAD_LEN + len)?;
-                }
-
+                net_packet.set_data_len(HEAD_LEN + buf.len())?;
+                net_packet.set_payload(buf)?;
                 net_packet.set_encrypt_flag(true);
+
+                if let Some(finger) = &self.finger {
+                    let mut nonce_raw = [0; 12];
+                    nonce_raw[0..4].copy_from_slice(&net_packet.source().octets());
+                    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();
+                    nonce_raw[10] = net_packet.is_gateway() as u8;
+                    nonce_raw[11] = net_packet.source_ttl();
+                    let finger = finger.calculate_finger(&nonce_raw, buf);
+                    let src_data_len = net_packet.data_len();
+                    //设置实际长度
+                    net_packet.set_data_len(src_data_len + finger.len())?;
+
+                    net_packet.buffer_mut()[src_data_len..].copy_from_slice(&finger);
+                }
                 Ok(())
             }
             Err(e) => Err(io::Error::new(
@@ -152,3 +159,11 @@ impl AesEcbCipher {
         };
     }
 }
+
+#[test]
+fn test_aes_ecb() {
+    let d = AesEcbCipher::new_128([0; 16], Some(Finger::new("123")));
+    let mut p = NetPacket::new_encrypt([0; 100]).unwrap();
+    d.encrypt_ipv4(&mut p).unwrap();
+    d.decrypt_ipv4(&mut p).unwrap();
+}

vnt/src/cipher/aes_gcm_cipher.rs

@@ -6,7 +6,7 @@ use aes_gcm::{AeadInPlace, Aes128Gcm, Aes256Gcm, Key, KeyInit, Nonce, Tag};
 use rand::RngCore;
 
 use crate::cipher::finger::Finger;
-use crate::protocol::{body::SecretBody, body::ENCRYPTION_RESERVED, NetPacket};
+use crate::protocol::{body::SecretBody, body::AES_GCM_ENCRYPTION_RESERVED, NetPacket};
 
 #[derive(Clone)]
 pub struct AesGcmCipher {
@@ -44,8 +44,8 @@ impl AesGcmCipher {
             //未加密的数据直接丢弃
             return Err(io::Error::new(io::ErrorKind::Other, "not encrypt"));
         }
-        if net_packet.payload().len() < ENCRYPTION_RESERVED {
-            log::error!("数据异常,长度小于{}", ENCRYPTION_RESERVED);
+        if net_packet.payload().len() < AES_GCM_ENCRYPTION_RESERVED {
+            log::error!("数据异常,长度小于{}", AES_GCM_ENCRYPTION_RESERVED);
             return Err(io::Error::new(io::ErrorKind::Other, "data err"));
         }
         let mut nonce_raw = [0; 12];
@@ -81,7 +81,7 @@ impl AesGcmCipher {
             ));
         }
         net_packet.set_encrypt_flag(false);
-        net_packet.set_data_len(net_packet.data_len() - ENCRYPTION_RESERVED)?;
+        net_packet.set_data_len(net_packet.data_len() - AES_GCM_ENCRYPTION_RESERVED)?;
         return Ok(());
     }
     /// net_packet 必须预留足够长度
@@ -90,7 +90,7 @@ impl AesGcmCipher {
         &self,
         net_packet: &mut NetPacket<B>,
     ) -> io::Result<()> {
-        if net_packet.reserve() < ENCRYPTION_RESERVED {
+        if net_packet.reserve() < AES_GCM_ENCRYPTION_RESERVED {
             return Err(io::Error::new(io::ErrorKind::Other, "too short"));
         }
         let mut nonce_raw = [0; 12];
@@ -101,7 +101,7 @@ impl AesGcmCipher {
         nonce_raw[10] = net_packet.is_gateway() as u8;
         nonce_raw[11] = net_packet.source_ttl();
         let nonce: &GenericArray<u8, U12> = Nonce::from_slice(&nonce_raw);
-        let data_len = net_packet.data_len() + ENCRYPTION_RESERVED;
+        let data_len = net_packet.data_len() + AES_GCM_ENCRYPTION_RESERVED;
         net_packet.set_data_len(data_len)?;
         let mut secret_body = SecretBody::new(net_packet.payload_mut(), self.finger.is_some())?;
         secret_body.set_random(rand::thread_rng().next_u32());

vnt/src/cipher/cipher.rs

@@ -1,6 +1,9 @@
+#[cfg(not(any(feature = "openssl-vendored", feature = "openssl")))]
 use crate::cipher::aes_ecb::AesEcbCipher;
 #[cfg(not(feature = "ring-cipher"))]
 use crate::cipher::aes_gcm_cipher::AesGcmCipher;
+#[cfg(any(feature = "openssl-vendored", feature = "openssl"))]
+use crate::cipher::openssl_aes_ecb::AesEcbCipher;
 #[cfg(feature = "ring-cipher")]
 use crate::cipher::ring_aes_gcm_cipher::AesGcmCipher;
 use crate::cipher::{aes_cbc, Finger};

vnt/src/cipher/mod.rs

@@ -1,4 +1,5 @@
 mod aes_cbc;
+#[cfg(not(any(feature = "openssl-vendored", feature = "openssl")))]
 mod aes_ecb;
 #[cfg(not(feature = "ring-cipher"))]
 mod aes_gcm_cipher;
@@ -8,6 +9,9 @@ mod finger;
 mod ring_aes_gcm_cipher;
 mod rsa_cipher;
 
+#[cfg(any(feature = "openssl-vendored", feature = "openssl"))]
+mod openssl_aes_ecb;
+
 pub use cipher::Cipher;
 pub use cipher::CipherModel;
 pub use finger::Finger;

vnt/src/cipher/openssl_aes_ecb.rs

@@ -0,0 +1,226 @@
+use crate::cipher::Finger;
+use crate::protocol::{NetPacket, HEAD_LEN};
+use libc::c_int;
+use openssl_sys::EVP_CIPHER_CTX;
+use std::{io, ptr};
+
+pub struct AesEcbCipher {
+    key: Vec<u8>,
+    pub(crate) en_ctx: *mut EVP_CIPHER_CTX,
+    pub(crate) de_ctx: *mut EVP_CIPHER_CTX,
+    pub(crate) finger: Option<Finger>,
+}
+
+impl Drop for AesEcbCipher {
+    fn drop(&mut self) {
+        unsafe {
+            openssl_sys::EVP_CIPHER_CTX_free(self.de_ctx);
+            openssl_sys::EVP_CIPHER_CTX_free(self.en_ctx);
+        }
+    }
+}
+
+impl Clone for AesEcbCipher {
+    fn clone(&self) -> Self {
+        if self.key.len() == 16 {
+            AesEcbCipher::new_128(self.key.clone().try_into().unwrap(), self.finger.clone())
+        } else {
+            AesEcbCipher::new_256(self.key.clone().try_into().unwrap(), self.finger.clone())
+        }
+    }
+}
+
+unsafe impl Sync for AesEcbCipher {}
+
+unsafe impl Send for AesEcbCipher {}
+
+impl AesEcbCipher {
+    pub fn key(&self) -> &[u8] {
+        &self.key
+    }
+}
+
+impl AesEcbCipher {
+    pub fn new_128(key: [u8; 16], finger: Option<Finger>) -> Self {
+        unsafe {
+            let cipher = openssl_sys::EVP_aes_128_ecb();
+            let en_ctx = openssl_sys::EVP_CIPHER_CTX_new();
+            openssl_sys::EVP_EncryptInit_ex(
+                en_ctx,
+                cipher,
+                ptr::null_mut(),
+                key.as_ptr(),
+                ptr::null(),
+            );
+
+            let de_ctx = openssl_sys::EVP_CIPHER_CTX_new();
+            openssl_sys::EVP_DecryptInit_ex(
+                de_ctx,
+                cipher,
+                ptr::null_mut(),
+                key.as_ptr(),
+                ptr::null(),
+            );
+            Self {
+                key: key.to_vec(),
+                en_ctx,
+                de_ctx,
+                finger,
+            }
+        }
+    }
+    pub fn new_256(key: [u8; 32], finger: Option<Finger>) -> Self {
+        unsafe {
+            let cipher = openssl_sys::EVP_aes_256_ecb();
+            let en_ctx = openssl_sys::EVP_CIPHER_CTX_new();
+            openssl_sys::EVP_EncryptInit_ex(
+                en_ctx,
+                cipher,
+                ptr::null_mut(),
+                key.as_ptr(),
+                ptr::null(),
+            );
+            let de_ctx = openssl_sys::EVP_CIPHER_CTX_new();
+            openssl_sys::EVP_DecryptInit_ex(
+                de_ctx,
+                cipher,
+                ptr::null_mut(),
+                key.as_ptr(),
+                ptr::null(),
+            );
+            Self {
+                key: key.to_vec(),
+                en_ctx,
+                de_ctx,
+                finger,
+            }
+        }
+    }
+
+    pub fn decrypt_ipv4<B: AsRef<[u8]> + AsMut<[u8]>>(
+        &self,
+        net_packet: &mut NetPacket<B>,
+    ) -> io::Result<()> {
+        if !net_packet.is_encrypt() {
+            //未加密的数据直接丢弃
+            return Err(io::Error::new(io::ErrorKind::Other, "not encrypt"));
+        }
+        if net_packet.payload().len() < 16 {
+            log::error!("数据异常,长度{}小于{}", net_packet.payload().len(), 16);
+            return Err(io::Error::new(io::ErrorKind::Other, "data err"));
+        }
+
+        if let Some(finger) = &self.finger {
+            let mut nonce_raw = [0; 12];
+            nonce_raw[0..4].copy_from_slice(&net_packet.source().octets());
+            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();
+            nonce_raw[10] = net_packet.is_gateway() as u8;
+            nonce_raw[11] = net_packet.source_ttl();
+            let len = net_packet.payload().len();
+            if len < 12 {
+                return Err(io::Error::new(io::ErrorKind::Other, "data len err"));
+            }
+            let secret_body = &net_packet.payload()[..len - 12];
+            let finger = finger.calculate_finger(&nonce_raw, secret_body);
+            if &finger != &net_packet.payload()[len - 12..] {
+                return Err(io::Error::new(io::ErrorKind::Other, "finger err"));
+            }
+            net_packet.set_data_len(net_packet.data_len() - finger.len())?;
+        }
+        let input = net_packet.payload();
+        let mut out = [0u8; 1024 * 5];
+        let mut out_len = 0;
+        let ctx = self.de_ctx;
+        unsafe {
+            let out_ptr = out.as_mut_ptr();
+            let in_len = input.len() as c_int;
+            openssl_sys::EVP_DecryptUpdate(ctx, out_ptr, &mut out_len, input.as_ptr(), in_len);
+            let mut last_len = 0;
+            openssl_sys::EVP_DecryptFinal_ex(ctx, out_ptr.offset(out_len as isize), &mut last_len);
+            out_len += last_len;
+        }
+        let out_len = out_len as usize;
+        let text = &out[..out_len];
+        {
+            //校验头部
+            let src_net_packet = NetPacket::new(text)?;
+            if src_net_packet.source() != net_packet.source() {
+                return Err(io::Error::new(io::ErrorKind::Other, "data err"));
+            }
+            if src_net_packet.destination() != net_packet.destination() {
+                return Err(io::Error::new(io::ErrorKind::Other, "data err"));
+            }
+            if src_net_packet.protocol() != net_packet.protocol() {
+                return Err(io::Error::new(io::ErrorKind::Other, "data err"));
+            }
+            if src_net_packet.transport_protocol() != net_packet.transport_protocol() {
+                return Err(io::Error::new(io::ErrorKind::Other, "data err"));
+            }
+            if src_net_packet.is_gateway() != net_packet.is_gateway() {
+                return Err(io::Error::new(io::ErrorKind::Other, "data err"));
+            }
+            if src_net_packet.source_ttl() != net_packet.source_ttl() {
+                return Err(io::Error::new(io::ErrorKind::Other, "data err"));
+            }
+        }
+        net_packet.set_encrypt_flag(false);
+        net_packet.set_data_len(out_len)?;
+        net_packet.set_payload(&text[12..])?;
+        Ok(())
+    }
+    /// net_packet 必须预留足够长度  大于 12+16+16
+    /// data_len是有效载荷的长度
+    pub fn encrypt_ipv4<B: AsRef<[u8]> + AsMut<[u8]>>(
+        &self,
+        net_packet: &mut NetPacket<B>,
+    ) -> io::Result<()> {
+        let input = net_packet.buffer();
+        let mut out = [0u8; 1024 * 5];
+        let mut out_len = 0;
+        let ctx = self.en_ctx;
+        //将头部也参与加密
+        unsafe {
+            let out_ptr = out.as_mut_ptr();
+            let in_len = input.len() as c_int;
+            openssl_sys::EVP_EncryptUpdate(ctx, out_ptr, &mut out_len, input.as_ptr(), in_len);
+            let mut last_len = 0;
+            openssl_sys::EVP_EncryptFinal_ex(ctx, out_ptr.offset(out_len as isize), &mut last_len);
+            out_len += last_len;
+        }
+        let out_len = out_len as usize;
+        if out_len == 0 {
+            return Err(io::Error::new(io::ErrorKind::Other, "ciphertext len err"));
+        }
+        //密文
+        let ciphertext = &out[..out_len];
+        net_packet.set_data_len(HEAD_LEN + out_len)?;
+        net_packet.payload_mut().copy_from_slice(ciphertext);
+        net_packet.set_encrypt_flag(true);
+        if let Some(finger) = &self.finger {
+            let mut nonce_raw = [0; 12];
+            nonce_raw[0..4].copy_from_slice(&net_packet.source().octets());
+            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();
+            nonce_raw[10] = net_packet.is_gateway() as u8;
+            nonce_raw[11] = net_packet.source_ttl();
+            let finger = finger.calculate_finger(&nonce_raw, ciphertext);
+            let src_data_len = net_packet.data_len();
+            //设置实际长度
+            net_packet.set_data_len(src_data_len + finger.len())?;
+
+            net_packet.buffer_mut()[src_data_len..].copy_from_slice(&finger);
+        }
+        Ok(())
+    }
+}
+
+#[test]
+fn test_openssl_aes_ecb() {
+    let d = AesEcbCipher::new_128([0; 16], Some(Finger::new("123")));
+    let mut p = NetPacket::new_encrypt([0; 100]).unwrap();
+    d.encrypt_ipv4(&mut p).unwrap();
+    d.decrypt_ipv4(&mut p).unwrap();
+}

vnt/src/cipher/ring_aes_gcm_cipher.rs

@@ -4,7 +4,7 @@ use ring::aead;
 use ring::aead::{LessSafeKey, UnboundKey};
 use std::io;
 
-use crate::protocol::body::{SecretBody, ENCRYPTION_RESERVED};
+use crate::protocol::body::{SecretBody, AES_GCM_ENCRYPTION_RESERVED};
 use crate::protocol::NetPacket;
 
 #[derive(Clone)]
@@ -58,8 +58,8 @@ impl AesGcmCipher {
             //未加密的数据直接丢弃
             return Err(io::Error::new(io::ErrorKind::Other, "not encrypt"));
         }
-        if net_packet.payload().len() < ENCRYPTION_RESERVED {
-            log::error!("数据异常,长度小于{}", ENCRYPTION_RESERVED);
+        if net_packet.payload().len() < AES_GCM_ENCRYPTION_RESERVED {
+            log::error!("数据异常,长度小于{}", AES_GCM_ENCRYPTION_RESERVED);
             return Err(io::Error::new(io::ErrorKind::Other, "data err"));
         }
         let mut nonce_raw = [0; 12];
@@ -93,7 +93,7 @@ impl AesGcmCipher {
             ));
         }
         net_packet.set_encrypt_flag(false);
-        net_packet.set_data_len(net_packet.data_len() - ENCRYPTION_RESERVED)?;
+        net_packet.set_data_len(net_packet.data_len() - AES_GCM_ENCRYPTION_RESERVED)?;
         return Ok(());
     }
     /// net_packet 必须预留足够长度
@@ -111,7 +111,7 @@ impl AesGcmCipher {
         nonce_raw[10] = net_packet.is_gateway() as u8;
         nonce_raw[11] = net_packet.source_ttl();
         let nonce = aead::Nonce::assume_unique_for_key(nonce_raw);
-        let data_len = net_packet.data_len() + ENCRYPTION_RESERVED;
+        let data_len = net_packet.data_len() + AES_GCM_ENCRYPTION_RESERVED;
         net_packet.set_data_len(data_len)?;
         let mut secret_body = SecretBody::new(net_packet.payload_mut(), self.finger.is_some())?;
         secret_body.set_random(rand::thread_rng().next_u32());

vnt/src/cipher/rsa_cipher.rs

@@ -1,4 +1,4 @@
-use crate::protocol::body::{RsaSecretBody, ENCRYPTION_RESERVED};
+use crate::protocol::body::{RsaSecretBody, RSA_ENCRYPTION_RESERVED};
 use crate::protocol::NetPacket;
 use rand::Rng;
 use rsa::pkcs8::der::Decode;
@@ -59,10 +59,10 @@ impl RsaCipher {
         &self,
         net_packet: &mut NetPacket<B>,
     ) -> io::Result<NetPacket<Vec<u8>>> {
-        if net_packet.reserve() < ENCRYPTION_RESERVED {
+        if net_packet.reserve() < RSA_ENCRYPTION_RESERVED {
             return Err(io::Error::new(io::ErrorKind::Other, "too short"));
         }
-        let data_len = net_packet.data_len() + ENCRYPTION_RESERVED;
+        let data_len = net_packet.data_len() + RSA_ENCRYPTION_RESERVED;
         net_packet.set_data_len(data_len)?;
         let mut nonce_raw = [0; 12];
         nonce_raw[0..4].copy_from_slice(&net_packet.source().octets());

vnt/src/core/mod.rs

@@ -7,7 +7,8 @@ use crossbeam_utils::atomic::AtomicCell;
 use dashmap::DashMap;
 use parking_lot::Mutex;
 use rand::Rng;
-use tokio::net::{TcpStream, UdpSocket};
+use std::net::UdpSocket;
+use tokio::net::TcpStream;
 use tokio::sync::mpsc::channel;
 
 use crate::channel::channel::{Channel, Context};
@@ -30,6 +31,7 @@ use crate::handle::{
     CurrentDeviceInfo, PeerDeviceInfo,
 };
 use crate::igmp_server::IgmpServer;
+use crate::ip_proxy::DashMapNew;
 use crate::nat::NatTest;
 use crate::tun_tap_device;
 use crate::tun_tap_device::{DeviceReader, DeviceWriter};
@@ -66,13 +68,23 @@ pub struct VntUtil {
 
 impl VntUtil {
     pub async fn new(config: Config) -> io::Result<VntUtil> {
-        let main_channel = UdpSocket::bind("0.0.0.0:0").await?;
-        let main_channel_ipv6 = match UdpSocket::bind("[::]:0").await {
-            Ok(main_channel_ipv6) => Some(main_channel_ipv6),
-            Err(e) => {
-                log::warn!("绑定ipv6地址失败:{}", e);
-                None
+        //单个udp用同步的性能更好，但是代理和多端口监听用异步更方便，这里将两者结合起来
+        let main_channel = UdpSocket::bind("0.0.0.0:0")?;
+        main_channel.set_write_timeout(Some(Duration::from_secs(5)))?;
+        main_channel.set_read_timeout(Some(Duration::from_secs(2)))?;
+        let main_channel_ipv6 = if config.punch_model != PunchModel::IPv4 {
+            match UdpSocket::bind("[::]:0") {
+                Ok(main_channel_ipv6) => {
+                    main_channel_ipv6.set_write_timeout(Some(Duration::from_secs(5)))?;
+                    Some(main_channel_ipv6)
+                }
+                Err(e) => {
+                    log::warn!("绑定ipv6地址失败:{}", e);
+                    None
+                }
             }
+        } else {
+            None
         };
         let server_cipher = if config.server_encrypt {
             let mut key = [0 as u8; 32];
@@ -181,7 +193,7 @@ impl VntUtil {
                 if self.config.password.is_none() {
                     1450
                 } else {
-                    1420
+                    1410
                 }
             }
             Some(mtu) => mtu,
@@ -205,6 +217,8 @@ impl VntUtil {
         Ok(driver_info)
     }
     pub async fn build(self) -> crate::Result<Vnt> {
+        //将读的超时时间清空
+        self.main_channel.set_read_timeout(None)?;
         let response = match self.response {
             None => {
                 return Err(Error::Stop("response None".to_string()));
@@ -266,9 +280,10 @@ impl VntUtil {
         ));
         let device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>> =
             Arc::new(Mutex::new((response.epoch, response.device_info_list)));
-        let peer_nat_info_map: Arc<DashMap<Ipv4Addr, NatInfo>> = Arc::new(DashMap::new());
+        let peer_nat_info_map: Arc<DashMap<Ipv4Addr, NatInfo>> = Arc::new(DashMap::new0());
         let connect_status = Arc::new(AtomicCell::new(ConnectStatus::Connected));
-
+        let public_ip = response.public_ip;
+        let public_port = response.public_port;
         let local_port = context.main_local_ipv4_port().unwrap_or(0);
 
         let local_ipv4_addr = crate::nat::local_ipv4_addr(local_port);
@@ -277,12 +292,11 @@ impl VntUtil {
         // NAT检测
         let nat_test = NatTest::new(
             config.stun_server.clone(),
-            response.public_ip,
-            response.public_port,
+            public_ip,
+            public_port,
             local_ipv4_addr,
             ipv6_addr,
-        )
-        .await;
+        );
         let in_external_route = if config.in_ips.is_empty() {
             None
         } else {
@@ -292,8 +306,11 @@ impl VntUtil {
             (None, None, None)
         } else {
             let (tcp_proxy, udp_proxy, ip_proxy_map) = crate::ip_proxy::init_proxy(
+                #[cfg(not(target_os = "android"))]
                 channel_sender.clone(),
+                #[cfg(not(target_os = "android"))]
                 current_device.clone(),
+                #[cfg(not(target_os = "android"))]
                 client_cipher.clone(),
             )
             .await?;
@@ -334,8 +351,7 @@ impl VntUtil {
                 client_cipher.clone(),
                 self.server_cipher.clone(),
                 config.parallel,
-            )
-            .await;
+            );
         }
         #[cfg(any(target_os = "android"))]
         tun_handler::start(
@@ -350,8 +366,7 @@ impl VntUtil {
             client_cipher.clone(),
             self.server_cipher.clone(),
             config.parallel,
-        )
-        .await;
+        );
 
         //外部数据接收处理
         let channel_recv_handler = ChannelDataHandler::new(
@@ -439,10 +454,11 @@ impl VntUtil {
             }
             let context = context.clone();
             let nat_test = nat_test.clone();
-            //延迟切换类型,避免无效流量
             tokio::spawn(async move {
-                tokio::time::sleep(Duration::from_secs(15)).await;
-                context.switch(nat_test.nat_info().nat_type);
+                let info = nat_test
+                    .re_test(public_ip, public_port, local_ipv4_addr, ipv6_addr)
+                    .await;
+                context.switch(info.nat_type);
             });
         }
         Ok(Vnt {
@@ -497,14 +513,15 @@ impl Vnt {
         self.context.route_table_one()
     }
     pub fn stop(&self) -> io::Result<()> {
-        self.context.close();
+        let _ = self.context.close();
         self.vnt_status_manager.stop_all();
-        self.device_writer.close()?;
+        let _ = self.device_writer.close();
         let virtual_gateway = self.current_device.load().virtual_gateway;
-        let _ = std::net::UdpSocket::bind("0.0.0.0:0")?.send_to(
-            &[0],
+        let _ = UdpSocket::bind("0.0.0.0:0")?.send_to(
+            b"stop",
             SocketAddr::V4(SocketAddrV4::new(virtual_gateway, 10000)),
         );
+
         Ok(())
     }
     pub async fn wait_stop(&mut self) {

vnt/src/handle/handshake_handler.rs

@@ -1,14 +1,15 @@
 use std::net::SocketAddr;
-use std::time::Duration;
 
 use protobuf::Message;
+use std::net::UdpSocket;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use tokio::net::{TcpStream, UdpSocket};
+use tokio::net::TcpStream;
 
 use crate::channel::channel::Context;
+use crate::channel::RouteKey;
 use crate::cipher::{Cipher, RsaCipher};
 use crate::proto::message::{HandshakeRequest, HandshakeResponse, SecretHandshakeRequest};
-use crate::protocol::body::ENCRYPTION_RESERVED;
+use crate::protocol::body::RSA_ENCRYPTION_RESERVED;
 use crate::protocol::{service_packet, NetPacket, Protocol, Version, MAX_TTL};
 
 pub enum HandshakeEnum {
@@ -44,7 +45,10 @@ fn secret_handshake_request_packet(
     request.token = token;
     request.key = key.to_vec();
     let bytes = request.write_to_bytes()?;
-    let mut net_packet = NetPacket::new_encrypt(vec![0u8; 12 + bytes.len() + ENCRYPTION_RESERVED])?;
+    let mut net_packet = NetPacket::new0(
+        12 + bytes.len(),
+        vec![0u8; 12 + bytes.len() + RSA_ENCRYPTION_RESERVED],
+    )?;
     net_packet.set_version(Version::V1);
     net_packet.set_gateway_flag(true);
     net_packet.set_protocol(Protocol::Service);
@@ -166,26 +170,18 @@ async fn send_recv(
         }
         Ok(len)
     } else {
-        if let Err(e) = main_channel.send_to(send_buf, server_address).await {
+        if let Err(e) = main_channel.send_to(send_buf, server_address) {
             return Err(HandshakeEnum::Other(format!("send error:{}", e)));
         }
-        match tokio::time::timeout(Duration::from_millis(300), main_channel.recv_from(recv_buf))
-            .await
-        {
-            Ok(rs) => match rs {
-                Ok((len, addr)) => {
-                    if server_address != addr {
-                        return Err(HandshakeEnum::Other(format!("invalid data,from {}", addr)));
-                    }
+        match main_channel.recv_from(recv_buf) {
+            Ok((len, addr)) => {
+                if server_address != addr {
+                    Err(HandshakeEnum::Other(format!("invalid data,from {}", addr)))
+                } else {
                     Ok(len)
                 }
-                Err(e) => {
-                    return Err(HandshakeEnum::Other(format!("receiver error:{}", e)));
-                }
-            },
-            Err(_) => {
-                return Err(HandshakeEnum::Timeout);
             }
+            Err(e) => Err(HandshakeEnum::Other(format!("receiver error:{}", e))),
         }
     }
 }
@@ -241,22 +237,20 @@ pub async fn secret_handshake(
     }
 }
 
-pub async fn secret_handshake_req(
+pub fn secret_handshake_req(
     context: &Context,
     server_address: SocketAddr,
     rsa_cipher: &RsaCipher,
     server_cipher: &Cipher,
     token: String,
+    route_key: &RouteKey,
 ) -> crate::Result<()> {
     let secret_packet =
         secret_handshake_request_packet(rsa_cipher, token, server_cipher.key().unwrap())?;
-    context
-        .send_main(secret_packet.buffer(), server_address)
-        .await?;
-    if context.is_main_tcp() {
-        context
-            .send_main_udp(secret_packet.buffer(), server_address)
-            .await?;
+    if route_key.is_tcp() {
+        context.send_main(secret_packet.buffer(), server_address)?;
+    } else {
+        context.send_main_udp(secret_packet.buffer(), server_address)?;
     }
     Ok(())
 }

vnt/src/handle/heartbeat_handler.rs

@@ -74,7 +74,7 @@ fn heartbeat_packet(
     gateway: bool,
     src: Ipv4Addr,
     dest: Ipv4Addr,
-) -> NetPacket<[u8; 48]> {
+) -> NetPacket<[u8; 12 + 4 + ENCRYPTION_RESERVED]> {
     let mut net_packet = NetPacket::new_encrypt([0u8; 12 + 4 + ENCRYPTION_RESERVED]).unwrap();
     net_packet.set_version(Version::V1);
     net_packet.set_protocol(Protocol::Control);
@@ -123,9 +123,7 @@ async fn start_heartbeat_(
             packet.set_source(current_dev.virtual_ip());
             packet.set_destination(current_dev.virtual_gateway);
             server_cipher.encrypt_ipv4(&mut packet)?;
-            let _ = sender
-                .send_main_udp(packet.buffer(), current_dev.connect_server)
-                .await;
+            let _ = sender.send_main_udp(packet.buffer(), current_dev.connect_server);
         }
         if count % 20 == 19 {
             if let Ok(mut addr) = server_address_str.to_socket_addrs() {
@@ -155,10 +153,7 @@ async fn start_heartbeat_(
             src,
             current_dev.virtual_gateway,
         );
-        if let Err(e) = sender
-            .send_main(server_packet.buffer(), current_dev.connect_server)
-            .await
-        {
+        if let Err(e) = sender.send_main(server_packet.buffer(), current_dev.connect_server) {
             log::warn!("connect_server:{:?},e:{:?}", current_dev.connect_server, e);
         }
         if count < 7 || count % 7 == 0 {
@@ -178,9 +173,8 @@ async fn start_heartbeat_(
                     peer.virtual_ip,
                 );
                 if let Some(route) = sender.route_one(&peer.virtual_ip) {
-                    if let Err(e) = sender
-                        .send_by_key(client_packet.buffer(), &route.route_key())
-                        .await
+                    if let Err(e) =
+                        sender.try_send_by_key(client_packet.buffer(), &route.route_key())
                     {
                         log::warn!("virtual_ip:{},route:{:?},e:{:?}", peer.virtual_ip, route, e);
                     }
@@ -189,9 +183,8 @@ async fn start_heartbeat_(
                     }
                 } else {
                     //没有直连路由则发送到网关
-                    if let Err(e) = sender
-                        .send_main(client_packet.buffer(), current_dev.connect_server)
-                        .await
+                    if let Err(e) =
+                        sender.send_main(client_packet.buffer(), current_dev.connect_server)
                     {
                         log::warn!(
                             "virtual_ip:{},connect_server:{:?},e:{:?}",
@@ -247,9 +240,8 @@ async fn start_heartbeat_(
                     *peer_ip,
                 );
                 for route in route_list {
-                    if let Err(e) = sender
-                        .send_by_key(client_packet.buffer(), &route.route_key())
-                        .await
+                    if let Err(e) =
+                        sender.try_send_by_key(client_packet.buffer(), &route.route_key())
                     {
                         log::warn!("peer_ip:{:?},route:{:?},e:{:?}", peer_ip, route, e);
                     }

vnt/src/handle/punch_handler.rs

@@ -137,9 +137,7 @@ async fn start_punch_(
             info.virtual_ip,
         )
         .unwrap();
-        let _ = sender
-            .send_main(packet.buffer(), current_device.connect_server)
-            .await;
+        let _ = sender.send_main(packet.buffer(), current_device.connect_server);
     }
     tokio::time::sleep(sleep_time).await;
     Ok(())

vnt/src/handle/recv_handler.rs

@@ -7,7 +7,6 @@ use parking_lot::Mutex;
 use protobuf::Message;
 use tokio::sync::mpsc::Sender;
 
-use packet::icmp::icmp::HeaderOther;
 use packet::icmp::{icmp, Kind};
 use packet::ip::ipv4;
 use packet::ip::ipv4::packet::IpV4Packet;
@@ -144,17 +143,13 @@ impl ChannelDataHandler {
                 // 转发
                 if let Some(route) = context.route_one(&destination) {
                     if route.metric <= net_packet.ttl() {
-                        context
-                            .send_by_key(net_packet.buffer(), &route.route_key())
-                            .await?;
+                        context.try_send_by_key(net_packet.buffer(), &route.route_key())?;
                     }
                 } else if (ttl > 1 || destination == current_device.virtual_gateway())
                     && source != current_device.virtual_gateway()
                 {
                     //网关默认要转发一次，生存时间不够的发到网关也会被丢弃
-                    context
-                        .send_main(net_packet.buffer(), current_device.connect_server)
-                        .await?;
+                    context.send_main(net_packet.buffer(), current_device.connect_server)?;
                 }
             }
             return Ok(());
@@ -171,8 +166,8 @@ impl ChannelDataHandler {
                         rsa_cipher,
                         &self.server_cipher,
                         self.token.clone(),
-                    )
-                    .await?;
+                        route_key,
+                    )?;
                 }
             } else {
                 //服务端解密
@@ -211,7 +206,7 @@ impl ChannelDataHandler {
                                         net_packet.set_destination(source);
                                         //不管加不加密，和接收到的数据长度都一致
                                         self.client_cipher.encrypt_ipv4(&mut net_packet)?;
-                                        context.send_by_key(net_packet.buffer(), route_key).await?;
+                                        context.try_send_by_key(net_packet.buffer(), route_key)?;
                                         return Ok(());
                                     }
                                 }
@@ -263,13 +258,14 @@ impl ChannelDataHandler {
                                                 .udp_proxy_map
                                                 .insert(key, SocketAddrV4::new(dest_ip, dest_port));
                                         }
+                                        #[cfg(not(target_os = "android"))]
                                         ipv4::protocol::Protocol::Icmp => {
                                             let dest_ip = ipv4.destination_ip();
                                             //转发到代理目标地址
                                             let icmp_packet =
                                                 icmp::IcmpPacket::new(ipv4.payload())?;
                                             match icmp_packet.header_other() {
-                                                HeaderOther::Identifier(id, seq) => {
+                                                icmp::HeaderOther::Identifier(id, seq) => {
                                                     ip_proxy_map
                                                         .icmp_proxy_map
                                                         .insert((dest_ip, id, seq), source);
@@ -360,9 +356,7 @@ impl ChannelDataHandler {
                 poll_device.set_protocol(Protocol::Service);
                 poll_device.set_transport_protocol(service_packet::Protocol::PollDeviceList.into());
                 self.server_cipher.encrypt_ipv4(&mut poll_device)?;
-                context
-                    .send_main(poll_device.buffer(), current_device.connect_server)
-                    .await?;
+                context.send_main(poll_device.buffer(), current_device.connect_server)?;
             }
         }
         Ok(())
@@ -383,7 +377,7 @@ impl ChannelDataHandler {
                 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.buffer(), route_key).await?;
+                context.try_send_by_key(net_packet.buffer(), route_key)?;
                 let route = Route::from(*route_key, metric, 199);
                 context.add_route_if_absent(source, route);
             }
@@ -409,7 +403,7 @@ impl ChannelDataHandler {
                 net_packet.set_destination(source);
                 net_packet.first_set_ttl(1);
                 self.client_cipher.encrypt_ipv4(&mut net_packet)?;
-                context.send_by_key(net_packet.buffer(), route_key).await?;
+                context.try_send_by_key(net_packet.buffer(), route_key)?;
                 let route = Route::from(*route_key, 1, 199);
                 context.add_route_if_absent(source, route);
             }
@@ -433,7 +427,7 @@ impl ChannelDataHandler {
                     addr_packet.set_ipv4(ipv4);
                     addr_packet.set_port(route_key.addr.port());
                     self.client_cipher.encrypt_ipv4(&mut packet)?;
-                    context.send_by_key(packet.buffer(), route_key).await?;
+                    context.try_send_by_key(packet.buffer(), route_key)?;
                 }
                 std::net::IpAddr::V6(_) => {}
             },
@@ -535,9 +529,7 @@ impl ChannelDataHandler {
                     // }
                     if self.punch(source, peer_nat_info).await {
                         self.client_cipher.encrypt_ipv4(&mut punch_packet)?;
-                        context
-                            .send_by_key(punch_packet.buffer(), route_key)
-                            .await?;
+                        context.try_send_by_key(punch_packet.buffer(), route_key)?;
                     }
                 } else {
                     self.punch(source, peer_nat_info).await;
@@ -624,9 +616,6 @@ impl ChannelDataHandler {
         net_packet: NetPacket<&[u8]>,
         route_key: &RouteKey,
     ) -> crate::Result<()> {
-        if net_packet.source() != current_device.virtual_gateway {
-            return Ok(());
-        }
         match ControlPacket::new(net_packet.transport_protocol(), net_packet.payload())? {
             ControlPacket::PongPacket(pong_packet) => {
                 let metric = net_packet.source_ttl() - net_packet.ttl() + 1;
@@ -677,7 +666,7 @@ impl ChannelDataHandler {
                         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
+                        let nat_info = nat_test
                             .re_test(
                                 Ipv4Addr::from(response.public_ip),
                                 response.public_port as u16,
@@ -778,9 +767,7 @@ impl ChannelDataHandler {
                 }
 
                 self.connect_status.store(ConnectStatus::Connecting);
-                self.register
-                    .fast_register(current_device.virtual_ip)
-                    .await?;
+                self.register.fast_register(current_device.virtual_ip)?;
             }
             InErrorPacket::AddressExhausted => {
                 //地址用尽

vnt/src/handle/registration_handler.rs

@@ -6,8 +6,9 @@ use crate::channel::sender::ChannelSender;
 use crate::cipher::Cipher;
 use crate::handle::PeerDeviceInfo;
 use protobuf::Message;
+use std::net::UdpSocket;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use tokio::net::{TcpStream, UdpSocket};
+use tokio::net::TcpStream;
 
 use crate::proto::message::{RegistrationRequest, RegistrationResponse};
 use crate::protocol::body::ENCRYPTION_RESERVED;
@@ -81,28 +82,18 @@ pub async fn registration(
         }
         &mut recv_buf[4..len]
     } else {
-        if let Err(e) = main_channel.send_to(buf, server_address).await {
+        if let Err(e) = main_channel.send_to(buf, server_address) {
             return Err(ReqEnum::Other(format!("send error:{}", e)));
         }
-        match tokio::time::timeout(
-            Duration::from_millis(300),
-            main_channel.recv_from(&mut recv_buf),
-        )
-        .await
-        {
-            Ok(rs) => match rs {
-                Ok((len, addr)) => {
-                    if server_address != addr {
-                        return Err(ReqEnum::Other(format!("invalid data,from {}", addr)));
-                    }
-                    &mut recv_buf[..len]
+        match main_channel.recv_from(&mut recv_buf) {
+            Ok((len, addr)) => {
+                if server_address != addr {
+                    return Err(ReqEnum::Other(format!("invalid data,from {}", addr)));
                 }
-                Err(e) => {
-                    return Err(ReqEnum::Other(format!("receiver error:{}", e)));
-                }
-            },
-            Err(_) => {
-                return Err(ReqEnum::Timeout);
+                &mut recv_buf[..len]
+            }
+            Err(e) => {
+                return Err(ReqEnum::Other(format!("receiver error:{}", e)));
             }
         }
     };
@@ -236,7 +227,7 @@ impl Register {
             client_secret,
         }
     }
-    pub async fn fast_register(&self, ip: Ipv4Addr) -> crate::Result<()> {
+    pub fn fast_register(&self, ip: Ipv4Addr) -> crate::Result<()> {
         let last = self.time.load();
         if last.elapsed() < Duration::from_secs(2)
             || self.time.compare_exchange(last, Instant::now()).is_err()
@@ -256,7 +247,7 @@ impl Register {
             self.client_secret,
         )?;
         let buf = request_packet.buffer();
-        self.sender.send_main(buf, self.server_address).await?;
+        self.sender.send_main(buf, self.server_address)?;
         Ok(())
     }
 }

vnt/src/handle/tun_tap/channel_group.rs

@@ -3,16 +3,16 @@ use byte_pool::Block;
 #[derive(Clone)]
 pub struct BufSenderGroup(
     usize,
-    Vec<tokio::sync::mpsc::Sender<(Block<'static>, usize, usize)>>,
+    Vec<std::sync::mpsc::SyncSender<(Block<'static>, usize, usize)>>,
 );
 
-pub struct BufReceiverGroup(pub Vec<tokio::sync::mpsc::Receiver<(Block<'static>, usize, usize)>>);
+pub struct BufReceiverGroup(pub Vec<std::sync::mpsc::Receiver<(Block<'static>, usize, usize)>>);
 
 impl BufSenderGroup {
-    pub async fn send(&mut self, val: (Block<'static>, usize, usize)) -> bool {
+    pub fn send(&mut self, val: (Block<'static>, usize, usize)) -> bool {
         let index = self.0 % self.1.len();
         self.0 = self.0.wrapping_add(1);
-        self.1[index].send(val).await.is_ok()
+        self.1[index].send(val).is_ok()
     }
 }
 
@@ -21,7 +21,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) =
-            tokio::sync::mpsc::channel::<(Block<'static>, usize, usize)>(10);
+            std::sync::mpsc::sync_channel::<(Block<'static>, usize, usize)>(1);
         buf_sender_group.push(buf_sender);
         buf_receiver_group.push(buf_receiver);
     }

vnt/src/handle/tun_tap/mod.rs

@@ -17,12 +17,13 @@ use parking_lot::RwLock;
 use std::io;
 use std::net::{Ipv4Addr, SocketAddrV4};
 use std::sync::Arc;
+
 pub mod channel_group;
 #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))]
 pub mod tap_handler;
 pub mod tun_handler;
 
-async fn broadcast(
+fn broadcast(
     server_cipher: &Cipher,
     multicast_members: Option<Arc<RwLock<Multicast>>>,
     sender: &ChannelSender,
@@ -47,8 +48,7 @@ async fn broadcast(
         }
         if route.is_p2p()
             && sender
-                .send_by_key(net_packet.buffer(), &route.route_key())
-                .await
+                .try_send_by_key(net_packet.buffer(), &route.route_key())
                 .is_ok()
         {
             peer_ips.push(peer_ip);
@@ -62,9 +62,7 @@ async fn broadcast(
     }
     //转发到服务端的可选择广播，还要进行服务端加密
     if peer_ips.is_empty() {
-        sender
-            .send_main(net_packet.buffer(), current_device.connect_server)
-            .await?;
+        sender.send_main(net_packet.buffer(), current_device.connect_server)?;
     } else {
         let buf = vec![
             0 as u8;
@@ -85,9 +83,7 @@ async fn broadcast(
         broadcast.set_address(&peer_ips)?;
         broadcast.set_data(net_packet.buffer())?;
         server_cipher.encrypt_ipv4(&mut server_packet)?;
-        sender
-            .send_main(server_packet.buffer(), current_device.connect_server)
-            .await?;
+        sender.send_main(server_packet.buffer(), current_device.connect_server)?;
     }
     Ok(())
 }
@@ -96,7 +92,7 @@ async fn broadcast(
 /// |12字节开头|ip报文|至少1024字节结尾|
 ///
 #[inline]
-pub async fn base_handle(
+pub fn base_handle(
     sender: &ChannelSender,
     buf: &mut [u8],
     data_len: usize, //数据总长度=12+ip包长度
@@ -126,9 +122,7 @@ pub async fn base_handle(
         if protocol == Protocol::Icmp {
             net_packet.set_gateway_flag(true);
             server_cipher.encrypt_ipv4(&mut net_packet)?;
-            sender
-                .send_main(net_packet.buffer(), current_device.connect_server)
-                .await?;
+            sender.send_main(net_packet.buffer(), current_device.connect_server)?;
         }
         return Ok(());
     }
@@ -140,9 +134,7 @@ pub async fn base_handle(
                     net_packet.set_destination(current_device.virtual_gateway);
                     net_packet.set_gateway_flag(true);
                     server_cipher.encrypt_ipv4(&mut net_packet)?;
-                    sender
-                        .send_main(net_packet.buffer(), current_device.connect_server)
-                        .await?;
+                    sender.send_main(net_packet.buffer(), current_device.connect_server)?;
                 }
             }
             Protocol::Udp => {
@@ -160,8 +152,7 @@ pub async fn base_handle(
                     sender,
                     &mut net_packet,
                     &current_device,
-                )
-                .await?;
+                )?;
             }
             _ => {}
         }
@@ -176,8 +167,7 @@ pub async fn base_handle(
             sender,
             &mut net_packet,
             &current_device,
-        )
-        .await?;
+        )?;
         return Ok(());
     }
     if !check_dest(
@@ -256,13 +246,10 @@ pub async fn base_handle(
     client_cipher.encrypt_ipv4(&mut net_packet)?;
     //优先发到直连到地址
     if sender
-        .send_by_id(net_packet.buffer(), &dest_ip)
-        .await
+        .try_send_by_id(net_packet.buffer(), &dest_ip)
         .is_err()
     {
-        sender
-            .send_main(net_packet.buffer(), current_device.connect_server)
-            .await?;
+        sender.send_main(net_packet.buffer(), current_device.connect_server)?;
     }
     return Ok(());
 }

vnt/src/handle/tun_tap/tap_handler.rs

@@ -43,33 +43,27 @@ pub fn start(
         thread::Builder::new()
             .name("tap_handler".into())
             .spawn(move || {
-                tokio::runtime::Builder::new_current_thread()
-                    .enable_all()
-                    .build()
-                    .unwrap()
-                    .block_on(async move {
-                        if let Err(e) = start_simple(
-                            sender,
-                            device_reader,
-                            device_writer,
-                            igmp_server,
-                            current_device,
-                            ip_route,
-                            ip_proxy_map,
-                            client_cipher,
-                            server_cipher,
-                        )
-                        .await
-                        {
-                            log::warn!("tap:{:?}", e);
-                        }
-                        worker.stop_all();
-                    });
+                if let Err(e) = start_simple(
+                    &sender,
+                    device_reader,
+                    &device_writer,
+                    igmp_server,
+                    current_device,
+                    ip_route,
+                    ip_proxy_map,
+                    client_cipher,
+                    server_cipher,
+                ) {
+                    log::warn!("tap:{:?}", e);
+                }
+                let _ = sender.close();
+                let _ = device_writer.close();
+                worker.stop_all();
             })
             .unwrap();
     } else {
         let (buf_sender, buf_receiver) = buf_channel_group(parallel);
-        for mut buf_receiver in buf_receiver.0 {
+        for buf_receiver in buf_receiver.0 {
             let sender = sender.clone();
             let device_writer = device_writer.clone();
             let igmp_server = igmp_server.clone();
@@ -78,8 +72,8 @@ pub fn start(
             let ip_proxy_map = ip_proxy_map.clone();
             let client_cipher = client_cipher.clone();
             let server_cipher = server_cipher.clone();
-            tokio::spawn(async move {
-                while let Some((mut buf, _, len)) = buf_receiver.recv().await {
+            thread::spawn(move || {
+                while let Ok((mut buf, _, len)) = buf_receiver.recv() {
                     match handle(
                         &mut buf,
                         len,
@@ -91,37 +85,33 @@ pub fn start(
                         &ip_proxy_map,
                         &client_cipher,
                         &server_cipher,
-                    )
-                    .await
-                    {
+                    ) {
                         Ok(_) => {}
                         Err(e) => {
                             log::warn!("{:?}", e)
                         }
                     }
                 }
+                let _ = sender.close();
+                let _ = device_writer.close();
             });
         }
         thread::Builder::new()
             .name("tap_handler".into())
             .spawn(move || {
-                tokio::runtime::Builder::new_current_thread()
-                    .enable_all()
-                    .build()
-                    .unwrap()
-                    .block_on(async move {
-                        if let Err(e) = start_(sender, device_reader, buf_sender).await {
-                            log::warn!("tap:{:?}", e);
-                        }
-                        worker.stop_all();
-                    });
+                if let Err(e) = start_(&sender, device_reader, buf_sender) {
+                    log::warn!("tap:{:?}", e);
+                }
+                let _ = sender.close();
+                let _ = device_writer.close();
+                worker.stop_all();
             })
             .unwrap();
     }
 }
 
-async fn start_(
-    sender: ChannelSender,
+fn start_(
+    sender: &ChannelSender,
     device_reader: DeviceReader,
     mut buf_sender: BufSenderGroup,
 ) -> io::Result<()> {
@@ -132,7 +122,7 @@ async fn start_(
         }
         let start = 0;
         let len = device_reader.read(&mut buf)?;
-        if !buf_sender.send((buf, start, len)).await {
+        if !buf_sender.send((buf, start, len)) {
             return Err(io::Error::new(
                 io::ErrorKind::Other,
                 "tap buf_sender发送失败",
@@ -141,10 +131,10 @@ async fn start_(
     }
 }
 
-async fn start_simple(
-    sender: ChannelSender,
+fn start_simple(
+    sender: &ChannelSender,
     device_reader: DeviceReader,
-    device_writer: DeviceWriter,
+    device_writer: &DeviceWriter,
     igmp_server: Option<IgmpServer>,
     current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
     ip_route: Option<ExternalRoute>,
@@ -160,21 +150,19 @@ async fn start_simple(
             len,
             &igmp_server,
             &current_device,
-            &device_writer,
-            &sender,
+            device_writer,
+            sender,
             &ip_route,
             &ip_proxy_map,
             &client_cipher,
             &server_cipher,
-        )
-        .await
-        {
+        ) {
             log::warn!("tap handle{:?}", e);
         }
     }
 }
 
-async fn handle(
+fn handle(
     buf: &mut [u8],
     len: usize,
     igmp_server: &Option<IgmpServer>,
@@ -261,8 +249,7 @@ async fn handle(
                 proxy_map,
                 client_cipher,
                 server_cipher,
-            )
-            .await;
+            );
         }
         _ => {
             // log::warn!("不支持的二层协议：{:?}",p)

vnt/src/handle/tun_tap/tun_handler.rs

@@ -40,7 +40,7 @@ fn icmp(device_writer: &DeviceWriter, mut ipv4_packet: IpV4Packet<&mut [u8]>) ->
 
 /// 接收tun数据，并且转发到udp上
 #[inline]
-async fn handle(
+fn handle(
     sender: &ChannelSender,
     data: &mut [u8],
     len: usize,
@@ -75,11 +75,10 @@ async fn handle(
         proxy_map,
         client_cipher,
         server_cipher,
-    )
-    .await;
+    );
 }
 
-pub async fn start(
+pub fn start(
     worker: VntWorker,
     sender: ChannelSender,
     device_reader: DeviceReader,
@@ -96,34 +95,27 @@ pub async fn start(
         thread::Builder::new()
             .name("tun_handler".into())
             .spawn(move || {
-                tokio::runtime::Builder::new_current_thread()
-                    .enable_all()
-                    .build()
-                    .unwrap()
-                    .block_on(async move {
-                        if let Err(e) = start_simple(
-                            sender,
-                            device_reader,
-                            &device_writer,
-                            igmp_server,
-                            current_device,
-                            ip_route,
-                            ip_proxy_map,
-                            client_cipher,
-                            server_cipher,
-                        )
-                        .await
-                        {
-                            log::warn!("stop:{}", e);
-                        }
-                        let _ = device_writer.close();
-                        worker.stop_all();
-                    })
+                if let Err(e) = start_simple(
+                    &sender,
+                    device_reader,
+                    &device_writer,
+                    igmp_server,
+                    current_device,
+                    ip_route,
+                    ip_proxy_map,
+                    client_cipher,
+                    server_cipher,
+                ) {
+                    log::warn!("stop:{}", e);
+                }
+                let _ = sender.close();
+                let _ = device_writer.close();
+                worker.stop_all();
             })
             .unwrap();
     } else {
         let (buf_sender, buf_receiver) = buf_channel_group(parallel);
-        for mut buf_receiver in buf_receiver.0 {
+        for buf_receiver in buf_receiver.0 {
             let sender = sender.clone();
             let device_writer = device_writer.clone();
             let igmp_server = igmp_server.clone();
@@ -132,8 +124,8 @@ pub async fn start(
             let ip_proxy_map = ip_proxy_map.clone();
             let client_cipher = client_cipher.clone();
             let server_cipher = server_cipher.clone();
-            tokio::spawn(async move {
-                while let Some((mut buf, start, len)) = buf_receiver.recv().await {
+            thread::spawn(move || {
+                while let Ok((mut buf, start, len)) = buf_receiver.recv() {
                     match handle(
                         &sender,
                         &mut buf[start..],
@@ -145,39 +137,34 @@ pub async fn start(
                         &ip_proxy_map,
                         &client_cipher,
                         &server_cipher,
-                    )
-                    .await
-                    {
+                    ) {
                         Ok(_) => {}
                         Err(e) => {
                             log::warn!("{:?}", e)
                         }
                     }
                 }
+                let _ = sender.close();
+                let _ = device_writer.close();
             });
         }
 
         thread::Builder::new()
             .name("tun_handler".into())
             .spawn(move || {
-                tokio::runtime::Builder::new_current_thread()
-                    .enable_all()
-                    .build()
-                    .unwrap()
-                    .block_on(async move {
-                        if let Err(e) = start_(sender, device_reader, buf_sender).await {
-                            log::warn!("stop:{}", e);
-                        }
-                        let _ = device_writer.close();
-                        worker.stop_all();
-                    })
+                if let Err(e) = start_(&sender, device_reader, buf_sender) {
+                    log::warn!("stop:{}", e);
+                }
+                let _ = sender.close();
+                let _ = device_writer.close();
+                worker.stop_all();
             })
             .unwrap();
     }
 }
 
-async fn start_(
-    sender: ChannelSender,
+fn start_(
+    sender: &ChannelSender,
     device_reader: DeviceReader,
     mut buf_sender: BufSenderGroup,
 ) -> io::Result<()> {
@@ -191,7 +178,7 @@ async fn start_(
         let len = device_reader.read(&mut buf[12..])? + 12;
         #[cfg(any(target_os = "macos"))]
         let start = 4;
-        if !buf_sender.send((buf, start, len)).await {
+        if !buf_sender.send((buf, start, len)) {
             return Err(io::Error::new(
                 io::ErrorKind::Other,
                 "tun buf_sender发送失败",
@@ -200,8 +187,8 @@ async fn start_(
     }
 }
 
-async fn start_simple(
-    sender: ChannelSender,
+fn start_simple(
+    sender: &ChannelSender,
     device_reader: DeviceReader,
     device_writer: &DeviceWriter,
     igmp_server: Option<IgmpServer>,
@@ -221,7 +208,7 @@ async fn start_simple(
         #[cfg(any(target_os = "macos"))]
         let mut buf = &mut buf[4..];
         match handle(
-            &sender,
+            sender,
             &mut buf,
             len,
             device_writer,
@@ -231,9 +218,7 @@ async fn start_simple(
             &ip_proxy_map,
             &client_cipher,
             &server_cipher,
-        )
-        .await
-        {
+        ) {
             Ok(_) => {}
             Err(e) => {
                 log::warn!("{:?}", e)

vnt/src/igmp_server/mod.rs

@@ -1,15 +1,19 @@
-use crate::tun_tap_device::DeviceWriter;
-use dashmap::DashMap;
-use packet::igmp::igmp_v2::IgmpV2Packet;
-use packet::igmp::igmp_v3::{IgmpV3QueryPacket, IgmpV3RecordType, IgmpV3ReportPacket};
-use packet::igmp::IgmpType;
-use packet::ip::ipv4::protocol::Protocol;
-use parking_lot::RwLock;
 use std::collections::{HashMap, HashSet};
 use std::net::Ipv4Addr;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
+use dashmap::DashMap;
+use parking_lot::RwLock;
+
+use packet::igmp::igmp_v2::IgmpV2Packet;
+use packet::igmp::igmp_v3::{IgmpV3QueryPacket, IgmpV3RecordType, IgmpV3ReportPacket};
+use packet::igmp::IgmpType;
+use packet::ip::ipv4::protocol::Protocol;
+
+use crate::ip_proxy::DashMapNew;
+use crate::tun_tap_device::DeviceWriter;
+
 //1. 定时发送query，启动时20秒一次，连发3次，之后8分钟一次
 //2. 接收网关的igmp report 维护组播源信息
 #[derive(Clone, Debug)]
@@ -52,7 +56,7 @@ pub struct IgmpServer {
 
 impl IgmpServer {
     pub fn new(device_writer: DeviceWriter) -> Self {
-        let multicast: Arc<DashMap<Ipv4Addr, Arc<RwLock<Multicast>>>> = Arc::new(DashMap::new());
+        let multicast: Arc<DashMap<Ipv4Addr, Arc<RwLock<Multicast>>>> = Arc::new(DashMap::new0());
         std::thread::spawn(move || {
             //预留以太网帧头和ip头
             let mut buf = [0; 14 + 24 + 12];

vnt/src/ip_proxy/icmp_proxy.rs

@@ -109,7 +109,7 @@ impl IcmpProxy {
                                                             )
                                                             .is_err()
                                                         {
-                                                            let _ = self.sender.try_send_main(
+                                                            let _ = self.sender.send_main(
                                                                 net_packet.buffer(),
                                                                 connect_server,
                                                             );

vnt/src/ip_proxy/mod.rs

@@ -1,21 +1,48 @@
-use crate::channel::sender::ChannelSender;
-use crate::cipher::Cipher;
-use crate::handle::CurrentDeviceInfo;
-use crate::ip_proxy::icmp_proxy::IcmpProxy;
 use crate::ip_proxy::tcp_proxy::TcpProxy;
 use crate::ip_proxy::udp_proxy::UdpProxy;
-use crossbeam_utils::atomic::AtomicCell;
 use dashmap::DashMap;
+#[cfg(not(target_os = "android"))]
 use socket2::{SockAddr, Socket};
-use std::net::{Ipv4Addr, SocketAddrV4};
+#[cfg(not(target_os = "android"))]
+use std::net::Ipv4Addr;
+use std::net::SocketAddrV4;
 use std::sync::Arc;
 use std::{io, thread};
 use tokio::net::{TcpListener, UdpSocket};
 
+#[cfg(not(target_os = "android"))]
 pub mod icmp_proxy;
 pub mod tcp_proxy;
 pub mod udp_proxy;
 
+pub trait DashMapNew {
+    fn new0() -> Self;
+    fn new_cap(capacity: usize) -> Self;
+}
+
+impl<'a, K: 'a + Eq + std::hash::Hash, V: 'a> DashMapNew for DashMap<K, V> {
+    fn new0() -> Self {
+        Self::new_cap(0)
+    }
+
+    fn new_cap(capacity: usize) -> Self {
+        let shard_amount = (thread::available_parallelism().map_or(4, |v| {
+            // https://github.com/rust-lang/rust/issues/115868
+            let n: usize = v.get() * 4;
+            if n == 0 {
+                log::warn!("available_parallelism=0");
+                println!("warn available_parallelism=0");
+            }
+            if n < 4 {
+                return 4;
+            }
+            n
+        }))
+        .next_power_of_two();
+        DashMap::with_capacity_and_shard_amount(capacity, shard_amount)
+    }
+}
+
 #[derive(Eq, PartialEq, Ord, PartialOrd, Copy, Clone, Debug)]
 pub enum Protocol {
     Icmp,
@@ -31,11 +58,14 @@ pub struct IpProxyMap {
     pub(crate) tcp_proxy_map: Arc<DashMap<SocketAddrV4, SocketAddrV4>>,
     pub(crate) udp_proxy_map: Arc<DashMap<SocketAddrV4, SocketAddrV4>>,
     // icmp用Identifier来区分，没有Identifier的一律不转发
+    #[cfg(not(target_os = "android"))]
     pub(crate) icmp_proxy_map: Arc<DashMap<(Ipv4Addr, u16, u16), Ipv4Addr>>,
+    #[cfg(not(target_os = "android"))]
     icmp_socket: Arc<Socket>,
 }
 
 impl IpProxyMap {
+    #[cfg(not(target_os = "android"))]
     pub fn send_icmp(&self, buf: &[u8], dest: &Ipv4Addr) -> io::Result<usize> {
         self.icmp_socket
             .send_to(buf, &SockAddr::from(SocketAddrV4::new(*dest, 0)))
@@ -43,31 +73,38 @@ impl IpProxyMap {
 }
 
 pub async fn init_proxy(
-    sender: ChannelSender,
-    current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
-    client_cipher: Cipher,
+    #[cfg(not(target_os = "android"))] sender: crate::channel::sender::ChannelSender,
+    #[cfg(not(target_os = "android"))] current_device: Arc<
+        crossbeam_utils::atomic::AtomicCell<crate::handle::CurrentDeviceInfo>,
+    >,
+    #[cfg(not(target_os = "android"))] client_cipher: crate::cipher::Cipher,
 ) -> io::Result<(TcpProxy, UdpProxy, IpProxyMap)> {
-    let tcp_proxy_map: Arc<DashMap<SocketAddrV4, SocketAddrV4>> = Arc::new(DashMap::new());
-    let udp_proxy_map: Arc<DashMap<SocketAddrV4, SocketAddrV4>> = Arc::new(DashMap::new());
-    let icmp_proxy_map: Arc<DashMap<(Ipv4Addr, u16, u16), Ipv4Addr>> = Arc::new(DashMap::new());
+    let tcp_proxy_map: Arc<DashMap<SocketAddrV4, SocketAddrV4>> = Arc::new(DashMap::new0());
+    let udp_proxy_map: Arc<DashMap<SocketAddrV4, SocketAddrV4>> = Arc::new(DashMap::new0());
+    #[cfg(not(target_os = "android"))]
+    let icmp_proxy_map: Arc<DashMap<(Ipv4Addr, u16, u16), Ipv4Addr>> = Arc::new(DashMap::new0());
     let tcp_listener = TcpListener::bind("0.0.0.0:0").await?;
     let udp_socket = UdpSocket::bind("0.0.0.0:0").await?;
     let tcp_proxy_port = tcp_listener.local_addr()?.port();
     let udp_proxy_port = udp_socket.local_addr()?.port();
     let tcp_proxy = TcpProxy::new(tcp_listener, tcp_proxy_map.clone());
     let udp_proxy = UdpProxy::new(udp_socket, udp_proxy_map.clone());
-    let addr = SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0);
-    let icmp_proxy = IcmpProxy::new(
-        addr,
-        icmp_proxy_map.clone(),
-        sender.clone(),
-        current_device.clone(),
-        client_cipher,
-    )?;
-    let icmp_socket = icmp_proxy.icmp_socket();
-    thread::spawn(move || {
-        icmp_proxy.start();
-    });
+    #[cfg(not(target_os = "android"))]
+    let icmp_socket = {
+        let addr = SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0);
+        let icmp_proxy = icmp_proxy::IcmpProxy::new(
+            addr,
+            icmp_proxy_map.clone(),
+            sender.clone(),
+            current_device.clone(),
+            client_cipher,
+        )?;
+        let icmp_socket = icmp_proxy.icmp_socket();
+        thread::spawn(move || {
+            icmp_proxy.start();
+        });
+        icmp_socket
+    };
 
     Ok((
         tcp_proxy,
@@ -77,7 +114,9 @@ pub async fn init_proxy(
             udp_proxy_port,
             tcp_proxy_map,
             udp_proxy_map,
+            #[cfg(not(target_os = "android"))]
             icmp_proxy_map,
+            #[cfg(not(target_os = "android"))]
             icmp_socket,
         },
     ))

vnt/src/ip_proxy/udp_proxy.rs

@@ -1,3 +1,4 @@
+use crate::ip_proxy::DashMapNew;
 use dashmap::DashMap;
 use std::io;
 use std::net::{SocketAddr, SocketAddrV4};
@@ -20,7 +21,8 @@ impl UdpProxy {
         let map = self.map;
         let udp_socket = self.udp_socket;
         let mut buf = [0u8; 65536];
-        let inner_map: Arc<DashMap<SocketAddrV4, Arc<UdpSocket>>> = Arc::new(DashMap::new());
+
+        let inner_map: Arc<DashMap<SocketAddrV4, Arc<UdpSocket>>> = Arc::new(DashMap::new0());
 
         loop {
             match udp_socket.recv_from(&mut buf).await {

vnt/src/lib.rs

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

vnt/src/nat/mod.rs

@@ -75,7 +75,7 @@ impl Into<NatType> for PunchNatType {
 }
 
 impl NatTest {
-    pub async fn new(
+    pub fn new(
         mut stun_server: Vec<String>,
         public_ip: Ipv4Addr,
         public_port: u16,
@@ -84,14 +84,14 @@ impl NatTest {
     ) -> NatTest {
         let server = stun_server[0].clone();
         stun_server.resize(3, server);
-        let nat_info = Self::re_test_(
-            &stun_server,
-            public_ip,
+        let nat_info = NatInfo::new(
+            vec![public_ip],
             public_port,
+            0,
             local_ipv4_addr,
             ipv6_addr,
-        )
-        .await;
+            NatType::Cone,
+        );
         let info = Arc::new(Mutex::new(nat_info));
         NatTest { stun_server, info }
     }

vnt/src/protocol/body.rs

@@ -1,6 +1,9 @@
 use std::{fmt, io};
 
-pub const ENCRYPTION_RESERVED: usize = 32;
+pub const ENCRYPTION_RESERVED: usize = 32 + 12;
+pub const AES_GCM_ENCRYPTION_RESERVED: usize = 32;
+pub const RSA_ENCRYPTION_RESERVED: usize = 32;
+
 /* aes_gcm加密数据体
   0                                            15                                              31
   0  1  2  3  4  5  6  7  8  9  0  1  2  3  4  5  6  7  8  9  0  1  2  3  4  5  6  7  8  9  0  1

vnt/src/protocol/mod.rs

@@ -111,7 +111,7 @@ impl<B: AsRef<[u8]>> NetPacket<B> {
                 "length overflow",
             ));
         }
-        //加密需要预留32字节
+        //加密需要预留ENCRYPTION_RESERVED字节
         let data_len = buffer.as_ref().len() - ENCRYPTION_RESERVED;
         Self::new0(data_len, buffer)
     }
@@ -187,7 +187,7 @@ impl<B: AsRef<[u8]>> NetPacket<B> {
 
 impl<B: AsRef<[u8]> + AsMut<[u8]>> NetPacket<B> {
     pub fn buffer_mut(&mut self) -> &mut [u8] {
-        self.buffer.as_mut()
+        &mut self.buffer.as_mut()[..self.data_len]
     }
     pub fn set_encrypt_flag(&mut self, is_encrypt: bool) {
         if is_encrypt {

vnt/src/tun_tap_device/linux.rs

@@ -169,7 +169,7 @@ pub fn delete_device(_device_type: DeviceType) {
             .output()
             .expect("sh exec error!");
         if !delete_tun.status.success() {
-            log::warn!("删除网卡失败:{:?}",delete_tun);
+            log::warn!("删除网卡失败:{:?}", delete_tun);
         }
     }
 }

vnt/src/tun_tap_device/linux_mac.rs

@@ -13,6 +13,7 @@ use tun::platform::macos::Device;
 use tun::platform::posix::{Reader, Writer};
 
 use packet::ethernet::packet::EthernetPacket;
+
 #[derive(Clone)]
 pub enum DeviceW {
     Tun(Writer),
@@ -63,10 +64,17 @@ impl DeviceWriter {
             #[cfg(any(target_os = "linux", target_os = "android"))]
             buf.put_u16(libc::ETH_P_IP as u16);
             buf.extend_from_slice(packet);
-            writer.write_all(&buf)
+            let len = writer.write(&buf)?;
+            if len != buf.len() {
+                log::error!("tun write error");
+            }
         } else {
-            writer.write_all(packet)
+            let len = writer.write(packet)?;
+            if len != packet.len() {
+                log::error!("tun write error");
+            }
         }
+        Ok(())
     }
     ///tun网卡写入ipv4数据
     pub fn write_ipv4_tun(&self, buf: &[u8]) -> io::Result<()> {