diff --git a/vnt/src/core/conn.rs b/vnt/src/core/conn.rs new file mode 100644 index 0000000..01cc1ad --- /dev/null +++ b/vnt/src/core/conn.rs @@ -0,0 +1,354 @@ +use std::collections::HashMap; +use std::io; +use std::net::Ipv4Addr; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::mpsc::{sync_channel, Receiver}; +use std::sync::Arc; +use std::time::Duration; + +use crossbeam_utils::atomic::AtomicCell; +use parking_lot::{Mutex, RwLock}; +use rand::Rng; + +use tun::device::IFace; + +use crate::channel::context::Context; +use crate::channel::idle::Idle; +use crate::channel::punch::{NatInfo, Punch}; +use crate::channel::{init_channel, init_context, Route, RouteKey}; +use crate::cipher::Cipher; +#[cfg(feature = "server_encrypt")] +use crate::cipher::RsaCipher; +use crate::core::Config; +use crate::external_route::{AllowExternalRoute, ExternalRoute}; +use crate::handle::recv_data::RecvDataHandler; +use crate::handle::{ + maintain, tun_tap, BaseConfigInfo, ConnectStatus, CurrentDeviceInfo, PeerDeviceInfo, +}; +use crate::nat::NatTest; +use crate::util::{Scheduler, StopManager, U64Adder, WatchU64Adder}; +use crate::{nat, tun_tap_device, DeviceInfo, VntCallback}; + +#[derive(Clone)] +pub struct Vnt { + stop_manager: StopManager, + config: Config, + current_device: Arc>, + nat_test: NatTest, + device_list: Arc)>>, + context: Context, + peer_nat_info_map: Arc>>, + down_count_watcher: WatchU64Adder, + up_count_watcher: Arc, +} + +impl Vnt { + pub fn new(config: Config, callback: Call) -> io::Result { + //服务端非对称加密 + #[cfg(feature = "server_encrypt")] + let rsa_cipher: Arc>> = Arc::new(Mutex::new(None)); + //服务端对称加密 + let server_cipher: Cipher = if config.server_encrypt { + let mut key = [0u8; 32]; + rand::thread_rng().fill(&mut key); + Cipher::new_key(key, config.token.clone())? + } else { + Cipher::None + }; + let finger = if config.finger { + Some(config.token.clone()) + } else { + None + }; + //客户端对称加密 + let client_cipher = + Cipher::new_password(config.cipher_model, config.password.clone(), finger); + //当前设备信息 + let current_device = Arc::new(AtomicCell::new(CurrentDeviceInfo::new0( + config.server_address, + ))); + //设备列表 + let device_list: Arc)>> = + Arc::new(Mutex::new((0, Vec::with_capacity(16)))); + //基础信息 + let config_info = BaseConfigInfo::new( + config.name.clone(), + config.token.clone(), + config.ip, + config.password.is_some(), + config.device_id.clone(), + config.server_address_str.clone(), + ); + let ports = config.ports.as_ref().map_or(vec![0, 0], |v| { + if v.is_empty() { + vec![0, 0] + } else { + v.clone() + } + }); + //通道上下文 + let (context, tcp_listener) = init_context(ports, config.first_latency, config.tcp)?; + let local_ipv4 = nat::local_ipv4(); + let local_ipv6 = nat::local_ipv6(); + let udp_ports = context.main_local_udp_port()?; + let tcp_port = tcp_listener.local_addr()?.port(); + //nat检测工具 + let nat_test = NatTest::new( + config.stun_server.clone(), + local_ipv4, + local_ipv6, + udp_ports, + tcp_port, + ); + + // 虚拟网卡 + let device = tun_tap_device::create_device(&config)?; + let tun_info = DeviceInfo::new(device.name()?, device.version()?); + callback.create_tun(tun_info); + // 服务停止管理器 + let stop_manager = { + let callback = callback.clone(); + StopManager::new(move || callback.stop()) + }; + // 定时器 + let scheduler = Scheduler::new(stop_manager.clone())?; + let external_route = ExternalRoute::new(config.in_ips.clone()); + let out_external_route = AllowExternalRoute::new(config.out_ips.clone()); + + #[cfg(feature = "ip_proxy")] + let proxy_map = if !config.out_ips.is_empty() && !config.no_proxy { + Some(crate::ip_proxy::init_proxy( + context.clone(), + scheduler.clone(), + stop_manager.clone(), + current_device.clone(), + client_cipher.clone(), + )?) + } else { + None + }; + let (punch_sender, punch_receiver) = sync_channel(3); + let peer_nat_info_map: Arc>> = + Arc::new(RwLock::new(HashMap::with_capacity(16))); + let down_counter = U64Adder::with_capacity(8); + let down_count_watcher = down_counter.watch(); + let handler = RecvDataHandler::new( + #[cfg(feature = "server_encrypt")] + rsa_cipher, + server_cipher.clone(), + client_cipher.clone(), + current_device.clone(), + device.clone(), + device_list.clone(), + config_info.clone(), + nat_test.clone(), + callback.clone(), + config.relay, + punch_sender, + peer_nat_info_map.clone(), + external_route.clone(), + out_external_route, + #[cfg(feature = "ip_proxy")] + proxy_map.clone(), + down_counter, + ); + + //初始化网络数据通道 + let (udp_socket_sender, tcp_socket_sender) = + init_channel(tcp_listener, context.clone(), stop_manager.clone(), handler)?; + // 打洞逻辑 + let punch = Punch::new( + context.clone(), + config.punch_model, + config.tcp, + tcp_socket_sender.clone(), + ); + let up_counter = Arc::new(AtomicU64::new(0)); + let up_count_watcher = up_counter.clone(); + tun_tap::tun_handler::start( + stop_manager.clone(), + context.clone(), + device.clone(), + current_device.clone(), + external_route, + #[cfg(feature = "ip_proxy")] + proxy_map, + client_cipher.clone(), + server_cipher.clone(), + config.parallel, + up_counter, + )?; + maintain::idle_gateway( + &scheduler, + context.clone(), + current_device.clone(), + config_info.clone(), + tcp_socket_sender.clone(), + callback.clone(), + 0, + ); + { + let context = context.clone(); + let nat_test = nat_test.clone(); + let device_list = device_list.clone(); + let current_device = current_device.clone(); + let relay = config.relay; + if !relay { + // 定时nat探测 + maintain::retrieve_nat_type( + &scheduler, + context.clone(), + nat_test.clone(), + udp_socket_sender, + ); + } + //延迟启动 + scheduler.timeout(Duration::from_secs(3), move |scheduler| { + start( + scheduler, + context, + nat_test, + device_list, + current_device, + client_cipher, + server_cipher, + punch_receiver, + config_info, + punch, + callback, + relay, + ); + }); + } + + Ok(Self { + stop_manager, + config, + current_device, + nat_test, + device_list, + context, + peer_nat_info_map, + down_count_watcher, + up_count_watcher, + }) + } +} + +pub fn start( + scheduler: &Scheduler, + context: Context, + nat_test: NatTest, + device_list: Arc)>>, + current_device: Arc>, + client_cipher: Cipher, + server_cipher: Cipher, + punch_receiver: Receiver<(Ipv4Addr, NatInfo)>, + config_info: BaseConfigInfo, + punch: Punch, + callback: Call, + relay: bool, +) { + // 定时心跳 + maintain::heartbeat( + &scheduler, + context.clone(), + current_device.clone(), + device_list.clone(), + client_cipher.clone(), + server_cipher.clone(), + ); + // 路由空闲检测逻辑 + let idle = Idle::new(Duration::from_secs(10), context.clone()); + // 定时空闲检查 + maintain::idle_route( + &scheduler, + idle, + context.clone(), + current_device.clone(), + callback, + ); + // 定时客户端中继检测 + maintain::client_relay( + &scheduler, + context.clone(), + current_device.clone(), + device_list.clone(), + client_cipher.clone(), + ); + // 定时地址探测 + maintain::addr_request( + &scheduler, + context.clone(), + current_device.clone(), + server_cipher.clone(), + config_info.clone(), + ); + if !relay { + // 定时打洞 + maintain::punch( + &scheduler, + context.clone(), + nat_test.clone(), + device_list.clone(), + current_device.clone(), + client_cipher.clone(), + punch_receiver, + punch, + ); + } +} + +impl Vnt { + pub fn name(&self) -> &str { + &self.config.name + } + pub fn server_encrypt(&self) -> bool { + self.config.server_encrypt + } + pub fn client_encrypt(&self) -> bool { + self.config.password.is_some() + } + pub fn current_device(&self) -> CurrentDeviceInfo { + self.current_device.load() + } + pub fn peer_nat_info(&self, ip: &Ipv4Addr) -> Option { + self.peer_nat_info_map.read().get(ip).cloned() + } + pub fn connection_status(&self) -> ConnectStatus { + self.current_device.load().status + } + pub fn nat_info(&self) -> NatInfo { + self.nat_test.nat_info() + } + pub fn device_list(&self) -> Vec { + let device_list_lock = self.device_list.lock(); + let (_epoch, device_list) = device_list_lock.clone(); + drop(device_list_lock); + device_list + } + pub fn route(&self, ip: &Ipv4Addr) -> Option { + self.context.route_table.route_one(ip) + } + pub fn is_gateway(&self, ip: &Ipv4Addr) -> bool { + self.current_device.load().is_gateway(ip) + } + pub fn route_key(&self, route_key: &RouteKey) -> Option { + self.context.route_table.route_to_id(route_key) + } + pub fn route_table(&self) -> Vec<(Ipv4Addr, Vec)> { + self.context.route_table.route_table() + } + pub fn up_stream(&self) -> u64 { + self.up_count_watcher.load(Ordering::Relaxed) + } + pub fn down_stream(&self) -> u64 { + self.down_count_watcher.get() + } + pub fn stop(&self) { + self.stop_manager.stop() + } + pub fn wait(&self) { + self.stop_manager.wait() + } +} diff --git a/vnt/src/core/mod.rs b/vnt/src/core/mod.rs index 644dd43..4f7b815 100644 --- a/vnt/src/core/mod.rs +++ b/vnt/src/core/mod.rs @@ -1,569 +1,16 @@ -use std::collections::HashMap; use std::io; -use std::net::UdpSocket; -use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4}; -use std::net::{TcpListener, TcpStream}; -use std::sync::Arc; -use std::time::Duration; +use std::net::{Ipv4Addr, SocketAddr}; -use crossbeam_utils::atomic::AtomicCell; -use parking_lot::{Mutex, RwLock}; -use rand::Rng; -use tokio::sync::mpsc::channel; +pub use conn::Vnt; -use crate::channel::channel::{Channel, Context}; -use crate::channel::idle::Idle; -use crate::channel::punch::{NatInfo, Punch, PunchModel}; -use crate::channel::sender::ChannelSender; -use crate::channel::{Route, RouteKey}; -use crate::cipher::{Cipher, CipherModel, RsaCipher}; -use crate::core::status::VntStatusManger; -use crate::error::Error; -use crate::external_route::{AllowExternalRoute, ExternalRoute}; -use crate::handle::handshake_handler::HandshakeEnum; -use crate::handle::recv_handler::ChannelDataHandler; -use crate::handle::registration_handler::{RegResponse, ReqEnum}; -#[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))] -use crate::handle::tun_tap::tap_handler; -use crate::handle::tun_tap::tun_handler; -use crate::handle::{ - handshake_handler, heartbeat_handler, punch_handler, registration_handler, ConnectStatus, - CurrentDeviceInfo, PeerDeviceInfo, -}; -use crate::igmp_server::IgmpServer; -use crate::nat::NatTest; -use crate::tun_tap_device; -use crate::tun_tap_device::{DeviceReader, DeviceWriter}; +use crate::channel::punch::PunchModel; +use crate::cipher::CipherModel; -pub mod status; -pub mod sync; - -#[derive(Clone)] -pub struct Vnt { - config: Config, - current_device: Arc>, - context: Context, - vnt_status_manager: VntStatusManger, - device_writer: DeviceWriter, - /// 0. 机器纪元，每一次上线或者下线都会增1，用于感知网络中机器变化 - /// 服务端和客户端的不一致，则服务端会推送新的设备列表 - /// 1. 网络中的虚拟ip列表 - device_list: Arc)>>, - nat_test: NatTest, - connect_status: Arc>, - peer_nat_info_map: Arc>>, -} - -pub struct VntUtil { - config: Config, - main_channel: UdpSocket, - main_tcp_channel: Option, - response: Option, - iface: Option<(DeviceWriter, DeviceReader)>, - server_cipher: Cipher, - rsa_cipher: Option, -} - -impl VntUtil { - pub fn new(config: Config) -> io::Result { - let address: SocketAddr = format!("[::]:{}", config.port).parse().unwrap(); - //单个udp用同步的性能更好，但是代理和多端口监听用异步更方便，这里将两者结合起来 - let socket = socket2::Socket::new(socket2::Domain::IPV6, socket2::Type::DGRAM, None)?; - socket.set_only_v6(false)?; - socket.bind(&address.into())?; - let main_channel: UdpSocket = socket.into(); - main_channel.set_write_timeout(Some(Duration::from_secs(5)))?; - main_channel.set_read_timeout(Some(Duration::from_secs(2)))?; - let server_cipher = if config.server_encrypt { - let mut key = [0u8; 32]; - rand::thread_rng().fill(&mut key); - Cipher::new_key(key, config.token.clone())? - } else { - Cipher::None - }; - Ok(VntUtil { - config, - main_channel, - main_tcp_channel: None, - response: None, - iface: None, - server_cipher, - rsa_cipher: None, - }) - } - ///链接 - pub fn connect(&mut self) -> io::Result<()> { - if self.config.tcp { - let tcp = TcpStream::connect(self.config.server_address)?; - tcp.set_read_timeout(Some(Duration::from_secs(10)))?; - let _ = self.main_tcp_channel.insert(tcp); - } - Ok(()) - } - - ///握手 用于获取公钥 - pub fn handshake(&mut self) -> Result, HandshakeEnum> { - let rsa_cipher = handshake_handler::handshake( - &self.main_channel, - self.main_tcp_channel.as_mut(), - self.config.server_address, - self.config.server_encrypt, - )?; - self.rsa_cipher = rsa_cipher.clone(); - Ok(rsa_cipher) - } - /// 加密握手 用于同步密钥 - pub fn secret_handshake(&mut self) -> Result<(), HandshakeEnum> { - handshake_handler::secret_handshake( - &self.main_channel, - self.main_tcp_channel.as_mut(), - self.config.server_address, - self.rsa_cipher.as_ref().unwrap(), - &self.server_cipher, - self.config.token.clone(), - ) - } - /// 注册 - pub fn register(&mut self) -> Result { - match registration_handler::registration( - &self.main_channel, - self.main_tcp_channel.as_mut(), - &self.server_cipher, - self.config.server_address, - self.config.token.clone(), - self.config.device_id.clone(), - self.config.name.clone(), - self.config.ip.unwrap_or(Ipv4Addr::UNSPECIFIED), - self.config.password.is_some(), - ) { - Ok(res) => { - let _ = self.response.insert(res.clone()); - Ok(res) - } - Err(e) => Err(e), - } - } - #[cfg(any(target_os = "android"))] - pub fn create_iface(&mut self, vpn_fd: i32) { - let (device_writer, device_reader) = tun_tap_device::create(vpn_fd); - let _ = self.iface.insert((device_writer, device_reader)); - } - #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))] - pub fn create_iface(&mut self) -> io::Result { - if self.iface.is_some() { - return Err(io::Error::from(io::ErrorKind::AlreadyExists)); - } - let response = match &self.response { - None => { - return Err(io::Error::from(io::ErrorKind::AlreadyExists)); - } - Some(res) => res, - }; - let device_type = if self.config.tap { - { - //删除tun网卡避免ip冲突，因为非正常退出会保留网卡 - tun_tap_device::delete_device(tun_tap_device::DeviceType::Tun); - } - tun_tap_device::DeviceType::Tap - } else { - { - //删除tap网卡避免ip冲突，非正常退出会保留网卡 - tun_tap_device::delete_device(tun_tap_device::DeviceType::Tap); - } - tun_tap_device::DeviceType::Tun - }; - let mtu = match self.config.mtu { - None => { - if self.config.password.is_none() { - 1450 - } else { - 1410 - } - } - Some(mtu) => mtu, - }; - let in_ips = self - .config - .in_ips - .iter() - .map(|(dest, mask, _)| (Ipv4Addr::from(*dest & *mask), Ipv4Addr::from(*mask))) - .collect::>(); - - let (device_writer, device_reader, driver_info) = tun_tap_device::create_device( - device_type, - response.virtual_ip, - response.virtual_netmask, - response.virtual_gateway, - in_ips, - mtu, - )?; - let _ = self.iface.insert((device_writer, device_reader)); - 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())); - } - Some(res) => res, - }; - let (device_writer, device_reader) = match self.iface { - None => { - return Err(Error::Stop("iface None".to_string())); - } - Some(res) => res, - }; - let config = self.config.clone(); - let vnt_status_manager = VntStatusManger::new(); - let finger = if config.finger { - Some(config.token.clone()) - } else { - None - }; - let client_cipher = - Cipher::new_password(config.cipher_model, config.password.clone(), finger); - let virtual_ip = response.virtual_ip; - let virtual_gateway = response.virtual_gateway; - let virtual_netmask = response.virtual_netmask; - let current_device = Arc::new(AtomicCell::new(CurrentDeviceInfo::new( - virtual_ip, - virtual_gateway, - virtual_netmask, - config.server_address, - ))); - - let (cone_sender, cone_receiver) = channel(3); - let (symmetric_sender, symmetric_receiver) = channel(2); - let (tcp_sender, tcp_receiver) = if let Some(main_tcp_channel) = self.main_tcp_channel { - (Some(main_tcp_channel.try_clone()?), Some(main_tcp_channel)) - } else { - (None, None) - }; - let tcp_listener = TcpListener::bind(format!("[::]:{}", config.port))?; - let local_tcp_port = tcp_listener.local_addr()?.port(); - let context = Context::new( - self.main_channel, - tcp_sender, - current_device.clone(), - 1, - config.first_latency, - local_tcp_port, - ); - let idle = Idle::new(Duration::from_secs(16), context.clone()); - let channel_sender = ChannelSender::new(context.clone()); - - let register = Arc::new(registration_handler::Register::new( - self.server_cipher.clone(), - channel_sender.clone(), - config.server_address, - config.token.clone(), - config.device_id.clone(), - config.name.clone(), - config.password.is_some(), - )); - let device_list: Arc)>> = - Arc::new(Mutex::new((response.epoch, response.device_info_list))); - let peer_nat_info_map: Arc>> = - Arc::new(RwLock::new(HashMap::with_capacity(16))); - let connect_status = Arc::new(AtomicCell::new(ConnectStatus::Connected)); - let public_ip = response.public_ip; - let public_port = response.public_port; - let local_udp_port = context.main_local_udp_port().unwrap_or(0); - let local_ipv4 = crate::nat::local_ipv4(); - let ipv6 = crate::nat::local_ipv6(); - - // NAT检测 - let nat_test = NatTest::new( - config.stun_server.clone(), - public_ip, - public_port, - local_ipv4, - ipv6, - local_udp_port, - local_tcp_port, - ); - let in_external_route = if config.in_ips.is_empty() { - None - } else { - Some(ExternalRoute::new(config.in_ips)) - }; - #[cfg(feature = "ip_proxy")] - let (tcp_proxy, udp_proxy, ip_proxy_map) = if config.out_ips.is_empty() || config.no_proxy { - (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?; - (Some(tcp_proxy), Some(udp_proxy), Some(ip_proxy_map)) - }; - let out_external_route = AllowExternalRoute::new(config.out_ips); - - let igmp_server = if config.simulate_multicast { - Some(IgmpServer::new(device_writer.clone())) - } else { - None - }; - #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))] - if config.tap { - tap_handler::start( - vnt_status_manager.worker("tap_handler"), - channel_sender.clone(), - device_reader, - device_writer.clone(), - igmp_server.clone(), - current_device.clone(), - in_external_route, - #[cfg(feature = "ip_proxy")] - ip_proxy_map.clone(), - client_cipher.clone(), - self.server_cipher.clone(), - config.parallel, - ); - } else { - tun_handler::start( - vnt_status_manager.worker("tun_handler"), - channel_sender.clone(), - device_reader, - device_writer.clone(), - igmp_server.clone(), - current_device.clone(), - in_external_route, - #[cfg(feature = "ip_proxy")] - ip_proxy_map.clone(), - client_cipher.clone(), - self.server_cipher.clone(), - config.parallel, - ); - } - #[cfg(any(target_os = "android"))] - tun_handler::start( - vnt_status_manager.worker("android tun_handler"), - channel_sender.clone(), - device_reader, - device_writer.clone(), - igmp_server.clone(), - current_device.clone(), - in_external_route, - #[cfg(feature = "ip_proxy")] - ip_proxy_map.clone(), - client_cipher.clone(), - self.server_cipher.clone(), - config.parallel, - ); - - //外部数据接收处理 - let channel_recv_handler = ChannelDataHandler::new( - current_device.clone(), - device_list.clone(), - register.clone(), - nat_test.clone(), - igmp_server, - device_writer.clone(), - connect_status.clone(), - peer_nat_info_map.clone(), - #[cfg(feature = "ip_proxy")] - ip_proxy_map, - out_external_route, - cone_sender, - symmetric_sender, - client_cipher.clone(), - self.server_cipher.clone(), - self.rsa_cipher.clone(), - config.relay, - config.token.clone(), - 14, - ); - let punch = Punch::new( - context.clone(), - config.punch_model, - config.tcp, - channel_recv_handler.clone(), - ); - { - let channel = Channel::new(context.clone(), channel_recv_handler, tcp_listener); - let channel_worker = vnt_status_manager.worker("channel_worker"); - let relay = config.relay; - tokio::spawn( - async move { channel.start(channel_worker, tcp_receiver, 65, relay).await }, - ); - } - { - let nat_test = nat_test.clone(); - let device_list = device_list.clone(); - let current_device = current_device.clone(); - // 定时心跳 - heartbeat_handler::start_heartbeat_main( - vnt_status_manager.worker("main-heartbeat"), - channel_sender.clone(), - device_list.clone(), - current_device.clone(), - config.server_address_str, - client_cipher.clone(), - self.server_cipher.clone(), - ); - heartbeat_handler::start_heartbeat( - vnt_status_manager.worker("heartbeat"), - channel_sender.clone(), - device_list.clone(), - current_device.clone(), - client_cipher.clone(), - self.server_cipher.clone(), - ); - // 空闲检查 - heartbeat_handler::start_idle( - vnt_status_manager.worker("idle"), - idle, - channel_sender.clone(), - ); - if !config.relay { - // 打洞处理 - punch_handler::start( - vnt_status_manager.worker("cone_receiver"), - cone_receiver, - punch.clone(), - current_device.clone(), - client_cipher.clone(), - ); - punch_handler::start( - vnt_status_manager.worker("symmetric_receiver"), - symmetric_receiver, - punch, - current_device.clone(), - client_cipher.clone(), - ); - tokio::spawn(punch_handler::start_punch( - vnt_status_manager.worker("punch_handler"), - nat_test, - device_list, - channel_sender, - current_device, - client_cipher.clone(), - )); - } - } - #[cfg(feature = "ip_proxy")] - { - //代理 - if let Some(tcp_proxy) = tcp_proxy { - tokio::spawn(tcp_proxy.start()); - } - if let Some(udp_proxy) = udp_proxy { - tokio::spawn(udp_proxy.start()); - } - } - { - let context = context.clone(); - let nat_test = nat_test.clone(); - tokio::spawn(async move { - let info = nat_test - .re_test( - public_ip, - public_port, - local_ipv4, - ipv6, - local_udp_port, - local_tcp_port, - ) - .await; - context.switch(info.nat_type); - }); - } - Ok(Vnt { - config: self.config, - current_device, - context, - vnt_status_manager, - device_writer, - nat_test, - device_list, - connect_status, - peer_nat_info_map, - }) - } -} - -impl Vnt { - pub fn name(&self) -> &str { - &self.config.name - } - pub fn server_encrypt(&self) -> bool { - self.config.server_encrypt - } - pub fn client_encrypt(&self) -> bool { - self.config.password.is_some() - } - pub fn current_device(&self) -> CurrentDeviceInfo { - self.current_device.load() - } - pub fn peer_nat_info(&self, ip: &Ipv4Addr) -> Option { - self.peer_nat_info_map.read().get(ip).cloned() - } - pub fn connection_status(&self) -> ConnectStatus { - self.connect_status.load() - } - pub fn nat_info(&self) -> NatInfo { - self.nat_test.nat_info() - } - pub fn device_list(&self) -> Vec { - let device_list_lock = self.device_list.lock(); - let (_epoch, device_list) = device_list_lock.clone(); - drop(device_list_lock); - device_list - } - pub fn route(&self, ip: &Ipv4Addr) -> Option { - self.context.route_one(ip) - } - pub fn route_key(&self, route_key: &RouteKey) -> Option { - self.context.route_to_id(route_key) - } - pub fn route_table(&self) -> Vec<(Ipv4Addr, Route)> { - self.context.route_table_one() - } - pub fn stop(&self) -> io::Result<()> { - let _ = self.context.close(); - self.vnt_status_manager.stop_all(); - let _ = self.device_writer.close(); - let virtual_gateway = self.current_device.load().virtual_gateway; - 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) { - self.vnt_status_manager.wait().await; - let _ = self.stop(); - } - pub async fn wait_stop_ms(&mut self, ms: Duration) -> bool { - tokio::select! { - _=self.vnt_status_manager.wait()=>{ - let _ = self.stop(); - return true; - } - _=tokio::time::sleep(ms)=>{ - return false; - } - } - } -} - -impl Drop for Vnt { - fn drop(&mut self) { - let _ = self.stop(); - } -} +mod conn; #[derive(Clone, Debug)] pub struct Config { + #[cfg(any(target_os = "windows", target_os = "linux"))] pub tap: bool, pub token: String, pub device_id: String, @@ -574,8 +21,7 @@ pub struct Config { pub in_ips: Vec<(u32, u32, Ipv4Addr)>, pub out_ips: Vec<(u32, u32)>, pub password: Option, - pub simulate_multicast: bool, - pub mtu: Option, + pub mtu: Option, pub tcp: bool, pub ip: Option, pub relay: bool, @@ -586,13 +32,17 @@ pub struct Config { pub cipher_model: CipherModel, pub finger: bool, pub punch_model: PunchModel, - pub port: u16, + pub ports: Option>, pub first_latency: bool, + #[cfg(not(target_os = "android"))] + pub device_name: Option, + #[cfg(target_os = "android")] + pub device_fd: i32, } impl Config { pub fn new( - tap: bool, + #[cfg(any(target_os = "windows", target_os = "linux"))] tap: bool, token: String, device_id: String, name: String, @@ -602,8 +52,7 @@ impl Config { in_ips: Vec<(u32, u32, Ipv4Addr)>, out_ips: Vec<(u32, u32)>, password: Option, - simulate_multicast: bool, - mtu: Option, + mtu: Option, tcp: bool, ip: Option, relay: bool, @@ -613,24 +62,27 @@ impl Config { cipher_model: CipherModel, finger: bool, punch_model: PunchModel, - port: u16, + ports: Option>, first_latency: bool, - ) -> Result { + #[cfg(not(target_os = "android"))] device_name: Option, + #[cfg(target_os = "android")] device_fd: i32, + ) -> io::Result { for x in stun_server.iter_mut() { if !x.contains(":") { x.push_str(":3478"); } } if token.is_empty() || token.len() > 128 { - return Err(Error::Stop(String::from("token too long"))); + return Err(io::Error::new(io::ErrorKind::Other, "token too long")); } if device_id.is_empty() || device_id.len() > 128 { - return Err(Error::Stop(String::from("device_id too long"))); + return Err(io::Error::new(io::ErrorKind::Other, "device_id too long")); } if name.is_empty() || name.len() > 128 { - return Err(Error::Stop(String::from("name too long"))); + return Err(io::Error::new(io::ErrorKind::Other, "name too long")); } Ok(Self { + #[cfg(any(target_os = "windows", target_os = "linux"))] tap, token, device_id, @@ -641,7 +93,6 @@ impl Config { in_ips, out_ips, password, - simulate_multicast, mtu, tcp, ip, @@ -653,8 +104,12 @@ impl Config { cipher_model, finger, punch_model, - port, + ports, first_latency, + #[cfg(not(target_os = "android"))] + device_name, + #[cfg(target_os = "android")] + device_fd, }) } } diff --git a/vnt/src/core/status.rs b/vnt/src/core/status.rs deleted file mode 100644 index b954d66..0000000 --- a/vnt/src/core/status.rs +++ /dev/null @@ -1,92 +0,0 @@ -use crate::util::wait::WaitGroup; -use std::sync::Arc; -use tokio::sync::watch; -use tokio::sync::watch::{Receiver, Sender}; - -#[derive(Copy, Clone, Eq, PartialEq)] -pub enum VntStatus { - Starting, - Stopping, -} - -pub struct VntWorker { - name: String, - wg: WaitGroup, - status_s: Arc>, - status_r: Receiver, -} - -impl VntWorker { - pub fn worker(&self, name: &str) -> Self { - self.wg.add(); - VntWorker { - name: name.to_string(), - wg: self.wg.clone(), - status_s: self.status_s.clone(), - status_r: self.status_r.clone(), - } - } -} - -impl Drop for VntWorker { - fn drop(&mut self) { - log::info!("任务停止:{}", self.name); - self.wg.done(); - } -} - -impl VntWorker { - pub fn stop_all(&self) { - let _ = self.status_s.send(VntStatus::Stopping); - } - pub async fn stop_wait(&mut self) { - loop { - if *self.status_r.borrow() == VntStatus::Stopping { - return; - } - match self.status_r.changed().await { - Ok(_) => { - if *self.status_r.borrow() == VntStatus::Stopping { - return; - } - } - Err(_) => { - return; - } - } - } - } -} - -#[derive(Clone)] -pub struct VntStatusManger { - wg: WaitGroup, - status_s: Arc>, - status_r: Receiver, -} - -impl VntStatusManger { - pub fn new() -> Self { - let (status_s, status_r) = watch::channel(VntStatus::Starting); - Self { - wg: WaitGroup::new(), - status_s: Arc::new(status_s), - status_r, - } - } - pub fn stop_all(&self) { - let _ = self.status_s.send(VntStatus::Stopping); - } - pub async fn wait(&mut self) { - self.wg.wait().await - } - pub fn worker(&self, name: &str) -> VntWorker { - self.wg.add(); - VntWorker { - name: name.to_string(), - wg: self.wg.clone(), - status_s: self.status_s.clone(), - status_r: self.status_r.clone(), - } - } -} diff --git a/vnt/src/core/sync.rs b/vnt/src/core/sync.rs deleted file mode 100644 index 386792d..0000000 --- a/vnt/src/core/sync.rs +++ /dev/null @@ -1,81 +0,0 @@ -use crate::cipher::RsaCipher; -use crate::core::{Config, Vnt, VntUtil}; -use crate::handle::handshake_handler::HandshakeEnum; -use crate::handle::registration_handler::{RegResponse, ReqEnum}; -use std::io; -use std::ops::Deref; -use std::time::Duration; -use tokio::runtime::Runtime; - -pub struct VntUtilSync { - vnt_util: VntUtil, - runtime: Runtime, -} - -pub struct VntSync { - vnt: Vnt, - runtime: Runtime, -} - -impl VntUtilSync { - pub fn new(config: Config) -> io::Result { - let runtime = tokio::runtime::Builder::new_multi_thread() - .enable_all() - .build()?; - let vnt_util = VntUtil::new(config)?; - Ok(VntUtilSync { vnt_util, runtime }) - } - pub fn connect(&mut self) -> io::Result<()> { - self.vnt_util.connect() - } - pub fn handshake(&mut self) -> Result, HandshakeEnum> { - self.vnt_util.handshake() - } - pub fn secret_handshake(&mut self) -> Result<(), HandshakeEnum> { - self.vnt_util.secret_handshake() - } - pub fn register(&mut self) -> Result { - self.vnt_util.register() - } - #[cfg(any(target_os = "android"))] - pub fn create_iface(&mut self, vpn_fd: i32) { - self.vnt_util.create_iface(vpn_fd) - } - #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))] - pub fn create_iface(&mut self) -> io::Result { - self.vnt_util.create_iface() - } - pub fn build(self) -> crate::Result { - let runtime = self.runtime; - let vnt = runtime.block_on(self.vnt_util.build())?; - { - let mut vnt = vnt.clone(); - std::thread::spawn(move || runtime.block_on(vnt.wait_stop())); - } - Ok(VntSync { - vnt, - runtime: tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .unwrap(), - }) - } -} - -impl VntSync { - pub fn wait_stop(&mut self) { - self.runtime.block_on(self.vnt.wait_stop()) - } - pub fn wait_stop_ms(&mut self, ms: u64) -> bool { - self.runtime - .block_on(self.vnt.wait_stop_ms(Duration::from_millis(ms))) - } -} - -impl Deref for VntSync { - type Target = Vnt; - - fn deref(&self) -> &Self::Target { - &self.vnt - } -}