From c3368481ad06de709ad8a3860e572441df9bbb76 Mon Sep 17 00:00:00 2001 From: lubeilin <1791778603@qq.com> Date: Mon, 28 Aug 2023 20:34:23 +0800 Subject: [PATCH] =?UTF-8?q?1.=E4=BF=AE=E6=94=B9=E9=BB=98=E8=AE=A4=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E6=95=B0=202.=E4=BF=AE=E6=94=B9=E6=8C=87=E7=BA=B9?= =?UTF-8?q?=E7=94=9F=E6=88=90=E6=96=B9=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- vnt-cli/src/main.rs | 8 +- vnt/src/channel/channel.rs | 137 ++++++++++++------ vnt/src/cipher/aes_c.rs | 6 +- vnt/src/cipher/aes_gcm_cipher.rs | 7 +- vnt/src/cipher/finger.rs | 29 ++-- vnt/src/cipher/mod.rs | 2 +- ...{ring_cipher.rs => ring_aes_gcm_cipher.rs} | 7 +- vnt/src/ip_proxy/tcp_proxy.rs | 2 - vnt/src/protocol/body.rs | 5 + 9 files changed, 128 insertions(+), 75 deletions(-) rename vnt/src/cipher/{ring_cipher.rs => ring_aes_gcm_cipher.rs} (96%) diff --git a/vnt-cli/src/main.rs b/vnt-cli/src/main.rs index 4cd09e6..47177d5 100644 --- a/vnt-cli/src/main.rs +++ b/vnt-cli/src/main.rs @@ -195,7 +195,7 @@ fn main() { } let tcp_channel = matches.opt_present("tcp"); let relay = matches.opt_present("relay"); - let parallel = matches.opt_get::("par").unwrap().unwrap_or(2); + let parallel = matches.opt_get::("par").unwrap().unwrap_or(1); if parallel == 0 { println!("--par invalid"); return; @@ -213,11 +213,11 @@ fn main() { out_ip, password, simulate_multicast, mtu, tcp_channel, virtual_ip, relay, server_encrypt, parallel); let runtime = tokio::runtime::Builder::new_multi_thread().enable_all().worker_threads(thread_num).build().unwrap(); - runtime.block_on(main0(config,!unused_cmd)); + runtime.block_on(main0(config, !unused_cmd)); std::process::exit(0); } -async fn main0(config: Config,show_cmd:bool) { +async fn main0(config: Config, show_cmd: bool) { let server_encrypt = config.server_encrypt; let mut vnt_util = VntUtil::new(config).await.unwrap(); let mut conn_count = 0; @@ -452,7 +452,7 @@ fn print_usage(program: &str, _opts: Options) { println!(" --tcp 和服务端使用tcp通信,默认使用udp,遇到udp qos时可指定使用tcp"); println!(" --ip 指定虚拟ip,指定的ip不能和其他设备重复,必须有效并且在服务端所属网段下,默认情况由服务端分配"); println!(" --relay 仅使用服务器转发,不使用p2p,默认情况允许使用p2p"); - println!(" --par 任务并行度(必须为正整数),默认值为2"); + println!(" --par 任务并行度(必须为正整数),默认值为1"); println!(" --thread 线程数(必须为正整数),默认为核心数乘2"); println!(); println!(" --list {}", yellow("后台运行时,查看其他设备列表".to_string())); diff --git a/vnt/src/channel/channel.rs b/vnt/src/channel/channel.rs index 5ea7a05..1eed55d 100644 --- a/vnt/src/channel/channel.rs +++ b/vnt/src/channel/channel.rs @@ -444,22 +444,28 @@ impl Channel { relay: bool, parallel: usize, ) { - let (buf_sender, buf_receiver) = buf_channel_group(parallel); - for mut buf_receiver in buf_receiver.0 { - let context = self.context.clone(); - let handler = self.handler.clone(); - tokio::spawn(async move { - while let Some((mut buf, start, end, route_key)) = buf_receiver.recv().await { - handler.handle(&mut buf, start, end, route_key, &context).await; - } - }); - } + let 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, buf_receiver) = buf_channel_group(parallel); + for mut buf_receiver in buf_receiver.0 { + let context = context.clone(); + let handler = handler.clone(); + tokio::spawn(async move { + while let Some((mut buf, start, end, route_key)) = buf_receiver.recv().await { + handler.handle(&mut buf, start, end, route_key, &context).await; + } + }); + } + Some(buf_sender) + } else { + None + }; if let Some((tcp_stream, receiver)) = tcp { - tokio::spawn(Self::start_tcp(worker.worker("main_channel_tcp"), tcp_stream, receiver, context.inner.current_device.clone(), buf_sender.clone(), head_reserve)); + tokio::spawn(Self::start_tcp(worker.worker("main_channel_tcp"), tcp_stream, receiver, context.inner.current_device.clone(), buf_sender.clone().unwrap(), head_reserve)); } - tokio::spawn(Self::start_(worker.worker("main_channel_1"), context.clone(), main_channel.clone(), buf_sender.clone(), head_reserve, true)); + tokio::spawn(Self::start_(worker.worker("main_channel_1"), context.clone(), main_channel.clone(), handler.clone(), buf_sender.clone(), head_reserve, true)); if relay { worker.stop_wait().await; return; @@ -489,7 +495,7 @@ impl Channel { Ok(udp) => { let udp = Arc::new(udp); let context = context.clone(); - tokio::spawn(Self::start_(worker.worker("symmetric_channel"),context, udp,buf_sender.clone(), head_reserve, false)); + tokio::spawn(Self::start_(worker.worker("symmetric_channel"),context, udp,handler.clone(),buf_sender.clone(), head_reserve, false)); } Err(e) => { log::error!("{}",e); @@ -513,7 +519,8 @@ impl Channel { } async fn start_(mut worker: VntWorker, context: Context, udp: Arc, - mut buf_sender: BufSenderGroup, + handler: ChannelDataHandler, + buf_sender: Option, head_reserve: usize, is_core: bool) { let mut status_receiver = context.inner.status_receiver.clone(); @@ -526,44 +533,88 @@ impl Channel { #[cfg(any(unix))] let id = 1 + udp.as_raw_fd() as usize; context.inner.udp_map.insert(id, udp.clone()); - loop { - let mut buf = POOL.alloc(4096); - tokio::select! { - 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{ - log::error!("udp buf_sender发送数据失败"); - break; + match buf_sender { + None => { + let mut buf = [0; 4096]; + loop { + tokio::select! { + rs=udp.recv_from(&mut buf[head_reserve..])=>{ + match rs { + Ok((len, addr)) => { + handler.handle(&mut buf, head_reserve, head_reserve + len, RouteKey::new(id, addr), &context).await; + } + Err(e) => { + log::error!("{:?}",e) + } } } - Err(e) => { - log::error!("{:?}",e) + changed=status_receiver.changed()=>{ + match changed { + Ok(_) => { + match *status_receiver.borrow() { + Status::Cone => { + if !is_core{ + break; + } + } + Status::Close=>{ + break; + } + Status::Symmetric => {} + } + } + Err(_) => { + break; + } + } + } + _=worker.stop_wait()=>{ + break; } } } - changed=status_receiver.changed()=>{ - match changed { - Ok(_) => { - match *status_receiver.borrow() { - Status::Cone => { - if !is_core{ - break; - } + } + Some(mut buf_sender) => { + loop { + let mut buf = POOL.alloc(4096); + tokio::select! { + 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{ + log::error!("udp buf_sender发送数据失败"); + break; } - Status::Close=>{ - break; - } - Status::Symmetric => {} + } + Err(e) => { + log::error!("{:?}",e) } } - Err(_) => { - break; - } } - } - _=worker.stop_wait()=>{ - break; + changed=status_receiver.changed()=>{ + match changed { + Ok(_) => { + match *status_receiver.borrow() { + Status::Cone => { + if !is_core{ + break; + } + } + Status::Close=>{ + break; + } + Status::Symmetric => {} + } + } + Err(_) => { + break; + } + } + } + _=worker.stop_wait()=>{ + break; + } + } } } } diff --git a/vnt/src/cipher/aes_c.rs b/vnt/src/cipher/aes_c.rs index 87262d4..d317c21 100644 --- a/vnt/src/cipher/aes_c.rs +++ b/vnt/src/cipher/aes_c.rs @@ -3,7 +3,7 @@ use crate::cipher::Finger; use crate::protocol::NetPacket; use sha2::Digest; #[cfg(feature = "ring-cipher")] -use crate::cipher::ring_cipher::AesGcmCipher; +use crate::cipher::ring_aes_gcm_cipher::AesGcmCipher; #[cfg(not(feature = "ring-cipher"))] use crate::cipher::aes_gcm_cipher::AesGcmCipher; @@ -15,7 +15,7 @@ pub enum Cipher { impl Cipher { pub fn new_password(password: Option, token: String) -> Self { - let finger = Finger::new(token); + let finger = Finger::new(&token); if let Some(password) = password { let mut hasher = sha2::Sha256::new(); hasher.update(password.as_bytes()); @@ -32,7 +32,7 @@ impl Cipher { } } pub fn new_key(key: [u8; 32], token: String) -> io::Result { - let finger = Finger::new(token); + let finger = Finger::new(&token); match key.len() { 16 => { let aes = AesGcmCipher::new_128(key[..16].try_into().unwrap(), finger); diff --git a/vnt/src/cipher/aes_gcm_cipher.rs b/vnt/src/cipher/aes_gcm_cipher.rs index 4045e6c..661de1d 100644 --- a/vnt/src/cipher/aes_gcm_cipher.rs +++ b/vnt/src/cipher/aes_gcm_cipher.rs @@ -58,10 +58,7 @@ impl AesGcmCipher { let mut secret_body = SecretBody::new(net_packet.payload_mut())?; let tag = secret_body.tag(); - if tag.len() != 16 { - return Err(io::Error::new(io::ErrorKind::Other, "tag err")); - } - let finger = self.finger.calculate_finger(&nonce_raw, &secret_body); + let finger = self.finger.calculate_finger(&nonce_raw, secret_body.en_body()); if &finger != secret_body.finger() { return Err(io::Error::new(io::ErrorKind::Other, "finger err")); } @@ -102,7 +99,7 @@ impl AesGcmCipher { return match rs { Ok(tag) => { secret_body.set_tag(tag.as_slice())?; - let finger = self.finger.calculate_finger(&nonce_raw, &secret_body); + let finger = self.finger.calculate_finger(&nonce_raw, secret_body.en_body()); secret_body.set_finger(&finger)?; net_packet.set_encrypt_flag(true); Ok(()) diff --git a/vnt/src/cipher/finger.rs b/vnt/src/cipher/finger.rs index 612be75..431c996 100644 --- a/vnt/src/cipher/finger.rs +++ b/vnt/src/cipher/finger.rs @@ -2,23 +2,27 @@ use std::io; use sha2::Digest; -use crate::protocol::{body::ENCRYPTION_RESERVED, body::SecretBody, NetPacket}; +use crate::protocol::{body::ENCRYPTION_RESERVED, NetPacket}; #[derive(Clone)] pub struct Finger { - token: String, + hash: [u8; 32], } impl Finger { - pub fn new(token: String) -> Self { - Finger { token } + pub fn new(str: &str) -> Self { + let mut hasher = sha2::Sha256::new(); + hasher.update(str.as_bytes()); + let hash: [u8; 32] = hasher.finalize().into(); + Finger { hash } } pub fn check_finger>(&self, net_packet: &NetPacket) -> io::Result<()> { if !net_packet.is_encrypt() { //未加密的数据直接丢弃 return Err(io::Error::new(io::ErrorKind::Other, "not encrypt")); } - if net_packet.payload().len() < ENCRYPTION_RESERVED { + let payload_len = net_packet.payload().len(); + if payload_len < ENCRYPTION_RESERVED { log::error!("数据异常,长度小于{}",ENCRYPTION_RESERVED); return Err(io::Error::new(io::ErrorKind::Other, "data err")); } @@ -27,19 +31,20 @@ impl Finger { nonce_raw[4..8].copy_from_slice(&net_packet.destination().octets()); nonce_raw[8] = net_packet.protocol().into(); nonce_raw[9] = net_packet.transport_protocol(); - let secret_body = SecretBody::new(net_packet.payload())?; - let finger = self.calculate_finger(&nonce_raw, &secret_body); - if &finger != secret_body.finger() { + nonce_raw[10] = net_packet.is_gateway() as u8; + nonce_raw[11] = net_packet.source_ttl(); + let payload = net_packet.payload(); + let finger = self.calculate_finger(&nonce_raw, &payload[..payload_len - 12]); + if &finger[..] != &payload[payload_len - 12..] { return Err(io::Error::new(io::ErrorKind::Other, "finger err")); } Ok(()) } - pub fn calculate_finger>(&self, nonce_raw: &[u8; 12], secret_body: &SecretBody) -> [u8; 12] { + pub fn calculate_finger(&self, nonce_raw: &[u8; 12], secret_body: &[u8]) -> [u8; 12] { let mut hasher = sha2::Sha256::new(); - hasher.update(secret_body.body()); hasher.update(nonce_raw); - hasher.update(secret_body.tag()); - hasher.update(&self.token); + hasher.update(secret_body); + hasher.update(&self.hash); let key: [u8; 32] = hasher.finalize().into(); return key[20..].try_into().unwrap(); } diff --git a/vnt/src/cipher/mod.rs b/vnt/src/cipher/mod.rs index e8a2b3a..d3881aa 100644 --- a/vnt/src/cipher/mod.rs +++ b/vnt/src/cipher/mod.rs @@ -1,5 +1,5 @@ #[cfg(feature = "ring-cipher")] -mod ring_cipher; +mod ring_aes_gcm_cipher; #[cfg(not(feature = "ring-cipher"))] mod aes_gcm_cipher; mod rsa_cipher; diff --git a/vnt/src/cipher/ring_cipher.rs b/vnt/src/cipher/ring_aes_gcm_cipher.rs similarity index 96% rename from vnt/src/cipher/ring_cipher.rs rename to vnt/src/cipher/ring_aes_gcm_cipher.rs index c0404c4..9a00fb1 100644 --- a/vnt/src/cipher/ring_cipher.rs +++ b/vnt/src/cipher/ring_aes_gcm_cipher.rs @@ -67,10 +67,7 @@ impl AesGcmCipher { let nonce = aead::Nonce::assume_unique_for_key(nonce_raw); let mut secret_body = SecretBody::new(net_packet.payload_mut())?; let tag = secret_body.tag(); - if tag.len() != 16 { - return Err(io::Error::new(io::ErrorKind::Other, "tag err")); - } - let finger = self.finger.calculate_finger(&nonce_raw, &secret_body); + let finger = self.finger.calculate_finger(&nonce_raw, secret_body.en_body()); if &finger != secret_body.finger() { return Err(io::Error::new(io::ErrorKind::Other, "finger err")); } @@ -122,7 +119,7 @@ impl AesGcmCipher { return Err(io::Error::new(io::ErrorKind::Other, format!("加密tag长度错误:{}", tag.len()))); } secret_body.set_tag(tag)?; - let finger = self.finger.calculate_finger(&nonce_raw, &secret_body); + let finger = self.finger.calculate_finger(&nonce_raw, secret_body.en_body()); secret_body.set_finger(&finger)?; net_packet.set_encrypt_flag(true); Ok(()) diff --git a/vnt/src/ip_proxy/tcp_proxy.rs b/vnt/src/ip_proxy/tcp_proxy.rs index 42df18d..1bce598 100644 --- a/vnt/src/ip_proxy/tcp_proxy.rs +++ b/vnt/src/ip_proxy/tcp_proxy.rs @@ -35,12 +35,10 @@ impl TcpProxy { continue; } }; - let tcp_proxy_map = tcp_proxy_map.clone(); tokio::spawn(async move { if let Err(e) = proxy(tcp_stream, peer_tcp_stream).await { log::warn!("{}->{},{}",sender_addr,dest_addr,e); } - tcp_proxy_map.remove(&sender_addr); }); } } diff --git a/vnt/src/protocol/body.rs b/vnt/src/protocol/body.rs index 215bdb7..ad9a54c 100644 --- a/vnt/src/protocol/body.rs +++ b/vnt/src/protocol/body.rs @@ -54,6 +54,11 @@ impl> SecretBody { let end = self.buffer.as_ref().len() - 12; &self.buffer.as_ref()[end - 16..end] } + /// 数据部分+tag部分 + pub fn en_body(&self) -> &[u8] { + let end = self.buffer.as_ref().len() - 12; + &self.buffer.as_ref()[..end] + } pub fn finger(&self) -> &[u8] { let end = self.buffer.as_ref().len(); &self.buffer.as_ref()[end - 12..end]