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

v1.1

Browse Source
This commit is contained in:
lubeilin
2023-03-10 23:01:57 +08:00
parent ebf84db204
commit 5b2c2435d5
39 changed files with 2493 additions and 2552 deletions
Download Patch File Download Diff File Expand all files Collapse all files
switch/src/handle/heartbeat_handler.rs
+79 -55
View File
@@ -1,80 +1,104 @@
use std::net::{SocketAddr, UdpSocket};
use std::{io, thread};
use std::net::Ipv4Addr;
use std::sync::Arc;
use std::time::Duration;
use chrono::Local;
use tokio::sync::watch::Receiver;
use tokio::time::sleep;
use crossbeam::atomic::AtomicCell;
use parking_lot::Mutex;
use rand::prelude::SliceRandom;
use nat_traversal::channel::Route;
use crate::error::*;
use crate::handle::{ApplicationStatus, DIRECT_ROUTE_TABLE};
use nat_traversal::channel::sender::Sender;
use nat_traversal::idle::Idle;
use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo};
use crate::protocol::{control_packet, MAX_TTL, 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,
{
tokio::spawn(async move {
match handle_loop(status_watch, udp, cur_info.connect_server).await {
Ok(_) => {}
Err(e) => {
log::warn!("{:?}", e)
}
pub fn start_idle(idle: Idle<Ipv4Addr>, sender: Sender<Ipv4Addr>) {
thread::spawn(move || {
if let Err(e) = start_idle_(idle, sender) {
log::info!("空闲检测线程停止:{:?}",e);
}
stop_fn();
});
}
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)];
let mut net_packet = NetPacket::new(&mut buf)?;
fn start_idle_(idle: Idle<Ipv4Addr>, sender: Sender<Ipv4Addr>) -> io::Result<()> {
loop {
let (idle_status, peer_ip, route) = idle.next_idle()?;
log::warn!("peer_ip:{:?},route:{:?},idle_status:{:?}",peer_ip,route,idle_status);
sender.remove_route(&peer_ip);
}
}
pub fn start_heartbeat(sender: Sender<Ipv4Addr>, device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) {
thread::spawn(move || {
if let Err(e) = start_heartbeat_(sender, device_list, current_device) {
log::info!("空闲检测线程停止:{:?}",e);
}
});
}
fn start_heartbeat_(sender: Sender<Ipv4Addr>, device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) -> io::Result<()> {
let mut net_packet = NetPacket::new([0u8; 16])?;
net_packet.set_version(Version::V1);
net_packet.set_protocol(Protocol::Control);
net_packet.set_transport_protocol(control_packet::Protocol::Ping.into());
net_packet.set_ttl(255);
net_packet.first_set_ttl(MAX_TTL);
let mut count = 0;
loop {
let current_time = Local::now().timestamp_millis();
let current_device = current_device.load();
net_packet.set_source(current_device.virtual_ip());
{
let current_time = Local::now().timestamp_millis() as u16;
let mut ping = PingPacket::new(net_packet.payload_mut())?;
ping.set_time(current_time);
let epoch = { DEVICE_LIST.lock().0 };
let epoch = { device_list.lock().0 };
ping.set_epoch(epoch);
}
let _ = udp.send_to(net_packet.buffer(), server_addr);
// 不clone会死锁？
for x in DIRECT_ROUTE_TABLE.clone().iter() {
let virtual_ip = x.key().clone();
let route = x.value().clone();
drop(x);
if current_time - route.recv_time <= MAX_INTERVAL {
let _ = udp.send_to(net_packet.buffer(), route.address);
} else {
DIRECT_ROUTE_TABLE.remove_if(&virtual_ip, |_, route| {
current_time - route.recv_time > MAX_INTERVAL
});
if count % 7 == 0 {
let mut route_list: Option<Vec<(Ipv4Addr, Route)>> = None;
let peer_list = device_list.lock().1.clone();
for peer in peer_list {
net_packet.first_set_ttl(MAX_TTL);
net_packet.set_destination(peer.virtual_ip);
if sender.send_to_id(net_packet.buffer(), &peer.virtual_ip).is_err() {
//没有路由则发送到网关
let _ = sender.send_to_addr(net_packet.buffer(), current_device.connect_server);
//再随机发送到其他地址，看有没有客户端符合转发条件
let route_list = route_list.get_or_insert_with(|| {
let mut l = sender.route_list();
l.shuffle(&mut rand::thread_rng());
l
});
let mut num = 0;
net_packet.first_set_ttl(2);
for (peer_ip, route) in route_list.iter() {
if peer_ip != &peer.virtual_ip && route.metric == 1 {
let _ = sender.send_to_route(net_packet.buffer(), &route.route_key());
num += 1;
}
if num >= 3 {
break;
}
}
}
}
}
tokio::select! {
_ = sleep(Duration::from_millis(INTERVAL))=>{
net_packet.set_destination(current_device.virtual_gateway());
if let Err(e) = sender.send_to_addr(net_packet.buffer(), current_device.connect_server) {
log::warn!("connect_server:{:?},e:{:?}",current_device.connect_server,e);
}
status = status_watch.changed() =>{
status?;
if *status_watch.borrow() != ApplicationStatus::Starting{
return Ok(())
} else {
for (peer_ip, route) in sender.route_list().iter() {
net_packet.set_destination(*peer_ip);
if let Err(e) = sender.send_to_route(net_packet.buffer(), &route.route_key()) {
log::warn!("peer_ip:{:?},route:{:?},e:{:?}",peer_ip,route,e);
}
}
}
count += 1;
thread::sleep(Duration::from_secs(5));
}
}
}
switch/src/handle/mod.rs
+19 -131
View File
@@ -1,38 +1,17 @@
use std::net::{Ipv4Addr, SocketAddr};
use std::sync::atomic::AtomicI64;
use std::time::Duration;
use chrono::Local;
use dashmap::DashMap;
use lazy_static::lazy_static;
use moka::sync::Cache;
use parking_lot::{const_mutex, Mutex};
use crate::proto::message::NatType;
pub mod heartbeat_handler;
pub mod punch_handler;
pub mod registration_handler;
pub mod tun_handler;
pub mod udp_recv_handler;
lazy_static! {
/// 0. 机器纪元，每一次上线或者下线都会增1，由服务端维护，用于感知网络中机器变化
/// 服务端和客户端的不一致，则服务端会推送新的设备列表
/// 1. 网络中的虚拟ip列表
pub static ref DEVICE_LIST:Mutex<(u32,Vec<PeerDeviceInfo>)> = const_mutex((0,Vec::new()));
/// 服务器延迟
pub static ref SERVER_RT:AtomicI64 = AtomicI64::new(-1);
/// id
pub static ref ID:AtomicI64 = AtomicI64::new(0);
/// 直连路由表
pub static ref DIRECT_ROUTE_TABLE:DashMap<Ipv4Addr,Route> = DashMap::new();
/// 地址映射
pub static ref ADDR_TABLE:Cache<SocketAddr,Ipv4Addr> = Cache::builder()
.time_to_idle(Duration::from_secs(60*5)).build();
/// 当前设备的nat信息
pub static ref NAT_INFO:Mutex<Option<NatInfo>> = const_mutex(None);
static ref NAT_TEST_ADDRESS:Mutex<Vec<SocketAddr>> = const_mutex(Vec::new());
pub mod recv_handler;
/// 是否在一个网段
fn check_dest(dest: Ipv4Addr, virtual_netmask: Ipv4Addr, virtual_network: Ipv4Addr) -> bool {
u32::from_be_bytes(dest.octets()) & u32::from_be_bytes(virtual_netmask.octets())
== u32::from_be_bytes(virtual_network.octets())
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PeerDeviceInfo {
pub virtual_ip: Ipv4Addr,
@@ -74,82 +53,15 @@ impl From<u8> for PeerDeviceStatus {
}
}
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub enum ApplicationStatus {
Starting,
Stopping,
}
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub enum ConnectStatus {
Connecting,
Connected,
}
impl Into<u8> for ConnectStatus {
fn into(self) -> u8 {
match self {
ConnectStatus::Connecting => 0,
ConnectStatus::Connected => 1,
}
}
}
#[derive(Clone, Debug)]
pub struct NatInfo {
pub public_ips: Vec<u32>,
pub public_port: u16,
pub public_port_range: u16,
pub nat_type: NatType,
}
impl NatInfo {
pub fn new(
public_ips: Vec<u32>,
public_port: u16,
public_port_range: u16,
nat_type: NatType,
) -> Self {
Self {
public_ips,
public_port,
public_port_range,
nat_type,
}
}
}
pub fn init_nat_test_addr(addrs: Vec<SocketAddr>) {
NAT_TEST_ADDRESS.lock().extend_from_slice(&addrs);
}
/// 初始化nat信息
pub fn init_nat_info(public_ip: u32, public_port: u16) {
let addrs = NAT_TEST_ADDRESS.lock().clone();
match crate::nat::check::public_ip_list(&addrs) {
Ok((nat_type, ips, port_range)) => {
let mut public_ips = Vec::new();
public_ips.push(public_ip);
for ip in ips {
let ip = u32::from_be_bytes(ip.octets());
if ip != public_ip {
public_ips.push(ip);
}
}
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);
}
Err(e) => {
println!("获取nat数据失败，将无法进行udp打洞:{:?}", e);
}
}
}
#[derive(Copy, Clone, Debug)]
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub struct CurrentDeviceInfo {
pub virtual_ip: Ipv4Addr,
virtual_ip: Ipv4Addr,
pub virtual_gateway: Ipv4Addr,
pub virtual_netmask: Ipv4Addr,
//网络地址
@@ -182,40 +94,16 @@ impl CurrentDeviceInfo {
connect_server,
}
}
}
#[derive(Clone, Debug)]
pub struct Route {
pub route_type: RouteType,
pub address: SocketAddr,
//用心跳探测延迟，收包时更新
pub rt: i64,
//收包时更新，如果太久没有收到消息则剔除
pub recv_time: i64,
}
#[derive(Copy, Clone, Debug, Eq, PartialEq)]
pub enum RouteType {
ServerRelay,
P2P,
}
impl Into<u8> for RouteType {
fn into(self) -> u8 {
match self {
RouteType::ServerRelay => 0,
RouteType::P2P => 1,
}
#[inline]
pub fn virtual_ip(&self) -> Ipv4Addr {
self.virtual_ip
}
#[inline]
pub fn virtual_gateway(&self) -> Ipv4Addr {
self.virtual_gateway
}
}
impl Route {
pub fn new(address: SocketAddr) -> Self {
Self {
route_type: RouteType::P2P,
address,
rt: -1,
recv_time: Local::now().timestamp_millis(),
}
}
}
switch/src/handle/punch_handler.rs
+113 -375
View File
@@ -1,402 +1,140 @@
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4, UdpSocket};
use std::time::Duration;
use std::{io, thread};
use std::net::{IpAddr, Ipv4Addr};
use std::sync::Arc;
use std::time::Duration;
use crossbeam::atomic::AtomicCell;
use parking_lot::Mutex;
use protobuf::Message;
use rand::prelude::SliceRandom;
use tokio::sync::mpsc::error::TrySendError;
use tokio::sync::mpsc::{Receiver, Sender};
use tokio::sync::watch;
use nat_traversal::channel::sender::Sender;
use nat_traversal::punch::{NatInfo, NatType, Punch};
use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo};
use crate::nat::NatTest;
use crate::proto::message::{PunchInfo, PunchNatType};
use crate::protocol::{control_packet, MAX_TTL, NetPacket, Protocol, turn_packet, Version};
use crate::error::*;
use crate::handle::{ApplicationStatus, DIRECT_ROUTE_TABLE};
use crate::proto::message::{NatType, Punch};
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};
/// 每一种类型一个通道，减少相互干扰
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),
)
}
pub struct ConeReceiver(Receiver<Punch>);
pub struct ReqSymmetricReceiver(Receiver<Punch>);
pub struct ResSymmetricReceiver(Receiver<Punch>);
#[derive(Clone)]
pub struct PunchSender {
cone_sender: Sender<Punch>,
req_symmetric_sender: Sender<Punch>,
res_symmetric_sender: Sender<Punch>,
}
impl PunchSender {
pub fn new(
cone_sender: Sender<Punch>,
req_symmetric_sender: Sender<Punch>,
res_symmetric_sender: Sender<Punch>,
) -> Self {
Self {
cone_sender,
req_symmetric_sender,
res_symmetric_sender,
pub fn start_cone(punch: Punch<Ipv4Addr>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) {
thread::spawn(move || {
if let Err(e) = start_(true, punch, current_device) {
log::warn!("锥形网络打洞处理线程停止 {:?}",e);
}
}
// pub fn send(&self, punch: Punch) -> std::result::Result<(), SendError<Punch>> {
// match punch.nat_type.enum_value_or_default() {
// NatType::Symmetric => {
// if punch.reply {
// // 为true表示回应，也就是主动发起的打洞操作
// self.res_symmetric_sender.blocking_send(punch)
// } else {
// self.req_symmetric_sender.blocking_send(punch)
// }
// }
// NatType::Cone => {
// self.cone_sender.blocking_send(punch)
// }
// }
// }
pub fn try_send(&self, punch: Punch) -> std::result::Result<(), TrySendError<Punch>> {
match punch.nat_type.enum_value_or_default() {
NatType::Symmetric => {
if punch.reply {
// 为true表示回应，也就是主动发起的打洞操作
self.res_symmetric_sender.try_send(punch)
} else {
self.req_symmetric_sender.try_send(punch)
}
});
}
pub fn start_symmetric(punch: Punch<Ipv4Addr>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) {
thread::spawn(move || {
if let Err(e) = start_(false, punch, current_device) {
log::warn!("对称网络打洞处理线程停止 {:?}",e);
}
});
}
fn start_(is_cone: bool, mut punch: Punch<Ipv4Addr>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) -> io::Result<()> {
let mut packet = NetPacket::new([0u8; 12])?;
packet.set_version(Version::V1);
packet.first_set_ttl(1);
packet.set_protocol(Protocol::Control);
packet.set_transport_protocol(control_packet::Protocol::PunchRequest.into());
loop {
let (peer_ip, nat_info) = if is_cone {
punch.next_cone(None)?
} else {
punch.next_symmetric(None)?
};
if let Some(route) = punch.sender().route(&peer_ip) {
if route.metric == 1 {
//直连地址不需要打洞
continue;
}
NatType::Cone => self.cone_sender.try_send(punch),
}
packet.set_source(current_device.load().virtual_ip());
packet.set_destination(peer_ip);
log::info!("发起打洞，目标:{:?},{:?}",peer_ip,nat_info);
if let Err(e) = punch.punch(packet.buffer(), peer_ip, nat_info) {
log::warn!("peer_ip:{:?},e:{:?}",peer_ip,e);
}
}
}
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);
if DIRECT_ROUTE_TABLE.contains_key(&dest) {
continue;
pub fn start_punch(nat_test: NatTest, device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>, sender: Sender<Ipv4Addr>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) {
thread::spawn(move || {
if let Err(e) = start_punch_(nat_test, device_list, sender, current_device) {
log::warn!("对称网络打洞处理线程停止 {:?}",e);
}
// println!("punch {:?}", punch);
match punch.nat_type.enum_value_or_default() {
NatType::Symmetric => {
// 假设绑定n个端口，通过NAT对外映射出n个 公网ip:公网端口，随机尝试k次的情况下
// 猜中的概率 p = 1-((65535-n)/65535)*((65535-n-1)/(65535-1))*...*((65535-n-k+1)/(65535-k+1))
// n取76，k取600，猜中的概率就超过50%了
// 前提 自己是锥形网络，否则猜中了也通信不了
let mut send_f = |min_port: u16, max_port: u16, k: usize| -> io::Result<()> {
let mut nums: Vec<u16> = (min_port..max_port).collect();
nums.push(max_port);
let mut rng = rand::thread_rng();
nums.shuffle(&mut rng);
for pub_ip in &punch.public_ip_list {
let pub_ip = Ipv4Addr::from(*pub_ip);
for port in &nums[..k] {
udp.send_to(buf, SocketAddr::V4(SocketAddrV4::new(pub_ip, *port)))?;
select_sleep(&mut counter);
}
}
Ok(())
};
if punch.public_port_range < 600 {
//端口变化不大时，在预测的范围内随机发送
let min_port = if punch.public_port > punch.public_port_range {
punch.public_port - punch.public_port_range
} else {
1
};
let max_port = if punch.public_port + punch.public_port_range > 65535 {
65535
} else {
punch.public_port + punch.public_port_range
};
let k = if max_port - min_port + 1 > 60 {
60
} else {
max_port - min_port + 1
};
send_f(min_port as u16, max_port as u16, k as usize)?;
});
}
fn start_punch_(nat_test: NatTest, device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>, sender: Sender<Ipv4Addr>, current_device: Arc<AtomicCell<CurrentDeviceInfo>>) -> crate::Result<()> {
loop {
if sender.is_close() {
return Ok(());
}
let current_device = current_device.load();
let nat_info = nat_test.nat_info();
{
let mut list = device_list.lock().clone().1;
list.shuffle(&mut rand::thread_rng());
let mut count = 0;
for info in list {
if info.virtual_ip <= current_device.virtual_ip {
continue;
}
// 全端口范围，随机取600个端口发送
send_f(1, 65535, 600)?;
if let Some(route) = sender.route(&info.virtual_ip) {
if route.metric == 1 {
//直连地址不需要打洞
continue;
}
}
count += 1;
if count > 3 {
break;
}
let buf = punch_packet(current_device.virtual_ip(), &nat_info, info.virtual_ip)?;
sender.send_to_addr(&buf, current_device.connect_server)?;
}
}
match nat_info.nat_type {
NatType::Symmetric => {
thread::sleep(Duration::from_secs(28));
}
NatType::Cone => {
for pub_ip in punch.public_ip_list {
udp.send_to(
buf,
SocketAddr::V4(SocketAddrV4::new(
Ipv4Addr::from(pub_ip),
punch.public_port as u16,
)),
)?;
select_sleep(&mut counter);
}
}
}
}
Ok(())
}
/// 给对称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,
{
let receiver = receiver.0;
tokio::spawn(async move {
match handle_loop(status_watch, receiver, udp, cur_info).await {
Ok(_) => {}
Err(e) => {
log::warn!("{:?}", e)
}
}
stop_fn()
});
}
// pub fn req_symmetric_handle_loop(
// receiver: ReqSymmetricReceiver,
// udp: UdpSocket,
// cur_info: CurrentDeviceInfo,
// ) -> Result<()> {
// let receiver = receiver.0;
// handle_loop(receiver, udp, cur_info)
// }
/// 给对称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,
{
let receiver = receiver.0;
tokio::spawn(async move {
match res_symmetric_handle_loop(status_watch, receiver, udp, cur_info).await {
Ok(_) => {}
Err(e) => {
log::warn!("{:?}", e)
}
}
stop_fn()
});
}
async fn res_symmetric_handle_loop(
mut status_watch: watch::Receiver<ApplicationStatus>,
mut receiver: Receiver<Punch>,
udp: UdpSocket,
cur_info: CurrentDeviceInfo,
) -> Result<()> {
let mut buf = [0u8; 12];
let mut packet = NetPacket::new(&mut buf)?;
packet.set_version(Version::V1);
packet.set_ttl(255);
packet.set_protocol(Protocol::Control);
packet.set_transport_protocol(control_packet::Protocol::PunchRequest.into());
{
let mut punch_packet = PunchRequestPacket::new(packet.payload_mut())?;
punch_packet.set_source(cur_info.virtual_ip);
}
loop {
tokio::select! {
rs = tokio::time::timeout(Duration::from_secs(20), receiver.recv()) =>{
match rs {
Ok(punch) => {
if let Some(punch) = punch{
let mut list = Vec::new();
list.push(punch);
loop {
match receiver.try_recv() {
Ok(punch) => {
list.push(punch);
}
Err(_) => {
break;
}
}
}
if let Err(e) = handle(&status_watch,&udp, list, packet.buffer()) {
log::warn!("{:?}",e)
}
}else {
return Err(Error::Stop("打洞线程通道关闭".to_string()));
}
}
Err(_) => {
punch_request_handle(&udp, &cur_info)?;
}
}
}
status = status_watch.changed() =>{
status?;
if *status_watch.borrow() != ApplicationStatus::Starting{
return Ok(())
}
thread::sleep(Duration::from_secs(20));
}
}
}
}
/// 给锥形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,
{
let receiver = receiver.0;
tokio::spawn(async move {
match handle_loop(status_watch, receiver, udp, cur_info).await {
Ok(_) => {}
Err(e) => {
log::warn!("{:?}", e)
}
}
stop_fn();
});
}
async fn handle_loop(
mut status_watch: watch::Receiver<ApplicationStatus>,
mut receiver: Receiver<Punch>,
udp: UdpSocket,
cur_info: CurrentDeviceInfo,
) -> Result<()> {
let mut buf = [0u8; 12];
let mut packet = NetPacket::new(&mut buf)?;
packet.set_version(Version::V1);
packet.set_ttl(255);
packet.set_protocol(Protocol::Control);
packet.set_transport_protocol(control_packet::Protocol::PunchRequest.into());
{
let mut punch_packet = PunchRequestPacket::new(packet.payload_mut())?;
punch_packet.set_source(cur_info.virtual_ip);
}
loop {
tokio::select! {
punch = receiver.recv() =>{
if let Some(punch) = punch{
let mut list = Vec::new();
list.push(punch);
loop {
match receiver.try_recv() {
Ok(punch) => {
list.push(punch);
}
Err(_) => {
break;
}
}
}
if let Err(e) = handle(&status_watch,&udp, list, packet.buffer()) {
log::warn!("{:?}",e)
}
}else {
return Err(Error::Stop("打洞线程通道关闭".to_string()));
}
}
status = status_watch.changed() =>{
status?;
if *status_watch.borrow() != ApplicationStatus::Starting{
return Ok(())
}
}
}
}
}
fn select_sleep(counter: &mut u64) {
*counter += 1;
if *counter & 10 == 10 {
thread::sleep(Duration::from_millis(2));
} else {
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::warn!("发送打洞数据失败 {:?}", e)
}
Ok(())
} else {
Err(Error::Stop("未初始化nat信息".to_string()))
}
}
fn send_punch(udp: &UdpSocket, cur_info: &CurrentDeviceInfo, nat_info: NatInfo) -> Result<()> {
let lock = DEVICE_LIST.lock();
let list = lock.1.clone();
drop(lock);
for peer_info in list {
let ip = peer_info.virtual_ip;
//只向ip比自己大的发起打洞，避免双方同时发起打洞浪费流量
if ip > cur_info.virtual_ip && !DIRECT_ROUTE_TABLE.contains_key(&ip) {
log::info!("发起打洞 {:?}, peer_info:{:?}", nat_info, peer_info);
let bytes = punch_packet(cur_info.virtual_ip, nat_info.clone(), ip)?;
udp.send_to(&bytes, cur_info.connect_server)?;
}
}
Ok(())
}
fn punch_packet(virtual_ip: Ipv4Addr, nat_info: NatInfo, dest: Ipv4Addr) -> Result<Vec<u8>> {
let mut punch_reply = Punch::new();
pub fn punch_packet(virtual_ip: Ipv4Addr, nat_info: &NatInfo, dest: Ipv4Addr) -> crate::Result<Vec<u8>> {
let mut punch_reply = PunchInfo::new();
punch_reply.reply = false;
punch_reply.virtual_ip = u32::from_be_bytes(virtual_ip.octets());
punch_reply.public_ip_list = nat_info.public_ips;
punch_reply.public_ip_list = nat_info.public_ips.iter().map(|i| {
match i {
IpAddr::V4(ip) => {
u32::from_be_bytes(ip.octets())
}
IpAddr::V6(_) => {
panic!()
}
}
}).collect();
punch_reply.public_port = nat_info.public_port as u32;
punch_reply.public_port_range = nat_info.public_port_range as u32;
punch_reply.nat_type = protobuf::EnumOrUnknown::new(nat_info.nat_type);
punch_reply.local_ip = match nat_info.local_ip {
IpAddr::V4(ip) => u32::from_be_bytes(ip.octets()),
IpAddr::V6(_) => {
panic!()
}
};
punch_reply.local_port = nat_info.local_port as u32;
punch_reply.nat_type = protobuf::EnumOrUnknown::new(PunchNatType::from(nat_info.nat_type));
let bytes = punch_reply.write_to_bytes()?;
let mut net_packet = NetPacket::new(vec![0u8; 4 + 8 + bytes.len()])?;
let mut net_packet = NetPacket::new(vec![0u8; 12 + 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_ttl(255);
let mut turn_packet = TurnPacket::new(net_packet.payload_mut())?;
turn_packet.set_source(virtual_ip);
turn_packet.set_destination(dest);
turn_packet.set_payload(&bytes);
net_packet.first_set_ttl(MAX_TTL);
net_packet.set_source(virtual_ip);
net_packet.set_destination(dest);
net_packet.set_payload(&bytes);
Ok(net_packet.into_buffer())
}
}
switch/src/handle/recv_handler.rs
+385
View File
@@ -0,0 +1,385 @@
use std::{io, thread};
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Arc;
use chrono::Local;
use crossbeam::atomic::AtomicCell;
use crossbeam_skiplist::SkipMap;
use parking_lot::Mutex;
use protobuf::Message;
use nat_traversal::channel::{Channel, Route, RouteKey};
use nat_traversal::punch::NatInfo;
use packet::icmp::{icmp, Kind};
use packet::ip::ipv4;
use packet::ip::ipv4::packet::IpV4Packet;
use crate::error::Error;
use crate::handle::{check_dest, ConnectStatus, CurrentDeviceInfo, PeerDeviceInfo};
use crate::handle::registration_handler::Register;
use crate::nat::NatTest;
use crate::proto::message::{DeviceList, PunchInfo, PunchNatType, RegistrationResponse};
use crate::protocol::{control_packet, MAX_TTL, NetPacket, Protocol, service_packet, turn_packet, Version};
use crate::protocol::control_packet::ControlPacket;
use crate::protocol::error_packet::InErrorPacket;
use crate::tun_device::TunWriter;
pub fn start(mut handler: RecvHandler) {
thread::spawn(move || {
let mut buf = [0; 4096];
loop {
match handler.channel.recv_from(&mut buf, None) {
Ok((len, route)) => {
if let Err(e) = handler.handle(&mut buf[..len], &route) {
log::warn!("数据处理失败:{:?},e:{:?}",route,e);
if let Error::Stop(_) = e {
let _ = handler.channel.close();
break;
}
}
}
Err(e) => {
log::warn!("{:?}",e);
// 检查关闭状态
if handler.channel.is_close() {
break;
}
}
}
}
});
}
pub struct RecvHandler {
channel: Channel<Ipv4Addr>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
register: Arc<Register>,
nat_test: NatTest,
tun_writer: TunWriter,
connect_status: Arc<AtomicCell<ConnectStatus>>,
peer_nat_info_map: Arc<SkipMap<Ipv4Addr, NatInfo>>,
}
impl RecvHandler {
pub fn new(channel: Channel<Ipv4Addr>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
register: Arc<Register>,
nat_test: NatTest,
tun_writer: TunWriter,
connect_status: Arc<AtomicCell<ConnectStatus>>,
peer_nat_info_map: Arc<SkipMap<Ipv4Addr, NatInfo>>,
) -> Self {
Self {
channel,
current_device,
device_list,
register,
nat_test,
tun_writer,
connect_status,
peer_nat_info_map,
}
}
pub fn try_clone(&self) -> io::Result<Self> {
Ok(Self {
channel: self.channel.try_clone()?,
current_device: self.current_device.clone(),
device_list: self.device_list.clone(),
register: self.register.clone(),
nat_test: self.nat_test.clone(),
tun_writer: self.tun_writer.clone(),
connect_status: self.connect_status.clone(),
peer_nat_info_map: self.peer_nat_info_map.clone(),
})
}
}
impl RecvHandler {
fn handle(&self, buf: &mut [u8], route_key: &RouteKey) -> crate::Result<()> {
let mut net_packet = NetPacket::new(buf)?;
if net_packet.ttl() == 0 {
return Ok(());
}
let source = net_packet.source();
let current_device = self.current_device.load();
if source == current_device.virtual_ip() {
return Ok(());
}
let destination = net_packet.destination();
if current_device.virtual_ip() != destination && self.connect_status.load() == ConnectStatus::Connected {
if !check_dest(source, current_device.virtual_netmask, current_device.virtual_network) {
log::warn!("转发数据，源地址错误:{:?},当前网络:{:?},route_key:{:?}",source,current_device.virtual_network,route_key);
return Ok(());
}
if !check_dest(destination, current_device.virtual_netmask, current_device.virtual_network) {
log::warn!("转发数据，目的地址错误:{:?},当前网络:{:?},route_key:{:?}",destination,current_device.virtual_network,route_key);
return Ok(());
}
let ttl = net_packet.ttl();
if ttl > 1 {
// 转发
net_packet.set_ttl(ttl - 1);
if let Some(route) = self.channel.route(&destination) {
if route.metric <= net_packet.ttl() {
self.channel.send_to_route(net_packet.buffer(), &route.route_key())?;
}
} else if (ttl > 2 || destination == current_device.virtual_gateway())
&& source != current_device.virtual_gateway() {
//网关默认要转发一次，生存时间不够的发到网关也会被丢弃
self.channel.send_to_addr(net_packet.buffer(), current_device.connect_server)?;
}
}
return Ok(());
}
match net_packet.protocol() {
Protocol::Ipv4Turn => {
let mut ipv4 = IpV4Packet::new(net_packet.payload_mut())?;
if ipv4.protocol() == ipv4::protocol::Protocol::Icmp {
let mut icmp_packet = icmp::IcmpPacket::new(ipv4.payload_mut())?;
if icmp_packet.kind() == Kind::EchoRequest {
//开启ping
icmp_packet.set_kind(Kind::EchoReply);
icmp_packet.update_checksum();
ipv4.set_source_ip(destination);
ipv4.set_destination_ip(source);
ipv4.update_checksum();
net_packet.set_source(destination);
net_packet.set_destination(source);
self.channel.send_to_route(net_packet.buffer(), route_key)?;
return Ok(());
}
}
self.tun_writer.write(net_packet.payload())?;
}
Protocol::Service => {
self.service(current_device, source, net_packet, route_key)?;
}
Protocol::Error => {
self.error(current_device, source, net_packet, route_key)?;
}
Protocol::Control => {
self.control(current_device, source, net_packet, route_key)?;
}
Protocol::OtherTurn => {
self.other_turn(current_device, source, net_packet, route_key)?;
}
Protocol::UnKnow(e) => {
log::info!("不支持的协议:{}",e);
}
}
Ok(())
}
fn service(&self, current_device: CurrentDeviceInfo, source: Ipv4Addr, net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> {
if route_key.addr != current_device.connect_server || source != current_device.virtual_gateway() {
return Ok(());
}
match service_packet::Protocol::from(net_packet.transport_protocol()) {
service_packet::Protocol::RegistrationRequest => {}
service_packet::Protocol::RegistrationResponse => {
let response = RegistrationResponse::parse_from_bytes(net_packet.payload())?;
let local_addr = self.channel.local_addr()?;
let local_ip = if local_addr.ip().is_unspecified() {
local_ip_address::local_ip().unwrap_or(local_addr.ip())
} else {
local_addr.ip()
};
let nat_info = self.nat_test.re_test(Ipv4Addr::from(response.public_ip), response.public_port as u16, local_ip, local_addr.port());
self.channel.set_nat_type(nat_info.nat_type)?;
let new_ip = Ipv4Addr::from(response.virtual_ip);
let current_ip = current_device.virtual_ip();
if current_ip != new_ip {
// ip发生变化
log::info!("ip发生变化,old_ip:{:?},new_ip:{:?}",current_ip,new_ip);
let old_netmask = current_device.virtual_netmask;
let old_gateway = current_device.virtual_gateway();
let virtual_ip = Ipv4Addr::from(response.virtual_ip);
let virtual_gateway = Ipv4Addr::from(response.virtual_gateway);
let virtual_netmask = Ipv4Addr::from(response.virtual_netmask);
self.tun_writer.change_ip(virtual_ip, virtual_netmask, virtual_gateway, old_netmask, old_gateway)?;
let new_current_device = CurrentDeviceInfo::new(virtual_ip, virtual_gateway,
virtual_netmask, current_device.connect_server);
if let Err(e) = self.current_device.compare_exchange(current_device, new_current_device) {
log::warn!("替换失败:{:?}",e);
}
}
self.connect_status.store(ConnectStatus::Connected);
}
service_packet::Protocol::PollDeviceList => {}
service_packet::Protocol::PushDeviceList => {
let device_list_t = DeviceList::parse_from_bytes(net_packet.payload())?;
let ip_list = device_list_t
.device_info_list
.into_iter()
.map(|info| {
PeerDeviceInfo::new(
Ipv4Addr::from(info.virtual_ip),
info.name,
info.device_status as u8,
)
})
.collect();
let mut dev = self.device_list.lock();
if dev.0 < device_list_t.epoch as u16 || device_list_t.epoch as u16 - dev.0 > u16::MAX >> 2 {
dev.0 = device_list_t.epoch as u16;
dev.1 = ip_list;
}
}
service_packet::Protocol::UnKnow(u) => {
log::warn!("未知服务协议:{}",u);
}
}
Ok(())
}
fn error(&self, current_device: CurrentDeviceInfo, source: Ipv4Addr, net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> {
if route_key.addr != current_device.connect_server || source != current_device.virtual_gateway() {
return Ok(());
}
match InErrorPacket::new(net_packet.transport_protocol(), net_packet.payload())? {
InErrorPacket::TokenError => {
return Err(Error::Stop("Token error".to_string()));
}
InErrorPacket::Disconnect => {
self.connect_status.store(ConnectStatus::Connecting);
self.register.fast_register()?;
}
InErrorPacket::AddressExhausted => {
//地址用尽
return Err(Error::Stop("IP address has been exhausted".to_string()));
}
InErrorPacket::OtherError(e) => {
log::error!("OtherError {:?}", e.message());
}
}
Ok(())
}
fn control(&self, current_device: CurrentDeviceInfo, source: Ipv4Addr, mut net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> {
match ControlPacket::new(net_packet.transport_protocol(), net_packet.payload())? {
ControlPacket::PingPacket(_) => {
net_packet.set_transport_protocol(control_packet::Protocol::Pong.into());
net_packet.set_source(current_device.virtual_ip());
net_packet.set_destination(source);
net_packet.first_set_ttl(MAX_TTL);
self.channel.send_to_route(net_packet.buffer(), route_key)?;
}
ControlPacket::PongPacket(pong_packet) => {
let current_time = Local::now().timestamp_millis() as u16;
if current_time < pong_packet.time() {
return Ok(());
}
let rt = (current_time - pong_packet.time()) as i64;
let metric = net_packet.source_ttl() - net_packet.ttl() + 1;
if let Some(current_route) = self.channel.route(&source) {
if &current_route.route_key() == route_key {
self.channel.update_route(&source, metric, rt);
} else if current_route.metric >= metric && current_route.rt > rt {
let route = Route::from(*route_key, metric, rt);
self.channel.add_route(source, route);
}
} else {
let route = Route::from(*route_key, metric, rt);
self.channel.add_route(source, route);
}
if route_key.addr == current_device.connect_server && source == current_device.virtual_gateway() {
let epoch = self.device_list.lock().0;
if pong_packet.epoch() != epoch {
let mut poll_device = NetPacket::new([0; 12])?;
poll_device.set_source(current_device.virtual_ip());
poll_device.set_destination(source);
poll_device.set_version(Version::V1);
poll_device.first_set_ttl(MAX_TTL);
poll_device.set_protocol(Protocol::Service);
poll_device.set_transport_protocol(service_packet::Protocol::PollDeviceList.into());
self.channel.send_to_route(poll_device.buffer(), route_key)?;
}
}
}
ControlPacket::PunchRequest => {
log::info!("PunchRequest route_key:{:?}",route_key);
//回应
net_packet.set_transport_protocol(control_packet::Protocol::PunchResponse.into());
net_packet.set_source(current_device.virtual_ip());
net_packet.set_destination(source);
net_packet.first_set_ttl(1);
self.channel.send_to_route(net_packet.buffer(), route_key)?;
let route = Route::from(*route_key, 1, -1);
self.channel.add_route(source, route);
}
ControlPacket::PunchResponse => {
log::info!("PunchResponse route_key:{:?}",route_key);
let route = Route::from(*route_key, 1, -1);
self.channel.add_route(net_packet.source(), route);
}
}
Ok(())
}
fn other_turn(&self, current_device: CurrentDeviceInfo, source: Ipv4Addr, net_packet: NetPacket<&mut [u8]>, route_key: &RouteKey) -> crate::Result<()> {
match turn_packet::Protocol::from(net_packet.transport_protocol()) {
turn_packet::Protocol::Punch => {
let punch_info = PunchInfo::parse_from_bytes(net_packet.payload())?;
let public_ips = punch_info.public_ip_list.
iter().map(|v| { IpAddr::from(v.to_be_bytes()) }).collect();
let peer_nat_info = NatInfo::new(public_ips,
punch_info.public_port as u16,
punch_info.public_port_range as u16,
IpAddr::from(punch_info.local_ip.to_be_bytes()),
punch_info.local_port as u16,
punch_info.nat_type.enum_value_or_default().into());
self.peer_nat_info_map.insert(source, peer_nat_info.clone());
if !punch_info.reply {
let mut punch_reply = PunchInfo::new();
punch_reply.reply = true;
let nat_info = self.nat_test.nat_info();
punch_reply.public_ip_list = nat_info.public_ips.iter().map(|i| {
match i {
IpAddr::V4(ip) => {
u32::from_be_bytes(ip.octets())
}
IpAddr::V6(_) => {
panic!()
}
}
}).collect();
punch_reply.public_port = nat_info.public_port as u32;
punch_reply.public_port_range = nat_info.public_port_range as u32;
punch_reply.nat_type =
protobuf::EnumOrUnknown::new(PunchNatType::from(nat_info.nat_type));
let bytes = punch_reply.write_to_bytes()?;
let mut net_packet =
NetPacket::new(vec![0u8; 12 + 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.first_set_ttl(MAX_TTL);
net_packet.set_source(current_device.virtual_ip());
net_packet.set_destination(source);
net_packet.set_payload(&bytes);
if !peer_nat_info.local_ip.is_unspecified() {
let mut packet = NetPacket::new([0u8; 12])?;
packet.set_version(Version::V1);
packet.first_set_ttl(1);
packet.set_protocol(Protocol::Control);
packet.set_transport_protocol(control_packet::Protocol::PunchRequest.into());
packet.set_source(current_device.virtual_ip());
packet.set_destination(source);
let _ = self.channel.send_to_addr(packet.buffer(), SocketAddr::new(peer_nat_info.local_ip, peer_nat_info.local_port));
}
if let Err(e) = self.channel.punch(source, peer_nat_info) {
log::warn!("发送到打洞通道失败 {:?}",e);
return Ok(());
}
self.channel.send_to_route(net_packet.buffer(), route_key)?;
} else {
let _ = self.channel.punch(source, peer_nat_info);
}
}
turn_packet::Protocol::UnKnow(e) => {
log::warn!("不支持的转发协议 {:?},source:{:?}",e,source);
}
}
Ok(())
}
}
switch/src/handle/registration_handler.rs
+93 -96
View File
@@ -1,139 +1,136 @@
use std::io;
use std::net::{SocketAddr, UdpSocket};
use std::net::{Ipv4Addr, SocketAddr};
use std::sync::atomic::{AtomicI64, Ordering};
use std::time::Duration;
use chrono::Local;
use crossbeam::atomic::AtomicCell;
use parking_lot::RwLock;
use protobuf::Message;
use nat_traversal::channel::Channel;
use nat_traversal::channel::sender::Sender;
use crate::error::*;
use crate::handle::ConnectStatus;
use crate::proto::message::{RegistrationRequest, RegistrationResponse};
use crate::protocol::error_packet::InErrorPacket;
use crate::protocol::{service_packet, NetPacket, Protocol, Version};
use crate::protocol::{service_packet, NetPacket, Protocol, Version, MAX_TTL};
lazy_static::lazy_static! {
static ref REQUEST:RwLock<Option<(String,String,String)>> = parking_lot::const_rwlock(None);
static ref REGISTRATION_TIME:AtomicI64=AtomicI64::new(0);
pub(crate) static ref CONNECTION_STATUS:AtomicCell<ConnectStatus> = AtomicCell::new(ConnectStatus::Connecting);
}
///向中继服务器注册，token标识一个虚拟网关，mac_address防止多次注册时得到的ip不一致
///向中继服务器注册，token标识一个虚拟网关，device_id防止多次注册时得到的ip不一致
pub fn registration(
udp: &UdpSocket,
channel: &mut Channel<Ipv4Addr>,
server_address: SocketAddr,
token: String,
mac_address: String,
device_id: String,
name: String,
) -> Result<RegistrationResponse> {
// todo 和服务器通信加密
let request_packet =
registration_request_packet(token.clone(), mac_address.clone(), name.clone(), false)?;
registration_request_packet(token.clone(), device_id.clone(), name.clone(), false)?;
let buf = request_packet.buffer();
let mut counter = 0;
let mut recv_buf = [0u8; 10240];
udp.set_read_timeout(Some(Duration::from_millis(500)))?;
loop {
counter += 1;
if counter & 10 == 10 {
return Err(Error::Stop("注册请求超时".to_string()));
}
udp.send_to(buf, server_address)?;
let (len, addr) = match udp.recv_from(&mut recv_buf) {
Ok(ok) => ok,
Err(e) => {
if e.kind() == io::ErrorKind::WouldBlock || e.kind() == io::ErrorKind::TimedOut {
continue;
}
return Err(Error::Io(e));
}
};
if server_address != addr {
continue;
}
let net_packet = NetPacket::new(&recv_buf[..len])?;
match net_packet.protocol() {
Protocol::Service => {
match service_packet::Protocol::from(net_packet.transport_protocol()) {
service_packet::Protocol::RegistrationResponse => {
let response =
RegistrationResponse::parse_from_bytes(net_packet.payload())?;
let _ = REQUEST.write().replace((token, mac_address, name));
udp.set_read_timeout(None)?;
CONNECTION_STATUS.store(ConnectStatus::Connected);
return Ok(response);
}
_ => {}
}
}
Protocol::Error => {
return match InErrorPacket::new(
net_packet.transport_protocol(),
net_packet.payload(),
) {
Ok(e) => match e {
InErrorPacket::TokenError => Err(Error::Stop("token错误".to_string())),
InErrorPacket::Disconnect => Err(Error::Stop("断开连接".to_string())),
InErrorPacket::AddressExhausted => Err(Error::Stop("地址用尽".to_string())),
InErrorPacket::OtherError(e) => match e.message() {
Ok(str) => Err(Error::Stop(str)),
Err(e) => Err(Error::Stop(format!("{:?}", e))),
},
},
Err(e) => Err(Error::Stop(format!("{:?}", e))),
};
}
_ => {
return Err(Error::Stop(format!("数据错误：{:?}", net_packet)));
}
}
channel.send_to_addr(buf, server_address)?;
let (len, route) = channel.recv_from(&mut recv_buf, Some(Duration::from_millis(300)))?;
if server_address != route.addr {
return Err(Error::Warn(format!("数据来源错误：{:?}", route.addr)));
}
let net_packet = NetPacket::new(&recv_buf[..len])?;
return match net_packet.protocol() {
Protocol::Service => {
match service_packet::Protocol::from(net_packet.transport_protocol()) {
service_packet::Protocol::RegistrationResponse => {
let response =
RegistrationResponse::parse_from_bytes(net_packet.payload())?;
Ok(response)
}
_ => {
Err(Error::Warn(format!("数据错误：{:?}", net_packet)))
}
}
}
Protocol::Error => {
match InErrorPacket::new(
net_packet.transport_protocol(),
net_packet.payload(),
) {
Ok(e) => match e {
InErrorPacket::TokenError => Err(Error::Stop("token错误".to_string())),
InErrorPacket::Disconnect => Err(Error::Warn("断开连接".to_string())),
InErrorPacket::AddressExhausted => Err(Error::Stop("地址用尽".to_string())),
InErrorPacket::OtherError(e) => match e.message() {
Ok(str) => Err(Error::Warn(str)),
Err(e) => Err(Error::Warn(format!("{:?}", e))),
},
},
Err(e) => Err(Error::Warn(format!("{:?}", e))),
}
}
_ => {
Err(Error::Warn(format!("数据错误：{:?}", net_packet)))
}
};
}
fn registration_request_packet(
token: String,
mac_address: String,
device_id: String,
name: String,
is_fast: bool,
) -> Result<NetPacket<Vec<u8>>> {
) -> crate::Result<NetPacket<Vec<u8>>> {
let mut request = RegistrationRequest::new();
request.token = token;
request.mac_address = mac_address;
request.device_id = device_id;
request.name = name;
request.is_fast = is_fast;
let bytes = request.write_to_bytes()?;
let buf = vec![0u8; 4 + bytes.len()];
let buf = vec![0u8; 12 + bytes.len()];
let mut net_packet = NetPacket::new(buf)?;
net_packet.set_version(Version::V1);
net_packet.set_protocol(Protocol::Service);
net_packet.set_transport_protocol(service_packet::Protocol::RegistrationRequest.into());
net_packet.set_ttl(255);
net_packet.first_set_ttl(MAX_TTL);
net_packet.set_payload(&bytes);
Ok(net_packet)
}
pub fn fast_registration(udp: &UdpSocket, server_address: SocketAddr) -> Result<()> {
let last = REGISTRATION_TIME.load(Ordering::Relaxed);
let new = Local::now().timestamp_millis();
if new - last < 2000
|| REGISTRATION_TIME
pub struct Register {
sender: Sender<Ipv4Addr>,
server_address: SocketAddr,
token: String,
device_id: String,
name: String,
time: AtomicI64,
}
impl Register {
pub fn new(sender: Sender<Ipv4Addr>,
server_address: SocketAddr,
token: String,
device_id: String,
name: String, ) -> Self {
Self {
sender,
server_address,
token,
device_id,
name,
time: AtomicI64::new(0),
}
}
pub fn fast_register(&self) -> io::Result<()> {
let last = self.time.load(Ordering::Relaxed);
let new = Local::now().timestamp_millis();
if new - last < 1000
|| self.time
.compare_exchange(last, new, Ordering::Relaxed, Ordering::Relaxed)
.is_err()
{
//短时间不重复注册
return Ok(());
{
//短时间不重复注册
return Ok(());
}
log::info!("重新连接");
let request_packet =
registration_request_packet(self.token.clone(),
self.device_id.clone(),
self.name.clone(), false).unwrap();
let buf = request_packet.buffer();
self.sender.send_to_addr(buf, self.server_address)?;
Ok(())
}
CONNECTION_STATUS.store(ConnectStatus::Connecting);
let lock = REQUEST.read();
let option = lock.clone();
drop(lock);
if let Some((token, mac_address, name)) = option {
let request_packet = registration_request_packet(token, mac_address, name, true)?;
udp.send_to(request_packet.buffer(), server_address)?;
REGISTRATION_TIME.store(Local::now().timestamp_millis(), Ordering::Relaxed);
return Ok(());
}
return Err(Error::Stop("注册信息不存在".to_string()));
}
switch/src/handle/tun_handler.rs
+31 -131
View File
@@ -1,29 +1,22 @@
use std::{io, thread};
/// 接收tun数据，并且转发到udp上
use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket};
use std::thread;
use chrono::Local;
use tokio::sync::watch;
use std::net::Ipv4Addr;
use std::sync::Arc;
use crossbeam::atomic::AtomicCell;
use nat_traversal::channel::sender::Sender;
use packet::icmp::icmp::IcmpPacket;
use packet::icmp::Kind;
use packet::ip::ipv4;
use packet::ip::ipv4::packet::IpV4Packet;
use crate::error::*;
use crate::handle::{CurrentDeviceInfo, DIRECT_ROUTE_TABLE};
use crate::protocol::turn_packet::TurnPacket;
use crate::protocol::{NetPacket, Protocol, Version};
use crate::tun_device::TunReader;
use crate::ApplicationStatus;
use crate::handle::{check_dest, CurrentDeviceInfo};
use crate::protocol::{MAX_TTL, NetPacket, Protocol, Version};
use crate::tun_device::{TunReader, TunWriter};
/// 是否在一个网段
fn check_dest(dest: Ipv4Addr, cur_info: &CurrentDeviceInfo) -> bool {
u32::from_be_bytes(dest.octets()) & u32::from_be_bytes(cur_info.virtual_netmask.octets())
== u32::from_be_bytes(cur_info.virtual_network.octets())
}
fn icmp(udp: &UdpSocket, mut ipv4_packet: IpV4Packet<&mut [u8]>) -> Result<()> {
fn icmp(tun_writer: &TunWriter, mut ipv4_packet: IpV4Packet<&mut [u8]>) -> Result<()> {
if ipv4_packet.protocol() == ipv4::protocol::Protocol::Icmp {
let mut icmp = IcmpPacket::new(ipv4_packet.payload_mut())?;
if icmp.kind() == Kind::EchoRequest {
@@ -33,21 +26,14 @@ fn icmp(udp: &UdpSocket, mut ipv4_packet: IpV4Packet<&mut [u8]>) -> Result<()> {
ipv4_packet.set_source_ip(ipv4_packet.destination_ip());
ipv4_packet.set_destination_ip(src);
ipv4_packet.update_checksum();
let mut addr = udp.local_addr()?;
addr.set_ip(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)));
udp.send_to(ipv4_packet.buffer, addr)?;
tun_writer.write(ipv4_packet.buffer)?;
}
}
Ok(())
}
#[inline]
fn handle(
udp: &UdpSocket,
data: &mut [u8],
cur_info: &CurrentDeviceInfo,
net_packet: &mut NetPacket<Vec<u8>>,
) -> Result<()> {
fn handle(sender: &Sender<Ipv4Addr>, data: &mut [u8], tun_writer: &TunWriter, current_device: CurrentDeviceInfo, net_packet: &mut NetPacket<Vec<u8>>) -> Result<()> {
let data_len = data.len();
let ipv4_packet = match IpV4Packet::new(data) {
Ok(ipv4_packet) => ipv4_packet,
@@ -63,135 +49,49 @@ fn handle(
// // 137端口是在局域网中提供计算机的名字或IP地址查询服务
// return Ok(());
// }
if src_ip != cur_info.virtual_ip || !check_dest(dest_ip, &cur_info) {
if src_ip != current_device.virtual_ip() || !check_dest(dest_ip, current_device.virtual_netmask, current_device.virtual_network) {
return Ok(());
}
if src_ip == dest_ip {
return icmp(&udp, ipv4_packet);
return icmp(&tun_writer, ipv4_packet);
}
let mut ipv4_turn_packet = TurnPacket::new(net_packet.payload_mut())?;
ipv4_turn_packet.set_source(src_ip);
ipv4_turn_packet.set_destination(dest_ip);
ipv4_turn_packet.set_payload(ipv4_packet.buffer);
net_packet.set_source(src_ip);
net_packet.set_destination(dest_ip);
net_packet.set_payload(ipv4_packet.buffer);
//优先发到直连到地址
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()
{
return Ok(());
}
}
if sender.send_to_id(&net_packet.buffer()[..(4 + 8 + data_len)], &dest_ip).is_err() {
sender.send_to_addr(&net_packet.buffer()[..(4 + 8 + data_len)], current_device.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,
{
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),
);
});
pub fn start(sender: Sender<Ipv4Addr>,
tun_reader: TunReader,
tun_writer: TunWriter,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>, ) {
thread::spawn(move || {
if let Err(e) = handle_loop(udp, tun_reader, cur_info) {
log::warn!("tun数据处理线程停止 {:?}", e);
if let Err(e) = start_(sender, tun_reader, tun_writer, current_device) {
log::warn!("{:?}",e);
}
stop_fn();
});
}
#[cfg(target_os = "windows")]
fn handle_loop(udp: UdpSocket, tun_reader: TunReader, cur_info: CurrentDeviceInfo) -> Result<()> {
fn start_(sender: Sender<Ipv4Addr>,
tun_reader: TunReader,
tun_writer: TunWriter,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>, ) -> io::Result<()> {
let mut net_packet = NetPacket::new(vec![0u8; 4 + 8 + 1500])?;
net_packet.set_version(Version::V1);
net_packet.set_protocol(Protocol::Ipv4Turn);
net_packet.set_transport_protocol(ipv4::protocol::Protocol::Ipv4.into());
net_packet.set_ttl(255);
net_packet.set_ttl(MAX_TTL);
loop {
let mut data = tun_reader.next()?;
match handle(&udp, data.bytes_mut(), &cur_info, &mut net_packet) {
match handle(&sender, data.bytes_mut(), &tun_writer, current_device.load(), &mut net_packet) {
Ok(_) => {}
Err(e) => {
log::warn!("{:?}", e)
}
}
}
}
#[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,
{
#[cfg(target_os = "macos")]
use std::os::fd::AsRawFd;
#[cfg(target_os = "linux")]
use std::os::unix::io::AsRawFd;
let raw_fd = tun_reader.0.as_raw_fd();
tokio::spawn(async move {
let _ = status_watch.changed().await;
// 让tun接收线程关闭，问题：如果改变tun配置，可能导致tun接收线程无法关闭
unsafe {
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),
);
});
thread::spawn(move || {
if let Err(e) = handle_loop(udp, tun_reader, cur_info) {
log::warn!(" tun数据处理线程停止 {:?}", e);
}
stop_fn();
});
}
#[cfg(any(target_os = "macos", target_os = "linux", target_os = "android"))]
pub fn handle_loop(
udp: UdpSocket,
mut 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);
net_packet.set_transport_protocol(0);
net_packet.set_ttl(255);
let mut buf = [0u8; 1500];
loop {
let data = tun_reader.read(&mut buf)?;
match handle(&udp, data, &cur_info, &mut net_packet) {
Ok(_) => {}
Err(e) => {
log::warn!("{:?}", e)
}
}
}
}
}
switch/src/handle/udp_recv_handler.rs
-389
View File
@@ -1,389 +0,0 @@
use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket};
use std::sync::atomic::Ordering;
use std::thread;
use chrono::Local;
use packet::icmp::{icmp, Kind};
use packet::ip::ipv4;
use packet::ip::ipv4::packet::IpV4Packet;
use protobuf::Message;
use tokio::sync::mpsc::error::TrySendError;
use tokio::sync::mpsc::{Receiver, Sender};
use tokio::sync::watch;
use crate::error::*;
use crate::handle::punch_handler::PunchSender;
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::{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, PeerDeviceInfo};
const UDP_STOP_BUF: [u8; 1] = [0u8];
pub async fn udp_recv_start<F>(
mut status_watch: watch::Receiver<ApplicationStatus>,
udp: UdpSocket,
server_addr: SocketAddr,
other_sender: Sender<(SocketAddr, Vec<u8>)>,
tun_writer: TunWriter,
current_device: CurrentDeviceInfo,
stop_fn: F,
) where
F: FnOnce() + Send + 'static,
{
{
let udp = udp.try_clone().unwrap();
tokio::spawn(async move {
let _ = status_watch.changed().await;
let mut addr = udp.local_addr().unwrap();
addr.set_ip(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)));
udp.send_to(&UDP_STOP_BUF, addr).unwrap();
});
}
thread::spawn(move || {
if let Err(e) = recv_loop(udp, server_addr, other_sender, tun_writer, current_device) {
log::warn!("udp数据处理线程停止 {:?}", e);
}
stop_fn();
});
}
fn recv_loop(
udp: UdpSocket,
server_addr: SocketAddr,
other_sender: Sender<(SocketAddr, Vec<u8>)>,
mut tun_writer: TunWriter,
current_device: CurrentDeviceInfo,
) -> Result<()> {
let mut buf = [0u8; 65536];
let mut local_addr = udp.local_addr()?;
local_addr.set_ip(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)));
loop {
match udp.recv_from(&mut buf) {
Ok((len, addr)) => {
if addr == local_addr {
if len == 1 && &buf[..len] == &UDP_STOP_BUF {
return Ok(());
}
//本地的包直接再发到网卡，这个主要用于处理当前虚拟ip的icmp ping
if let Ok(ip) = IpV4Packet::new(&buf[..len]) {
if ip.destination_ip() == current_device.virtual_ip {
let _ = tun_writer.write(&buf[..len]);
}
}
continue;
}
match recv_handle(
&udp,
addr,
&mut buf[..len],
&server_addr,
&other_sender,
&mut tun_writer,
&current_device,
) {
Ok(_) => {}
Err(Error::Stop(str)) => {
return Err(Error::Stop(str));
}
Err(e) => {
log::warn!("{:?}", e);
}
}
}
Err(e) => {
log::warn!("{:?}", e);
}
};
}
}
fn recv_handle(
udp: &UdpSocket,
recv_addr: SocketAddr,
buf: &mut [u8],
_server_addr: &SocketAddr,
other_sender: &Sender<(SocketAddr, Vec<u8>)>,
tun_writer: &mut TunWriter,
current_device: &CurrentDeviceInfo,
) -> Result<()> {
let mut net_packet = NetPacket::new(buf)?;
match net_packet.protocol() {
Protocol::Ipv4Turn => {
let mut ipv4_turn_packet = TurnPacket::new(net_packet.payload_mut())?;
let source = ipv4_turn_packet.source();
let destination = ipv4_turn_packet.destination();
let mut ipv4 = IpV4Packet::new(ipv4_turn_packet.payload_mut())?;
if ipv4.source_ip() == source
&& ipv4.destination_ip() == destination
&& current_device.virtual_ip == ipv4.destination_ip()
{
if ipv4.protocol() == ipv4::protocol::Protocol::Icmp {
let mut icmp_packet = icmp::IcmpPacket::new(ipv4.payload_mut())?;
if icmp_packet.kind() == Kind::EchoRequest {
//开启ping
icmp_packet.set_kind(Kind::EchoReply);
icmp_packet.update_checksum();
ipv4.set_source_ip(destination);
ipv4.set_destination_ip(source);
ipv4.update_checksum();
ipv4_turn_packet.set_source(destination);
ipv4_turn_packet.set_destination(source);
udp.send_to(net_packet.buffer(), recv_addr)?;
} else {
tun_writer.write(ipv4_turn_packet.payload())?;
}
} else {
tun_writer.write(ipv4_turn_packet.payload())?;
}
}
}
Protocol::UnKnow(_) => {}
_ => {
//发送到子线程处理
let v = net_packet.buffer().to_vec();
match other_sender.try_send((recv_addr, v)) {
Ok(_) => {}
Err(TrySendError::Closed(_)) => {
return Err(Error::Stop("子处理线程停止".to_string()));
}
Err(e) => {
log::warn!("子线程处理 {:?}", e);
}
}
}
}
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,
{
tokio::spawn(async move {
match other_loop(status_watch, udp, receiver, current_device, sender).await {
Ok(_) => {
log::info!("udp子处理线程停止");
}
Err(e) => {
log::warn!("{:?}", e);
}
}
stop_fn();
});
}
async fn other_loop(
mut status_watch: watch::Receiver<ApplicationStatus>,
udp: UdpSocket,
mut receiver: Receiver<(SocketAddr, Vec<u8>)>,
current_device: CurrentDeviceInfo,
sender: PunchSender,
) -> Result<()> {
loop {
tokio::select! {
rs = receiver.recv()=>{
if let Some((peer_addr, buf)) = rs {
match other_handle(&udp, buf, peer_addr, &current_device, &sender) {
Ok(_) => {}
Err(Error::Stop(str)) => {
return Err(Error::Stop(str));
}
Err(e) => {
log::warn!("other_loop {:?}",e);
}
}
}
}
status = status_watch.changed() =>{
status?;
if *status_watch.borrow() != ApplicationStatus::Starting{
return Ok(())
}
}
}
}
}
fn other_handle(
udp: &UdpSocket,
buf: Vec<u8>,
peer_addr: SocketAddr,
current_device: &CurrentDeviceInfo,
sender: &PunchSender,
) -> Result<()> {
let server_addr = current_device.connect_server;
let mut net_packet = NetPacket::new(buf)?;
match net_packet.protocol() {
Protocol::Service => {
if peer_addr != current_device.connect_server {
return Ok(());
}
match service_packet::Protocol::from(net_packet.transport_protocol()) {
service_packet::Protocol::RegistrationRequest => {}
service_packet::Protocol::RegistrationResponse => {
let response = RegistrationResponse::parse_from_bytes(net_packet.payload())?;
crate::handle::init_nat_info(response.public_ip, response.public_port as u16);
CONNECTION_STATUS.store(ConnectStatus::Connected);
//需要保证重连ip不变
}
service_packet::Protocol::UpdateDeviceList => {
let device_list = DeviceList::parse_from_bytes(net_packet.payload())?;
let ip_list = device_list
.device_info_list
.into_iter()
.map(|info| {
PeerDeviceInfo::new(
Ipv4Addr::from(info.virtual_ip),
info.name,
info.device_status as u8,
)
})
.collect();
let mut dev = DEVICE_LIST.lock();
if dev.0 < device_list.epoch || device_list.epoch - dev.0 > u32::MAX >> 2 {
dev.0 = device_list.epoch;
dev.1 = ip_list;
}
}
service_packet::Protocol::UnKnow(_) => {}
}
}
Protocol::Error => {
match InErrorPacket::new(net_packet.transport_protocol(), net_packet.payload())? {
InErrorPacket::TokenError => {
if server_addr == peer_addr {
//停止整个应用
return Err(Error::Stop("token无效".to_string()));
}
}
InErrorPacket::Disconnect => {
if server_addr == peer_addr {
fast_registration(&udp, server_addr)?;
}
}
InErrorPacket::AddressExhausted => {
return Err(Error::Stop("IP address has been exhausted".to_string()));
}
InErrorPacket::OtherError(e) => {
log::error!("OtherError {:?}", e.message());
}
}
}
Protocol::Control => {
match ControlPacket::new(net_packet.transport_protocol(), net_packet.payload())? {
ControlPacket::PingPacket(_ping) => {
net_packet.set_transport_protocol(control_packet::Protocol::Pong.into());
udp.send_to(&net_packet.buffer()[..12], peer_addr)?;
}
ControlPacket::PongPacket(pong_packet) => {
let current_time = Local::now().timestamp_millis();
let rt = current_time - pong_packet.time();
if rt >= 0 {
if peer_addr == server_addr {
SERVER_RT.store(rt, Ordering::Relaxed)
} else {
//其他设备
if let Some(virtual_ip) = ADDR_TABLE.get(&peer_addr) {
if let Some(mut info) = DIRECT_ROUTE_TABLE.get_mut(&virtual_ip) {
info.rt = rt;
info.recv_time = current_time;
}
}
}
}
}
ControlPacket::PunchRequest(punch_request) => {
// println!("打洞请求:{:?}", punch_request);
let src = punch_request.source();
drop(punch_request);
//回应
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());
udp.send_to(net_packet.buffer(), peer_addr)?;
let route = Route::new(peer_addr);
DIRECT_ROUTE_TABLE.insert(src, route);
ADDR_TABLE.insert(peer_addr, src);
}
ControlPacket::PunchResponse(punch_response) => {
// println!("打洞响应:{:?}", punch_response);
let route = Route::new(peer_addr);
DIRECT_ROUTE_TABLE.insert(punch_response.source(), route);
ADDR_TABLE.insert(peer_addr, punch_response.source());
}
}
}
Protocol::Ipv4Turn => {}
Protocol::OtherTurn => {
let turn_packet = TurnPacket::new(net_packet.payload())?;
// println!("{:?}",turn_packet);
let src = turn_packet.source();
let dest = turn_packet.destination();
if dest == current_device.virtual_ip {
match turn_packet::Protocol::from(net_packet.transport_protocol()) {
turn_packet::Protocol::Punch => {
let punch = Punch::parse_from_bytes(turn_packet.payload())?;
if punch.virtual_ip.to_be_bytes() == src.octets() {
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());
if let Err(_) = sender.try_send(punch) {
return Ok(());
}
let nat_info = NAT_INFO.lock();
if let Some(info) = nat_info.as_ref() {
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);
drop(nat_info);
let bytes = punch_reply.write_to_bytes()?;
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_ttl(255);
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);
udp.send_to(net_packet.buffer(), peer_addr)?;
}
} else {
let _ = sender.try_send(punch);
}
}
}
turn_packet::Protocol::UnKnow(_) => {}
}
}
}
Protocol::UnKnow(p) => {
log::warn!("未知协议 {}", p);
}
}
Ok(())
}