From 55a510aa34aa7cb327c4a9271d34d80997baabfe Mon Sep 17 00:00:00 2001 From: lubeilin <1791778603@qq.com> Date: Thu, 29 Feb 2024 22:31:08 +0800 Subject: [PATCH] =?UTF-8?q?[mio]=20=E5=8E=BB=E9=99=A4=E5=A4=9A=E4=BD=99?= =?UTF-8?q?=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- vnt/src/handle/handshake_handler.rs | 262 --------- vnt/src/handle/heartbeat_handler.rs | 301 ----------- vnt/src/handle/punch_handler.rs | 180 ------- vnt/src/handle/recv_handler.rs | 714 ------------------------- vnt/src/handle/registration_handler.rs | 262 --------- vnt/src/handle/tun_tap/tap_handler.rs | 260 --------- 6 files changed, 1979 deletions(-) delete mode 100644 vnt/src/handle/handshake_handler.rs delete mode 100644 vnt/src/handle/heartbeat_handler.rs delete mode 100644 vnt/src/handle/punch_handler.rs delete mode 100644 vnt/src/handle/recv_handler.rs delete mode 100644 vnt/src/handle/registration_handler.rs delete mode 100644 vnt/src/handle/tun_tap/tap_handler.rs diff --git a/vnt/src/handle/handshake_handler.rs b/vnt/src/handle/handshake_handler.rs deleted file mode 100644 index 8b68eb9..0000000 --- a/vnt/src/handle/handshake_handler.rs +++ /dev/null @@ -1,262 +0,0 @@ -use std::net::{SocketAddr, SocketAddrV6}; - -use crate::channel::channel::Context; -use crate::channel::RouteKey; -use crate::cipher::{Cipher, RsaCipher}; -use crate::proto::message::{HandshakeRequest, HandshakeResponse, SecretHandshakeRequest}; -use crate::protocol::body::RSA_ENCRYPTION_RESERVED; -use crate::protocol::{service_packet, NetPacket, Protocol, Version, MAX_TTL}; -use protobuf::Message; -use std::io::{Read, Write}; -use std::net::TcpStream; -use std::net::UdpSocket; - -pub enum HandshakeEnum { - NotSecret, - KeyError, - Timeout, - ServerError(String), - Other(String), -} - -fn handshake_request_packet(secret: bool) -> crate::Result>> { - let mut request = HandshakeRequest::new(); - request.secret = secret; - request.version = crate::VNT_VERSION.to_string(); - let bytes = request.write_to_bytes()?; - let buf = vec![0u8; 12 + bytes.len()]; - let mut net_packet = NetPacket::new(buf)?; - net_packet.set_version(Version::V1); - net_packet.set_gateway_flag(true); - net_packet.set_protocol(Protocol::Service); - net_packet.set_transport_protocol(service_packet::Protocol::HandshakeRequest.into()); - net_packet.first_set_ttl(MAX_TTL); - net_packet.set_payload(&bytes)?; - Ok(net_packet) -} - -fn secret_handshake_request_packet( - rsa_cipher: &RsaCipher, - token: String, - key: &[u8], -) -> crate::Result>> { - let mut request = SecretHandshakeRequest::new(); - request.token = token; - request.key = key.to_vec(); - let bytes = request.write_to_bytes()?; - let mut net_packet = NetPacket::new0( - 12 + bytes.len(), - vec![0u8; 12 + bytes.len() + RSA_ENCRYPTION_RESERVED], - )?; - net_packet.set_version(Version::V1); - net_packet.set_gateway_flag(true); - net_packet.set_protocol(Protocol::Service); - net_packet.set_transport_protocol(service_packet::Protocol::SecretHandshakeRequest.into()); - net_packet.first_set_ttl(MAX_TTL); - net_packet.set_payload(&bytes)?; - Ok(rsa_cipher.encrypt(&mut net_packet)?) -} - -/// 第一次握手，拿到公钥 -pub fn handshake( - main_channel: &UdpSocket, - main_tcp_channel: Option<&mut TcpStream>, - server_address: SocketAddr, - secret: bool, -) -> Result, HandshakeEnum> { - let request_packet = handshake_request_packet(secret).unwrap(); - let send_buf = request_packet.buffer(); - let mut recv_buf = [0u8; 10240]; - let len = send_recv( - main_channel, - main_tcp_channel, - server_address, - send_buf, - &mut recv_buf, - )?; - let net_packet = match NetPacket::new(&recv_buf[..len]) { - Ok(net_packet) => net_packet, - Err(e) => { - return Err(HandshakeEnum::Other(format!("net_packet {}", e))); - } - }; - match net_packet.protocol() { - Protocol::Service => { - match service_packet::Protocol::from(net_packet.transport_protocol()) { - service_packet::Protocol::HandshakeResponse => { - match HandshakeResponse::parse_from_bytes(net_packet.payload()) { - Ok(response) => { - if !response.secret && secret { - //客户端要加密，服务端不支持加密 - return Err(HandshakeEnum::NotSecret); - } - if secret { - //转换公钥 - match RsaCipher::new(&response.public_key) { - Ok(rsa) => { - match rsa.finger() { - Ok(finger) => { - if finger != response.key_finger { - return Err(HandshakeEnum::Other( - "finger error".to_string(), - )); - } - } - Err(e) => { - return Err(HandshakeEnum::Other(format!( - "finger {}", - e - ))); - } - } - Ok(Some(rsa)) - } - Err(e) => { - return Err(HandshakeEnum::Other(format!( - "RsaCipher {}", - e - ))); - } - } - } else { - Ok(None) - } - } - Err(e) => { - return Err(HandshakeEnum::Other(format!("parse_from_bytes {}", e))); - } - } - } - _ => { - return Err(HandshakeEnum::Other("not match".to_string())); - } - } - } - _ => { - return Err(HandshakeEnum::Other("not match".to_string())); - } - } -} - -fn send_recv( - main_channel: &UdpSocket, - main_tcp_channel: Option<&mut TcpStream>, - server_address: SocketAddr, - send_buf: &[u8], - recv_buf: &mut [u8], -) -> Result { - if let Some(main_tcp_channel) = main_tcp_channel { - let mut head = [0; 4]; - let len = send_buf.len(); - head[2] = (len >> 8) as u8; - head[3] = (len & 0xFF) as u8; - if let Err(e) = main_tcp_channel.write_all(&head) { - return Err(HandshakeEnum::Other(format!("send error:{}", e))); - } - if let Err(e) = main_tcp_channel.write_all(send_buf) { - return Err(HandshakeEnum::Other(format!("send error:{}", e))); - } - if let Err(e) = main_tcp_channel.read_exact(&mut head) { - return Err(HandshakeEnum::Other(format!("read error:{}", e))); - } - let len = (((head[2] as u16) << 8) | head[3] as u16) as usize; - if len > recv_buf.len() { - return Err(HandshakeEnum::Other("too long".to_string())); - } - if let Err(e) = main_tcp_channel.read_exact(&mut recv_buf[..len]) { - return Err(HandshakeEnum::Other(format!("read error:{}", e))); - } - Ok(len) - } else { - let server_address = if let SocketAddr::V4(ipv4) = server_address { - SocketAddr::V6(SocketAddrV6::new( - ipv4.ip().to_ipv6_mapped(), - ipv4.port(), - 0, - 0, - )) - } else { - server_address - }; - if let Err(e) = main_channel.send_to(send_buf, server_address) { - return Err(HandshakeEnum::Other(format!("send error:{}", e))); - } - match main_channel.recv_from(recv_buf) { - Ok((len, addr)) => { - if server_address != addr { - log::warn!("请求{:?}和响应{:?}地址不一致", server_address, addr); - } - Ok(len) - } - Err(e) => Err(HandshakeEnum::Other(format!("receiver error:{}", e))), - } - } -} - -/// 第二次握手，同步对称密钥，后续将使用对称加密 -pub fn secret_handshake( - main_channel: &UdpSocket, - main_tcp_channel: Option<&mut TcpStream>, - server_address: SocketAddr, - rsa_cipher: &RsaCipher, - server_cipher: &Cipher, - token: String, -) -> Result<(), HandshakeEnum> { - let secret_packet = - match secret_handshake_request_packet(rsa_cipher, token, server_cipher.key().unwrap()) { - Ok(secret_packet) => secret_packet, - Err(e) => { - return Err(HandshakeEnum::Other(format!( - "secret_handshake_request_packet {}", - e - ))); - } - }; - let send_buf = secret_packet.buffer(); - let mut recv_buf = [0u8; 10240]; - let len = send_recv( - main_channel, - main_tcp_channel, - server_address, - send_buf, - &mut recv_buf, - )?; - let mut net_packet = match NetPacket::new(&mut recv_buf[..len]) { - Ok(net_packet) => net_packet, - Err(e) => { - return Err(HandshakeEnum::Other(format!("secret_net_packet {}", e))); - } - }; - match server_cipher.decrypt_ipv4(&mut net_packet) { - Ok(_) => { - if net_packet.is_gateway() - && net_packet.protocol() == Protocol::Service - && service_packet::Protocol::from(net_packet.transport_protocol()) - == service_packet::Protocol::SecretHandshakeResponse - { - Ok(()) - } else { - Err(HandshakeEnum::Other("not match".to_string())) - } - } - Err(e) => Err(HandshakeEnum::Other(format!("decrypt_ipv4 {}", e))), - } -} - -pub fn secret_handshake_req( - context: &Context, - server_address: SocketAddr, - rsa_cipher: &RsaCipher, - server_cipher: &Cipher, - token: String, - route_key: &RouteKey, -) -> crate::Result<()> { - let secret_packet = - secret_handshake_request_packet(rsa_cipher, token, server_cipher.key().unwrap())?; - if route_key.is_tcp() { - context.send_main(secret_packet.buffer(), server_address)?; - } else { - context.send_main_udp(secret_packet.buffer(), server_address)?; - } - Ok(()) -} diff --git a/vnt/src/handle/heartbeat_handler.rs b/vnt/src/handle/heartbeat_handler.rs deleted file mode 100644 index 7d259a0..0000000 --- a/vnt/src/handle/heartbeat_handler.rs +++ /dev/null @@ -1,301 +0,0 @@ -use std::io; -use std::net::{Ipv4Addr, ToSocketAddrs}; -use std::sync::Arc; -use std::time::Duration; - -use crossbeam_utils::atomic::AtomicCell; -use parking_lot::Mutex; -use rand::prelude::SliceRandom; - -use crate::channel::idle::Idle; -use crate::channel::sender::ChannelSender; -use crate::channel::Route; -use crate::cipher::Cipher; -use crate::core::status::VntWorker; -use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo}; -use crate::protocol::body::ENCRYPTION_RESERVED; -use crate::protocol::control_packet::PingPacket; -use crate::protocol::{control_packet, NetPacket, Protocol, Version, MAX_TTL}; - -pub fn start_idle(mut worker: VntWorker, idle: Idle, sender: ChannelSender) { - tokio::spawn(async move { - tokio::select! { - _=worker.stop_wait()=>{ - return; - } - rs=start_idle_(idle, sender)=>{ - if let Err(e) = rs { - log::warn!("空闲检测任务停止:{:?}", e); - } - } - } - worker.stop_all(); - }); -} - -async fn start_idle_(idle: Idle, sender: ChannelSender) -> io::Result<()> { - log::info!("启动空闲检查任务"); - loop { - let (peer_ip, route) = idle.next_idle().await?; - log::info!("路由空闲 peer_ip:{:?},route:{:?}", peer_ip, route); - sender.remove_route(&peer_ip, route); - } -} - -pub fn start_heartbeat( - mut worker: VntWorker, - sender: ChannelSender, - device_list: Arc)>>, - current_device: Arc>, - client_cipher: Cipher, - server_cipher: Cipher, -) { - tokio::spawn(async move { - tokio::select! { - _=worker.stop_wait()=>{ - return; - } - rs=start_heartbeat_(sender, device_list, current_device,client_cipher,server_cipher)=>{ - if let Err(e) = rs { - log::warn!("心跳任务停止:{:?}", e); - } - } - } - worker.stop_all(); - }); -} - -pub fn start_heartbeat_main( - mut worker: VntWorker, - sender: ChannelSender, - device_list: Arc)>>, - current_device: Arc>, - server_address_str: String, - client_cipher: Cipher, - server_cipher: Cipher, -) { - tokio::spawn(async move { - tokio::select! { - _=worker.stop_wait()=>{ - return; - } - rs=start_heartbeat_main_(sender, device_list, current_device,server_address_str,client_cipher,server_cipher)=>{ - if let Err(e) = rs { - log::warn!("主心跳任务停止:{:?}", e); - } - } - } - worker.stop_all(); - }); -} - -fn heartbeat_packet( - ttl: u8, - device_list: &Mutex<(u16, Vec)>, - client_cipher: &Cipher, - server_cipher: &Cipher, - gateway: bool, - src: Ipv4Addr, - dest: Ipv4Addr, -) -> NetPacket<[u8; 12 + 4 + ENCRYPTION_RESERVED]> { - let mut net_packet = NetPacket::new_encrypt([0u8; 12 + 4 + ENCRYPTION_RESERVED]).unwrap(); - 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(ttl); - net_packet.set_source(src); - net_packet.set_destination(dest); - { - let mut ping = PingPacket::new(net_packet.payload_mut()).unwrap(); - let epoch = { device_list.lock().0 }; - ping.set_epoch(epoch); - ping.set_time(crate::handle::now_time() as u16); - } - if gateway { - net_packet.set_gateway_flag(true); - server_cipher.encrypt_ipv4(&mut net_packet).unwrap(); - } else { - client_cipher.encrypt_ipv4(&mut net_packet).unwrap(); - } - net_packet -} - -async fn start_heartbeat_main_( - sender: ChannelSender, - device_list: Arc)>>, - current_device: Arc>, - server_address_str: String, - client_cipher: Cipher, - server_cipher: Cipher, -) -> io::Result<()> { - let mut count = 0; - log::info!("启动主心跳任务"); - loop { - if sender.is_close() { - return Ok(()); - } - let mut current_dev = current_device.load(); - let src = current_dev.virtual_ip(); - if count % 40 == 19 { - if let Ok(mut addr) = server_address_str.to_socket_addrs() { - if let Some(addr) = addr.next() { - if addr != current_dev.connect_server { - let mut tmp = current_dev.clone(); - tmp.connect_server = addr; - log::info!( - "服务端地址变化,旧地址:{}，新地址:{}", - current_dev.connect_server, - addr - ); - if current_device.compare_exchange(current_dev, tmp).is_ok() { - current_dev.connect_server = addr; - } - } - } - } - } - let server_packet = heartbeat_packet( - MAX_TTL, - &device_list, - &client_cipher, - &server_cipher, - true, - src, - current_dev.virtual_gateway, - ); - if let Err(e) = sender.send_main(server_packet.buffer(), current_dev.connect_server) { - log::warn!("connect_server:{:?},e:{:?}", current_dev.connect_server, e); - } - count += 1; - tokio::time::sleep(Duration::from_millis(3000)).await; - } -} - -async fn start_heartbeat_( - sender: ChannelSender, - device_list: Arc)>>, - current_device: Arc>, - client_cipher: Cipher, - server_cipher: Cipher, -) -> io::Result<()> { - let mut count = 0; - log::info!("启动心跳任务"); - loop { - if sender.is_close() { - return Ok(()); - } - let current_dev = current_device.load(); - //如果和服务端使用tcp连接，则维持udp洞的频率要更高些 - if (sender.is_main_tcp() && count % 4 == 0) || (!sender.is_main_tcp() && count % 40 == 1) { - let mut packet = NetPacket::new_encrypt([0; 12 + ENCRYPTION_RESERVED])?; - packet.set_version(Version::V1); - packet.set_gateway_flag(true); - packet.set_protocol(Protocol::Control); - packet.set_transport_protocol(control_packet::Protocol::AddrRequest.into()); - packet.first_set_ttl(MAX_TTL); - packet.set_source(current_dev.virtual_ip()); - packet.set_destination(current_dev.virtual_gateway); - server_cipher.encrypt_ipv4(&mut packet)?; - let _ = sender.send_main_udp(packet.buffer(), current_dev.connect_server); - } - - let src = current_dev.virtual_ip(); - - if count % 10 == 7 { - let mut route_list: Option)>> = None; - let peer_list = { device_list.lock().1.clone() }; - for peer in peer_list { - if peer.virtual_ip == current_dev.virtual_ip { - continue; - } - let client_packet = heartbeat_packet( - MAX_TTL, - &device_list, - &client_cipher, - &server_cipher, - false, - src, - peer.virtual_ip, - ); - if let Some(route) = sender.route_one(&peer.virtual_ip) { - if let Err(e) = - sender.try_send_by_key(client_packet.buffer(), &route.route_key()) - { - log::warn!("virtual_ip:{},route:{:?},e:{:?}", peer.virtual_ip, route, e); - } - if route.is_p2p() && !sender.is_first_latency() { - continue; - } - } else { - //没有直连路由则发送到网关 - if let Err(e) = - sender.send_main(client_packet.buffer(), current_dev.connect_server) - { - log::warn!( - "virtual_ip:{},connect_server:{:?},e:{:?}", - peer.virtual_ip, - current_dev.connect_server, - e - ); - } - } - - //再随机发送到其他地址，看有没有客户端符合转发条件 - let route_list = route_list.get_or_insert_with(|| { - let mut l = sender.route_table(); - l.shuffle(&mut rand::thread_rng()); - l - }); - let mut num = 0; - 'a: for (peer_ip, route_list) in route_list.iter() { - for route in route_list { - if peer_ip != &peer.virtual_ip && route.is_p2p() { - if let Err(e) = - sender.try_send_by_key(client_packet.buffer(), &route.route_key()) - { - log::warn!( - "virtual_ip:{},route:{:?},e:{:?}", - peer.virtual_ip, - route, - e - ); - } - num += 1; - break; - } - if num >= 2 { - break 'a; - } - } - } - tokio::time::sleep(Duration::from_millis(1)).await; - } - } else { - for (peer_ip, route_list) in sender.route_table().iter() { - if peer_ip == ¤t_dev.virtual_gateway { - continue; - } - let client_packet = heartbeat_packet( - MAX_TTL, - &device_list, - &client_cipher, - &server_cipher, - false, - src, - *peer_ip, - ); - for route in route_list { - if let Err(e) = - sender.try_send_by_key(client_packet.buffer(), &route.route_key()) - { - log::warn!("peer_ip:{:?},route:{:?},e:{:?}", peer_ip, route, e); - } - tokio::time::sleep(Duration::from_millis(2)).await; - } - } - } - - count += 1; - tokio::time::sleep(Duration::from_millis(3000)).await; - } -} diff --git a/vnt/src/handle/punch_handler.rs b/vnt/src/handle/punch_handler.rs deleted file mode 100644 index 269cd4b..0000000 --- a/vnt/src/handle/punch_handler.rs +++ /dev/null @@ -1,180 +0,0 @@ -use crate::channel::punch::{NatInfo, Punch}; -use crate::channel::sender::ChannelSender; -use crate::cipher::Cipher; -use crate::core::status::VntWorker; -use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo}; -use crate::nat::NatTest; -use crate::proto::message::{PunchInfo, PunchNatType}; -use crate::protocol::body::ENCRYPTION_RESERVED; -use crate::protocol::{control_packet, other_turn_packet, NetPacket, Protocol, Version, MAX_TTL}; -use crossbeam_utils::atomic::AtomicCell; -use parking_lot::Mutex; -use protobuf::Message; -use rand::prelude::SliceRandom; -use std::io; -use std::net::Ipv4Addr; -use std::sync::Arc; -use std::time::Duration; -use tokio::sync::mpsc::Receiver; - -pub fn start( - mut worker: VntWorker, - receiver: Receiver<(Ipv4Addr, NatInfo)>, - punch: Punch, - current_device: Arc>, - client_cipher: Cipher, -) { - tokio::spawn(async move { - tokio::select! { - _=start0(receiver, punch, current_device,client_cipher)=>{} - _=worker.stop_wait()=>{ - return; - } - } - worker.stop_all(); - }); -} - -pub async fn start0( - mut receiver: Receiver<(Ipv4Addr, NatInfo)>, - mut punch: Punch, - current_device: Arc>, - client_cipher: Cipher, -) { - log::info!("启动打洞任务"); - while let Some((peer_ip, nat_info)) = receiver.recv().await { - if let Err(e) = start_( - &client_cipher, - &mut punch, - ¤t_device, - peer_ip, - nat_info, - ) - .await - { - log::warn!("网络打洞异常 {:?}", e); - } - } -} - -async fn start_( - client_cipher: &Cipher, - punch: &mut Punch, - current_device: &Arc>, - peer_ip: Ipv4Addr, - nat_info: NatInfo, -) -> io::Result<()> { - let mut packet = NetPacket::new_encrypt([0u8; 12 + ENCRYPTION_RESERVED])?; - packet.set_version(Version::V1); - packet.first_set_ttl(1); - packet.set_protocol(Protocol::Control); - packet.set_transport_protocol(control_packet::Protocol::PunchRequest.into()); - packet.set_source(current_device.load().virtual_ip()); - packet.set_destination(peer_ip); - log::info!("发起打洞，目标:{:?},{:?}", peer_ip, nat_info); - client_cipher.encrypt_ipv4(&mut packet)?; - punch.punch(packet.buffer(), peer_ip, nat_info).await -} - -pub async fn start_punch( - mut worker: VntWorker, - nat_test: NatTest, - device_list: Arc)>>, - sender: ChannelSender, - current_device: Arc>, - client_cipher: Cipher, -) { - let mut num = 0; - let sleep_time = [3, 5, 7, 11, 13, 17, 19, 23, 29]; - log::info!("启动发起打洞请求任务"); - loop { - if sender.is_close() { - break; - } - tokio::select! { - rs= start_punch_(Duration::from_secs(sleep_time[num % sleep_time.len()]),&nat_test, &device_list, - &sender, ¤t_device,&client_cipher)=>{ - if let Err(e) = rs { - log::warn!("打洞处理任务异常 {:?}", e); - } - } - _=worker.stop_wait()=>{ - break; - } - } - num += 1; - } -} - -async fn start_punch_( - sleep_time: Duration, - nat_test: &NatTest, - device_list: &Arc)>>, - sender: &ChannelSender, - current_device: &Arc>, - client_cipher: &Cipher, -) -> crate::Result<()> { - let current_device = current_device.load(); - let nat_info = nat_test.nat_info(); - let mut list = device_list.lock().clone().1; - list.shuffle(&mut rand::thread_rng()); - let mut count = 0; - for info in list { - if info.virtual_ip <= current_device.virtual_ip { - continue; - } - if !sender.need_punch(&info.virtual_ip) { - continue; - } - count += 1; - if count > 2 { - break; - } - let packet = punch_packet( - client_cipher, - current_device.virtual_ip(), - &nat_info, - info.virtual_ip, - ) - .unwrap(); - let _ = sender.send_main(packet.buffer(), current_device.connect_server); - } - tokio::time::sleep(sleep_time).await; - Ok(()) -} - -pub fn punch_packet( - client_cipher: &Cipher, - virtual_ip: Ipv4Addr, - nat_info: &NatInfo, - dest: Ipv4Addr, -) -> crate::Result>> { - let mut punch_reply = PunchInfo::new(); - punch_reply.reply = false; - 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.local_ip = u32::from(nat_info.local_ipv4().unwrap_or(Ipv4Addr::UNSPECIFIED)); - punch_reply.local_port = nat_info.udp_port as u32; - punch_reply.tcp_port = nat_info.tcp_port as u32; - if let Some(ipv6) = nat_info.ipv6 { - punch_reply.ipv6_port = nat_info.udp_port as u32; - punch_reply.ipv6 = ipv6.octets().to_vec(); - } - 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_encrypt(vec![0u8; 12 + bytes.len() + ENCRYPTION_RESERVED])?; - net_packet.set_version(Version::V1); - net_packet.set_protocol(Protocol::OtherTurn); - 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); - net_packet.set_payload(&bytes)?; - client_cipher.encrypt_ipv4(&mut net_packet)?; - Ok(net_packet) -} diff --git a/vnt/src/handle/recv_handler.rs b/vnt/src/handle/recv_handler.rs deleted file mode 100644 index b955643..0000000 --- a/vnt/src/handle/recv_handler.rs +++ /dev/null @@ -1,714 +0,0 @@ -use std::collections::HashMap; -use std::net::{Ipv4Addr, Ipv6Addr}; -use std::sync::Arc; -use std::time::{Duration, Instant}; - -use crossbeam_utils::atomic::AtomicCell; -use parking_lot::{Mutex, RwLock}; -use protobuf::Message; -use tokio::sync::mpsc::Sender; - -use packet::icmp::{icmp, Kind}; -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::cipher::{Cipher, RsaCipher}; -use crate::error::Error; -use crate::external_route::AllowExternalRoute; -use crate::handle::handshake_handler::secret_handshake_req; -use crate::handle::registration_handler::Register; -use crate::handle::{ConnectStatus, CurrentDeviceInfo, PeerDeviceInfo, PeerDeviceStatus}; -use crate::igmp_server::IgmpServer; -#[cfg(feature = "ip_proxy")] -use crate::ip_proxy::{IpProxyMap, ProxyHandler}; -use crate::nat; -use crate::nat::NatTest; -use crate::proto::message::{DeviceList, PunchInfo, PunchNatType, RegistrationResponse}; -use crate::protocol::body::ENCRYPTION_RESERVED; -use crate::protocol::control_packet::ControlPacket; -use crate::protocol::error_packet::InErrorPacket; -use crate::protocol::{ - control_packet, ip_turn_packet, other_turn_packet, service_packet, NetPacket, Protocol, - Version, MAX_TTL, -}; -use crate::tun_tap_device::DeviceWriter; - -#[derive(Clone)] -pub struct ChannelDataHandler { - current_device: Arc>, - device_list: Arc)>>, - register: Arc, - nat_test: NatTest, - igmp_server: Option, - device_writer: DeviceWriter, - connect_status: Arc>, - peer_nat_info_map: Arc>>, - #[cfg(feature = "ip_proxy")] - ip_proxy_map: Option, - out_external_route: AllowExternalRoute, - cone_sender: Sender<(Ipv4Addr, NatInfo)>, - symmetric_sender: Sender<(Ipv4Addr, NatInfo)>, - client_cipher: Cipher, - server_cipher: Cipher, - rsa_cipher: Option, - relay: bool, - token: String, - time: Arc>, - pub head_reserve: usize, -} - -impl ChannelDataHandler { - pub fn new( - current_device: Arc>, - device_list: Arc)>>, - register: Arc, - nat_test: NatTest, - igmp_server: Option, - device_writer: DeviceWriter, - connect_status: Arc>, - peer_nat_info_map: Arc>>, - #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, - out_external_route: AllowExternalRoute, - cone_sender: Sender<(Ipv4Addr, NatInfo)>, - symmetric_sender: Sender<(Ipv4Addr, NatInfo)>, - client_cipher: Cipher, - server_cipher: Cipher, - rsa_cipher: Option, - relay: bool, - token: String, - head_reserve: usize, - ) -> Self { - Self { - current_device, - device_list, - register, - nat_test, - igmp_server, - device_writer, - connect_status, - peer_nat_info_map, - #[cfg(feature = "ip_proxy")] - ip_proxy_map, - out_external_route, - cone_sender, - symmetric_sender, - client_cipher, - server_cipher, - rsa_cipher, - relay, - token, - time: Arc::new(AtomicCell::new(Instant::now())), - head_reserve, - } - } -} - -impl ChannelDataHandler { - pub fn handle( - &self, - buf: &mut [u8], - start: usize, - end: usize, - route_key: RouteKey, - context: &Context, - ) { - assert_eq!(start, 14); - match self.handle0(&mut buf[..end], &route_key, context) { - Ok(_) => {} - Err(e) => { - log::warn!("{:?}", e); - } - } - } - 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 || net_packet.source_ttl() < net_packet.ttl() { - return Ok(()); - } - let source = net_packet.source(); - context.update_read_time(&source, route_key); - let current_device = self.current_device.load(); - let destination = net_packet.destination(); - let not_broadcast = !destination.is_broadcast() - && !destination.is_multicast() - && destination != current_device.broadcast_address; - if current_device.virtual_ip() != destination - && !net_packet.is_gateway() - && not_broadcast - && !destination.is_unspecified() - { - //校验指纹，不需要解密 - self.client_cipher.check_finger(&net_packet)?; - net_packet.set_ttl(net_packet.ttl() - 1); - let ttl = net_packet.ttl(); - if ttl > 0 { - // 转发 - if let Some(route) = context.route_one(&destination) { - if route.metric <= net_packet.ttl() { - context.try_send_by_key(net_packet.buffer(), &route.route_key())?; - } - } else if (ttl > 1 || destination == current_device.virtual_gateway()) - && source != current_device.virtual_gateway() - { - //网关默认要转发一次，生存时间不够的发到网关也会被丢弃 - context.send_main(net_packet.buffer(), current_device.connect_server)?; - } - } - return Ok(()); - } - if net_packet.is_gateway() { - if net_packet.protocol() == Protocol::Error - && net_packet.transport_protocol() - == crate::protocol::error_packet::Protocol::NoKey.into() - { - if let Some(rsa_cipher) = &self.rsa_cipher { - let last = self.time.load(); - if last.elapsed() < Duration::from_secs(3) - || self.time.compare_exchange(last, Instant::now()).is_err() - { - //短时间不重复上传服务端密钥 - return Ok(()); - } - log::warn!("上传服务端密钥"); - secret_handshake_req( - context, - current_device.connect_server, - rsa_cipher, - &self.server_cipher, - self.token.clone(), - route_key, - )?; - } - } else { - //服务端解密 - self.server_cipher.decrypt_ipv4(&mut net_packet)?; - let data_len = net_packet.data_len(); - self.server_packet_handle(context, current_device, buf, data_len, route_key)?; - } - return Ok(()); - } - self.client_cipher.decrypt_ipv4(&mut net_packet)?; - match net_packet.protocol() { - 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::Igmp => { - if let Some(igmp_server) = &self.igmp_server { - igmp_server.handle(ipv4.payload(), source)?; - } - return Ok(()); - } - ipv4::protocol::Protocol::Icmp => { - 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); - //不管加不加密，和接收到的数据长度都一致 - self.client_cipher.encrypt_ipv4(&mut net_packet)?; - context.try_send_by_key(net_packet.buffer(), route_key)?; - return Ok(()); - } - } - } - _ => {} - } - if not_broadcast && ipv4.destination_ip() != destination { - if self.out_external_route.allow(&ipv4.destination_ip()) { - #[cfg(feature = "ip_proxy")] - if let Some(ip_proxy_map) = &self.ip_proxy_map { - if ip_proxy_map.recv_handle(&mut ipv4, source, destination)? { - return Ok(()); - } - } - } else { - log::warn!( - "没有ip代理规则{:?}:{}->{}->{}", - ipv4.protocol(), - source, - destination, - ipv4.destination_ip() - ); - return Err(Error::Warn("没有ip代理规则".to_string())); - } - } - - //传输协议12字节 - self.device_writer.write_ipv4(&mut buf[12..])?; - return Ok(()); - } - ip_turn_packet::Protocol::Ipv4Broadcast => { - //客户端不帮忙转发广播包，所以不会出现这种类型的数据 - } - ip_turn_packet::Protocol::Unknown(_) => {} - } - } - Protocol::Service => {} - Protocol::Error => {} - Protocol::Control => { - self.control(context, current_device, source, net_packet, route_key)?; - } - Protocol::OtherTurn => { - self.other_turn(context, current_device, source, net_packet, route_key)?; - } - Protocol::UnKnow(e) => { - log::info!("不支持的协议:{}", e); - } - } - Ok(()) - } - - fn pong_packet( - &self, - gateway: bool, - metric: u8, - context: &Context, - current_device: CurrentDeviceInfo, - source: Ipv4Addr, - pong_packet: control_packet::PongPacket<&[u8]>, - route_key: &RouteKey, - ) -> crate::Result<()> { - let current_time = crate::handle::now_time() as u16; - if current_time < pong_packet.time() { - return Ok(()); - } - let rt = (current_time - pong_packet.time()) as i64; - let route = Route::from(*route_key, metric, rt); - context.add_route(source, route); - if gateway { - let epoch = self.device_list.lock().0; - if pong_packet.epoch() != epoch { - let mut poll_device = NetPacket::new_encrypt([0; 12 + ENCRYPTION_RESERVED])?; - poll_device.set_source(current_device.virtual_ip()); - poll_device.set_destination(source); - poll_device.set_version(Version::V1); - poll_device.set_gateway_flag(true); - poll_device.first_set_ttl(MAX_TTL); - poll_device.set_protocol(Protocol::Service); - poll_device.set_transport_protocol(service_packet::Protocol::PollDeviceList.into()); - self.server_cipher.encrypt_ipv4(&mut poll_device)?; - context.send_main(poll_device.buffer(), current_device.connect_server)?; - } - } - Ok(()) - } - 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(_) => { - 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.client_cipher.encrypt_ipv4(&mut net_packet)?; - context.try_send_by_key(net_packet.buffer(), route_key)?; - let route = Route::from(*route_key, metric, 199); - context.add_route_if_absent(source, route); - } - ControlPacket::PongPacket(pong_packet) => { - self.pong_packet( - false, - metric, - context, - current_device, - source, - pong_packet, - route_key, - )?; - } - ControlPacket::PunchRequest => { - if self.relay { - return Ok(()); - } - //回应 - net_packet.set_transport_protocol(control_packet::Protocol::PunchResponse.into()); - net_packet.set_source(current_device.virtual_ip()); - net_packet.set_destination(source); - net_packet.first_set_ttl(1); - self.client_cipher.encrypt_ipv4(&mut net_packet)?; - context.try_send_by_key(net_packet.buffer(), route_key)?; - let route = Route::from(*route_key, 1, 199); - context.add_route_if_absent(source, route); - } - ControlPacket::PunchResponse => { - if self.relay { - return Ok(()); - } - let route = Route::from(*route_key, 1, 199); - context.add_route_if_absent(source, route); - } - ControlPacket::AddrRequest => match route_key.addr.ip() { - std::net::IpAddr::V4(ipv4) => { - let mut packet = NetPacket::new_encrypt([0; 12 + 6 + ENCRYPTION_RESERVED])?; - packet.set_version(Version::V1); - packet.set_protocol(Protocol::Control); - packet.set_transport_protocol(control_packet::Protocol::AddrResponse.into()); - packet.first_set_ttl(MAX_TTL); - packet.set_source(current_device.virtual_ip()); - packet.set_destination(source); - let mut addr_packet = control_packet::AddrPacket::new(packet.payload_mut())?; - addr_packet.set_ipv4(ipv4); - addr_packet.set_port(route_key.addr.port()); - self.client_cipher.encrypt_ipv4(&mut packet)?; - context.try_send_by_key(packet.buffer(), route_key)?; - } - std::net::IpAddr::V6(_) => {} - }, - ControlPacket::AddrResponse(addr_packet) => self - .nat_test - .update_addr(addr_packet.ipv4(), addr_packet.port()), - } - Ok(()) - } - fn other_turn( - &self, - context: &Context, - current_device: CurrentDeviceInfo, - source: Ipv4Addr, - net_packet: NetPacket<&mut [u8]>, - route_key: &RouteKey, - ) -> crate::Result<()> { - if self.relay { - return Ok(()); - } - 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| Ipv4Addr::from(v.to_be_bytes())) - .collect(); - let local_ipv4 = Some(Ipv4Addr::from(punch_info.local_ip.to_be_bytes())); - let udp_port = punch_info.local_port as u16; - let tcp_port = punch_info.tcp_port as u16; - let ipv6 = if punch_info.ipv6.len() == 16 { - let ipv6: [u8; 16] = punch_info.ipv6.try_into().unwrap(); - Some(Ipv6Addr::from(ipv6)) - } else { - None - }; - - let peer_nat_info = NatInfo::new( - public_ips, - punch_info.public_port as u16, - punch_info.public_port_range as u16, - local_ipv4, - ipv6, - udp_port, - tcp_port, - punch_info.nat_type.enum_value_or_default().into(), - ); - { - let peer_nat_info = peer_nat_info.clone(); - self.peer_nat_info_map.write().insert(source, peer_nat_info); - } - if !punch_info.reply { - 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(|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 = - u32::from(nat_info.local_ipv4().unwrap_or(Ipv4Addr::UNSPECIFIED)); - punch_reply.local_port = nat_info.udp_port as u32; - if let Some(ipv6) = nat_info.ipv6() { - punch_reply.ipv6 = ipv6.octets().to_vec(); - punch_reply.ipv6_port = nat_info.udp_port as u32; - } - let bytes = punch_reply.write_to_bytes()?; - let mut punch_packet = - NetPacket::new_encrypt(vec![0u8; 12 + bytes.len() + ENCRYPTION_RESERVED])?; - punch_packet.set_version(Version::V1); - punch_packet.set_protocol(Protocol::OtherTurn); - punch_packet.set_transport_protocol(other_turn_packet::Protocol::Punch.into()); - punch_packet.first_set_ttl(MAX_TTL); - punch_packet.set_source(current_device.virtual_ip()); - punch_packet.set_destination(source); - punch_packet.set_payload(&bytes)?; - if self.punch(source, peer_nat_info) { - self.client_cipher.encrypt_ipv4(&mut punch_packet)?; - context.try_send_by_key(punch_packet.buffer(), route_key)?; - } - } else { - self.punch(source, peer_nat_info); - } - } - other_turn_packet::Protocol::Unknown(e) => { - log::warn!("不支持的转发协议 {:?},source:{:?}", e, source); - } - } - Ok(()) - } - 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(), - } - } -} - -/// 处理服务端数据 -impl ChannelDataHandler { - fn server_packet_handle( - &self, - context: &Context, - current_device: CurrentDeviceInfo, - buf: &mut [u8], - data_len: usize, - route_key: &RouteKey, - ) -> crate::Result<()> { - let net_packet = NetPacket::new0(data_len, &buf[14..])?; - let source = net_packet.source(); - match net_packet.protocol() { - Protocol::Service => { - self.service(context, current_device, net_packet, route_key)?; - } - Protocol::Error => { - self.error(context, current_device, source, net_packet, route_key)?; - } - Protocol::Control => { - self.control_gateway(context, current_device, net_packet, route_key)?; - } - Protocol::IpTurn => { - match ip_turn_packet::Protocol::from(net_packet.transport_protocol()) { - ip_turn_packet::Protocol::Ipv4 => { - let ipv4 = IpV4Packet::new(net_packet.payload())?; - match ipv4.protocol() { - ipv4::protocol::Protocol::Igmp => { - if let Some(igmp_server) = &self.igmp_server { - igmp_server.handle(ipv4.payload(), source)?; - } - return Ok(()); - } - ipv4::protocol::Protocol::Icmp => { - if ipv4.destination_ip() == current_device.virtual_ip { - let icmp_packet = icmp::IcmpPacket::new(ipv4.payload())?; - if icmp_packet.kind() == Kind::EchoReply { - self.device_writer.write_ipv4(&mut buf[12..])?; - return Ok(()); - } - } - } - _ => {} - } - } - ip_turn_packet::Protocol::Ipv4Broadcast => {} - ip_turn_packet::Protocol::Unknown(_) => {} - } - } - Protocol::OtherTurn => {} - Protocol::UnKnow(_) => {} - } - return Ok(()); - } - fn control_gateway( - &self, - context: &Context, - current_device: CurrentDeviceInfo, - net_packet: NetPacket<&[u8]>, - route_key: &RouteKey, - ) -> crate::Result<()> { - match ControlPacket::new(net_packet.transport_protocol(), net_packet.payload())? { - ControlPacket::PongPacket(pong_packet) => { - let metric = net_packet.source_ttl() - net_packet.ttl() + 1; - self.pong_packet( - true, - metric, - context, - current_device, - net_packet.source(), - pong_packet, - route_key, - )?; - } - ControlPacket::AddrResponse(addr_packet) => self - .nat_test - .update_addr(addr_packet.ipv4(), addr_packet.port()), - _ => {} - } - Ok(()) - } - fn service( - &self, - context: &Context, - current_device: CurrentDeviceInfo, - net_packet: NetPacket<&[u8]>, - route_key: &RouteKey, - ) -> crate::Result<()> { - 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())?; - - if self.nat_test.can_update() { - let context = context.clone(); - let nat_test = self.nat_test.clone(); - std::thread::spawn(move || { - tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .unwrap() - .block_on(async move { - let local_ipv4 = nat::local_ipv4(); - let ipv6 = nat::local_ipv6(); - let udp_port = nat_test.nat_info().udp_port; - let tcp_port = nat_test.nat_info().tcp_port; - let nat_info = nat_test - .re_test( - Ipv4Addr::from(response.public_ip), - response.public_port as u16, - local_ipv4, - ipv6, - udp_port, - tcp_port, - ) - .await; - 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 { - // ip发生变化 - log::info!("ip发生变化,old_ip:{:?},new_ip:{:?}", current_ip, new_ip); - #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))] - let old_netmask = current_device.virtual_netmask; - #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))] - let old_gateway = current_device.virtual_gateway(); - let virtual_ip = Ipv4Addr::from(response.virtual_ip); - let virtual_gateway = Ipv4Addr::from(response.virtual_gateway); - let virtual_netmask = Ipv4Addr::from(response.virtual_netmask); - #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))] - 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) - { - log::warn!("替换失败:{:?}", e); - } - } - self.connect_status.store(ConnectStatus::Connected); - } - service_packet::Protocol::PollDeviceList => {} - service_packet::Protocol::PushDeviceList => { - let device_list_t = DeviceList::parse_from_bytes(net_packet.payload())?; - let ip_list: Vec = device_list_t - .device_info_list - .into_iter() - .map(|info| { - PeerDeviceInfo::new( - Ipv4Addr::from(info.virtual_ip), - info.name, - info.device_status as u8, - info.client_secret, - ) - }) - .collect(); - let route = Route::from(*route_key, 2, 199); - for x in &ip_list { - if x.status == PeerDeviceStatus::Online { - context.add_route_if_absent(x.virtual_ip, route); - } - } - let mut dev = self.device_list.lock(); - if dev.0 != device_list_t.epoch as u16 { - dev.0 = device_list_t.epoch as u16; - dev.1 = ip_list; - } - } - service_packet::Protocol::Unknown(u) => { - log::warn!("未知服务协议:{}", u); - } - _ => {} - } - Ok(()) - } - fn error( - &self, - _context: &Context, - current_device: CurrentDeviceInfo, - _source: Ipv4Addr, - net_packet: NetPacket<&[u8]>, - _route_key: &RouteKey, - ) -> crate::Result<()> { - log::info!("current_device:{:?}", current_device); - match InErrorPacket::new(net_packet.transport_protocol(), net_packet.payload())? { - InErrorPacket::TokenError => { - 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(current_device.virtual_ip)?; - } - InErrorPacket::AddressExhausted => { - //地址用尽 - return Err(Error::Stop("IP address has been exhausted".to_string())); - } - InErrorPacket::OtherError(e) => { - log::error!("OtherError {:?}", e.message()); - } - InErrorPacket::IpAlreadyExists => { - log::error!("IpAlreadyExists"); - } - InErrorPacket::InvalidIp => { - log::error!("InvalidIp"); - } - InErrorPacket::NoKey => {} - } - Ok(()) - } -} diff --git a/vnt/src/handle/registration_handler.rs b/vnt/src/handle/registration_handler.rs deleted file mode 100644 index a1c6f45..0000000 --- a/vnt/src/handle/registration_handler.rs +++ /dev/null @@ -1,262 +0,0 @@ -use crossbeam_utils::atomic::AtomicCell; -use std::io::{Read, Write}; -use std::net::{Ipv4Addr, SocketAddr, SocketAddrV6}; -use std::time::{Duration, Instant}; - -use crate::channel::sender::ChannelSender; -use crate::cipher::Cipher; -use crate::handle::PeerDeviceInfo; -use protobuf::Message; -use std::net::TcpStream; -use std::net::UdpSocket; - -use crate::proto::message::{RegistrationRequest, RegistrationResponse}; -use crate::protocol::body::ENCRYPTION_RESERVED; -use crate::protocol::error_packet::InErrorPacket; -use crate::protocol::{service_packet, NetPacket, Protocol, Version, MAX_TTL}; - -pub enum ReqEnum { - TokenError, - AddressExhausted, - IpAlreadyExists, - InvalidIp, - Timeout, - ServerError(String), - Other(String), -} - -#[derive(Clone, Debug)] -pub struct RegResponse { - pub virtual_ip: Ipv4Addr, - pub virtual_gateway: Ipv4Addr, - pub virtual_netmask: Ipv4Addr, - pub epoch: u16, - pub device_info_list: Vec, - pub public_ip: Ipv4Addr, - pub public_port: u16, -} - -///向中继服务器注册，token标识一个虚拟网关，device_id防止多次注册时得到的ip不一致 -pub fn registration( - main_channel: &UdpSocket, - main_tcp_channel: Option<&mut TcpStream>, - server_cipher: &Cipher, - server_address: SocketAddr, - token: String, - device_id: String, - name: String, - ip: Ipv4Addr, - client_secret: bool, -) -> Result { - let request_packet = registration_request_packet( - server_cipher, - token.clone(), - device_id.clone(), - name.clone(), - ip, - false, - false, - client_secret, - ) - .unwrap(); - let buf = request_packet.buffer(); - let mut recv_buf = [0u8; 10240]; - let recv_buf = if let Some(main_tcp_channel) = main_tcp_channel { - let mut vec = vec![0; 4 + buf.len()]; - let len = buf.len(); - vec[2] = (len >> 8) as u8; - vec[3] = (len & 0xFF) as u8; - vec[4..].copy_from_slice(buf); - if let Err(e) = main_tcp_channel.write_all(&vec) { - return Err(ReqEnum::Other(format!("send error:{}", e))); - } - if let Err(e) = main_tcp_channel.read_exact(&mut recv_buf[..4]) { - return Err(ReqEnum::Other(format!("read error:{}", e))); - } - let len = 4 + (((recv_buf[2] as u16) << 8) | recv_buf[3] as u16) as usize; - if len > recv_buf.len() { - return Err(ReqEnum::Other("too long".to_string())); - } - if let Err(e) = main_tcp_channel.read_exact(&mut recv_buf[4..len]) { - return Err(ReqEnum::Other(format!("read error:{}", e))); - } - &mut recv_buf[4..len] - } else { - let server_address = match server_address { - SocketAddr::V4(ipv4) => SocketAddr::V6(SocketAddrV6::new( - ipv4.ip().to_ipv6_mapped(), - ipv4.port(), - 0, - 0, - )), - SocketAddr::V6(_) => server_address, - }; - if let Err(e) = main_channel.send_to(buf, server_address) { - return Err(ReqEnum::Other(format!("send error:{}", e))); - } - match main_channel.recv_from(&mut recv_buf) { - Ok((len, addr)) => { - if server_address != addr { - log::warn!("请求{:?}和响应{:?}地址不一致", server_address, addr); - } - &mut recv_buf[..len] - } - Err(e) => { - return Err(ReqEnum::Other(format!("receiver error:{}", e))); - } - } - }; - let mut net_packet = match NetPacket::new(recv_buf) { - Ok(net_packet) => net_packet, - Err(e) => { - return Err(ReqEnum::ServerError(format!("{}", e))); - } - }; - if let Err(e) = server_cipher.decrypt_ipv4(&mut net_packet) { - return Err(ReqEnum::ServerError(format!("decrypt_ipv4 {}", e))); - } - match net_packet.protocol() { - Protocol::Service => { - match service_packet::Protocol::from(net_packet.transport_protocol()) { - service_packet::Protocol::RegistrationResponse => { - match RegistrationResponse::parse_from_bytes(net_packet.payload()) { - Ok(response) => { - let device_info_list: Vec = response - .device_info_list - .into_iter() - .map(|info| { - PeerDeviceInfo::new( - Ipv4Addr::from(info.virtual_ip), - info.name, - info.device_status as u8, - info.client_secret, - ) - }) - .collect(); - Ok(RegResponse { - virtual_ip: Ipv4Addr::from(response.virtual_ip), - virtual_gateway: Ipv4Addr::from(response.virtual_gateway), - virtual_netmask: Ipv4Addr::from(response.virtual_netmask), - epoch: response.epoch as u16, - device_info_list, - public_ip: Ipv4Addr::from(response.public_ip), - public_port: response.public_port as u16, - }) - } - Err(_) => Err(ReqEnum::ServerError("invalid data".to_string())), - } - } - _ => Err(ReqEnum::ServerError("invalid data".to_string())), - } - } - Protocol::Error => { - match InErrorPacket::new(net_packet.transport_protocol(), net_packet.payload()) { - Ok(e) => match e { - InErrorPacket::TokenError => Err(ReqEnum::TokenError), - InErrorPacket::Disconnect => { - Err(ReqEnum::ServerError("disconnect".to_string())) - } - InErrorPacket::AddressExhausted => Err(ReqEnum::AddressExhausted), - InErrorPacket::OtherError(e) => match e.message() { - Ok(str) => Err(ReqEnum::ServerError(str)), - Err(e) => Err(ReqEnum::Other(format!("{}", e))), - }, - InErrorPacket::IpAlreadyExists => Err(ReqEnum::IpAlreadyExists), - InErrorPacket::InvalidIp => Err(ReqEnum::InvalidIp), - InErrorPacket::NoKey => Err(ReqEnum::ServerError("no key".to_string())), - }, - Err(e) => Err(ReqEnum::Other(format!("{}", e))), - } - } - _ => Err(ReqEnum::ServerError("invalid data".to_string())), - } -} - -fn registration_request_packet( - server_cipher: &Cipher, - token: String, - device_id: String, - name: String, - ip: Ipv4Addr, - is_fast: bool, - allow_ip_change: bool, - client_secret: bool, -) -> crate::Result>> { - let mut request = RegistrationRequest::new(); - request.token = token; - request.device_id = device_id; - request.name = name; - request.virtual_ip = ip.into(); - request.allow_ip_change = allow_ip_change; - request.is_fast = is_fast; - request.version = crate::VNT_VERSION.to_string(); - request.client_secret = client_secret; - let bytes = request.write_to_bytes()?; - let buf = vec![0u8; 12 + bytes.len() + ENCRYPTION_RESERVED]; - let mut net_packet = NetPacket::new_encrypt(buf)?; - net_packet.set_version(Version::V1); - net_packet.set_gateway_flag(true); - net_packet.set_protocol(Protocol::Service); - net_packet.set_transport_protocol(service_packet::Protocol::RegistrationRequest.into()); - net_packet.first_set_ttl(MAX_TTL); - net_packet.set_payload(&bytes)?; - server_cipher.encrypt_ipv4(&mut net_packet)?; - Ok(net_packet) -} - -pub struct Register { - server_cipher: Cipher, - sender: ChannelSender, - server_address: SocketAddr, - token: String, - device_id: String, - name: String, - time: AtomicCell, - client_secret: bool, -} - -impl Register { - pub fn new( - server_cipher: Cipher, - sender: ChannelSender, - server_address: SocketAddr, - token: String, - device_id: String, - name: String, - client_secret: bool, - ) -> Self { - Self { - server_cipher, - sender, - server_address, - token, - device_id, - name, - time: AtomicCell::new(Instant::now()), - client_secret, - } - } - pub fn fast_register(&self, ip: Ipv4Addr) -> crate::Result<()> { - let last = self.time.load(); - if last.elapsed() < Duration::from_secs(3) - || self.time.compare_exchange(last, Instant::now()).is_err() - { - //短时间不重复注册 - return Ok(()); - } - log::info!("重新连接"); - let request_packet = registration_request_packet( - &self.server_cipher, - self.token.clone(), - self.device_id.clone(), - self.name.clone(), - ip, - false, - true, - self.client_secret, - )?; - let buf = request_packet.buffer(); - self.sender.send_main(buf, self.server_address)?; - Ok(()) - } -} diff --git a/vnt/src/handle/tun_tap/tap_handler.rs b/vnt/src/handle/tun_tap/tap_handler.rs deleted file mode 100644 index 54bcd9f..0000000 --- a/vnt/src/handle/tun_tap/tap_handler.rs +++ /dev/null @@ -1,260 +0,0 @@ -use std::sync::Arc; -use std::{io, thread}; - -use crossbeam_utils::atomic::AtomicCell; - -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::channel::sender::ChannelSender; -use crate::cipher::Cipher; -use crate::core::status::VntWorker; -use crate::external_route::ExternalRoute; -use crate::handle::tun_tap::channel_group::{buf_channel_group, BufSenderGroup}; -use crate::handle::CurrentDeviceInfo; -use crate::igmp_server::IgmpServer; -#[cfg(feature = "ip_proxy")] -use crate::ip_proxy::IpProxyMap; -use crate::tun_tap_device::{DeviceReader, DeviceWriter}; - -pub fn start( - worker: VntWorker, - sender: ChannelSender, - device_reader: DeviceReader, - device_writer: DeviceWriter, - igmp_server: Option, - current_device: Arc>, - ip_route: Option, - #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, - client_cipher: Cipher, - server_cipher: Cipher, - parallel: usize, -) { - if parallel == 1 { - thread::Builder::new() - .name("tap_handler".into()) - .spawn(move || { - if let Err(e) = start_simple( - &sender, - device_reader, - &device_writer, - igmp_server, - current_device, - ip_route, - #[cfg(feature = "ip_proxy")] - ip_proxy_map, - client_cipher, - server_cipher, - ) { - log::warn!("tap:{:?}", e); - } - let _ = sender.close(); - let _ = device_writer.close(); - worker.stop_all(); - }) - .unwrap(); - } else { - let (buf_sender, buf_receiver) = buf_channel_group(parallel); - for buf_receiver in buf_receiver.0 { - let sender = sender.clone(); - let device_writer = device_writer.clone(); - let igmp_server = igmp_server.clone(); - let current_device = current_device.clone(); - let ip_route = ip_route.clone(); - #[cfg(feature = "ip_proxy")] - let ip_proxy_map = ip_proxy_map.clone(); - let client_cipher = client_cipher.clone(); - let server_cipher = server_cipher.clone(); - thread::spawn(move || { - while let Ok((mut buf, _, len)) = buf_receiver.recv() { - match handle( - &mut buf, - len, - &igmp_server, - ¤t_device, - &device_writer, - &sender, - &ip_route, - #[cfg(feature = "ip_proxy")] - &ip_proxy_map, - &client_cipher, - &server_cipher, - ) { - Ok(_) => {} - Err(e) => { - log::warn!("{:?}", e) - } - } - } - let _ = sender.close(); - let _ = device_writer.close(); - }); - } - thread::Builder::new() - .name("tap_handler".into()) - .spawn(move || { - if let Err(e) = start_(&sender, device_reader, buf_sender) { - log::warn!("tap:{:?}", e); - } - let _ = sender.close(); - let _ = device_writer.close(); - worker.stop_all(); - }) - .unwrap(); - } -} - -fn start_( - sender: &ChannelSender, - device_reader: DeviceReader, - mut buf_sender: BufSenderGroup, -) -> io::Result<()> { - loop { - let mut buf = vec![0; 4096]; - if sender.is_close() { - return Ok(()); - } - let start = 0; - let len = device_reader.read(&mut buf)?; - if !buf_sender.send((buf, start, len)) { - return Err(io::Error::new( - io::ErrorKind::Other, - "tap buf_sender发送失败", - )); - } - } -} - -fn start_simple( - sender: &ChannelSender, - device_reader: DeviceReader, - device_writer: &DeviceWriter, - igmp_server: Option, - current_device: Arc>, - ip_route: Option, - #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, - client_cipher: Cipher, - server_cipher: Cipher, -) -> io::Result<()> { - let mut buf = [0; 4096]; - loop { - if sender.is_close() { - return Ok(()); - } - let len = device_reader.read(&mut buf)?; - if let Err(e) = handle( - &mut buf, - len, - &igmp_server, - ¤t_device, - device_writer, - sender, - &ip_route, - #[cfg(feature = "ip_proxy")] - &ip_proxy_map, - &client_cipher, - &server_cipher, - ) { - log::warn!("tap handle{:?}", e); - } - } -} - -fn handle( - buf: &mut [u8], - len: usize, - igmp_server: &Option, - current_device: &AtomicCell, - device_writer: &DeviceWriter, - sender: &ChannelSender, - ip_route: &Option, - #[cfg(feature = "ip_proxy")] proxy_map: &Option, - client_cipher: &Cipher, - server_cipher: &Cipher, -) -> crate::Result<()> { - let mut ethernet_packet = EthernetPacket::new(&mut buf[..len])?; - 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], - !sender_h[5], - 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], - !sender_h[5], - 234, - ]); - out_ethernet_packet.set_destination(sender_h); - device_writer.write_ethernet_tap(&out_ethernet_packet.buffer)?; - } - ethernet::protocol::Protocol::Ipv4 => { - let mut ipv4_packet = IpV4Packet::unchecked(ethernet_packet.payload_mut()); - let src_ip = ipv4_packet.source_ip(); - let dest_ip = ipv4_packet.destination_ip(); - let protocol = ipv4_packet.protocol(); - if src_ip == dest_ip { - if 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(); - ipv4_packet.set_source_ip(dest_ip); - ipv4_packet.set_destination_ip(src_ip); - ipv4_packet.update_checksum(); - let source = ethernet_packet.source().to_vec(); - let dest = ethernet_packet.destination().to_vec(); - ethernet_packet.set_source(&dest); - ethernet_packet.set_destination(&source); - device_writer.write_ethernet_tap(ðernet_packet.buffer)?; - } - } - return Ok(()); - } - // 以太网帧头部14字节，预留12字节 - return crate::handle::tun_tap::base_handle( - sender, - &mut buf[2..], - len - 2, - igmp_server, - current_device, - ip_route, - #[cfg(feature = "ip_proxy")] - proxy_map, - client_cipher, - server_cipher, - ); - } - _ => { - // log::warn!("不支持的二层协议：{:?}",p) - } - } - Ok(()) -}