Compare commits

...
20 Commits
Author SHA1 Message Date
lubeilin 707b07b8d3 ipv6改为完整地址 2023-09-24 15:42:19 +08:00
lubeilin de5a6971f0 更新读取时间不需要再插入 2023-09-24 14:58:03 +08:00
lubeilin 056036c4d2 减少注册和探测nat的频率 2023-09-24 14:53:37 +08:00
lubeilin 1cfb188845 去除多余依赖 2023-09-24 13:45:47 +08:00
lubeilin 73a2c31854 连接通过关闭同时关闭tap 2023-09-24 13:42:09 +08:00
lubeilin 92eea536f8 删除多余依赖 2023-09-24 12:57:49 +08:00
lubeilin 00936a923e 避免直接关闭网卡 2023-09-24 12:55:42 +08:00
lubeilin ba69ba78af 增加线程名称 2023-09-23 23:09:37 +08:00
lubeilin 17b206bace 1.2.4.3 2023-09-23 21:38:31 +08:00
lubeilin 58d5a4f5da 增加wintun日志 2023-09-23 21:38:20 +08:00
lubeilin 2438d14175 避免短时间重复上传服务端密钥 2023-09-23 21:33:00 +08:00
lubeilin 99f8526799 去除tap广播路由 2023-09-22 22:49:05 +08:00
lubeilin 56fcbd64ed 增加日志 2023-09-22 22:43:23 +08:00
lubeilin 3766b2b7c1 修改命令超时时间 2023-09-22 22:43:02 +08:00
lubeilin baf0698fe4 去除广播路由 2023-09-22 22:17:13 +08:00
lubeilin 301938b9fc 增加小版本 2023-09-22 18:19:13 +08:00
lubeilin 16a37c713a 增加日志 2023-09-22 18:18:29 +08:00
lubeilin d412a769dd fmt 2023-09-22 18:18:10 +08:00
lubeilin c8eecc87fd 调整心跳间隔,服务端和客户端心跳分离 2023-09-22 18:17:12 +08:00
lubeilin 6a11db70c8 调整代理超时时间 2023-09-22 18:16:06 +08:00
20 changed files with 266 additions and 123 deletions
+1 -1
View File
@@ -26,7 +26,7 @@ impl CommandClient {
} }
}; };
let udp = UdpSocket::bind("127.0.0.1:0")?; let udp = UdpSocket::bind("127.0.0.1:0")?;
udp.set_read_timeout(Some(Duration::from_secs(2)))?; udp.set_read_timeout(Some(Duration::from_secs(5)))?;
udp.connect(SocketAddr::V4(SocketAddrV4::new( udp.connect(SocketAddr::V4(SocketAddrV4::new(
Ipv4Addr::new(127, 0, 0, 1), Ipv4Addr::new(127, 0, 0, 1),
port, port,
+1 -1
View File
@@ -17,7 +17,7 @@ pub enum CommandEnum {
pub fn command(cmd: CommandEnum) { pub fn command(cmd: CommandEnum) {
if let Err(e) = command_(cmd) { if let Err(e) = command_(cmd) {
println!("cmd: {}", e); println!("cmd: {:?}", e);
} }
} }
+7 -2
View File
@@ -17,15 +17,20 @@ impl CommandServer {
let udp = UdpSocket::bind("127.0.0.1:0").await?; let udp = UdpSocket::bind("127.0.0.1:0").await?;
let path_buf = crate::app_home()?.join("command-port"); let path_buf = crate::app_home()?.join("command-port");
let mut file = std::fs::File::create(path_buf)?; let mut file = std::fs::File::create(path_buf)?;
file.write_all(udp.local_addr()?.port().to_string().as_bytes())?; let addr = udp.local_addr()?;
file.write_all(addr.port().to_string().as_bytes())?;
file.sync_all()?; file.sync_all()?;
log::info!("启动后台cmd:{:?}", addr);
let mut buf = [0u8; 64]; let mut buf = [0u8; 64];
loop { loop {
let (len, addr) = udp.recv_from(&mut buf).await?; let (len, addr) = udp.recv_from(&mut buf).await?;
match std::str::from_utf8(&buf[..len]) { match std::str::from_utf8(&buf[..len]) {
Ok(cmd) => { Ok(cmd) => {
log::info!("收到cmd={:?}", cmd);
if let Ok(out) = command(cmd, &vnt) { if let Ok(out) = command(cmd, &vnt) {
let _ = udp.send_to(out.as_bytes(), addr).await; if let Err(e) = udp.send_to(out.as_bytes(), addr).await {
log::warn!("cmd={},err={:?}", cmd, e);
}
if "stopped" == &out { if "stopped" == &out {
break; break;
} }
+1
View File
@@ -362,6 +362,7 @@ async fn main0(config: Config, show_cmd: bool) {
let vnt_c = vnt.clone(); let vnt_c = vnt.clone();
tokio::spawn(async { tokio::spawn(async {
if let Err(e) = command::server::CommandServer::new().start(vnt_c).await { if let Err(e) = command::server::CommandServer::new().start(vnt_c).await {
log::warn!("cmd:{:?}", e);
println!("command error :{}", e); println!("command error :{}", e);
} }
}); });
+20 -16
View File
@@ -522,10 +522,6 @@ impl Context {
pub fn update_read_time(&self, id: &Ipv4Addr, route_key: &RouteKey) { 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)) { if let Some(mut time) = self.inner.route_table_time.get_mut(&(*route_key, *id)) {
*time.value_mut() = Instant::now(); *time.value_mut() = Instant::now();
} else {
self.inner
.route_table_time
.insert((*route_key, *id), Instant::now());
} }
} }
} }
@@ -595,8 +591,7 @@ impl Channel {
tcp_r tcp_r
.read_exact(&mut buf[head_reserve..head_reserve + len]) .read_exact(&mut buf[head_reserve..head_reserve + len])
.await?; .await?;
handler handler.handle(&mut buf, head_reserve, head_reserve + len, key, &context);
.handle(&mut buf, head_reserve, head_reserve + len, key, &context);
} }
} }
async fn start_tcp( async fn start_tcp(
@@ -679,16 +674,20 @@ impl Channel {
let main_channel = context.inner.main_channel.clone(); let main_channel = context.inner.main_channel.clone();
let buf_sender = if parallel > 1 { let buf_sender = if parallel > 1 {
let (buf_sender, buf_receiver) = buf_channel_group(parallel); let (buf_sender, buf_receiver) = buf_channel_group(parallel);
let mut num = 0;
for buf_receiver in buf_receiver.0 { for buf_receiver in buf_receiver.0 {
let context = context.clone(); let context = context.clone();
let handler = handler.clone(); let handler = handler.clone();
std::thread::spawn(move || { std::thread::Builder::new()
.name(format!("recv-handler-{}", num))
.spawn(move || {
while let Ok((mut buf, start, end, route_key)) = buf_receiver.recv() { while let Ok((mut buf, start, end, route_key)) = buf_receiver.recv() {
handler handler.handle(&mut buf, start, end, route_key, &context);
.handle(&mut buf, start, end, route_key, &context);
} }
log::warn!("异步处理停止"); log::warn!("异步处理停止");
}); })
.unwrap();
num += 1;
} }
Some(buf_sender) Some(buf_sender)
} else { } else {
@@ -711,7 +710,9 @@ impl Channel {
let main_channel_ipv6 = main_channel_ipv6.clone(); let main_channel_ipv6 = main_channel_ipv6.clone();
let handler = handler.clone(); let handler = handler.clone();
let buf_sender = buf_sender.clone(); let buf_sender = buf_sender.clone();
std::thread::spawn(move || { std::thread::Builder::new()
.name("ipv6-recv".into())
.spawn(move || {
log::info!("启动udp v6"); log::info!("启动udp v6");
Self::main_start_( Self::main_start_(
worker, worker,
@@ -722,7 +723,8 @@ impl Channel {
buf_sender, buf_sender,
head_reserve, head_reserve,
) )
}); })
.unwrap();
} }
{ {
let worker = worker.worker("main_channel_1"); let worker = worker.worker("main_channel_1");
@@ -730,7 +732,9 @@ impl Channel {
let main_channel = main_channel.clone(); let main_channel = main_channel.clone();
let handler = handler.clone(); let handler = handler.clone();
let buf_sender = buf_sender.clone(); let buf_sender = buf_sender.clone();
std::thread::spawn(move || { std::thread::Builder::new()
.name("ipv4-recv".into())
.spawn(move || {
log::info!("启动udp v4"); log::info!("启动udp v4");
Self::main_start_( Self::main_start_(
worker, worker,
@@ -741,7 +745,8 @@ impl Channel {
buf_sender, buf_sender,
head_reserve, head_reserve,
) )
}); })
.unwrap();
} }
if relay { if relay {
worker.stop_wait().await; worker.stop_wait().await;
@@ -815,8 +820,7 @@ impl Channel {
break; break;
} }
} }
handler handler.handle(
.handle(
&mut buf, &mut buf,
head_reserve, head_reserve,
end, end,
+9 -1
View File
@@ -403,12 +403,20 @@ impl VntUtil {
let device_list = device_list.clone(); let device_list = device_list.clone();
let current_device = current_device.clone(); let current_device = current_device.clone();
// 定时心跳 // 定时心跳
heartbeat_handler::start_heartbeat_main(
vnt_status_manager.worker("main-heartbeat"),
channel_sender.clone(),
device_list.clone(),
current_device.clone(),
config.server_address_str,
client_cipher.clone(),
self.server_cipher.clone(),
);
heartbeat_handler::start_heartbeat( heartbeat_handler::start_heartbeat(
vnt_status_manager.worker("heartbeat"), vnt_status_manager.worker("heartbeat"),
channel_sender.clone(), channel_sender.clone(),
device_list.clone(), device_list.clone(),
current_device.clone(), current_device.clone(),
config.server_address_str,
client_cipher.clone(), client_cipher.clone(),
self.server_cipher.clone(), self.server_cipher.clone(),
); );
+64 -20
View File
@@ -43,6 +43,28 @@ async fn start_idle_(idle: Idle, sender: ChannelSender) -> io::Result<()> {
} }
pub fn start_heartbeat( pub fn start_heartbeat(
mut worker: VntWorker,
sender: ChannelSender,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
client_cipher: Cipher,
server_cipher: Cipher,
) {
tokio::spawn(async move {
tokio::select! {
_=worker.stop_wait()=>{
return;
}
rs=start_heartbeat_(sender, device_list, current_device,client_cipher,server_cipher)=>{
if let Err(e) = rs {
log::warn!("心跳任务停止:{:?}", e);
}
}
}
worker.stop_all();
});
}
pub fn start_heartbeat_main(
mut worker: VntWorker, mut worker: VntWorker,
sender: ChannelSender, sender: ChannelSender,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>, device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
@@ -56,9 +78,9 @@ pub fn start_heartbeat(
_=worker.stop_wait()=>{ _=worker.stop_wait()=>{
return; return;
} }
rs=start_heartbeat_(sender, device_list, current_device,server_address_str,client_cipher,server_cipher)=>{ rs=start_heartbeat_main_(sender, device_list, current_device,server_address_str,client_cipher,server_cipher)=>{
if let Err(e) = rs { if let Err(e) = rs {
log::warn!("心跳任务停止:{:?}", e); log::warn!("心跳任务停止:{:?}", e);
} }
} }
} }
@@ -97,7 +119,7 @@ fn heartbeat_packet(
net_packet net_packet
} }
async fn start_heartbeat_( async fn start_heartbeat_main_(
sender: ChannelSender, sender: ChannelSender,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>, device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
@@ -106,26 +128,14 @@ async fn start_heartbeat_(
server_cipher: Cipher, server_cipher: Cipher,
) -> io::Result<()> { ) -> io::Result<()> {
let mut count = 0; let mut count = 0;
log::info!("启动心跳任务"); log::info!("启动心跳任务");
loop { loop {
if sender.is_close() { if sender.is_close() {
return Ok(()); return Ok(());
} }
let mut current_dev = current_device.load(); let mut current_dev = current_device.load();
//如果和服务端使用tcp连接,则维持udp洞的频率要更高些 let src = current_dev.virtual_ip();
if (sender.is_main_tcp() && count % 2 == 0) || (!sender.is_main_tcp() && count % 20 == 1) { if count % 40 == 19 {
let mut packet = NetPacket::new_encrypt([0; 12 + ENCRYPTION_RESERVED])?;
packet.set_version(Version::V1);
packet.set_gateway_flag(true);
packet.set_protocol(Protocol::Control);
packet.set_transport_protocol(control_packet::Protocol::AddrRequest.into());
packet.first_set_ttl(MAX_TTL);
packet.set_source(current_dev.virtual_ip());
packet.set_destination(current_dev.virtual_gateway);
server_cipher.encrypt_ipv4(&mut packet)?;
let _ = sender.send_main_udp(packet.buffer(), current_dev.connect_server);
}
if count % 20 == 19 {
if let Ok(mut addr) = server_address_str.to_socket_addrs() { if let Ok(mut addr) = server_address_str.to_socket_addrs() {
if let Some(addr) = addr.next() { if let Some(addr) = addr.next() {
if addr != current_dev.connect_server { if addr != current_dev.connect_server {
@@ -143,7 +153,6 @@ async fn start_heartbeat_(
} }
} }
} }
let src = current_dev.virtual_ip();
let server_packet = heartbeat_packet( let server_packet = heartbeat_packet(
MAX_TTL, MAX_TTL,
&device_list, &device_list,
@@ -156,6 +165,41 @@ async fn start_heartbeat_(
if let Err(e) = sender.send_main(server_packet.buffer(), current_dev.connect_server) { if let Err(e) = sender.send_main(server_packet.buffer(), current_dev.connect_server) {
log::warn!("connect_server:{:?},e:{:?}", current_dev.connect_server, e); log::warn!("connect_server:{:?},e:{:?}", current_dev.connect_server, e);
} }
count += 1;
tokio::time::sleep(Duration::from_millis(3000)).await;
}
}
async fn start_heartbeat_(
sender: ChannelSender,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
client_cipher: Cipher,
server_cipher: Cipher,
) -> io::Result<()> {
let mut count = 0;
log::info!("启动心跳任务");
loop {
if sender.is_close() {
return Ok(());
}
let current_dev = current_device.load();
//如果和服务端使用tcp连接,则维持udp洞的频率要更高些
if (sender.is_main_tcp() && count % 4 == 0) || (!sender.is_main_tcp() && count % 40 == 1) {
let mut packet = NetPacket::new_encrypt([0; 12 + ENCRYPTION_RESERVED])?;
packet.set_version(Version::V1);
packet.set_gateway_flag(true);
packet.set_protocol(Protocol::Control);
packet.set_transport_protocol(control_packet::Protocol::AddrRequest.into());
packet.first_set_ttl(MAX_TTL);
packet.set_source(current_dev.virtual_ip());
packet.set_destination(current_dev.virtual_gateway);
server_cipher.encrypt_ipv4(&mut packet)?;
let _ = sender.send_main_udp(packet.buffer(), current_dev.connect_server);
}
let src = current_dev.virtual_ip();
if count < 7 || count % 7 == 0 { if count < 7 || count % 7 == 0 {
let mut route_list: Option<Vec<(Ipv4Addr, Vec<Route>)>> = None; let mut route_list: Option<Vec<(Ipv4Addr, Vec<Route>)>> = None;
let peer_list = { device_list.lock().1.clone() }; let peer_list = { device_list.lock().1.clone() };
@@ -251,6 +295,6 @@ async fn start_heartbeat_(
} }
count += 1; count += 1;
tokio::time::sleep(Duration::from_millis(5000)).await; tokio::time::sleep(Duration::from_millis(3000)).await;
} }
} }
+42 -8
View File
@@ -1,5 +1,6 @@
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6}; use std::net::{Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
use std::sync::Arc; use std::sync::Arc;
use std::time::{Duration, Instant};
use crossbeam_utils::atomic::AtomicCell; use crossbeam_utils::atomic::AtomicCell;
use dashmap::DashMap; use dashmap::DashMap;
@@ -53,6 +54,7 @@ pub struct ChannelDataHandler {
rsa_cipher: Option<RsaCipher>, rsa_cipher: Option<RsaCipher>,
relay: bool, relay: bool,
token: String, token: String,
time: Arc<AtomicCell<Instant>>,
} }
impl ChannelDataHandler { impl ChannelDataHandler {
@@ -93,6 +95,7 @@ impl ChannelDataHandler {
rsa_cipher, rsa_cipher,
relay, relay,
token, token,
time: Arc::new(AtomicCell::new(Instant::now())),
} }
} }
} }
@@ -132,6 +135,7 @@ impl ChannelDataHandler {
&& !destination.is_multicast() && !destination.is_multicast()
&& destination != current_device.broadcast_address; && destination != current_device.broadcast_address;
if current_device.virtual_ip() != destination if current_device.virtual_ip() != destination
&& !net_packet.is_gateway()
&& not_broadcast && not_broadcast
&& !destination.is_unspecified() && !destination.is_unspecified()
{ {
@@ -160,6 +164,14 @@ impl ChannelDataHandler {
== crate::protocol::error_packet::Protocol::NoKey.into() == crate::protocol::error_packet::Protocol::NoKey.into()
{ {
if let Some(rsa_cipher) = &self.rsa_cipher { if let Some(rsa_cipher) = &self.rsa_cipher {
let last = self.time.load();
if last.elapsed() < Duration::from_secs(3)
|| self.time.compare_exchange(last, Instant::now()).is_err()
{
//短时间不重复上传服务端密钥
return Ok(());
}
log::warn!("上传服务端密钥");
secret_handshake_req( secret_handshake_req(
context, context,
current_device.connect_server, current_device.connect_server,
@@ -273,8 +285,10 @@ impl ChannelDataHandler {
} }
_ => { _ => {
log::warn!( log::warn!(
"不支持的ip代理Icmp协议:{}", "不支持的ip代理Icmp协议:{}->{}->{}",
destination source,
destination,
dest_ip
); );
return Err(Error::Warn( return Err(Error::Warn(
"不支持的ip代理Icmp协议".to_string(), "不支持的ip代理Icmp协议".to_string(),
@@ -283,18 +297,36 @@ impl ChannelDataHandler {
} }
} }
_ => { _ => {
log::warn!("不支持的ip代理ipv4协议:{}", destination); log::warn!(
"不支持的ip代理ipv4协议{:?}:{}->{}->{}",
ipv4.protocol(),
source,
destination,
ipv4.destination_ip()
);
return Err(Error::Warn( return Err(Error::Warn(
"不支持的ip代理ipv4协议".to_string(), "不支持的ip代理ipv4协议".to_string(),
)); ));
} }
} }
} else { } else {
log::warn!("没有ip代理规则:{}", destination); log::warn!(
"没有ip代理规则{:?}:{}->{}->{}",
ipv4.protocol(),
source,
destination,
ipv4.destination_ip()
);
return Err(Error::Warn("没有ip代理规则".to_string())); return Err(Error::Warn("没有ip代理规则".to_string()));
} }
} else { } else {
log::warn!("不支持ip代理:{}", destination); log::warn!(
"不支持ip代理{:?}:{}->{}->{}",
ipv4.protocol(),
source,
destination,
ipv4.destination_ip()
);
return Err(Error::Warn("不支持ip代理".to_string())); return Err(Error::Warn("不支持ip代理".to_string()));
} }
} }
@@ -632,12 +664,14 @@ impl ChannelDataHandler {
service_packet::Protocol::RegistrationResponse => { service_packet::Protocol::RegistrationResponse => {
let response = RegistrationResponse::parse_from_bytes(net_packet.payload())?; let response = RegistrationResponse::parse_from_bytes(net_packet.payload())?;
{ if self.nat_test.can_update() {
let context = context.clone(); let context = context.clone();
let nat_test = self.nat_test.clone(); let nat_test = self.nat_test.clone();
std::thread::spawn(move ||{ std::thread::spawn(move || {
tokio::runtime::Builder::new_current_thread() tokio::runtime::Builder::new_current_thread()
.enable_all().build().unwrap() .enable_all()
.build()
.unwrap()
.block_on(async move { .block_on(async move {
let local_port = context.main_local_ipv4_port().unwrap_or(0); let local_port = context.main_local_ipv4_port().unwrap_or(0);
let local_ipv4_addr = nat::local_ipv4_addr(local_port); let local_ipv4_addr = nat::local_ipv4_addr(local_port);
+1 -1
View File
@@ -229,7 +229,7 @@ impl Register {
} }
pub fn fast_register(&self, ip: Ipv4Addr) -> crate::Result<()> { pub fn fast_register(&self, ip: Ipv4Addr) -> crate::Result<()> {
let last = self.time.load(); let last = self.time.load();
if last.elapsed() < Duration::from_secs(2) if last.elapsed() < Duration::from_secs(3)
|| self.time.compare_exchange(last, Instant::now()).is_err() || self.time.compare_exchange(last, Instant::now()).is_err()
{ {
//短时间不重复注册 //短时间不重复注册
+3
View File
@@ -139,6 +139,9 @@ fn start_simple(
) -> io::Result<()> { ) -> io::Result<()> {
let mut buf = [0; 4096]; let mut buf = [0; 4096];
loop { loop {
if sender.is_close() {
return Ok(());
}
let len = device_reader.read(&mut buf)?; let len = device_reader.read(&mut buf)?;
if let Err(e) = handle( if let Err(e) = handle(
&mut buf, &mut buf,
+16 -5
View File
@@ -1,8 +1,9 @@
use crossbeam_utils::atomic::AtomicCell;
use dashmap::DashMap; use dashmap::DashMap;
use std::io; use std::io;
use std::net::{SocketAddr, SocketAddrV4}; use std::net::{SocketAddr, SocketAddrV4};
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::{Duration, Instant};
use tokio::io::AsyncReadExt; use tokio::io::AsyncReadExt;
use tokio::io::AsyncWriteExt; use tokio::io::AsyncWriteExt;
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
@@ -84,15 +85,21 @@ impl TcpProxy {
async fn proxy(client: TcpStream, server: TcpStream) -> io::Result<()> { async fn proxy(client: TcpStream, server: TcpStream) -> io::Result<()> {
let (client_read, client_write) = client.into_split(); let (client_read, client_write) = client.into_split();
let (server_read, server_write) = server.into_split(); let (server_read, server_write) = server.into_split();
let time = Arc::new(AtomicCell::new(Instant::now()));
let time1 = time.clone();
tokio::spawn(async move { tokio::spawn(async move {
if let Err(e) = copy(client_read, server_write).await { if let Err(e) = copy(client_read, server_write, &time1).await {
log::warn!("{:?}", e); log::warn!("{:?}", e);
} }
}); });
copy(server_read, client_write).await copy(server_read, client_write, &time).await
} }
async fn copy(mut read: OwnedReadHalf, mut write: OwnedWriteHalf) -> io::Result<()> { async fn copy(
mut read: OwnedReadHalf,
mut write: OwnedWriteHalf,
time: &AtomicCell<Instant>,
) -> io::Result<()> {
let mut buf = [0; 10240]; let mut buf = [0; 10240];
loop { loop {
tokio::select! { tokio::select! {
@@ -102,11 +109,15 @@ async fn copy(mut read: OwnedReadHalf, mut write: OwnedWriteHalf) -> io::Result<
break; break;
} }
write.write_all(&buf[..len]).await?; write.write_all(&buf[..len]).await?;
time.store(Instant::now());
} }
_ = tokio::time::sleep(Duration::from_secs(300)) =>{ _ = tokio::time::sleep(Duration::from_secs(600)) =>{
if time.load().elapsed()>=Duration::from_secs(580){
//读写均超时再退出
break; break;
} }
} }
} }
}
Ok(()) Ok(())
} }
+18 -6
View File
@@ -1,10 +1,12 @@
use crate::ip_proxy::DashMapNew; use crate::ip_proxy::DashMapNew;
use crossbeam_utils::atomic::AtomicCell;
use dashmap::DashMap; use dashmap::DashMap;
use std::io; use std::io;
use std::net::{SocketAddr, SocketAddrV4}; use std::net::{SocketAddr, SocketAddrV4};
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use tokio::net::UdpSocket; use tokio::net::UdpSocket;
use tokio::time::Instant;
/// 一个udp代理,作用是利用系统协议栈,将udp数据报解析出来再转发到目的地址 /// 一个udp代理,作用是利用系统协议栈,将udp数据报解析出来再转发到目的地址
pub struct UdpProxy { pub struct UdpProxy {
@@ -22,7 +24,8 @@ impl UdpProxy {
let udp_socket = self.udp_socket; let udp_socket = self.udp_socket;
let mut buf = [0u8; 65536]; let mut buf = [0u8; 65536];
let inner_map: Arc<DashMap<SocketAddrV4, Arc<UdpSocket>>> = Arc::new(DashMap::new0()); let inner_map: Arc<DashMap<SocketAddrV4, (Arc<UdpSocket>, Arc<AtomicCell<Instant>>)>> =
Arc::new(DashMap::new0());
loop { loop {
match udp_socket.recv_from(&mut buf).await { match udp_socket.recv_from(&mut buf).await {
@@ -49,29 +52,36 @@ impl UdpProxy {
async fn start0( async fn start0(
buf: &[u8], buf: &[u8],
sender_addr: SocketAddrV4, sender_addr: SocketAddrV4,
inner_map: &Arc<DashMap<SocketAddrV4, Arc<UdpSocket>>>, inner_map: &Arc<DashMap<SocketAddrV4, (Arc<UdpSocket>, Arc<AtomicCell<Instant>>)>>,
map: &Arc<DashMap<SocketAddrV4, SocketAddrV4>>, map: &Arc<DashMap<SocketAddrV4, SocketAddrV4>>,
udp_socket: &Arc<UdpSocket>, udp_socket: &Arc<UdpSocket>,
) -> io::Result<()> { ) -> io::Result<()> {
if let Some(entry) = inner_map.get(&sender_addr) { if let Some(entry) = inner_map.get(&sender_addr) {
let udp = entry.value().clone(); entry.value().1.store(Instant::now());
let udp = entry.value().0.clone();
drop(entry); drop(entry);
udp.send(buf).await?; udp.send(buf).await?;
} else if let Some(entry) = map.get(&sender_addr) { } else if let Some(entry) = map.get(&sender_addr) {
let dest_addr = *entry.value(); let dest_addr = *entry.value();
drop(entry); drop(entry);
let peer_udp_socket = UdpSocket::bind("0.0.0.0:0").await?; //先使用相同的端口,冲突了再随机端口
let peer_udp_socket = match UdpSocket::bind(format!("0.0.0.0:{}", sender_addr.port())).await
{
Ok(udp) => udp,
Err(_) => UdpSocket::bind("0.0.0.0:0").await?,
};
peer_udp_socket.connect(dest_addr).await?; peer_udp_socket.connect(dest_addr).await?;
peer_udp_socket.send(buf).await?; peer_udp_socket.send(buf).await?;
let peer_udp_socket = Arc::new(peer_udp_socket); let peer_udp_socket = Arc::new(peer_udp_socket);
let inner_map = inner_map.clone(); let inner_map = inner_map.clone();
inner_map.insert(sender_addr, peer_udp_socket.clone()); let time = Arc::new(AtomicCell::new(Instant::now()));
inner_map.insert(sender_addr, (peer_udp_socket.clone(), time.clone()));
let udp_socket = udp_socket.clone(); let udp_socket = udp_socket.clone();
let map = map.clone(); let map = map.clone();
tokio::spawn(async move { tokio::spawn(async move {
let mut buf = [0u8; 65536]; let mut buf = [0u8; 65536];
loop { loop {
match tokio::time::timeout(Duration::from_secs(300), peer_udp_socket.recv(&mut buf)) match tokio::time::timeout(Duration::from_secs(600), peer_udp_socket.recv(&mut buf))
.await .await
{ {
Ok(rs) => match rs { Ok(rs) => match rs {
@@ -98,12 +108,14 @@ async fn start0(
} }
}, },
Err(_) => { Err(_) => {
if time.load().elapsed() > Duration::from_secs(580) {
//超时关闭 //超时关闭
log::warn!("udp代理超时关闭,来源:{},目标:{}", sender_addr, dest_addr); log::warn!("udp代理超时关闭,来源:{},目标:{}", sender_addr, dest_addr);
break; break;
} }
} }
} }
}
inner_map.remove(&sender_addr); inner_map.remove(&sender_addr);
map.remove(&sender_addr); map.remove(&sender_addr);
}); });
+1 -1
View File
@@ -1,5 +1,5 @@
use crate::error::Error; use crate::error::Error;
pub const VNT_VERSION: &'static str = "1.2.4"; pub const VNT_VERSION: &'static str = "1.2.4.4";
pub type Result<T> = std::result::Result<T, Error>; pub type Result<T> = std::result::Result<T, Error>;
pub mod channel; pub mod channel;
+15 -2
View File
@@ -1,7 +1,9 @@
use crossbeam_utils::atomic::AtomicCell;
use std::io; use std::io;
use std::net::UdpSocket; use std::net::UdpSocket;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6}; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
use std::sync::Arc; use std::sync::Arc;
use std::time::{Duration, Instant};
use parking_lot::Mutex; use parking_lot::Mutex;
@@ -22,7 +24,7 @@ pub fn local_ipv4() -> io::Result<Ipv4Addr> {
pub fn local_ipv6() -> io::Result<Ipv6Addr> { pub fn local_ipv6() -> io::Result<Ipv6Addr> {
let socket = UdpSocket::bind("[::]:0")?; let socket = UdpSocket::bind("[::]:0")?;
socket.connect("[2001:4860:4860::8888]:80")?; socket.connect("[2001:4860:4860:0000:0000:0000:0000:8888]:80")?;
let addr = socket.local_addr()?; let addr = socket.local_addr()?;
match addr.ip() { match addr.ip() {
IpAddr::V4(_) => Ok(Ipv6Addr::UNSPECIFIED), IpAddr::V4(_) => Ok(Ipv6Addr::UNSPECIFIED),
@@ -54,6 +56,7 @@ pub fn local_ipv6_addr(port: u16) -> SocketAddrV6 {
pub struct NatTest { pub struct NatTest {
stun_server: Vec<String>, stun_server: Vec<String>,
info: Arc<Mutex<NatInfo>>, info: Arc<Mutex<NatInfo>>,
time: Arc<AtomicCell<Instant>>,
} }
impl From<NatType> for PunchNatType { impl From<NatType> for PunchNatType {
@@ -93,7 +96,16 @@ impl NatTest {
NatType::Cone, NatType::Cone,
); );
let info = Arc::new(Mutex::new(nat_info)); let info = Arc::new(Mutex::new(nat_info));
NatTest { stun_server, info } NatTest {
stun_server,
info,
time: Arc::new(AtomicCell::new(Instant::now())),
}
}
pub fn can_update(&self) -> bool {
let last = self.time.load();
last.elapsed() > Duration::from_secs(10)
&& self.time.compare_exchange(last, Instant::now()).is_ok()
} }
pub fn nat_info(&self) -> NatInfo { pub fn nat_info(&self) -> NatInfo {
self.info.lock().clone() self.info.lock().clone()
@@ -128,6 +140,7 @@ impl NatTest {
ipv6_addr, ipv6_addr,
) )
.await; .await;
log::info!("探测nat类型={:?}", info);
*self.info.lock() = info.clone(); *self.info.lock() = info.clone();
info info
} }
+11 -11
View File
@@ -5,7 +5,6 @@ use bytes::BufMut;
use packet::ethernet; use packet::ethernet;
use parking_lot::Mutex; use parking_lot::Mutex;
use std::net::Ipv4Addr; use std::net::Ipv4Addr;
use std::os::unix::io::AsRawFd;
#[cfg(any(target_os = "linux"))] #[cfg(any(target_os = "linux"))]
use tun::platform::linux::Device; use tun::platform::linux::Device;
#[cfg(any(target_os = "macos"))] #[cfg(any(target_os = "macos"))]
@@ -112,16 +111,17 @@ impl DeviceWriter {
} }
} }
pub fn close(&self) -> io::Result<()> { pub fn close(&self) -> io::Result<()> {
unsafe { //早期使用close直接切断网卡,现在并不需要这么做也能正常关闭
match &self.writer { // unsafe {
DeviceW::Tun(writer) => { // match &self.writer {
libc::close(writer.as_raw_fd()); // DeviceW::Tun(writer) => {
} // libc::close(writer.as_raw_fd());
DeviceW::Tap((writer, _)) => { // }
libc::close(writer.as_raw_fd()); // DeviceW::Tap((writer, _)) => {
} // libc::close(writer.as_raw_fd());
} // }
} // }
// }
Ok(()) Ok(())
} }
pub fn is_tun(&self) -> bool { pub fn is_tun(&self) -> bool {
+4 -4
View File
@@ -117,7 +117,7 @@ impl DeviceWriter {
// 当前网段路由 // 当前网段路由
dev.add_route(address, netmask, gateway, 1)?; dev.add_route(address, netmask, gateway, 1)?;
// 广播和组播路由 // 广播和组播路由
dev.add_route(Ipv4Addr::BROADCAST, Ipv4Addr::BROADCAST, gateway, 1)?; // dev.add_route(Ipv4Addr::BROADCAST, Ipv4Addr::BROADCAST, gateway, 1)?;
dev.add_route( dev.add_route(
Ipv4Addr::from([224, 0, 0, 0]), Ipv4Addr::from([224, 0, 0, 0]),
Ipv4Addr::from([240, 0, 0, 0]), Ipv4Addr::from([240, 0, 0, 0]),
@@ -230,8 +230,8 @@ fn create_tun(
} }
// 当前网段路由 // 当前网段路由
tun_device.add_route(address, netmask, gateway, 1)?; tun_device.add_route(address, netmask, gateway, 1)?;
// 广播和组播路由 // 广播和组播路由 修改了广播路由会导致发不出广播
tun_device.add_route(Ipv4Addr::BROADCAST, Ipv4Addr::BROADCAST, gateway, 1)?; // tun_device.add_route(Ipv4Addr::BROADCAST, Ipv4Addr::BROADCAST, gateway, 1)?;
tun_device.add_route( tun_device.add_route(
Ipv4Addr::from([224, 0, 0, 0]), Ipv4Addr::from([224, 0, 0, 0]),
Ipv4Addr::from([240, 0, 0, 0]), Ipv4Addr::from([240, 0, 0, 0]),
@@ -309,7 +309,7 @@ fn create_tap(
tap_device.add_route(*address, *netmask, gateway, 1)?; tap_device.add_route(*address, *netmask, gateway, 1)?;
} }
// 广播和组播路由 // 广播和组播路由
tap_device.add_route(Ipv4Addr::BROADCAST, Ipv4Addr::BROADCAST, gateway, 1)?; // tap_device.add_route(Ipv4Addr::BROADCAST, Ipv4Addr::BROADCAST, gateway, 1)?;
tap_device.add_route( tap_device.add_route(
Ipv4Addr::from([224, 0, 0, 0]), Ipv4Addr::from([224, 0, 0, 0]),
Ipv4Addr::from([240, 0, 0, 0]), Ipv4Addr::from([240, 0, 0, 0]),
+2 -1
View File
@@ -29,5 +29,6 @@ features = [
"winerror", "winerror",
"ipexport", "ipexport",
"iphlpapi", "iphlpapi",
"handleapi" "handleapi",
"ifdef"
] ]
+2 -1
View File
@@ -119,7 +119,8 @@ impl TapDevice {
} }
pub fn delete(self) -> io::Result<()> { pub fn delete(self) -> io::Result<()> {
iface::delete_interface(&self.luid) // iface::delete_interface(&self.luid)
Ok(())
} }
} }
+12 -6
View File
@@ -1,12 +1,12 @@
use std::io; use std::io;
use std::net::Ipv4Addr; use std::net::Ipv4Addr;
use winapi::um::{handleapi, synchapi, winbase, winnt}; use winapi::um::{synchapi, winbase, winnt};
use crate::{decode_utf16, encode_utf16, ffi, netsh, route, IFace}; use crate::{decode_utf16, encode_utf16, ffi, netsh, route, IFace};
use rand::Rng; use rand::Rng;
mod log;
pub mod packet; pub mod packet;
mod wintun_log;
mod wintun_raw; mod wintun_raw;
/// The maximum size of wintun's internal ring buffer (in bytes) /// The maximum size of wintun's internal ring buffer (in bytes)
@@ -80,7 +80,7 @@ impl TunDevice {
let guid_struct: wintun_raw::GUID = unsafe { std::mem::transmute(guid) }; let guid_struct: wintun_raw::GUID = unsafe { std::mem::transmute(guid) };
let guid_ptr = &guid_struct as *const wintun_raw::GUID; let guid_ptr = &guid_struct as *const wintun_raw::GUID;
log::set_default_logger_if_unset(&win_tun); wintun_log::set_default_logger_if_unset(&win_tun);
//SAFETY: the function is loaded from the wintun dll properly, we are providing valid //SAFETY: the function is loaded from the wintun dll properly, we are providing valid
//pointers, and all the strings are correct null terminated UTF-16. This safety rationale //pointers, and all the strings are correct null terminated UTF-16. This safety rationale
@@ -88,6 +88,7 @@ impl TunDevice {
let adapter = let adapter =
win_tun.WintunCreateAdapter(pool_utf16.as_ptr(), name_utf16.as_ptr(), guid_ptr); win_tun.WintunCreateAdapter(pool_utf16.as_ptr(), name_utf16.as_ptr(), guid_ptr);
if adapter.is_null() { if adapter.is_null() {
log::error!("adapter.is_null {:?}", io::Error::last_os_error());
return Err(io::Error::new( return Err(io::Error::new(
io::ErrorKind::Other, io::ErrorKind::Other,
"Failed to crate adapter", "Failed to crate adapter",
@@ -102,6 +103,7 @@ impl TunDevice {
// 开启session // 开启session
let session = win_tun.WintunStartSession(adapter, 128 * 1024); let session = win_tun.WintunStartSession(adapter, 128 * 1024);
if session.is_null() { if session.is_null() {
log::error!("session.is_null {:?}", io::Error::last_os_error());
return Err(io::Error::new( return Err(io::Error::new(
io::ErrorKind::Other, io::ErrorKind::Other,
"WintunStartSession failed", "WintunStartSession failed",
@@ -138,10 +140,14 @@ impl TunDevice {
)); ));
} }
}; };
log::set_default_logger_if_unset(&win_tun); wintun_log::set_default_logger_if_unset(&win_tun);
let name_utf16 = encode_utf16(name); let name_utf16 = encode_utf16(name);
let adapter = win_tun.WintunOpenAdapter(name_utf16.as_ptr()); let adapter = win_tun.WintunOpenAdapter(name_utf16.as_ptr());
if adapter.is_null() { if adapter.is_null() {
log::error!(
"delete_for_name adapter.is_null {:?}",
io::Error::last_os_error()
);
return Err(io::Error::new( return Err(io::Error::new(
io::ErrorKind::Other, io::ErrorKind::Other,
"Failed to open adapter", "Failed to open adapter",
@@ -187,8 +193,8 @@ pub struct Version {
impl IFace for TunDevice { impl IFace for TunDevice {
fn shutdown(&self) -> io::Result<()> { fn shutdown(&self) -> io::Result<()> {
let _ = unsafe { synchapi::SetEvent(self.shutdown_event) }; // let _ = unsafe { synchapi::SetEvent(self.shutdown_event) };
let _ = unsafe { handleapi::CloseHandle(self.shutdown_event) }; // let _ = unsafe { handleapi::CloseHandle(self.shutdown_event) };
Ok(()) Ok(())
} }