From dacce892ffb9e8a5570e8eb55050bc9654681c21 Mon Sep 17 00:00:00 2001 From: lubeilin <1791778603@qq.com> Date: Sun, 8 Oct 2023 17:16:02 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=AF=E9=80=89ip=E4=BB=A3=E7=90=86=EF=BC=8C?= =?UTF-8?q?=E5=85=B3=E9=97=AD=E5=90=8E=E5=8F=AF=E4=BD=BF=E7=94=A8=E5=A4=96?= =?UTF-8?q?=E9=83=A8=E5=91=BD=E4=BB=A4=E6=9D=A5=E8=BF=9B=E8=A1=8Cip?= =?UTF-8?q?=E8=BD=AC=E5=8F=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- vnt-cli/Cargo.toml | 4 +- vnt/Cargo.toml | 4 +- vnt/packet/src/tcp/tcp.rs | 9 +- vnt/src/channel/channel.rs | 117 ++++++++++---------- vnt/src/channel/idle.rs | 27 +++-- vnt/src/core/mod.rs | 44 +++++--- vnt/src/external_route/mod.rs | 3 + vnt/src/handle/recv_handler.rs | 123 +++++++++------------ vnt/src/handle/tun_tap/mod.rs | 82 ++++---------- vnt/src/handle/tun_tap/tap_handler.rs | 12 ++- vnt/src/handle/tun_tap/tun_handler.rs | 18 ++-- vnt/src/igmp_server/mod.rs | 72 +++++++++---- vnt/src/ip_proxy/icmp_proxy.rs | 114 ++++++++++---------- vnt/src/ip_proxy/mod.rs | 148 +++++++++++++++++--------- vnt/src/ip_proxy/tcp_proxy.rs | 67 ++++++++++-- vnt/src/ip_proxy/udp_proxy.rs | 66 +++++++++++- vnt/src/lib.rs | 3 +- 17 files changed, 538 insertions(+), 375 deletions(-) diff --git a/vnt-cli/Cargo.toml b/vnt-cli/Cargo.toml index d459bb4..207aa25 100644 --- a/vnt-cli/Cargo.toml +++ b/vnt-cli/Cargo.toml @@ -31,7 +31,7 @@ sudo = "0.6.0" winapi = { version = "0.3.9", features = ["handleapi", "processthreadsapi", "winnt", "securitybaseapi", "impl-default"] } [features] -default = ["server_encrypt","aes_gcm","aes_cbc","aes_ecb","sm4_cbc"] +default = ["server_encrypt","aes_gcm","aes_cbc","aes_ecb","sm4_cbc","ip_proxy"] openssl = ["vnt/openssl"] openssl-vendored = ["vnt/openssl-vendored"] ring-cipher = ["vnt/ring-cipher"] @@ -40,6 +40,6 @@ aes_ecb=["vnt/aes_ecb"] sm4_cbc=["vnt/sm4_cbc"] aes_gcm=["vnt/aes_gcm"] server_encrypt=["vnt/server_encrypt"] - +ip_proxy=["vnt/ip_proxy"] [build-dependencies] embed-manifest = "1.4.0" \ No newline at end of file diff --git a/vnt/Cargo.toml b/vnt/Cargo.toml index 3e0cb57..949d3fc 100644 --- a/vnt/Cargo.toml +++ b/vnt/Cargo.toml @@ -12,7 +12,7 @@ log = "0.4.17" libc = "0.2.137" crossbeam-utils = "0.8" crossbeam-epoch = "0.9.15" -dashmap = "5.5.3" +dashmap = {version = "5.5.3",optional = true} parking_lot = "0.12.1" rand = "0.8.5" sha2 = { version = "0.10.6", features = ["oid"] } @@ -54,4 +54,4 @@ aes_ecb=["ecb"] sm4_cbc=["libsm"] aes_gcm=["aes-gcm"] server_encrypt =["rsa","spki"] - +ip_proxy=["dashmap"] diff --git a/vnt/packet/src/tcp/tcp.rs b/vnt/packet/src/tcp/tcp.rs index 38139a7..a68ac9d 100644 --- a/vnt/packet/src/tcp/tcp.rs +++ b/vnt/packet/src/tcp/tcp.rs @@ -76,7 +76,14 @@ impl> TcpPacket { Ok(packet) } } - +impl + AsMut<[u8]>> TcpPacket { + pub fn set_source_ip(&mut self, value: Ipv4Addr) { + self.source_ip = value; + } + pub fn set_destination_ip(&mut self, value: Ipv4Addr) { + self.destination_ip = value; + } +} impl + AsMut<[u8]>> TcpPacket { fn set_checksum(&mut self, value: u16) { self.buffer.as_mut()[16..18].copy_from_slice(&value.to_be_bytes()) diff --git a/vnt/src/channel/channel.rs b/vnt/src/channel/channel.rs index 198ac96..39acc43 100644 --- a/vnt/src/channel/channel.rs +++ b/vnt/src/channel/channel.rs @@ -3,7 +3,6 @@ use std::io::{Read, Write}; use std::net::TcpStream; use std::net::UdpSocket as StdUdpSocket; use std::net::{Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr}; -use std::ops::Sub; use std::sync::atomic::Ordering; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -11,7 +10,6 @@ use std::{io, thread}; use crossbeam_epoch::{Atomic, Owned}; use crossbeam_utils::atomic::AtomicCell; -use dashmap::DashMap; use tokio::net::UdpSocket; use tokio::sync::watch::{channel, Receiver, Sender}; @@ -20,7 +18,6 @@ use crate::channel::{Route, RouteKey, Status, TCP_ID, UDP_ID, UDP_V6_ID}; use crate::core::status::VntWorker; use crate::handle::recv_handler::ChannelDataHandler; use crate::handle::CurrentDeviceInfo; -use crate::ip_proxy::DashMapNew; pub struct ContextInner { //udp用于打洞、服务端通信(可选) @@ -28,8 +25,7 @@ pub struct ContextInner { pub(crate) main_channel_ipv6: Option>, //在udp的基础上，可以选择使用tcp和服务端通信 pub(crate) main_tcp_channel: Option>>, - pub(crate) route_table: Atomic>>, - pub(crate) route_table_time: DashMap<(RouteKey, Ipv4Addr), Instant>, + pub(crate) route_table: Atomic>)>>>, pub(crate) status_receiver: Receiver, pub(crate) status_sender: Sender, pub(crate) udp_map: Atomic>>, @@ -58,7 +54,6 @@ impl Context { main_channel_ipv6, main_tcp_channel, route_table: Atomic::new(HashMap::with_capacity(16)), - route_table_time: DashMap::new_cap(16), status_receiver, status_sender, udp_map: Atomic::new(HashMap::with_capacity(16)), @@ -133,9 +128,9 @@ impl Context { fn insert_udp_(&self, id: usize, udp: Option>) { let guard = &crossbeam_epoch::pin(); let udp_map = &self.inner.udp_map; - let mut udp_map_shared = self.inner.udp_map.load(Ordering::Relaxed, guard); + let mut udp_map_shared = udp_map.load(Ordering::Acquire, guard); loop { - let mut map = unsafe { udp_map_shared.as_ref().unwrap().clone() }; + let mut map = unsafe { udp_map_shared.deref().clone() }; match udp.clone() { None => { map.remove(&id); @@ -147,7 +142,7 @@ impl Context { match udp_map.compare_exchange( udp_map_shared, Owned::new(map), - Ordering::Relaxed, + Ordering::AcqRel, Ordering::Relaxed, guard, ) { @@ -191,8 +186,7 @@ impl Context { self.inner .udp_map .load(Ordering::Relaxed, guard) - .as_ref() - .unwrap() + .deref() .clone() }; if table.is_empty() { @@ -222,27 +216,24 @@ impl Context { self.inner .route_table .load(Ordering::Relaxed, guard) - .as_ref() - .unwrap() + .deref() }; if let Some(v) = table.get(id) { if v.is_empty() { return Err(io::Error::new(io::ErrorKind::NotFound, "route not found")); } - let route = v[0]; + let (route, time) = &v[0]; if route.rt == 199 { //这通常是刚加入路由，直接放弃使用,避免抖动 return Err(io::Error::new(io::ErrorKind::NotFound, "route not found")); } if !route.is_p2p() { - if let Some(time) = self.inner.route_table_time.get(&(route.route_key(), *id)) { - //借道传输时，长时间不通信的通道不使用 - if time.value().elapsed() > Duration::from_secs(6) { - return Err(io::Error::new(io::ErrorKind::NotFound, "route time out")); - } + //借道传输时，长时间不通信的通道不使用 + if time.load().elapsed() > Duration::from_secs(6) { + return Err(io::Error::new(io::ErrorKind::NotFound, "route time out")); } } - return Ok(route); + return Ok(*route); } Err(io::Error::new(io::ErrorKind::NotFound, "route not found")) } @@ -311,8 +302,7 @@ impl Context { self.inner .udp_map .load(Ordering::Relaxed, guard) - .as_ref() - .unwrap() + .deref() }; udp_map.get(&route_key.index).cloned() } @@ -327,13 +317,12 @@ impl Context { let key = route.route_key(); let guard = &crossbeam_epoch::pin(); let route_table = &self.inner.route_table; - let mut table_share = route_table.load(Ordering::Relaxed, guard); + let mut table_share = route_table.load(Ordering::Acquire, guard); loop { - let mut table = unsafe { table_share.as_ref().unwrap().clone() }; - + let mut table = unsafe { table_share.deref().clone() }; let list = table.entry(id).or_insert_with(|| Vec::with_capacity(4)); let mut exist = false; - for x in list.iter_mut() { + for (x, time) in list.iter_mut() { if x.metric < route.metric { //不能比当前的路径更长 return; @@ -345,18 +334,19 @@ impl Context { x.metric = route.metric; x.rt = route.rt; exist = true; + time.store(Instant::now()); break; } } if exist { - list.sort_by_key(|k| k.sort_key()); + list.sort_by_key(|(k, _)| k.sort_key()); } else { if route.metric == 1 { //添加了直连的则排除非直连的 - list.retain(|k| k.metric == 1); + list.retain(|(k, _)| k.metric == 1); } - list.push(route); - list.sort_by_key(|k| k.sort_key()); + list.push((route, Arc::new(AtomicCell::new(Instant::now())))); + list.sort_by_key(|(k, _)| k.sort_key()); let max_len = self.inner.channel_num + 1; if list.len() > max_len { list.truncate(max_len); @@ -365,7 +355,7 @@ impl Context { match route_table.compare_exchange( table_share, Owned::new(table), - Ordering::Relaxed, + Ordering::AcqRel, Ordering::Relaxed, guard, ) { @@ -378,10 +368,6 @@ impl Context { } } } - - self.inner - .route_table_time - .insert((key, id), Instant::now().sub(Duration::from_secs(10))); } pub fn route(&self, id: &Ipv4Addr) -> Option> { let guard = &crossbeam_epoch::pin(); @@ -389,11 +375,10 @@ impl Context { self.inner .route_table .load(Ordering::Relaxed, guard) - .as_ref() - .unwrap() + .deref() }; if let Some(v) = table.get(id) { - Some(v.clone()) + Some(v.iter().map(|(i, _)| *i).collect()) } else { None } @@ -404,11 +389,10 @@ impl Context { self.inner .route_table .load(Ordering::Relaxed, guard) - .as_ref() - .unwrap() + .deref() }; if let Some(v) = table.get(id) { - v.first().map(|v| *v) + v.first().map(|(i, _)| *i) } else { None } @@ -419,11 +403,10 @@ impl Context { self.inner .route_table .load(Ordering::Relaxed, guard) - .as_ref() - .unwrap() + .deref() }; for (k, v) in table.iter() { - for route in v { + for (route, _) in v { if &route.route_key() == route_key && route.is_p2p() { return Some(*k); } @@ -437,11 +420,10 @@ impl Context { self.inner .route_table .load(Ordering::Relaxed, guard) - .as_ref() - .unwrap() + .deref() }; if let Some(v) = table.get(id) { - if v.iter().filter(|k| k.is_p2p()).count() >= self.inner.channel_num { + if v.iter().filter(|(k, _)| k.is_p2p()).count() >= self.inner.channel_num { return false; } } @@ -453,10 +435,12 @@ impl Context { self.inner .route_table .load(Ordering::Relaxed, guard) - .as_ref() - .unwrap() + .deref() }; - table.iter().map(|(k, v)| (k.clone(), v.clone())).collect() + table + .iter() + .map(|(k, v)| (k.clone(), v.iter().map(|(i, _)| *i).collect())) + .collect() } pub fn route_table_one(&self) -> Vec<(Ipv4Addr, Route)> { let mut list = Vec::with_capacity(8); @@ -465,11 +449,10 @@ impl Context { self.inner .route_table .load(Ordering::Relaxed, guard) - .as_ref() - .unwrap() + .deref() }; for (k, v) in table { - if let Some(route) = v.first() { + if let Some((route, _)) = v.first() { list.push((*k, *route)); } } @@ -482,11 +465,10 @@ impl Context { self.inner .route_table .load(Ordering::Relaxed, guard) - .as_ref() - .unwrap() + .deref() }; for (k, v) in table { - if let Some(route) = v.first() { + if let Some((route, _)) = v.first() { if route.metric == 1 { list.push((*k, *route)); } @@ -498,33 +480,42 @@ impl Context { pub fn remove_route(&self, id: &Ipv4Addr, route_key: RouteKey) { let guard = &crossbeam_epoch::pin(); let route_table = &self.inner.route_table; - let mut table_share = route_table.load(Ordering::Relaxed, guard); + let mut table_share = route_table.load(Ordering::Acquire, guard); loop { - let mut table = unsafe { table_share.as_ref().unwrap().clone() }; + let mut table = unsafe { table_share.deref().clone() }; if let Some(routes) = table.get_mut(id) { - routes.retain(|x| x.route_key() != route_key); + routes.retain(|(x, _)| x.route_key() != route_key); match route_table.compare_exchange( table_share, Owned::new(table), - Ordering::Relaxed, + Ordering::AcqRel, Ordering::Relaxed, guard, ) { Ok(_p) => unsafe { guard.defer_destroy(table_share); - self.inner.route_table_time.remove(&(route_key, *id)); return; }, Err(e) => { table_share = e.current; } } + }else{ + return; } } } pub fn update_read_time(&self, id: &Ipv4Addr, route_key: &RouteKey) { - if let Some(mut time) = self.inner.route_table_time.get_mut(&(*route_key, *id)) { - *time.value_mut() = Instant::now(); + let guard = &crossbeam_epoch::pin(); + let table_share = self.inner.route_table.load(Ordering::Relaxed, guard); + let table = unsafe { table_share.deref().clone() }; + if let Some(routes) = table.get(id) { + for (route, time) in routes { + if &route.route_key() == route_key { + time.store(Instant::now()); + break; + } + } } } } diff --git a/vnt/src/channel/idle.rs b/vnt/src/channel/idle.rs index c474535..93e178b 100644 --- a/vnt/src/channel/idle.rs +++ b/vnt/src/channel/idle.rs @@ -3,6 +3,7 @@ use crate::channel::RouteKey; use std::io; use std::io::{Error, ErrorKind}; use std::net::Ipv4Addr; +use std::sync::atomic::Ordering; use std::time::Duration; pub struct Idle { @@ -21,13 +22,25 @@ impl Idle { pub async fn next_idle(&self) -> io::Result<(Ipv4Addr, RouteKey)> { loop { let mut max = Duration::from_secs(0); - for entry in self.context.inner.route_table_time.iter() { - let last_read = entry.value().elapsed(); - if last_read >= self.read_idle { - return Ok((entry.key().1.clone(), entry.key().0.clone())); - } else { - if max < last_read { - max = last_read; + { + let guard = &crossbeam_epoch::pin(); + let table = unsafe { + self.context + .inner + .route_table + .load(Ordering::Relaxed, guard) + .deref() + }; + for (ip, routes) in table.iter() { + for (route, time) in routes { + let last_read = time.load().elapsed(); + if last_read >= self.read_idle { + return Ok((*ip, route.route_key())); + } else { + if max < last_read { + max = last_read; + } + } } } } diff --git a/vnt/src/core/mod.rs b/vnt/src/core/mod.rs index a140638..154b457 100644 --- a/vnt/src/core/mod.rs +++ b/vnt/src/core/mod.rs @@ -1,37 +1,38 @@ +use std::collections::HashMap; use std::io; use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4}; -use std::sync::Arc; -use std::time::Duration; - -use crossbeam_utils::atomic::AtomicCell; -use dashmap::DashMap; -use parking_lot::Mutex; -use rand::Rng; use std::net::TcpStream; use std::net::UdpSocket; +use std::sync::Arc; +use std::sync::atomic::Ordering; +use std::time::Duration; + +use crossbeam_epoch::Atomic; +use crossbeam_utils::atomic::AtomicCell; +use parking_lot::Mutex; +use rand::Rng; use tokio::sync::mpsc::channel; +use crate::channel::{Route, RouteKey}; use crate::channel::channel::{Channel, Context}; use crate::channel::idle::Idle; use crate::channel::punch::{NatInfo, Punch, PunchModel}; use crate::channel::sender::ChannelSender; -use crate::channel::{Route, RouteKey}; use crate::cipher::{Cipher, CipherModel, RsaCipher}; use crate::core::status::VntStatusManger; use crate::error::Error; use crate::external_route::{AllowExternalRoute, ExternalRoute}; +use crate::handle::{ + ConnectStatus, CurrentDeviceInfo, handshake_handler, heartbeat_handler, PeerDeviceInfo, + punch_handler, registration_handler, +}; use crate::handle::handshake_handler::HandshakeEnum; use crate::handle::recv_handler::ChannelDataHandler; use crate::handle::registration_handler::{RegResponse, ReqEnum}; #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))] use crate::handle::tun_tap::tap_handler; use crate::handle::tun_tap::tun_handler; -use crate::handle::{ - handshake_handler, heartbeat_handler, punch_handler, registration_handler, ConnectStatus, - CurrentDeviceInfo, PeerDeviceInfo, -}; use crate::igmp_server::IgmpServer; -use crate::ip_proxy::DashMapNew; use crate::nat::NatTest; use crate::tun_tap_device; use crate::tun_tap_device::{DeviceReader, DeviceWriter}; @@ -52,7 +53,7 @@ pub struct Vnt { device_list: Arc)>>, nat_test: NatTest, connect_status: Arc>, - peer_nat_info_map: Arc>, + peer_nat_info_map: Arc>>, } pub struct VntUtil { @@ -277,7 +278,7 @@ impl VntUtil { )); let device_list: Arc)>> = Arc::new(Mutex::new((response.epoch, response.device_info_list))); - let peer_nat_info_map: Arc> = Arc::new(DashMap::new0()); + let peer_nat_info_map: Arc>> = Arc::new(Atomic::new(HashMap::new())); let connect_status = Arc::new(AtomicCell::new(ConnectStatus::Connected)); let public_ip = response.public_ip; let public_port = response.public_port; @@ -299,6 +300,7 @@ impl VntUtil { } else { Some(ExternalRoute::new(config.in_ips)) }; + #[cfg(feature = "ip_proxy")] let (tcp_proxy, udp_proxy, ip_proxy_map) = if config.out_ips.is_empty() { (None, None, None) } else { @@ -330,6 +332,7 @@ impl VntUtil { igmp_server.clone(), current_device.clone(), in_external_route, + #[cfg(feature = "ip_proxy")] ip_proxy_map.clone(), client_cipher.clone(), self.server_cipher.clone(), @@ -344,6 +347,7 @@ impl VntUtil { igmp_server.clone(), current_device.clone(), in_external_route, + #[cfg(feature = "ip_proxy")] ip_proxy_map.clone(), client_cipher.clone(), self.server_cipher.clone(), @@ -359,6 +363,7 @@ impl VntUtil { igmp_server.clone(), current_device.clone(), in_external_route, + #[cfg(feature = "ip_proxy")] ip_proxy_map.clone(), client_cipher.clone(), self.server_cipher.clone(), @@ -375,6 +380,7 @@ impl VntUtil { device_writer.clone(), connect_status.clone(), peer_nat_info_map.clone(), + #[cfg(feature = "ip_proxy")] ip_proxy_map, out_external_route, cone_sender, @@ -449,6 +455,7 @@ impl VntUtil { )); } } + #[cfg(feature = "ip_proxy")] { //代理 if let Some(tcp_proxy) = tcp_proxy { @@ -457,6 +464,8 @@ impl VntUtil { if let Some(udp_proxy) = udp_proxy { tokio::spawn(udp_proxy.start()); } + } + { let context = context.clone(); let nat_test = nat_test.clone(); tokio::spawn(async move { @@ -494,7 +503,10 @@ impl Vnt { self.current_device.load() } pub fn peer_nat_info(&self, ip: &Ipv4Addr) -> Option { - self.peer_nat_info_map.get(ip).map(|e| e.value().clone()) + let guard = &crossbeam_epoch::pin(); + let shared = self.peer_nat_info_map.load(Ordering::Acquire,guard); + let map = unsafe{shared.deref()}; + map.get(ip).map(|e| e.clone()) } pub fn connection_status(&self) -> ConnectStatus { self.connect_status.load() diff --git a/vnt/src/external_route/mod.rs b/vnt/src/external_route/mod.rs index 07b4d0a..78667f3 100644 --- a/vnt/src/external_route/mod.rs +++ b/vnt/src/external_route/mod.rs @@ -37,6 +37,9 @@ impl AllowExternalRoute { } } pub fn allow(&self, ip: &Ipv4Addr) -> bool { + if self.route_table.is_empty() { + return false; + } let ip = u32::from_be_bytes(ip.octets()); for (dest, mask) in self.route_table.iter() { if *mask & ip == *mask & *dest { diff --git a/vnt/src/handle/recv_handler.rs b/vnt/src/handle/recv_handler.rs index ba838c8..dd3e576 100644 --- a/vnt/src/handle/recv_handler.rs +++ b/vnt/src/handle/recv_handler.rs @@ -1,9 +1,11 @@ +use std::collections::HashMap; use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6}; use std::sync::Arc; +use std::sync::atomic::Ordering; use std::time::{Duration, Instant}; +use crossbeam_epoch::{Atomic, Owned}; use crossbeam_utils::atomic::AtomicCell; -use dashmap::DashMap; use parking_lot::Mutex; use protobuf::Message; use tokio::sync::mpsc::Sender; @@ -12,27 +14,28 @@ use packet::icmp::{icmp, Kind}; use packet::ip::ipv4; use packet::ip::ipv4::packet::IpV4Packet; +use crate::channel::{Route, RouteKey}; 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::{ConnectStatus, CurrentDeviceInfo, PeerDeviceInfo, PeerDeviceStatus}; 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; -use crate::ip_proxy::IpProxyMap; +#[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::{ + control_packet, ip_turn_packet, MAX_TTL, NetPacket, other_turn_packet, Protocol, + service_packet, Version, +}; 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)] @@ -44,7 +47,8 @@ pub struct ChannelDataHandler { igmp_server: Option, device_writer: DeviceWriter, connect_status: Arc>, - peer_nat_info_map: Arc>, + peer_nat_info_map: Arc>>, + #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, out_external_route: AllowExternalRoute, cone_sender: Sender<(Ipv4Addr, NatInfo)>, @@ -66,7 +70,8 @@ impl ChannelDataHandler { igmp_server: Option, device_writer: DeviceWriter, connect_status: Arc>, - peer_nat_info_map: Arc>, + peer_nat_info_map: Arc>>, + #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, out_external_route: AllowExternalRoute, cone_sender: Sender<(Ipv4Addr, NatInfo)>, @@ -86,6 +91,7 @@ impl ChannelDataHandler { device_writer, connect_status, peer_nat_info_map, + #[cfg(feature = "ip_proxy")] ip_proxy_map, out_external_route, cone_sender, @@ -225,75 +231,36 @@ impl ChannelDataHandler { _ => {} } if not_broadcast && ipv4.destination_ip() != destination { - if let Some(ip_proxy_map) = &self.ip_proxy_map { - if self.out_external_route.allow(&ipv4.destination_ip()) { + if self.out_external_route.allow(&ipv4.destination_ip()) { + #[cfg(feature = "ip_proxy")] + if let Some(ip_proxy_map) = &self.ip_proxy_map { match ipv4.protocol() { ipv4::protocol::Protocol::Tcp => { - let dest_ip = ipv4.destination_ip(); - //转发到代理目标地址 - let mut tcp_packet = packet::tcp::tcp::TcpPacket::new( + if ip_proxy_map.tcp_handler.recv_handle( + &mut ipv4, source, destination, - ipv4.payload_mut(), - )?; - let source_port = tcp_packet.source_port(); - let dest_port = tcp_packet.destination_port(); - tcp_packet - .set_destination_port(ip_proxy_map.tcp_proxy_port); - tcp_packet.update_checksum(); - ipv4.set_destination_ip(destination); - ipv4.update_checksum(); - let key = SocketAddrV4::new(source, source_port); - //https://github.com/crossbeam-rs/crossbeam/issues/1023 - ip_proxy_map - .tcp_proxy_map - .insert(key, SocketAddrV4::new(dest_ip, dest_port)); + )? { + return Ok(()); + } } ipv4::protocol::Protocol::Udp => { - let dest_ip = ipv4.destination_ip(); - //转发到代理目标地址 - let mut udp_packet = packet::udp::udp::UdpPacket::new( + if ip_proxy_map.udp_handler.recv_handle( + &mut ipv4, source, destination, - ipv4.payload_mut(), - )?; - let source_port = udp_packet.source_port(); - let dest_port = udp_packet.destination_port(); - udp_packet - .set_destination_port(ip_proxy_map.udp_proxy_port); - udp_packet.update_checksum(); - ipv4.set_destination_ip(destination); - ipv4.update_checksum(); - let key = SocketAddrV4::new(source, source_port); - ip_proxy_map - .udp_proxy_map - .insert(key, SocketAddrV4::new(dest_ip, dest_port)); + )? { + return Ok(()); + } } #[cfg(not(target_os = "android"))] ipv4::protocol::Protocol::Icmp => { - let dest_ip = ipv4.destination_ip(); - //转发到代理目标地址 - let icmp_packet = - icmp::IcmpPacket::new(ipv4.payload())?; - match icmp_packet.header_other() { - icmp::HeaderOther::Identifier(id, seq) => { - ip_proxy_map - .icmp_proxy_map - .insert((dest_ip, id, seq), source); - ip_proxy_map - .send_icmp(ipv4.payload(), &dest_ip)?; - } - _ => { - log::warn!( - "不支持的ip代理Icmp协议:{}->{}->{}", - source, - destination, - dest_ip - ); - return Err(Error::Warn( - "不支持的ip代理Icmp协议".to_string(), - )); - } + if ip_proxy_map.icmp_handler.recv_handle( + &mut ipv4, + source, + destination, + )? { + return Ok(()); } } _ => { @@ -311,23 +278,23 @@ impl ChannelDataHandler { } } else { log::warn!( - "没有ip代理规则{:?}:{}->{}->{}", + "不支持ip代理{:?}:{}->{}->{}", ipv4.protocol(), source, destination, ipv4.destination_ip() ); - return Err(Error::Warn("没有ip代理规则".to_string())); + return Err(Error::Warn("不支持ip代理".to_string())); } } else { log::warn!( - "不支持ip代理{:?}:{}->{}->{}", + "没有ip代理规则{:?}:{}->{}->{}", ipv4.protocol(), source, destination, ipv4.destination_ip() ); - return Err(Error::Warn("不支持ip代理".to_string())); + return Err(Error::Warn("没有ip代理规则".to_string())); } } @@ -503,7 +470,17 @@ impl ChannelDataHandler { ipv6_addr, punch_info.nat_type.enum_value_or_default().into(), ); - self.peer_nat_info_map.insert(source, peer_nat_info.clone()); + { + let guard = &crossbeam_epoch::pin(); + let nat_map = &self.peer_nat_info_map; + let nat_map_shared = nat_map.load(Ordering::Acquire, guard); + let mut map = unsafe { nat_map_shared.deref().clone() }; + map.insert(source, peer_nat_info.clone()); + nat_map.store(Owned::new(map),Ordering::Release); + unsafe { + guard.defer_destroy(nat_map_shared); + } + } if !punch_info.reply { let mut punch_reply = PunchInfo::new(); punch_reply.reply = true; diff --git a/vnt/src/handle/tun_tap/mod.rs b/vnt/src/handle/tun_tap/mod.rs index 385177d..1de31fb 100644 --- a/vnt/src/handle/tun_tap/mod.rs +++ b/vnt/src/handle/tun_tap/mod.rs @@ -1,22 +1,23 @@ +use std::net::Ipv4Addr; +use std::sync::Arc; + +use parking_lot::RwLock; + +use packet::ip::ipv4::packet::IpV4Packet; +use packet::ip::ipv4::protocol::Protocol; + use crate::channel::sender::ChannelSender; use crate::cipher::Cipher; use crate::error::*; use crate::external_route::ExternalRoute; use crate::handle::{check_dest, CurrentDeviceInfo}; use crate::igmp_server::{IgmpServer, Multicast}; -use crate::ip_proxy::IpProxyMap; +#[cfg(feature = "ip_proxy")] +use crate::ip_proxy::{IpProxyMap, ProxyHandler}; use crate::protocol; use crate::protocol::body::ENCRYPTION_RESERVED; use crate::protocol::ip_turn_packet::BroadcastPacket; use crate::protocol::{ip_turn_packet, NetPacket, Version, MAX_TTL}; -use packet::ip::ipv4::packet::IpV4Packet; -use packet::ip::ipv4::protocol::Protocol; -use packet::tcp::tcp::TcpPacket; -use packet::udp::udp::UdpPacket; -use parking_lot::RwLock; -use std::io; -use std::net::{Ipv4Addr, SocketAddrV4}; -use std::sync::Arc; pub mod channel_group; #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))] @@ -64,10 +65,8 @@ fn broadcast( if peer_ips.is_empty() { sender.send_main(net_packet.buffer(), current_device.connect_server)?; } else { - let buf = vec![ - 0 as u8; - 12 + 1 + peer_ips.len() * 4 + net_packet.data_len() + ENCRYPTION_RESERVED - ]; + let buf = + vec![0u8; 12 + 1 + peer_ips.len() * 4 + net_packet.data_len() + ENCRYPTION_RESERVED]; //剩余的发送到服务端，需要告知哪些已发送过 let mut server_packet = NetPacket::new_encrypt(buf)?; server_packet.set_version(Version::V1); @@ -99,16 +98,13 @@ pub fn base_handle( igmp_server: &Option, current_device: CurrentDeviceInfo, ip_route: &Option, + #[cfg(feature = "ip_proxy")] proxy_map: &Option, client_cipher: &Cipher, server_cipher: &Cipher, ) -> Result<()> { let ipv4_packet = IpV4Packet::new(&buf[12..data_len])?; let protocol = ipv4_packet.protocol(); - let ip_head_len = ipv4_packet.header_len() as usize * 4; - if 12 + ip_head_len >= data_len { - Err(io::Error::new(io::ErrorKind::Other, "ip_head_len err"))? - } let src_ip = ipv4_packet.source_ip(); let mut dest_ip = ipv4_packet.destination_ip(); let mut net_packet = NetPacket::new0(data_len, buf)?; @@ -190,55 +186,17 @@ pub fn base_handle( } else { return Ok(()); } - } else if let Some(proxy_map) = proxy_map { + } + #[cfg(feature = "ip_proxy")] + if let Some(proxy_map) = proxy_map { match protocol { Protocol::Tcp => { - let dest_addr = { - let tcp_packet = TcpPacket::new( - src_ip, - dest_ip, - &mut net_packet.payload_mut()[ip_head_len..], - )?; - 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(); - let source_ip = *source_addr.ip(); - let mut tcp_packet = TcpPacket::new( - source_ip, - dest_ip, - &mut net_packet.payload_mut()[ip_head_len..], - )?; - tcp_packet.set_source_port(source_addr.port()); - tcp_packet.update_checksum(); - let mut ipv4_packet = IpV4Packet::new(net_packet.payload_mut())?; - ipv4_packet.set_source_ip(source_ip); - ipv4_packet.update_checksum(); - } + let mut ipv4_packet = IpV4Packet::new(net_packet.payload_mut())?; + proxy_map.tcp_handler.send_handle(&mut ipv4_packet)?; } Protocol::Udp => { - let dest_addr = { - let udp_packet = UdpPacket::new( - src_ip, - dest_ip, - &mut net_packet.payload_mut()[ip_head_len..], - )?; - 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(); - let source_ip = *source_addr.ip(); - let mut udp_packet = UdpPacket::new( - source_ip, - dest_ip, - &mut net_packet.payload_mut()[ip_head_len..], - )?; - udp_packet.set_source_port(source_addr.port()); - udp_packet.update_checksum(); - let mut ipv4_packet = IpV4Packet::new(net_packet.payload_mut())?; - ipv4_packet.set_source_ip(source_ip); - ipv4_packet.update_checksum(); - } + let mut ipv4_packet = IpV4Packet::new(net_packet.payload_mut())?; + proxy_map.udp_handler.send_handle(&mut ipv4_packet)?; } _ => {} } diff --git a/vnt/src/handle/tun_tap/tap_handler.rs b/vnt/src/handle/tun_tap/tap_handler.rs index 97b0309..cbc04d6 100644 --- a/vnt/src/handle/tun_tap/tap_handler.rs +++ b/vnt/src/handle/tun_tap/tap_handler.rs @@ -18,6 +18,7 @@ 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}; @@ -29,6 +30,7 @@ pub fn start( igmp_server: Option, current_device: Arc>, ip_route: Option, + #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, client_cipher: Cipher, server_cipher: Cipher, @@ -45,6 +47,7 @@ pub fn start( igmp_server, current_device, ip_route, + #[cfg(feature = "ip_proxy")] ip_proxy_map, client_cipher, server_cipher, @@ -64,6 +67,7 @@ pub fn start( 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(); @@ -77,6 +81,7 @@ pub fn start( &device_writer, &sender, &ip_route, + #[cfg(feature = "ip_proxy")] &ip_proxy_map, &client_cipher, &server_cipher, @@ -133,6 +138,7 @@ fn start_simple( igmp_server: Option, current_device: Arc>, ip_route: Option, + #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, client_cipher: Cipher, server_cipher: Cipher, @@ -151,6 +157,7 @@ fn start_simple( device_writer, sender, &ip_route, + #[cfg(feature = "ip_proxy")] &ip_proxy_map, &client_cipher, &server_cipher, @@ -168,6 +175,7 @@ fn handle( device_writer: &DeviceWriter, sender: &ChannelSender, ip_route: &Option, + #[cfg(feature = "ip_proxy")] proxy_map: &Option, client_cipher: &Cipher, server_cipher: &Cipher, @@ -213,9 +221,6 @@ fn handle( ethernet::protocol::Protocol::Ipv4 => { let mut ipv4_packet = IpV4Packet::unchecked(ethernet_packet.payload_mut()); let src_ip = ipv4_packet.source_ip(); - if src_ip != current_device.virtual_ip() { - return Ok(()); - } let dest_ip = ipv4_packet.destination_ip(); let protocol = ipv4_packet.protocol(); if src_ip == dest_ip { @@ -244,6 +249,7 @@ fn handle( igmp_server, current_device, ip_route, + #[cfg(feature = "ip_proxy")] proxy_map, client_cipher, server_cipher, diff --git a/vnt/src/handle/tun_tap/tun_handler.rs b/vnt/src/handle/tun_tap/tun_handler.rs index 60681a5..fd6dd8d 100644 --- a/vnt/src/handle/tun_tap/tun_handler.rs +++ b/vnt/src/handle/tun_tap/tun_handler.rs @@ -16,6 +16,7 @@ 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}; fn icmp(device_writer: &DeviceWriter, mut ipv4_packet: IpV4Packet<&mut [u8]>) -> Result<()> { @@ -44,20 +45,14 @@ fn handle( igmp_server: &Option, current_device: CurrentDeviceInfo, ip_route: &Option, + #[cfg(feature = "ip_proxy")] proxy_map: &Option, client_cipher: &Cipher, server_cipher: &Cipher, ) -> Result<()> { - let ipv4_packet = if let Ok(ipv4_packet) = IpV4Packet::new(&mut data[12..len]) { - ipv4_packet - } else { - return Ok(()); - }; + let ipv4_packet = IpV4Packet::new(&mut data[12..len])?; let src_ip = ipv4_packet.source_ip(); let dest_ip = ipv4_packet.destination_ip(); - if src_ip != current_device.virtual_ip() { - return Ok(()); - } if src_ip == dest_ip { return icmp(&device_writer, ipv4_packet); } @@ -68,6 +63,7 @@ fn handle( igmp_server, current_device, ip_route, + #[cfg(feature = "ip_proxy")] proxy_map, client_cipher, server_cipher, @@ -82,6 +78,7 @@ pub fn start( igmp_server: Option, current_device: Arc>, ip_route: Option, + #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, client_cipher: Cipher, server_cipher: Cipher, @@ -98,6 +95,7 @@ pub fn start( igmp_server, current_device, ip_route, + #[cfg(feature = "ip_proxy")] ip_proxy_map, client_cipher, server_cipher, @@ -117,6 +115,7 @@ pub fn start( 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(); @@ -130,6 +129,7 @@ pub fn start( &igmp_server, current_device.load(), &ip_route, + #[cfg(feature = "ip_proxy")] &ip_proxy_map, &client_cipher, &server_cipher, @@ -190,6 +190,7 @@ fn start_simple( igmp_server: Option, current_device: Arc>, ip_route: Option, + #[cfg(feature = "ip_proxy")] ip_proxy_map: Option, client_cipher: Cipher, server_cipher: Cipher, @@ -211,6 +212,7 @@ fn start_simple( &igmp_server, current_device.load(), &ip_route, + #[cfg(feature = "ip_proxy")] &ip_proxy_map, &client_cipher, &server_cipher, diff --git a/vnt/src/igmp_server/mod.rs b/vnt/src/igmp_server/mod.rs index 16ec482..ef4dd6d 100644 --- a/vnt/src/igmp_server/mod.rs +++ b/vnt/src/igmp_server/mod.rs @@ -1,9 +1,10 @@ use std::collections::{HashMap, HashSet}; use std::net::Ipv4Addr; use std::sync::Arc; +use std::sync::atomic::Ordering; use std::time::{Duration, Instant}; -use dashmap::DashMap; +use crossbeam_epoch::{Atomic, Owned}; use parking_lot::RwLock; use packet::igmp::igmp_v2::IgmpV2Packet; @@ -11,7 +12,6 @@ use packet::igmp::igmp_v3::{IgmpV3QueryPacket, IgmpV3RecordType, IgmpV3ReportPac use packet::igmp::IgmpType; use packet::ip::ipv4::protocol::Protocol; -use crate::ip_proxy::DashMapNew; use crate::tun_tap_device::DeviceWriter; //1. 定时发送query，启动时20秒一次，连发3次，之后8分钟一次 @@ -51,12 +51,12 @@ impl Multicast { #[derive(Clone)] pub struct IgmpServer { - multicast: Arc>>>, + multicast: Arc>>>>, } impl IgmpServer { pub fn new(device_writer: DeviceWriter) -> Self { - let multicast: Arc>>> = Arc::new(DashMap::new0()); + let multicast: Arc>>>> = Arc::new(Atomic::new(HashMap::with_capacity(16))); std::thread::spawn(move || { //预留以太网帧头和ip头 let mut buf = [0; 14 + 24 + 12]; @@ -97,16 +97,20 @@ impl IgmpServer { Self { multicast } } pub fn load(&self, multicast_addr: &Ipv4Addr) -> Option>> { - if let Some(entry) = self.multicast.get(multicast_addr) { - Some(entry.value().clone()) + let guard = &crossbeam_epoch::pin(); + let multicast = unsafe{self.multicast.load(Ordering::Relaxed, guard).deref()}; + if let Some(entry) = multicast.get(multicast_addr) { + Some(entry.clone()) } else { None } } pub fn handle(&self, buf: &[u8], source: Ipv4Addr) -> crate::Result<()> { - for x in self.multicast.iter() { + let guard = &crossbeam_epoch::pin(); + let multicast = unsafe{self.multicast.load(Ordering::Relaxed, guard).deref()}; + for (_,v) in multicast.iter() { let mut list = Vec::new(); - let mut write_guard = x.value().write(); + let mut write_guard = v.write(); for (ip, time) in &write_guard.members { if time.elapsed() > Duration::from_secs(30) { list.push(*ip); @@ -126,13 +130,7 @@ impl IgmpServer { if !multicast_addr.is_multicast() { return Ok(()); } - let multi = { - self.multicast - .entry(multicast_addr) - .or_insert_with(|| Arc::new(RwLock::new(Multicast::new()))) - .value() - .clone() - }; + let multi = self.add_multicast(multicast_addr); let mut guard = multi.write(); guard.members.insert(source, Instant::now()); } @@ -143,8 +141,8 @@ impl IgmpServer { if !multicast_addr.is_multicast() { return Ok(()); } - if let Some(entry) = self.multicast.get(&multicast_addr) { - let mut guard = entry.value().write(); + if let Some(entry) = self.get_multicast(&multicast_addr) { + let mut guard = entry.write(); guard.map.remove(&source); guard.members.remove(&source); } @@ -157,12 +155,7 @@ impl IgmpServer { if !multicast_addr.is_multicast() { return Ok(()); } - let multi = self - .multicast - .entry(multicast_addr) - .or_insert_with(|| Arc::new(RwLock::new(Multicast::new()))) - .value() - .clone(); + let multi = self.add_multicast(multicast_addr); let mut guard = multi.write(); match group_record.record_type() { @@ -240,4 +233,37 @@ impl IgmpServer { } Ok(()) } + fn get_multicast(&self,multicast_addr:&Ipv4Addr)->Option>>{ + let guard = &crossbeam_epoch::pin(); + let multicast = &self.multicast; + let table_share = multicast.load(Ordering::Acquire, guard); + unsafe { + table_share.deref().get(multicast_addr).map(|v|v.clone()) + } + } + fn add_multicast(&self,multicast_addr:Ipv4Addr)->Arc>{ + let guard = &crossbeam_epoch::pin(); + let multicast = &self.multicast; + let mut table_share = multicast.load(Ordering::Acquire, guard); + let value = Arc::new(RwLock::new(Multicast::new())); + loop { + let mut table = unsafe { table_share.deref().clone() }; + table.insert(multicast_addr,value.clone()); + match self.multicast.compare_exchange( + table_share, + Owned::new(table.clone()), + Ordering::AcqRel, + Ordering::Relaxed, + guard, + ) { + Ok(_p) => unsafe { + guard.defer_destroy(table_share); + return value; + }, + Err(e) => { + table_share = e.current; + } + } + } + } } diff --git a/vnt/src/ip_proxy/icmp_proxy.rs b/vnt/src/ip_proxy/icmp_proxy.rs index b8a5774..769f255 100644 --- a/vnt/src/ip_proxy/icmp_proxy.rs +++ b/vnt/src/ip_proxy/icmp_proxy.rs @@ -1,20 +1,20 @@ -use crossbeam_utils::atomic::AtomicCell; -use dashmap::DashMap; use std::io; use std::mem::MaybeUninit; use std::net::{IpAddr, Ipv4Addr, SocketAddrV4}; use std::sync::Arc; +use crossbeam_utils::atomic::AtomicCell; +use dashmap::DashMap; use socket2::{Domain, SockAddr, Socket, Type}; +use packet::icmp::icmp; +use packet::icmp::icmp::HeaderOther; +use packet::ip::ipv4::packet::IpV4Packet; + use crate::channel::sender::ChannelSender; use crate::cipher::Cipher; use crate::handle::CurrentDeviceInfo; -use crate::protocol::body::ENCRYPTION_RESERVED; -use crate::protocol::{NetPacket, Protocol, Version, MAX_TTL}; -use packet::icmp::icmp; -use packet::icmp::icmp::HeaderOther; -use packet::ip::ipv4; +use crate::ip_proxy::{send, ProxyHandler}; pub struct IcmpProxy { icmp_socket: Arc, @@ -47,19 +47,19 @@ impl IcmpProxy { client_cipher, }) } - pub fn icmp_socket(&self) -> Arc { - self.icmp_socket.clone() + pub fn icmp_handler(&self) -> IcmpHandler { + IcmpHandler(self.icmp_socket.clone(), self.icmp_proxy_map.clone()) } pub fn start(self) { - let mut buf = [0u8; 1500]; - let data: &mut [MaybeUninit] = unsafe { std::mem::transmute(&mut buf[..]) }; + let mut buf = [0u8; 4096]; + let data: &mut [MaybeUninit] = unsafe { std::mem::transmute(&mut buf[12..]) }; loop { match self.recv(data) { Ok((len, peer_ip)) => { match peer_ip { IpAddr::V4(peer_ip) => { - match ipv4::packet::IpV4Packet::new(&mut buf[..len]) { + match IpV4Packet::new(&mut buf[12..12 + len]) { Ok(mut ipv4_packet) => { match icmp::IcmpPacket::new(ipv4_packet.payload()) { Ok(icmp_packet) => { @@ -73,47 +73,14 @@ impl IcmpProxy { drop(entry); ipv4_packet.set_destination_ip(dest_ip); ipv4_packet.update_checksum(); - let current_device = - self.current_device.load(); - let virtual_ip = - current_device.virtual_ip(); - let connect_server = - current_device.connect_server; - let mut net_packet = - NetPacket::new_encrypt(vec![ - 0u8; - 12 + len + ENCRYPTION_RESERVED - ]) - .unwrap(); - net_packet.set_version(Version::V1); - net_packet.set_protocol(Protocol::IpTurn); - net_packet.set_transport_protocol(crate::protocol::ip_turn_packet::Protocol::Ipv4.into()); - net_packet.first_set_ttl(MAX_TTL); - net_packet.set_source(virtual_ip); - net_packet.set_destination(dest_ip); - net_packet - .set_payload(ipv4_packet.buffer) - .unwrap(); - if let Err(e) = self - .client_cipher - .encrypt_ipv4(&mut net_packet) - { - log::warn!("加密失败:{}", e); - continue; - } - if self - .sender - .try_send_by_id( - net_packet.buffer(), - &dest_ip, - ) - .is_err() - { - let _ = self.sender.send_main( - net_packet.buffer(), - connect_server, - ); - } + send( + &mut buf, + len, + dest_ip, + &self.sender, + &self.current_device, + &self.client_cipher, + ); } } _ => { @@ -144,7 +111,42 @@ impl IcmpProxy { }; Ok((size, addr)) } - // fn send_to(&self, buf: &[u8], addr: SocketAddrV4) -> io::Result { - // self.icmp_socket.send_to(buf, &SockAddr::from(addr)) - // } +} +/// icmp用Identifier来区分，没有Identifier的一律不转发 +#[derive(Clone)] +pub struct IcmpHandler(Arc, Arc>); + +impl ProxyHandler for IcmpHandler { + fn recv_handle( + &self, + ipv4: &mut IpV4Packet<&mut [u8]>, + source: Ipv4Addr, + destination: Ipv4Addr, + ) -> io::Result { + let dest_ip = ipv4.destination_ip(); + //转发到代理目标地址 + let icmp_packet = icmp::IcmpPacket::new(ipv4.payload())?; + match icmp_packet.header_other() { + HeaderOther::Identifier(id, seq) => { + self.1.insert((dest_ip, id, seq), source); + self.0.send_to( + ipv4.payload(), + &SockAddr::from(SocketAddrV4::new(dest_ip, 0)), + )?; + } + _ => { + log::warn!( + "不支持的ip代理Icmp协议:{}->{}->{}", + source, + destination, + dest_ip + ); + } + } + Ok(true) + } + + fn send_handle(&self, _ipv4: &mut IpV4Packet<&mut [u8]>) -> io::Result<()> { + Ok(()) + } } diff --git a/vnt/src/ip_proxy/mod.rs b/vnt/src/ip_proxy/mod.rs index 90027c0..1e5ac7b 100644 --- a/vnt/src/ip_proxy/mod.rs +++ b/vnt/src/ip_proxy/mod.rs @@ -1,14 +1,23 @@ -use crate::ip_proxy::tcp_proxy::TcpProxy; -use crate::ip_proxy::udp_proxy::UdpProxy; -use dashmap::DashMap; -#[cfg(not(target_os = "android"))] -use socket2::{SockAddr, Socket}; #[cfg(not(target_os = "android"))] use std::net::Ipv4Addr; use std::net::SocketAddrV4; use std::sync::Arc; use std::{io, thread}; -use tokio::net::{TcpListener, UdpSocket}; + +use crossbeam_utils::atomic::AtomicCell; +use dashmap::DashMap; +use tokio::net::UdpSocket; + +use packet::ip::ipv4::packet::IpV4Packet; + +use crate::channel::sender::ChannelSender; +use crate::cipher::Cipher; +use crate::handle::CurrentDeviceInfo; +use crate::ip_proxy::icmp_proxy::IcmpHandler; +use crate::ip_proxy::tcp_proxy::{TcpHandler, TcpProxy}; +use crate::ip_proxy::udp_proxy::{UdpHandler, UdpProxy}; +use crate::protocol; +use crate::protocol::{NetPacket, Version, MAX_TTL}; #[cfg(not(target_os = "android"))] pub mod icmp_proxy; @@ -20,6 +29,16 @@ pub trait DashMapNew { fn new_cap(capacity: usize) -> Self; } +pub trait ProxyHandler { + fn recv_handle( + &self, + ipv4: &mut IpV4Packet<&mut [u8]>, + source: Ipv4Addr, + destination: Ipv4Addr, + ) -> io::Result; + fn send_handle(&self, ipv4: &mut IpV4Packet<&mut [u8]>) -> io::Result<()>; +} + impl<'a, K: 'a + Eq + std::hash::Hash, V: 'a> DashMapNew for DashMap { fn new0() -> Self { Self::new_cap(0) @@ -52,72 +71,105 @@ pub enum Protocol { #[derive(Clone)] pub struct IpProxyMap { - pub(crate) tcp_proxy_port: u16, - pub(crate) udp_proxy_port: u16, - //真实源地址 -> 目的地址 - pub(crate) tcp_proxy_map: Arc>, - pub(crate) udp_proxy_map: Arc>, - // icmp用Identifier来区分，没有Identifier的一律不转发 #[cfg(not(target_os = "android"))] - pub(crate) icmp_proxy_map: Arc>, - #[cfg(not(target_os = "android"))] - icmp_socket: Arc, -} - -impl IpProxyMap { - #[cfg(not(target_os = "android"))] - pub fn send_icmp(&self, buf: &[u8], dest: &Ipv4Addr) -> io::Result { - self.icmp_socket - .send_to(buf, &SockAddr::from(SocketAddrV4::new(*dest, 0))) - } + pub(crate) icmp_handler: IcmpHandler, + pub(crate) tcp_handler: TcpHandler, + pub(crate) udp_handler: UdpHandler, } +#[cfg(not(target_os = "android"))] pub async fn init_proxy( - #[cfg(not(target_os = "android"))] sender: crate::channel::sender::ChannelSender, - #[cfg(not(target_os = "android"))] current_device: Arc< - crossbeam_utils::atomic::AtomicCell, - >, - #[cfg(not(target_os = "android"))] client_cipher: crate::cipher::Cipher, + sender: ChannelSender, + current_device: Arc>, + client_cipher: Cipher, ) -> io::Result<(TcpProxy, UdpProxy, IpProxyMap)> { let tcp_proxy_map: Arc> = Arc::new(DashMap::new0()); let udp_proxy_map: Arc> = Arc::new(DashMap::new0()); - #[cfg(not(target_os = "android"))] + let icmp_proxy_map: Arc> = Arc::new(DashMap::new0()); - 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 = TcpProxy::new(tcp_listener, tcp_proxy_map.clone()); - let udp_proxy = UdpProxy::new(udp_socket, udp_proxy_map.clone()); - #[cfg(not(target_os = "android"))] - let icmp_socket = { - let addr = SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0); + let addr = SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0); + let tcp_proxy = TcpProxy::new(addr, tcp_proxy_map.clone()).await?; + let tcp_handler = tcp_proxy.tcp_handler(); + let udp_proxy = UdpProxy::new(udp_socket, udp_proxy_map.clone())?; + let udp_handler = udp_proxy.udp_handler(); + + let icmp_handler = { let icmp_proxy = icmp_proxy::IcmpProxy::new( addr, icmp_proxy_map.clone(), sender.clone(), current_device.clone(), - client_cipher, + client_cipher.clone(), )?; - let icmp_socket = icmp_proxy.icmp_socket(); + let icmp_handler = icmp_proxy.icmp_handler(); thread::spawn(move || { icmp_proxy.start(); }); - icmp_socket + icmp_handler }; Ok(( tcp_proxy, udp_proxy, IpProxyMap { - tcp_proxy_port, - udp_proxy_port, - tcp_proxy_map, - udp_proxy_map, - #[cfg(not(target_os = "android"))] - icmp_proxy_map, - #[cfg(not(target_os = "android"))] - icmp_socket, + tcp_handler, + udp_handler, + icmp_handler, }, )) } + +#[cfg(target_os = "android")] +pub async fn init_proxy() -> io::Result<(TcpProxy, UdpProxy, IpProxyMap)> { + let tcp_proxy_map: Arc> = Arc::new(DashMap::new0()); + let udp_proxy_map: Arc> = Arc::new(DashMap::new0()); + let addr = SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0); + let udp_socket = UdpSocket::bind("0.0.0.0:0").await?; + let tcp_proxy = TcpProxy::new(addr, tcp_proxy_map.clone()).await?; + let tcp_handler = tcp_proxy.tcp_handler(); + let udp_proxy = UdpProxy::new(udp_socket, udp_proxy_map.clone())?; + let udp_handler = udp_proxy.udp_handler(); + + Ok(( + tcp_proxy, + udp_proxy, + IpProxyMap { + tcp_handler, + udp_handler, + }, + )) +} + +pub fn send( + buf: &mut [u8], + data_len: usize, + dest_ip: Ipv4Addr, + sender: &ChannelSender, + current_device: &AtomicCell, + client_cipher: &Cipher, +) { + let current_device = current_device.load(); + let virtual_ip = current_device.virtual_ip(); + + let mut net_packet = NetPacket::new0(12 + data_len, buf).unwrap(); + net_packet.set_version(Version::V1); + net_packet.set_protocol(protocol::Protocol::IpTurn); + net_packet.set_transport_protocol(protocol::ip_turn_packet::Protocol::Ipv4.into()); + net_packet.first_set_ttl(MAX_TTL); + net_packet.set_source(virtual_ip); + net_packet.set_destination(dest_ip); + if let Err(e) = client_cipher.encrypt_ipv4(&mut net_packet) { + log::warn!("加密失败:{}", e); + return; + } + if sender + .try_send_by_id(net_packet.buffer(), &dest_ip) + .is_err() + { + let connect_server = current_device.connect_server; + if let Err(e) = sender.send_main(net_packet.buffer(), connect_server) { + log::warn!("发送到目标失败:{},{}", e, connect_server); + } + } +} diff --git a/vnt/src/ip_proxy/tcp_proxy.rs b/vnt/src/ip_proxy/tcp_proxy.rs index a45e83f..1d73bce 100644 --- a/vnt/src/ip_proxy/tcp_proxy.rs +++ b/vnt/src/ip_proxy/tcp_proxy.rs @@ -1,7 +1,10 @@ +use crate::ip_proxy::ProxyHandler; use crossbeam_utils::atomic::AtomicCell; use dashmap::DashMap; +use packet::ip::ipv4::packet::IpV4Packet; +use packet::tcp::tcp::TcpPacket; use std::io; -use std::net::{SocketAddr, SocketAddrV4}; +use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4}; use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::io::AsyncReadExt; @@ -10,19 +13,26 @@ use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; use tokio::net::{TcpListener, TcpStream}; pub struct TcpProxy { + tcp_proxy_port: u16, tcp_listener: TcpListener, + //真实源地址 -> 目的地址 tcp_proxy_map: Arc>, } impl TcpProxy { - pub fn new( - tcp_listener: TcpListener, + pub async fn new( + addr: SocketAddrV4, tcp_proxy_map: Arc>, - ) -> Self { - Self { + ) -> io::Result { + let tcp_listener = TcpListener::bind(addr).await?; + Ok(Self { + tcp_proxy_port: tcp_listener.local_addr()?.port(), tcp_listener, tcp_proxy_map, - } + }) + } + pub fn tcp_handler(&self) -> TcpHandler { + TcpHandler(self.tcp_proxy_port, self.tcp_proxy_map.clone()) } pub async fn start(self) { let tcp_listener = self.tcp_listener; @@ -121,3 +131,48 @@ async fn copy( } Ok(()) } + +#[derive(Clone)] +pub struct TcpHandler(u16, Arc>); + +impl ProxyHandler for TcpHandler { + fn recv_handle( + &self, + ipv4: &mut IpV4Packet<&mut [u8]>, + source: Ipv4Addr, + destination: Ipv4Addr, + ) -> io::Result { + let dest_ip = ipv4.destination_ip(); + //转发到代理目标地址 + let mut tcp_packet = 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.0); + tcp_packet.update_checksum(); + ipv4.set_destination_ip(destination); + ipv4.update_checksum(); + let key = SocketAddrV4::new(source, source_port); + //https://github.com/crossbeam-rs/crossbeam/issues/1023 + self.1.insert(key, SocketAddrV4::new(dest_ip, dest_port)); + Ok(false) + } + + fn send_handle(&self, ipv4: &mut IpV4Packet<&mut [u8]>) -> io::Result<()> { + let src_ip = ipv4.source_ip(); + let dest_ip = ipv4.destination_ip(); + let dest_addr = { + let tcp_packet = TcpPacket::new(src_ip, dest_ip, ipv4.payload_mut())?; + SocketAddrV4::new(dest_ip, tcp_packet.destination_port()) + }; + if let Some(entry) = self.1.get(&dest_addr) { + let source_addr = entry.value(); + let source_ip = *source_addr.ip(); + let mut tcp_packet = TcpPacket::new(source_ip, dest_ip, ipv4.payload_mut())?; + tcp_packet.set_source_port(source_addr.port()); + tcp_packet.update_checksum(); + ipv4.set_source_ip(source_ip); + ipv4.update_checksum(); + } + Ok(()) + } +} diff --git a/vnt/src/ip_proxy/udp_proxy.rs b/vnt/src/ip_proxy/udp_proxy.rs index 62463fd..89c3028 100644 --- a/vnt/src/ip_proxy/udp_proxy.rs +++ b/vnt/src/ip_proxy/udp_proxy.rs @@ -1,8 +1,10 @@ -use crate::ip_proxy::DashMapNew; +use crate::ip_proxy::{DashMapNew, ProxyHandler}; use crossbeam_utils::atomic::AtomicCell; use dashmap::DashMap; +use packet::ip::ipv4::packet::IpV4Packet; +use packet::udp::udp::UdpPacket; use std::io; -use std::net::{SocketAddr, SocketAddrV4}; +use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4}; use std::sync::Arc; use std::time::Duration; use tokio::net::UdpSocket; @@ -10,14 +12,26 @@ use tokio::time::Instant; /// 一个udp代理，作用是利用系统协议栈，将udp数据报解析出来再转发到目的地址 pub struct UdpProxy { + udp_proxy_port: u16, udp_socket: Arc, map: Arc>, } impl UdpProxy { - pub fn new(udp_socket: UdpSocket, map: Arc>) -> Self { + pub fn new( + udp_socket: UdpSocket, + map: Arc>, + ) -> io::Result { let udp_socket = Arc::new(udp_socket); - Self { udp_socket, map } + let udp_proxy_port = udp_socket.local_addr()?.port(); + Ok(Self { + udp_proxy_port, + udp_socket, + map, + }) + } + pub fn udp_handler(&self) -> UdpHandler { + UdpHandler(self.udp_proxy_port, self.map.clone()) } pub async fn start(self) { let map = self.map; @@ -122,3 +136,47 @@ async fn start0( } Ok(()) } + +#[derive(Clone)] +pub struct UdpHandler(u16, Arc>); + +impl ProxyHandler for UdpHandler { + fn recv_handle( + &self, + ipv4: &mut IpV4Packet<&mut [u8]>, + source: Ipv4Addr, + destination: Ipv4Addr, + ) -> io::Result { + let dest_ip = ipv4.destination_ip(); + //转发到代理目标地址 + let mut udp_packet = 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.0); + udp_packet.update_checksum(); + ipv4.set_destination_ip(destination); + ipv4.update_checksum(); + let key = SocketAddrV4::new(source, source_port); + self.1.insert(key, SocketAddrV4::new(dest_ip, dest_port)); + Ok(false) + } + + fn send_handle(&self, ipv4: &mut IpV4Packet<&mut [u8]>) -> io::Result<()> { + let src_ip = ipv4.source_ip(); + let dest_ip = ipv4.destination_ip(); + let dest_addr = { + let udp_packet = UdpPacket::new(src_ip, dest_ip, ipv4.payload_mut())?; + SocketAddrV4::new(dest_ip, udp_packet.destination_port()) + }; + if let Some(entry) = self.1.get(&dest_addr) { + let source_addr = entry.value(); + let source_ip = *source_addr.ip(); + let mut udp_packet = UdpPacket::new(source_ip, dest_ip, ipv4.payload_mut())?; + udp_packet.set_source_port(source_addr.port()); + udp_packet.update_checksum(); + ipv4.set_source_ip(source_ip); + ipv4.update_checksum(); + } + Ok(()) + } +} diff --git a/vnt/src/lib.rs b/vnt/src/lib.rs index f3821ea..94f81d0 100644 --- a/vnt/src/lib.rs +++ b/vnt/src/lib.rs @@ -1,5 +1,5 @@ use crate::error::Error; -pub const VNT_VERSION: &'static str = "1.2.5"; +pub const VNT_VERSION: &'static str = "1.2.5.1"; pub type Result = std::result::Result; pub mod channel; @@ -9,6 +9,7 @@ pub mod error; pub mod external_route; pub mod handle; pub mod igmp_server; +#[cfg(feature = "ip_proxy")] pub mod ip_proxy; pub mod nat; pub mod proto;