yeyinlaoying/vnt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

可选ip代理，关闭后可使用外部命令来进行ip转发

Browse Source
This commit is contained in:
lubeilin
2023-10-08 17:16:02 +08:00
parent a9495f1d30
commit dacce892ff
17 changed files with 538 additions and 375 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vnt-cli/Cargo.toml
+2 -2
View File
@@ -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"
vnt/Cargo.toml
+2 -2
View File
@@ -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"]
vnt/packet/src/tcp/tcp.rs
+8 -1
View File
@@ -76,7 +76,14 @@ impl<B: AsRef<[u8]>> TcpPacket<B> {
Ok(packet)
}
}
impl<B: AsRef<[u8]> + AsMut<[u8]>> TcpPacket<B> {
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<B: AsRef<[u8]> + AsMut<[u8]>> TcpPacket<B> {
fn set_checksum(&mut self, value: u16) {
self.buffer.as_mut()[16..18].copy_from_slice(&value.to_be_bytes())
vnt/src/channel/channel.rs
+54 -63
View File
@@ -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<Arc<StdUdpSocket>>,
//在udp的基础上，可以选择使用tcp和服务端通信
pub(crate) main_tcp_channel: Option<std::sync::mpsc::SyncSender<Vec<u8>>>,
pub(crate) route_table: Atomic<HashMap<Ipv4Addr, Vec<Route>>>,
pub(crate) route_table_time: DashMap<(RouteKey, Ipv4Addr), Instant>,
pub(crate) route_table: Atomic<HashMap<Ipv4Addr, Vec<(Route, Arc<AtomicCell<Instant>>)>>>,
pub(crate) status_receiver: Receiver<Status>,
pub(crate) status_sender: Sender<Status>,
pub(crate) udp_map: Atomic<HashMap<usize, Arc<UdpSocket>>>,
@@ -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<Arc<UdpSocket>>) {
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<Vec<Route>> {
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;
}
}
}
}
}
vnt/src/channel/idle.rs
+20 -7
View File
@@ -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;
}
}
}
}
}
vnt/src/core/mod.rs
+28 -16
View File
@@ -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<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
nat_test: NatTest,
connect_status: Arc<AtomicCell<ConnectStatus>>,
peer_nat_info_map: Arc<DashMap<Ipv4Addr, NatInfo>>,
peer_nat_info_map: Arc<Atomic<HashMap<Ipv4Addr, NatInfo>>>,
}
pub struct VntUtil {
@@ -277,7 +278,7 @@ impl VntUtil {
));
let device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>> =
Arc::new(Mutex::new((response.epoch, response.device_info_list)));
let peer_nat_info_map: Arc<DashMap<Ipv4Addr, NatInfo>> = Arc::new(DashMap::new0());
let peer_nat_info_map: Arc<Atomic<HashMap<Ipv4Addr, NatInfo>>> = 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<NatInfo> {
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()
vnt/src/external_route/mod.rs
+3
View File
@@ -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 {
vnt/src/handle/recv_handler.rs
+50 -73
View File
@@ -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<IgmpServer>,
device_writer: DeviceWriter,
connect_status: Arc<AtomicCell<ConnectStatus>>,
peer_nat_info_map: Arc<DashMap<Ipv4Addr, NatInfo>>,
peer_nat_info_map: Arc<Atomic<HashMap<Ipv4Addr, NatInfo>>>,
#[cfg(feature = "ip_proxy")]
ip_proxy_map: Option<IpProxyMap>,
out_external_route: AllowExternalRoute,
cone_sender: Sender<(Ipv4Addr, NatInfo)>,
@@ -66,7 +70,8 @@ impl ChannelDataHandler {
igmp_server: Option<IgmpServer>,
device_writer: DeviceWriter,
connect_status: Arc<AtomicCell<ConnectStatus>>,
peer_nat_info_map: Arc<DashMap<Ipv4Addr, NatInfo>>,
peer_nat_info_map: Arc<Atomic<HashMap<Ipv4Addr, NatInfo>>>,
#[cfg(feature = "ip_proxy")]
ip_proxy_map: Option<IpProxyMap>,
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;
vnt/src/handle/tun_tap/mod.rs
+20 -62
View File
@@ -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<IgmpServer>,
current_device: CurrentDeviceInfo,
ip_route: &Option<ExternalRoute>,
#[cfg(feature = "ip_proxy")]
proxy_map: &Option<IpProxyMap>,
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)?;
}
_ => {}
}
vnt/src/handle/tun_tap/tap_handler.rs
+9 -3
View File
@@ -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<IgmpServer>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
ip_route: Option<ExternalRoute>,
#[cfg(feature = "ip_proxy")]
ip_proxy_map: Option<IpProxyMap>,
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<IgmpServer>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
ip_route: Option<ExternalRoute>,
#[cfg(feature = "ip_proxy")]
ip_proxy_map: Option<IpProxyMap>,
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<ExternalRoute>,
#[cfg(feature = "ip_proxy")]
proxy_map: &Option<IpProxyMap>,
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,
vnt/src/handle/tun_tap/tun_handler.rs
+10 -8
View File
@@ -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<IgmpServer>,
current_device: CurrentDeviceInfo,
ip_route: &Option<ExternalRoute>,
#[cfg(feature = "ip_proxy")]
proxy_map: &Option<IpProxyMap>,
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<IgmpServer>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
ip_route: Option<ExternalRoute>,
#[cfg(feature = "ip_proxy")]
ip_proxy_map: Option<IpProxyMap>,
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<IgmpServer>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
ip_route: Option<ExternalRoute>,
#[cfg(feature = "ip_proxy")]
ip_proxy_map: Option<IpProxyMap>,
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,
vnt/src/igmp_server/mod.rs
+49 -23
View File
@@ -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<DashMap<Ipv4Addr, Arc<RwLock<Multicast>>>>,
multicast: Arc<Atomic<HashMap<Ipv4Addr, Arc<RwLock<Multicast>>>>>,
}
impl IgmpServer {
pub fn new(device_writer: DeviceWriter) -> Self {
let multicast: Arc<DashMap<Ipv4Addr, Arc<RwLock<Multicast>>>> = Arc::new(DashMap::new0());
let multicast: Arc<Atomic<HashMap<Ipv4Addr, Arc<RwLock<Multicast>>>>> = 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<Arc<RwLock<Multicast>>> {
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<Arc<RwLock<Multicast>>>{
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<RwLock<Multicast>>{
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;
}
}
}
}
}
vnt/src/ip_proxy/icmp_proxy.rs
+58 -56
View File
@@ -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<Socket>,
@@ -47,19 +47,19 @@ impl IcmpProxy {
client_cipher,
})
}
pub fn icmp_socket(&self) -> Arc<Socket> {
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<u8>] = unsafe { std::mem::transmute(&mut buf[..]) };
let mut buf = [0u8; 4096];
let data: &mut [MaybeUninit<u8>] = 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<usize> {
// self.icmp_socket.send_to(buf, &SockAddr::from(addr))
// }
}
/// icmp用Identifier来区分，没有Identifier的一律不转发
#[derive(Clone)]
pub struct IcmpHandler(Arc<Socket>, Arc<DashMap<(Ipv4Addr, u16, u16), Ipv4Addr>>);
impl ProxyHandler for IcmpHandler {
fn recv_handle(
&self,
ipv4: &mut IpV4Packet<&mut [u8]>,
source: Ipv4Addr,
destination: Ipv4Addr,
) -> io::Result<bool> {
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(())
}
}
vnt/src/ip_proxy/mod.rs
+100 -48
View File
@@ -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<bool>;
fn send_handle(&self, ipv4: &mut IpV4Packet<&mut [u8]>) -> io::Result<()>;
}
impl<'a, K: 'a + Eq + std::hash::Hash, V: 'a> DashMapNew for DashMap<K, V> {
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<DashMap<SocketAddrV4, SocketAddrV4>>,
pub(crate) udp_proxy_map: Arc<DashMap<SocketAddrV4, SocketAddrV4>>,
// icmp用Identifier来区分，没有Identifier的一律不转发
#[cfg(not(target_os = "android"))]
pub(crate) icmp_proxy_map: Arc<DashMap<(Ipv4Addr, u16, u16), Ipv4Addr>>,
#[cfg(not(target_os = "android"))]
icmp_socket: Arc<Socket>,
}
impl IpProxyMap {
#[cfg(not(target_os = "android"))]
pub fn send_icmp(&self, buf: &[u8], dest: &Ipv4Addr) -> io::Result<usize> {
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<crate::handle::CurrentDeviceInfo>,
>,
#[cfg(not(target_os = "android"))] client_cipher: crate::cipher::Cipher,
sender: ChannelSender,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
client_cipher: Cipher,
) -> io::Result<(TcpProxy, UdpProxy, IpProxyMap)> {
let tcp_proxy_map: Arc<DashMap<SocketAddrV4, SocketAddrV4>> = Arc::new(DashMap::new0());
let udp_proxy_map: Arc<DashMap<SocketAddrV4, SocketAddrV4>> = Arc::new(DashMap::new0());
#[cfg(not(target_os = "android"))]
let icmp_proxy_map: Arc<DashMap<(Ipv4Addr, u16, u16), Ipv4Addr>> = 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<DashMap<SocketAddrV4, SocketAddrV4>> = Arc::new(DashMap::new0());
let udp_proxy_map: Arc<DashMap<SocketAddrV4, SocketAddrV4>> = 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<CurrentDeviceInfo>,
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);
}
}
}
vnt/src/ip_proxy/tcp_proxy.rs
+61 -6
View File
@@ -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<DashMap<SocketAddrV4, SocketAddrV4>>,
}
impl TcpProxy {
pub fn new(
tcp_listener: TcpListener,
pub async fn new(
addr: SocketAddrV4,
tcp_proxy_map: Arc<DashMap<SocketAddrV4, SocketAddrV4>>,
) -> Self {
Self {
) -> io::Result<Self> {
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<DashMap<SocketAddrV4, SocketAddrV4>>);
impl ProxyHandler for TcpHandler {
fn recv_handle(
&self,
ipv4: &mut IpV4Packet<&mut [u8]>,
source: Ipv4Addr,
destination: Ipv4Addr,
) -> io::Result<bool> {
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(())
}
}
vnt/src/ip_proxy/udp_proxy.rs
+62 -4
View File
@@ -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<UdpSocket>,
map: Arc<DashMap<SocketAddrV4, SocketAddrV4>>,
}
impl UdpProxy {
pub fn new(udp_socket: UdpSocket, map: Arc<DashMap<SocketAddrV4, SocketAddrV4>>) -> Self {
pub fn new(
udp_socket: UdpSocket,
map: Arc<DashMap<SocketAddrV4, SocketAddrV4>>,
) -> io::Result<Self> {
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<DashMap<SocketAddrV4, SocketAddrV4>>);
impl ProxyHandler for UdpHandler {
fn recv_handle(
&self,
ipv4: &mut IpV4Packet<&mut [u8]>,
source: Ipv4Addr,
destination: Ipv4Addr,
) -> io::Result<bool> {
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(())
}
}
vnt/src/lib.rs
+2 -1
View File
@@ -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<T> = std::result::Result<T, Error>;
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;