diff --git a/common/src/cli.rs b/common/src/cli.rs index e66bab8..23f533b 100644 --- a/common/src/cli.rs +++ b/common/src/cli.rs @@ -75,6 +75,7 @@ pub fn parse_args_config() -> anyhow::Result, bool)> opts.optopt("f", "", "配置文件", ""); opts.optopt("", "compressor", "压缩算法", ""); opts.optflag("", "disable-stats", "关闭流量统计"); + opts.optflag("", "allow-wg", "允许接入WireGuard"); //"后台运行时,查看其他设备列表" opts.optflag("", "add", "后台运行时,添加地址"); opts.optflag("", "list", "后台运行时,查看其他设备列表"); @@ -281,6 +282,7 @@ pub fn parse_args_config() -> anyhow::Result, bool)> let port_mapping_list = matches.opt_strs("mapping"); let vnt_mapping_list = matches.opt_strs("vnt-mapping"); let disable_stats = matches.opt_present("disable-stats"); + let allow_wire_guard = matches.opt_present("allow-wg"); let compressor = if let Some(compressor) = matches.opt_str("compressor").as_ref() { Compressor::from_str(compressor) .map_err(|e| anyhow!("{}", e)) @@ -321,6 +323,7 @@ pub fn parse_args_config() -> anyhow::Result, bool)> port_mapping_list, compressor, !disable_stats, + allow_wire_guard, ) { Ok(config) => config, Err(e) => { @@ -435,6 +438,7 @@ fn print_usage(program: &str, _opts: Options) { ) ); println!(" --disable-stats 关闭流量统计"); + println!(" --allow-wg 允许接入WireGuard客户端"); println!(); #[cfg(feature = "command")] { diff --git a/common/src/command/entity.rs b/common/src/command/entity.rs index b42f054..9770897 100644 --- a/common/src/command/entity.rs +++ b/common/src/command/entity.rs @@ -45,6 +45,7 @@ pub struct DeviceItem { pub client_secret_hash: Vec, pub current_client_secret: bool, pub current_client_secret_hash: Vec, + pub wire_guard: bool, } #[derive(Serialize, Deserialize, Debug, Default)] diff --git a/common/src/command/mod.rs b/common/src/command/mod.rs index b408b88..2ffa8bc 100644 --- a/common/src/command/mod.rs +++ b/common/src/command/mod.rs @@ -219,6 +219,7 @@ pub fn command_list(vnt: &Vnt) -> Vec { client_secret_hash: peer.client_secret_hash, current_client_secret, current_client_secret_hash: client_encrypt_hash.to_vec(), + wire_guard: peer.wireguard, }; list.push(item); } diff --git a/common/src/config/file_config.rs b/common/src/config/file_config.rs index 2c9b24e..c5b2b1b 100644 --- a/common/src/config/file_config.rs +++ b/common/src/config/file_config.rs @@ -46,6 +46,8 @@ pub struct FileConfig { pub compressor: Option, pub vnt_mapping: Vec, pub disable_stats: bool, + // 允许传递wg流量 + pub allow_wire_guard: bool, } impl Default for FileConfig { @@ -90,6 +92,7 @@ impl Default for FileConfig { compressor: None, vnt_mapping: vec![], disable_stats: false, + allow_wire_guard: false, } } } @@ -177,6 +180,7 @@ pub fn read_config(file_path: &str) -> anyhow::Result<(Config, Vec, bool file_conf.mapping, compressor, !file_conf.disable_stats, + file_conf.allow_wire_guard, )?; Ok((config, file_conf.vnt_mapping, file_conf.cmd)) diff --git a/common/src/console_out/mod.rs b/common/src/console_out/mod.rs index 2efde46..3a3bc7c 100644 --- a/common/src/console_out/mod.rs +++ b/common/src/console_out/mod.rs @@ -132,15 +132,21 @@ pub fn console_device_list(mut list: Vec) { ("Rt".to_string(), Style::new()), ]); for item in list { + let name = if item.wire_guard { + format!("{}(wg)", item.name) + } else { + item.name + }; if &item.status == "Online" { - if item.client_secret != item.current_client_secret - || (!item.current_client_secret_hash.is_empty() - && !item.client_secret_hash.is_empty() - && item.current_client_secret_hash != item.client_secret_hash) + if !item.wire_guard + && (item.client_secret != item.current_client_secret + || (!item.current_client_secret_hash.is_empty() + && !item.client_secret_hash.is_empty() + && item.current_client_secret_hash != item.client_secret_hash)) { //加密状态不一致，无法通信的 out_list.push(vec![ - (item.name, Style::new().red()), + (name, Style::new().red()), (item.virtual_ip, Style::new().red()), (item.status, Style::new().red()), ("Mismatch".to_string(), Style::new().red()), @@ -149,7 +155,7 @@ pub fn console_device_list(mut list: Vec) { } else { if item.nat_traversal_type.contains("p2p") { out_list.push(vec![ - (item.name, Style::new().green()), + (name, Style::new().green()), (item.virtual_ip, Style::new().green()), (item.status, Style::new().green()), (item.nat_traversal_type, Style::new().green()), @@ -157,7 +163,7 @@ pub fn console_device_list(mut list: Vec) { ]); } else { out_list.push(vec![ - (item.name, Style::new().yellow()), + (name, Style::new().yellow()), (item.virtual_ip, Style::new().yellow()), (item.status, Style::new().yellow()), (item.nat_traversal_type, Style::new().yellow()), @@ -167,7 +173,7 @@ pub fn console_device_list(mut list: Vec) { } } else { out_list.push(vec![ - (item.name, Style::new().color256(102)), + (name, Style::new().color256(102)), (item.virtual_ip, Style::new().color256(102)), (item.status, Style::new().color256(102)), ("".to_string(), Style::new().color256(102)), diff --git a/vn-link-cli/Cargo.toml b/vn-link-cli/Cargo.toml index 8a1d4b7..e4e3559 100644 --- a/vn-link-cli/Cargo.toml +++ b/vn-link-cli/Cargo.toml @@ -11,7 +11,7 @@ log = "0.4.17" [features] default = ["default-feature"] -default-feature = ["server_encrypt", "aes_gcm", "aes_cbc", "aes_ecb", "sm4_cbc", "chacha20_poly1305", "port_mapping", "log", "command", "file_config", "lz4"] +default-feature = ["server_encrypt", "aes_gcm", "aes_cbc", "aes_ecb", "sm4_cbc", "chacha20_poly1305", "port_mapping", "log", "command", "file_config", "lz4", "ws"] openssl = ["vn-link/openssl", "common/openssl"] openssl-vendored = ["vn-link/openssl-vendored", "common/openssl-vendored"] diff --git a/vnt/proto/message.proto b/vnt/proto/message.proto index 47e9e97..197c884 100644 --- a/vnt/proto/message.proto +++ b/vnt/proto/message.proto @@ -43,6 +43,7 @@ message DeviceInfo { uint32 device_status = 3; bool client_secret = 4; bytes client_secret_hash = 5; + bool wireguard = 6; } message DeviceList { diff --git a/vnt/src/channel/punch.rs b/vnt/src/channel/punch.rs index 5b2a760..07764a8 100644 --- a/vnt/src/channel/punch.rs +++ b/vnt/src/channel/punch.rs @@ -264,9 +264,8 @@ impl Punch { if let Some(ipv4_addr) = nat_info.local_tcp_ipv4addr() { self.connect_tcp(buf, ipv4_addr) } - if nat_info.nat_type == NatType::Cone && nat_info.public_ips.len() == 1 { - let addr = - SocketAddr::V4(SocketAddrV4::new(nat_info.public_ips[0], nat_info.tcp_port)); + for ip in &nat_info.public_ips { + let addr = SocketAddr::V4(SocketAddrV4::new(*ip, nat_info.tcp_port)); self.connect_tcp(buf, addr) } } diff --git a/vnt/src/channel/sender.rs b/vnt/src/channel/sender.rs index 08e9252..5cfa27c 100644 --- a/vnt/src/channel/sender.rs +++ b/vnt/src/channel/sender.rs @@ -1,9 +1,11 @@ +use std::collections::HashMap; use std::io; use std::net::{Ipv4Addr, SocketAddr}; use std::sync::mpsc::{SyncSender, TrySendError}; use std::sync::Arc; use crossbeam_utils::atomic::AtomicCell; +use parking_lot::Mutex; use tokio::sync::mpsc::Sender; use crate::channel::context::ChannelContext; @@ -11,7 +13,7 @@ use crate::channel::notify::AcceptNotify; use crate::cipher::Cipher; use crate::compression::Compressor; use crate::external_route::ExternalRoute; -use crate::handle::CurrentDeviceInfo; +use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo}; use crate::protocol; use crate::protocol::{ip_turn_packet, NetPacket}; @@ -21,7 +23,10 @@ pub struct IpPacketSender { current_device: Arc>, compressor: Compressor, client_cipher: Cipher, + server_cipher: Cipher, ip_route: ExternalRoute, + device_map: Arc)>>, + allow_wire_guard: bool, } impl IpPacketSender { @@ -30,14 +35,20 @@ impl IpPacketSender { current_device: Arc>, compressor: Compressor, client_cipher: Cipher, + server_cipher: Cipher, ip_route: ExternalRoute, + device_map: Arc)>>, + allow_wire_guard: bool, ) -> Self { Self { context, current_device, compressor, client_cipher, + server_cipher, ip_route, + device_map, + allow_wire_guard, } } pub fn self_virtual_ip(&self) -> Ipv4Addr { @@ -58,19 +69,54 @@ impl IpPacketSender { if let Some(v) = self.ip_route.route(&dest_ip) { dest_ip = v; } - if dest_ip.is_multicast() || dest_ip.is_broadcast() || dest_ip == device_info.broadcast_ip { + if dest_ip.is_multicast() { //广播 dest_ip = Ipv4Addr::BROADCAST; } - let mut net_packet = NetPacket::new0(data_len, buf)?; - let mut auxiliary = NetPacket::new(auxiliary_buf)?; net_packet.set_default_version(); net_packet.set_protocol(protocol::Protocol::IpTurn); net_packet.set_transport_protocol(ip_turn_packet::Protocol::Ipv4.into()); net_packet.first_set_ttl(6); net_packet.set_source(src_ip); net_packet.set_destination(dest_ip); + if self.allow_wire_guard { + if dest_ip.is_broadcast() || dest_ip == device_info.broadcast_ip { + let exists_wg = self + .device_map + .lock() + .1 + .values() + .any(|v| v.status.is_online() && v.wireguard); + if exists_wg { + send_to_wg_broadcast( + &self.context, + &net_packet, + &self.server_cipher, + &device_info, + )?; + } + } else { + let guard = self.device_map.lock(); + if let Some(peer_info) = guard.1.get(&dest_ip) { + if peer_info.wireguard { + if peer_info.status.is_offline() { + return Ok(()); + } + drop(guard); + send_to_wg( + &self.context, + &mut net_packet, + &self.server_cipher, + &device_info, + )?; + return Ok(()); + } + } + } + } + + let mut auxiliary = NetPacket::new(auxiliary_buf)?; let mut net_packet = if self.compressor.compress(&net_packet, &mut auxiliary)? { auxiliary.set_default_version(); @@ -84,7 +130,7 @@ impl IpPacketSender { net_packet }; self.client_cipher.encrypt_ipv4(&mut net_packet)?; - if dest_ip.is_broadcast() { + if dest_ip.is_broadcast() || dest_ip == device_info.broadcast_ip { //走服务端广播 self.context .send_default(&net_packet, device_info.connect_server)?; @@ -105,6 +151,40 @@ impl IpPacketSender { } } +pub fn send_to_wg_broadcast( + sender: &ChannelContext, + net_packet: &NetPacket<&mut [u8]>, + server_cipher: &Cipher, + current_device: &CurrentDeviceInfo, +) -> anyhow::Result<()> { + let mut copy_packet = NetPacket::new0(net_packet.data_len(), [0; 65536])?; + copy_packet.set_default_version(); + copy_packet.set_protocol(protocol::Protocol::IpTurn); + copy_packet.set_transport_protocol(ip_turn_packet::Protocol::WGIpv4.into()); + copy_packet.first_set_ttl(6); + copy_packet.set_source(net_packet.source()); + copy_packet.set_destination(net_packet.destination()); + copy_packet.set_gateway_flag(true); + copy_packet.set_payload(net_packet.payload())?; + server_cipher.encrypt_ipv4(&mut copy_packet)?; + sender.send_default(©_packet, current_device.connect_server)?; + + Ok(()) +} +pub fn send_to_wg( + sender: &ChannelContext, + net_packet: &mut NetPacket<&mut [u8]>, + server_cipher: &Cipher, + current_device: &CurrentDeviceInfo, +) -> anyhow::Result<()> { + net_packet.set_transport_protocol(ip_turn_packet::Protocol::WGIpv4.into()); + net_packet.set_gateway_flag(true); + server_cipher.encrypt_ipv4(net_packet)?; + sender.send_default(&net_packet, current_device.connect_server)?; + + Ok(()) +} + pub struct AcceptSocketSender { sender: SyncSender, notify: AcceptNotify, diff --git a/vnt/src/core/conn.rs b/vnt/src/core/conn.rs index 6aab462..cacd907 100644 --- a/vnt/src/core/conn.rs +++ b/vnt/src/core/conn.rs @@ -66,12 +66,13 @@ pub struct VntInner { config: Config, current_device: Arc>, nat_test: NatTest, - device_list: Arc)>>, + device_map: Arc)>>, context: Arc>>, peer_nat_info_map: Arc>>, client_secret_hash: Option<[u8; 16]>, compressor: Compressor, client_cipher: Cipher, + server_cipher: Cipher, external_route: ExternalRoute, up_traffic_meter: Option, down_traffic_meter: Option, @@ -128,8 +129,8 @@ impl VntInner { config.server_address, ))); //设备列表 - let device_list: Arc)>> = - Arc::new(Mutex::new((0, Vec::with_capacity(16)))); + let device_map: Arc)>> = + Arc::new(Mutex::new((0, HashMap::with_capacity(16)))); //基础信息 let config_info = BaseConfigInfo::new( config.name.clone(), @@ -147,6 +148,7 @@ impl VntInner { #[cfg(feature = "integrated_tun")] #[cfg(any(target_os = "windows", target_os = "linux", target_os = "macos"))] config.device_name.clone(), + config.allow_wire_guard, ); // 服务停止管理器 let stop_manager = { @@ -228,7 +230,7 @@ impl VntInner { proxy_map.clone(), client_cipher.clone(), server_cipher.clone(), - device_list.clone(), + device_map.clone(), config.compressor, device.clone().into_device_adapter(), ) @@ -241,7 +243,7 @@ impl VntInner { client_cipher.clone(), current_device.clone(), device, - device_list.clone(), + device_map.clone(), config_info.clone(), nat_test.clone(), callback.clone(), @@ -287,7 +289,7 @@ impl VntInner { { let context = context.clone(); let nat_test = nat_test.clone(); - let device_list = device_list.clone(); + let device_map = device_map.clone(); let config_info = config_info.clone(); let current_device = current_device.clone(); if !config.use_channel_type.is_only_relay() { @@ -300,13 +302,14 @@ impl VntInner { ); } let client_cipher = client_cipher.clone(); + let server_cipher = server_cipher.clone(); //延迟启动 scheduler.timeout(Duration::from_secs(3), move |scheduler| { start( scheduler, context, nat_test, - device_list, + device_map, current_device, client_cipher, server_cipher, @@ -323,12 +326,13 @@ impl VntInner { config, current_device, nat_test, - device_list, + device_map, context: Arc::new(Mutex::new(Some(context))), peer_nat_info_map, client_secret_hash: config_info.client_secret_hash, compressor, client_cipher, + server_cipher, external_route, up_traffic_meter, down_traffic_meter, @@ -340,7 +344,7 @@ pub fn start( scheduler: &Scheduler, context: ChannelContext, nat_test: NatTest, - device_list: Arc)>>, + device_map: Arc)>>, current_device: Arc>, client_cipher: Cipher, server_cipher: Cipher, @@ -354,7 +358,7 @@ pub fn start( &scheduler, context.clone(), current_device.clone(), - device_list.clone(), + device_map.clone(), client_cipher.clone(), server_cipher.clone(), ); @@ -374,7 +378,7 @@ pub fn start( &scheduler, context.clone(), current_device.clone(), - device_list.clone(), + device_map.clone(), client_cipher.clone(), ); } @@ -393,7 +397,7 @@ pub fn start( &scheduler, context.clone(), nat_test.clone(), - device_list.clone(), + device_map.clone(), current_device.clone(), client_cipher.clone(), punch_receiver, @@ -432,10 +436,10 @@ impl VntInner { self.nat_test.nat_info() } pub fn device_list(&self) -> Vec { - let device_list_lock = self.device_list.lock(); + let device_list_lock = self.device_map.lock(); let (_epoch, device_list) = device_list_lock.clone(); drop(device_list_lock); - device_list + device_list.into_values().collect() } pub fn route(&self, ip: &Ipv4Addr) -> Option { self.context.lock().as_ref()?.route_table.route_one(ip) @@ -507,7 +511,10 @@ impl VntInner { self.current_device.clone(), self.compressor.clone(), self.client_cipher.clone(), + self.server_cipher.clone(), self.external_route.clone(), + self.device_map.clone(), + self.config.allow_wire_guard, )) } else { None diff --git a/vnt/src/core/mod.rs b/vnt/src/core/mod.rs index 44238fa..56701ad 100644 --- a/vnt/src/core/mod.rs +++ b/vnt/src/core/mod.rs @@ -51,6 +51,7 @@ pub struct Config { pub port_mapping_list: Vec<(bool, SocketAddr, String)>, pub compressor: Compressor, pub enable_traffic: bool, + pub allow_wire_guard: bool, } impl Config { @@ -88,6 +89,8 @@ impl Config { #[cfg(feature = "port_mapping")] port_mapping_list: Vec, compressor: Compressor, enable_traffic: bool, + // 允许传递wg流量 + allow_wire_guard: bool, ) -> anyhow::Result { for x in stun_server.iter_mut() { if !x.contains(":") { @@ -180,6 +183,7 @@ impl Config { port_mapping_list, compressor, enable_traffic, + allow_wire_guard, }) } } diff --git a/vnt/src/handle/maintain/heartbeat.rs b/vnt/src/handle/maintain/heartbeat.rs index f18ac8d..d5e21ee 100644 --- a/vnt/src/handle/maintain/heartbeat.rs +++ b/vnt/src/handle/maintain/heartbeat.rs @@ -1,3 +1,4 @@ +use std::collections::HashMap; use std::net::Ipv4Addr; use std::sync::Arc; use std::time::Duration; @@ -19,14 +20,14 @@ pub fn heartbeat( scheduler: &Scheduler, context: ChannelContext, current_device_info: Arc>, - device_list: Arc)>>, + device_map: Arc)>>, client_cipher: Cipher, server_cipher: Cipher, ) { heartbeat0( &context, ¤t_device_info.load(), - &device_list, + &device_map, &client_cipher, &server_cipher, ); @@ -36,7 +37,7 @@ pub fn heartbeat( s, context, current_device_info, - device_list, + device_map, client_cipher, server_cipher, ) @@ -49,7 +50,7 @@ pub fn heartbeat( fn heartbeat0( context: &ChannelContext, current_device: &CurrentDeviceInfo, - device_list: &Mutex<(u16, Vec)>, + device_map: &Mutex<(u16, HashMap)>, client_cipher: &Cipher, server_cipher: &Cipher, ) { @@ -57,7 +58,7 @@ fn heartbeat0( let src_ip = current_device.virtual_ip; // 可能服务器ip发生变化，导致发送失败 let mut is_send_gateway = false; - match heartbeat_packet_server(device_list, server_cipher, src_ip, gateway_ip) { + match heartbeat_packet_server(device_map, server_cipher, src_ip, gateway_ip) { Ok(net_packet) => { if let Err(e) = context.send_default(&net_packet, current_device.connect_server) { log::warn!("heartbeat err={:?}", e) @@ -75,7 +76,7 @@ fn heartbeat0( if is_send_gateway { continue; } - heartbeat_packet_server(device_list, server_cipher, src_ip, gateway_ip) + heartbeat_packet_server(device_map, server_cipher, src_ip, gateway_ip) } else { heartbeat_packet_client(client_cipher, src_ip, dest_ip) }; @@ -92,9 +93,9 @@ fn heartbeat0( } } } - let peer_list = { device_list.lock().1.clone() }; - for peer in &peer_list { - if !peer.status.is_online() { + let peer_list = { device_map.lock().1.clone() }; + for peer in peer_list.values() { + if !peer.status.is_online() || peer.wireguard { continue; } if current_device.is_gateway(&peer.virtual_ip) { @@ -124,11 +125,11 @@ pub fn client_relay( scheduler: &Scheduler, context: ChannelContext, current_device: Arc>, - device_list: Arc)>>, + device_map: Arc)>>, client_cipher: Cipher, ) { let rs = scheduler.timeout(Duration::from_secs(30), move |s| { - client_relay_(s, context, current_device, device_list, client_cipher) + client_relay_(s, context, current_device, device_map, client_cipher) }); if !rs { log::info!("定时任务停止"); @@ -140,19 +141,19 @@ fn client_relay_( scheduler: &Scheduler, context: ChannelContext, current_device: Arc>, - device_list: Arc)>>, + device_map: Arc)>>, client_cipher: Cipher, ) { if let Err(e) = client_relay0( &context, ¤t_device.load(), - &device_list, + &device_map, &client_cipher, ) { log::error!("{:?}", e); } let rs = scheduler.timeout(Duration::from_secs(30), move |s| { - client_relay_(s, context, current_device, device_list, client_cipher) + client_relay_(s, context, current_device, device_map, client_cipher) }); if !rs { log::info!("定时任务停止"); @@ -162,17 +163,20 @@ fn client_relay_( fn client_relay0( context: &ChannelContext, current_device: &CurrentDeviceInfo, - device_list: &Mutex<(u16, Vec)>, + device_map: &Mutex<(u16, HashMap)>, client_cipher: &Cipher, ) -> anyhow::Result<()> { // 离线了不再探测 if current_device.status.offline() { return Ok(()); } - let peer_list = { device_list.lock().1.clone() }; + let peer_list = { device_map.lock().1.clone() }; let mut routes = context.route_table.route_table_p2p(); - for peer in &peer_list { - if !peer.status.is_online() || peer.virtual_ip == current_device.virtual_ip { + for peer in peer_list.values() { + if peer.wireguard + || !peer.status.is_online() + || peer.virtual_ip == current_device.virtual_ip + { continue; } if context @@ -232,14 +236,14 @@ fn heartbeat_packet_client( } fn heartbeat_packet_server( - device_list: &Mutex<(u16, Vec)>, + device_map: &Mutex<(u16, HashMap)>, server_cipher: &Cipher, src: Ipv4Addr, dest: Ipv4Addr, ) -> anyhow::Result> { let mut net_packet = heartbeat_packet(src, dest)?; let mut ping = PingPacket::new(net_packet.payload_mut())?; - ping.set_epoch(device_list.lock().0); + ping.set_epoch(device_map.lock().0); net_packet.set_gateway_flag(true); server_cipher.encrypt_ipv4(&mut net_packet)?; Ok(net_packet) diff --git a/vnt/src/handle/maintain/punch.rs b/vnt/src/handle/maintain/punch.rs index 51a21e9..8bbbe84 100644 --- a/vnt/src/handle/maintain/punch.rs +++ b/vnt/src/handle/maintain/punch.rs @@ -90,7 +90,7 @@ pub fn punch( scheduler: &Scheduler, context: ChannelContext, nat_test: NatTest, - device_list: Arc)>>, + device_map: Arc)>>, current_device: Arc>, client_cipher: Cipher, receiver: PunchReceiver, @@ -102,7 +102,7 @@ pub fn punch( scheduler, context, nat_test, - device_list, + device_map, current_device.clone(), client_cipher.clone(), 0, @@ -170,7 +170,7 @@ fn punch_request( scheduler: &Scheduler, context: ChannelContext, nat_test: NatTest, - device_list: Arc)>>, + device_map: Arc)>>, current_device: Arc>, client_cipher: Cipher, count: usize, @@ -182,7 +182,7 @@ fn punch_request( if let Err(e) = punch0( &context, &nat_test, - &device_list, + &device_map, curr, &client_cipher, &punch_record, @@ -201,7 +201,7 @@ fn punch_request( s, context, nat_test, - device_list, + device_map, current_device, client_cipher, count + 1, @@ -218,7 +218,7 @@ fn punch_request( fn punch0( context: &ChannelContext, nat_test: &NatTest, - device_list: &Arc)>>, + device_map: &Arc)>>, current_device: CurrentDeviceInfo, client_cipher: &Cipher, punch_record: &Mutex>, @@ -237,11 +237,11 @@ fn punch0( return Ok(()); } let current_ip = current_device.virtual_ip; - let mut list: Vec = device_list + let mut list: Vec = device_map .lock() .1 - .iter() - .filter(|info| info.status.is_online() && info.virtual_ip > current_ip) + .values() + .filter(|info| !info.wireguard && info.status.is_online() && info.virtual_ip > current_ip) .cloned() .collect(); list.shuffle(&mut rand::thread_rng()); diff --git a/vnt/src/handle/mod.rs b/vnt/src/handle/mod.rs index e39305f..754b7b3 100644 --- a/vnt/src/handle/mod.rs +++ b/vnt/src/handle/mod.rs @@ -29,6 +29,7 @@ pub struct PeerDeviceInfo { pub status: PeerDeviceStatus, pub client_secret: bool, pub client_secret_hash: Vec, + pub wireguard: bool, } impl PeerDeviceInfo { @@ -38,6 +39,7 @@ impl PeerDeviceInfo { status: u8, client_secret: bool, client_secret_hash: Vec, + wireguard: bool, ) -> Self { Self { virtual_ip, @@ -45,6 +47,7 @@ impl PeerDeviceInfo { status: PeerDeviceStatus::from(status), client_secret, client_secret_hash, + wireguard, } } } @@ -66,6 +69,7 @@ pub struct BaseConfigInfo { #[cfg(feature = "integrated_tun")] #[cfg(any(target_os = "windows", target_os = "linux", target_os = "macos"))] pub device_name: Option, + pub allow_wire_guard: bool, } impl BaseConfigInfo { @@ -85,6 +89,7 @@ impl BaseConfigInfo { #[cfg(feature = "integrated_tun")] #[cfg(any(target_os = "windows", target_os = "linux", target_os = "macos"))] device_name: Option, + allow_wire_guard: bool, ) -> Self { Self { name, @@ -102,6 +107,7 @@ impl BaseConfigInfo { #[cfg(feature = "integrated_tun")] #[cfg(any(target_os = "windows", target_os = "linux", target_os = "macos"))] device_name, + allow_wire_guard, } } } @@ -116,6 +122,9 @@ impl PeerDeviceStatus { pub fn is_online(&self) -> bool { self == &PeerDeviceStatus::Online } + pub fn is_offline(&self) -> bool { + self == &PeerDeviceStatus::Offline + } } impl Into for PeerDeviceStatus { diff --git a/vnt/src/handle/recv_data/client.rs b/vnt/src/handle/recv_data/client.rs index 13879d3..fc61721 100644 --- a/vnt/src/handle/recv_data/client.rs +++ b/vnt/src/handle/recv_data/client.rs @@ -194,6 +194,9 @@ impl ClientPacketHandler { } self.device.write(net_packet.payload())?; } + ip_turn_packet::Protocol::WGIpv4 => { + // WG客户端的数据不会直接发过来，不用处理 + } ip_turn_packet::Protocol::Ipv4Broadcast => { //客户端不帮忙转发广播包，所以不会出现这种类型的数据 } diff --git a/vnt/src/handle/recv_data/mod.rs b/vnt/src/handle/recv_data/mod.rs index e2bd5d6..4b745c1 100644 --- a/vnt/src/handle/recv_data/mod.rs +++ b/vnt/src/handle/recv_data/mod.rs @@ -80,7 +80,7 @@ impl RecvDataHandler { client_cipher: Cipher, current_device: Arc>, device: Device, - device_list: Arc)>>, + device_map: Arc)>>, config_info: BaseConfigInfo, nat_test: NatTest, callback: Call, @@ -101,7 +101,7 @@ impl RecvDataHandler { server_cipher, current_device.clone(), device.clone(), - device_list, + device_map, config_info, nat_test.clone(), callback, diff --git a/vnt/src/handle/recv_data/server.rs b/vnt/src/handle/recv_data/server.rs index 66ecfc0..2abce5c 100644 --- a/vnt/src/handle/recv_data/server.rs +++ b/vnt/src/handle/recv_data/server.rs @@ -1,4 +1,5 @@ use anyhow::anyhow; +use std::collections::HashMap; use std::io; use std::net::Ipv4Addr; use std::sync::Arc; @@ -42,7 +43,7 @@ pub struct ServerPacketHandler { server_cipher: Cipher, current_device: Arc>, device: Device, - device_list: Arc)>>, + device_map: Arc)>>, config_info: BaseConfigInfo, nat_test: NatTest, callback: Call, @@ -60,7 +61,7 @@ impl ServerPacketHandler { server_cipher: Cipher, current_device: Arc>, device: Device, - device_list: Arc)>>, + device_map: Arc)>>, config_info: BaseConfigInfo, nat_test: NatTest, callback: Call, @@ -75,7 +76,7 @@ impl ServerPacketHandler { server_cipher, current_device, device, - device_list, + device_map, config_info, nat_test, callback, @@ -246,6 +247,11 @@ impl PacketHandler for ServerPacketHandl _ => {} } } + ip_turn_packet::Protocol::WGIpv4 => { + if self.config_info.allow_wire_guard { + self.device.write(net_packet.payload())?; + } + } ip_turn_packet::Protocol::Ipv4Broadcast => {} ip_turn_packet::Protocol::Unknown(_) => {} } @@ -353,7 +359,8 @@ impl ServerPacketHandler { ); log::info!("tun信息{:?}", tun_info); self.callback.create_tun(tun_info); - self.tun_device_helper.start(device)?; + self.tun_device_helper + .start(device, self.config_info.allow_wire_guard)?; } Err(e) => { log::error!("{:?}", e); @@ -435,14 +442,18 @@ impl ServerPacketHandler { info.device_status as u8, info.client_secret, info.client_secret_hash, + info.wireguard, ) }) .collect(); { - let mut dev = self.device_list.lock(); + let mut dev = self.device_map.lock(); //这里可能会收到旧的消息，但是随着时间推移总会收到新的 dev.0 = epoch; - dev.1 = ip_list.clone(); + dev.1.clear(); + for info in ip_list.clone() { + dev.1.insert(info.virtual_ip, info); + } } self.callback.peer_client_list( ip_list @@ -506,7 +517,7 @@ impl ServerPacketHandler { self.callback.error(err); //掉线epoch要归零 { - let mut dev = self.device_list.lock(); + let mut dev = self.device_map.lock(); dev.0 = 0; drop(dev); } @@ -554,7 +565,7 @@ impl ServerPacketHandler { let rt = (current_time - pong_packet.time()) as i64; let route = Route::from(route_key, metric, rt); context.route_table.add_route(net_packet.source(), route); - let epoch = self.device_list.lock().0; + let epoch = self.device_map.lock().0; if pong_packet.epoch() != epoch { //纪元不一致，可能有新客户端连接，向服务端拉取客户端列表 let mut poll_device = NetPacket::new_encrypt([0; 12 + ENCRYPTION_RESERVED])?; diff --git a/vnt/src/handle/tun_tap/tun_handler.rs b/vnt/src/handle/tun_tap/tun_handler.rs index 6aee8f2..ef57181 100644 --- a/vnt/src/handle/tun_tap/tun_handler.rs +++ b/vnt/src/handle/tun_tap/tun_handler.rs @@ -1,10 +1,10 @@ +use crossbeam_utils::atomic::AtomicCell; +use parking_lot::Mutex; +use std::collections::HashMap; use std::net::Ipv4Addr; use std::sync::Arc; use std::{io, thread}; -use crossbeam_utils::atomic::AtomicCell; -use parking_lot::Mutex; - use packet::icmp::icmp::IcmpPacket; use packet::icmp::Kind; use packet::ip::ipv4::packet::IpV4Packet; @@ -13,6 +13,7 @@ use tun::device::IFace; use tun::Device; use crate::channel::context::ChannelContext; +use crate::channel::sender::{send_to_wg, send_to_wg_broadcast}; use crate::cipher::Cipher; use crate::compression::Compressor; use crate::external_route::ExternalRoute; @@ -27,7 +28,6 @@ use crate::protocol::body::ENCRYPTION_RESERVED; use crate::protocol::ip_turn_packet::BroadcastPacket; use crate::protocol::{ip_turn_packet, NetPacket, MAX_TTL}; use crate::util::StopManager; - fn icmp(device_writer: &Device, mut ipv4_packet: IpV4Packet<&mut [u8]>) -> anyhow::Result<()> { if ipv4_packet.protocol() == Protocol::Icmp { let mut icmp = IcmpPacket::new(ipv4_packet.payload_mut())?; @@ -53,9 +53,10 @@ pub fn start( #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, client_cipher: Cipher, server_cipher: Cipher, - device_list: Arc)>>, + device_map: Arc)>>, compressor: Compressor, device_stop: DeviceStop, + allow_wire_guard: bool, ) -> io::Result<()> { thread::Builder::new() .name("tunHandlerS".into()) @@ -70,9 +71,10 @@ pub fn start( ip_proxy_map, client_cipher, server_cipher, - device_list, + device_map, compressor, device_stop, + allow_wire_guard, ) { log::warn!("stop:{}", e); } @@ -86,13 +88,13 @@ fn broadcast( sender: &ChannelContext, net_packet: &mut NetPacket<&mut [u8]>, current_device: &CurrentDeviceInfo, - device_list: &Mutex<(u16, Vec)>, + device_map: &Mutex<(u16, HashMap)>, ) -> anyhow::Result<()> { - let list: Vec = device_list + let list: Vec = device_map .lock() .1 - .iter() - .filter(|info| info.status.is_online()) + .values() + .filter(|info| !info.wireguard && info.status.is_online()) .map(|info| info.virtual_ip) .collect(); const MAX_COUNT: usize = 8; @@ -177,8 +179,9 @@ pub(crate) fn handle( #[cfg(feature = "ip_proxy")] proxy_map: &Option, client_cipher: &Cipher, server_cipher: &Cipher, - device_list: &Mutex<(u16, Vec)>, + device_map: &Mutex<(u16, HashMap)>, compressor: &Compressor, + allow_wire_guard: bool, ) -> anyhow::Result<()> { //忽略掉结构不对的情况（ipv6数据、win tap会读到空数据），不然日志打印太多了 let ipv4_packet = match IpV4Packet::new(&mut buf[12..data_len]) { @@ -237,6 +240,33 @@ pub(crate) fn handle( dest_ip = Ipv4Addr::BROADCAST; net_packet.set_destination(Ipv4Addr::BROADCAST); } + let is_broadcast = dest_ip.is_broadcast() || current_device.broadcast_ip == dest_ip; + if allow_wire_guard { + if is_broadcast { + // wg客户端和vnt客户端分开广播 + let exists_wg = device_map + .lock() + .1 + .values() + .any(|v| v.status.is_online() && v.wireguard); + if exists_wg { + send_to_wg_broadcast(context, &net_packet, server_cipher, ¤t_device)?; + } + } else { + // 如果是wg客户端则发到vnts转发 + let guard = device_map.lock(); + if let Some(peer_info) = guard.1.get(&dest_ip) { + if peer_info.wireguard { + if peer_info.status.is_offline() { + return Ok(()); + } + drop(guard); + send_to_wg(context, &mut net_packet, server_cipher, ¤t_device)?; + return Ok(()); + } + } + } + } let mut net_packet = if compressor.compress(&net_packet, &mut out)? { out.set_default_version(); @@ -257,7 +287,7 @@ pub(crate) fn handle( context, &mut net_packet, ¤t_device, - device_list, + device_map, )?; return Ok(()); } diff --git a/vnt/src/handle/tun_tap/unix.rs b/vnt/src/handle/tun_tap/unix.rs index a75d02c..80d7a22 100644 --- a/vnt/src/handle/tun_tap/unix.rs +++ b/vnt/src/handle/tun_tap/unix.rs @@ -13,7 +13,9 @@ use mio::event::Source; use mio::unix::SourceFd; use mio::{Events, Interest, Poll, Token, Waker}; use parking_lot::Mutex; +use std::collections::HashMap; use std::io; +use std::net::Ipv4Addr; use std::os::fd::AsRawFd; use std::sync::Arc; use tun::Device; @@ -30,9 +32,10 @@ pub(crate) fn start_simple( #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, client_cipher: Cipher, server_cipher: Cipher, - device_list: Arc)>>, + device_map: Arc)>>, compressor: Compressor, device_stop: DeviceStop, + allow_wire_guard: bool, ) -> anyhow::Result<()> { let poll = Poll::new()?; let waker = Arc::new(Waker::new(poll.registry(), STOP)?); @@ -61,8 +64,9 @@ pub(crate) fn start_simple( ip_proxy_map, client_cipher, server_cipher, - device_list, + device_map, compressor, + allow_wire_guard, ) { log::error!("{:?}", e); }; @@ -83,8 +87,9 @@ fn start_simple0( #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, client_cipher: Cipher, server_cipher: Cipher, - device_list: Arc)>>, + device_map: Arc)>>, compressor: Compressor, + allow_wire_guard: bool, ) -> anyhow::Result<()> { let mut buf = [0; BUFFER_SIZE]; let mut extend = [0; BUFFER_SIZE]; @@ -134,8 +139,9 @@ fn start_simple0( &ip_proxy_map, &client_cipher, &server_cipher, - &device_list, + &device_map, &compressor, + allow_wire_guard, ) { Ok(_) => {} Err(e) => { diff --git a/vnt/src/handle/tun_tap/windows.rs b/vnt/src/handle/tun_tap/windows.rs index 0df3e4a..3c91b56 100644 --- a/vnt/src/handle/tun_tap/windows.rs +++ b/vnt/src/handle/tun_tap/windows.rs @@ -10,6 +10,8 @@ use crate::ip_proxy::IpProxyMap; use crate::util::StopManager; use crossbeam_utils::atomic::AtomicCell; use parking_lot::Mutex; +use std::collections::HashMap; +use std::net::Ipv4Addr; use std::sync::Arc; use tun::device::IFace; use tun::Device; @@ -23,9 +25,10 @@ pub(crate) fn start_simple( #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, client_cipher: Cipher, server_cipher: Cipher, - device_list: Arc)>>, + device_map: Arc)>>, compressor: Compressor, device_stop: DeviceStop, + allow_wire_guard: bool, ) -> anyhow::Result<()> { let worker = { let device = device.clone(); @@ -54,8 +57,9 @@ pub(crate) fn start_simple( ip_proxy_map, client_cipher, server_cipher, - device_list, + device_map, compressor, + allow_wire_guard, ) { log::error!("{:?}", e); } @@ -74,8 +78,9 @@ fn start_simple0( #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, client_cipher: Cipher, server_cipher: Cipher, - device_list: Arc)>>, + device_map: Arc)>>, compressor: Compressor, + allow_wire_guard: bool, ) -> anyhow::Result<()> { let mut buf = [0; BUFFER_SIZE]; let mut extend = [0; BUFFER_SIZE]; @@ -96,8 +101,9 @@ fn start_simple0( &ip_proxy_map, &client_cipher, &server_cipher, - &device_list, + &device_map, &compressor, + allow_wire_guard, ) { Ok(_) => {} Err(e) => { diff --git a/vnt/src/protocol/ip_turn_packet.rs b/vnt/src/protocol/ip_turn_packet.rs index b009f3f..edccdd5 100644 --- a/vnt/src/protocol/ip_turn_packet.rs +++ b/vnt/src/protocol/ip_turn_packet.rs @@ -1,9 +1,12 @@ +#![allow(dead_code)] + use std::io; use std::net::Ipv4Addr; #[derive(Copy, Clone, Eq, PartialEq, Debug)] pub enum Protocol { Ipv4, + WGIpv4, Ipv4Broadcast, Unknown(u8), } @@ -12,16 +15,18 @@ impl From for Protocol { fn from(value: u8) -> Self { match value { 4 => Protocol::Ipv4, + 5 => Protocol::WGIpv4, 201 => Protocol::Ipv4Broadcast, val => Protocol::Unknown(val), } } } -impl Into for Protocol { - fn into(self) -> u8 { - match self { +impl From for u8 { + fn from(val: Protocol) -> Self { + match val { Protocol::Ipv4 => 4, + Protocol::WGIpv4 => 5, Protocol::Ipv4Broadcast => 201, Protocol::Unknown(val) => val, } diff --git a/vnt/src/tun_tap_device/tun_create_helper.rs b/vnt/src/tun_tap_device/tun_create_helper.rs index 76babcf..12d1b1b 100644 --- a/vnt/src/tun_tap_device/tun_create_helper.rs +++ b/vnt/src/tun_tap_device/tun_create_helper.rs @@ -1,4 +1,6 @@ +use std::collections::HashMap; use std::io; +use std::net::Ipv4Addr; use std::sync::Arc; use crossbeam_utils::atomic::AtomicCell; @@ -67,7 +69,7 @@ struct TunDeviceHelperInner { ip_proxy_map: Option, client_cipher: Cipher, server_cipher: Cipher, - device_list: Arc)>>, + device_map: Arc)>>, compressor: Compressor, } @@ -80,7 +82,7 @@ impl TunDeviceHelper { #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, client_cipher: Cipher, server_cipher: Cipher, - device_list: Arc)>>, + device_map: Arc)>>, compressor: Compressor, device_adapter: DeviceAdapter, ) -> Self { @@ -93,7 +95,7 @@ impl TunDeviceHelper { ip_proxy_map, client_cipher, server_cipher, - device_list, + device_map, compressor, }; Self { @@ -117,7 +119,7 @@ impl TunDeviceHelper { } } /// 要保证先stop 再start - pub fn start(&self, device: Arc) -> io::Result<()> { + pub fn start(&self, device: Arc, allow_wire_guard: bool) -> io::Result<()> { self.device_adapter.insert(device.clone()); let device_stop = DeviceStop::default(); let s = self.device_stop.lock().replace(device_stop.clone()); @@ -133,9 +135,10 @@ impl TunDeviceHelper { inner.ip_proxy_map, inner.client_cipher, inner.server_cipher, - inner.device_list, + inner.device_map, inner.compressor, device_stop, + allow_wire_guard, ) } }