From c655d9650bcf6317962014ebbbc2480350eefc8b Mon Sep 17 00:00:00 2001 From: lubeilin <1791778603@qq.com> Date: Mon, 9 Jan 2023 20:26:25 +0800 Subject: [PATCH] =?UTF-8?q?=E6=A0=BC=E5=BC=8F=E5=8C=96=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- switch-desktop/src/main.rs | 51 ++++++-- switch-desktop/src/windows_admin_check.rs | 2 +- switch-jni/src/lib.rs | 115 ++++++++++------- switch/packet/src/lib.rs | 1 - switch/rust-tun/src/platform/linux/device.rs | 8 +- switch/src/handle/heartbeat_handler.rs | 23 ++-- switch/src/handle/mod.rs | 23 ++-- switch/src/handle/punch_handler.rs | 117 ++++++++++------- switch/src/handle/registration_handler.rs | 6 +- switch/src/handle/tun_handler.rs | 65 ++++++---- switch/src/handle/udp_recv_handler.rs | 75 ++++++----- switch/src/lib.rs | 126 ++++++++++++------- switch/src/nat/check.rs | 4 +- switch/src/nat/mod.rs | 2 +- switch/src/protocol/control_packet.rs | 1 - switch/src/tun_device/linux.rs | 3 +- switch/src/tun_device/mac.rs | 13 +- switch/src/tun_device/mod.rs | 10 +- 18 files changed, 398 insertions(+), 247 deletions(-) diff --git a/switch-desktop/src/main.rs b/switch-desktop/src/main.rs index b2648b1..b52589e 100644 --- a/switch-desktop/src/main.rs +++ b/switch-desktop/src/main.rs @@ -1,14 +1,18 @@ use clap::Parser; use console::style; -use switch::*; use switch::handle::RouteType; +use switch::*; #[cfg(windows)] mod windows_admin_check; #[derive(Parser, Debug)] -#[command(author = "Lu Beilin", version, about = "一个虚拟网络工具,启动后会获取一个ip,相同token下的设备之间可以用ip直接通信")] +#[command( + author = "Lu Beilin", + version, + about = "一个虚拟网络工具,启动后会获取一个ip,相同token下的设备之间可以用ip直接通信" +)] struct Args { /// 32位字符 /// 相同token的设备之间才能通信。 @@ -27,7 +31,9 @@ fn log_init() { } 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"))) + .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() @@ -52,7 +58,9 @@ fn main() { .ok() .and_then(|p| p.to_str().map(|p| p.to_string())) { - let _ = runas::Command::new(&absolute_path).args(&args[1..]).status() + let _ = runas::Command::new(&absolute_path) + .args(&args[1..]) + .status() .expect("failed to execute"); } else { panic!("failed to execute") @@ -72,10 +80,19 @@ fn main() { 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()); + 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)); + 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() { @@ -128,16 +145,28 @@ fn command(cmd: &str, switch: &Switch) -> Result<(), ()> { 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()); + 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!( + "{} , 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()); } diff --git a/switch-desktop/src/windows_admin_check.rs b/switch-desktop/src/windows_admin_check.rs index c818547..c3531bf 100644 --- a/switch-desktop/src/windows_admin_check.rs +++ b/switch-desktop/src/windows_admin_check.rs @@ -5,7 +5,7 @@ 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 winapi::um::winnt::{TokenElevation, HANDLE, TOKEN_ELEVATION, TOKEN_QUERY}; // Use std::io::Error::last_os_error for errors. // NOTE: For this example I'm simple passing on the OS error. diff --git a/switch-jni/src/lib.rs b/switch-jni/src/lib.rs index d11572f..ec77bb7 100644 --- a/switch-jni/src/lib.rs +++ b/switch-jni/src/lib.rs @@ -2,26 +2,26 @@ 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 jni::JNIEnv; -use switch::{Config, Switch}; use switch::handle::{CurrentDeviceInfo, Route}; +use switch::{Config, Switch}; 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"); + 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())) - } + Ok(value) => Ok(Some(value.to_string())), Err(_) => { - env.throw_new("Ljava/lang/RuntimeException", "not utf-8").expect("throw"); + env.throw_new("Ljava/lang/RuntimeException", "not utf-8") + .expect("throw"); Ok(None) } } @@ -35,7 +35,11 @@ fn start(env: &JNIEnv, config: JObject) -> Result, Error> { return Ok(Some(switch)); } Err(e) => { - env.throw_new("Ljava/lang/RuntimeException", format!("switch start failed {:?}", e)).expect("throw"); + env.throw_new( + "Ljava/lang/RuntimeException", + format!("switch start failed {:?}", e), + ) + .expect("throw"); } } } @@ -44,7 +48,11 @@ fn start(env: &JNIEnv, config: JObject) -> Result, Error> { } #[no_mangle] -pub unsafe extern "C" fn Java_org_switches_jni_Switch_start0(env: JNIEnv, _class: JClass, config: JObject) -> jlong { +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 { @@ -57,61 +65,74 @@ pub unsafe extern "C" fn Java_org_switches_jni_Switch_start0(env: JNIEnv, _class } #[no_mangle] -pub unsafe extern "C" fn Java_org_switches_jni_Switch_stop0(env: JNIEnv, _class: JClass, raw_switch: jlong) { +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 { +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() - } + 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 { +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() - } + 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 { +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() - } + 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 { +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 { +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 @@ -133,10 +154,13 @@ fn device_list(env: &JNIEnv, device_list: Vec) -> Result = device_list.iter().map(|ip| { - let ip: u32 = (*ip).into(); - ip as jint - }).collect(); + 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) } @@ -148,9 +172,7 @@ fn current_device(env: &JNIEnv, dev_info: &CurrentDeviceInfo) -> Result { - ip.into() - } + IpAddr::V4(ip) => ip.into(), IpAddr::V6(_) => { panic!() } @@ -159,10 +181,15 @@ fn current_device(env: &JNIEnv, dev_info: &CurrentDeviceInfo) -> Result u32 { ((x as u32) << 8) | y as u32 } - #[cfg(test)] mod tests { use super::*; diff --git a/switch/rust-tun/src/platform/linux/device.rs b/switch/rust-tun/src/platform/linux/device.rs index 60b34a4..1b078ae 100644 --- a/switch/rust-tun/src/platform/linux/device.rs +++ b/switch/rust-tun/src/platform/linux/device.rs @@ -77,10 +77,10 @@ impl Device { req.ifru.flags = device_type | if config.platform.packet_information { - 0 - } else { - IFF_NO_PI - } + 0 + } else { + IFF_NO_PI + } | if queues_num > 1 { IFF_MULTI_QUEUE } else { 0 }; for _ in 0..queues_num { diff --git a/switch/src/handle/heartbeat_handler.rs b/switch/src/handle/heartbeat_handler.rs index 9e506c7..73e58b1 100644 --- a/switch/src/handle/heartbeat_handler.rs +++ b/switch/src/handle/heartbeat_handler.rs @@ -5,27 +5,36 @@ use chrono::Local; use tokio::sync::watch::Receiver; use tokio::time::sleep; -use crate::{CurrentDeviceInfo, DEVICE_LIST}; use crate::error::*; use crate::handle::{ApplicationStatus, DIRECT_ROUTE_TABLE}; -use crate::protocol::{control_packet, NetPacket, Protocol, Version}; use crate::protocol::control_packet::PingPacket; +use crate::protocol::{control_packet, NetPacket, Protocol, Version}; +use crate::{CurrentDeviceInfo, DEVICE_LIST}; -pub async fn start(status_watch: Receiver, - udp: UdpSocket, cur_info: CurrentDeviceInfo, stop_fn: F) - where F: FnOnce() + Send + 'static { +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) + log::error!("{:?}", e) } } stop_fn(); }); } -async fn handle_loop(mut status_watch: Receiver, udp: UdpSocket, server_addr: SocketAddr) -> Result<()> { +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)]; diff --git a/switch/src/handle/mod.rs b/switch/src/handle/mod.rs index 674bcff..164cbf4 100644 --- a/switch/src/handle/mod.rs +++ b/switch/src/handle/mod.rs @@ -62,10 +62,12 @@ pub struct NatInfo { } impl NatInfo { - pub fn new(public_ips: Vec, - public_port: u16, - public_port_range: u16, - nat_type: NatType, ) -> Self { + pub fn new( + public_ips: Vec, + public_port: u16, + public_port_range: u16, + nat_type: NatType, + ) -> Self { Self { public_ips, public_port, @@ -87,9 +89,7 @@ pub fn init_nat_info(public_ip: u32, public_port: u16) { public_ips.push(ip); } } - let nat_info = NatInfo::new(public_ips, - public_port, - port_range, nat_type); + let nat_info = NatInfo::new(public_ips, public_port, port_range, nat_type); // println!("nat信息:{:?}",nat_info); let mut nat_info_lock = NAT_INFO.lock(); nat_info_lock.replace(nat_info); @@ -114,7 +114,12 @@ pub struct CurrentDeviceInfo { } impl CurrentDeviceInfo { - pub fn new(virtual_ip: Ipv4Addr, virtual_gateway: Ipv4Addr, virtual_netmask: Ipv4Addr, connect_server: SocketAddr) -> Self { + pub fn new( + virtual_ip: Ipv4Addr, + virtual_gateway: Ipv4Addr, + virtual_netmask: Ipv4Addr, + connect_server: SocketAddr, + ) -> Self { let broadcast_address = (!u32::from_be_bytes(virtual_netmask.octets())) | u32::from_be_bytes(virtual_gateway.octets()); let broadcast_address = Ipv4Addr::from(broadcast_address); @@ -152,7 +157,7 @@ impl Into for RouteType { fn into(self) -> u8 { match self { RouteType::ServerRelay => 0, - RouteType::P2P => 1 + RouteType::P2P => 1, } } } diff --git a/switch/src/handle/punch_handler.rs b/switch/src/handle/punch_handler.rs index 5afa1d2..67b7107 100644 --- a/switch/src/handle/punch_handler.rs +++ b/switch/src/handle/punch_handler.rs @@ -5,29 +5,37 @@ use std::time::Duration; 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::mpsc::{Receiver, Sender}; use tokio::sync::watch; -use crate::{CurrentDeviceInfo, DEVICE_LIST, handle::NAT_INFO, handle::NatInfo}; use crate::error::*; 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; use crate::protocol::turn_packet::TurnPacket; +use crate::protocol::{control_packet, turn_packet, NetPacket, Protocol, Version}; +use crate::{handle::NatInfo, handle::NAT_INFO, CurrentDeviceInfo, DEVICE_LIST}; lazy_static! { - pub static ref STEP_MAP:DashMap = DashMap::new(); + pub static ref STEP_MAP: DashMap = DashMap::new(); } /// 每一种类型一个通道，减少相互干扰 -pub fn bounded() -> (PunchSender, ConeReceiver, ReqSymmetricReceiver, ResSymmetricReceiver) { +pub fn bounded() -> ( + PunchSender, + ConeReceiver, + ReqSymmetricReceiver, + ResSymmetricReceiver, +) { 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)) + ( + PunchSender::new(cone_sender, req_symmetric_sender, res_symmetric_sender), + ConeReceiver(cone_receiver), + ReqSymmetricReceiver(req_symmetric_receiver), + ResSymmetricReceiver(res_symmetric_receiver), + ) } pub struct ConeReceiver(Receiver); @@ -44,9 +52,11 @@ pub struct PunchSender { } impl PunchSender { - pub fn new(cone_sender: Sender, - req_symmetric_sender: Sender, - res_symmetric_sender: Sender, ) -> Self { + pub fn new( + cone_sender: Sender, + req_symmetric_sender: Sender, + res_symmetric_sender: Sender, + ) -> Self { Self { cone_sender, req_symmetric_sender, @@ -78,14 +88,17 @@ impl PunchSender { self.req_symmetric_sender.try_send(punch) } } - NatType::Cone => { - self.cone_sender.try_send(punch) - } + NatType::Cone => self.cone_sender.try_send(punch), } } } -fn handle(status_watch: &watch::Receiver, 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); @@ -107,7 +120,8 @@ fn handle(status_watch: &watch::Receiver, udp: &UdpSocket, pu } } let right_port = ((punch.public_port + range) & 0xFFFF) as u16; - let left_port = ((0xFFFF + punch.public_port - range) & 0xFFFF) as u16; + let left_port = + ((0xFFFF + punch.public_port - range) & 0xFFFF) as u16; if right_port != 0 { // println!("{:?}", SocketAddr::V4(SocketAddrV4::new(pub_ip, right_port))); udp.send_to( @@ -140,10 +154,7 @@ fn handle(status_watch: &watch::Receiver, udp: &UdpSocket, pu return Ok(()); } } - udp.send_to( - buf, - SocketAddr::V4(SocketAddrV4::new(pub_ip, port)), - )?; + udp.send_to(buf, SocketAddr::V4(SocketAddrV4::new(pub_ip, port)))?; select_sleep(&mut counter); } } @@ -168,17 +179,21 @@ fn handle(status_watch: &watch::Receiver, udp: &UdpSocket, pu } /// 给对称nat发送打洞数据包 -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{ +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; tokio::spawn(async move { match handle_loop(status_watch, receiver, udp, cur_info).await { Ok(_) => {} Err(e) => { - log::error!("{:?}",e) + log::error!("{:?}", e) } } stop_fn() @@ -195,17 +210,21 @@ pub async fn req_symmetric_handler_start(status_watch: watch::Receiver(status_watch: watch::Receiver, - receiver: ResSymmetricReceiver, - udp: UdpSocket, - cur_info: CurrentDeviceInfo, - stop_fn: F) where F: FnOnce() +Send+'static{ +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) + log::error!("{:?}", e) } } stop_fn() @@ -290,17 +309,21 @@ async fn res_symmetric_handle_loop( } /// 给锥形nat发送打洞数据包 -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{ +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; tokio::spawn(async move { match handle_loop(status_watch, receiver, udp, cur_info).await { Ok(_) => {} Err(e) => { - log::error!("{:?}",e) + log::error!("{:?}", e) } } stop_fn(); @@ -361,16 +384,13 @@ fn select_sleep(counter: &mut u64) { thread::sleep(Duration::from_millis(1)); } - fn punch_request_handle(udp: &UdpSocket, cur_info: &CurrentDeviceInfo) -> Result<()> { let nat_info_lock = NAT_INFO.lock(); let nat_info = nat_info_lock.clone(); drop(nat_info_lock); if let Some(nat_info) = nat_info { - if let Err(e) = send_punch(&udp, - &cur_info, - nat_info) { - log::error!("发送打洞数据失败 {:?}",e) + if let Err(e) = send_punch(&udp, &cur_info, nat_info) { + log::error!("发送打洞数据失败 {:?}", e) } Ok(()) } else { @@ -378,7 +398,6 @@ fn punch_request_handle(udp: &UdpSocket, cur_info: &CurrentDeviceInfo) -> Result } } - fn send_punch(udp: &UdpSocket, cur_info: &CurrentDeviceInfo, nat_info: NatInfo) -> Result<()> { let lock = DEVICE_LIST.lock(); let list = lock.1.clone(); @@ -391,15 +410,19 @@ fn send_punch(udp: &UdpSocket, cur_info: &CurrentDeviceInfo, nat_info: NatInfo) } else { Step::Step1 }; - let bytes = punch_packet(cur_info.virtual_ip, - nat_info.clone(), ip, step)?; + let bytes = punch_packet(cur_info.virtual_ip, nat_info.clone(), ip, step)?; udp.send_to(&bytes, cur_info.connect_server)?; } } Ok(()) } -fn punch_packet(virtual_ip: Ipv4Addr, nat_info: NatInfo, dest: Ipv4Addr, step: Step) -> Result> { +fn punch_packet( + virtual_ip: Ipv4Addr, + nat_info: NatInfo, + dest: Ipv4Addr, + step: Step, +) -> Result> { let mut punch_reply = Punch::new(); punch_reply.reply = false; punch_reply.virtual_ip = u32::from_be_bytes(virtual_ip.octets()); diff --git a/switch/src/handle/registration_handler.rs b/switch/src/handle/registration_handler.rs index 89b722c..0c3799d 100644 --- a/switch/src/handle/registration_handler.rs +++ b/switch/src/handle/registration_handler.rs @@ -11,7 +11,7 @@ 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}; +use crate::protocol::{error_packet, service_packet, NetPacket, Protocol, Version}; lazy_static::lazy_static! { static ref REQUEST:RwLock> = parking_lot::const_rwlock(None); @@ -98,8 +98,8 @@ pub fn fast_registration(udp: &UdpSocket, server_address: SocketAddr) -> Result< let new = Local::now().timestamp_millis(); if new - last < 2000 || REGISTRATION_TIME - .compare_exchange(last, new, Ordering::Relaxed, Ordering::Relaxed) - .is_err() + .compare_exchange(last, new, Ordering::Relaxed, Ordering::Relaxed) + .is_err() { //短时间不重复注册 return Ok(()); diff --git a/switch/src/handle/tun_handler.rs b/switch/src/handle/tun_handler.rs index 341290f..d5b2e5e 100644 --- a/switch/src/handle/tun_handler.rs +++ b/switch/src/handle/tun_handler.rs @@ -10,12 +10,12 @@ 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}; use crate::protocol::turn_packet::TurnPacket; +use crate::protocol::{NetPacket, Protocol, Version}; use crate::tun_device::TunReader; +use crate::ApplicationStatus; /// 是否在一个网段 fn check_dest(dest: Ipv4Addr, cur_info: &CurrentDeviceInfo) -> bool { @@ -77,42 +77,51 @@ 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 { - if udp.send_to(&net_packet.buffer()[..(4 + 8 + data_len)], route.address).is_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)?; + udp.send_to( + &net_packet.buffer()[..(4 + 8 + data_len)], + cur_info.connect_server, + )?; return Ok(()); } #[cfg(target_os = "windows")] -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 { +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)); + 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); + log::error!("tun数据处理线程停止 {:?}", e); } stop_fn(); }); } #[cfg(target_os = "windows")] -fn handle_loop( - udp: UdpSocket, - tun_reader: TunReader, - cur_info: CurrentDeviceInfo, -) -> Result<()> { +fn handle_loop(udp: UdpSocket, tun_reader: TunReader, cur_info: CurrentDeviceInfo) -> Result<()> { let mut net_packet = NetPacket::new(vec![0u8; 4 + 8 + 1500])?; net_packet.set_version(Version::V1); net_packet.set_protocol(Protocol::Ipv4Turn); @@ -130,11 +139,16 @@ fn handle_loop( } #[cfg(any(target_os = "macos", target_os = "linux", target_os = "android"))] -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 { +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, +{ + use std::os::fd::AsRawFd; let raw_fd = tun_reader.0.as_raw_fd(); tokio::spawn(async move { let _ = status_watch.changed().await; @@ -143,11 +157,14 @@ pub async fn handler_start(mut status_watch: watch::Receiver {} Err(e) => { - log::error!("{:?}",e) + log::error!("{:?}", e) } } } diff --git a/switch/src/handle/udp_recv_handler.rs b/switch/src/handle/udp_recv_handler.rs index 75ab2f8..832e66a 100644 --- a/switch/src/handle/udp_recv_handler.rs +++ b/switch/src/handle/udp_recv_handler.rs @@ -7,21 +7,23 @@ 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::mpsc::{Receiver, Sender}; use tokio::sync::watch; -use crate::{ApplicationStatus, CurrentDeviceInfo}; use crate::error::*; -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::{CONNECTION_STATUS, fast_registration}; +use crate::handle::registration_handler::{fast_registration, CONNECTION_STATUS}; +use crate::handle::{ + ConnectStatus, Route, ADDR_TABLE, DEVICE_LIST, DIRECT_ROUTE_TABLE, NAT_INFO, SERVER_RT, +}; 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}; use crate::protocol::error_packet::InErrorPacket; use crate::protocol::turn_packet::TurnPacket; +use crate::protocol::{control_packet, service_packet, turn_packet, NetPacket, Protocol, Version}; use crate::tun_device::TunWriter; +use crate::{ApplicationStatus, CurrentDeviceInfo}; const UDP_STOP_BUF: [u8; 1] = [0u8]; @@ -32,8 +34,10 @@ pub async fn udp_recv_start( other_sender: Sender<(SocketAddr, Vec)>, mut tun_writer: TunWriter, current_device: CurrentDeviceInfo, - stop_fn: F) - where F: FnOnce() + Send + 'static { + stop_fn: F, +) where + F: FnOnce() + Send + 'static, +{ { let udp = udp.try_clone().unwrap(); tokio::spawn(async move { @@ -45,14 +49,8 @@ pub async fn udp_recv_start( } thread::spawn(move || { - if let Err(e) = recv_loop( - udp, - server_addr, - other_sender, - tun_writer, - current_device, - ) { - log::error!("udp数据处理线程停止 {:?}",e); + if let Err(e) = recv_loop(udp, server_addr, other_sender, tun_writer, current_device) { + log::error!("udp数据处理线程停止 {:?}", e); } stop_fn(); }); @@ -97,12 +95,12 @@ fn recv_loop( return Err(Error::Stop(str)); } Err(e) => { - log::error!("{:?}",e); + log::error!("{:?}", e); } } } Err(e) => { - log::error!("{:?}",e); + log::error!("{:?}", e); } }; } @@ -158,7 +156,7 @@ fn recv_handle( return Err(Error::Stop("子处理线程停止".to_string())); } Err(e) => { - log::error!("子线程处理 {:?}",e); + log::error!("子线程处理 {:?}", e); } } } @@ -166,17 +164,21 @@ fn recv_handle( Ok(()) } -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 { +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); + log::error!("{:?}", e); } } stop_fn(); @@ -267,7 +269,7 @@ fn other_handle( } } InErrorPacket::OtherError(e) => { - log::error!("OtherError {:?}",e.message()); + log::error!("OtherError {:?}", e.message()); } } } @@ -301,7 +303,8 @@ fn other_handle( //回应 let mut punch_response = PunchResponsePacket::new(net_packet.payload_mut())?; punch_response.set_source(current_device.virtual_ip); - net_packet.set_transport_protocol(control_packet::Protocol::PunchResponse.into()); + net_packet + .set_transport_protocol(control_packet::Protocol::PunchResponse.into()); udp.send_to(net_packet.buffer(), peer_addr)?; let route = Route::new(peer_addr); DIRECT_ROUTE_TABLE.insert(src, route); @@ -329,7 +332,8 @@ fn other_handle( if !punch.reply { let mut punch_reply = Punch::new(); punch_reply.reply = true; - punch_reply.virtual_ip = u32::from_be_bytes(current_device.virtual_ip.octets()); + punch_reply.virtual_ip = + u32::from_be_bytes(current_device.virtual_ip.octets()); punch_reply.step = punch.step; if let Err(_) = sender.try_send(punch) { return Ok(()); @@ -339,15 +343,20 @@ fn other_handle( punch_reply.public_ip_list = info.public_ips.clone(); punch_reply.public_port = info.public_port as u32; punch_reply.public_port_range = info.public_port_range as u32; - punch_reply.nat_type = protobuf::EnumOrUnknown::new(info.nat_type); + punch_reply.nat_type = + protobuf::EnumOrUnknown::new(info.nat_type); drop(nat_info); let bytes = punch_reply.write_to_bytes()?; - let mut net_packet = NetPacket::new(vec![0u8; 4 + 8 + bytes.len()])?; + let mut net_packet = + NetPacket::new(vec![0u8; 4 + 8 + bytes.len()])?; net_packet.set_version(Version::V1); net_packet.set_protocol(Protocol::OtherTurn); - net_packet.set_transport_protocol(turn_packet::Protocol::Punch.into()); + net_packet.set_transport_protocol( + turn_packet::Protocol::Punch.into(), + ); net_packet.set_ttl(255); - let mut turn_packet = TurnPacket::new(net_packet.payload_mut())?; + let mut turn_packet = + TurnPacket::new(net_packet.payload_mut())?; turn_packet.set_source(current_device.virtual_ip); turn_packet.set_destination(src); turn_packet.set_payload(&bytes); @@ -365,7 +374,7 @@ fn other_handle( } } Protocol::UnKnow(p) => { - log::error!("未知协议 {}",p); + log::error!("未知协议 {}", p); } } Ok(()) diff --git a/switch/src/lib.rs b/switch/src/lib.rs index 2055a42..78341a2 100644 --- a/switch/src/lib.rs +++ b/switch/src/lib.rs @@ -7,15 +7,18 @@ 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; +use crate::handle::{ + ApplicationStatus, ConnectStatus, CurrentDeviceInfo, Route, RouteType, DEVICE_LIST, + DIRECT_ROUTE_TABLE, SERVER_RT, +}; -pub mod tun_device; -pub mod nat; pub mod error; pub mod handle; +pub mod nat; pub mod proto; pub mod protocol; +pub mod tun_device; #[derive(Clone, Debug)] pub struct Config { @@ -25,10 +28,7 @@ pub struct Config { impl Config { pub fn new(token: String, mac_address: String) -> Self { - Self { - token, - mac_address, - } + Self { token, mac_address } } } @@ -50,9 +50,7 @@ impl Switch { switch.runtime = Some(runtime); Ok(switch) } - Err(e) => { - Err(e) - } + Err(e) => Err(e), }; } pub fn stop(self) { @@ -89,7 +87,11 @@ impl Switch { 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 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))) { @@ -100,14 +102,15 @@ impl Switch { if e.kind() == io::ErrorKind::AddrInUse { port += 1; } else { - log::error!("创建udp失败 {:?}",e); + 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 response = + handle::registration_handler::registration(&udp, server_address, token, mac_address)?; { let ip_list = response .virtual_ip_list @@ -121,8 +124,10 @@ impl Switch { 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 (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(); //心跳线程 { @@ -130,7 +135,8 @@ impl Switch { let wait_group1 = wait_group.clone(); handle::heartbeat_handler::start(status_receiver.clone(), udp, current_device, || { drop(wait_group1); - }).await; + }) + .await; } //初始化nat数据 handle::init_nat_info(response.public_ip, response.public_port as u16); @@ -138,7 +144,8 @@ impl Switch { 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(); + let (punch_sender, cone_receiver, req_symmetric_receiver, res_symmetric_receiver) = + handle::punch_handler::bounded(); //udp数据处理 { // 低优先级的udp数据通道 @@ -155,51 +162,74 @@ impl Switch { || { drop(wait_group1); }, - ).await; + ) + .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; + 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; + 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; + 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; + 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; + handle::tun_handler::handler_start( + status_receiver.clone(), + udp, + tun_reader, + current_device, + || { + drop(wait_group1); + }, + ) + .await; } Ok(Switch { current_device, @@ -208,4 +238,4 @@ impl Switch { runtime: None, }) } -} \ No newline at end of file +} diff --git a/switch/src/nat/check.rs b/switch/src/nat/check.rs index 558f7fb..a3c059a 100644 --- a/switch/src/nat/check.rs +++ b/switch/src/nat/check.rs @@ -1,7 +1,7 @@ -use std::{io, thread}; use std::collections::HashSet; use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket}; use std::time::Duration; +use std::{io, thread}; use crate::proto::message::NatType; @@ -154,4 +154,4 @@ fn nat_test_run() { let udp = UdpSocket::bind("0.0.0.0:101").unwrap(); let print = public_ip_list_(&udp).unwrap(); println!("{:?}", print); -} \ No newline at end of file +} diff --git a/switch/src/nat/mod.rs b/switch/src/nat/mod.rs index 1e41ae9..3e8ff0f 100644 --- a/switch/src/nat/mod.rs +++ b/switch/src/nat/mod.rs @@ -1 +1 @@ -pub mod check; \ No newline at end of file +pub mod check; diff --git a/switch/src/protocol/control_packet.rs b/switch/src/protocol/control_packet.rs index 499ec26..93e0ab1 100644 --- a/switch/src/protocol/control_packet.rs +++ b/switch/src/protocol/control_packet.rs @@ -70,7 +70,6 @@ pub struct PongPacket { buffer: B, } - impl> PingPacket { pub fn new(buffer: B) -> Result> { let len = buffer.as_ref().len(); diff --git a/switch/src/tun_device/linux.rs b/switch/src/tun_device/linux.rs index 11df490..9a849ec 100644 --- a/switch/src/tun_device/linux.rs +++ b/switch/src/tun_device/linux.rs @@ -5,8 +5,8 @@ use std::os::unix::process::CommandExt; use std::process::Command; use bytes::BufMut; -use tun::Device; use tun::platform::posix::{Reader, Writer}; +use tun::Device; use crate::tun_device::{TunReader, TunWriter}; @@ -36,4 +36,3 @@ pub fn create_tun( TunReader(reader, packet_information), )) } - diff --git a/switch/src/tun_device/mac.rs b/switch/src/tun_device/mac.rs index 0e0db6b..a20f1a2 100644 --- a/switch/src/tun_device/mac.rs +++ b/switch/src/tun_device/mac.rs @@ -5,8 +5,8 @@ use std::os::unix::process::CommandExt; use std::process::Command; use bytes::BufMut; -use tun::Device; use tun::platform::posix::{Reader, Writer}; +use tun::Device; use crate::tun_device::{TunReader, TunWriter}; @@ -36,7 +36,10 @@ pub fn create_tun( .output() .expect("sh exec error!"); if !up_eth_out.status.success() { - return Err(crate::error::Error::Stop(format!("设置地址失败:{:?}", up_eth_out))); + return Err(crate::error::Error::Stop(format!( + "设置地址失败:{:?}", + up_eth_out + ))); } let if_config_out = Command::new("sh") .arg("-c") @@ -44,7 +47,10 @@ pub fn create_tun( .output() .expect("sh exec error!"); if !if_config_out.status.success() { - return Err(crate::error::Error::Stop(format!("设置路由失败:{:?}", if_config_out))); + return Err(crate::error::Error::Stop(format!( + "设置路由失败:{:?}", + if_config_out + ))); } // println!("{:?}", if_config_out); // let cmd_str: String = " ifconfig|grep flags=8051|awk -F ':' '{print $1}'|tail -1".to_string(); @@ -65,4 +71,3 @@ pub fn create_tun( TunReader(reader, packet_information), )) } - diff --git a/switch/src/tun_device/mod.rs b/switch/src/tun_device/mod.rs index 6f36228..a3e00a9 100644 --- a/switch/src/tun_device/mod.rs +++ b/switch/src/tun_device/mod.rs @@ -1,18 +1,18 @@ -#[cfg(any(target_os = "linux",target_os = "android"))] +#[cfg(any(target_os = "linux", target_os = "android"))] pub use linux::create_tun; #[cfg(target_os = "macos")] pub use mac::create_tun; #[cfg(any(unix))] pub use unix::{TunReader, TunWriter}; #[cfg(target_os = "windows")] -pub use windows::{TunReader, TunWriter}; -#[cfg(target_os = "windows")] pub use windows::create_tun; +#[cfg(target_os = "windows")] +pub use windows::{TunReader, TunWriter}; +#[cfg(any(target_os = "linux", target_os = "android"))] +pub mod linux; #[cfg(target_os = "macos")] pub mod mac; -#[cfg(any(target_os = "linux",target_os = "android"))] -pub mod linux; #[cfg(any(unix))] pub mod unix; #[cfg(target_os = "windows")]