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

格式化代码

Browse Source
This commit is contained in:
lubeilin
2023-01-09 20:26:25 +08:00
parent 61086089b3
commit c655d9650b
18 changed files with 398 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files
switch/packet/src/lib.rs
-1
View File
@@ -133,7 +133,6 @@ fn u32c(x: u8, y: u8) -> u32 {
((x as u32) << 8) | y as u32
}
#[cfg(test)]
mod tests {
use super::*;
switch/rust-tun/src/platform/linux/device.rs
+4 -4
View File
@@ -77,10 +77,10 @@ impl Device {
req.ifru.flags = device_type
| if config.platform.packet_information {
0
} else {
IFF_NO_PI
}
0
} else {
IFF_NO_PI
}
| if queues_num > 1 { IFF_MULTI_QUEUE } else { 0 };
for _ in 0..queues_num {
switch/src/handle/heartbeat_handler.rs
+16 -7
View File
@@ -5,27 +5,36 @@ use chrono::Local;
use tokio::sync::watch::Receiver;
use tokio::time::sleep;
use crate::{CurrentDeviceInfo, DEVICE_LIST};
use crate::error::*;
use crate::handle::{ApplicationStatus, DIRECT_ROUTE_TABLE};
use crate::protocol::{control_packet, NetPacket, Protocol, Version};
use crate::protocol::control_packet::PingPacket;
use crate::protocol::{control_packet, NetPacket, Protocol, Version};
use crate::{CurrentDeviceInfo, DEVICE_LIST};
pub async fn start<F>(status_watch: Receiver<ApplicationStatus>,
udp: UdpSocket, cur_info: CurrentDeviceInfo, stop_fn: F)
where F: FnOnce() + Send + 'static {
pub async fn start<F>(
status_watch: Receiver<ApplicationStatus>,
udp: UdpSocket,
cur_info: CurrentDeviceInfo,
stop_fn: F,
) where
F: FnOnce() + Send + 'static,
{
tokio::spawn(async move {
match handle_loop(status_watch, udp, cur_info.connect_server).await {
Ok(_) => {}
Err(e) => {
log::error!("{:?}",e)
log::error!("{:?}", e)
}
}
stop_fn();
});
}
async fn handle_loop(mut status_watch: Receiver<ApplicationStatus>, udp: UdpSocket, server_addr: SocketAddr) -> Result<()> {
async fn handle_loop(
mut status_watch: Receiver<ApplicationStatus>,
udp: UdpSocket,
server_addr: SocketAddr,
) -> Result<()> {
const INTERVAL: u64 = 3000;
const MAX_INTERVAL: i64 = 3000 * 3;
let mut buf = [0u8; (4 + 8 + 4)];
switch/src/handle/mod.rs
+14 -9
View File
@@ -62,10 +62,12 @@ pub struct NatInfo {
}
impl NatInfo {
pub fn new(public_ips: Vec<u32>,
public_port: u16,
public_port_range: u16,
nat_type: NatType, ) -> Self {
pub fn new(
public_ips: Vec<u32>,
public_port: u16,
public_port_range: u16,
nat_type: NatType,
) -> Self {
Self {
public_ips,
public_port,
@@ -87,9 +89,7 @@ pub fn init_nat_info(public_ip: u32, public_port: u16) {
public_ips.push(ip);
}
}
let nat_info = NatInfo::new(public_ips,
public_port,
port_range, nat_type);
let nat_info = NatInfo::new(public_ips, public_port, port_range, nat_type);
// println!("nat信息:{:?}",nat_info);
let mut nat_info_lock = NAT_INFO.lock();
nat_info_lock.replace(nat_info);
@@ -114,7 +114,12 @@ pub struct CurrentDeviceInfo {
}
impl CurrentDeviceInfo {
pub fn new(virtual_ip: Ipv4Addr, virtual_gateway: Ipv4Addr, virtual_netmask: Ipv4Addr, connect_server: SocketAddr) -> Self {
pub fn new(
virtual_ip: Ipv4Addr,
virtual_gateway: Ipv4Addr,
virtual_netmask: Ipv4Addr,
connect_server: SocketAddr,
) -> Self {
let broadcast_address = (!u32::from_be_bytes(virtual_netmask.octets()))
| u32::from_be_bytes(virtual_gateway.octets());
let broadcast_address = Ipv4Addr::from(broadcast_address);
@@ -152,7 +157,7 @@ impl Into<u8> for RouteType {
fn into(self) -> u8 {
match self {
RouteType::ServerRelay => 0,
RouteType::P2P => 1
RouteType::P2P => 1,
}
}
}
switch/src/handle/punch_handler.rs
+70 -47
View File
@@ -5,29 +5,37 @@ use std::time::Duration;
use dashmap::DashMap;
use lazy_static::lazy_static;
use protobuf::Message;
use tokio::sync::mpsc::{Receiver, Sender};
use tokio::sync::mpsc::error::TrySendError;
use tokio::sync::mpsc::{Receiver, Sender};
use tokio::sync::watch;
use crate::{CurrentDeviceInfo, DEVICE_LIST, handle::NAT_INFO, handle::NatInfo};
use crate::error::*;
use crate::handle::{ApplicationStatus, DIRECT_ROUTE_TABLE};
use crate::proto::message::{NatType, Punch, Step};
use crate::protocol::{control_packet, NetPacket, Protocol, turn_packet, Version};
use crate::protocol::control_packet::PunchRequestPacket;
use crate::protocol::turn_packet::TurnPacket;
use crate::protocol::{control_packet, turn_packet, NetPacket, Protocol, Version};
use crate::{handle::NatInfo, handle::NAT_INFO, CurrentDeviceInfo, DEVICE_LIST};
lazy_static! {
pub static ref STEP_MAP:DashMap<Ipv4Addr,Step> = DashMap::new();
pub static ref STEP_MAP: DashMap<Ipv4Addr, Step> = DashMap::new();
}
/// 每一种类型一个通道，减少相互干扰
pub fn bounded() -> (PunchSender, ConeReceiver, ReqSymmetricReceiver, ResSymmetricReceiver) {
pub fn bounded() -> (
PunchSender,
ConeReceiver,
ReqSymmetricReceiver,
ResSymmetricReceiver,
) {
let (cone_sender, cone_receiver) = tokio::sync::mpsc::channel(3);
let (req_symmetric_sender, req_symmetric_receiver) = tokio::sync::mpsc::channel(1);
let (res_symmetric_sender, res_symmetric_receiver) = tokio::sync::mpsc::channel(1);
(PunchSender::new(cone_sender, req_symmetric_sender, res_symmetric_sender),
ConeReceiver(cone_receiver), ReqSymmetricReceiver(req_symmetric_receiver),
ResSymmetricReceiver(res_symmetric_receiver))
(
PunchSender::new(cone_sender, req_symmetric_sender, res_symmetric_sender),
ConeReceiver(cone_receiver),
ReqSymmetricReceiver(req_symmetric_receiver),
ResSymmetricReceiver(res_symmetric_receiver),
)
}
pub struct ConeReceiver(Receiver<Punch>);
@@ -44,9 +52,11 @@ pub struct PunchSender {
}
impl PunchSender {
pub fn new(cone_sender: Sender<Punch>,
req_symmetric_sender: Sender<Punch>,
res_symmetric_sender: Sender<Punch>, ) -> Self {
pub fn new(
cone_sender: Sender<Punch>,
req_symmetric_sender: Sender<Punch>,
res_symmetric_sender: Sender<Punch>,
) -> Self {
Self {
cone_sender,
req_symmetric_sender,
@@ -78,14 +88,17 @@ impl PunchSender {
self.req_symmetric_sender.try_send(punch)
}
}
NatType::Cone => {
self.cone_sender.try_send(punch)
}
NatType::Cone => self.cone_sender.try_send(punch),
}
}
}
fn handle(status_watch: &watch::Receiver<ApplicationStatus>, udp: &UdpSocket, punch_list: Vec<Punch>, buf: &[u8]) -> Result<()> {
fn handle(
status_watch: &watch::Receiver<ApplicationStatus>,
udp: &UdpSocket,
punch_list: Vec<Punch>,
buf: &[u8],
) -> Result<()> {
let mut counter = 0u64;
for punch in punch_list {
let dest = Ipv4Addr::from(punch.virtual_ip);
@@ -107,7 +120,8 @@ fn handle(status_watch: &watch::Receiver<ApplicationStatus>, udp: &UdpSocket, pu
}
}
let right_port = ((punch.public_port + range) & 0xFFFF) as u16;
let left_port = ((0xFFFF + punch.public_port - range) & 0xFFFF) as u16;
let left_port =
((0xFFFF + punch.public_port - range) & 0xFFFF) as u16;
if right_port != 0 {
// println!("{:?}", SocketAddr::V4(SocketAddrV4::new(pub_ip, right_port)));
udp.send_to(
@@ -140,10 +154,7 @@ fn handle(status_watch: &watch::Receiver<ApplicationStatus>, udp: &UdpSocket, pu
return Ok(());
}
}
udp.send_to(
buf,
SocketAddr::V4(SocketAddrV4::new(pub_ip, port)),
)?;
udp.send_to(buf, SocketAddr::V4(SocketAddrV4::new(pub_ip, port)))?;
select_sleep(&mut counter);
}
}
@@ -168,17 +179,21 @@ fn handle(status_watch: &watch::Receiver<ApplicationStatus>, udp: &UdpSocket, pu
}
/// 给对称nat发送打洞数据包
pub async fn req_symmetric_handler_start<F>(status_watch: watch::Receiver<ApplicationStatus>,
receiver: ReqSymmetricReceiver,
udp: UdpSocket,
cur_info: CurrentDeviceInfo,
stop_fn: F) where F: FnOnce() +Send+'static{
pub async fn req_symmetric_handler_start<F>(
status_watch: watch::Receiver<ApplicationStatus>,
receiver: ReqSymmetricReceiver,
udp: UdpSocket,
cur_info: CurrentDeviceInfo,
stop_fn: F,
) where
F: FnOnce() + Send + 'static,
{
let receiver = receiver.0;
tokio::spawn(async move {
match handle_loop(status_watch, receiver, udp, cur_info).await {
Ok(_) => {}
Err(e) => {
log::error!("{:?}",e)
log::error!("{:?}", e)
}
}
stop_fn()
@@ -195,17 +210,21 @@ pub async fn req_symmetric_handler_start<F>(status_watch: watch::Receiver<Applic
// }
/// 给对称nat发送打洞数据包，处理主动发起的打洞操作
pub async fn res_symmetric_handler_start<F>(status_watch: watch::Receiver<ApplicationStatus>,
receiver: ResSymmetricReceiver,
udp: UdpSocket,
cur_info: CurrentDeviceInfo,
stop_fn: F) where F: FnOnce() +Send+'static{
pub async fn res_symmetric_handler_start<F>(
status_watch: watch::Receiver<ApplicationStatus>,
receiver: ResSymmetricReceiver,
udp: UdpSocket,
cur_info: CurrentDeviceInfo,
stop_fn: F,
) where
F: FnOnce() + Send + 'static,
{
let receiver = receiver.0;
tokio::spawn(async move {
match res_symmetric_handle_loop(status_watch, receiver, udp, cur_info).await {
Ok(_) => {}
Err(e) => {
log::error!("{:?}",e)
log::error!("{:?}", e)
}
}
stop_fn()
@@ -290,17 +309,21 @@ async fn res_symmetric_handle_loop(
}
/// 给锥形nat发送打洞数据包
pub async fn cone_handler_start<F>(status_watch: watch::Receiver<ApplicationStatus>,
receiver: ConeReceiver,
udp: UdpSocket,
cur_info: CurrentDeviceInfo,
stop_fn: F) where F: FnOnce()+Send +'static{
pub async fn cone_handler_start<F>(
status_watch: watch::Receiver<ApplicationStatus>,
receiver: ConeReceiver,
udp: UdpSocket,
cur_info: CurrentDeviceInfo,
stop_fn: F,
) where
F: FnOnce() + Send + 'static,
{
let receiver = receiver.0;
tokio::spawn(async move {
match handle_loop(status_watch, receiver, udp, cur_info).await {
Ok(_) => {}
Err(e) => {
log::error!("{:?}",e)
log::error!("{:?}", e)
}
}
stop_fn();
@@ -361,16 +384,13 @@ fn select_sleep(counter: &mut u64) {
thread::sleep(Duration::from_millis(1));
}
fn punch_request_handle(udp: &UdpSocket, cur_info: &CurrentDeviceInfo) -> Result<()> {
let nat_info_lock = NAT_INFO.lock();
let nat_info = nat_info_lock.clone();
drop(nat_info_lock);
if let Some(nat_info) = nat_info {
if let Err(e) = send_punch(&udp,
&cur_info,
nat_info) {
log::error!("发送打洞数据失败 {:?}",e)
if let Err(e) = send_punch(&udp, &cur_info, nat_info) {
log::error!("发送打洞数据失败 {:?}", e)
}
Ok(())
} else {
@@ -378,7 +398,6 @@ fn punch_request_handle(udp: &UdpSocket, cur_info: &CurrentDeviceInfo) -> Result
}
}
fn send_punch(udp: &UdpSocket, cur_info: &CurrentDeviceInfo, nat_info: NatInfo) -> Result<()> {
let lock = DEVICE_LIST.lock();
let list = lock.1.clone();
@@ -391,15 +410,19 @@ fn send_punch(udp: &UdpSocket, cur_info: &CurrentDeviceInfo, nat_info: NatInfo)
} else {
Step::Step1
};
let bytes = punch_packet(cur_info.virtual_ip,
nat_info.clone(), ip, step)?;
let bytes = punch_packet(cur_info.virtual_ip, nat_info.clone(), ip, step)?;
udp.send_to(&bytes, cur_info.connect_server)?;
}
}
Ok(())
}
fn punch_packet(virtual_ip: Ipv4Addr, nat_info: NatInfo, dest: Ipv4Addr, step: Step) -> Result<Vec<u8>> {
fn punch_packet(
virtual_ip: Ipv4Addr,
nat_info: NatInfo,
dest: Ipv4Addr,
step: Step,
) -> Result<Vec<u8>> {
let mut punch_reply = Punch::new();
punch_reply.reply = false;
punch_reply.virtual_ip = u32::from_be_bytes(virtual_ip.octets());
switch/src/handle/registration_handler.rs
+3 -3
View File
@@ -11,7 +11,7 @@ use protobuf::Message;
use crate::error::*;
use crate::handle::ConnectStatus;
use crate::proto::message::{RegistrationRequest, RegistrationResponse};
use crate::protocol::{error_packet, NetPacket, Protocol, service_packet, Version};
use crate::protocol::{error_packet, service_packet, NetPacket, Protocol, Version};
lazy_static::lazy_static! {
static ref REQUEST:RwLock<Option<(String,String)>> = parking_lot::const_rwlock(None);
@@ -98,8 +98,8 @@ pub fn fast_registration(udp: &UdpSocket, server_address: SocketAddr) -> Result<
let new = Local::now().timestamp_millis();
if new - last < 2000
|| REGISTRATION_TIME
.compare_exchange(last, new, Ordering::Relaxed, Ordering::Relaxed)
.is_err()
.compare_exchange(last, new, Ordering::Relaxed, Ordering::Relaxed)
.is_err()
{
//短时间不重复注册
return Ok(());
switch/src/handle/tun_handler.rs
+41 -24
View File
@@ -10,12 +10,12 @@ use packet::icmp::Kind;
use packet::ip::ipv4;
use packet::ip::ipv4::packet::IpV4Packet;
use crate::ApplicationStatus;
use crate::error::*;
use crate::handle::{CurrentDeviceInfo, DIRECT_ROUTE_TABLE};
use crate::protocol::{NetPacket, Protocol, Version};
use crate::protocol::turn_packet::TurnPacket;
use crate::protocol::{NetPacket, Protocol, Version};
use crate::tun_device::TunReader;
use crate::ApplicationStatus;
/// 是否在一个网段
fn check_dest(dest: Ipv4Addr, cur_info: &CurrentDeviceInfo) -> bool {
@@ -77,42 +77,51 @@ fn handle(
if let Some(route) = DIRECT_ROUTE_TABLE.get(&dest_ip) {
let current_time = Local::now().timestamp_millis();
if current_time - route.recv_time < 3_000 {
if udp.send_to(&net_packet.buffer()[..(4 + 8 + data_len)], route.address).is_ok() {
if udp
.send_to(&net_packet.buffer()[..(4 + 8 + data_len)], route.address)
.is_ok()
{
return Ok(());
}
}
}
udp.send_to(&net_packet.buffer()[..(4 + 8 + data_len)], cur_info.connect_server)?;
udp.send_to(
&net_packet.buffer()[..(4 + 8 + data_len)],
cur_info.connect_server,
)?;
return Ok(());
}
#[cfg(target_os = "windows")]
pub async fn handler_start<F>(mut status_watch: watch::Receiver<ApplicationStatus>,
udp: UdpSocket,
tun_reader: TunReader,
cur_info: CurrentDeviceInfo, stop_fn: F)
where F: FnOnce() + Send + 'static {
pub async fn handler_start<F>(
mut status_watch: watch::Receiver<ApplicationStatus>,
udp: UdpSocket,
tun_reader: TunReader,
cur_info: CurrentDeviceInfo,
stop_fn: F,
) where
F: FnOnce() + Send + 'static,
{
let session = tun_reader.0.clone();
tokio::spawn(async move {
let _ = status_watch.changed().await;
session.shutdown();
let udp = UdpSocket::bind("0.0.0.0:0").unwrap();
let _ = udp.send_to(&[0],SocketAddr::new(IpAddr::V4(cur_info.virtual_gateway),10));
let _ = udp.send_to(
&[0],
SocketAddr::new(IpAddr::V4(cur_info.virtual_gateway), 10),
);
});
thread::spawn(move || {
if let Err(e) = handle_loop(udp, tun_reader, cur_info) {
log::error!("tun数据处理线程停止 {:?}",e);
log::error!("tun数据处理线程停止 {:?}", e);
}
stop_fn();
});
}
#[cfg(target_os = "windows")]
fn handle_loop(
udp: UdpSocket,
tun_reader: TunReader,
cur_info: CurrentDeviceInfo,
) -> Result<()> {
fn handle_loop(udp: UdpSocket, tun_reader: TunReader, cur_info: CurrentDeviceInfo) -> Result<()> {
let mut net_packet = NetPacket::new(vec![0u8; 4 + 8 + 1500])?;
net_packet.set_version(Version::V1);
net_packet.set_protocol(Protocol::Ipv4Turn);
@@ -130,11 +139,16 @@ fn handle_loop(
}
#[cfg(any(target_os = "macos", target_os = "linux", target_os = "android"))]
pub async fn handler_start<F>(mut status_watch: watch::Receiver<ApplicationStatus>,
udp: UdpSocket,
tun_reader: TunReader,
cur_info: CurrentDeviceInfo, stop_fn: F)
where F: FnOnce() + Send + 'static {
pub async fn handler_start<F>(
mut status_watch: watch::Receiver<ApplicationStatus>,
udp: UdpSocket,
tun_reader: TunReader,
cur_info: CurrentDeviceInfo,
stop_fn: F,
) where
F: FnOnce() + Send + 'static,
{
use std::os::fd::AsRawFd;
let raw_fd = tun_reader.0.as_raw_fd();
tokio::spawn(async move {
let _ = status_watch.changed().await;
@@ -143,11 +157,14 @@ pub async fn handler_start<F>(mut status_watch: watch::Receiver<ApplicationStatu
libc::close(raw_fd);
}
let udp = UdpSocket::bind("0.0.0.0:0").unwrap();
let _ = udp.send_to(&[0],SocketAddr::new(IpAddr::V4(cur_info.virtual_gateway),10));
let _ = udp.send_to(
&[0],
SocketAddr::new(IpAddr::V4(cur_info.virtual_gateway), 10),
);
});
thread::spawn(move || {
if let Err(e) = handle_loop(udp, tun_reader, cur_info) {
log::error!(" tun数据处理线程停止 {:?}",e);
log::error!(" tun数据处理线程停止 {:?}", e);
}
stop_fn();
});
@@ -170,7 +187,7 @@ pub fn handle_loop(
match handle(&udp, data, &cur_info, &mut net_packet) {
Ok(_) => {}
Err(e) => {
log::error!("{:?}",e)
log::error!("{:?}", e)
}
}
}
switch/src/handle/udp_recv_handler.rs
+42 -33
View File
@@ -7,21 +7,23 @@ use packet::icmp::{icmp, Kind};
use packet::ip::ipv4;
use packet::ip::ipv4::packet::IpV4Packet;
use protobuf::Message;
use tokio::sync::mpsc::{Receiver, Sender};
use tokio::sync::mpsc::error::TrySendError;
use tokio::sync::mpsc::{Receiver, Sender};
use tokio::sync::watch;
use crate::{ApplicationStatus, CurrentDeviceInfo};
use crate::error::*;
use crate::handle::{ADDR_TABLE, ConnectStatus, DEVICE_LIST, DIRECT_ROUTE_TABLE, NAT_INFO, Route, SERVER_RT};
use crate::handle::punch_handler::PunchSender;
use crate::handle::registration_handler::{CONNECTION_STATUS, fast_registration};
use crate::handle::registration_handler::{fast_registration, CONNECTION_STATUS};
use crate::handle::{
ConnectStatus, Route, ADDR_TABLE, DEVICE_LIST, DIRECT_ROUTE_TABLE, NAT_INFO, SERVER_RT,
};
use crate::proto::message::{DeviceList, Punch, RegistrationResponse};
use crate::protocol::{control_packet, NetPacket, Protocol, service_packet, turn_packet, Version};
use crate::protocol::control_packet::{ControlPacket, PunchResponsePacket};
use crate::protocol::error_packet::InErrorPacket;
use crate::protocol::turn_packet::TurnPacket;
use crate::protocol::{control_packet, service_packet, turn_packet, NetPacket, Protocol, Version};
use crate::tun_device::TunWriter;
use crate::{ApplicationStatus, CurrentDeviceInfo};
const UDP_STOP_BUF: [u8; 1] = [0u8];
@@ -32,8 +34,10 @@ pub async fn udp_recv_start<F>(
other_sender: Sender<(SocketAddr, Vec<u8>)>,
mut tun_writer: TunWriter,
current_device: CurrentDeviceInfo,
stop_fn: F)
where F: FnOnce() + Send + 'static {
stop_fn: F,
) where
F: FnOnce() + Send + 'static,
{
{
let udp = udp.try_clone().unwrap();
tokio::spawn(async move {
@@ -45,14 +49,8 @@ pub async fn udp_recv_start<F>(
}
thread::spawn(move || {
if let Err(e) = recv_loop(
udp,
server_addr,
other_sender,
tun_writer,
current_device,
) {
log::error!("udp数据处理线程停止 {:?}",e);
if let Err(e) = recv_loop(udp, server_addr, other_sender, tun_writer, current_device) {
log::error!("udp数据处理线程停止 {:?}", e);
}
stop_fn();
});
@@ -97,12 +95,12 @@ fn recv_loop(
return Err(Error::Stop(str));
}
Err(e) => {
log::error!("{:?}",e);
log::error!("{:?}", e);
}
}
}
Err(e) => {
log::error!("{:?}",e);
log::error!("{:?}", e);
}
};
}
@@ -158,7 +156,7 @@ fn recv_handle(
return Err(Error::Stop("子处理线程停止".to_string()));
}
Err(e) => {
log::error!("子线程处理 {:?}",e);
log::error!("子线程处理 {:?}", e);
}
}
}
@@ -166,17 +164,21 @@ fn recv_handle(
Ok(())
}
pub async fn udp_other_recv_start<F>(status_watch: watch::Receiver<ApplicationStatus>,
udp: UdpSocket,
receiver: Receiver<(SocketAddr, Vec<u8>)>,
current_device: CurrentDeviceInfo,
sender: PunchSender,
stop_fn: F) where F: FnOnce() + Send + 'static {
pub async fn udp_other_recv_start<F>(
status_watch: watch::Receiver<ApplicationStatus>,
udp: UdpSocket,
receiver: Receiver<(SocketAddr, Vec<u8>)>,
current_device: CurrentDeviceInfo,
sender: PunchSender,
stop_fn: F,
) where
F: FnOnce() + Send + 'static,
{
tokio::spawn(async move {
match other_loop(status_watch, udp, receiver, current_device, sender).await {
Ok(_) => {}
Err(e) => {
log::error!("{:?}",e);
log::error!("{:?}", e);
}
}
stop_fn();
@@ -267,7 +269,7 @@ fn other_handle(
}
}
InErrorPacket::OtherError(e) => {
log::error!("OtherError {:?}",e.message());
log::error!("OtherError {:?}", e.message());
}
}
}
@@ -301,7 +303,8 @@ fn other_handle(
//回应
let mut punch_response = PunchResponsePacket::new(net_packet.payload_mut())?;
punch_response.set_source(current_device.virtual_ip);
net_packet.set_transport_protocol(control_packet::Protocol::PunchResponse.into());
net_packet
.set_transport_protocol(control_packet::Protocol::PunchResponse.into());
udp.send_to(net_packet.buffer(), peer_addr)?;
let route = Route::new(peer_addr);
DIRECT_ROUTE_TABLE.insert(src, route);
@@ -329,7 +332,8 @@ fn other_handle(
if !punch.reply {
let mut punch_reply = Punch::new();
punch_reply.reply = true;
punch_reply.virtual_ip = u32::from_be_bytes(current_device.virtual_ip.octets());
punch_reply.virtual_ip =
u32::from_be_bytes(current_device.virtual_ip.octets());
punch_reply.step = punch.step;
if let Err(_) = sender.try_send(punch) {
return Ok(());
@@ -339,15 +343,20 @@ fn other_handle(
punch_reply.public_ip_list = info.public_ips.clone();
punch_reply.public_port = info.public_port as u32;
punch_reply.public_port_range = info.public_port_range as u32;
punch_reply.nat_type = protobuf::EnumOrUnknown::new(info.nat_type);
punch_reply.nat_type =
protobuf::EnumOrUnknown::new(info.nat_type);
drop(nat_info);
let bytes = punch_reply.write_to_bytes()?;
let mut net_packet = NetPacket::new(vec![0u8; 4 + 8 + bytes.len()])?;
let mut net_packet =
NetPacket::new(vec![0u8; 4 + 8 + bytes.len()])?;
net_packet.set_version(Version::V1);
net_packet.set_protocol(Protocol::OtherTurn);
net_packet.set_transport_protocol(turn_packet::Protocol::Punch.into());
net_packet.set_transport_protocol(
turn_packet::Protocol::Punch.into(),
);
net_packet.set_ttl(255);
let mut turn_packet = TurnPacket::new(net_packet.payload_mut())?;
let mut turn_packet =
TurnPacket::new(net_packet.payload_mut())?;
turn_packet.set_source(current_device.virtual_ip);
turn_packet.set_destination(src);
turn_packet.set_payload(&bytes);
@@ -365,7 +374,7 @@ fn other_handle(
}
}
Protocol::UnKnow(p) => {
log::error!("未知协议 {}",p);
log::error!("未知协议 {}", p);
}
}
Ok(())
switch/src/lib.rs
+78 -48
View File
@@ -7,15 +7,18 @@ use tokio::sync::watch;
use error::*;
use crate::handle::{ApplicationStatus, ConnectStatus, CurrentDeviceInfo, DEVICE_LIST, DIRECT_ROUTE_TABLE, Route, RouteType, SERVER_RT};
use crate::handle::registration_handler::CONNECTION_STATUS;
use crate::handle::{
ApplicationStatus, ConnectStatus, CurrentDeviceInfo, Route, RouteType, DEVICE_LIST,
DIRECT_ROUTE_TABLE, SERVER_RT,
};
pub mod tun_device;
pub mod nat;
pub mod error;
pub mod handle;
pub mod nat;
pub mod proto;
pub mod protocol;
pub mod tun_device;
#[derive(Clone, Debug)]
pub struct Config {
@@ -25,10 +28,7 @@ pub struct Config {
impl Config {
pub fn new(token: String, mac_address: String) -> Self {
Self {
token,
mac_address,
}
Self { token, mac_address }
}
}
@@ -50,9 +50,7 @@ impl Switch {
switch.runtime = Some(runtime);
Ok(switch)
}
Err(e) => {
Err(e)
}
Err(e) => Err(e),
};
}
pub fn stop(self) {
@@ -89,7 +87,11 @@ impl Switch {
impl Switch {
pub async fn start_(token: String, mac_address: String) -> Result<Self> {
let server_address = "nat1.wherewego.top:29876".to_socket_addrs().unwrap().next().unwrap();
let server_address = "nat1.wherewego.top:29876"
.to_socket_addrs()
.unwrap()
.next()
.unwrap();
let mut port = 101 as u16;
let udp = loop {
match UdpSocket::bind(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::from(0), port))) {
@@ -100,14 +102,15 @@ impl Switch {
if e.kind() == io::ErrorKind::AddrInUse {
port += 1;
} else {
log::error!("创建udp失败 {:?}",e);
log::error!("创建udp失败 {:?}", e);
return Err(Error::Stop("udp bind error".to_string()));
}
}
}
};
//注册
let response = handle::registration_handler::registration(&udp, server_address, token, mac_address)?;
let response =
handle::registration_handler::registration(&udp, server_address, token, mac_address)?;
{
let ip_list = response
.virtual_ip_list
@@ -121,8 +124,10 @@ impl Switch {
let virtual_ip = Ipv4Addr::from(response.virtual_ip);
let virtual_gateway = Ipv4Addr::from(response.virtual_gateway);
let virtual_netmask = Ipv4Addr::from(response.virtual_netmask);
let (status_sender, status_receiver) = tokio::sync::watch::channel(ApplicationStatus::Starting);
let current_device = CurrentDeviceInfo::new(virtual_ip, virtual_gateway, virtual_netmask, server_address);
let (status_sender, status_receiver) =
tokio::sync::watch::channel(ApplicationStatus::Starting);
let current_device =
CurrentDeviceInfo::new(virtual_ip, virtual_gateway, virtual_netmask, server_address);
let wait_group = WaitGroup::new();
//心跳线程
{
@@ -130,7 +135,8 @@ impl Switch {
let wait_group1 = wait_group.clone();
handle::heartbeat_handler::start(status_receiver.clone(), udp, current_device, || {
drop(wait_group1);
}).await;
})
.await;
}
//初始化nat数据
handle::init_nat_info(response.public_ip, response.public_port as u16);
@@ -138,7 +144,8 @@ impl Switch {
let (tun_writer, tun_reader) =
tun_device::create_tun(virtual_ip, virtual_netmask, virtual_gateway)?;
// 打洞数据通道
let (punch_sender, cone_receiver, req_symmetric_receiver, res_symmetric_receiver) = handle::punch_handler::bounded();
let (punch_sender, cone_receiver, req_symmetric_receiver, res_symmetric_receiver) =
handle::punch_handler::bounded();
//udp数据处理
{
// 低优先级的udp数据通道
@@ -155,51 +162,74 @@ impl Switch {
|| {
drop(wait_group1);
},
).await;
)
.await;
let udp1 = udp.try_clone()?;
let wait_group1 = wait_group.clone();
handle::udp_recv_handler::udp_other_recv_start(status_receiver.clone(), udp1,
receiver, current_device, punch_sender,
|| {
drop(wait_group1);
}).await;
handle::udp_recv_handler::udp_other_recv_start(
status_receiver.clone(),
udp1,
receiver,
current_device,
punch_sender,
|| {
drop(wait_group1);
},
)
.await;
}
//打洞处理
{
let udp1 = udp.try_clone()?;
let wait_group1 = wait_group.clone();
handle::punch_handler::cone_handler_start(status_receiver.clone(),
cone_receiver, udp1,
current_device,
|| {
drop(wait_group1);
}).await;
handle::punch_handler::cone_handler_start(
status_receiver.clone(),
cone_receiver,
udp1,
current_device,
|| {
drop(wait_group1);
},
)
.await;
let udp1 = udp.try_clone()?;
let wait_group1 = wait_group.clone();
handle::punch_handler::req_symmetric_handler_start(status_receiver.clone(),
req_symmetric_receiver, udp1,
current_device,
|| {
drop(wait_group1);
}).await;
handle::punch_handler::req_symmetric_handler_start(
status_receiver.clone(),
req_symmetric_receiver,
udp1,
current_device,
|| {
drop(wait_group1);
},
)
.await;
let udp1 = udp.try_clone()?;
let wait_group1 = wait_group.clone();
handle::punch_handler::res_symmetric_handler_start(status_receiver.clone(),
res_symmetric_receiver,
udp1,
current_device,
|| {
drop(wait_group1);
}).await;
handle::punch_handler::res_symmetric_handler_start(
status_receiver.clone(),
res_symmetric_receiver,
udp1,
current_device,
|| {
drop(wait_group1);
},
)
.await;
}
//tun数据处理
{
let wait_group1 = wait_group.clone();
handle::tun_handler::handler_start(status_receiver.clone(), udp,
tun_reader, current_device,
|| {
drop(wait_group1);
}).await;
handle::tun_handler::handler_start(
status_receiver.clone(),
udp,
tun_reader,
current_device,
|| {
drop(wait_group1);
},
)
.await;
}
Ok(Switch {
current_device,
@@ -208,4 +238,4 @@ impl Switch {
runtime: None,
})
}
}
}
switch/src/nat/check.rs
+2 -2
View File
@@ -1,7 +1,7 @@
use std::{io, thread};
use std::collections::HashSet;
use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket};
use std::time::Duration;
use std::{io, thread};
use crate::proto::message::NatType;
@@ -154,4 +154,4 @@ fn nat_test_run() {
let udp = UdpSocket::bind("0.0.0.0:101").unwrap();
let print = public_ip_list_(&udp).unwrap();
println!("{:?}", print);
}
}
switch/src/nat/mod.rs
+1 -1
View File
@@ -1 +1 @@
pub mod check;
pub mod check;
switch/src/protocol/control_packet.rs
-1
View File
@@ -70,7 +70,6 @@ pub struct PongPacket<B> {
buffer: B,
}
impl<B: AsRef<[u8]>> PingPacket<B> {
pub fn new(buffer: B) -> Result<PingPacket<B>> {
let len = buffer.as_ref().len();
switch/src/tun_device/linux.rs
+1 -2
View File
@@ -5,8 +5,8 @@ use std::os::unix::process::CommandExt;
use std::process::Command;
use bytes::BufMut;
use tun::Device;
use tun::platform::posix::{Reader, Writer};
use tun::Device;
use crate::tun_device::{TunReader, TunWriter};
@@ -36,4 +36,3 @@ pub fn create_tun(
TunReader(reader, packet_information),
))
}
switch/src/tun_device/mac.rs
+9 -4
View File
@@ -5,8 +5,8 @@ use std::os::unix::process::CommandExt;
use std::process::Command;
use bytes::BufMut;
use tun::Device;
use tun::platform::posix::{Reader, Writer};
use tun::Device;
use crate::tun_device::{TunReader, TunWriter};
@@ -36,7 +36,10 @@ pub fn create_tun(
.output()
.expect("sh exec error!");
if !up_eth_out.status.success() {
return Err(crate::error::Error::Stop(format!("设置地址失败:{:?}", up_eth_out)));
return Err(crate::error::Error::Stop(format!(
"设置地址失败:{:?}",
up_eth_out
)));
}
let if_config_out = Command::new("sh")
.arg("-c")
@@ -44,7 +47,10 @@ pub fn create_tun(
.output()
.expect("sh exec error!");
if !if_config_out.status.success() {
return Err(crate::error::Error::Stop(format!("设置路由失败:{:?}", if_config_out)));
return Err(crate::error::Error::Stop(format!(
"设置路由失败:{:?}",
if_config_out
)));
}
// println!("{:?}", if_config_out);
// let cmd_str: String = " ifconfig|grep flags=8051|awk -F ':' '{print $1}'|tail -1".to_string();
@@ -65,4 +71,3 @@ pub fn create_tun(
TunReader(reader, packet_information),
))
}
switch/src/tun_device/mod.rs
+5 -5
View File
@@ -1,18 +1,18 @@
#[cfg(any(target_os = "linux",target_os = "android"))]
#[cfg(any(target_os = "linux", target_os = "android"))]
pub use linux::create_tun;
#[cfg(target_os = "macos")]
pub use mac::create_tun;
#[cfg(any(unix))]
pub use unix::{TunReader, TunWriter};
#[cfg(target_os = "windows")]
pub use windows::{TunReader, TunWriter};
#[cfg(target_os = "windows")]
pub use windows::create_tun;
#[cfg(target_os = "windows")]
pub use windows::{TunReader, TunWriter};
#[cfg(any(target_os = "linux", target_os = "android"))]
pub mod linux;
#[cfg(target_os = "macos")]
pub mod mac;
#[cfg(any(target_os = "linux",target_os = "android"))]
pub mod linux;
#[cfg(any(unix))]
pub mod unix;
#[cfg(target_os = "windows")]