diff --git a/switch/Cargo.toml b/switch/Cargo.toml index 4ea4092..d12f077 100644 --- a/switch/Cargo.toml +++ b/switch/Cargo.toml @@ -1,13 +1,12 @@ [package] name = "switch" -version = "1.0.2" +version = "1.0.6" edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] packet = { path = "./packet" } -p2p_channel = { path = "./p2p_channel" } bytes = "1.3.0" log = "0.4.17" libc = "0.2.137" @@ -24,7 +23,7 @@ sha2 = { version = "0.10.6", features = ["oid"] } thiserror = "1.0.37" chrono = "0.4.23" #lazy_static = "1.4.0" -#moka = "0.9.6" +moka = "0.9.6" protobuf = "3.2.0" #local-ip-address = "0.4.9" socket2 ={ version = "0.5.2", features = ["all"] } diff --git a/switch/p2p_channel b/switch/p2p_channel deleted file mode 160000 index 9d2e02f..0000000 --- a/switch/p2p_channel +++ /dev/null @@ -1 +0,0 @@ -Subproject commit 9d2e02f6290bde873584f457cd1236a594b1bb42 diff --git a/switch/proto/message.proto b/switch/proto/message.proto index 6a490b4..72c2e19 100644 --- a/switch/proto/message.proto +++ b/switch/proto/message.proto @@ -4,6 +4,7 @@ message RegistrationRequest{ string device_id = 2; string name = 3; bool is_fast = 4; + string version = 5; } message RegistrationResponse{ diff --git a/switch/src/core/mod.rs b/switch/src/core/mod.rs index 22c2220..d422887 100644 --- a/switch/src/core/mod.rs +++ b/switch/src/core/mod.rs @@ -1,28 +1,34 @@ -use std::io; +use std::{io, thread}; use std::net::{Ipv4Addr, SocketAddr}; use std::sync::Arc; use crossbeam::atomic::AtomicCell; use crossbeam_skiplist::SkipMap; use parking_lot::Mutex; +use tokio::net::UdpSocket; +use tokio::sync::mpsc::channel; -use p2p_channel::boot::Boot; -use p2p_channel::channel::{Channel, Route, RouteKey}; -use p2p_channel::punch::NatInfo; -use crate::{tap_device, tun_device}; +use crate::channel::channel::{Channel, Context}; +use crate::channel::idle::Idle; +use crate::channel::punch::{NatInfo, Punch}; +use crate::channel::{Route, RouteKey}; +use crate::channel::sender::ChannelSender; + use crate::external_route::ExternalRoute; -use crate::handle::{ConnectStatus, CurrentDeviceInfo, CurrentDeviceInfoExt, heartbeat_handler, PeerDeviceInfo, punch_handler, recv_handler, registration_handler, tap_handler, tun_handler}; +use crate::handle::{ConnectStatus, CurrentDeviceInfo, heartbeat_handler, PeerDeviceInfo, punch_handler, registration_handler}; +use crate::handle::recv_handler::ChannelDataHandler; +use crate::handle::tun_tap::{tap_handler, tun_handler}; +use crate::igmp_server::IgmpServer; use crate::nat::NatTest; -use crate::tap_device::TapWriter; -use crate::tun_device::TunWriter; +use crate::tun_tap_device; +use crate::tun_tap_device::DeviceWriter; pub struct Switch { name: String, current_device: Arc>, - tun_writer: Option, - tap_writer: Option, - nat_channel: Channel, + context: Context, + device_writer: DeviceWriter, /// 0. 机器纪元，每一次上线或者下线都会增1，用于感知网络中机器变化 /// 服务端和客户端的不一致，则服务端会推送新的设备列表 /// 1. 网络中的虚拟ip列表 @@ -35,10 +41,16 @@ pub struct Switch { impl Switch { pub async fn start(config: Config) -> crate::Result { log::info!("config:{:?}",config); + let main_channel = Arc::new(UdpSocket::bind("0.0.0.0:0").await?); + let response = registration_handler::registration(&main_channel, config.server_address, config.token.clone(), config.device_id.clone(), config.name.clone()).await?; + let (cone_sender, cone_receiver) = channel(3); + let (symmetric_sender, symmetric_receiver) = channel(2); + let context = Context::new(main_channel, 1); + let punch = Punch::new(context.clone()); + let idle = Idle::new(16000, context.clone()); + let channel_sender = ChannelSender::new(context.clone()); - let (mut channel, punch, idle) = Boot::new::(80, 15000, 0)?; - let response = registration_handler::registration(&mut channel, config.server_address, config.token.clone(), config.device_id.clone(), config.name.clone())?; - let register = Arc::new(registration_handler::Register::new(channel.sender()?, config.server_address, config.token.clone(), config.device_id.clone(), config.name.clone())); + let register = Arc::new(registration_handler::Register::new(channel_sender.clone(), config.server_address, config.token.clone(), config.device_id.clone(), config.name.clone())); let device_list: Arc)>> = Arc::new(Mutex::new((0, Vec::new()))); let peer_nat_info_map: Arc> = Arc::new(SkipMap::new()); let connect_status = Arc::new(AtomicCell::new(ConnectStatus::Connected)); @@ -47,7 +59,7 @@ impl Switch { let virtual_netmask = Ipv4Addr::from(response.virtual_netmask); let local_ip = crate::nat::local_ip()?; - let local_port = channel.local_addr()?.port(); + let local_port = context.main_local_port()?; // NAT检测 let nat_test = NatTest::new(config.nat_test_server.clone(), Ipv4Addr::from(response.public_ip), response.public_port as u16, local_ip, local_port); let in_ips = config.in_ips.iter().map(|(dest, mask, _)| { (Ipv4Addr::from(*dest), Ipv4Addr::from(*mask)) }).collect::>(); @@ -56,51 +68,60 @@ impl Switch { let out_external_route = ExternalRoute::new(config.out_ips); let in_external_route = ExternalRoute::new(config.in_ips); let current_device = Arc::new(AtomicCell::new(CurrentDeviceInfo::new(virtual_ip, virtual_gateway, virtual_netmask, config.server_address))); - let ip_proxy_map = crate::ip_proxy::init_proxy(channel.sender()?,out_ips,current_device.clone()).await?; - let ( current_device_ext,tun_writer, tap_writer) = if config.tap { + let ip_proxy_map = crate::ip_proxy::init_proxy(channel_sender.clone(), out_ips, current_device.clone()).await?; + let (device_writer, igmp_server) = if config.tap { #[cfg(windows)] { //删除switch的tun网卡避免ip冲突，因为非正常退出会保留网卡 - tun_device::delete_tun(); + tun_tap_device::delete_device(tun_tap_device::DeviceType::Tap); } - let (tap_writer, tap_reader, mac) = tap_device::create_tap(virtual_ip, virtual_netmask, virtual_gateway)?; - let current_device_ext = Arc::new(AtomicCell::new(CurrentDeviceInfoExt::new( mac))); + let (tap_writer, tap_reader) = tun_tap_device::create_device(tun_tap_device::DeviceType::Tap, virtual_ip, virtual_netmask, virtual_gateway, in_ips)?; + let igmp_server = IgmpServer::new(tap_writer.clone()); //tap数据处理 - tap_handler::start(channel.sender()?, tap_reader.clone(), tap_writer.clone(), current_device.clone(), in_external_route, ip_proxy_map.clone()); - (current_device_ext, None, Some(tap_writer)) + tap_handler::start(channel_sender.clone(), tap_reader.clone(), tap_writer.clone(), igmp_server.clone(), current_device.clone(), in_external_route, ip_proxy_map.clone()); + (tap_writer, igmp_server) } else { #[cfg(windows)] { //删除switch的tap网卡避免ip冲突，非正常退出会保留网卡 - tap_device::delete_tap(); + tun_tap_device::delete_device(tun_tap_device::DeviceType::Tap); } // tun通道 - let (tun_writer, tun_reader) = tun_device::create_tun(virtual_ip, virtual_netmask, virtual_gateway,in_ips)?; - let current_device_ext = Arc::new(AtomicCell::new(CurrentDeviceInfoExt::new( [0, 0, 0, 0, 0, 0]))); + let (tun_writer, tun_reader) = tun_tap_device::create_device(tun_tap_device::DeviceType::Tun, virtual_ip, virtual_netmask, virtual_gateway, in_ips)?; + let igmp_server = IgmpServer::new(tun_writer.clone()); //tun数据接收处理 - tun_handler::start(channel.sender()?, tun_reader.clone(), tun_writer.clone(), current_device.clone(), in_external_route, ip_proxy_map.clone()); - (current_device_ext,Some(tun_writer), None) + tun_handler::start(channel_sender.clone(), tun_reader.clone(), tun_writer.clone(), igmp_server.clone(), current_device.clone(), in_external_route, ip_proxy_map.clone()); + (tun_writer, igmp_server) }; //外部数据接收处理 - let channel_recv_handler = recv_handler::RecvHandler::new(channel.try_clone()?, current_device.clone(), current_device_ext,device_list.clone(), register.clone(), - nat_test.clone(), tun_writer.clone(), tap_writer.clone(), - connect_status.clone(), peer_nat_info_map.clone(), ip_proxy_map, out_external_route); - recv_handler::start(channel_recv_handler); + let channel_recv_handler = ChannelDataHandler::new(current_device.clone(), device_list.clone(), + register.clone(), nat_test.clone(), igmp_server, + device_writer.clone(), connect_status.clone(), + peer_nat_info_map.clone(), ip_proxy_map, out_external_route, + cone_sender, symmetric_sender); + let channel = Channel::new(context.clone(), channel_recv_handler); + thread::spawn(move || { + tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build().unwrap() + .block_on(channel.start(14, 60)); + }); + context.switch(nat_test.nat_info().nat_type); // 定时心跳 - heartbeat_handler::start_heartbeat(channel.sender()?, device_list.clone(), current_device.clone()); + heartbeat_handler::start_heartbeat(channel_sender.clone(), device_list.clone(), current_device.clone()).await; // 空闲检查 - heartbeat_handler::start_idle(idle, channel.sender()?); + heartbeat_handler::start_idle(idle, channel_sender.clone()).await; // 打洞处理 - punch_handler::start_cone(punch.try_clone()?, current_device.clone()); - punch_handler::start_symmetric(punch, current_device.clone()); - punch_handler::start_punch(nat_test.clone(), device_list.clone(), channel.sender()?, current_device.clone()); + punch_handler::start(cone_receiver, punch.clone(), current_device.clone()).await; + punch_handler::start(symmetric_receiver, punch, current_device.clone()).await; + punch_handler::start_punch(nat_test.clone(), device_list.clone(), channel_sender.clone(), current_device.clone()).await; + log::info!("switch启动成功"); Ok(Switch { name: config.name, current_device, - tun_writer, - tap_writer, - nat_channel: channel, + context, + device_writer, nat_test, device_list, connect_status, @@ -132,22 +153,17 @@ impl Switch { device_list } pub fn route(&self, ip: &Ipv4Addr) -> Option { - self.nat_channel.route(ip) + self.context.route_one(ip) } pub fn route_key(&self, route_key: &RouteKey) -> Option { - self.nat_channel.route_to_id(route_key) + self.context.route_to_id(route_key) } pub fn route_table(&self) -> Vec<(Ipv4Addr, Route)> { - self.nat_channel.route_table() + self.context.route_table_one() } pub fn stop(&self) -> io::Result<()> { - if let Some(tap) = &self.tap_writer { - tap.close()?; - } - if let Some(tun) = &self.tun_writer { - tun.close()?; - } - self.nat_channel.close()?; + self.context.close(); + self.device_writer.close()?; Ok(()) } } diff --git a/switch/src/error/mod.rs b/switch/src/error/mod.rs index 24e6e3b..889a925 100644 --- a/switch/src/error/mod.rs +++ b/switch/src/error/mod.rs @@ -5,8 +5,6 @@ use thiserror::Error; #[derive(Error, Debug)] pub enum Error { - #[error("packet error")] - PacketError(#[from] packet::error::Error), #[error("Io error")] Io(#[from] io::Error), #[error("Channel error")] diff --git a/switch/src/external_route/mod.rs b/switch/src/external_route/mod.rs index a7cd6af..77a6ec1 100644 --- a/switch/src/external_route/mod.rs +++ b/switch/src/external_route/mod.rs @@ -2,6 +2,7 @@ use std::net::Ipv4Addr; // 目标ip，子网掩码，网关 +#[derive(Clone)] pub struct ExternalRoute { route_table: Vec<(u32, u32, Ipv4Addr)>, } diff --git a/switch/src/handle/heartbeat_handler.rs b/switch/src/handle/heartbeat_handler.rs index ad13df2..8496832 100644 --- a/switch/src/handle/heartbeat_handler.rs +++ b/switch/src/handle/heartbeat_handler.rs @@ -1,60 +1,54 @@ use std::net::Ipv4Addr; use std::sync::Arc; use std::time::Duration; -use std::{io, thread}; +use std::io; use chrono::Local; use crossbeam::atomic::AtomicCell; use parking_lot::Mutex; use rand::prelude::SliceRandom; +use crate::channel::idle::Idle; +use crate::channel::Route; +use crate::channel::sender::ChannelSender; -use p2p_channel::channel::sender::Sender; -use p2p_channel::channel::Route; -use p2p_channel::idle::Idle; use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo}; use crate::protocol::control_packet::PingPacket; -use crate::protocol::{control_packet, NetPacket, Protocol, Version, MAX_TTL}; +use crate::protocol::{control_packet, NetPacket, Protocol, Version}; -pub fn start_idle(idle: Idle, sender: Sender) { - thread::Builder::new() - .name("idle".into()) - .spawn(move || { - if let Err(e) = start_idle_(idle, sender) { - log::info!("空闲检测线程停止:{:?}", e); +pub async fn start_idle(idle: Idle, sender: ChannelSender) { + tokio::spawn(async move { + match start_idle_(idle, sender).await { + Ok(_) => {} + Err(e) => { + log::warn!("空闲检测任务停止:{:?}", e); } - }) - .unwrap(); + } + }); } -fn start_idle_(idle: Idle, sender: Sender) -> io::Result<()> { +async fn start_idle_(idle: Idle, sender: ChannelSender) -> io::Result<()> { loop { - let (idle_status, peer_ips, route) = idle.next_idle()?; - log::warn!( - "peer_ip:{:?},route:{:?},idle_status:{:?}", - peer_ips, - route, - idle_status + let (peer_ip, route) = idle.next_idle().await?; + log::info!( + "peer_ip:{:?},route:{:?}", + peer_ip, + route ); - for peer_ip in peer_ips { - sender.remove_route(&peer_ip); - } + sender.remove_route(&peer_ip, route); } } -pub fn start_heartbeat( - sender: Sender, +pub async fn start_heartbeat( + sender: ChannelSender, device_list: Arc)>>, current_device: Arc>, ) { - thread::Builder::new() - .name("heartbeat".into()) - .spawn(move || { - if let Err(e) = start_heartbeat_(sender, device_list, current_device) { - log::info!("空闲检测线程停止:{:?}", e); - } - }) - .unwrap(); + tokio::spawn(async move { + if let Err(e) = start_heartbeat_(sender, device_list, current_device).await { + log::warn!("心跳任务停止:{:?}", e); + } + }); } fn set_now_time(packet: &mut NetPacket<[u8; 16]>) -> io::Result<()> { @@ -64,8 +58,8 @@ fn set_now_time(packet: &mut NetPacket<[u8; 16]>) -> io::Result<()> { Ok(()) } -fn start_heartbeat_( - sender: Sender, +async fn start_heartbeat_( + sender: ChannelSender, device_list: Arc)>>, current_device: Arc>, ) -> io::Result<()> { @@ -73,7 +67,8 @@ fn start_heartbeat_( net_packet.set_version(Version::V1); net_packet.set_protocol(Protocol::Control); net_packet.set_transport_protocol(control_packet::Protocol::Ping.into()); - net_packet.first_set_ttl(MAX_TTL); + //只寻找两跳以内能到的目标 + net_packet.first_set_ttl(2); let mut count = 0; loop { let current_device = current_device.load(); @@ -83,19 +78,28 @@ fn start_heartbeat_( let epoch = { device_list.lock().0 }; ping.set_epoch(epoch); } + set_now_time(&mut net_packet)?; + net_packet.set_destination(current_device.virtual_gateway()); + if let Err(e) = sender.send_main(net_packet.buffer(), current_device.connect_server).await + { + log::warn!( + "connect_server:{:?},e:{:?}", + current_device.connect_server, + e + ); + } if count < 7 || count % 7 == 0 { - let mut route_list: Option> = None; - let peer_list = device_list.lock().1.clone(); + let mut route_list: Option)>> = None; + let peer_list = {device_list.lock().1.clone()}; for peer in peer_list { set_now_time(&mut net_packet)?; - net_packet.first_set_ttl(2); net_packet.set_destination(peer.virtual_ip); if sender - .send_to_id(net_packet.buffer(), &peer.virtual_ip) + .send_by_id(net_packet.buffer(), &peer.virtual_ip).await .is_err() { //没有路由则发送到网关 - let _ = sender.send_to_addr(net_packet.buffer(), current_device.connect_server); + let _ = sender.try_send_main(net_packet.buffer(), current_device.connect_server); //再随机发送到其他地址，看有没有客户端符合转发条件 let route_list = route_list.get_or_insert_with(|| { let mut l = sender.route_table(); @@ -103,43 +107,37 @@ fn start_heartbeat_( l }); let mut num = 0; - //只寻找两跳以内能到的目标 - net_packet.first_set_ttl(2); - for (peer_ip, route) in route_list.iter() { - if peer_ip != &peer.virtual_ip && route.metric == 1 { - set_now_time(&mut net_packet)?; - let _ = sender.send_to_route(net_packet.buffer(), &route.route_key()); - num += 1; - } - if num >= 3 { - break; + 'a: for (peer_ip, route_list) in route_list.iter() { + for route in route_list { + if peer_ip != &peer.virtual_ip && route.metric == 1 { + set_now_time(&mut net_packet)?; + let _ = sender.try_send_by_key(net_packet.buffer(), &route.route_key()); + num += 1; + break; + } + if num >= 3 { + break 'a; + } } } } - thread::sleep(Duration::from_millis(1)); - } - set_now_time(&mut net_packet)?; - net_packet.set_destination(current_device.virtual_gateway()); - if let Err(e) = sender.send_to_addr(net_packet.buffer(), current_device.connect_server) - { - log::warn!( - "connect_server:{:?},e:{:?}", - current_device.connect_server, - e - ); + tokio::time::sleep(Duration::from_millis(1)).await; } + } else { - for (peer_ip, route) in sender.route_table().iter() { + for (peer_ip, route_list) in sender.route_table().iter() { set_now_time(&mut net_packet)?; net_packet.set_destination(*peer_ip); - if let Err(e) = sender.send_to_route(net_packet.buffer(), &route.route_key()) { - log::warn!("peer_ip:{:?},route:{:?},e:{:?}", peer_ip, route, e); + for route in route_list { + if let Err(e) = sender.send_by_key(net_packet.buffer(), &route.route_key()).await { + log::warn!("peer_ip:{:?},route:{:?},e:{:?}", peer_ip, route, e); + } + tokio::time::sleep(Duration::from_millis(2)).await; } - thread::sleep(Duration::from_millis(1)); } } count += 1; - thread::sleep(Duration::from_millis(5000)); + tokio::time::sleep(Duration::from_millis(5000)).await; } } diff --git a/switch/src/handle/mod.rs b/switch/src/handle/mod.rs index d6e9121..f20651b 100644 --- a/switch/src/handle/mod.rs +++ b/switch/src/handle/mod.rs @@ -1,11 +1,10 @@ use std::net::{Ipv4Addr, SocketAddr}; pub mod heartbeat_handler; -pub mod tun_handler; -pub mod tap_handler; pub mod punch_handler; pub mod recv_handler; pub mod registration_handler; +pub mod tun_tap; /// 是否在一个网段 fn check_dest(dest: Ipv4Addr, virtual_netmask: Ipv4Addr, virtual_network: Ipv4Addr) -> bool { @@ -60,19 +59,6 @@ pub enum ConnectStatus { Connected, } -#[derive(Copy, Clone, Debug, Eq, PartialEq)] -pub struct CurrentDeviceInfoExt { - pub mac: [u8; 6], -} - -impl CurrentDeviceInfoExt { - pub fn new(mac: [u8; 6]) -> Self { - Self { - mac, - } - } -} - #[derive(Copy, Clone, Debug, Eq, PartialEq)] pub struct CurrentDeviceInfo { virtual_ip: Ipv4Addr, diff --git a/switch/src/handle/punch_handler.rs b/switch/src/handle/punch_handler.rs index bdbd6e4..38cd179 100644 --- a/switch/src/handle/punch_handler.rs +++ b/switch/src/handle/punch_handler.rs @@ -1,93 +1,72 @@ use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo}; use crate::nat::NatTest; use crate::proto::message::{PunchInfo, PunchNatType}; -use crate::protocol::{control_packet, turn_packet, NetPacket, Protocol, Version, MAX_TTL}; +use crate::protocol::{control_packet, other_turn_packet, NetPacket, Protocol, Version, MAX_TTL}; use crossbeam::atomic::AtomicCell; -use p2p_channel::channel::sender::Sender; -use p2p_channel::punch::{NatInfo, NatType, Punch}; use parking_lot::Mutex; use protobuf::Message; use rand::prelude::SliceRandom; -use std::net::{IpAddr, Ipv4Addr}; +use std::net::Ipv4Addr; use std::sync::Arc; use std::time::Duration; -use std::{io, thread}; +use std::io; +use tokio::sync::mpsc::Receiver; +use crate::channel::punch::{NatInfo, Punch}; +use crate::channel::sender::ChannelSender; -pub fn start_cone(punch: Punch, current_device: Arc>) { - thread::Builder::new() - .name("punch-cone".into()) - .spawn(move || { - if let Err(e) = start_(true, punch, current_device) { - log::warn!("锥形网络打洞处理线程停止 {:?}", e); - } - }) - .unwrap(); +pub async fn start(receiver: Receiver<(Ipv4Addr, NatInfo)>, punch: Punch, current_device: Arc>) { + tokio::spawn(async move { + start0(receiver, punch, current_device).await; + }); } -pub fn start_symmetric(punch: Punch, current_device: Arc>) { - thread::Builder::new() - .name("punch-symmetric".into()) - .spawn(move || { - if let Err(e) = start_(false, punch, current_device) { - log::warn!("对称网络打洞处理线程停止 {:?}", e); - } - }) - .unwrap(); +pub async fn start0(mut receiver: Receiver<(Ipv4Addr, NatInfo)>, mut punch: Punch, current_device: Arc>) { + while let Some((peer_ip, nat_info)) = receiver.recv().await { + if let Err(e) = start_(&mut punch, ¤t_device, peer_ip, nat_info).await { + log::warn!("网络打洞异常 {:?}", e); + } + } } -fn start_( - is_cone: bool, - mut punch: Punch, - current_device: Arc>, +async fn start_( + punch: &mut Punch, + current_device: &Arc>, + peer_ip: Ipv4Addr, + nat_info: NatInfo, ) -> io::Result<()> { let mut packet = NetPacket::new([0u8; 12])?; packet.set_version(Version::V1); packet.first_set_ttl(1); packet.set_protocol(Protocol::Control); packet.set_transport_protocol(control_packet::Protocol::PunchRequest.into()); - loop { - let (peer_ip, nat_info) = if is_cone { - punch.next_cone(None)? - } else { - punch.next_symmetric(None)? - }; - if let Some(route) = punch.sender().route(&peer_ip) { - if route.metric == 1 { - //直连地址不需要打洞 - continue; - } - } - packet.set_source(current_device.load().virtual_ip()); - packet.set_destination(peer_ip); - log::info!("发起打洞，目标:{:?},{:?}", peer_ip, nat_info); - if let Err(e) = punch.punch(packet.buffer(), peer_ip, nat_info) { - log::warn!("peer_ip:{:?},e:{:?}", peer_ip, e); - } - } + packet.set_source(current_device.load().virtual_ip()); + packet.set_destination(peer_ip); + log::info!("发起打洞，目标:{:?},{:?}", peer_ip, nat_info); + + punch.punch(packet.buffer(), peer_ip, nat_info).await } -pub fn start_punch( +pub async fn start_punch( nat_test: NatTest, device_list: Arc)>>, - sender: Sender, + sender: ChannelSender, current_device: Arc>, ) { - thread::Builder::new() - .name("punch-send-request".into()) - .spawn(move || { - if let Err(e) = start_punch_(nat_test, device_list, sender, current_device) { - log::warn!("对称网络打洞处理线程停止 {:?}", e); - } - }) - .unwrap(); + tokio::spawn(async move { + if let Err(e) = start_punch_(nat_test, device_list, sender, current_device).await { + log::warn!("打洞处理任务停止 {:?}", e); + } + }); } -fn start_punch_( +async fn start_punch_( nat_test: NatTest, device_list: Arc)>>, - sender: Sender, + sender: ChannelSender, current_device: Arc>, ) -> crate::Result<()> { + let mut num = 0; + let sleep_time = [3, 5, 7, 11, 13, 17, 19, 23, 29]; loop { if sender.is_close() { return Ok(()); @@ -102,28 +81,19 @@ fn start_punch_( if info.virtual_ip <= current_device.virtual_ip { continue; } - if let Some(route) = sender.route(&info.virtual_ip) { - if route.metric == 1 { - //直连地址不需要打洞 - continue; - } + if !sender.need_punch(&info.virtual_ip) { + continue; } count += 1; - if count > 3 { + if count > 2 { break; } let buf = punch_packet(current_device.virtual_ip(), &nat_info, info.virtual_ip)?; - sender.send_to_addr(&buf, current_device.connect_server)?; - } - } - match nat_info.nat_type { - NatType::Symmetric => { - thread::sleep(Duration::from_secs(28)); - } - NatType::Cone => { - thread::sleep(Duration::from_secs(20)); + sender.send_main(&buf, current_device.connect_server).await?; } } + num += 1; + tokio::time::sleep(Duration::from_secs(sleep_time[num % sleep_time.len()])).await; } } @@ -137,28 +107,18 @@ pub fn punch_packet( punch_reply.public_ip_list = nat_info .public_ips .iter() - .map(|i| match i { - IpAddr::V4(ip) => u32::from_be_bytes(ip.octets()), - IpAddr::V6(_) => { - panic!() - } - }) + .map(|ip| u32::from_be_bytes(ip.octets())) .collect(); punch_reply.public_port = nat_info.public_port as u32; punch_reply.public_port_range = nat_info.public_port_range as u32; - punch_reply.local_ip = match nat_info.local_ip { - IpAddr::V4(ip) => u32::from_be_bytes(ip.octets()), - IpAddr::V6(_) => { - panic!() - } - }; + punch_reply.local_ip = u32::from_be_bytes(nat_info.local_ip.octets()); punch_reply.local_port = nat_info.local_port as u32; punch_reply.nat_type = protobuf::EnumOrUnknown::new(PunchNatType::from(nat_info.nat_type)); let bytes = punch_reply.write_to_bytes()?; let mut net_packet = NetPacket::new(vec![0u8; 12 + 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(other_turn_packet::Protocol::Punch.into()); net_packet.first_set_ttl(MAX_TTL); net_packet.set_source(virtual_ip); net_packet.set_destination(dest); diff --git a/switch/src/handle/recv_handler.rs b/switch/src/handle/recv_handler.rs index 8c5f8b3..fc47724 100644 --- a/switch/src/handle/recv_handler.rs +++ b/switch/src/handle/recv_handler.rs @@ -1,5 +1,4 @@ -use std::thread; -use std::net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4}; +use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4}; use std::sync::Arc; use chrono::Local; @@ -7,128 +6,100 @@ use crossbeam::atomic::AtomicCell; use crossbeam_skiplist::SkipMap; use parking_lot::Mutex; use protobuf::Message; +use tokio::net::UdpSocket; +use tokio::sync::mpsc::Sender; -use p2p_channel::channel::{Channel, Route, RouteKey}; -use p2p_channel::punch::NatInfo; -use packet::ethernet; use packet::icmp::{icmp, Kind}; use packet::icmp::icmp::HeaderOther; use packet::ip::ipv4; use packet::ip::ipv4::packet::IpV4Packet; +use crate::channel::channel::Context; +use crate::channel::punch::{NatInfo, NatType}; +use crate::channel::{Route, RouteKey}; use crate::error::Error; use crate::external_route::ExternalRoute; -use crate::handle::{check_dest, ConnectStatus, CurrentDeviceInfo, CurrentDeviceInfoExt, PeerDeviceInfo}; +use crate::handle::{check_dest, ConnectStatus, CurrentDeviceInfo, PeerDeviceInfo}; use crate::handle::registration_handler::Register; +use crate::igmp_server::IgmpServer; use crate::ip_proxy::IpProxyMap; use crate::nat; use crate::nat::NatTest; use crate::proto::message::{DeviceList, PunchInfo, PunchNatType, RegistrationResponse}; -use crate::protocol::{control_packet, MAX_TTL, NetPacket, Protocol, service_packet, turn_packet, Version}; +use crate::protocol::{control_packet, MAX_TTL, NetPacket, Protocol, service_packet, other_turn_packet, Version, ip_turn_packet}; use crate::protocol::control_packet::ControlPacket; use crate::protocol::error_packet::InErrorPacket; -use crate::tap_device::TapWriter; -use crate::tun_device::TunWriter; +use crate::tun_tap_device::DeviceWriter; -pub fn start(mut handler: RecvHandler) { - thread::Builder::new().name("udp-recv-handler".into()).spawn(move || { - let mut buf = [0; 4096]; - loop { - match handler.channel.recv_from(&mut buf, None) { - Ok((len, route)) => { - if let Err(e) = handler.handle(&mut buf[..len], &route) { - log::warn!("数据处理失败:{:?},e:{:?}",route,e); - if let Error::Stop(_) = e { - let _ = handler.channel.close(); - break; - } - } - } - Err(e) => { - log::warn!("{:?}",e); - // 检查关闭状态 - if handler.channel.is_close() { - break; - } - } - } - } - }).unwrap(); -} - -pub struct RecvHandler { - channel: Channel, +#[derive(Clone)] +pub struct ChannelDataHandler { current_device: Arc>, - current_device_ext: Arc>, device_list: Arc)>>, register: Arc, nat_test: NatTest, - tun_writer: Option, - tap_writer: Option, + igmp_server: IgmpServer, + device_writer: DeviceWriter, connect_status: Arc>, peer_nat_info_map: Arc>, ip_proxy_map: IpProxyMap, out_external_route: ExternalRoute, + cone_sender: Sender<(Ipv4Addr, NatInfo)>, + symmetric_sender: Sender<(Ipv4Addr, NatInfo)>, } -impl RecvHandler { - pub fn new(channel: Channel, - current_device: Arc>, - current_device_ext: Arc>, +impl ChannelDataHandler { + pub fn new(current_device: Arc>, device_list: Arc)>>, register: Arc, nat_test: NatTest, - tun_writer: Option, - tap_writer: Option, + igmp_server: IgmpServer, + device_writer: DeviceWriter, connect_status: Arc>, peer_nat_info_map: Arc>, ip_proxy_map: IpProxyMap, out_external_route: ExternalRoute, - ) -> Self { + cone_sender: Sender<(Ipv4Addr, NatInfo)>, + symmetric_sender: Sender<(Ipv4Addr, NatInfo)>, ) -> Self { Self { - channel, current_device, - current_device_ext, device_list, register, nat_test, - tun_writer, - tap_writer, + igmp_server, + device_writer, connect_status, peer_nat_info_map, ip_proxy_map, out_external_route, + cone_sender, + symmetric_sender, } } - // pub fn try_clone(&self) -> io::Result { - // Ok(Self { - // channel: self.channel.try_clone()?, - // current_device: self.current_device.clone(), - // device_list: self.device_list.clone(), - // register: self.register.clone(), - // nat_test: self.nat_test.clone(), - // tun_writer: self.tun_writer.clone(), - // tap_writer: self.tap_writer.clone(), - // connect_status: self.connect_status.clone(), - // peer_nat_info_map: self.peer_nat_info_map.clone(), - // }) - // } } -impl RecvHandler { - fn handle(&self, buf: &mut [u8], route_key: &RouteKey) -> crate::Result<()> { - let mut net_packet = NetPacket::new(buf)?; + +impl ChannelDataHandler { + pub async fn handle(&mut self, buf: &mut [u8], start: usize, end: usize, route_key: RouteKey, _udp: &Arc, + context: &Context, ) { + assert_eq!(start, 14); + match self.handle0(&mut buf[..end], &route_key, context).await { + Ok(_) => {} + Err(e) => { + log::error!("{:?}",e); + } + } + } + async fn handle0(&self, buf: &mut [u8], route_key: &RouteKey, context: &Context) -> crate::Result<()> { + let mut net_packet = NetPacket::new(&mut buf[14..])?; if net_packet.ttl() == 0 { return Ok(()); } let source = net_packet.source(); let current_device = self.current_device.load(); - if source == current_device.virtual_ip() { - return Ok(()); - } let destination = net_packet.destination(); - if !destination.is_broadcast() && destination != current_device.broadcast_address - && current_device.virtual_ip() != destination && self.connect_status.load() == ConnectStatus::Connected { + if current_device.virtual_ip() != destination + && !destination.is_broadcast() && !destination.is_multicast() && destination != current_device.broadcast_address + && self.connect_status.load() == ConnectStatus::Connected { if !check_dest(source, current_device.virtual_netmask, current_device.virtual_network) { log::warn!("转发数据，源地址错误:{:?},当前网络:{:?},route_key:{:?}",source,current_device.virtual_network,route_key); return Ok(()); @@ -141,111 +112,118 @@ impl RecvHandler { let ttl = net_packet.ttl(); if ttl > 0 { // 转发 - if let Some(route) = self.channel.route(&destination) { + if let Some(route) = context.route_one(&destination) { if route.metric <= net_packet.ttl() { - self.channel.send_to_route(net_packet.buffer(), &route.route_key())?; + context.send_by_key(net_packet.buffer(), &route.route_key()).await?; } } else if (ttl > 1 || destination == current_device.virtual_gateway()) && source != current_device.virtual_gateway() { //网关默认要转发一次，生存时间不够的发到网关也会被丢弃 - self.channel.send_to_addr(net_packet.buffer(), current_device.connect_server)?; + context.send_main(net_packet.buffer(), current_device.connect_server).await?; } } return Ok(()); } match net_packet.protocol() { - Protocol::Ipv4Turn => { - let mut ipv4 = IpV4Packet::new(net_packet.payload_mut())?; - if ipv4.destination_ip() == destination && ipv4.protocol() == ipv4::protocol::Protocol::Icmp { - let mut icmp_packet = icmp::IcmpPacket::new(ipv4.payload_mut())?; - if icmp_packet.kind() == Kind::EchoRequest { - //开启ping - icmp_packet.set_kind(Kind::EchoReply); - icmp_packet.update_checksum(); - ipv4.set_source_ip(destination); - ipv4.set_destination_ip(source); - ipv4.update_checksum(); - net_packet.set_source(destination); - net_packet.set_destination(source); - self.channel.send_to_route(net_packet.buffer(), route_key)?; - return Ok(()); - } - } - if ipv4.destination_ip() != destination { - if let Some(gate_way) = self.out_external_route.route(&ipv4.destination_ip()) { + Protocol::IpTurn => { + match ip_turn_packet::Protocol::from(net_packet.transport_protocol()) { + ip_turn_packet::Protocol::Ipv4 => { + let mut ipv4 = IpV4Packet::new(net_packet.payload_mut())?; match ipv4.protocol() { - ipv4::protocol::Protocol::Tcp => { - let dest_ip = ipv4.destination_ip(); - //转发到代理目标地址 - let mut tcp_packet = packet::tcp::tcp::TcpPacket::new(source, destination, ipv4.payload_mut())?; - let source_port = tcp_packet.source_port(); - let dest_port = tcp_packet.destination_port(); - tcp_packet.set_destination_port(self.ip_proxy_map.tcp_proxy_port); - tcp_packet.update_checksum(); - ipv4.set_destination_ip(destination); - ipv4.update_checksum(); - self.ip_proxy_map.tcp_proxy_map.insert(SocketAddrV4::new(source, source_port), - (SocketAddrV4::new(gate_way, 0), SocketAddrV4::new(dest_ip, dest_port))); - } - ipv4::protocol::Protocol::Udp => { - let dest_ip = ipv4.destination_ip(); - //转发到代理目标地址 - let mut udp_packet = packet::udp::udp::UdpPacket::new(source, destination, ipv4.payload_mut())?; - let source_port = udp_packet.source_port(); - let dest_port = udp_packet.destination_port(); - udp_packet.set_destination_port(self.ip_proxy_map.udp_proxy_port); - udp_packet.update_checksum(); - ipv4.set_destination_ip(destination); - ipv4.update_checksum(); - println!("{:?}",ipv4); - self.ip_proxy_map.udp_proxy_map.insert(SocketAddrV4::new(source, source_port), - (SocketAddrV4::new(gate_way, 0), SocketAddrV4::new(dest_ip, dest_port))); + ipv4::protocol::Protocol::Igmp => { + self.igmp_server.handle(ipv4.payload(), source)?; + return Ok(()); } ipv4::protocol::Protocol::Icmp => { - let dest_ip = ipv4.destination_ip(); - //转发到代理目标地址 - let icmp_packet = icmp::IcmpPacket::new(ipv4.payload())?; - match icmp_packet.header_other() { - HeaderOther::Identifier(id, seq) => { - self.ip_proxy_map.icmp_proxy_map.insert((dest_ip, id, seq), source); - self.ip_proxy_map.send_icmp(ipv4.payload(), &gate_way, &dest_ip)?; - } - _ => { + if ipv4.destination_ip() == destination { + let mut icmp_packet = icmp::IcmpPacket::new(ipv4.payload_mut())?; + if icmp_packet.kind() == Kind::EchoRequest { + //开启ping + icmp_packet.set_kind(Kind::EchoReply); + icmp_packet.update_checksum(); + ipv4.set_source_ip(destination); + ipv4.set_destination_ip(source); + ipv4.update_checksum(); + net_packet.set_source(destination); + net_packet.set_destination(source); + context.send_by_key(net_packet.buffer(), route_key).await?; return Ok(()); } } } _ => { - return Ok(()); + if ipv4.destination_ip() != destination { + if let Some(gate_way) = self.out_external_route.route(&ipv4.destination_ip()) { + match ipv4.protocol() { + ipv4::protocol::Protocol::Tcp => { + let dest_ip = ipv4.destination_ip(); + //转发到代理目标地址 + let mut tcp_packet = packet::tcp::tcp::TcpPacket::new(source, destination, ipv4.payload_mut())?; + let source_port = tcp_packet.source_port(); + let dest_port = tcp_packet.destination_port(); + tcp_packet.set_destination_port(self.ip_proxy_map.tcp_proxy_port); + tcp_packet.update_checksum(); + ipv4.set_destination_ip(destination); + ipv4.update_checksum(); + self.ip_proxy_map.tcp_proxy_map.insert(SocketAddrV4::new(source, source_port), + (SocketAddrV4::new(gate_way, 0), SocketAddrV4::new(dest_ip, dest_port))); + } + ipv4::protocol::Protocol::Udp => { + let dest_ip = ipv4.destination_ip(); + //转发到代理目标地址 + let mut udp_packet = packet::udp::udp::UdpPacket::new(source, destination, ipv4.payload_mut())?; + let source_port = udp_packet.source_port(); + let dest_port = udp_packet.destination_port(); + udp_packet.set_destination_port(self.ip_proxy_map.udp_proxy_port); + udp_packet.update_checksum(); + ipv4.set_destination_ip(destination); + ipv4.update_checksum(); + self.ip_proxy_map.udp_proxy_map.insert(SocketAddrV4::new(source, source_port), + (SocketAddrV4::new(gate_way, 0), SocketAddrV4::new(dest_ip, dest_port))); + } + ipv4::protocol::Protocol::Icmp => { + let dest_ip = ipv4.destination_ip(); + //转发到代理目标地址 + let icmp_packet = icmp::IcmpPacket::new(ipv4.payload())?; + match icmp_packet.header_other() { + HeaderOther::Identifier(id, seq) => { + self.ip_proxy_map.icmp_proxy_map.insert((dest_ip, id, seq), source); + self.ip_proxy_map.send_icmp(ipv4.payload(), &gate_way, &dest_ip)?; + } + _ => { + return Ok(()); + } + } + } + _ => { + return Ok(()); + } + } + } + } } } + //传输协议12字节 + self.device_writer.write_ipv4(&mut buf[12..])?; + return Ok(()); } - } - if let Some(tun_writer) = &self.tun_writer { - tun_writer.write(net_packet.payload())?; - } else { - if let Some(tap_writer) = &self.tap_writer { - let mut ethernet_packet = ethernet::packet::EthernetPacket::unchecked(vec![0; 14 + ipv4.buffer.len()]); - let source = source.octets(); - ethernet_packet.set_source(&[source[0], source[1], source[2], source[3], 123, 234]); - ethernet_packet.set_destination(&self.current_device_ext.load().mac); - ethernet_packet.set_protocol(ethernet::protocol::Protocol::Ipv4); - ethernet_packet.payload_mut().copy_from_slice(ipv4.buffer); - tap_writer.write(ðernet_packet.buffer)?; + ip_turn_packet::Protocol::Ipv4Broadcast => { + //客户端不帮忙转发广播包，所以不会出现这种类型的数据 } + ip_turn_packet::Protocol::Unknown(_) => {} } } Protocol::Service => { - self.service(current_device, source, net_packet, route_key)?; + self.service(context, current_device, source, net_packet, route_key).await?; } Protocol::Error => { - self.error(current_device, source, net_packet, route_key)?; + self.error(context, current_device, source, net_packet, route_key).await?; } Protocol::Control => { - self.control(current_device, source, net_packet, route_key)?; + self.control(context, current_device, source, net_packet, route_key).await?; } Protocol::OtherTurn => { - self.other_turn(current_device, source, net_packet, route_key)?; + self.other_turn(context, current_device, source, net_packet, route_key).await?; } Protocol::UnKnow(e) => { log::info!("不支持的协议:{}",e); @@ -253,18 +231,20 @@ impl RecvHandler { } Ok(()) } - fn service(&self, current_device: CurrentDeviceInfo, source: Ipv4Addr, net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> { - if route_key.addr != current_device.connect_server || source != current_device.virtual_gateway() { + async fn service(&self, context: &Context, current_device: CurrentDeviceInfo, _source: Ipv4Addr, net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> { + //todo 校验来源 + if route_key.addr != current_device.connect_server { return Ok(()); } match service_packet::Protocol::from(net_packet.transport_protocol()) { service_packet::Protocol::RegistrationRequest => {} service_packet::Protocol::RegistrationResponse => { let response = RegistrationResponse::parse_from_bytes(net_packet.payload())?; - let local_port = self.channel.local_addr()?.port(); + let local_port = context.main_local_port()?; let local_ip = nat::local_ip()?; - let nat_info = self.nat_test.re_test(Ipv4Addr::from(response.public_ip), response.public_port as u16, local_ip, local_port); - self.channel.set_nat_type(nat_info.nat_type)?; + let nat_info = self.nat_test.re_test(Ipv4Addr::from(response.public_ip), + response.public_port as u16, local_ip, local_port); + context.switch(nat_info.nat_type); let new_ip = Ipv4Addr::from(response.virtual_ip); let current_ip = current_device.virtual_ip(); if current_ip != new_ip { @@ -275,13 +255,7 @@ impl RecvHandler { let virtual_ip = Ipv4Addr::from(response.virtual_ip); let virtual_gateway = Ipv4Addr::from(response.virtual_gateway); let virtual_netmask = Ipv4Addr::from(response.virtual_netmask); - if let Some(tun_writer) = &self.tun_writer { - tun_writer.change_ip(virtual_ip, virtual_netmask, virtual_gateway, old_netmask, old_gateway)?; - } else { - if let Some(tap_writer) = &self.tap_writer { - tap_writer.change_ip(virtual_ip, virtual_netmask, virtual_gateway, old_netmask, old_gateway)?; - } - } + self.device_writer.change_ip(virtual_ip, virtual_netmask, virtual_gateway, old_netmask, old_gateway)?; let new_current_device = CurrentDeviceInfo::new(virtual_ip, virtual_gateway, virtual_netmask, current_device.connect_server); if let Err(e) = self.current_device.compare_exchange(current_device, new_current_device) { @@ -310,14 +284,14 @@ impl RecvHandler { dev.1 = ip_list; } } - service_packet::Protocol::UnKnow(u) => { + service_packet::Protocol::Unknown(u) => { log::warn!("未知服务协议:{}",u); } } Ok(()) } - fn error(&self, current_device: CurrentDeviceInfo, source: Ipv4Addr, net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> { - if route_key.addr != current_device.connect_server || source != current_device.virtual_gateway() { + async fn error(&self, _context: &Context, current_device: CurrentDeviceInfo, _source: Ipv4Addr, net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> { + if route_key.addr != current_device.connect_server { return Ok(()); } match InErrorPacket::new(net_packet.transport_protocol(), net_packet.payload())? { @@ -325,8 +299,14 @@ impl RecvHandler { return Err(Error::Stop("Token error".to_string())); } InErrorPacket::Disconnect => { + { + //掉线epoch要归零 + let mut dev = self.device_list.lock(); + dev.0 = 0; + } + self.connect_status.store(ConnectStatus::Connecting); - self.register.fast_register()?; + self.register.fast_register().await?; } InErrorPacket::AddressExhausted => { //地址用尽 @@ -338,42 +318,28 @@ impl RecvHandler { } Ok(()) } - fn control(&self, current_device: CurrentDeviceInfo, source: Ipv4Addr, mut net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> { + async fn control(&self, context: &Context, current_device: CurrentDeviceInfo, source: Ipv4Addr, mut net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> { + let metric = net_packet.source_ttl() - net_packet.ttl() + 1; match ControlPacket::new(net_packet.transport_protocol(), net_packet.payload())? { ControlPacket::PingPacket(_) => { - let metric = net_packet.source_ttl() - net_packet.ttl() + 1; + context.update_read_time(&source, route_key); net_packet.set_transport_protocol(control_packet::Protocol::Pong.into()); net_packet.set_source(current_device.virtual_ip()); net_packet.set_destination(source); net_packet.first_set_ttl(MAX_TTL); - self.channel.send_to_route(net_packet.buffer(), route_key)?; - if metric == 1 { - if let Some(current_route) = self.channel.route(&source) { - if current_route.metric > 1 { - let route = Route::from(*route_key, 1, -1); - self.channel.add_route(source, route); - } - } - } + context.send_by_key(net_packet.buffer(), route_key).await?; + let route = Route::from(*route_key, metric, 99); + context.add_route_if_absent(source, route); } ControlPacket::PongPacket(pong_packet) => { + context.update_read_time(&source, route_key); let current_time = Local::now().timestamp_millis() as u16; if current_time < pong_packet.time() { return Ok(()); } let rt = (current_time - pong_packet.time()) as i64; - let metric = net_packet.source_ttl() - net_packet.ttl() + 1; - if let Some(current_route) = self.channel.route(&source) { - if ¤t_route.route_key() == route_key { - self.channel.update_route(&source, metric, rt); - } else if current_route.metric >= metric && current_route.rt > rt { - let route = Route::from(*route_key, metric, rt); - self.channel.add_route(source, route); - } - } else { - let route = Route::from(*route_key, metric, rt); - self.channel.add_route(source, route); - } + let route = Route::from(*route_key, metric, rt); + context.add_route(source, route); if route_key.addr == current_device.connect_server && source == current_device.virtual_gateway() { let epoch = self.device_list.lock().0; if pong_packet.epoch() != epoch { @@ -384,7 +350,7 @@ impl RecvHandler { poll_device.first_set_ttl(MAX_TTL); poll_device.set_protocol(Protocol::Service); poll_device.set_transport_protocol(service_packet::Protocol::PollDeviceList.into()); - self.channel.send_to_route(poll_device.buffer(), route_key)?; + context.send_by_key(poll_device.buffer(), route_key).await?; } } } @@ -395,28 +361,28 @@ impl RecvHandler { net_packet.set_source(current_device.virtual_ip()); net_packet.set_destination(source); net_packet.first_set_ttl(1); - self.channel.send_to_route(net_packet.buffer(), route_key)?; - let route = Route::from(*route_key, 1, -1); - self.channel.add_route(source, route); + context.send_by_key(net_packet.buffer(), route_key).await?; + let route = Route::from(*route_key, metric, 99); + context.add_route_if_absent(source, route); } ControlPacket::PunchResponse => { // log::info!("PunchResponse route_key:{:?}",route_key); - let route = Route::from(*route_key, 1, -1); - self.channel.add_route(net_packet.source(), route); + let route = Route::from(*route_key, metric, 99); + context.add_route_if_absent(source, route); } } Ok(()) } - fn other_turn(&self, current_device: CurrentDeviceInfo, source: Ipv4Addr, net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> { - match turn_packet::Protocol::from(net_packet.transport_protocol()) { - turn_packet::Protocol::Punch => { + async fn other_turn(&self, context: &Context, current_device: CurrentDeviceInfo, source: Ipv4Addr, net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> { + match other_turn_packet::Protocol::from(net_packet.transport_protocol()) { + other_turn_packet::Protocol::Punch => { let punch_info = PunchInfo::parse_from_bytes(net_packet.payload())?; let public_ips = punch_info.public_ip_list. - iter().map(|v| { IpAddr::from(v.to_be_bytes()) }).collect(); + iter().map(|v| { Ipv4Addr::from(v.to_be_bytes()) }).collect(); let peer_nat_info = NatInfo::new(public_ips, punch_info.public_port as u16, punch_info.public_port_range as u16, - IpAddr::from(punch_info.local_ip.to_be_bytes()), + Ipv4Addr::from(punch_info.local_ip.to_be_bytes()), punch_info.local_port as u16, punch_info.nat_type.enum_value_or_default().into()); self.peer_nat_info_map.insert(source, peer_nat_info.clone()); @@ -424,20 +390,12 @@ impl RecvHandler { let mut punch_reply = PunchInfo::new(); punch_reply.reply = true; let nat_info = self.nat_test.nat_info(); - punch_reply.public_ip_list = nat_info.public_ips.iter().map(|i| { - match i { - IpAddr::V4(ip) => u32::from_be_bytes(ip.octets()), - IpAddr::V6(_) => 0 - } - }).collect(); + punch_reply.public_ip_list = nat_info.public_ips.iter().map(|ip| u32::from_be_bytes(ip.octets())).collect(); punch_reply.public_port = nat_info.public_port as u32; punch_reply.public_port_range = nat_info.public_port_range as u32; punch_reply.nat_type = protobuf::EnumOrUnknown::new(PunchNatType::from(nat_info.nat_type)); - punch_reply.local_ip = match nat_info.local_ip { - IpAddr::V4(ip) => u32::from_be_bytes(ip.octets()), - IpAddr::V6(_) => 0 - }; + punch_reply.local_ip = u32::from_be_bytes(nat_info.local_ip.octets()); punch_reply.local_port = nat_info.local_port as u32; let bytes = punch_reply.write_to_bytes()?; let mut net_packet = @@ -445,7 +403,7 @@ impl RecvHandler { net_packet.set_version(Version::V1); net_packet.set_protocol(Protocol::OtherTurn); net_packet.set_transport_protocol( - turn_packet::Protocol::Punch.into(), + other_turn_packet::Protocol::Punch.into(), ); net_packet.first_set_ttl(MAX_TTL); net_packet.set_source(current_device.virtual_ip()); @@ -459,21 +417,29 @@ impl RecvHandler { packet.set_transport_protocol(control_packet::Protocol::PunchRequest.into()); packet.set_source(current_device.virtual_ip()); packet.set_destination(source); - let _ = self.channel.send_to_addr(packet.buffer(), SocketAddr::new(peer_nat_info.local_ip, peer_nat_info.local_port)); + let _ = context.send_main(packet.buffer(), SocketAddr::V4(SocketAddrV4::new(peer_nat_info.local_ip, peer_nat_info.local_port))).await; } - if let Err(e) = self.channel.punch(source, peer_nat_info) { - log::warn!("发送到打洞通道失败 {:?}",e); - return Ok(()); + if self.punch(source, peer_nat_info).await { + context.send_by_key(net_packet.buffer(), route_key).await?; } - self.channel.send_to_route(net_packet.buffer(), route_key)?; } else { - let _ = self.channel.punch(source, peer_nat_info); + self.punch(source, peer_nat_info).await; } } - turn_packet::Protocol::UnKnow(e) => { + other_turn_packet::Protocol::Unknown(e) => { log::warn!("不支持的转发协议 {:?},source:{:?}",e,source); } } Ok(()) } + async fn punch(&self, peer_ip: Ipv4Addr, peer_nat_info: NatInfo) -> bool { + match peer_nat_info.nat_type { + NatType::Symmetric => { + self.symmetric_sender.try_send((peer_ip, peer_nat_info)).is_ok() + } + NatType::Cone => { + self.cone_sender.try_send((peer_ip, peer_nat_info)).is_ok() + } + } + } } \ No newline at end of file diff --git a/switch/src/handle/registration_handler.rs b/switch/src/handle/registration_handler.rs index b136b11..3ac6c5a 100644 --- a/switch/src/handle/registration_handler.rs +++ b/switch/src/handle/registration_handler.rs @@ -1,12 +1,12 @@ use std::io; -use std::net::{Ipv4Addr, SocketAddr}; +use std::net::SocketAddr; use std::sync::atomic::{AtomicI64, Ordering}; use std::time::Duration; use chrono::Local; -use p2p_channel::channel::sender::Sender; -use p2p_channel::channel::Channel; use protobuf::Message; +use tokio::net::UdpSocket; +use crate::channel::sender::ChannelSender; use crate::error::*; use crate::proto::message::{RegistrationRequest, RegistrationResponse}; @@ -14,8 +14,8 @@ use crate::protocol::error_packet::InErrorPacket; use crate::protocol::{service_packet, NetPacket, Protocol, Version, MAX_TTL}; ///向中继服务器注册，token标识一个虚拟网关，device_id防止多次注册时得到的ip不一致 -pub fn registration( - channel: &mut Channel, +pub async fn registration( + main_channel: &UdpSocket, server_address: SocketAddr, token: String, device_id: String, @@ -26,52 +26,60 @@ pub fn registration( let buf = request_packet.buffer(); let mut recv_buf = [0u8; 10240]; let mut count = 0; - let len = loop { - match channel.send_to_addr(buf, server_address) { + loop { + match main_channel.send_to(buf, server_address).await { Ok(_) => { - match channel.recv_from(&mut recv_buf, Some(Duration::from_millis(300))) { - Ok((len, route)) => { - if server_address == route.addr { - let net_packet = NetPacket::new(&recv_buf[..len])?; - match net_packet.protocol() { - Protocol::Service => { - match service_packet::Protocol::from(net_packet.transport_protocol()) { - service_packet::Protocol::RegistrationResponse => { - let response = RegistrationResponse::parse_from_bytes(net_packet.payload())?; - return Ok(response); + match tokio::time::timeout(Duration::from_millis(300), main_channel.recv_from(&mut recv_buf)).await { + Ok(rs) => { + match rs { + Ok((len, addr)) => { + if server_address == addr { + let net_packet = NetPacket::new(&recv_buf[..len])?; + match net_packet.protocol() { + Protocol::Service => { + match service_packet::Protocol::from(net_packet.transport_protocol()) { + service_packet::Protocol::RegistrationResponse => { + let response = RegistrationResponse::parse_from_bytes(net_packet.payload())?; + return Ok(response); + } + _ => println!("响应数据错误"), + } + } + Protocol::Error => { + match InErrorPacket::new(net_packet.transport_protocol(), net_packet.payload()) { + Ok(e) => match e { + InErrorPacket::TokenError => return Err(Error::Stop("token错误".to_string())), + InErrorPacket::Disconnect => { + println!("断开连接"); + } + InErrorPacket::AddressExhausted => { + println!("地址用尽"); + log::warn!("地址用尽"); + } + InErrorPacket::OtherError(e) => match e.message() { + Ok(str) => { + println!("其他异常:{:?}", str); + log::warn!("其他异常{:?}",str); + } + Err(e) => println!("其他异常:{:?}", e), + }, + }, + Err(e) => println!("数据解析异常:{:?}", e), + } } _ => println!("响应数据错误"), - } + }; } - Protocol::Error => { - match InErrorPacket::new(net_packet.transport_protocol(), net_packet.payload()) { - Ok(e) => match e { - InErrorPacket::TokenError => return Err(Error::Stop("token错误".to_string())), - InErrorPacket::Disconnect => { - println!("断开连接"); - } - InErrorPacket::AddressExhausted => { - println!("地址用尽"); - log::warn!("地址用尽"); - } - InErrorPacket::OtherError(e) => match e.message() { - Ok(str) => { - println!("其他异常:{:?}", str); - log::warn!("其他异常{:?}",str); - } - Err(e) => println!("其他异常:{:?}", e), - }, - }, - Err(e) => println!("数据解析异常:{:?}", e), - } - } - _ => println!("响应数据错误"), - }; + } + Err(e) => { + println!("接收服务器数据失败:{:?}", e); + log::warn!("接收服务器数据失败:{:?}",e); + } } } - Err(e) => { - println!("接收服务器数据失败:{:?}", e); - log::warn!("接收服务器数据失败:{:?}",e); + Err(_) => { + println!("接收超时"); + log::warn!("接收超时"); } } } @@ -97,6 +105,7 @@ fn registration_request_packet( request.device_id = device_id; request.name = name; request.is_fast = is_fast; + request.version = "1.0.6".to_string(); let bytes = request.write_to_bytes()?; let buf = vec![0u8; 12 + bytes.len()]; let mut net_packet = NetPacket::new(buf)?; @@ -109,7 +118,7 @@ fn registration_request_packet( } pub struct Register { - sender: Sender, + sender: ChannelSender, server_address: SocketAddr, token: String, device_id: String, @@ -119,7 +128,7 @@ pub struct Register { impl Register { pub fn new( - sender: Sender, + sender: ChannelSender, server_address: SocketAddr, token: String, device_id: String, @@ -134,7 +143,7 @@ impl Register { time: AtomicI64::new(0), } } - pub fn fast_register(&self) -> io::Result<()> { + pub async fn fast_register(&self) -> io::Result<()> { let last = self.time.load(Ordering::Relaxed); let new = Local::now().timestamp_millis(); if new - last < 1000 @@ -155,7 +164,7 @@ impl Register { ) .unwrap(); let buf = request_packet.buffer(); - self.sender.send_to_addr(buf, self.server_address)?; + self.sender.send_main(buf, self.server_address).await?; Ok(()) } } diff --git a/switch/src/handle/tap_handler.rs b/switch/src/handle/tap_handler.rs deleted file mode 100644 index af7bcd5..0000000 --- a/switch/src/handle/tap_handler.rs +++ /dev/null @@ -1,163 +0,0 @@ -use std::net::{Ipv4Addr, SocketAddrV4}; -use std::sync::Arc; -use std::{io, thread}; -use crossbeam::atomic::AtomicCell; -use p2p_channel::channel::sender::Sender; -use packet::arp::arp::ArpPacket; -use packet::ethernet; -use packet::ethernet::packet::EthernetPacket; -use packet::icmp::icmp::IcmpPacket; -use packet::icmp::Kind; -use packet::ip::ipv4; -use packet::ip::ipv4::packet::IpV4Packet; -use crate::external_route::ExternalRoute; -use crate::handle::{check_dest, CurrentDeviceInfo}; -use crate::ip_proxy::IpProxyMap; -use crate::protocol::{MAX_TTL, NetPacket, Protocol, Version}; -use crate::tap_device::{TapReader, TapWriter}; - -pub fn start(sender: Sender, - tap_reader: TapReader, - tap_writer: TapWriter, - current_device: Arc>, - ip_route: ExternalRoute, - ip_proxy_map: IpProxyMap) { - thread::Builder::new().name("tap-handler".into()).spawn(move || { - if let Err(e) = start_(sender, tap_reader, tap_writer, current_device,ip_route,ip_proxy_map) { - log::warn!("{:?}",e); - } - }).unwrap(); -} - -fn start_(sender: Sender, - tap_reader: TapReader, - tap_writer: TapWriter, - current_device: Arc>, - ip_route: ExternalRoute, - ip_proxy_map: IpProxyMap) -> io::Result<()> { - let mut net_packet = NetPacket::new(vec![0u8; 4 + 8 + 1500]).unwrap(); - net_packet.set_version(Version::V1); - net_packet.set_protocol(Protocol::Ipv4Turn); - net_packet.set_transport_protocol(ipv4::protocol::Protocol::Ipv4.into()); - net_packet.set_ttl(MAX_TTL); - let mut buf = [0; 2048]; - loop { - let len = tap_reader.read(&mut buf)?; - if len == 0 { - continue; - } - let mut ethernet_packet = EthernetPacket::unchecked(&mut buf[..len]); - if let Err(e) = handle(&mut net_packet, ¤t_device, &tap_writer, &mut ethernet_packet, &sender,&ip_route,&ip_proxy_map) { - log::error!("tap handle{:?}",e); - } - } -} - -fn handle(net_packet: &mut NetPacket>, current_device: &AtomicCell, tap_writer: &TapWriter, ethernet_packet: &mut EthernetPacket<&mut [u8]>, sender: &Sender, ip_route: &ExternalRoute, proxy_map: &IpProxyMap) -> crate::Result<()> { - let current_device = current_device.load(); - match ethernet_packet.protocol() { - ethernet::protocol::Protocol::Arp => { - let mut out_ethernet_packet = EthernetPacket::unchecked(ethernet_packet.buffer.to_vec()); - let arp_packet = ArpPacket::unchecked(ethernet_packet.payload()); - let mut out_arp_packet = ArpPacket::unchecked(out_ethernet_packet.payload_mut()); - let sender_h = arp_packet.sender_hardware_addr(); - let sender_p = arp_packet.sender_protocol_addr(); - let target_p = arp_packet.target_protocol_addr(); - if target_p == &[0, 0, 0, 0] || sender_p == &[0, 0, 0, 0] || target_p == sender_p { - return Ok(()); - } - //回复一个虚假的MAC地址 - out_arp_packet.set_sender_hardware_addr(&[target_p[0], target_p[1], target_p[2], target_p[3], 123, 234]); - out_arp_packet.set_sender_protocol_addr(target_p); - out_arp_packet.set_target_hardware_addr(sender_h); - out_arp_packet.set_target_protocol_addr(sender_p); - out_arp_packet.set_op_code(2); - out_ethernet_packet.set_source(&[target_p[0], target_p[1], target_p[2], target_p[3], 123, 234]); - out_ethernet_packet.set_destination(sender_h); - - tap_writer.write(&out_ethernet_packet.buffer)?; - } - ethernet::protocol::Protocol::Ipv4 => { - // println!("in ethernet_packet {:?}", ethernet_packet); - let mut ipv4_packet = IpV4Packet::unchecked(ethernet_packet.payload_mut()); - let src_ip = ipv4_packet.source_ip(); - let mut dest_ip = ipv4_packet.destination_ip(); - if src_ip != current_device.virtual_ip() { - return Ok(()); - } - if !check_dest(dest_ip, current_device.virtual_netmask, current_device.virtual_network) && !dest_ip.is_broadcast() { - if let Some(r_dest_ip) = ip_route.route(&dest_ip) { - //路由的目标不能是自己 - if r_dest_ip == src_ip { - return Ok(()); - } - dest_ip = r_dest_ip; - } else { - return Ok(()); - } - }else{ - match ipv4_packet.protocol() { - ipv4::protocol::Protocol::Tcp => { - let dest_addr = { - let tcp_packet = packet::tcp::tcp::TcpPacket::new(src_ip, dest_ip, ipv4_packet.payload())?; - SocketAddrV4::new(dest_ip, tcp_packet.destination_port()) - }; - if let Some(entry) = proxy_map.tcp_proxy_map.get(&dest_addr) { - let source_addr = entry.value().1; - let source_ip = *source_addr.ip(); - let mut tcp_packet = packet::tcp::tcp::TcpPacket::new(source_ip, dest_ip, ipv4_packet.payload_mut())?; - tcp_packet.set_source_port(source_addr.port()); - tcp_packet.update_checksum(); - ipv4_packet.set_source_ip(source_ip); - ipv4_packet.update_checksum(); - } - } - ipv4::protocol::Protocol::Udp => { - let dest_addr = { - let udp_packet = packet::udp::udp::UdpPacket::new(src_ip, dest_ip, ipv4_packet.payload())?; - SocketAddrV4::new(dest_ip, udp_packet.destination_port()) - }; - if let Some(entry) = proxy_map.udp_proxy_map.get(&dest_addr) { - let source_addr = entry.value().1; - let source_ip = *source_addr.ip(); - let mut udp_packet = packet::udp::udp::UdpPacket::new(src_ip, dest_ip, ipv4_packet.payload_mut())?; - udp_packet.set_source_port(source_addr.port()); - udp_packet.update_checksum(); - ipv4_packet.set_source_ip(source_ip); - ipv4_packet.update_checksum(); - } - } - _ => {} - } - } - if src_ip == dest_ip { - if ipv4_packet.protocol() == ipv4::protocol::Protocol::Icmp { - let mut icmp = IcmpPacket::unchecked(ipv4_packet.payload_mut()); - if icmp.kind() == Kind::EchoRequest { - icmp.set_kind(Kind::EchoReply); - icmp.update_checksum(); - let src = ipv4_packet.source_ip(); - ipv4_packet.set_source_ip(ipv4_packet.destination_ip()); - ipv4_packet.set_destination_ip(src); - ipv4_packet.update_checksum(); - tap_writer.write(ethernet_packet.buffer)?; - return Ok(()); - } - } - } - net_packet.set_source(src_ip); - net_packet.set_destination(dest_ip); - let data_len = ipv4_packet.buffer.len(); - net_packet.set_payload(ipv4_packet.buffer); - //优先发到直连到地址 - if sender.send_to_id(&net_packet.buffer()[..(12 + data_len)], &dest_ip).is_err() { - sender.send_to_addr(&net_packet.buffer()[..(12 + data_len)], current_device.connect_server)?; - } - } - _ => { - // log::warn!("不支持的二层协议：{:?}",p) - } - } - Ok(()) -} - diff --git a/switch/src/handle/tun_handler.rs b/switch/src/handle/tun_handler.rs deleted file mode 100644 index 4a7c6fe..0000000 --- a/switch/src/handle/tun_handler.rs +++ /dev/null @@ -1,175 +0,0 @@ -use std::{io, thread}; -use std::net::{Ipv4Addr, SocketAddrV4}; -use std::sync::Arc; - -use crossbeam::atomic::AtomicCell; - -use p2p_channel::channel::sender::Sender; -use packet::icmp:: Kind; -use packet::icmp::icmp:: IcmpPacket; -use packet::ip::ipv4; -use packet::ip::ipv4::packet::IpV4Packet; - -use crate::error::*; -use crate::external_route::ExternalRoute; -use crate::handle::{check_dest, CurrentDeviceInfo}; -use crate::ip_proxy::IpProxyMap; -use crate::protocol::{MAX_TTL, NetPacket, Protocol, Version}; -use crate::tun_device::{TunReader, TunWriter}; - -fn icmp(tun_writer: &TunWriter, mut ipv4_packet: IpV4Packet<&mut [u8]>) -> Result<()> { - if ipv4_packet.protocol() == ipv4::protocol::Protocol::Icmp { - let mut icmp = IcmpPacket::new(ipv4_packet.payload_mut())?; - if icmp.kind() == Kind::EchoRequest { - icmp.set_kind(Kind::EchoReply); - icmp.update_checksum(); - let src = ipv4_packet.source_ip(); - ipv4_packet.set_source_ip(ipv4_packet.destination_ip()); - ipv4_packet.set_destination_ip(src); - ipv4_packet.update_checksum(); - tun_writer.write(ipv4_packet.buffer)?; - } - } - Ok(()) -} - -/// 接收tun数据，并且转发到udp上 -#[inline] -fn handle(sender: &Sender, data: &mut [u8], tun_writer: &TunWriter, current_device: CurrentDeviceInfo, net_packet: &mut NetPacket<[u8; 1512]>, ip_route: &ExternalRoute, proxy_map: &IpProxyMap) -> Result<()> { - let data_len = data.len(); - let mut ipv4_packet = match IpV4Packet::new(data) { - Ok(ipv4_packet) => ipv4_packet, - Err(packet::error::Error::Unimplemented) => { - return Ok(()); - } - Err(e) => Err(e)?, - }; - let src_ip = ipv4_packet.source_ip(); - let mut dest_ip = ipv4_packet.destination_ip(); - // if dest_ip == cur_info.broadcast_address { - // // 启动服务后会收到对137端口的广播 - // // 137端口是在局域网中提供计算机的名字或IP地址查询服务 - // return Ok(()); - // } - if src_ip != current_device.virtual_ip() { - return Ok(()); - } - if src_ip == dest_ip { - return icmp(&tun_writer, ipv4_packet); - } - if !check_dest(dest_ip, current_device.virtual_netmask, current_device.virtual_network) && !dest_ip.is_broadcast() { - // println!("非目标 {:?}",ipv4_packet); - if let Some(r_dest_ip) = ip_route.route(&dest_ip) { - //路由的目标不能是自己 - if r_dest_ip == src_ip { - return Ok(()); - } - dest_ip = r_dest_ip; - } else { - return Ok(()); - } - } else { - match ipv4_packet.protocol() { - ipv4::protocol::Protocol::Tcp => { - let dest_addr = { - let tcp_packet = packet::tcp::tcp::TcpPacket::new(src_ip, dest_ip, ipv4_packet.payload())?; - SocketAddrV4::new(dest_ip, tcp_packet.destination_port()) - }; - if let Some(entry) = proxy_map.tcp_proxy_map.get(&dest_addr) { - let source_addr = entry.value().1; - let source_ip = *source_addr.ip(); - let mut tcp_packet = packet::tcp::tcp::TcpPacket::new(source_ip, dest_ip, ipv4_packet.payload_mut())?; - tcp_packet.set_source_port(source_addr.port()); - tcp_packet.update_checksum(); - ipv4_packet.set_source_ip(source_ip); - ipv4_packet.update_checksum(); - } - } - ipv4::protocol::Protocol::Udp => { - let dest_addr = { - let udp_packet = packet::udp::udp::UdpPacket::new(src_ip, dest_ip, ipv4_packet.payload())?; - SocketAddrV4::new(dest_ip, udp_packet.destination_port()) - }; - if let Some(entry) = proxy_map.udp_proxy_map.get(&dest_addr) { - let source_addr = entry.value().1; - let source_ip = *source_addr.ip(); - let mut udp_packet = packet::udp::udp::UdpPacket::new(source_ip, dest_ip, ipv4_packet.payload_mut())?; - udp_packet.set_source_port(source_addr.port()); - udp_packet.update_checksum(); - ipv4_packet.set_source_ip(source_ip); - ipv4_packet.update_checksum(); - } - } - _ => {} - } - } - - net_packet.set_source(src_ip); - net_packet.set_destination(dest_ip); - net_packet.set_payload(ipv4_packet.buffer); - //优先发到直连到地址 - if sender.send_to_id(&net_packet.buffer()[..(12 + data_len)], &dest_ip).is_err() { - sender.send_to_addr(&net_packet.buffer()[..(12 + data_len)], current_device.connect_server)?; - } - return Ok(()); -} - -pub fn start(sender: Sender, - tun_reader: TunReader, - tun_writer: TunWriter, - current_device: Arc>, - ip_route: ExternalRoute, - ip_proxy_map: IpProxyMap) { - thread::Builder::new().name("tun-handler".into()).spawn(move || { - if let Err(e) = start_(sender, tun_reader, tun_writer, current_device, ip_route, ip_proxy_map) { - log::warn!("{:?}",e); - } - }).unwrap(); -} - -#[cfg(target_os = "windows")] -fn start_(sender: Sender, - tun_reader: TunReader, - tun_writer: TunWriter, - current_device: Arc>, - ip_route: ExternalRoute, - ip_proxy_map: IpProxyMap) -> io::Result<()> { - let mut net_packet = NetPacket::new([0u8; 4 + 8 + 1500])?; - net_packet.set_version(Version::V1); - net_packet.set_protocol(Protocol::Ipv4Turn); - net_packet.set_transport_protocol(ipv4::protocol::Protocol::Ipv4.into()); - net_packet.set_ttl(MAX_TTL); - loop { - let mut data = tun_reader.next()?; - match handle(&sender, data.bytes_mut(), &tun_writer, current_device.load(), &mut net_packet, &ip_route, &ip_proxy_map) { - Ok(_) => {} - Err(e) => { - log::warn!("{:?}", e) - } - } - } -} - -#[cfg(any(target_os = "linux", target_os = "macos"))] -fn start_(sender: Sender, - tun_reader: TunReader, - tun_writer: TunWriter, - current_device: Arc>, - ip_route: ExternalRoute, - ip_proxy_map: IpProxyMap) -> io::Result<()> { - let mut net_packet = NetPacket::new([0u8; 4 + 8 + 1500])?; - net_packet.set_version(Version::V1); - net_packet.set_protocol(Protocol::Ipv4Turn); - net_packet.set_transport_protocol(ipv4::protocol::Protocol::Ipv4.into()); - net_packet.set_ttl(MAX_TTL); - let mut buf = [0; 4096]; - loop { - let len = tun_reader.read(&mut buf)?; - match handle(&sender, &mut buf[..len], &tun_writer, current_device.load(), &mut net_packet, &ip_route, &ip_proxy_map) { - Ok(_) => {} - Err(e) => { - log::warn!("{:?}", e) - } - } - } -} \ No newline at end of file diff --git a/switch/src/ip_proxy/icmp_proxy.rs b/switch/src/ip_proxy/icmp_proxy.rs index d74087d..0dda487 100644 --- a/switch/src/ip_proxy/icmp_proxy.rs +++ b/switch/src/ip_proxy/icmp_proxy.rs @@ -6,11 +6,11 @@ use crossbeam::atomic::AtomicCell; use crossbeam_skiplist::SkipMap; use socket2::{Domain, SockAddr, Socket, Type}; -use p2p_channel::channel::sender::Sender; use packet::icmp::icmp; use packet::icmp::icmp::HeaderOther; use packet::ip::ipv4; +use crate::channel::sender::ChannelSender; use crate::handle::CurrentDeviceInfo; use crate::protocol::{MAX_TTL, NetPacket, Protocol, Version}; @@ -18,12 +18,12 @@ pub struct IcmpProxy { icmp_socket: Arc, // 对端-> 真实来源 icmp_proxy_map: Arc>, - sender: Sender, + sender: ChannelSender, current_device: Arc>, } impl IcmpProxy { - pub fn new(addr: SocketAddrV4, icmp_proxy_map: Arc>, sender: Sender, current_device: Arc>) -> io::Result { + pub fn new(addr: SocketAddrV4, icmp_proxy_map: Arc>, sender: ChannelSender, current_device: Arc>) -> io::Result { let icmp_socket = Arc::new(Socket::new(Domain::IPV4, Type::RAW, Some(socket2::Protocol::ICMPV4))?); icmp_socket.bind(&SockAddr::from(addr))?; // // 设置 SIO_RCVALL 参数 @@ -66,7 +66,7 @@ impl IcmpProxy { unsafe { std::mem::transmute(&mut buf[..]) }; let mut net_packet = NetPacket::new([0u8; 4 + 8 + 1500]).unwrap(); net_packet.set_version(Version::V1); - net_packet.set_protocol(Protocol::Ipv4Turn); + net_packet.set_protocol(Protocol::IpTurn); net_packet.set_transport_protocol(ipv4::protocol::Protocol::Ipv4.into()); net_packet.set_ttl(MAX_TTL); loop { @@ -90,7 +90,7 @@ impl IcmpProxy { net_packet.set_destination(dest_ip); let data_len = ipv4_packet.buffer.len(); net_packet.set_payload(ipv4_packet.buffer); - let _ = self.sender.send_to_id(&net_packet.buffer()[..(12 + data_len)], &dest_ip); + let _ = self.sender.try_send_by_id(&net_packet.buffer()[..(12 + data_len)], &dest_ip); } } _ => { diff --git a/switch/src/ip_proxy/mod.rs b/switch/src/ip_proxy/mod.rs index c75c7f8..abfced0 100644 --- a/switch/src/ip_proxy/mod.rs +++ b/switch/src/ip_proxy/mod.rs @@ -6,7 +6,7 @@ use crossbeam::atomic::AtomicCell; use crossbeam_skiplist::SkipMap; use socket2::{SockAddr, Socket}; use tokio::net::{TcpListener, UdpSocket}; -use p2p_channel::channel::sender::Sender; +use crate::channel::sender::ChannelSender; use crate::handle::CurrentDeviceInfo; use crate::ip_proxy::icmp_proxy::IcmpProxy; use crate::ip_proxy::tcp_proxy::TcpProxy; @@ -45,34 +45,34 @@ impl IpProxyMap { } } -pub async fn init_proxy(sender: Sender, bind_ips: Vec, current_device: Arc>) -> io::Result { +pub async fn init_proxy(sender: ChannelSender, bind_ips: Vec, current_device: Arc>) -> io::Result { let mut icmp_sockets = HashMap::new(); let tcp_proxy_map: Arc> = Arc::new(SkipMap::new()); let udp_proxy_map: Arc> = Arc::new(SkipMap::new()); let icmp_proxy_map: Arc> = Arc::new(SkipMap::new()); - let tcp_listener = TcpListener::bind("0.0.0.0:0").await?; - let udp_socket = UdpSocket::bind("0.0.0.0:0").await?; - let tcp_proxy_port = tcp_listener.local_addr()?.port(); - let udp_proxy_port = udp_socket.local_addr()?.port(); - - { + let (tcp_proxy_port, udp_proxy_port) = if !bind_ips.is_empty() { + let tcp_listener = TcpListener::bind("0.0.0.0:0").await?; + let udp_socket = UdpSocket::bind("0.0.0.0:0").await?; + let tcp_proxy_port = tcp_listener.local_addr()?.port(); + let udp_proxy_port = udp_socket.local_addr()?.port(); let tcp_proxy_map = tcp_proxy_map.clone(); tokio::spawn(async { let tcp_proxy = TcpProxy::new(tcp_listener, tcp_proxy_map); tcp_proxy.start().await }); - } - { let udp_proxy_map = udp_proxy_map.clone(); tokio::spawn(async { let udp_proxy = UdpProxy::new(udp_socket, udp_proxy_map); udp_proxy.start().await }); - } + (tcp_proxy_port, udp_proxy_port) + } else { + (0, 0) + }; for ip in bind_ips { let addr = SocketAddrV4::new(ip, 0); let icmp_proxy_map = icmp_proxy_map.clone(); - let icmp_proxy = IcmpProxy::new(addr, icmp_proxy_map, sender.try_clone()?, current_device.clone())?; + let icmp_proxy = IcmpProxy::new(addr, icmp_proxy_map, sender.clone(), current_device.clone())?; icmp_sockets.insert(ip, icmp_proxy.icmp_socket()); thread::spawn(move || { icmp_proxy.start(); diff --git a/switch/src/lib.rs b/switch/src/lib.rs index 547a74a..eebd814 100644 --- a/switch/src/lib.rs +++ b/switch/src/lib.rs @@ -1,7 +1,5 @@ use crate::error::Error; -pub use p2p_channel::channel::{Route, RouteKey}; - pub type Result = std::result::Result; pub mod error; @@ -9,8 +7,9 @@ pub mod handle; pub mod nat; pub mod proto; pub mod protocol; -pub mod tun_device; -pub mod tap_device; pub mod ip_proxy; pub mod external_route; +pub mod igmp_server; +pub mod tun_tap_device; pub mod core; +pub mod channel; diff --git a/switch/src/nat/check.rs b/switch/src/nat/check.rs index d7dae7a..d417d7f 100644 --- a/switch/src/nat/check.rs +++ b/switch/src/nat/check.rs @@ -1,8 +1,8 @@ -use p2p_channel::punch::NatType; use std::collections::HashSet; use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket}; use std::time::Duration; use std::{io, thread}; +use crate::channel::punch::NatType; // #[derive(Debug, Copy, Clone, PartialEq)] // pub enum NatType { diff --git a/switch/src/nat/mod.rs b/switch/src/nat/mod.rs index 9743e2b..ab58053 100644 --- a/switch/src/nat/mod.rs +++ b/switch/src/nat/mod.rs @@ -1,5 +1,4 @@ use crate::proto::message::PunchNatType; -use p2p_channel::punch::{NatInfo, NatType}; use parking_lot::Mutex; use std::io; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; @@ -8,12 +7,20 @@ use std::sync::Arc; pub mod check; use std::net::UdpSocket; +use crate::channel::punch::{NatInfo, NatType}; -pub fn local_ip() -> io::Result { +pub fn local_ip() -> io::Result { let socket = UdpSocket::bind("0.0.0.0:0")?; socket.connect("8.8.8.8:80")?; let addr = socket.local_addr()?; - Ok(addr.ip()) + match addr.ip() { + IpAddr::V4(ip) => { + Ok(ip) + } + IpAddr::V6(_) => { + Ok(Ipv4Addr::UNSPECIFIED) + } + } } #[derive(Clone)] @@ -45,7 +52,7 @@ impl NatTest { nat_test_server: Vec, public_ip: Ipv4Addr, public_port: u16, - local_ip: IpAddr, + local_ip: Ipv4Addr, local_port: u16, ) -> NatTest { let info = NatTest::re_test_( @@ -67,7 +74,7 @@ impl NatTest { &self, public_ip: Ipv4Addr, public_port: u16, - local_ip: IpAddr, + local_ip: Ipv4Addr, local_port: u16, ) -> NatInfo { let info = NatTest::re_test_( @@ -84,16 +91,16 @@ impl NatTest { nat_test_server: &Vec, public_ip: Ipv4Addr, public_port: u16, - local_ip: IpAddr, + local_ip: Ipv4Addr, local_port: u16, ) -> NatInfo { return match check::public_ip_list(nat_test_server) { Ok((nat_type, ips, port_range)) => { let mut public_ips = Vec::new(); - public_ips.push(IpAddr::from(public_ip)); + public_ips.push(Ipv4Addr::from(public_ip)); for ip in ips { if ip != public_ip { - public_ips.push(IpAddr::from(ip)); + public_ips.push(ip); } } NatInfo::new( @@ -108,7 +115,7 @@ impl NatTest { Err(e) => { log::warn!("{:?}", e); NatInfo::new( - vec![IpAddr::from(public_ip)], + vec![public_ip], public_port, 0, local_ip, diff --git a/switch/src/proto/message.rs b/switch/src/proto/message.rs index 7164962..d082655 100644 --- a/switch/src/proto/message.rs +++ b/switch/src/proto/message.rs @@ -37,6 +37,8 @@ pub struct RegistrationRequest { pub name: ::std::string::String, // @@protoc_insertion_point(field:RegistrationRequest.is_fast) pub is_fast: bool, + // @@protoc_insertion_point(field:RegistrationRequest.version) + pub version: ::std::string::String, // special fields // @@protoc_insertion_point(special_field:RegistrationRequest.special_fields) pub special_fields: ::protobuf::SpecialFields, @@ -54,7 +56,7 @@ impl RegistrationRequest { } fn generated_message_descriptor_data() -> ::protobuf::reflect::GeneratedMessageDescriptorData { - let mut fields = ::std::vec::Vec::with_capacity(4); + let mut fields = ::std::vec::Vec::with_capacity(5); let mut oneofs = ::std::vec::Vec::with_capacity(0); fields.push(::protobuf::reflect::rt::v2::make_simpler_field_accessor::<_, _>( "token", @@ -76,6 +78,11 @@ impl RegistrationRequest { |m: &RegistrationRequest| { &m.is_fast }, |m: &mut RegistrationRequest| { &mut m.is_fast }, )); + fields.push(::protobuf::reflect::rt::v2::make_simpler_field_accessor::<_, _>( + "version", + |m: &RegistrationRequest| { &m.version }, + |m: &mut RegistrationRequest| { &mut m.version }, + )); ::protobuf::reflect::GeneratedMessageDescriptorData::new_2::( "RegistrationRequest", fields, @@ -106,6 +113,9 @@ impl ::protobuf::Message for RegistrationRequest { 32 => { self.is_fast = is.read_bool()?; }, + 42 => { + self.version = is.read_string()?; + }, tag => { ::protobuf::rt::read_unknown_or_skip_group(tag, is, self.special_fields.mut_unknown_fields())?; }, @@ -130,6 +140,9 @@ impl ::protobuf::Message for RegistrationRequest { if self.is_fast != false { my_size += 1 + 1; } + if !self.version.is_empty() { + my_size += ::protobuf::rt::string_size(5, &self.version); + } my_size += ::protobuf::rt::unknown_fields_size(self.special_fields.unknown_fields()); self.special_fields.cached_size().set(my_size as u32); my_size @@ -148,6 +161,9 @@ impl ::protobuf::Message for RegistrationRequest { if self.is_fast != false { os.write_bool(4, self.is_fast)?; } + if !self.version.is_empty() { + os.write_string(5, &self.version)?; + } os.write_unknown_fields(self.special_fields.unknown_fields())?; ::std::result::Result::Ok(()) } @@ -169,6 +185,7 @@ impl ::protobuf::Message for RegistrationRequest { self.device_id.clear(); self.name.clear(); self.is_fast = false; + self.version.clear(); self.special_fields.clear(); } @@ -178,6 +195,7 @@ impl ::protobuf::Message for RegistrationRequest { device_id: ::std::string::String::new(), name: ::std::string::String::new(), is_fast: false, + version: ::std::string::String::new(), special_fields: ::protobuf::SpecialFields::new(), }; &instance @@ -1017,28 +1035,29 @@ impl PunchNatType { } static file_descriptor_proto_data: &'static [u8] = b"\ - \n\rmessage.proto\"u\n\x13RegistrationRequest\x12\x14\n\x05token\x18\x01\ - \x20\x01(\tR\x05token\x12\x1b\n\tdevice_id\x18\x02\x20\x01(\tR\x08device\ - Id\x12\x12\n\x04name\x18\x03\x20\x01(\tR\x04name\x12\x17\n\x07is_fast\ - \x18\x04\x20\x01(\x08R\x06isFast\"\x92\x02\n\x14RegistrationResponse\x12\ - \x1d\n\nvirtual_ip\x18\x01\x20\x01(\x07R\tvirtualIp\x12'\n\x0fvirtual_ga\ - teway\x18\x02\x20\x01(\x07R\x0evirtualGateway\x12'\n\x0fvirtual_netmask\ - \x18\x03\x20\x01(\x07R\x0evirtualNetmask\x12\x14\n\x05epoch\x18\x04\x20\ - \x01(\rR\x05epoch\x125\n\x10device_info_list\x18\x05\x20\x03(\x0b2\x0b.D\ - eviceInfoR\x0edeviceInfoList\x12\x1b\n\tpublic_ip\x18\x06\x20\x01(\x07R\ - \x08publicIp\x12\x1f\n\x0bpublic_port\x18\x07\x20\x01(\rR\npublicPort\"d\ - \n\nDeviceInfo\x12\x12\n\x04name\x18\x01\x20\x01(\tR\x04name\x12\x1d\n\n\ - virtual_ip\x18\x02\x20\x01(\x07R\tvirtualIp\x12#\n\rdevice_status\x18\ - \x03\x20\x01(\rR\x0cdeviceStatus\"Y\n\nDeviceList\x12\x14\n\x05epoch\x18\ - \x01\x20\x01(\rR\x05epoch\x125\n\x10device_info_list\x18\x02\x20\x03(\ - \x0b2\x0b.DeviceInfoR\x0edeviceInfoList\"\xf8\x01\n\tPunchInfo\x12$\n\ - \x0epublic_ip_list\x18\x02\x20\x03(\x07R\x0cpublicIpList\x12\x1f\n\x0bpu\ - blic_port\x18\x03\x20\x01(\rR\npublicPort\x12*\n\x11public_port_range\ - \x18\x04\x20\x01(\rR\x0fpublicPortRange\x12(\n\x08nat_type\x18\x05\x20\ - \x01(\x0e2\r.PunchNatTypeR\x07natType\x12\x14\n\x05reply\x18\x06\x20\x01\ - (\x08R\x05reply\x12\x19\n\x08local_ip\x18\x07\x20\x01(\x07R\x07localIp\ - \x12\x1d\n\nlocal_port\x18\x08\x20\x01(\rR\tlocalPort*'\n\x0cPunchNatTyp\ - e\x12\r\n\tSymmetric\x10\0\x12\x08\n\x04Cone\x10\x01b\x06proto3\ + \n\rmessage.proto\"\x8f\x01\n\x13RegistrationRequest\x12\x14\n\x05token\ + \x18\x01\x20\x01(\tR\x05token\x12\x1b\n\tdevice_id\x18\x02\x20\x01(\tR\ + \x08deviceId\x12\x12\n\x04name\x18\x03\x20\x01(\tR\x04name\x12\x17\n\x07\ + is_fast\x18\x04\x20\x01(\x08R\x06isFast\x12\x18\n\x07version\x18\x05\x20\ + \x01(\tR\x07version\"\x92\x02\n\x14RegistrationResponse\x12\x1d\n\nvirtu\ + al_ip\x18\x01\x20\x01(\x07R\tvirtualIp\x12'\n\x0fvirtual_gateway\x18\x02\ + \x20\x01(\x07R\x0evirtualGateway\x12'\n\x0fvirtual_netmask\x18\x03\x20\ + \x01(\x07R\x0evirtualNetmask\x12\x14\n\x05epoch\x18\x04\x20\x01(\rR\x05e\ + poch\x125\n\x10device_info_list\x18\x05\x20\x03(\x0b2\x0b.DeviceInfoR\ + \x0edeviceInfoList\x12\x1b\n\tpublic_ip\x18\x06\x20\x01(\x07R\x08publicI\ + p\x12\x1f\n\x0bpublic_port\x18\x07\x20\x01(\rR\npublicPort\"d\n\nDeviceI\ + nfo\x12\x12\n\x04name\x18\x01\x20\x01(\tR\x04name\x12\x1d\n\nvirtual_ip\ + \x18\x02\x20\x01(\x07R\tvirtualIp\x12#\n\rdevice_status\x18\x03\x20\x01(\ + \rR\x0cdeviceStatus\"Y\n\nDeviceList\x12\x14\n\x05epoch\x18\x01\x20\x01(\ + \rR\x05epoch\x125\n\x10device_info_list\x18\x02\x20\x03(\x0b2\x0b.Device\ + InfoR\x0edeviceInfoList\"\xf8\x01\n\tPunchInfo\x12$\n\x0epublic_ip_list\ + \x18\x02\x20\x03(\x07R\x0cpublicIpList\x12\x1f\n\x0bpublic_port\x18\x03\ + \x20\x01(\rR\npublicPort\x12*\n\x11public_port_range\x18\x04\x20\x01(\rR\ + \x0fpublicPortRange\x12(\n\x08nat_type\x18\x05\x20\x01(\x0e2\r.PunchNatT\ + ypeR\x07natType\x12\x14\n\x05reply\x18\x06\x20\x01(\x08R\x05reply\x12\ + \x19\n\x08local_ip\x18\x07\x20\x01(\x07R\x07localIp\x12\x1d\n\nlocal_por\ + t\x18\x08\x20\x01(\rR\tlocalPort*'\n\x0cPunchNatType\x12\r\n\tSymmetric\ + \x10\0\x12\x08\n\x04Cone\x10\x01b\x06proto3\ "; /// `FileDescriptorProto` object which was a source for this generated file diff --git a/switch/src/protocol/control_packet.rs b/switch/src/protocol/control_packet.rs index 32b36cd..4cc67ca 100644 --- a/switch/src/protocol/control_packet.rs +++ b/switch/src/protocol/control_packet.rs @@ -17,7 +17,7 @@ pub enum Protocol { PunchRequest, /// 打洞响应 PunchResponse, - UnKnow(u8), + Unknown(u8), } impl From for Protocol { @@ -27,7 +27,7 @@ impl From for Protocol { 2 => Protocol::Pong, 3 => Protocol::PunchRequest, 4 => Protocol::PunchResponse, - val => Protocol::UnKnow(val), + val => Protocol::Unknown(val), } } } @@ -39,7 +39,7 @@ impl Into for Protocol { Protocol::Pong => 2, Protocol::PunchRequest => 3, Protocol::PunchResponse => 4, - Protocol::UnKnow(val) => val, + Protocol::Unknown(val) => val, } } } @@ -58,18 +58,17 @@ impl> ControlPacket { Protocol::Pong => Ok(ControlPacket::PongPacket(PongPacket::new(buffer)?)), Protocol::PunchRequest => Ok(ControlPacket::PunchRequest), Protocol::PunchResponse => Ok(ControlPacket::PunchResponse), - Protocol::UnKnow(_) => Err(io::Error::new(io::ErrorKind::InvalidData, "Unsupported")), + Protocol::Unknown(_) => Err(io::Error::new(io::ErrorKind::InvalidData, "Unsupported")), } } } /// 网络探针 -#[derive(Copy, Clone)] pub struct PingPacket { buffer: B, } -type PongPacket = PingPacket; +pub type PongPacket = PingPacket; impl> PingPacket { pub fn new(buffer: B) -> io::Result> { diff --git a/switch/src/protocol/ip_turn_packet.rs b/switch/src/protocol/ip_turn_packet.rs new file mode 100644 index 0000000..11aa6c8 --- /dev/null +++ b/switch/src/protocol/ip_turn_packet.rs @@ -0,0 +1,99 @@ +use std::io; +use std::net::Ipv4Addr; + +#[derive(Copy, Clone, Eq, PartialEq, Debug)] +pub enum Protocol { + Ipv4, + Ipv4Broadcast, + Unknown(u8), +} + +impl From for Protocol { + fn from(value: u8) -> Self { + match value { + 4 => Protocol::Ipv4, + 201 => Protocol::Ipv4Broadcast, + val => Protocol::Unknown(val), + } + } +} + +impl Into for Protocol { + fn into(self) -> u8 { + match self { + Protocol::Ipv4 => 4, + Protocol::Ipv4Broadcast => 201, + Protocol::Unknown(val) => val, + } + } +} + +pub struct BroadcastPacketEnd { + buffer: B, +} + +impl> BroadcastPacketEnd { + pub fn unchecked(buffer: B) -> Self { + Self { buffer } + } + pub fn new(buffer: B) -> io::Result { + let len = buffer.as_ref().len(); + let packet = Self::unchecked(buffer); + if len < 1 || packet.len() != len { + Err(io::Error::new( + io::ErrorKind::InvalidData, + "InvalidData", + )) + } else { + Ok(packet) + } + } +} + +impl> BroadcastPacketEnd { + pub fn len(&self) -> usize { + 1 + self.num() as usize * 4 + } + pub fn num(&self) -> u8 { + let len = self.buffer.as_ref().len(); + self.buffer.as_ref()[len - 1] + } + /// 已经发送给了这些地址 + /// 从尾往头拿 + pub fn addresses(&self) -> Vec { + let num = self.num() as usize; + let mut list = Vec::with_capacity(num); + let buf = self.buffer.as_ref(); + let mut offset = buf.len() + 4 - 2; + for _ in 0..num { + offset -= 4; + list.push(Ipv4Addr::new(buf[offset - 3], buf[offset - 2], buf[offset - 1], buf[offset])); + } + list + } +} + +impl + AsMut<[u8]>> BroadcastPacketEnd { + /// 从头往尾放 + pub fn set_address(&mut self, addr: &[Ipv4Addr]) -> io::Result<()> { + let buf = self.buffer.as_mut(); + if buf.len() < 1 + addr.len() * 4 || addr.len() > u8::MAX as usize { + Err(io::Error::new( + io::ErrorKind::InvalidData, + "InvalidData", + )) + } else { + let mut offset = 0; + for ip in addr { + buf[offset..offset + 4].copy_from_slice(&ip.octets()); + offset += 4; + } + self.buffer.as_mut()[offset] = addr.len() as u8; + Ok(()) + } + } +} + + + + diff --git a/switch/src/protocol/mod.rs b/switch/src/protocol/mod.rs index 6fc0fc3..5ee818b 100644 --- a/switch/src/protocol/mod.rs +++ b/switch/src/protocol/mod.rs @@ -18,7 +18,8 @@ use std::{fmt, io}; pub mod control_packet; pub mod error_packet; pub mod service_packet; -pub mod turn_packet; +pub mod ip_turn_packet; +pub mod other_turn_packet; #[derive(Eq, PartialEq, Copy, Clone, Debug)] pub enum Version { @@ -52,8 +53,8 @@ pub enum Protocol { Error, /// 控制协议 Control, - /// 转发ipv4数据 - Ipv4Turn, + /// 转发ip数据 + IpTurn, /// 转发其他数据 OtherTurn, UnKnow(u8), @@ -65,7 +66,7 @@ impl From for Protocol { 1 => Protocol::Service, 2 => Protocol::Error, 3 => Protocol::Control, - 4 => Protocol::Ipv4Turn, + 4 => Protocol::IpTurn, 5 => Protocol::OtherTurn, val => Protocol::UnKnow(val), } @@ -78,7 +79,7 @@ impl Into for Protocol { Protocol::Service => 1, Protocol::Error => 2, Protocol::Control => 3, - Protocol::Ipv4Turn => 4, + Protocol::IpTurn => 4, Protocol::OtherTurn => 5, Protocol::UnKnow(val) => val, } @@ -143,6 +144,9 @@ impl> NetPacket { } impl + AsMut<[u8]>> NetPacket { + pub fn buffer_mut(&mut self)->&mut [u8]{ + self.buffer.as_mut() + } pub fn set_version(&mut self, version: Version) { self.buffer.as_mut()[0] = version.into(); } diff --git a/switch/src/protocol/turn_packet.rs b/switch/src/protocol/other_turn_packet.rs similarity index 76% rename from switch/src/protocol/turn_packet.rs rename to switch/src/protocol/other_turn_packet.rs index acc961b..d224104 100644 --- a/switch/src/protocol/turn_packet.rs +++ b/switch/src/protocol/other_turn_packet.rs @@ -1,14 +1,14 @@ #[derive(Copy, Clone, Eq, PartialEq, Debug)] pub enum Protocol { Punch, - UnKnow(u8), + Unknown(u8), } impl From for Protocol { fn from(value: u8) -> Self { match value { 1 => Protocol::Punch, - val => Protocol::UnKnow(val), + val => Protocol::Unknown(val), } } } @@ -17,7 +17,7 @@ impl Into for Protocol { fn into(self) -> u8 { match self { Protocol::Punch => 1, - Protocol::UnKnow(val) => val, + Protocol::Unknown(val) => val, } } } diff --git a/switch/src/protocol/service_packet.rs b/switch/src/protocol/service_packet.rs index bc3ea1e..d17a149 100644 --- a/switch/src/protocol/service_packet.rs +++ b/switch/src/protocol/service_packet.rs @@ -8,7 +8,7 @@ pub enum Protocol { PollDeviceList, /// 推送设备列表 PushDeviceList, - UnKnow(u8), + Unknown(u8), } impl From for Protocol { @@ -18,7 +18,7 @@ impl From for Protocol { 2 => Self::RegistrationResponse, 3 => Self::PollDeviceList, 4 => Self::PushDeviceList, - val => Self::UnKnow(val), + val => Self::Unknown(val), } } } @@ -30,7 +30,7 @@ impl Into for Protocol { Self::RegistrationResponse => 2, Self::PollDeviceList => 3, Self::PushDeviceList => 4, - Self::UnKnow(val) => val, + Self::Unknown(val) => val, } } }