diff --git a/Cargo.toml b/Cargo.toml index a98859f..2e0149c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,46 +1,2 @@ -[package] -name = "switch" -version = "0.1.0" -edition = "2021" - -# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html - -[dependencies] -packet = { path = "./packet" } -bytes = "1.3.0" -log = "0.4.17" -log4rs = "1.2.0" -dirs = "4.0.0" -libc = "0.2.137" - -dashmap = "5.4.0" -crossbeam = "0.8.2" -parking_lot = "0.12.1" - -rsa = "0.7.2" -rand = "0.8.5" -sha2 = { version = "0.10.6", features = ["oid"] } -#colored = "2.0.0" - -thiserror = "1.0.37" -chrono = "0.4.23" -lazy_static = "1.4.0" -moka = "0.9.6" -protobuf = "3.2.0" - -console = "0.15.2" -mac_address = "1.1.4" -clap = { version = "4.0.32", features = ["derive"] } -[target.'cfg(any(unix))'.dependencies] -tun = { path = "./rust-tun" } -sudo = "0.6.0" - -[target.'cfg(target_os = "windows")'.dependencies] -winapi = { version = "0.3.9", features = ["handleapi", "processthreadsapi", "winnt", "securitybaseapi", "impl-default"] } -wintun = "0.2.1" -libloading = "0.7.4" -runas = "0.2.1" - -[build-dependencies] -protobuf-codegen = "3.2.0" -protoc-bin-vendored = "3.0.0" \ No newline at end of file +[workspace] +members = ["switch","switch-desktop","switch-jni"] \ No newline at end of file diff --git a/README.md b/README.md index 9855303..dc049ed 100644 --- a/README.md +++ b/README.md @@ -46,5 +46,6 @@ Virtual Network Tools - 服务端中继转发 ### Todo +- 支持安卓 - 数据加密 - 客户端中继转发 diff --git a/src/main.rs b/src/main.rs deleted file mode 100644 index 3444ae3..0000000 --- a/src/main.rs +++ /dev/null @@ -1,278 +0,0 @@ -use std::{io, thread}; -use std::net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4, UdpSocket}; -use std::sync::atomic::Ordering; - -use clap::Parser; -use console::style; - -use crate::handle::{CurrentDeviceInfo, DEVICE_LIST, DIRECT_ROUTE_TABLE, NAT_INFO, NatInfo, SERVER_RT}; -use crate::handle::registration_handler::registration; -use crate::tun_device::create_tun; - -pub mod tun_device; -pub mod nat; -pub mod error; -pub mod handle; -pub mod proto; -pub mod protocol; -#[cfg(windows)] -pub mod admin_check; - -#[derive(Parser, Debug)] -#[command(author = "Lu Beilin", version, about = "一个虚拟网络工具,启动后会获取一个ip,相同token下的设备之间可以用ip直接通信")] -struct Args { - /// 32位字符 - /// 相同token的设备之间才能通信。 - /// 建议使用uuid保证唯一性。 - /// 32-bit characters. - /// Only devices with the same token can communicate with each other. - /// It is recommended to use uuid to ensure uniqueness - #[arg(short, long)] - token: String, -} - -fn log_init() { - let home = dirs::home_dir().unwrap().join(".switch"); - if !home.exists() { - std::fs::create_dir(&home).expect(" Failed to create '.switch' directory"); - } - let logfile = log4rs::append::file::FileAppender::builder() - // Pattern: https://docs.rs/log4rs/*/log4rs/encode/pattern/index.html - .encoder(Box::new(log4rs::encode::pattern::PatternEncoder::new("{d(%+)(utc)} [{f}:{L}] {h({l})} {M}:{m}{n}\n"))) - .build(home.join("switch.log")) - .unwrap(); - let config = log4rs::Config::builder() - .appender(log4rs::config::Appender::builder().build("logfile", Box::new(logfile))) - .build( - log4rs::config::Root::builder() - .appender("logfile") - .build(log::LevelFilter::Info), - ) - .unwrap(); - let _ = log4rs::init_config(config); -} - -fn main() { - log_init(); - let args = Args::parse(); - #[cfg(windows)] - if !admin_check::is_app_elevated() { - let args: Vec<_> = std::env::args().collect(); - println!("{}", style("正在启动管理员权限执行...").red()); - if let Some(absolute_path) = std::env::current_exe() - .ok() - .and_then(|p| p.to_str().map(|p| p.to_string())) - { - let _ = runas::Command::new(&absolute_path).args(&args[1..]).status() - .expect("failed to execute"); - } else { - panic!("failed to execute") - } - return; - } - - #[cfg(any(unix))] - if sudo::RunningAs::Root != sudo::check() { - println!("{}", style("需要使用root权限执行...").red()); - sudo::escalate_if_needed().unwrap(); - } - - println!("{}", style("启动服务...").green()); - - let token = args.token; - // let d = Local::now().timestamp().to_string(); - let mac_address = mac_address::get_mac_address().unwrap().unwrap().to_string(); - let server_address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(43, 139, 56, 10)), 29876); - // let server_address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127,0,0,1)), 29876); - let mut port = 101 as u16; - let udp = loop { - match UdpSocket::bind(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::from(0), port))) { - Ok(udp) => { - break udp; - } - Err(e) => { - if e.kind() == io::ErrorKind::AddrInUse { - port += 1; - } else { - log::error!("创建udp失败 {:?}",e); - println!("创建udp失败:{:?}", e); - panic!() - } - } - } - }; - //注册 - let response = registration(&udp, server_address, token, mac_address).unwrap(); - { - let ip_list = response - .virtual_ip_list - .iter() - .map(|ip| Ipv4Addr::from(*ip)) - .collect(); - let mut dev = DEVICE_LIST.lock(); - dev.0 = response.epoch; - dev.1 = ip_list; - } - let virtual_ip = Ipv4Addr::from(response.virtual_ip); - let virtual_gateway = Ipv4Addr::from(response.virtual_gateway); - let virtual_netmask = Ipv4Addr::from(response.virtual_netmask); - println!("virtual_gateway:{:?}", virtual_gateway); - println!("virtual_netmask:{:?}", virtual_netmask); - println!("当前设备ip(virtual_ip):{}", style(virtual_ip).green()); - //心跳线程 - { - let udp = udp.try_clone().unwrap(); - let _ = thread::spawn(move || { - if let Err(e) = handle::heartbeat_handler::handle_loop(udp, server_address) { - log::error!("心跳线程停止 {:?}",e); - println!("心跳线程停止:{:?}", e); - } - std::process::exit(1); - }); - } - //初始化nat数据 - handle::init_nat_info(response.public_ip, response.public_port as u16); - // tun服务 - let (tun_writer, tun_reader) = - create_tun(virtual_ip, virtual_netmask, virtual_gateway).unwrap(); - // 打洞数据通道 - let (punch_sender, cone_receiver, req_symmetric_receiver, res_symmetric_receiver) = handle::punch_handler::bounded(); - //udp数据处理 - { - // 低优先级的udp数据通道 - let (sender, receiver) = crossbeam::channel::bounded(100); - let udp1 = udp.try_clone().unwrap(); - let _ = thread::spawn(move || { - let current_device = CurrentDeviceInfo::new(virtual_ip, virtual_gateway, virtual_netmask, server_address); - if let Err(e) = handle::udp_recv_handler::recv_loop( - udp1, - server_address, - sender, - tun_writer, - current_device, - ) { - log::error!("udp数据处理线程停止 {:?}",e); - println!("udp数据处理线程停止:{:?}", e); - } - std::process::exit(1); - }); - let udp1 = udp.try_clone().unwrap(); - let _ = thread::spawn(move || { - let current_device = CurrentDeviceInfo::new(virtual_ip, virtual_gateway, virtual_netmask, server_address); - if let Err(e) = handle::udp_recv_handler::other_loop(udp1, receiver, current_device, punch_sender) { - log::error!("udp数据处理线程停止 {:?}",e); - println!("udp数据处理线程停止:{:?}", e); - } - std::process::exit(1); - }); - } - //打洞处理 - { - let udp1 = udp.try_clone().unwrap(); - let _ = thread::spawn(move || { - let current_device = CurrentDeviceInfo::new(virtual_ip, virtual_gateway, virtual_netmask, server_address); - if let Err(e) = handle::punch_handler::cone_handle_loop(cone_receiver, udp1, current_device) { - log::error!("打洞响应线程停止 {:?}",e); - println!("打洞响应线程停止:{:?}", e); - } - }); - let udp1 = udp.try_clone().unwrap(); - let _ = thread::spawn(move || { - let current_device = CurrentDeviceInfo::new(virtual_ip, virtual_gateway, virtual_netmask, server_address); - if let Err(e) = handle::punch_handler::req_symmetric_handle_loop(req_symmetric_receiver, udp1, current_device) { - log::error!("打洞触发线程停止 {:?}",e); - println!("打洞触发线程停止:{:?}", e); - } - }); - let udp1 = udp.try_clone().unwrap(); - let _ = thread::spawn(move || { - let current_device = CurrentDeviceInfo::new(virtual_ip, virtual_gateway, virtual_netmask, server_address); - if let Err(e) = handle::punch_handler::res_symmetric_handle_loop(res_symmetric_receiver, udp1, current_device) { - log::error!("打洞触发线程停止 {:?}",e); - println!("打洞触发线程停止:{:?}", e); - } - }); - } - //tun数据处理 - { - let udp = udp.try_clone().unwrap(); - let _ = thread::spawn(move || { - let current_device = CurrentDeviceInfo::new(virtual_ip, virtual_gateway, virtual_netmask, server_address); - if let Err(e) = handle::tun_handler::handle_loop(udp, tun_reader, current_device) { - log::error!("tun数据处理线程停止 {:?}",e); - println!("tun数据处理线程停止:{:?}", e); - } - std::process::exit(1); - }); - } - use console::Term; - let term = Term::stdout(); - let current_device = CurrentDeviceInfo::new(virtual_ip, virtual_gateway, virtual_netmask, server_address); - loop { - println!("{}", style("Please enter the command (Usage: list,status,exit,help):").color256(102)); - match term.read_line() { - Ok(cmd) => { - command(cmd.trim(), ¤t_device); - } - Err(e) => { - println!("read_line:{:?}", e); - std::process::exit(1); - } - } - } -} - -fn command(cmd: &str, current_device: &CurrentDeviceInfo) { - match cmd { - "list" => { - let server_delay = SERVER_RT.load(Ordering::Relaxed); - let device_list_lock = DEVICE_LIST.lock(); - let (_epoch, device_list) = device_list_lock.clone(); - drop(device_list_lock); - if device_list.is_empty() { - println!("No other devices found"); - return; - } - for ip in device_list { - if let Some(route_ref) = DIRECT_ROUTE_TABLE.get(&ip) { - let str = if route_ref.value().delay >= 0 { - format!("{}(p2p delay:{}ms)", ip, route_ref.value().delay) - } else { - format!("{}(p2p)", ip) - }; - drop(route_ref); - println!("{}", style(str).green()); - } else { - let str = if server_delay >= 0 { - format!("{}(relay delay:{}ms)", ip, server_delay * 2) - } else { - format!("{}(relay)", ip) - }; - println!("{}", style(str).blue()); - } - } - } - "status" => { - let server_delay = SERVER_RT.load(Ordering::Relaxed); - println!("Virtual ip:{}", style(current_device.virtual_ip).green()); - println!("Virtual gateway:{}", style(current_device.virtual_gateway).green()); - println!("Relay server :{}", style(current_device.connect_server).green()); - if server_delay >= 0 { - println!("Delay of relay server :{}", style(server_delay).green()); - } - } - "help" | "h" => { - println!("Options: "); - println!("{} , Query the virtual IP of other devices", style("list").green()); - println!("{} , View current device status", style("status").green()); - println!("{} , Exit the program", style("exit").green()); - } - "exit" => { - std::process::exit(1); - } - _ => { - println!("command {} not fount. ", style(cmd).red()); - println!("Try to enter: '{}'", style("help").green()); - } - } -} diff --git a/switch-desktop/Cargo.toml b/switch-desktop/Cargo.toml new file mode 100644 index 0000000..c35fa77 --- /dev/null +++ b/switch-desktop/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "switch-desktop" +version = "0.1.0" +edition = "2021" + +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html + +[dependencies] +switch = {path="../switch"} +mac_address = "1.1.4" +clap = { version = "4.0.32", features = ["derive"] } +console = "0.15.2" +dirs = "4.0.0" +log = "0.4.17" +log4rs = "1.2.0" +tokio = { version = "1.24.1", features = ["full"] } + +[target.'cfg(any(target_os = "linux",target_os = "macos"))'.dependencies] +sudo = "0.6.0" + +[target.'cfg(target_os = "windows")'.dependencies] +winapi = { version = "0.3.9", features = ["handleapi", "processthreadsapi", "winnt", "securitybaseapi", "impl-default"] } +runas = "0.2.1" diff --git a/switch-desktop/src/main.rs b/switch-desktop/src/main.rs new file mode 100644 index 0000000..b2648b1 --- /dev/null +++ b/switch-desktop/src/main.rs @@ -0,0 +1,153 @@ +use clap::Parser; +use console::style; + +use switch::*; +use switch::handle::RouteType; + +#[cfg(windows)] +mod windows_admin_check; + +#[derive(Parser, Debug)] +#[command(author = "Lu Beilin", version, about = "一个虚拟网络工具,启动后会获取一个ip,相同token下的设备之间可以用ip直接通信")] +struct Args { + /// 32位字符 + /// 相同token的设备之间才能通信。 + /// 建议使用uuid保证唯一性。 + /// 32-bit characters. + /// Only devices with the same token can communicate with each other. + /// It is recommended to use uuid to ensure uniqueness + #[arg(short, long)] + token: String, +} + +fn log_init() { + let home = dirs::home_dir().unwrap().join(".switch"); + if !home.exists() { + std::fs::create_dir(&home).expect(" Failed to create '.switch' directory"); + } + let logfile = log4rs::append::file::FileAppender::builder() + // Pattern: https://docs.rs/log4rs/*/log4rs/encode/pattern/index.html + .encoder(Box::new(log4rs::encode::pattern::PatternEncoder::new("{d(%+)(utc)} [{f}:{L}] {h({l})} {M}:{m}{n}\n"))) + .build(home.join("switch.log")) + .unwrap(); + let config = log4rs::Config::builder() + .appender(log4rs::config::Appender::builder().build("logfile", Box::new(logfile))) + .build( + log4rs::config::Root::builder() + .appender("logfile") + .build(log::LevelFilter::Info), + ) + .unwrap(); + let _ = log4rs::init_config(config); +} + +fn main() { + log_init(); + let args = Args::parse(); + #[cfg(windows)] + if !windows_admin_check::is_app_elevated() { + let args: Vec<_> = std::env::args().collect(); + println!("{}", style("正在启动管理员权限执行...").red()); + if let Some(absolute_path) = std::env::current_exe() + .ok() + .and_then(|p| p.to_str().map(|p| p.to_string())) + { + let _ = runas::Command::new(&absolute_path).args(&args[1..]).status() + .expect("failed to execute"); + } else { + panic!("failed to execute") + } + return; + } + + #[cfg(any(unix))] + if sudo::RunningAs::Root != sudo::check() { + println!("{}", style("需要使用root权限执行...").red()); + sudo::escalate_if_needed().unwrap(); + } + println!("{}", style("starting...").green()); + let mac_address = mac_address::get_mac_address().unwrap().unwrap().to_string(); + let switch = Switch::start(Config::new(args.token, mac_address)).unwrap(); + use console::Term; + let term = Term::stdout(); + println!("{}", style("started").green()); + let current_device = switch.current_device(); + println!("当前虚拟ip(virtual ip): {:?}", style(current_device.virtual_ip).green()); + println!("虚拟网关(virtual gateway): {:?}", style(current_device.virtual_gateway).green()); + loop { + println!("{}", style("Please enter the command (Usage: list,status,exit,help):").color256(102)); + match term.read_line() { + Ok(cmd) => { + if command(cmd.trim(), &switch).is_err() { + println!("{}", style("stopping").red()); + switch.stop(); + break; + } + } + Err(e) => { + println!("read_line:{:?}", e); + println!("{}", style("stopping...").red()); + switch.stop(); + break; + } + } + } + println!("{}", style("stopped").red()); + std::process::exit(1); +} + +fn command(cmd: &str, switch: &Switch) -> Result<(), ()> { + match cmd { + "list" => { + let server_rt = switch.server_rt(); + let device_list = switch.device_list(); + if device_list.is_empty() { + println!("No other devices found"); + return Ok(()); + } + for ip in device_list { + let route = switch.route(&ip); + if route.route_type == RouteType::P2P { + let str = if route.rt >= 0 { + format!("{}(p2p delay:{}ms)", ip, route.rt) + } else { + format!("{}(p2p)", ip) + }; + println!("{}", style(str).green()); + } else { + let str = if server_rt >= 0 { + format!("{}(relay delay:{}ms)", ip, server_rt * 2) + } else { + format!("{}(relay)", ip) + }; + println!("{}", style(str).blue()); + } + } + } + "status" => { + let server_rt = switch.server_rt(); + let current_device = switch.current_device(); + println!("Virtual ip:{}", style(current_device.virtual_ip).green()); + println!("Virtual gateway:{}", style(current_device.virtual_gateway).green()); + println!("Connection status :{}", style(format!("{:?}", switch.connection_status())).green()); + println!("Relay server :{}", style(current_device.connect_server).green()); + if server_rt >= 0 { + println!("Delay of relay server :{}ms", style(server_rt).green()); + } + } + "help" | "h" => { + println!("Options: "); + println!("{} , Query the virtual IP of other devices", style("list").green()); + println!("{} , View current device status", style("status").green()); + println!("{} , Exit the program", style("exit").green()); + } + "exit" => { + return Err(()); + } + _ => { + println!("command '{}' not fount. ", style(cmd).red()); + println!("Try to enter: '{}'", style("help").green()); + } + } + Ok(()) +} diff --git a/src/admin_check.rs b/switch-desktop/src/windows_admin_check.rs similarity index 97% rename from src/admin_check.rs rename to switch-desktop/src/windows_admin_check.rs index 94ca4d3..c818547 100644 --- a/src/admin_check.rs +++ b/switch-desktop/src/windows_admin_check.rs @@ -1,76 +1,76 @@ -/// 使用 https://github.com/spa5k/is_sudo/blob/main/src/window.rs -use std::io::Error; -use std::ptr; - -use winapi::um::handleapi::CloseHandle; -use winapi::um::processthreadsapi::{GetCurrentProcess, OpenProcessToken}; -use winapi::um::securitybaseapi::GetTokenInformation; -use winapi::um::winnt::{HANDLE, TOKEN_ELEVATION, TOKEN_QUERY, TokenElevation}; - -// Use std::io::Error::last_os_error for errors. -// NOTE: For this example I'm simple passing on the OS error. -// However, customising the error could provide more context - -/// Returns true if the current process has admin rights, otherwise false. -pub fn is_app_elevated() -> bool { - _is_app_elevated().unwrap_or(false) -} - -/// On success returns a bool indicating if the current process has admin rights. -/// Otherwise returns an OS error. -/// -/// This is unlikely to fail but if it does it's even more unlikely that you have admin permissions anyway. -/// Therefore the public function above simply eats the error and returns a bool. -fn _is_app_elevated() -> Result { - let token = QueryAccessToken::from_current_process()?; - token.is_elevated() -} - -/// A safe wrapper around querying Windows access tokens. -pub struct QueryAccessToken(HANDLE); - -impl QueryAccessToken { - pub fn from_current_process() -> Result { - unsafe { - let mut handle: HANDLE = ptr::null_mut(); - let result = OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &mut handle); - - if result != 0 { - Ok(Self(handle)) - } else { - Err(Error::last_os_error()) - } - } - } - - /// On success returns a bool indicating if the access token has elevated privilidges. - /// Otherwise returns an OS error. - pub fn is_elevated(&self) -> Result { - unsafe { - let mut elevation = TOKEN_ELEVATION::default(); - let size = std::mem::size_of::() as u32; - let mut ret_size = size; - // The weird looking repetition of `as *mut _` is casting the reference to a c_void pointer. - if GetTokenInformation( - self.0, - TokenElevation, - &mut elevation as *mut _ as *mut _, - size, - &mut ret_size, - ) != 0 - { - Ok(elevation.TokenIsElevated != 0) - } else { - Err(Error::last_os_error()) - } - } - } -} - -impl Drop for QueryAccessToken { - fn drop(&mut self) { - if !self.0.is_null() { - unsafe { CloseHandle(self.0) }; - } - } -} +/// 使用 https://github.com/spa5k/is_sudo/blob/main/src/window.rs +use std::io::Error; +use std::ptr; + +use winapi::um::handleapi::CloseHandle; +use winapi::um::processthreadsapi::{GetCurrentProcess, OpenProcessToken}; +use winapi::um::securitybaseapi::GetTokenInformation; +use winapi::um::winnt::{HANDLE, TOKEN_ELEVATION, TOKEN_QUERY, TokenElevation}; + +// Use std::io::Error::last_os_error for errors. +// NOTE: For this example I'm simple passing on the OS error. +// However, customising the error could provide more context + +/// Returns true if the current process has admin rights, otherwise false. +pub fn is_app_elevated() -> bool { + _is_app_elevated().unwrap_or(false) +} + +/// On success returns a bool indicating if the current process has admin rights. +/// Otherwise returns an OS error. +/// +/// This is unlikely to fail but if it does it's even more unlikely that you have admin permissions anyway. +/// Therefore the public function above simply eats the error and returns a bool. +fn _is_app_elevated() -> Result { + let token = QueryAccessToken::from_current_process()?; + token.is_elevated() +} + +/// A safe wrapper around querying Windows access tokens. +pub struct QueryAccessToken(HANDLE); + +impl QueryAccessToken { + pub fn from_current_process() -> Result { + unsafe { + let mut handle: HANDLE = ptr::null_mut(); + let result = OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &mut handle); + + if result != 0 { + Ok(Self(handle)) + } else { + Err(Error::last_os_error()) + } + } + } + + /// On success returns a bool indicating if the access token has elevated privilidges. + /// Otherwise returns an OS error. + pub fn is_elevated(&self) -> Result { + unsafe { + let mut elevation = TOKEN_ELEVATION::default(); + let size = std::mem::size_of::() as u32; + let mut ret_size = size; + // The weird looking repetition of `as *mut _` is casting the reference to a c_void pointer. + if GetTokenInformation( + self.0, + TokenElevation, + &mut elevation as *mut _ as *mut _, + size, + &mut ret_size, + ) != 0 + { + Ok(elevation.TokenIsElevated != 0) + } else { + Err(Error::last_os_error()) + } + } + } +} + +impl Drop for QueryAccessToken { + fn drop(&mut self) { + if !self.0.is_null() { + unsafe { CloseHandle(self.0) }; + } + } +} diff --git a/switch-jni/Cargo.toml b/switch-jni/Cargo.toml new file mode 100644 index 0000000..add87e9 --- /dev/null +++ b/switch-jni/Cargo.toml @@ -0,0 +1,13 @@ +[package] +name = "switch-jni" +version = "0.1.0" +edition = "2021" + +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html +[lib] +crate-type = ['cdylib'] + +[dependencies] +switch = {path="../switch"} +jni = "0.20.0" +anyhow = "1.0.65" \ No newline at end of file diff --git a/switch-jni/src/lib.rs b/switch-jni/src/lib.rs new file mode 100644 index 0000000..d11572f --- /dev/null +++ b/switch-jni/src/lib.rs @@ -0,0 +1,168 @@ +use std::net::{IpAddr, Ipv4Addr}; +use std::str::Utf8Error; + +use jni::errors::Error; +use jni::JNIEnv; +use jni::objects::{JClass, JList, JObject, JString, JValue}; +use jni::sys::{jbyte, jint, jintArray, jlong, jobject, jobjectArray, jsize}; + +use switch::{Config, Switch}; +use switch::handle::{CurrentDeviceInfo, Route}; + +fn to_string(env: &JNIEnv, config: JObject, name: &str) -> Result, Error> { + let value = env.get_field(config, name, "Ljava/lang/String;")?.l()?; + if value.is_null() { + env.throw_new("Ljava/lang/NullPointerException", &name).expect("throw"); + return Ok(None); + } + let value = env.get_string(JString::from(value))?; + match value.to_str() { + Ok(value) => { + Ok(Some(value.to_string())) + } + Err(_) => { + env.throw_new("Ljava/lang/RuntimeException", "not utf-8").expect("throw"); + Ok(None) + } + } +} + +fn start(env: &JNIEnv, config: JObject) -> Result, Error> { + if let Some(token) = to_string(&env, config, "token")? { + if let Some(mac_address) = to_string(&env, config, "macAddress")? { + match Switch::start(Config::new(token, mac_address)) { + Ok(switch) => { + return Ok(Some(switch)); + } + Err(e) => { + env.throw_new("Ljava/lang/RuntimeException", format!("switch start failed {:?}", e)).expect("throw"); + } + } + } + } + Ok(None) +} + +#[no_mangle] +pub unsafe extern "C" fn Java_org_switches_jni_Switch_start0(env: JNIEnv, _class: JClass, config: JObject) -> jlong { + match start(&env, config) { + Ok(switch) => { + if let Some(switch) = switch { + return Box::into_raw(Box::new(switch)) as jlong; + } + } + Err(_) => {} + } + return 0; +} + +#[no_mangle] +pub unsafe extern "C" fn Java_org_switches_jni_Switch_stop0(env: JNIEnv, _class: JClass, raw_switch: jlong) { + let switch = Box::from_raw(raw_switch as *mut Switch); + switch.stop(); +} + +#[no_mangle] +pub unsafe extern "C" fn Java_org_switches_jni_Switch_currentDevice0(env: JNIEnv, _class: JClass, raw_switch: jlong) -> jobject { + let switch = raw_switch as *mut Switch; + let dev_info = (&*switch).current_device(); + match current_device(&env, dev_info) { + Ok(obj) => { + obj + } + Err(_) => { + std::ptr::null_mut() + } + } +} + +#[no_mangle] +pub unsafe extern "C" fn Java_org_switches_jni_Switch_deviceList0(env: JNIEnv, _class: JClass, raw_switch: jlong) -> jintArray { + let switch = raw_switch as *mut Switch; + match device_list(&env, (&*switch).device_list()) { + Ok(arr) => { + arr + } + Err(_) => { + std::ptr::null_mut() + } + } +} + +#[no_mangle] +pub unsafe extern "C" fn Java_org_switches_jni_Switch_route0(env: JNIEnv, _class: JClass, raw_switch: jlong, ip: jint) -> jobject { + let ip = Ipv4Addr::from(ip as u32); + let switch = raw_switch as *mut Switch; + match route(&env, (&*switch).route(&ip)) { + Ok(arr) => { + arr + } + Err(_) => { + std::ptr::null_mut() + } + } +} + +#[no_mangle] +pub unsafe extern "C" fn Java_org_switches_jni_Switch_serverRt0(env: JNIEnv, _class: JClass, raw_switch: jlong) -> jlong { + let switch = raw_switch as *mut Switch; + let rt = (&*switch).server_rt(); + rt as jlong +} + +#[no_mangle] +pub unsafe extern "C" fn Java_org_switches_jni_Switch_connectionStatus0(env: JNIEnv, _class: JClass, raw_switch: jlong) -> jbyte { + let switch = raw_switch as *mut Switch; + let connection_status: u8 = (&*switch).connection_status().into(); + connection_status as jbyte +} + +fn route(env: &JNIEnv, route: Route) -> Result { + let route_type: u8 = route.route_type.into(); + let rt = route.rt; + let route = env.new_object( + "org/switches/jni/Route", + "(BJ)V", + &[JValue::Byte(route_type as jbyte), JValue::Long(rt as jlong)], + )?; + Ok(route.into_raw()) +} + +fn device_list(env: &JNIEnv, device_list: Vec) -> Result { + if device_list.is_empty() { + return Ok(std::ptr::null_mut()); + } + let arr = env.new_int_array(device_list.len() as jsize)?; + let devices: Vec = device_list.iter().map(|ip| { + let ip: u32 = (*ip).into(); + ip as jint + }).collect(); + env.set_int_array_region(arr, 0, &devices)?; + Ok(arr) +} + +fn current_device(env: &JNIEnv, dev_info: &CurrentDeviceInfo) -> Result { + let virtual_ip: u32 = dev_info.virtual_ip.into(); + let virtual_gateway: u32 = dev_info.virtual_gateway.into(); + let virtual_netmask: u32 = dev_info.virtual_netmask.into(); + let virtual_network: u32 = dev_info.virtual_network.into(); + let broadcast_address: u32 = dev_info.broadcast_address.into(); + let connect_server_host: u32 = match dev_info.connect_server.ip() { + IpAddr::V4(ip) => { + ip.into() + } + IpAddr::V6(_) => { + panic!() + } + }; + let connect_server_port = dev_info.connect_server.port() as u32; + let current_device = env.new_object( + "org/switches/jni/CurrentDevice", + "(IIIIIII)V", + &[JValue::Int(virtual_ip as jint), JValue::Int(virtual_gateway as jint), + JValue::Int(virtual_netmask as jint), JValue::Int(virtual_network as jint), + JValue::Int(broadcast_address as jint), JValue::Int(connect_server_host as jint), + JValue::Int(connect_server_port as jint)], + )?; + Ok(current_device.into_raw()) +} diff --git a/switch/Cargo.toml b/switch/Cargo.toml new file mode 100644 index 0000000..8c7b252 --- /dev/null +++ b/switch/Cargo.toml @@ -0,0 +1,38 @@ +[package] +name = "switch" +version = "0.1.0" +edition = "2021" + +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html + +[dependencies] +packet = { path = "./packet" } +bytes = "1.3.0" +log = "0.4.17" +libc = "0.2.137" + +dashmap = "5.4.0" +crossbeam = "0.8.2" +parking_lot = "0.12.1" + +rsa = "0.7.2" +rand = "0.8.5" +sha2 = { version = "0.10.6", features = ["oid"] } + +thiserror = "1.0.37" +chrono = "0.4.23" +lazy_static = "1.4.0" +moka = "0.9.6" +protobuf = "3.2.0" + +tokio = { version = "1.24.1", features = ["full"] } +[target.'cfg(any(unix))'.dependencies] +tun = { path = "./rust-tun" } + +[target.'cfg(target_os = "windows")'.dependencies] +wintun = "0.2.1" +libloading = "0.7.4" + +[build-dependencies] +protobuf-codegen = "3.2.0" +protoc-bin-vendored = "3.0.0" \ No newline at end of file diff --git a/LICENSE b/switch/LICENSE similarity index 100% rename from LICENSE rename to switch/LICENSE diff --git a/build.rs b/switch/build.rs similarity index 100% rename from build.rs rename to switch/build.rs diff --git a/packet/Cargo.toml b/switch/packet/Cargo.toml similarity index 100% rename from packet/Cargo.toml rename to switch/packet/Cargo.toml diff --git a/packet/src/error.rs b/switch/packet/src/error.rs similarity index 100% rename from packet/src/error.rs rename to switch/packet/src/error.rs diff --git a/packet/src/icmp/icmp.rs b/switch/packet/src/icmp/icmp.rs similarity index 100% rename from packet/src/icmp/icmp.rs rename to switch/packet/src/icmp/icmp.rs diff --git a/packet/src/icmp/mod.rs b/switch/packet/src/icmp/mod.rs similarity index 100% rename from packet/src/icmp/mod.rs rename to switch/packet/src/icmp/mod.rs diff --git a/packet/src/ip/ipv4/mod.rs b/switch/packet/src/ip/ipv4/mod.rs similarity index 100% rename from packet/src/ip/ipv4/mod.rs rename to switch/packet/src/ip/ipv4/mod.rs diff --git a/packet/src/ip/ipv4/packet.rs b/switch/packet/src/ip/ipv4/packet.rs similarity index 100% rename from packet/src/ip/ipv4/packet.rs rename to switch/packet/src/ip/ipv4/packet.rs diff --git a/packet/src/ip/ipv4/protocol.rs b/switch/packet/src/ip/ipv4/protocol.rs similarity index 100% rename from packet/src/ip/ipv4/protocol.rs rename to switch/packet/src/ip/ipv4/protocol.rs diff --git a/packet/src/ip/mod.rs b/switch/packet/src/ip/mod.rs similarity index 100% rename from packet/src/ip/mod.rs rename to switch/packet/src/ip/mod.rs diff --git a/packet/src/lib.rs b/switch/packet/src/lib.rs similarity index 100% rename from packet/src/lib.rs rename to switch/packet/src/lib.rs diff --git a/packet/src/tcp/mod.rs b/switch/packet/src/tcp/mod.rs similarity index 100% rename from packet/src/tcp/mod.rs rename to switch/packet/src/tcp/mod.rs diff --git a/packet/src/tcp/tcp.rs b/switch/packet/src/tcp/tcp.rs similarity index 100% rename from packet/src/tcp/tcp.rs rename to switch/packet/src/tcp/tcp.rs diff --git a/packet/src/udp/mod.rs b/switch/packet/src/udp/mod.rs similarity index 92% rename from packet/src/udp/mod.rs rename to switch/packet/src/udp/mod.rs index 21e3805..7e5aaa1 100644 --- a/packet/src/udp/mod.rs +++ b/switch/packet/src/udp/mod.rs @@ -1 +1 @@ -pub mod udp; +pub mod udp; diff --git a/packet/src/udp/udp.rs b/switch/packet/src/udp/udp.rs similarity index 100% rename from packet/src/udp/udp.rs rename to switch/packet/src/udp/udp.rs diff --git a/proto/message.proto b/switch/proto/message.proto similarity index 94% rename from proto/message.proto rename to switch/proto/message.proto index 68415ad..81cf1bd 100644 --- a/proto/message.proto +++ b/switch/proto/message.proto @@ -1,40 +1,40 @@ -syntax = "proto3"; -message RegistrationRequest{ - string token = 1; - string mac_address = 2; -} - -message RegistrationResponse{ - fixed32 virtual_ip = 1; - fixed32 virtual_gateway = 2; - fixed32 virtual_netmask = 3; - uint32 epoch = 4; - repeated fixed32 virtual_ip_list = 5; - fixed32 public_ip = 6; - uint32 public_port = 7; -} - -message DeviceList{ - uint32 epoch = 1; - repeated fixed32 virtual_ip_list = 2; -} - -message Punch{ - fixed32 virtual_ip = 1; - repeated fixed32 public_ip_list = 2; - uint32 public_port = 3; - uint32 public_port_range = 4; - NatType nat_type = 5; - bool reply = 6; - Step step = 7; -} -enum NatType{ - Symmetric = 0; - Cone = 1; -} -enum Step{ - Step1 = 0; - Step2 = 1; - Step3 = 2; - Step4 = 3; +syntax = "proto3"; +message RegistrationRequest{ + string token = 1; + string mac_address = 2; +} + +message RegistrationResponse{ + fixed32 virtual_ip = 1; + fixed32 virtual_gateway = 2; + fixed32 virtual_netmask = 3; + uint32 epoch = 4; + repeated fixed32 virtual_ip_list = 5; + fixed32 public_ip = 6; + uint32 public_port = 7; +} + +message DeviceList{ + uint32 epoch = 1; + repeated fixed32 virtual_ip_list = 2; +} + +message Punch{ + fixed32 virtual_ip = 1; + repeated fixed32 public_ip_list = 2; + uint32 public_port = 3; + uint32 public_port_range = 4; + NatType nat_type = 5; + bool reply = 6; + Step step = 7; +} +enum NatType{ + Symmetric = 0; + Cone = 1; +} +enum Step{ + Step1 = 0; + Step2 = 1; + Step3 = 2; + Step4 = 3; } \ No newline at end of file diff --git a/rust-tun/Cargo.toml b/switch/rust-tun/Cargo.toml similarity index 100% rename from rust-tun/Cargo.toml rename to switch/rust-tun/Cargo.toml diff --git a/rust-tun/README.md b/switch/rust-tun/README.md similarity index 100% rename from rust-tun/README.md rename to switch/rust-tun/README.md diff --git a/rust-tun/examples/ping-tun.rs b/switch/rust-tun/examples/ping-tun.rs similarity index 100% rename from rust-tun/examples/ping-tun.rs rename to switch/rust-tun/examples/ping-tun.rs diff --git a/rust-tun/examples/read-async-codec.rs b/switch/rust-tun/examples/read-async-codec.rs similarity index 100% rename from rust-tun/examples/read-async-codec.rs rename to switch/rust-tun/examples/read-async-codec.rs diff --git a/rust-tun/examples/read-async.rs b/switch/rust-tun/examples/read-async.rs similarity index 100% rename from rust-tun/examples/read-async.rs rename to switch/rust-tun/examples/read-async.rs diff --git a/rust-tun/examples/read.rs b/switch/rust-tun/examples/read.rs similarity index 100% rename from rust-tun/examples/read.rs rename to switch/rust-tun/examples/read.rs diff --git a/rust-tun/src/address.rs b/switch/rust-tun/src/address.rs similarity index 100% rename from rust-tun/src/address.rs rename to switch/rust-tun/src/address.rs diff --git a/rust-tun/src/async/codec.rs b/switch/rust-tun/src/async/codec.rs similarity index 100% rename from rust-tun/src/async/codec.rs rename to switch/rust-tun/src/async/codec.rs diff --git a/rust-tun/src/async/device.rs b/switch/rust-tun/src/async/device.rs similarity index 100% rename from rust-tun/src/async/device.rs rename to switch/rust-tun/src/async/device.rs diff --git a/rust-tun/src/async/mod.rs b/switch/rust-tun/src/async/mod.rs similarity index 100% rename from rust-tun/src/async/mod.rs rename to switch/rust-tun/src/async/mod.rs diff --git a/rust-tun/src/configuration.rs b/switch/rust-tun/src/configuration.rs similarity index 100% rename from rust-tun/src/configuration.rs rename to switch/rust-tun/src/configuration.rs diff --git a/rust-tun/src/device.rs b/switch/rust-tun/src/device.rs similarity index 100% rename from rust-tun/src/device.rs rename to switch/rust-tun/src/device.rs diff --git a/rust-tun/src/error.rs b/switch/rust-tun/src/error.rs similarity index 100% rename from rust-tun/src/error.rs rename to switch/rust-tun/src/error.rs diff --git a/rust-tun/src/lib.rs b/switch/rust-tun/src/lib.rs similarity index 100% rename from rust-tun/src/lib.rs rename to switch/rust-tun/src/lib.rs diff --git a/rust-tun/src/platform/android/device.rs b/switch/rust-tun/src/platform/android/device.rs similarity index 100% rename from rust-tun/src/platform/android/device.rs rename to switch/rust-tun/src/platform/android/device.rs diff --git a/rust-tun/src/platform/android/mod.rs b/switch/rust-tun/src/platform/android/mod.rs similarity index 100% rename from rust-tun/src/platform/android/mod.rs rename to switch/rust-tun/src/platform/android/mod.rs diff --git a/rust-tun/src/platform/ios/device.rs b/switch/rust-tun/src/platform/ios/device.rs similarity index 100% rename from rust-tun/src/platform/ios/device.rs rename to switch/rust-tun/src/platform/ios/device.rs diff --git a/rust-tun/src/platform/ios/mod.rs b/switch/rust-tun/src/platform/ios/mod.rs similarity index 100% rename from rust-tun/src/platform/ios/mod.rs rename to switch/rust-tun/src/platform/ios/mod.rs diff --git a/rust-tun/src/platform/linux/device.rs b/switch/rust-tun/src/platform/linux/device.rs similarity index 100% rename from rust-tun/src/platform/linux/device.rs rename to switch/rust-tun/src/platform/linux/device.rs diff --git a/rust-tun/src/platform/linux/mod.rs b/switch/rust-tun/src/platform/linux/mod.rs similarity index 100% rename from rust-tun/src/platform/linux/mod.rs rename to switch/rust-tun/src/platform/linux/mod.rs diff --git a/rust-tun/src/platform/linux/sys.rs b/switch/rust-tun/src/platform/linux/sys.rs similarity index 100% rename from rust-tun/src/platform/linux/sys.rs rename to switch/rust-tun/src/platform/linux/sys.rs diff --git a/rust-tun/src/platform/macos/device.rs b/switch/rust-tun/src/platform/macos/device.rs similarity index 100% rename from rust-tun/src/platform/macos/device.rs rename to switch/rust-tun/src/platform/macos/device.rs diff --git a/rust-tun/src/platform/macos/mod.rs b/switch/rust-tun/src/platform/macos/mod.rs similarity index 100% rename from rust-tun/src/platform/macos/mod.rs rename to switch/rust-tun/src/platform/macos/mod.rs diff --git a/rust-tun/src/platform/macos/sys.rs b/switch/rust-tun/src/platform/macos/sys.rs similarity index 100% rename from rust-tun/src/platform/macos/sys.rs rename to switch/rust-tun/src/platform/macos/sys.rs diff --git a/rust-tun/src/platform/mod.rs b/switch/rust-tun/src/platform/mod.rs similarity index 100% rename from rust-tun/src/platform/mod.rs rename to switch/rust-tun/src/platform/mod.rs diff --git a/rust-tun/src/platform/posix/fd.rs b/switch/rust-tun/src/platform/posix/fd.rs similarity index 100% rename from rust-tun/src/platform/posix/fd.rs rename to switch/rust-tun/src/platform/posix/fd.rs diff --git a/rust-tun/src/platform/posix/mod.rs b/switch/rust-tun/src/platform/posix/mod.rs similarity index 100% rename from rust-tun/src/platform/posix/mod.rs rename to switch/rust-tun/src/platform/posix/mod.rs diff --git a/rust-tun/src/platform/posix/sockaddr.rs b/switch/rust-tun/src/platform/posix/sockaddr.rs similarity index 100% rename from rust-tun/src/platform/posix/sockaddr.rs rename to switch/rust-tun/src/platform/posix/sockaddr.rs diff --git a/rust-tun/src/platform/posix/split.rs b/switch/rust-tun/src/platform/posix/split.rs similarity index 100% rename from rust-tun/src/platform/posix/split.rs rename to switch/rust-tun/src/platform/posix/split.rs diff --git a/src/error/mod.rs b/switch/src/error/mod.rs similarity index 83% rename from src/error/mod.rs rename to switch/src/error/mod.rs index 7d46c90..576bf03 100644 --- a/src/error/mod.rs +++ b/switch/src/error/mod.rs @@ -7,6 +7,8 @@ use thiserror::Error; pub enum Error { #[error("packet error")] PacketError(#[from] packet::error::Error), + #[error("TokioWatchRecvError")] + TokioWatchRecvError(#[from] tokio::sync::watch::error::RecvError), #[error("Io error")] Io(#[from] io::Error), #[error("Channel error")] diff --git a/src/handle/heartbeat_handler.rs b/switch/src/handle/heartbeat_handler.rs similarity index 57% rename from src/handle/heartbeat_handler.rs rename to switch/src/handle/heartbeat_handler.rs index 09721c8..9e506c7 100644 --- a/src/handle/heartbeat_handler.rs +++ b/switch/src/handle/heartbeat_handler.rs @@ -1,16 +1,31 @@ use std::net::{SocketAddr, UdpSocket}; -use std::thread; use std::time::Duration; use chrono::Local; +use tokio::sync::watch::Receiver; +use tokio::time::sleep; -use crate::DEVICE_LIST; +use crate::{CurrentDeviceInfo, DEVICE_LIST}; use crate::error::*; -use crate::handle::DIRECT_ROUTE_TABLE; +use crate::handle::{ApplicationStatus, DIRECT_ROUTE_TABLE}; use crate::protocol::{control_packet, NetPacket, Protocol, Version}; use crate::protocol::control_packet::PingPacket; -pub fn handle_loop(udp: UdpSocket, server_addr: SocketAddr) -> Result<()> { +pub async fn start(status_watch: Receiver, + udp: UdpSocket, cur_info: CurrentDeviceInfo, stop_fn: F) + where F: FnOnce() + Send + 'static { + tokio::spawn(async move { + match handle_loop(status_watch, udp, cur_info.connect_server).await { + Ok(_) => {} + Err(e) => { + log::error!("{:?}",e) + } + } + stop_fn(); + }); +} + +async fn handle_loop(mut status_watch: Receiver, udp: UdpSocket, server_addr: SocketAddr) -> Result<()> { const INTERVAL: u64 = 3000; const MAX_INTERVAL: i64 = 3000 * 3; let mut buf = [0u8; (4 + 8 + 4)]; @@ -40,6 +55,16 @@ pub fn handle_loop(udp: UdpSocket, server_addr: SocketAddr) -> Result<()> { }); } } - thread::sleep(Duration::from_millis(INTERVAL)); + tokio::select! { + _ = sleep(Duration::from_millis(INTERVAL))=>{ + + } + status = status_watch.changed() =>{ + status?; + if *status_watch.borrow() != ApplicationStatus::Starting{ + return Ok(()) + } + } + } } } diff --git a/src/handle/mod.rs b/switch/src/handle/mod.rs similarity index 78% rename from src/handle/mod.rs rename to switch/src/handle/mod.rs index bb0d57f..674bcff 100644 --- a/src/handle/mod.rs +++ b/switch/src/handle/mod.rs @@ -32,6 +32,26 @@ lazy_static! { /// 当前设备的nat信息 pub static ref NAT_INFO:Mutex> = const_mutex(None); } +#[derive(Copy, Clone, Debug, Eq, PartialEq)] +pub enum ApplicationStatus { + Starting, + Stopping, +} + +#[derive(Copy, Clone, Debug, Eq, PartialEq)] +pub enum ConnectStatus { + Connecting, + Connected, +} + +impl Into for ConnectStatus { + fn into(self) -> u8 { + match self { + ConnectStatus::Connecting => 0, + ConnectStatus::Connected => 1, + } + } +} #[derive(Clone, Debug)] pub struct NatInfo { @@ -80,17 +100,17 @@ pub fn init_nat_info(public_ip: u32, public_port: u16) { } } -#[derive(Clone, Debug)] +#[derive(Copy, Clone, Debug)] pub struct CurrentDeviceInfo { - pub(crate) virtual_ip: Ipv4Addr, - pub(crate) virtual_gateway: Ipv4Addr, - pub(crate) virtual_netmask: Ipv4Addr, + pub virtual_ip: Ipv4Addr, + pub virtual_gateway: Ipv4Addr, + pub virtual_netmask: Ipv4Addr, //网络地址 - pub(crate) virtual_network: Ipv4Addr, + pub virtual_network: Ipv4Addr, //直接广播地址 - pub(crate) broadcast_address: Ipv4Addr, + pub broadcast_address: Ipv4Addr, //链接的服务器地址 - pub(crate) connect_server: SocketAddr, + pub connect_server: SocketAddr, } impl CurrentDeviceInfo { @@ -114,18 +134,35 @@ impl CurrentDeviceInfo { #[derive(Clone, Debug)] pub struct Route { - pub(crate) address: SocketAddr, + pub route_type: RouteType, + pub address: SocketAddr, //用心跳探测延迟，收包时更新 - pub(crate) delay: i64, + pub rt: i64, //收包时更新，如果太久没有收到消息则剔除 - pub(crate) recv_time: i64, + pub recv_time: i64, +} + +#[derive(Copy, Clone, Debug, Eq, PartialEq)] +pub enum RouteType { + ServerRelay, + P2P, +} + +impl Into for RouteType { + fn into(self) -> u8 { + match self { + RouteType::ServerRelay => 0, + RouteType::P2P => 1 + } + } } impl Route { pub fn new(address: SocketAddr) -> Self { Self { + route_type: RouteType::P2P, address, - delay: -1, + rt: -1, recv_time: Local::now().timestamp_millis(), } } diff --git a/src/handle/punch_handler.rs b/switch/src/handle/punch_handler.rs similarity index 56% rename from src/handle/punch_handler.rs rename to switch/src/handle/punch_handler.rs index 85c3ca9..5afa1d2 100644 --- a/src/handle/punch_handler.rs +++ b/switch/src/handle/punch_handler.rs @@ -2,14 +2,16 @@ use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4, UdpSocket}; use std::thread; use std::time::Duration; -use crossbeam::channel::{Receiver, RecvTimeoutError, Sender, SendError, TrySendError}; use dashmap::DashMap; use lazy_static::lazy_static; use protobuf::Message; +use tokio::sync::mpsc::{Receiver, Sender}; +use tokio::sync::mpsc::error::TrySendError; +use tokio::sync::watch; -use crate::{CurrentDeviceInfo, DEVICE_LIST, NAT_INFO, NatInfo}; +use crate::{CurrentDeviceInfo, DEVICE_LIST, handle::NAT_INFO, handle::NatInfo}; use crate::error::*; -use crate::handle::DIRECT_ROUTE_TABLE; +use crate::handle::{ApplicationStatus, DIRECT_ROUTE_TABLE}; use crate::proto::message::{NatType, Punch, Step}; use crate::protocol::{control_packet, NetPacket, Protocol, turn_packet, Version}; use crate::protocol::control_packet::PunchRequestPacket; @@ -20,9 +22,9 @@ lazy_static! { } /// 每一种类型一个通道，减少相互干扰 pub fn bounded() -> (PunchSender, ConeReceiver, ReqSymmetricReceiver, ResSymmetricReceiver) { - let (cone_sender, cone_receiver) = crossbeam::channel::bounded(3); - let (req_symmetric_sender, req_symmetric_receiver) = crossbeam::channel::bounded(1); - let (res_symmetric_sender, res_symmetric_receiver) = crossbeam::channel::bounded(1); + let (cone_sender, cone_receiver) = tokio::sync::mpsc::channel(3); + let (req_symmetric_sender, req_symmetric_receiver) = tokio::sync::mpsc::channel(1); + let (res_symmetric_sender, res_symmetric_receiver) = tokio::sync::mpsc::channel(1); (PunchSender::new(cone_sender, req_symmetric_sender, res_symmetric_sender), ConeReceiver(cone_receiver), ReqSymmetricReceiver(req_symmetric_receiver), ResSymmetricReceiver(res_symmetric_receiver)) @@ -51,21 +53,21 @@ impl PunchSender { res_symmetric_sender, } } - pub fn send(&self, punch: Punch) -> std::result::Result<(), SendError> { - match punch.nat_type.enum_value_or_default() { - NatType::Symmetric => { - if punch.reply { - // 为true表示回应，也就是主动发起的打洞操作 - self.res_symmetric_sender.send(punch) - } else { - self.req_symmetric_sender.send(punch) - } - } - NatType::Cone => { - self.cone_sender.send(punch) - } - } - } + // pub fn send(&self, punch: Punch) -> std::result::Result<(), SendError> { + // match punch.nat_type.enum_value_or_default() { + // NatType::Symmetric => { + // if punch.reply { + // // 为true表示回应，也就是主动发起的打洞操作 + // self.res_symmetric_sender.blocking_send(punch) + // } else { + // self.req_symmetric_sender.blocking_send(punch) + // } + // } + // NatType::Cone => { + // self.cone_sender.blocking_send(punch) + // } + // } + // } pub fn try_send(&self, punch: Punch) -> std::result::Result<(), TrySendError> { match punch.nat_type.enum_value_or_default() { NatType::Symmetric => { @@ -83,7 +85,7 @@ impl PunchSender { } } -fn handle(udp: &UdpSocket, punch_list: Vec, buf: &[u8]) -> Result<()> { +fn handle(status_watch: &watch::Receiver, udp: &UdpSocket, punch_list: Vec, buf: &[u8]) -> Result<()> { let mut counter = 0u64; for punch in punch_list { let dest = Ipv4Addr::from(punch.virtual_ip); @@ -99,6 +101,11 @@ fn handle(udp: &UdpSocket, punch_list: Vec, buf: &[u8]) -> Result<()> { for pub_ip in punch.public_ip_list { let pub_ip = Ipv4Addr::from(pub_ip); for range in 0..punch.public_port_range + 1 { + if counter & 10 == 10 { + if status_watch.has_changed()? { + return Ok(()); + } + } let right_port = ((punch.public_port + range) & 0xFFFF) as u16; let left_port = ((0xFFFF + punch.public_port - range) & 0xFFFF) as u16; if right_port != 0 { @@ -128,6 +135,11 @@ fn handle(udp: &UdpSocket, punch_list: Vec, buf: &[u8]) -> Result<()> { for pub_ip in punch.public_ip_list { let pub_ip = Ipv4Addr::from(pub_ip); for port in 1..0xFFFF { + if counter & 10 == 10 { + if status_watch.has_changed()? { + return Ok(()); + } + } udp.send_to( buf, SocketAddr::V4(SocketAddrV4::new(pub_ip, port)), @@ -156,22 +168,56 @@ fn handle(udp: &UdpSocket, punch_list: Vec, buf: &[u8]) -> Result<()> { } /// 给对称nat发送打洞数据包 -pub fn req_symmetric_handle_loop( - receiver: ReqSymmetricReceiver, - udp: UdpSocket, - cur_info: CurrentDeviceInfo, -) -> Result<()> { +pub async fn req_symmetric_handler_start(status_watch: watch::Receiver, + receiver: ReqSymmetricReceiver, + udp: UdpSocket, + cur_info: CurrentDeviceInfo, + stop_fn: F) where F: FnOnce() +Send+'static{ let receiver = receiver.0; - handle_loop(receiver, udp, cur_info) + tokio::spawn(async move { + match handle_loop(status_watch, receiver, udp, cur_info).await { + Ok(_) => {} + Err(e) => { + log::error!("{:?}",e) + } + } + stop_fn() + }); } +// pub fn req_symmetric_handle_loop( +// receiver: ReqSymmetricReceiver, +// udp: UdpSocket, +// cur_info: CurrentDeviceInfo, +// ) -> Result<()> { +// let receiver = receiver.0; +// handle_loop(receiver, udp, cur_info) +// } + /// 给对称nat发送打洞数据包，处理主动发起的打洞操作 -pub fn res_symmetric_handle_loop( - receiver: ResSymmetricReceiver, +pub async fn res_symmetric_handler_start(status_watch: watch::Receiver, + receiver: ResSymmetricReceiver, + udp: UdpSocket, + cur_info: CurrentDeviceInfo, + stop_fn: F) where F: FnOnce() +Send+'static{ + let receiver = receiver.0; + tokio::spawn(async move { + match res_symmetric_handle_loop(status_watch, receiver, udp, cur_info).await { + Ok(_) => {} + Err(e) => { + log::error!("{:?}",e) + } + } + stop_fn() + }); +} + +async fn res_symmetric_handle_loop( + mut status_watch: watch::Receiver, + mut receiver: Receiver, udp: UdpSocket, cur_info: CurrentDeviceInfo, ) -> Result<()> { - let receiver = receiver.0; let mut buf = [0u8; 12]; let mut packet = NetPacket::new(&mut buf)?; packet.set_version(Version::V1); @@ -182,64 +228,88 @@ pub fn res_symmetric_handle_loop( let mut punch_packet = PunchRequestPacket::new(packet.payload_mut())?; punch_packet.set_source(cur_info.virtual_ip); } + match tokio::time::timeout(Duration::from_secs(30), receiver.recv()).await { + Ok(_) => {} + Err(e) => {} + } loop { - match receiver.recv_timeout(Duration::from_secs(30)) { - Ok(punch) => { - let mut list = Vec::new(); - list.push(punch); - loop { - match receiver.try_recv() { - Ok(punch) => { + tokio::select! { + rs = tokio::time::timeout(Duration::from_secs(30), receiver.recv()) =>{ + match rs { + Ok(punch) => { + if let Some(punch) = punch{ + let mut list = Vec::new(); list.push(punch); - } - Err(_) => { - break; + loop { + match receiver.try_recv() { + Ok(punch) => { + list.push(punch); + } + Err(_) => { + break; + } + } + } + for punch in &list { + let dest = Ipv4Addr::from(punch.virtual_ip); + match punch.step.enum_value_or_default() { + Step::Step1 => { + STEP_MAP.insert(dest, Step::Step2); + } + Step::Step2 => { + STEP_MAP.insert(dest, Step::Step3); + } + Step::Step3 => { + STEP_MAP.insert(dest, Step::Step4); + } + Step::Step4 => { + STEP_MAP.insert(dest, Step::Step1); + } + } + } + if let Err(e) = handle(&status_watch,&udp, list, packet.buffer()) { + log::error!("{:?}",e) + } + }else { + return Err(Error::Stop("打洞线程通道关闭".to_string())); } } - } - for punch in &list { - let dest = Ipv4Addr::from(punch.virtual_ip); - match punch.step.enum_value_or_default() { - Step::Step1 => { - STEP_MAP.insert(dest, Step::Step2); - } - Step::Step2 => { - STEP_MAP.insert(dest, Step::Step3); - } - Step::Step3 => { - STEP_MAP.insert(dest, Step::Step4); - } - Step::Step4 => { - STEP_MAP.insert(dest, Step::Step1); - } + Err(_) => { + punch_request_handle(&udp, &cur_info)?; } } - if let Err(e) = handle(&udp, list, packet.buffer()) { - log::error!("{:?}",e) + } + status = status_watch.changed() =>{ + status?; + if *status_watch.borrow() != ApplicationStatus::Starting{ + return Ok(()) } } - Err(RecvTimeoutError::Timeout) => { - punch_request_handle(&udp, &cur_info)?; - } - Err(_) => { - return Err(Error::Stop("打洞线程通道关闭".to_string())); - } } } } /// 给锥形nat发送打洞数据包 -pub fn cone_handle_loop( - receiver: ConeReceiver, - udp: UdpSocket, - cur_info: CurrentDeviceInfo, -) -> Result<()> { +pub async fn cone_handler_start(status_watch: watch::Receiver, + receiver: ConeReceiver, + udp: UdpSocket, + cur_info: CurrentDeviceInfo, + stop_fn: F) where F: FnOnce()+Send +'static{ let receiver = receiver.0; - handle_loop(receiver, udp, cur_info) + tokio::spawn(async move { + match handle_loop(status_watch, receiver, udp, cur_info).await { + Ok(_) => {} + Err(e) => { + log::error!("{:?}",e) + } + } + stop_fn(); + }); } -pub fn handle_loop( - receiver: Receiver, +async fn handle_loop( + mut status_watch: watch::Receiver, + mut receiver: Receiver, udp: UdpSocket, cur_info: CurrentDeviceInfo, ) -> Result<()> { @@ -254,26 +324,33 @@ pub fn handle_loop( punch_packet.set_source(cur_info.virtual_ip); } loop { - match receiver.recv() { - Ok(punch) => { - let mut list = Vec::new(); - list.push(punch); - loop { - match receiver.try_recv() { - Ok(punch) => { - list.push(punch); + tokio::select! { + punch = receiver.recv() =>{ + if let Some(punch) = punch{ + let mut list = Vec::new(); + list.push(punch); + loop { + match receiver.try_recv() { + Ok(punch) => { + list.push(punch); + } + Err(_) => { + break; + } + } } - Err(_) => { - break; + if let Err(e) = handle(&status_watch,&udp, list, packet.buffer()) { + log::error!("{:?}",e) } - } - } - if let Err(e) = handle(&udp, list, packet.buffer()) { - log::error!("{:?}",e) - } + }else { + return Err(Error::Stop("打洞线程通道关闭".to_string())); + } } - Err(_) => { - return Err(Error::Stop("打洞线程通道关闭".to_string())); + status = status_watch.changed() =>{ + status?; + if *status_watch.borrow() != ApplicationStatus::Starting{ + return Ok(()) + } } } } @@ -282,15 +359,6 @@ pub fn handle_loop( fn select_sleep(counter: &mut u64) { *counter += 1; thread::sleep(Duration::from_millis(1)); - // if *counter > 1 { - // if cone_nat { - // thread::sleep(Duration::from_millis(2)); - // } else { - // if (*counter) & 10 == 10 { - // thread::sleep(Duration::from_millis(1)); - // } - // } - // } } diff --git a/src/handle/registration_handler.rs b/switch/src/handle/registration_handler.rs similarity index 93% rename from src/handle/registration_handler.rs rename to switch/src/handle/registration_handler.rs index 707f9e1..89b722c 100644 --- a/src/handle/registration_handler.rs +++ b/switch/src/handle/registration_handler.rs @@ -4,16 +4,19 @@ use std::sync::atomic::{AtomicI64, Ordering}; use std::time::Duration; use chrono::Local; +use crossbeam::atomic::AtomicCell; use parking_lot::RwLock; use protobuf::Message; use crate::error::*; +use crate::handle::ConnectStatus; use crate::proto::message::{RegistrationRequest, RegistrationResponse}; use crate::protocol::{error_packet, NetPacket, Protocol, service_packet, Version}; lazy_static::lazy_static! { static ref REQUEST:RwLock> = parking_lot::const_rwlock(None); static ref REGISTRATION_TIME:AtomicI64=AtomicI64::new(0); + pub(crate) static ref CONNECTION_STATUS:AtomicCell = AtomicCell::new(ConnectStatus::Connecting); } ///向中继服务器注册，token标识一个虚拟网关，mac_address防止多次注册时得到的ip不一致 @@ -56,6 +59,7 @@ pub fn registration( RegistrationResponse::parse_from_bytes(net_packet.payload())?; let _ = REQUEST.write().replace((token, mac_address)); udp.set_read_timeout(None)?; + CONNECTION_STATUS.store(ConnectStatus::Connected); return Ok(response); } _ => {} @@ -100,6 +104,7 @@ pub fn fast_registration(udp: &UdpSocket, server_address: SocketAddr) -> Result< //短时间不重复注册 return Ok(()); } + CONNECTION_STATUS.store(ConnectStatus::Connecting); let lock = REQUEST.read(); let option = lock.clone(); drop(lock); diff --git a/src/handle/tun_handler.rs b/switch/src/handle/tun_handler.rs similarity index 66% rename from src/handle/tun_handler.rs rename to switch/src/handle/tun_handler.rs index 7fd4e79..0be1828 100644 --- a/src/handle/tun_handler.rs +++ b/switch/src/handle/tun_handler.rs @@ -1,12 +1,17 @@ /// 接收tun数据，并且转发到udp上 -use std::net::{IpAddr, Ipv4Addr, UdpSocket}; +use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket}; +use std::os::fd::AsRawFd; +use std::thread; use chrono::Local; +use tokio::sync::watch; + use packet::icmp::icmp::IcmpPacket; use packet::icmp::Kind; use packet::ip::ipv4; use packet::ip::ipv4::packet::IpV4Packet; +use crate::ApplicationStatus; use crate::error::*; use crate::handle::{CurrentDeviceInfo, DIRECT_ROUTE_TABLE}; use crate::protocol::{NetPacket, Protocol, Version}; @@ -73,8 +78,9 @@ fn handle( if let Some(route) = DIRECT_ROUTE_TABLE.get(&dest_ip) { let current_time = Local::now().timestamp_millis(); if current_time - route.recv_time < 3_000 { - udp.send_to(&net_packet.buffer()[..(4 + 8 + data_len)], route.address)?; - return Ok(()); + if udp.send_to(&net_packet.buffer()[..(4 + 8 + data_len)], route.address).is_ok() { + return Ok(()); + } } } udp.send_to(&net_packet.buffer()[..(4 + 8 + data_len)], cur_info.connect_server)?; @@ -82,7 +88,28 @@ fn handle( } #[cfg(target_os = "windows")] -pub fn handle_loop( +pub async fn handler_start(mut status_watch: watch::Receiver, + udp: UdpSocket, + tun_reader: TunReader, + cur_info: CurrentDeviceInfo, stop_fn: F) + where F: FnOnce() + Send + 'static { + let session = tun_reader.0.clone(); + tokio::spawn(async move { + let _ = status_watch.changed().await; + session.shutdown(); + let udp = UdpSocket::bind("0.0.0.0:0").unwrap(); + let _ = udp.send_to(&[0],SocketAddr::new(IpAddr::V4(cur_info.virtual_gateway),10)); + }); + thread::spawn(move || { + if let Err(e) = handle_loop(udp, tun_reader, cur_info) { + log::error!("tun数据处理线程停止 {:?}",e); + } + stop_fn(); + }); +} + +#[cfg(target_os = "windows")] +fn handle_loop( udp: UdpSocket, tun_reader: TunReader, cur_info: CurrentDeviceInfo, @@ -103,7 +130,31 @@ pub fn handle_loop( } } -#[cfg(any(unix))] +#[cfg(any(target_os = "macos", target_os = "linux"))] +pub async fn handler_start(mut status_watch: watch::Receiver, + udp: UdpSocket, + tun_reader: TunReader, + cur_info: CurrentDeviceInfo, stop_fn: F) + where F: FnOnce() + Send + 'static { + let raw_fd = tun_reader.0.as_raw_fd(); + tokio::spawn(async move { + let _ = status_watch.changed().await; + // 让tun接收线程关闭 + unsafe { + libc::close(raw_fd); + } + let udp = UdpSocket::bind("0.0.0.0:0").unwrap(); + let _ = udp.send_to(&[0],SocketAddr::new(IpAddr::V4(cur_info.virtual_gateway),10)); + }); + thread::spawn(move || { + if let Err(e) = handle_loop(udp, tun_reader, cur_info) { + log::error!(" tun数据处理线程停止 {:?}",e); + } + stop_fn(); + }); +} + +#[cfg(any(target_os = "macos", target_os = "linux"))] pub fn handle_loop( udp: UdpSocket, mut tun_reader: TunReader, diff --git a/src/handle/udp_recv_handler.rs b/switch/src/handle/udp_recv_handler.rs similarity index 78% rename from src/handle/udp_recv_handler.rs rename to switch/src/handle/udp_recv_handler.rs index f9f86c1..75ab2f8 100644 --- a/src/handle/udp_recv_handler.rs +++ b/switch/src/handle/udp_recv_handler.rs @@ -1,18 +1,21 @@ -use std::net::{Ipv4Addr, SocketAddr, UdpSocket}; +use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket}; use std::sync::atomic::Ordering; +use std::thread; use chrono::Local; -use crossbeam::channel::{Receiver, Sender, TrySendError}; use packet::icmp::{icmp, Kind}; use packet::ip::ipv4; use packet::ip::ipv4::packet::IpV4Packet; use protobuf::Message; +use tokio::sync::mpsc::{Receiver, Sender}; +use tokio::sync::mpsc::error::TrySendError; +use tokio::sync::watch; -use crate::CurrentDeviceInfo; +use crate::{ApplicationStatus, CurrentDeviceInfo}; use crate::error::*; -use crate::handle::{ADDR_TABLE, DEVICE_LIST, DIRECT_ROUTE_TABLE, NAT_INFO, Route, SERVER_RT}; +use crate::handle::{ADDR_TABLE, ConnectStatus, DEVICE_LIST, DIRECT_ROUTE_TABLE, NAT_INFO, Route, SERVER_RT}; use crate::handle::punch_handler::PunchSender; -use crate::handle::registration_handler::fast_registration; +use crate::handle::registration_handler::{CONNECTION_STATUS, fast_registration}; use crate::proto::message::{DeviceList, Punch, RegistrationResponse}; use crate::protocol::{control_packet, NetPacket, Protocol, service_packet, turn_packet, Version}; use crate::protocol::control_packet::{ControlPacket, PunchResponsePacket}; @@ -20,7 +23,42 @@ use crate::protocol::error_packet::InErrorPacket; use crate::protocol::turn_packet::TurnPacket; use crate::tun_device::TunWriter; -pub fn recv_loop( +const UDP_STOP_BUF: [u8; 1] = [0u8]; + +pub async fn udp_recv_start( + mut status_watch: watch::Receiver, + udp: UdpSocket, + server_addr: SocketAddr, + other_sender: Sender<(SocketAddr, Vec)>, + mut tun_writer: TunWriter, + current_device: CurrentDeviceInfo, + stop_fn: F) + where F: FnOnce() + Send + 'static { + { + let udp = udp.try_clone().unwrap(); + tokio::spawn(async move { + let _ = status_watch.changed().await; + let mut addr = udp.local_addr().unwrap(); + addr.set_ip(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))); + udp.send_to(&UDP_STOP_BUF, addr).unwrap(); + }); + } + + thread::spawn(move || { + if let Err(e) = recv_loop( + udp, + server_addr, + other_sender, + tun_writer, + current_device, + ) { + log::error!("udp数据处理线程停止 {:?}",e); + } + stop_fn(); + }); +} + +fn recv_loop( udp: UdpSocket, server_addr: SocketAddr, other_sender: Sender<(SocketAddr, Vec)>, @@ -29,11 +67,14 @@ pub fn recv_loop( ) -> Result<()> { let mut buf = [0u8; 65536]; let mut local_addr = udp.local_addr()?; - local_addr.set_ip(std::net::IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))); + local_addr.set_ip(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))); loop { match udp.recv_from(&mut buf) { Ok((len, addr)) => { if addr == local_addr { + if len == 1 && &buf[..len] == &UDP_STOP_BUF { + return Ok(()); + } //本地的包直接再发到网卡，这个主要用于处理当前虚拟ip的icmp ping if let Ok(ip) = IpV4Packet::new(&buf[..len]) { if ip.destination_ip() == current_device.virtual_ip { @@ -55,12 +96,13 @@ pub fn recv_loop( Err(Error::Stop(str)) => { return Err(Error::Stop(str)); } - Err(_) => {} + Err(e) => { + log::error!("{:?}",e); + } } } Err(e) => { log::error!("{:?}",e); - // println!("{:?}", e); } }; } @@ -112,8 +154,8 @@ fn recv_handle( let v = net_packet.buffer().to_vec(); match other_sender.try_send((recv_addr, v)) { Ok(_) => {} - Err(TrySendError::Disconnected(_)) => { - return Err(Error::Stop("处理线程停止".to_string())); + Err(TrySendError::Closed(_)) => { + return Err(Error::Stop("子处理线程停止".to_string())); } Err(e) => { log::error!("子线程处理 {:?}",e); @@ -124,21 +166,50 @@ fn recv_handle( Ok(()) } -pub fn other_loop( +pub async fn udp_other_recv_start(status_watch: watch::Receiver, + udp: UdpSocket, + receiver: Receiver<(SocketAddr, Vec)>, + current_device: CurrentDeviceInfo, + sender: PunchSender, + stop_fn: F) where F: FnOnce() + Send + 'static { + tokio::spawn(async move { + match other_loop(status_watch, udp, receiver, current_device, sender).await { + Ok(_) => {} + Err(e) => { + log::error!("{:?}",e); + } + } + stop_fn(); + }); +} + +async fn other_loop( + mut status_watch: watch::Receiver, udp: UdpSocket, - receiver: Receiver<(SocketAddr, Vec)>, + mut receiver: Receiver<(SocketAddr, Vec)>, current_device: CurrentDeviceInfo, sender: PunchSender, ) -> Result<()> { loop { - let (peer_addr, buf) = receiver.recv()?; - match other_handle(&udp, buf, peer_addr, ¤t_device, &sender) { - Ok(_) => {} - Err(Error::Stop(str)) => { - return Err(Error::Stop(str)); + tokio::select! { + rs = receiver.recv()=>{ + if let Some((peer_addr, buf)) = rs { + match other_handle(&udp, buf, peer_addr, ¤t_device, &sender) { + Ok(_) => {} + Err(Error::Stop(str)) => { + return Err(Error::Stop(str)); + } + Err(e) => { + log::error!("other_loop {:?}",e); + } + } + } } - Err(e) => { - log::error!("other_loop {:?}",e); + status = status_watch.changed() =>{ + status?; + if *status_watch.borrow() != ApplicationStatus::Starting{ + return Ok(()) + } } } } @@ -163,6 +234,7 @@ fn other_handle( service_packet::Protocol::RegistrationResponse => { let response = RegistrationResponse::parse_from_bytes(net_packet.payload())?; crate::handle::init_nat_info(response.public_ip, response.public_port as u16); + CONNECTION_STATUS.store(ConnectStatus::Connected); //todo 重连之后ip可能会发生改变(目前2分钟内未重连则会释放ip)，需要更新本地ip(或者保证重连ip不变) } service_packet::Protocol::UpdateDeviceList => { @@ -215,7 +287,7 @@ fn other_handle( //其他设备 if let Some(virtual_ip) = ADDR_TABLE.get(&peer_addr) { if let Some(mut info) = DIRECT_ROUTE_TABLE.get_mut(&virtual_ip) { - info.delay = rt; + info.rt = rt; info.recv_time = current_time; } } diff --git a/switch/src/lib.rs b/switch/src/lib.rs new file mode 100644 index 0000000..2055a42 --- /dev/null +++ b/switch/src/lib.rs @@ -0,0 +1,211 @@ +use std::io; +use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4, ToSocketAddrs, UdpSocket}; +use std::sync::atomic::Ordering; + +use crossbeam::sync::WaitGroup; +use tokio::sync::watch; + +use error::*; + +use crate::handle::{ApplicationStatus, ConnectStatus, CurrentDeviceInfo, DEVICE_LIST, DIRECT_ROUTE_TABLE, Route, RouteType, SERVER_RT}; +use crate::handle::registration_handler::CONNECTION_STATUS; + +pub mod tun_device; +pub mod nat; +pub mod error; +pub mod handle; +pub mod proto; +pub mod protocol; + +#[derive(Clone, Debug)] +pub struct Config { + pub token: String, + pub mac_address: String, +} + +impl Config { + pub fn new(token: String, mac_address: String) -> Self { + Self { + token, + mac_address, + } + } +} + +pub struct Switch { + current_device: CurrentDeviceInfo, + status_sender: watch::Sender, + wait_group: WaitGroup, + runtime: Option, +} + +impl Switch { + pub fn start(config: Config) -> Result { + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .unwrap(); + return match runtime.block_on(Switch::start_(config.token, config.mac_address)) { + Ok(mut switch) => { + switch.runtime = Some(runtime); + Ok(switch) + } + Err(e) => { + Err(e) + } + }; + } + pub fn stop(self) { + let _ = self.status_sender.send(ApplicationStatus::Stopping); + self.wait_group.wait(); + } + pub fn current_device(&self) -> &CurrentDeviceInfo { + &self.current_device + } + pub fn server_rt(&self) -> i64 { + SERVER_RT.load(Ordering::Relaxed) + } + pub fn connection_status(&self) -> ConnectStatus { + CONNECTION_STATUS.load() + } + pub fn device_list(&self) -> Vec { + let device_list_lock = DEVICE_LIST.lock(); + let (_epoch, device_list) = device_list_lock.clone(); + drop(device_list_lock); + device_list + } + pub fn route(&self, ip: &Ipv4Addr) -> Route { + if let Some(route_ref) = DIRECT_ROUTE_TABLE.get(ip) { + route_ref.value().clone() + } else { + let mut route = Route::new(self.current_device.connect_server); + route.route_type = RouteType::ServerRelay; + route.rt = self.server_rt() * 2; + route.recv_time = -1; + route + } + } +} + +impl Switch { + pub async fn start_(token: String, mac_address: String) -> Result { + let server_address = "nat1.wherewego.top:29876".to_socket_addrs().unwrap().next().unwrap(); + let mut port = 101 as u16; + let udp = loop { + match UdpSocket::bind(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::from(0), port))) { + Ok(udp) => { + break udp; + } + Err(e) => { + if e.kind() == io::ErrorKind::AddrInUse { + port += 1; + } else { + log::error!("创建udp失败 {:?}",e); + return Err(Error::Stop("udp bind error".to_string())); + } + } + } + }; + //注册 + let response = handle::registration_handler::registration(&udp, server_address, token, mac_address)?; + { + let ip_list = response + .virtual_ip_list + .iter() + .map(|ip| Ipv4Addr::from(*ip)) + .collect(); + let mut dev = DEVICE_LIST.lock(); + dev.0 = response.epoch; + dev.1 = ip_list; + } + let virtual_ip = Ipv4Addr::from(response.virtual_ip); + let virtual_gateway = Ipv4Addr::from(response.virtual_gateway); + let virtual_netmask = Ipv4Addr::from(response.virtual_netmask); + let (status_sender, status_receiver) = tokio::sync::watch::channel(ApplicationStatus::Starting); + let current_device = CurrentDeviceInfo::new(virtual_ip, virtual_gateway, virtual_netmask, server_address); + let wait_group = WaitGroup::new(); + //心跳线程 + { + let udp = udp.try_clone()?; + let wait_group1 = wait_group.clone(); + handle::heartbeat_handler::start(status_receiver.clone(), udp, current_device, || { + drop(wait_group1); + }).await; + } + //初始化nat数据 + handle::init_nat_info(response.public_ip, response.public_port as u16); + // tun服务 + let (tun_writer, tun_reader) = + tun_device::create_tun(virtual_ip, virtual_netmask, virtual_gateway)?; + // 打洞数据通道 + let (punch_sender, cone_receiver, req_symmetric_receiver, res_symmetric_receiver) = handle::punch_handler::bounded(); + //udp数据处理 + { + // 低优先级的udp数据通道 + let (sender, receiver) = tokio::sync::mpsc::channel(50); + let udp1 = udp.try_clone()?; + let wait_group1 = wait_group.clone(); + handle::udp_recv_handler::udp_recv_start( + status_receiver.clone(), + udp1, + server_address, + sender, + tun_writer, + current_device, + || { + drop(wait_group1); + }, + ).await; + let udp1 = udp.try_clone()?; + let wait_group1 = wait_group.clone(); + handle::udp_recv_handler::udp_other_recv_start(status_receiver.clone(), udp1, + receiver, current_device, punch_sender, + || { + drop(wait_group1); + }).await; + } + //打洞处理 + { + let udp1 = udp.try_clone()?; + let wait_group1 = wait_group.clone(); + handle::punch_handler::cone_handler_start(status_receiver.clone(), + cone_receiver, udp1, + current_device, + || { + drop(wait_group1); + }).await; + let udp1 = udp.try_clone()?; + let wait_group1 = wait_group.clone(); + handle::punch_handler::req_symmetric_handler_start(status_receiver.clone(), + req_symmetric_receiver, udp1, + current_device, + || { + drop(wait_group1); + }).await; + let udp1 = udp.try_clone()?; + let wait_group1 = wait_group.clone(); + handle::punch_handler::res_symmetric_handler_start(status_receiver.clone(), + res_symmetric_receiver, + udp1, + current_device, + || { + drop(wait_group1); + }).await; + } + //tun数据处理 + { + let wait_group1 = wait_group.clone(); + handle::tun_handler::handler_start(status_receiver.clone(), udp, + tun_reader, current_device, + || { + drop(wait_group1); + }).await; + } + Ok(Switch { + current_device, + status_sender, + wait_group, + runtime: None, + }) + } +} \ No newline at end of file diff --git a/src/nat/check.rs b/switch/src/nat/check.rs similarity index 100% rename from src/nat/check.rs rename to switch/src/nat/check.rs diff --git a/src/nat/mod.rs b/switch/src/nat/mod.rs similarity index 100% rename from src/nat/mod.rs rename to switch/src/nat/mod.rs diff --git a/src/proto/message.rs b/switch/src/proto/message.rs similarity index 100% rename from src/proto/message.rs rename to switch/src/proto/message.rs diff --git a/src/proto/mod.rs b/switch/src/proto/mod.rs similarity index 100% rename from src/proto/mod.rs rename to switch/src/proto/mod.rs diff --git a/src/protocol/control_packet.rs b/switch/src/protocol/control_packet.rs similarity index 100% rename from src/protocol/control_packet.rs rename to switch/src/protocol/control_packet.rs diff --git a/src/protocol/error_packet.rs b/switch/src/protocol/error_packet.rs similarity index 100% rename from src/protocol/error_packet.rs rename to switch/src/protocol/error_packet.rs diff --git a/src/protocol/mod.rs b/switch/src/protocol/mod.rs similarity index 100% rename from src/protocol/mod.rs rename to switch/src/protocol/mod.rs diff --git a/src/protocol/service_packet.rs b/switch/src/protocol/service_packet.rs similarity index 100% rename from src/protocol/service_packet.rs rename to switch/src/protocol/service_packet.rs diff --git a/src/protocol/turn_packet.rs b/switch/src/protocol/turn_packet.rs similarity index 100% rename from src/protocol/turn_packet.rs rename to switch/src/protocol/turn_packet.rs diff --git a/src/tun_device/linux.rs b/switch/src/tun_device/linux.rs similarity index 100% rename from src/tun_device/linux.rs rename to switch/src/tun_device/linux.rs diff --git a/src/tun_device/mac.rs b/switch/src/tun_device/mac.rs similarity index 100% rename from src/tun_device/mac.rs rename to switch/src/tun_device/mac.rs diff --git a/src/tun_device/mod.rs b/switch/src/tun_device/mod.rs similarity index 78% rename from src/tun_device/mod.rs rename to switch/src/tun_device/mod.rs index 1c61abd..6f36228 100644 --- a/src/tun_device/mod.rs +++ b/switch/src/tun_device/mod.rs @@ -1,4 +1,4 @@ -#[cfg(target_os = "linux")] +#[cfg(any(target_os = "linux",target_os = "android"))] pub use linux::create_tun; #[cfg(target_os = "macos")] pub use mac::create_tun; @@ -11,7 +11,7 @@ pub use windows::create_tun; #[cfg(target_os = "macos")] pub mod mac; -#[cfg(target_os = "linux")] +#[cfg(any(target_os = "linux",target_os = "android"))] pub mod linux; #[cfg(any(unix))] pub mod unix; diff --git a/src/tun_device/unix.rs b/switch/src/tun_device/unix.rs similarity index 100% rename from src/tun_device/unix.rs rename to switch/src/tun_device/unix.rs diff --git a/src/tun_device/windows.rs b/switch/src/tun_device/windows.rs similarity index 92% rename from src/tun_device/windows.rs rename to switch/src/tun_device/windows.rs index 8f753e5..39e87b5 100644 --- a/src/tun_device/windows.rs +++ b/switch/src/tun_device/windows.rs @@ -23,7 +23,7 @@ impl TunWriter { } } -pub struct TunReader(Arc); +pub struct TunReader(pub(crate) Arc); impl TunReader { pub fn next(&self) -> io::Result { @@ -52,14 +52,13 @@ pub fn create_tun( }, Err(e) => { log::error!("wintun.dll not found"); - println!("{}", console::style("wintun.dll not found").red()); - return Err(Error::Stop(format!("{:?}", e))); + return Err(Error::Stop(format!("wintun.dll not found {:?}", e))); } } }; - let adapter = match Adapter::open(&win_tun, "Demo") { + let adapter = match Adapter::open(&win_tun, "Switch") { Ok(a) => a, - Err(_) => match Adapter::create(&win_tun, "Example", "Demo", None) { + Err(_) => match Adapter::create(&win_tun, "Switch", "Switch", None) { Ok(adapter) => adapter, Err(e) => return Err(Error::Stop(format!("{:?}", e))),