diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index f13db51..efaa04b 100644 --- a/.github/workflows/rust.yml +++ b/.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 <) -> Result, String> { pub fn to_ip(mask: &str) -> Result { if let Ok(m) = mask.parse::() { + 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)); diff --git a/common/src/identifier.rs b/common/src/identifier.rs index dfb89b4..94c3520 100644 --- a/common/src/identifier.rs +++ b/common/src/identifier.rs @@ -1,8 +1,7 @@ -use std::process::Command; - #[cfg(target_os = "windows")] pub fn get_unique_identifier() -> Option { 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 { #[cfg(target_os = "macos")] pub fn get_unique_identifier() -> Option { + 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 { #[cfg(target_os = "linux")] pub fn get_unique_identifier() -> Option { + use std::process::Command; let output = match Command::new("dmidecode") .arg("-s") .arg("system-uuid") diff --git a/vnt-cli/Cargo.toml b/vnt-cli/Cargo.toml index 4faa84f..ce73575 100644 --- a/vnt-cli/Cargo.toml +++ b/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"] diff --git a/vnt-cli/README.md b/vnt-cli/README.md index d455c54..6ac247d 100644 --- a/vnt-cli/README.md +++ b/vnt-cli/README.md @@ -43,7 +43,7 @@ ### -u `` -设置虚拟网卡的mtu值，大多数情况下使用默认值效率会更高，也可根据实际情况微调这个值，不加密默认为1430，加密默认为1410 +设置虚拟网卡的mtu值，大多数情况下使用默认值效率会更高，也可根据实际情况微调这个值，不加密默认为1450，加密默认为1410 ### --tcp 和服务端使用tcp通信。有些网络提供商对UDP限制比较大，这个时候可以选择使用TCP模式，提高稳定性。一般来说udp延迟和消耗更低 @@ -51,17 +51,17 @@ 指定虚拟ip,指定的ip不能和其他设备重复,必须有效并且在服务端所属网段下,默认情况由服务端分配 ### --par `` 任务并行度(必须为正整数),默认值为1,该值表示处理网卡读写的任务数,组网设备数较多、处理延迟较大时可适当调大此值 -### --thread `` -线程数(必须为正整数),默认为核心数乘2,该值表示处理网络读写、ip代理、打洞等用到的线程数,组网设备数较多、处理延迟较大时可适当调大此值 ### --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 开启数据指纹校验，可增加安全性，如果服务端开启指纹校验，则客户端也必须开启，开启会损耗一部分性能 diff --git a/vnt-cli/src/main.rs b/vnt-cli/src/main.rs index 8bf2001..26d12b6 100644 --- a/vnt-cli/src/main.rs +++ b/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 = 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::("par").unwrap().unwrap_or(1); if parallel == 0 { - println!("--par invalid"); + println!("'--par {}' invalid", parallel); return; } - let thread_num = matches - .opt_get::("thread") - .unwrap() - .unwrap_or(std::thread::available_parallelism().unwrap().get() * 2); + let cipher_model = matches .opt_get::("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::("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 使用该密码生成的密钥对客户端数据进行加密,并且服务端无法解密,使用相同密码的客户端才能通信"); println!(" -W 加密当前客户端和服务端通信的数据,请留意服务端指纹是否正确"); println!(" -m 模拟组播,默认情况下组播数据会被当作广播发送,开启后会模拟真实组播的数据发送"); - println!(" -u 自定义mtu(不加密默认为1430，加密默认为1410)"); + println!(" -u 自定义mtu(不加密默认为1450，加密默认为1410)"); println!(" --tcp 和服务端使用tcp通信,默认使用udp,遇到udp qos时可指定使用tcp"); println!(" --ip 指定虚拟ip,指定的ip不能和其他设备重复,必须有效并且在服务端所属网段下,默认情况由服务端分配"); println!(" --relay 仅使用服务器转发,不使用p2p,默认情况允许使用p2p"); println!(" --par 任务并行度(必须为正整数),默认值为1"); - println!(" --thread 线程数(必须为正整数),默认为核心数乘2"); - println!(" --model 加密模式(默认aes_gcm)，可选值aes_gcm/aes_cbc/aes_ecb，通常性能aes_ecb>aes_cbc>aes_gcm,安全性则相反"); + println!(" --model 加密模式(默认aes_gcm)，可选值aes_gcm/aes_cbc/aes_ecb，一般来说性能:aes_ecb>aes_cbc>aes_gcm"); println!(" --finger 增加数据指纹校验，可增加安全性，如果服务端开启指纹校验，则客户端也必须开启"); println!(" --punch 取值ipv4/ipv6，ipv4表示仅使用ipv4打洞"); diff --git a/vnt-jni/Cargo.toml b/vnt-jni/Cargo.toml index bb8bc60..7b4984d 100644 --- a/vnt-jni/Cargo.toml +++ b/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 } diff --git a/vnt-jni/src/vnt_util.rs b/vnt-jni/src/vnt_util.rs index 97574bf..59f6c1a 100644 --- a/vnt-jni/src/vnt_util.rs +++ b/vnt-jni/src/vnt_util.rs @@ -66,6 +66,36 @@ fn new_sync(env: &mut JNIEnv, config: JObject) -> Result { 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 { server_address, server_address_str, stun_server, - vec![], - vec![], + in_ips, + out_ips, password, false, None, diff --git a/vnt/Cargo.toml b/vnt/Cargo.toml index 8219e68..a17fa6d 100644 --- a/vnt/Cargo.toml +++ b/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"] diff --git a/vnt/src/channel/channel.rs b/vnt/src/channel/channel.rs index df821c6..970f6f7 100644 --- a/vnt/src/channel/channel.rs +++ b/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, - pub(crate) main_channel_ipv6: Option>, + pub(crate) main_channel: Arc, + pub(crate) main_channel_ipv6: Option>, //在udp的基础上，可以选择使用tcp和服务端通信 pub(crate) main_tcp_channel: Option>>, - pub(crate) route_table: DashMap>, + pub(crate) route_table: Atomic>>, pub(crate) route_table_time: DashMap<(RouteKey, Ipv4Addr), Instant>, pub(crate) status_receiver: Receiver, pub(crate) status_sender: Sender, - pub(crate) udp_map: DashMap>, + pub(crate) udp_map: Atomic>>, pub(crate) channel_num: usize, current_device: Arc>, } @@ -42,8 +48,8 @@ pub struct Context { impl Context { pub fn new( - main_channel: Arc, - main_channel_ipv6: Option>, + main_channel: Arc, + main_channel_ipv6: Option>, main_tcp_channel: Option>>, current_device: Arc>, _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 { + fn insert_udp(&self, id: usize, udp: Arc) { + 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>) { + 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 { 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 { - 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 { + + pub fn send_main(&self, buf: &[u8], addr: SocketAddr) -> io::Result { 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 { - 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 { - 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 { + 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 { + 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 { - 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 { - 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 { - 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> { + 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> { - 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 { - 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 { - 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)> { - 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, usize, usize, RouteKey)>>, + Vec, usize, usize, RouteKey)>>, ); -struct BufReceiverGroup(Vec, usize, usize, RouteKey)>>); +struct BufReceiverGroup(Vec, 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>, usize, usize, RouteKey)>(10); + std::sync::mpsc::sync_channel::<(Block<'static, Vec>, 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>, current_device: Arc>, - 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, + handler: ChannelDataHandler, + buf_sender: Option, + 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(); } diff --git a/vnt/src/channel/mod.rs b/vnt/src/channel/mod.rs index ced60ef..c985708 100644 --- a/vnt/src/channel/mod.rs +++ b/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 + } } diff --git a/vnt/src/channel/punch.rs b/vnt/src/channel/punch.rs index 42e78af..0b520a8 100644 --- a/vnt/src/channel/punch.rs +++ b/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; } } diff --git a/vnt/src/cipher/aes_ecb.rs b/vnt/src/cipher/aes_ecb.rs index cccbd13..14c83db 100644 --- a/vnt/src/cipher/aes_ecb.rs +++ b/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; @@ -12,11 +10,11 @@ type Aes256EcbDec = ecb::Decryptor; #[derive(Clone)] pub struct AesEcbCipher { - pub(crate) cipher: AesEcbEnum, + key: AesEcbEnum, pub(crate) finger: Option, } -#[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) -> Self { Self { - cipher: AesEcbEnum::AES128ECB(key), + key: AesEcbEnum::AES128ECB(key), finger, } } pub fn new_256(key: [u8; 32], finger: Option) -> 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::(secret_body.en_body_mut()), - AesEcbEnum::AES256ECB(key) => Aes256EcbDec::new(&(*key).into()) - .decrypt_padded_mut::(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::(net_packet.payload(), &mut out), + AesEcbEnum::AES256ECB(key) => Aes256EcbDec::new(&key.into()) + .decrypt_padded_b2b_mut::(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, ) -> 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::(net_packet.payload_mut(), p_len), - AesEcbEnum::AES256ECB(key) => Aes256EcbEnc::new(&(*key).into()) - .encrypt_padded_mut::(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::(net_packet.buffer(), &mut out), + AesEcbEnum::AES256ECB(key) => Aes256EcbEnc::new(&key.into()) + .encrypt_padded_b2b_mut::(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(); +} diff --git a/vnt/src/cipher/aes_gcm_cipher.rs b/vnt/src/cipher/aes_gcm_cipher.rs index a2a731c..9920c10 100644 --- a/vnt/src/cipher/aes_gcm_cipher.rs +++ b/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, ) -> 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 = 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()); diff --git a/vnt/src/cipher/cipher.rs b/vnt/src/cipher/cipher.rs index 4715b78..5dd4840 100644 --- a/vnt/src/cipher/cipher.rs +++ b/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}; diff --git a/vnt/src/cipher/mod.rs b/vnt/src/cipher/mod.rs index ad042f3..1f4e66d 100644 --- a/vnt/src/cipher/mod.rs +++ b/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; diff --git a/vnt/src/cipher/openssl_aes_ecb.rs b/vnt/src/cipher/openssl_aes_ecb.rs new file mode 100644 index 0000000..4f8fad6 --- /dev/null +++ b/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, + pub(crate) en_ctx: *mut EVP_CIPHER_CTX, + pub(crate) de_ctx: *mut EVP_CIPHER_CTX, + pub(crate) finger: Option, +} + +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) -> 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) -> 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 + AsMut<[u8]>>( + &self, + net_packet: &mut NetPacket, + ) -> 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 + AsMut<[u8]>>( + &self, + net_packet: &mut NetPacket, + ) -> 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(); +} diff --git a/vnt/src/cipher/ring_aes_gcm_cipher.rs b/vnt/src/cipher/ring_aes_gcm_cipher.rs index 8685baf..bc65a18 100644 --- a/vnt/src/cipher/ring_aes_gcm_cipher.rs +++ b/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()); diff --git a/vnt/src/cipher/rsa_cipher.rs b/vnt/src/cipher/rsa_cipher.rs index d7dcfd9..3b52c05 100644 --- a/vnt/src/cipher/rsa_cipher.rs +++ b/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, ) -> io::Result>> { - 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()); diff --git a/vnt/src/core/mod.rs b/vnt/src/core/mod.rs index 5755dbe..628ac3c 100644 --- a/vnt/src/core/mod.rs +++ b/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 { - 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 { + //将读的超时时间清空 + 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)>> = Arc::new(Mutex::new((response.epoch, response.device_info_list))); - let peer_nat_info_map: Arc> = Arc::new(DashMap::new()); + let peer_nat_info_map: Arc> = 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) { diff --git a/vnt/src/handle/handshake_handler.rs b/vnt/src/handle/handshake_handler.rs index 15585ce..5c8100e 100644 --- a/vnt/src/handle/handshake_handler.rs +++ b/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(()) } diff --git a/vnt/src/handle/heartbeat_handler.rs b/vnt/src/handle/heartbeat_handler.rs index 69c93ee..60d464c 100644 --- a/vnt/src/handle/heartbeat_handler.rs +++ b/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); } diff --git a/vnt/src/handle/punch_handler.rs b/vnt/src/handle/punch_handler.rs index 8483cf1..d7aaa0a 100644 --- a/vnt/src/handle/punch_handler.rs +++ b/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(()) diff --git a/vnt/src/handle/recv_handler.rs b/vnt/src/handle/recv_handler.rs index 121f6d7..4423df5 100644 --- a/vnt/src/handle/recv_handler.rs +++ b/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 => { //地址用尽 diff --git a/vnt/src/handle/registration_handler.rs b/vnt/src/handle/registration_handler.rs index 2596ef2..a89a13b 100644 --- a/vnt/src/handle/registration_handler.rs +++ b/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(()) } } diff --git a/vnt/src/handle/tun_tap/channel_group.rs b/vnt/src/handle/tun_tap/channel_group.rs index dfa37c3..2974c24 100644 --- a/vnt/src/handle/tun_tap/channel_group.rs +++ b/vnt/src/handle/tun_tap/channel_group.rs @@ -3,16 +3,16 @@ use byte_pool::Block; #[derive(Clone)] pub struct BufSenderGroup( usize, - Vec, usize, usize)>>, + Vec, usize, usize)>>, ); -pub struct BufReceiverGroup(pub Vec, usize, usize)>>); +pub struct BufReceiverGroup(pub Vec, usize, usize)>>); impl BufSenderGroup { - pub 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); } diff --git a/vnt/src/handle/tun_tap/mod.rs b/vnt/src/handle/tun_tap/mod.rs index 9d93615..385177d 100644 --- a/vnt/src/handle/tun_tap/mod.rs +++ b/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>>, 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, ¤t_device, - ) - .await?; + )?; } _ => {} } @@ -176,8 +167,7 @@ pub async fn base_handle( sender, &mut net_packet, ¤t_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(()); } diff --git a/vnt/src/handle/tun_tap/tap_handler.rs b/vnt/src/handle/tun_tap/tap_handler.rs index 60de044..fd50d67 100644 --- a/vnt/src/handle/tun_tap/tap_handler.rs +++ b/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, current_device: Arc>, ip_route: Option, @@ -160,21 +150,19 @@ async fn start_simple( len, &igmp_server, ¤t_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, @@ -261,8 +249,7 @@ async fn handle( proxy_map, client_cipher, server_cipher, - ) - .await; + ); } _ => { // log::warn!("不支持的二层协议：{:?}",p) diff --git a/vnt/src/handle/tun_tap/tun_handler.rs b/vnt/src/handle/tun_tap/tun_handler.rs index 4c56939..177fe80 100644 --- a/vnt/src/handle/tun_tap/tun_handler.rs +++ b/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, @@ -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) diff --git a/vnt/src/igmp_server/mod.rs b/vnt/src/igmp_server/mod.rs index bb49287..16ec482 100644 --- a/vnt/src/igmp_server/mod.rs +++ b/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>>> = Arc::new(DashMap::new()); + let multicast: Arc>>> = Arc::new(DashMap::new0()); std::thread::spawn(move || { //预留以太网帧头和ip头 let mut buf = [0; 14 + 24 + 12]; diff --git a/vnt/src/ip_proxy/icmp_proxy.rs b/vnt/src/ip_proxy/icmp_proxy.rs index 62feb94..ce03bfb 100644 --- a/vnt/src/ip_proxy/icmp_proxy.rs +++ b/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, ); diff --git a/vnt/src/ip_proxy/mod.rs b/vnt/src/ip_proxy/mod.rs index 1d1a803..90027c0 100644 --- a/vnt/src/ip_proxy/mod.rs +++ b/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 { + 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>, pub(crate) udp_proxy_map: Arc>, // icmp用Identifier来区分，没有Identifier的一律不转发 + #[cfg(not(target_os = "android"))] pub(crate) icmp_proxy_map: Arc>, + #[cfg(not(target_os = "android"))] icmp_socket: Arc, } impl IpProxyMap { + #[cfg(not(target_os = "android"))] pub fn send_icmp(&self, buf: &[u8], dest: &Ipv4Addr) -> io::Result { 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>, - 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, + >, + #[cfg(not(target_os = "android"))] client_cipher: crate::cipher::Cipher, ) -> io::Result<(TcpProxy, UdpProxy, IpProxyMap)> { - let tcp_proxy_map: Arc> = Arc::new(DashMap::new()); - let udp_proxy_map: Arc> = Arc::new(DashMap::new()); - let icmp_proxy_map: Arc> = Arc::new(DashMap::new()); + let tcp_proxy_map: Arc> = Arc::new(DashMap::new0()); + let udp_proxy_map: Arc> = Arc::new(DashMap::new0()); + #[cfg(not(target_os = "android"))] + let icmp_proxy_map: Arc> = 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, }, )) diff --git a/vnt/src/ip_proxy/udp_proxy.rs b/vnt/src/ip_proxy/udp_proxy.rs index a9930b3..5e04c9b 100644 --- a/vnt/src/ip_proxy/udp_proxy.rs +++ b/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>> = Arc::new(DashMap::new()); + + let inner_map: Arc>> = Arc::new(DashMap::new0()); loop { match udp_socket.recv_from(&mut buf).await { diff --git a/vnt/src/lib.rs b/vnt/src/lib.rs index e234a8e..aadf8c1 100644 --- a/vnt/src/lib.rs +++ b/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 = std::result::Result; pub mod channel; diff --git a/vnt/src/nat/mod.rs b/vnt/src/nat/mod.rs index a695b1f..1b1b90e 100644 --- a/vnt/src/nat/mod.rs +++ b/vnt/src/nat/mod.rs @@ -75,7 +75,7 @@ impl Into for PunchNatType { } impl NatTest { - pub async fn new( + pub fn new( mut stun_server: Vec, 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 } } diff --git a/vnt/src/protocol/body.rs b/vnt/src/protocol/body.rs index eb78e48..d35cea4 100644 --- a/vnt/src/protocol/body.rs +++ b/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 diff --git a/vnt/src/protocol/mod.rs b/vnt/src/protocol/mod.rs index 2d0f9d2..e1db894 100644 --- a/vnt/src/protocol/mod.rs +++ b/vnt/src/protocol/mod.rs @@ -111,7 +111,7 @@ impl> NetPacket { "length overflow", )); } - //加密需要预留32字节 + //加密需要预留ENCRYPTION_RESERVED字节 let data_len = buffer.as_ref().len() - ENCRYPTION_RESERVED; Self::new0(data_len, buffer) } @@ -187,7 +187,7 @@ impl> NetPacket { impl + AsMut<[u8]>> NetPacket { 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 { diff --git a/vnt/src/tun_tap_device/linux.rs b/vnt/src/tun_tap_device/linux.rs index f4e169c..0d4e951 100644 --- a/vnt/src/tun_tap_device/linux.rs +++ b/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); } } } diff --git a/vnt/src/tun_tap_device/linux_mac.rs b/vnt/src/tun_tap_device/linux_mac.rs index 718feff..876ee02 100644 --- a/vnt/src/tun_tap_device/linux_mac.rs +++ b/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<()> {