[mio] 去除多余代码
This commit is contained in:
lubeilin
@@ -1,262 +0,0 @@
|
||||
use std::net::{SocketAddr, SocketAddrV6};
|
||||
|
||||
use crate::channel::channel::Context;
|
||||
use crate::channel::RouteKey;
|
||||
use crate::cipher::{Cipher, RsaCipher};
|
||||
use crate::proto::message::{HandshakeRequest, HandshakeResponse, SecretHandshakeRequest};
|
||||
use crate::protocol::body::RSA_ENCRYPTION_RESERVED;
|
||||
use crate::protocol::{service_packet, NetPacket, Protocol, Version, MAX_TTL};
|
||||
use protobuf::Message;
|
||||
use std::io::{Read, Write};
|
||||
use std::net::TcpStream;
|
||||
use std::net::UdpSocket;
|
||||
|
||||
pub enum HandshakeEnum {
|
||||
NotSecret,
|
||||
KeyError,
|
||||
Timeout,
|
||||
ServerError(String),
|
||||
Other(String),
|
||||
}
|
||||
|
||||
fn handshake_request_packet(secret: bool) -> crate::Result<NetPacket<Vec<u8>>> {
|
||||
let mut request = HandshakeRequest::new();
|
||||
request.secret = secret;
|
||||
request.version = crate::VNT_VERSION.to_string();
|
||||
let bytes = request.write_to_bytes()?;
|
||||
let buf = vec![0u8; 12 + bytes.len()];
|
||||
let mut net_packet = NetPacket::new(buf)?;
|
||||
net_packet.set_version(Version::V1);
|
||||
net_packet.set_gateway_flag(true);
|
||||
net_packet.set_protocol(Protocol::Service);
|
||||
net_packet.set_transport_protocol(service_packet::Protocol::HandshakeRequest.into());
|
||||
net_packet.first_set_ttl(MAX_TTL);
|
||||
net_packet.set_payload(&bytes)?;
|
||||
Ok(net_packet)
|
||||
}
|
||||
|
||||
fn secret_handshake_request_packet(
|
||||
rsa_cipher: &RsaCipher,
|
||||
token: String,
|
||||
key: &[u8],
|
||||
) -> crate::Result<NetPacket<Vec<u8>>> {
|
||||
let mut request = SecretHandshakeRequest::new();
|
||||
request.token = token;
|
||||
request.key = key.to_vec();
|
||||
let bytes = request.write_to_bytes()?;
|
||||
let mut net_packet = NetPacket::new0(
|
||||
12 + bytes.len(),
|
||||
vec![0u8; 12 + bytes.len() + RSA_ENCRYPTION_RESERVED],
|
||||
)?;
|
||||
net_packet.set_version(Version::V1);
|
||||
net_packet.set_gateway_flag(true);
|
||||
net_packet.set_protocol(Protocol::Service);
|
||||
net_packet.set_transport_protocol(service_packet::Protocol::SecretHandshakeRequest.into());
|
||||
net_packet.first_set_ttl(MAX_TTL);
|
||||
net_packet.set_payload(&bytes)?;
|
||||
Ok(rsa_cipher.encrypt(&mut net_packet)?)
|
||||
}
|
||||
|
||||
/// 第一次握手，拿到公钥
|
||||
pub fn handshake(
|
||||
main_channel: &UdpSocket,
|
||||
main_tcp_channel: Option<&mut TcpStream>,
|
||||
server_address: SocketAddr,
|
||||
secret: bool,
|
||||
) -> Result<Option<RsaCipher>, HandshakeEnum> {
|
||||
let request_packet = handshake_request_packet(secret).unwrap();
|
||||
let send_buf = request_packet.buffer();
|
||||
let mut recv_buf = [0u8; 10240];
|
||||
let len = send_recv(
|
||||
main_channel,
|
||||
main_tcp_channel,
|
||||
server_address,
|
||||
send_buf,
|
||||
&mut recv_buf,
|
||||
)?;
|
||||
let net_packet = match NetPacket::new(&recv_buf[..len]) {
|
||||
Ok(net_packet) => net_packet,
|
||||
Err(e) => {
|
||||
return Err(HandshakeEnum::Other(format!("net_packet {}", e)));
|
||||
}
|
||||
};
|
||||
match net_packet.protocol() {
|
||||
Protocol::Service => {
|
||||
match service_packet::Protocol::from(net_packet.transport_protocol()) {
|
||||
service_packet::Protocol::HandshakeResponse => {
|
||||
match HandshakeResponse::parse_from_bytes(net_packet.payload()) {
|
||||
Ok(response) => {
|
||||
if !response.secret && secret {
|
||||
//客户端要加密，服务端不支持加密
|
||||
return Err(HandshakeEnum::NotSecret);
|
||||
}
|
||||
if secret {
|
||||
//转换公钥
|
||||
match RsaCipher::new(&response.public_key) {
|
||||
Ok(rsa) => {
|
||||
match rsa.finger() {
|
||||
Ok(finger) => {
|
||||
if finger != response.key_finger {
|
||||
return Err(HandshakeEnum::Other(
|
||||
"finger error".to_string(),
|
||||
));
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
return Err(HandshakeEnum::Other(format!(
|
||||
"finger {}",
|
||||
e
|
||||
)));
|
||||
}
|
||||
}
|
||||
Ok(Some(rsa))
|
||||
}
|
||||
Err(e) => {
|
||||
return Err(HandshakeEnum::Other(format!(
|
||||
"RsaCipher {}",
|
||||
e
|
||||
)));
|
||||
}
|
||||
}
|
||||
} else {
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
return Err(HandshakeEnum::Other(format!("parse_from_bytes {}", e)));
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
return Err(HandshakeEnum::Other("not match".to_string()));
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
return Err(HandshakeEnum::Other("not match".to_string()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn send_recv(
|
||||
main_channel: &UdpSocket,
|
||||
main_tcp_channel: Option<&mut TcpStream>,
|
||||
server_address: SocketAddr,
|
||||
send_buf: &[u8],
|
||||
recv_buf: &mut [u8],
|
||||
) -> Result<usize, HandshakeEnum> {
|
||||
if let Some(main_tcp_channel) = main_tcp_channel {
|
||||
let mut head = [0; 4];
|
||||
let len = send_buf.len();
|
||||
head[2] = (len >> 8) as u8;
|
||||
head[3] = (len & 0xFF) as u8;
|
||||
if let Err(e) = main_tcp_channel.write_all(&head) {
|
||||
return Err(HandshakeEnum::Other(format!("send error:{}", e)));
|
||||
}
|
||||
if let Err(e) = main_tcp_channel.write_all(send_buf) {
|
||||
return Err(HandshakeEnum::Other(format!("send error:{}", e)));
|
||||
}
|
||||
if let Err(e) = main_tcp_channel.read_exact(&mut head) {
|
||||
return Err(HandshakeEnum::Other(format!("read error:{}", e)));
|
||||
}
|
||||
let len = (((head[2] as u16) << 8) | head[3] as u16) as usize;
|
||||
if len > recv_buf.len() {
|
||||
return Err(HandshakeEnum::Other("too long".to_string()));
|
||||
}
|
||||
if let Err(e) = main_tcp_channel.read_exact(&mut recv_buf[..len]) {
|
||||
return Err(HandshakeEnum::Other(format!("read error:{}", e)));
|
||||
}
|
||||
Ok(len)
|
||||
} else {
|
||||
let server_address = if let SocketAddr::V4(ipv4) = server_address {
|
||||
SocketAddr::V6(SocketAddrV6::new(
|
||||
ipv4.ip().to_ipv6_mapped(),
|
||||
ipv4.port(),
|
||||
0,
|
||||
0,
|
||||
))
|
||||
} else {
|
||||
server_address
|
||||
};
|
||||
if let Err(e) = main_channel.send_to(send_buf, server_address) {
|
||||
return Err(HandshakeEnum::Other(format!("send error:{}", e)));
|
||||
}
|
||||
match main_channel.recv_from(recv_buf) {
|
||||
Ok((len, addr)) => {
|
||||
if server_address != addr {
|
||||
log::warn!("请求{:?}和响应{:?}地址不一致", server_address, addr);
|
||||
}
|
||||
Ok(len)
|
||||
}
|
||||
Err(e) => Err(HandshakeEnum::Other(format!("receiver error:{}", e))),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 第二次握手，同步对称密钥，后续将使用对称加密
|
||||
pub fn secret_handshake(
|
||||
main_channel: &UdpSocket,
|
||||
main_tcp_channel: Option<&mut TcpStream>,
|
||||
server_address: SocketAddr,
|
||||
rsa_cipher: &RsaCipher,
|
||||
server_cipher: &Cipher,
|
||||
token: String,
|
||||
) -> Result<(), HandshakeEnum> {
|
||||
let secret_packet =
|
||||
match secret_handshake_request_packet(rsa_cipher, token, server_cipher.key().unwrap()) {
|
||||
Ok(secret_packet) => secret_packet,
|
||||
Err(e) => {
|
||||
return Err(HandshakeEnum::Other(format!(
|
||||
"secret_handshake_request_packet {}",
|
||||
e
|
||||
)));
|
||||
}
|
||||
};
|
||||
let send_buf = secret_packet.buffer();
|
||||
let mut recv_buf = [0u8; 10240];
|
||||
let len = send_recv(
|
||||
main_channel,
|
||||
main_tcp_channel,
|
||||
server_address,
|
||||
send_buf,
|
||||
&mut recv_buf,
|
||||
)?;
|
||||
let mut net_packet = match NetPacket::new(&mut recv_buf[..len]) {
|
||||
Ok(net_packet) => net_packet,
|
||||
Err(e) => {
|
||||
return Err(HandshakeEnum::Other(format!("secret_net_packet {}", e)));
|
||||
}
|
||||
};
|
||||
match server_cipher.decrypt_ipv4(&mut net_packet) {
|
||||
Ok(_) => {
|
||||
if net_packet.is_gateway()
|
||||
&& net_packet.protocol() == Protocol::Service
|
||||
&& service_packet::Protocol::from(net_packet.transport_protocol())
|
||||
== service_packet::Protocol::SecretHandshakeResponse
|
||||
{
|
||||
Ok(())
|
||||
} else {
|
||||
Err(HandshakeEnum::Other("not match".to_string()))
|
||||
}
|
||||
}
|
||||
Err(e) => Err(HandshakeEnum::Other(format!("decrypt_ipv4 {}", e))),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn secret_handshake_req(
|
||||
context: &Context,
|
||||
server_address: SocketAddr,
|
||||
rsa_cipher: &RsaCipher,
|
||||
server_cipher: &Cipher,
|
||||
token: String,
|
||||
route_key: &RouteKey,
|
||||
) -> crate::Result<()> {
|
||||
let secret_packet =
|
||||
secret_handshake_request_packet(rsa_cipher, token, server_cipher.key().unwrap())?;
|
||||
if route_key.is_tcp() {
|
||||
context.send_main(secret_packet.buffer(), server_address)?;
|
||||
} else {
|
||||
context.send_main_udp(secret_packet.buffer(), server_address)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
@@ -1,301 +0,0 @@
|
||||
use std::io;
|
||||
use std::net::{Ipv4Addr, ToSocketAddrs};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use crossbeam_utils::atomic::AtomicCell;
|
||||
use parking_lot::Mutex;
|
||||
use rand::prelude::SliceRandom;
|
||||
|
||||
use crate::channel::idle::Idle;
|
||||
use crate::channel::sender::ChannelSender;
|
||||
use crate::channel::Route;
|
||||
use crate::cipher::Cipher;
|
||||
use crate::core::status::VntWorker;
|
||||
use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo};
|
||||
use crate::protocol::body::ENCRYPTION_RESERVED;
|
||||
use crate::protocol::control_packet::PingPacket;
|
||||
use crate::protocol::{control_packet, NetPacket, Protocol, Version, MAX_TTL};
|
||||
|
||||
pub fn start_idle(mut worker: VntWorker, idle: Idle, sender: ChannelSender) {
|
||||
tokio::spawn(async move {
|
||||
tokio::select! {
|
||||
_=worker.stop_wait()=>{
|
||||
return;
|
||||
}
|
||||
rs=start_idle_(idle, sender)=>{
|
||||
if let Err(e) = rs {
|
||||
log::warn!("空闲检测任务停止:{:?}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
worker.stop_all();
|
||||
});
|
||||
}
|
||||
|
||||
async fn start_idle_(idle: Idle, sender: ChannelSender) -> io::Result<()> {
|
||||
log::info!("启动空闲检查任务");
|
||||
loop {
|
||||
let (peer_ip, route) = idle.next_idle().await?;
|
||||
log::info!("路由空闲 peer_ip:{:?},route:{:?}", peer_ip, route);
|
||||
sender.remove_route(&peer_ip, route);
|
||||
}
|
||||
}
|
||||
|
||||
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,
|
||||
sender: ChannelSender,
|
||||
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
|
||||
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
|
||||
server_address_str: String,
|
||||
client_cipher: Cipher,
|
||||
server_cipher: Cipher,
|
||||
) {
|
||||
tokio::spawn(async move {
|
||||
tokio::select! {
|
||||
_=worker.stop_wait()=>{
|
||||
return;
|
||||
}
|
||||
rs=start_heartbeat_main_(sender, device_list, current_device,server_address_str,client_cipher,server_cipher)=>{
|
||||
if let Err(e) = rs {
|
||||
log::warn!("主心跳任务停止:{:?}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
worker.stop_all();
|
||||
});
|
||||
}
|
||||
|
||||
fn heartbeat_packet(
|
||||
ttl: u8,
|
||||
device_list: &Mutex<(u16, Vec<PeerDeviceInfo>)>,
|
||||
client_cipher: &Cipher,
|
||||
server_cipher: &Cipher,
|
||||
gateway: bool,
|
||||
src: Ipv4Addr,
|
||||
dest: Ipv4Addr,
|
||||
) -> NetPacket<[u8; 12 + 4 + ENCRYPTION_RESERVED]> {
|
||||
let mut net_packet = NetPacket::new_encrypt([0u8; 12 + 4 + ENCRYPTION_RESERVED]).unwrap();
|
||||
net_packet.set_version(Version::V1);
|
||||
net_packet.set_protocol(Protocol::Control);
|
||||
net_packet.set_transport_protocol(control_packet::Protocol::Ping.into());
|
||||
net_packet.first_set_ttl(ttl);
|
||||
net_packet.set_source(src);
|
||||
net_packet.set_destination(dest);
|
||||
{
|
||||
let mut ping = PingPacket::new(net_packet.payload_mut()).unwrap();
|
||||
let epoch = { device_list.lock().0 };
|
||||
ping.set_epoch(epoch);
|
||||
ping.set_time(crate::handle::now_time() as u16);
|
||||
}
|
||||
if gateway {
|
||||
net_packet.set_gateway_flag(true);
|
||||
server_cipher.encrypt_ipv4(&mut net_packet).unwrap();
|
||||
} else {
|
||||
client_cipher.encrypt_ipv4(&mut net_packet).unwrap();
|
||||
}
|
||||
net_packet
|
||||
}
|
||||
|
||||
async fn start_heartbeat_main_(
|
||||
sender: ChannelSender,
|
||||
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
|
||||
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
|
||||
server_address_str: String,
|
||||
client_cipher: Cipher,
|
||||
server_cipher: Cipher,
|
||||
) -> io::Result<()> {
|
||||
let mut count = 0;
|
||||
log::info!("启动主心跳任务");
|
||||
loop {
|
||||
if sender.is_close() {
|
||||
return Ok(());
|
||||
}
|
||||
let mut current_dev = current_device.load();
|
||||
let src = current_dev.virtual_ip();
|
||||
if count % 40 == 19 {
|
||||
if let Ok(mut addr) = server_address_str.to_socket_addrs() {
|
||||
if let Some(addr) = addr.next() {
|
||||
if addr != current_dev.connect_server {
|
||||
let mut tmp = current_dev.clone();
|
||||
tmp.connect_server = addr;
|
||||
log::info!(
|
||||
"服务端地址变化,旧地址:{}，新地址:{}",
|
||||
current_dev.connect_server,
|
||||
addr
|
||||
);
|
||||
if current_device.compare_exchange(current_dev, tmp).is_ok() {
|
||||
current_dev.connect_server = addr;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
let server_packet = heartbeat_packet(
|
||||
MAX_TTL,
|
||||
&device_list,
|
||||
&client_cipher,
|
||||
&server_cipher,
|
||||
true,
|
||||
src,
|
||||
current_dev.virtual_gateway,
|
||||
);
|
||||
if let Err(e) = sender.send_main(server_packet.buffer(), current_dev.connect_server) {
|
||||
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 % 10 == 7 {
|
||||
let mut route_list: Option<Vec<(Ipv4Addr, Vec<Route>)>> = None;
|
||||
let peer_list = { device_list.lock().1.clone() };
|
||||
for peer in peer_list {
|
||||
if peer.virtual_ip == current_dev.virtual_ip {
|
||||
continue;
|
||||
}
|
||||
let client_packet = heartbeat_packet(
|
||||
MAX_TTL,
|
||||
&device_list,
|
||||
&client_cipher,
|
||||
&server_cipher,
|
||||
false,
|
||||
src,
|
||||
peer.virtual_ip,
|
||||
);
|
||||
if let Some(route) = sender.route_one(&peer.virtual_ip) {
|
||||
if let Err(e) =
|
||||
sender.try_send_by_key(client_packet.buffer(), &route.route_key())
|
||||
{
|
||||
log::warn!("virtual_ip:{},route:{:?},e:{:?}", peer.virtual_ip, route, e);
|
||||
}
|
||||
if route.is_p2p() && !sender.is_first_latency() {
|
||||
continue;
|
||||
}
|
||||
} else {
|
||||
//没有直连路由则发送到网关
|
||||
if let Err(e) =
|
||||
sender.send_main(client_packet.buffer(), current_dev.connect_server)
|
||||
{
|
||||
log::warn!(
|
||||
"virtual_ip:{},connect_server:{:?},e:{:?}",
|
||||
peer.virtual_ip,
|
||||
current_dev.connect_server,
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
//再随机发送到其他地址，看有没有客户端符合转发条件
|
||||
let route_list = route_list.get_or_insert_with(|| {
|
||||
let mut l = sender.route_table();
|
||||
l.shuffle(&mut rand::thread_rng());
|
||||
l
|
||||
});
|
||||
let mut num = 0;
|
||||
'a: for (peer_ip, route_list) in route_list.iter() {
|
||||
for route in route_list {
|
||||
if peer_ip != &peer.virtual_ip && route.is_p2p() {
|
||||
if let Err(e) =
|
||||
sender.try_send_by_key(client_packet.buffer(), &route.route_key())
|
||||
{
|
||||
log::warn!(
|
||||
"virtual_ip:{},route:{:?},e:{:?}",
|
||||
peer.virtual_ip,
|
||||
route,
|
||||
e
|
||||
);
|
||||
}
|
||||
num += 1;
|
||||
break;
|
||||
}
|
||||
if num >= 2 {
|
||||
break 'a;
|
||||
}
|
||||
}
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(1)).await;
|
||||
}
|
||||
} else {
|
||||
for (peer_ip, route_list) in sender.route_table().iter() {
|
||||
if peer_ip == ¤t_dev.virtual_gateway {
|
||||
continue;
|
||||
}
|
||||
let client_packet = heartbeat_packet(
|
||||
MAX_TTL,
|
||||
&device_list,
|
||||
&client_cipher,
|
||||
&server_cipher,
|
||||
false,
|
||||
src,
|
||||
*peer_ip,
|
||||
);
|
||||
for route in route_list {
|
||||
if let Err(e) =
|
||||
sender.try_send_by_key(client_packet.buffer(), &route.route_key())
|
||||
{
|
||||
log::warn!("peer_ip:{:?},route:{:?},e:{:?}", peer_ip, route, e);
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(2)).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
count += 1;
|
||||
tokio::time::sleep(Duration::from_millis(3000)).await;
|
||||
}
|
||||
}
|
||||
@@ -1,180 +0,0 @@
|
||||
use crate::channel::punch::{NatInfo, Punch};
|
||||
use crate::channel::sender::ChannelSender;
|
||||
use crate::cipher::Cipher;
|
||||
use crate::core::status::VntWorker;
|
||||
use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo};
|
||||
use crate::nat::NatTest;
|
||||
use crate::proto::message::{PunchInfo, PunchNatType};
|
||||
use crate::protocol::body::ENCRYPTION_RESERVED;
|
||||
use crate::protocol::{control_packet, other_turn_packet, NetPacket, Protocol, Version, MAX_TTL};
|
||||
use crossbeam_utils::atomic::AtomicCell;
|
||||
use parking_lot::Mutex;
|
||||
use protobuf::Message;
|
||||
use rand::prelude::SliceRandom;
|
||||
use std::io;
|
||||
use std::net::Ipv4Addr;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::mpsc::Receiver;
|
||||
|
||||
pub fn start(
|
||||
mut worker: VntWorker,
|
||||
receiver: Receiver<(Ipv4Addr, NatInfo)>,
|
||||
punch: Punch,
|
||||
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
|
||||
client_cipher: Cipher,
|
||||
) {
|
||||
tokio::spawn(async move {
|
||||
tokio::select! {
|
||||
_=start0(receiver, punch, current_device,client_cipher)=>{}
|
||||
_=worker.stop_wait()=>{
|
||||
return;
|
||||
}
|
||||
}
|
||||
worker.stop_all();
|
||||
});
|
||||
}
|
||||
|
||||
pub async fn start0(
|
||||
mut receiver: Receiver<(Ipv4Addr, NatInfo)>,
|
||||
mut punch: Punch,
|
||||
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
|
||||
client_cipher: Cipher,
|
||||
) {
|
||||
log::info!("启动打洞任务");
|
||||
while let Some((peer_ip, nat_info)) = receiver.recv().await {
|
||||
if let Err(e) = start_(
|
||||
&client_cipher,
|
||||
&mut punch,
|
||||
¤t_device,
|
||||
peer_ip,
|
||||
nat_info,
|
||||
)
|
||||
.await
|
||||
{
|
||||
log::warn!("网络打洞异常 {:?}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn start_(
|
||||
client_cipher: &Cipher,
|
||||
punch: &mut Punch,
|
||||
current_device: &Arc<AtomicCell<CurrentDeviceInfo>>,
|
||||
peer_ip: Ipv4Addr,
|
||||
nat_info: NatInfo,
|
||||
) -> io::Result<()> {
|
||||
let mut packet = NetPacket::new_encrypt([0u8; 12 + ENCRYPTION_RESERVED])?;
|
||||
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.load().virtual_ip());
|
||||
packet.set_destination(peer_ip);
|
||||
log::info!("发起打洞，目标:{:?},{:?}", peer_ip, nat_info);
|
||||
client_cipher.encrypt_ipv4(&mut packet)?;
|
||||
punch.punch(packet.buffer(), peer_ip, nat_info).await
|
||||
}
|
||||
|
||||
pub async fn start_punch(
|
||||
mut worker: VntWorker,
|
||||
nat_test: NatTest,
|
||||
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
|
||||
sender: ChannelSender,
|
||||
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
|
||||
client_cipher: Cipher,
|
||||
) {
|
||||
let mut num = 0;
|
||||
let sleep_time = [3, 5, 7, 11, 13, 17, 19, 23, 29];
|
||||
log::info!("启动发起打洞请求任务");
|
||||
loop {
|
||||
if sender.is_close() {
|
||||
break;
|
||||
}
|
||||
tokio::select! {
|
||||
rs= start_punch_(Duration::from_secs(sleep_time[num % sleep_time.len()]),&nat_test, &device_list,
|
||||
&sender, ¤t_device,&client_cipher)=>{
|
||||
if let Err(e) = rs {
|
||||
log::warn!("打洞处理任务异常 {:?}", e);
|
||||
}
|
||||
}
|
||||
_=worker.stop_wait()=>{
|
||||
break;
|
||||
}
|
||||
}
|
||||
num += 1;
|
||||
}
|
||||
}
|
||||
|
||||
async fn start_punch_(
|
||||
sleep_time: Duration,
|
||||
nat_test: &NatTest,
|
||||
device_list: &Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
|
||||
sender: &ChannelSender,
|
||||
current_device: &Arc<AtomicCell<CurrentDeviceInfo>>,
|
||||
client_cipher: &Cipher,
|
||||
) -> crate::Result<()> {
|
||||
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;
|
||||
}
|
||||
if !sender.need_punch(&info.virtual_ip) {
|
||||
continue;
|
||||
}
|
||||
count += 1;
|
||||
if count > 2 {
|
||||
break;
|
||||
}
|
||||
let packet = punch_packet(
|
||||
client_cipher,
|
||||
current_device.virtual_ip(),
|
||||
&nat_info,
|
||||
info.virtual_ip,
|
||||
)
|
||||
.unwrap();
|
||||
let _ = sender.send_main(packet.buffer(), current_device.connect_server);
|
||||
}
|
||||
tokio::time::sleep(sleep_time).await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn punch_packet(
|
||||
client_cipher: &Cipher,
|
||||
virtual_ip: Ipv4Addr,
|
||||
nat_info: &NatInfo,
|
||||
dest: Ipv4Addr,
|
||||
) -> crate::Result<NetPacket<Vec<u8>>> {
|
||||
let mut punch_reply = PunchInfo::new();
|
||||
punch_reply.reply = false;
|
||||
punch_reply.public_ip_list = nat_info
|
||||
.public_ips
|
||||
.iter()
|
||||
.map(|ip| u32::from_be_bytes(ip.octets()))
|
||||
.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.local_ip = u32::from(nat_info.local_ipv4().unwrap_or(Ipv4Addr::UNSPECIFIED));
|
||||
punch_reply.local_port = nat_info.udp_port as u32;
|
||||
punch_reply.tcp_port = nat_info.tcp_port as u32;
|
||||
if let Some(ipv6) = nat_info.ipv6 {
|
||||
punch_reply.ipv6_port = nat_info.udp_port as u32;
|
||||
punch_reply.ipv6 = ipv6.octets().to_vec();
|
||||
}
|
||||
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_encrypt(vec![0u8; 12 + bytes.len() + ENCRYPTION_RESERVED])?;
|
||||
net_packet.set_version(Version::V1);
|
||||
net_packet.set_protocol(Protocol::OtherTurn);
|
||||
net_packet.set_transport_protocol(other_turn_packet::Protocol::Punch.into());
|
||||
net_packet.first_set_ttl(MAX_TTL);
|
||||
net_packet.set_source(virtual_ip);
|
||||
net_packet.set_destination(dest);
|
||||
net_packet.set_payload(&bytes)?;
|
||||
client_cipher.encrypt_ipv4(&mut net_packet)?;
|
||||
Ok(net_packet)
|
||||
}
|
||||
@@ -1,714 +0,0 @@
|
||||
use std::collections::HashMap;
|
||||
use std::net::{Ipv4Addr, Ipv6Addr};
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use crossbeam_utils::atomic::AtomicCell;
|
||||
use parking_lot::{Mutex, RwLock};
|
||||
use protobuf::Message;
|
||||
use tokio::sync::mpsc::Sender;
|
||||
|
||||
use packet::icmp::{icmp, Kind};
|
||||
use packet::ip::ipv4;
|
||||
use packet::ip::ipv4::packet::IpV4Packet;
|
||||
|
||||
use crate::channel::channel::Context;
|
||||
use crate::channel::punch::{NatInfo, NatType};
|
||||
use crate::channel::{Route, RouteKey};
|
||||
use crate::cipher::{Cipher, RsaCipher};
|
||||
use crate::error::Error;
|
||||
use crate::external_route::AllowExternalRoute;
|
||||
use crate::handle::handshake_handler::secret_handshake_req;
|
||||
use crate::handle::registration_handler::Register;
|
||||
use crate::handle::{ConnectStatus, CurrentDeviceInfo, PeerDeviceInfo, PeerDeviceStatus};
|
||||
use crate::igmp_server::IgmpServer;
|
||||
#[cfg(feature = "ip_proxy")]
|
||||
use crate::ip_proxy::{IpProxyMap, ProxyHandler};
|
||||
use crate::nat;
|
||||
use crate::nat::NatTest;
|
||||
use crate::proto::message::{DeviceList, PunchInfo, PunchNatType, RegistrationResponse};
|
||||
use crate::protocol::body::ENCRYPTION_RESERVED;
|
||||
use crate::protocol::control_packet::ControlPacket;
|
||||
use crate::protocol::error_packet::InErrorPacket;
|
||||
use crate::protocol::{
|
||||
control_packet, ip_turn_packet, other_turn_packet, service_packet, NetPacket, Protocol,
|
||||
Version, MAX_TTL,
|
||||
};
|
||||
use crate::tun_tap_device::DeviceWriter;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct ChannelDataHandler {
|
||||
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
|
||||
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
|
||||
register: Arc<Register>,
|
||||
nat_test: NatTest,
|
||||
igmp_server: Option<IgmpServer>,
|
||||
device_writer: DeviceWriter,
|
||||
connect_status: Arc<AtomicCell<ConnectStatus>>,
|
||||
peer_nat_info_map: Arc<RwLock<HashMap<Ipv4Addr, NatInfo>>>,
|
||||
#[cfg(feature = "ip_proxy")]
|
||||
ip_proxy_map: Option<IpProxyMap>,
|
||||
out_external_route: AllowExternalRoute,
|
||||
cone_sender: Sender<(Ipv4Addr, NatInfo)>,
|
||||
symmetric_sender: Sender<(Ipv4Addr, NatInfo)>,
|
||||
client_cipher: Cipher,
|
||||
server_cipher: Cipher,
|
||||
rsa_cipher: Option<RsaCipher>,
|
||||
relay: bool,
|
||||
token: String,
|
||||
time: Arc<AtomicCell<Instant>>,
|
||||
pub head_reserve: usize,
|
||||
}
|
||||
|
||||
impl ChannelDataHandler {
|
||||
pub fn new(
|
||||
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
|
||||
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
|
||||
register: Arc<Register>,
|
||||
nat_test: NatTest,
|
||||
igmp_server: Option<IgmpServer>,
|
||||
device_writer: DeviceWriter,
|
||||
connect_status: Arc<AtomicCell<ConnectStatus>>,
|
||||
peer_nat_info_map: Arc<RwLock<HashMap<Ipv4Addr, NatInfo>>>,
|
||||
#[cfg(feature = "ip_proxy")] ip_proxy_map: Option<IpProxyMap>,
|
||||
out_external_route: AllowExternalRoute,
|
||||
cone_sender: Sender<(Ipv4Addr, NatInfo)>,
|
||||
symmetric_sender: Sender<(Ipv4Addr, NatInfo)>,
|
||||
client_cipher: Cipher,
|
||||
server_cipher: Cipher,
|
||||
rsa_cipher: Option<RsaCipher>,
|
||||
relay: bool,
|
||||
token: String,
|
||||
head_reserve: usize,
|
||||
) -> Self {
|
||||
Self {
|
||||
current_device,
|
||||
device_list,
|
||||
register,
|
||||
nat_test,
|
||||
igmp_server,
|
||||
device_writer,
|
||||
connect_status,
|
||||
peer_nat_info_map,
|
||||
#[cfg(feature = "ip_proxy")]
|
||||
ip_proxy_map,
|
||||
out_external_route,
|
||||
cone_sender,
|
||||
symmetric_sender,
|
||||
client_cipher,
|
||||
server_cipher,
|
||||
rsa_cipher,
|
||||
relay,
|
||||
token,
|
||||
time: Arc::new(AtomicCell::new(Instant::now())),
|
||||
head_reserve,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl ChannelDataHandler {
|
||||
pub fn handle(
|
||||
&self,
|
||||
buf: &mut [u8],
|
||||
start: usize,
|
||||
end: usize,
|
||||
route_key: RouteKey,
|
||||
context: &Context,
|
||||
) {
|
||||
assert_eq!(start, 14);
|
||||
match self.handle0(&mut buf[..end], &route_key, context) {
|
||||
Ok(_) => {}
|
||||
Err(e) => {
|
||||
log::warn!("{:?}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
fn handle0(
|
||||
&self,
|
||||
buf: &mut [u8],
|
||||
route_key: &RouteKey,
|
||||
context: &Context,
|
||||
) -> crate::Result<()> {
|
||||
let mut net_packet = NetPacket::new(&mut buf[14..])?;
|
||||
if net_packet.ttl() == 0 || net_packet.source_ttl() < net_packet.ttl() {
|
||||
return Ok(());
|
||||
}
|
||||
let source = net_packet.source();
|
||||
context.update_read_time(&source, route_key);
|
||||
let current_device = self.current_device.load();
|
||||
let destination = net_packet.destination();
|
||||
let not_broadcast = !destination.is_broadcast()
|
||||
&& !destination.is_multicast()
|
||||
&& destination != current_device.broadcast_address;
|
||||
if current_device.virtual_ip() != destination
|
||||
&& !net_packet.is_gateway()
|
||||
&& not_broadcast
|
||||
&& !destination.is_unspecified()
|
||||
{
|
||||
//校验指纹，不需要解密
|
||||
self.client_cipher.check_finger(&net_packet)?;
|
||||
net_packet.set_ttl(net_packet.ttl() - 1);
|
||||
let ttl = net_packet.ttl();
|
||||
if ttl > 0 {
|
||||
// 转发
|
||||
if let Some(route) = context.route_one(&destination) {
|
||||
if route.metric <= net_packet.ttl() {
|
||||
context.try_send_by_key(net_packet.buffer(), &route.route_key())?;
|
||||
}
|
||||
} else if (ttl > 1 || destination == current_device.virtual_gateway())
|
||||
&& source != current_device.virtual_gateway()
|
||||
{
|
||||
//网关默认要转发一次，生存时间不够的发到网关也会被丢弃
|
||||
context.send_main(net_packet.buffer(), current_device.connect_server)?;
|
||||
}
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
if net_packet.is_gateway() {
|
||||
if net_packet.protocol() == Protocol::Error
|
||||
&& net_packet.transport_protocol()
|
||||
== crate::protocol::error_packet::Protocol::NoKey.into()
|
||||
{
|
||||
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(
|
||||
context,
|
||||
current_device.connect_server,
|
||||
rsa_cipher,
|
||||
&self.server_cipher,
|
||||
self.token.clone(),
|
||||
route_key,
|
||||
)?;
|
||||
}
|
||||
} else {
|
||||
//服务端解密
|
||||
self.server_cipher.decrypt_ipv4(&mut net_packet)?;
|
||||
let data_len = net_packet.data_len();
|
||||
self.server_packet_handle(context, current_device, buf, data_len, route_key)?;
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
self.client_cipher.decrypt_ipv4(&mut net_packet)?;
|
||||
match net_packet.protocol() {
|
||||
Protocol::IpTurn => {
|
||||
match ip_turn_packet::Protocol::from(net_packet.transport_protocol()) {
|
||||
ip_turn_packet::Protocol::Ipv4 => {
|
||||
let mut ipv4 = IpV4Packet::new(net_packet.payload_mut())?;
|
||||
match ipv4.protocol() {
|
||||
ipv4::protocol::Protocol::Igmp => {
|
||||
if let Some(igmp_server) = &self.igmp_server {
|
||||
igmp_server.handle(ipv4.payload(), source)?;
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
ipv4::protocol::Protocol::Icmp => {
|
||||
if ipv4.destination_ip() == destination {
|
||||
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.client_cipher.encrypt_ipv4(&mut net_packet)?;
|
||||
context.try_send_by_key(net_packet.buffer(), route_key)?;
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
if not_broadcast && ipv4.destination_ip() != destination {
|
||||
if self.out_external_route.allow(&ipv4.destination_ip()) {
|
||||
#[cfg(feature = "ip_proxy")]
|
||||
if let Some(ip_proxy_map) = &self.ip_proxy_map {
|
||||
if ip_proxy_map.recv_handle(&mut ipv4, source, destination)? {
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
} else {
|
||||
log::warn!(
|
||||
"没有ip代理规则{:?}:{}->{}->{}",
|
||||
ipv4.protocol(),
|
||||
source,
|
||||
destination,
|
||||
ipv4.destination_ip()
|
||||
);
|
||||
return Err(Error::Warn("没有ip代理规则".to_string()));
|
||||
}
|
||||
}
|
||||
|
||||
//传输协议12字节
|
||||
self.device_writer.write_ipv4(&mut buf[12..])?;
|
||||
return Ok(());
|
||||
}
|
||||
ip_turn_packet::Protocol::Ipv4Broadcast => {
|
||||
//客户端不帮忙转发广播包，所以不会出现这种类型的数据
|
||||
}
|
||||
ip_turn_packet::Protocol::Unknown(_) => {}
|
||||
}
|
||||
}
|
||||
Protocol::Service => {}
|
||||
Protocol::Error => {}
|
||||
Protocol::Control => {
|
||||
self.control(context, current_device, source, net_packet, route_key)?;
|
||||
}
|
||||
Protocol::OtherTurn => {
|
||||
self.other_turn(context, current_device, source, net_packet, route_key)?;
|
||||
}
|
||||
Protocol::UnKnow(e) => {
|
||||
log::info!("不支持的协议:{}", e);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn pong_packet(
|
||||
&self,
|
||||
gateway: bool,
|
||||
metric: u8,
|
||||
context: &Context,
|
||||
current_device: CurrentDeviceInfo,
|
||||
source: Ipv4Addr,
|
||||
pong_packet: control_packet::PongPacket<&[u8]>,
|
||||
route_key: &RouteKey,
|
||||
) -> crate::Result<()> {
|
||||
let current_time = crate::handle::now_time() as u16;
|
||||
if current_time < pong_packet.time() {
|
||||
return Ok(());
|
||||
}
|
||||
let rt = (current_time - pong_packet.time()) as i64;
|
||||
let route = Route::from(*route_key, metric, rt);
|
||||
context.add_route(source, route);
|
||||
if gateway {
|
||||
let epoch = self.device_list.lock().0;
|
||||
if pong_packet.epoch() != epoch {
|
||||
let mut poll_device = NetPacket::new_encrypt([0; 12 + ENCRYPTION_RESERVED])?;
|
||||
poll_device.set_source(current_device.virtual_ip());
|
||||
poll_device.set_destination(source);
|
||||
poll_device.set_version(Version::V1);
|
||||
poll_device.set_gateway_flag(true);
|
||||
poll_device.first_set_ttl(MAX_TTL);
|
||||
poll_device.set_protocol(Protocol::Service);
|
||||
poll_device.set_transport_protocol(service_packet::Protocol::PollDeviceList.into());
|
||||
self.server_cipher.encrypt_ipv4(&mut poll_device)?;
|
||||
context.send_main(poll_device.buffer(), current_device.connect_server)?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
fn control(
|
||||
&self,
|
||||
context: &Context,
|
||||
current_device: CurrentDeviceInfo,
|
||||
source: Ipv4Addr,
|
||||
mut net_packet: NetPacket<&mut [u8]>,
|
||||
route_key: &RouteKey,
|
||||
) -> crate::Result<()> {
|
||||
let metric = net_packet.source_ttl() - net_packet.ttl() + 1;
|
||||
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.client_cipher.encrypt_ipv4(&mut net_packet)?;
|
||||
context.try_send_by_key(net_packet.buffer(), route_key)?;
|
||||
let route = Route::from(*route_key, metric, 199);
|
||||
context.add_route_if_absent(source, route);
|
||||
}
|
||||
ControlPacket::PongPacket(pong_packet) => {
|
||||
self.pong_packet(
|
||||
false,
|
||||
metric,
|
||||
context,
|
||||
current_device,
|
||||
source,
|
||||
pong_packet,
|
||||
route_key,
|
||||
)?;
|
||||
}
|
||||
ControlPacket::PunchRequest => {
|
||||
if self.relay {
|
||||
return Ok(());
|
||||
}
|
||||
//回应
|
||||
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.client_cipher.encrypt_ipv4(&mut net_packet)?;
|
||||
context.try_send_by_key(net_packet.buffer(), route_key)?;
|
||||
let route = Route::from(*route_key, 1, 199);
|
||||
context.add_route_if_absent(source, route);
|
||||
}
|
||||
ControlPacket::PunchResponse => {
|
||||
if self.relay {
|
||||
return Ok(());
|
||||
}
|
||||
let route = Route::from(*route_key, 1, 199);
|
||||
context.add_route_if_absent(source, route);
|
||||
}
|
||||
ControlPacket::AddrRequest => match route_key.addr.ip() {
|
||||
std::net::IpAddr::V4(ipv4) => {
|
||||
let mut packet = NetPacket::new_encrypt([0; 12 + 6 + ENCRYPTION_RESERVED])?;
|
||||
packet.set_version(Version::V1);
|
||||
packet.set_protocol(Protocol::Control);
|
||||
packet.set_transport_protocol(control_packet::Protocol::AddrResponse.into());
|
||||
packet.first_set_ttl(MAX_TTL);
|
||||
packet.set_source(current_device.virtual_ip());
|
||||
packet.set_destination(source);
|
||||
let mut addr_packet = control_packet::AddrPacket::new(packet.payload_mut())?;
|
||||
addr_packet.set_ipv4(ipv4);
|
||||
addr_packet.set_port(route_key.addr.port());
|
||||
self.client_cipher.encrypt_ipv4(&mut packet)?;
|
||||
context.try_send_by_key(packet.buffer(), route_key)?;
|
||||
}
|
||||
std::net::IpAddr::V6(_) => {}
|
||||
},
|
||||
ControlPacket::AddrResponse(addr_packet) => self
|
||||
.nat_test
|
||||
.update_addr(addr_packet.ipv4(), addr_packet.port()),
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
fn other_turn(
|
||||
&self,
|
||||
context: &Context,
|
||||
current_device: CurrentDeviceInfo,
|
||||
source: Ipv4Addr,
|
||||
net_packet: NetPacket<&mut [u8]>,
|
||||
route_key: &RouteKey,
|
||||
) -> crate::Result<()> {
|
||||
if self.relay {
|
||||
return Ok(());
|
||||
}
|
||||
match other_turn_packet::Protocol::from(net_packet.transport_protocol()) {
|
||||
other_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| Ipv4Addr::from(v.to_be_bytes()))
|
||||
.collect();
|
||||
let local_ipv4 = Some(Ipv4Addr::from(punch_info.local_ip.to_be_bytes()));
|
||||
let udp_port = punch_info.local_port as u16;
|
||||
let tcp_port = punch_info.tcp_port as u16;
|
||||
let ipv6 = if punch_info.ipv6.len() == 16 {
|
||||
let ipv6: [u8; 16] = punch_info.ipv6.try_into().unwrap();
|
||||
Some(Ipv6Addr::from(ipv6))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
let peer_nat_info = NatInfo::new(
|
||||
public_ips,
|
||||
punch_info.public_port as u16,
|
||||
punch_info.public_port_range as u16,
|
||||
local_ipv4,
|
||||
ipv6,
|
||||
udp_port,
|
||||
tcp_port,
|
||||
punch_info.nat_type.enum_value_or_default().into(),
|
||||
);
|
||||
{
|
||||
let peer_nat_info = peer_nat_info.clone();
|
||||
self.peer_nat_info_map.write().insert(source, peer_nat_info);
|
||||
}
|
||||
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(|ip| u32::from_be_bytes(ip.octets()))
|
||||
.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));
|
||||
punch_reply.local_ip =
|
||||
u32::from(nat_info.local_ipv4().unwrap_or(Ipv4Addr::UNSPECIFIED));
|
||||
punch_reply.local_port = nat_info.udp_port as u32;
|
||||
if let Some(ipv6) = nat_info.ipv6() {
|
||||
punch_reply.ipv6 = ipv6.octets().to_vec();
|
||||
punch_reply.ipv6_port = nat_info.udp_port as u32;
|
||||
}
|
||||
let bytes = punch_reply.write_to_bytes()?;
|
||||
let mut punch_packet =
|
||||
NetPacket::new_encrypt(vec![0u8; 12 + bytes.len() + ENCRYPTION_RESERVED])?;
|
||||
punch_packet.set_version(Version::V1);
|
||||
punch_packet.set_protocol(Protocol::OtherTurn);
|
||||
punch_packet.set_transport_protocol(other_turn_packet::Protocol::Punch.into());
|
||||
punch_packet.first_set_ttl(MAX_TTL);
|
||||
punch_packet.set_source(current_device.virtual_ip());
|
||||
punch_packet.set_destination(source);
|
||||
punch_packet.set_payload(&bytes)?;
|
||||
if self.punch(source, peer_nat_info) {
|
||||
self.client_cipher.encrypt_ipv4(&mut punch_packet)?;
|
||||
context.try_send_by_key(punch_packet.buffer(), route_key)?;
|
||||
}
|
||||
} else {
|
||||
self.punch(source, peer_nat_info);
|
||||
}
|
||||
}
|
||||
other_turn_packet::Protocol::Unknown(e) => {
|
||||
log::warn!("不支持的转发协议 {:?},source:{:?}", e, source);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
fn punch(&self, peer_ip: Ipv4Addr, peer_nat_info: NatInfo) -> bool {
|
||||
match peer_nat_info.nat_type {
|
||||
NatType::Symmetric => self
|
||||
.symmetric_sender
|
||||
.try_send((peer_ip, peer_nat_info))
|
||||
.is_ok(),
|
||||
NatType::Cone => self.cone_sender.try_send((peer_ip, peer_nat_info)).is_ok(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 处理服务端数据
|
||||
impl ChannelDataHandler {
|
||||
fn server_packet_handle(
|
||||
&self,
|
||||
context: &Context,
|
||||
current_device: CurrentDeviceInfo,
|
||||
buf: &mut [u8],
|
||||
data_len: usize,
|
||||
route_key: &RouteKey,
|
||||
) -> crate::Result<()> {
|
||||
let net_packet = NetPacket::new0(data_len, &buf[14..])?;
|
||||
let source = net_packet.source();
|
||||
match net_packet.protocol() {
|
||||
Protocol::Service => {
|
||||
self.service(context, current_device, net_packet, route_key)?;
|
||||
}
|
||||
Protocol::Error => {
|
||||
self.error(context, current_device, source, net_packet, route_key)?;
|
||||
}
|
||||
Protocol::Control => {
|
||||
self.control_gateway(context, current_device, net_packet, route_key)?;
|
||||
}
|
||||
Protocol::IpTurn => {
|
||||
match ip_turn_packet::Protocol::from(net_packet.transport_protocol()) {
|
||||
ip_turn_packet::Protocol::Ipv4 => {
|
||||
let ipv4 = IpV4Packet::new(net_packet.payload())?;
|
||||
match ipv4.protocol() {
|
||||
ipv4::protocol::Protocol::Igmp => {
|
||||
if let Some(igmp_server) = &self.igmp_server {
|
||||
igmp_server.handle(ipv4.payload(), source)?;
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
ipv4::protocol::Protocol::Icmp => {
|
||||
if ipv4.destination_ip() == current_device.virtual_ip {
|
||||
let icmp_packet = icmp::IcmpPacket::new(ipv4.payload())?;
|
||||
if icmp_packet.kind() == Kind::EchoReply {
|
||||
self.device_writer.write_ipv4(&mut buf[12..])?;
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
ip_turn_packet::Protocol::Ipv4Broadcast => {}
|
||||
ip_turn_packet::Protocol::Unknown(_) => {}
|
||||
}
|
||||
}
|
||||
Protocol::OtherTurn => {}
|
||||
Protocol::UnKnow(_) => {}
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
fn control_gateway(
|
||||
&self,
|
||||
context: &Context,
|
||||
current_device: CurrentDeviceInfo,
|
||||
net_packet: NetPacket<&[u8]>,
|
||||
route_key: &RouteKey,
|
||||
) -> crate::Result<()> {
|
||||
match ControlPacket::new(net_packet.transport_protocol(), net_packet.payload())? {
|
||||
ControlPacket::PongPacket(pong_packet) => {
|
||||
let metric = net_packet.source_ttl() - net_packet.ttl() + 1;
|
||||
self.pong_packet(
|
||||
true,
|
||||
metric,
|
||||
context,
|
||||
current_device,
|
||||
net_packet.source(),
|
||||
pong_packet,
|
||||
route_key,
|
||||
)?;
|
||||
}
|
||||
ControlPacket::AddrResponse(addr_packet) => self
|
||||
.nat_test
|
||||
.update_addr(addr_packet.ipv4(), addr_packet.port()),
|
||||
_ => {}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
fn service(
|
||||
&self,
|
||||
context: &Context,
|
||||
current_device: CurrentDeviceInfo,
|
||||
net_packet: NetPacket<&[u8]>,
|
||||
route_key: &RouteKey,
|
||||
) -> crate::Result<()> {
|
||||
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())?;
|
||||
|
||||
if self.nat_test.can_update() {
|
||||
let context = context.clone();
|
||||
let nat_test = self.nat_test.clone();
|
||||
std::thread::spawn(move || {
|
||||
tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
.unwrap()
|
||||
.block_on(async move {
|
||||
let local_ipv4 = nat::local_ipv4();
|
||||
let ipv6 = nat::local_ipv6();
|
||||
let udp_port = nat_test.nat_info().udp_port;
|
||||
let tcp_port = nat_test.nat_info().tcp_port;
|
||||
let nat_info = nat_test
|
||||
.re_test(
|
||||
Ipv4Addr::from(response.public_ip),
|
||||
response.public_port as u16,
|
||||
local_ipv4,
|
||||
ipv6,
|
||||
udp_port,
|
||||
tcp_port,
|
||||
)
|
||||
.await;
|
||||
context.switch(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);
|
||||
#[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))]
|
||||
let old_netmask = current_device.virtual_netmask;
|
||||
#[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))]
|
||||
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);
|
||||
#[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))]
|
||||
self.device_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: Vec<PeerDeviceInfo> = device_list_t
|
||||
.device_info_list
|
||||
.into_iter()
|
||||
.map(|info| {
|
||||
PeerDeviceInfo::new(
|
||||
Ipv4Addr::from(info.virtual_ip),
|
||||
info.name,
|
||||
info.device_status as u8,
|
||||
info.client_secret,
|
||||
)
|
||||
})
|
||||
.collect();
|
||||
let route = Route::from(*route_key, 2, 199);
|
||||
for x in &ip_list {
|
||||
if x.status == PeerDeviceStatus::Online {
|
||||
context.add_route_if_absent(x.virtual_ip, route);
|
||||
}
|
||||
}
|
||||
let mut dev = self.device_list.lock();
|
||||
if dev.0 != device_list_t.epoch as u16 {
|
||||
dev.0 = device_list_t.epoch as u16;
|
||||
dev.1 = ip_list;
|
||||
}
|
||||
}
|
||||
service_packet::Protocol::Unknown(u) => {
|
||||
log::warn!("未知服务协议:{}", u);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
fn error(
|
||||
&self,
|
||||
_context: &Context,
|
||||
current_device: CurrentDeviceInfo,
|
||||
_source: Ipv4Addr,
|
||||
net_packet: NetPacket<&[u8]>,
|
||||
_route_key: &RouteKey,
|
||||
) -> crate::Result<()> {
|
||||
log::info!("current_device:{:?}", current_device);
|
||||
match InErrorPacket::new(net_packet.transport_protocol(), net_packet.payload())? {
|
||||
InErrorPacket::TokenError => {
|
||||
return Err(Error::Stop("Token error".to_string()));
|
||||
}
|
||||
InErrorPacket::Disconnect => {
|
||||
{
|
||||
//掉线epoch要归零
|
||||
let mut dev = self.device_list.lock();
|
||||
dev.0 = 0;
|
||||
}
|
||||
|
||||
self.connect_status.store(ConnectStatus::Connecting);
|
||||
self.register.fast_register(current_device.virtual_ip)?;
|
||||
}
|
||||
InErrorPacket::AddressExhausted => {
|
||||
//地址用尽
|
||||
return Err(Error::Stop("IP address has been exhausted".to_string()));
|
||||
}
|
||||
InErrorPacket::OtherError(e) => {
|
||||
log::error!("OtherError {:?}", e.message());
|
||||
}
|
||||
InErrorPacket::IpAlreadyExists => {
|
||||
log::error!("IpAlreadyExists");
|
||||
}
|
||||
InErrorPacket::InvalidIp => {
|
||||
log::error!("InvalidIp");
|
||||
}
|
||||
InErrorPacket::NoKey => {}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -1,262 +0,0 @@
|
||||
use crossbeam_utils::atomic::AtomicCell;
|
||||
use std::io::{Read, Write};
|
||||
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV6};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use crate::channel::sender::ChannelSender;
|
||||
use crate::cipher::Cipher;
|
||||
use crate::handle::PeerDeviceInfo;
|
||||
use protobuf::Message;
|
||||
use std::net::TcpStream;
|
||||
use std::net::UdpSocket;
|
||||
|
||||
use crate::proto::message::{RegistrationRequest, RegistrationResponse};
|
||||
use crate::protocol::body::ENCRYPTION_RESERVED;
|
||||
use crate::protocol::error_packet::InErrorPacket;
|
||||
use crate::protocol::{service_packet, NetPacket, Protocol, Version, MAX_TTL};
|
||||
|
||||
pub enum ReqEnum {
|
||||
TokenError,
|
||||
AddressExhausted,
|
||||
IpAlreadyExists,
|
||||
InvalidIp,
|
||||
Timeout,
|
||||
ServerError(String),
|
||||
Other(String),
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct RegResponse {
|
||||
pub virtual_ip: Ipv4Addr,
|
||||
pub virtual_gateway: Ipv4Addr,
|
||||
pub virtual_netmask: Ipv4Addr,
|
||||
pub epoch: u16,
|
||||
pub device_info_list: Vec<PeerDeviceInfo>,
|
||||
pub public_ip: Ipv4Addr,
|
||||
pub public_port: u16,
|
||||
}
|
||||
|
||||
///向中继服务器注册，token标识一个虚拟网关，device_id防止多次注册时得到的ip不一致
|
||||
pub fn registration(
|
||||
main_channel: &UdpSocket,
|
||||
main_tcp_channel: Option<&mut TcpStream>,
|
||||
server_cipher: &Cipher,
|
||||
server_address: SocketAddr,
|
||||
token: String,
|
||||
device_id: String,
|
||||
name: String,
|
||||
ip: Ipv4Addr,
|
||||
client_secret: bool,
|
||||
) -> Result<RegResponse, ReqEnum> {
|
||||
let request_packet = registration_request_packet(
|
||||
server_cipher,
|
||||
token.clone(),
|
||||
device_id.clone(),
|
||||
name.clone(),
|
||||
ip,
|
||||
false,
|
||||
false,
|
||||
client_secret,
|
||||
)
|
||||
.unwrap();
|
||||
let buf = request_packet.buffer();
|
||||
let mut recv_buf = [0u8; 10240];
|
||||
let recv_buf = if let Some(main_tcp_channel) = main_tcp_channel {
|
||||
let mut vec = vec![0; 4 + buf.len()];
|
||||
let len = buf.len();
|
||||
vec[2] = (len >> 8) as u8;
|
||||
vec[3] = (len & 0xFF) as u8;
|
||||
vec[4..].copy_from_slice(buf);
|
||||
if let Err(e) = main_tcp_channel.write_all(&vec) {
|
||||
return Err(ReqEnum::Other(format!("send error:{}", e)));
|
||||
}
|
||||
if let Err(e) = main_tcp_channel.read_exact(&mut recv_buf[..4]) {
|
||||
return Err(ReqEnum::Other(format!("read error:{}", e)));
|
||||
}
|
||||
let len = 4 + (((recv_buf[2] as u16) << 8) | recv_buf[3] as u16) as usize;
|
||||
if len > recv_buf.len() {
|
||||
return Err(ReqEnum::Other("too long".to_string()));
|
||||
}
|
||||
if let Err(e) = main_tcp_channel.read_exact(&mut recv_buf[4..len]) {
|
||||
return Err(ReqEnum::Other(format!("read error:{}", e)));
|
||||
}
|
||||
&mut recv_buf[4..len]
|
||||
} else {
|
||||
let server_address = match server_address {
|
||||
SocketAddr::V4(ipv4) => SocketAddr::V6(SocketAddrV6::new(
|
||||
ipv4.ip().to_ipv6_mapped(),
|
||||
ipv4.port(),
|
||||
0,
|
||||
0,
|
||||
)),
|
||||
SocketAddr::V6(_) => server_address,
|
||||
};
|
||||
if let Err(e) = main_channel.send_to(buf, server_address) {
|
||||
return Err(ReqEnum::Other(format!("send error:{}", e)));
|
||||
}
|
||||
match main_channel.recv_from(&mut recv_buf) {
|
||||
Ok((len, addr)) => {
|
||||
if server_address != addr {
|
||||
log::warn!("请求{:?}和响应{:?}地址不一致", server_address, addr);
|
||||
}
|
||||
&mut recv_buf[..len]
|
||||
}
|
||||
Err(e) => {
|
||||
return Err(ReqEnum::Other(format!("receiver error:{}", e)));
|
||||
}
|
||||
}
|
||||
};
|
||||
let mut net_packet = match NetPacket::new(recv_buf) {
|
||||
Ok(net_packet) => net_packet,
|
||||
Err(e) => {
|
||||
return Err(ReqEnum::ServerError(format!("{}", e)));
|
||||
}
|
||||
};
|
||||
if let Err(e) = server_cipher.decrypt_ipv4(&mut net_packet) {
|
||||
return Err(ReqEnum::ServerError(format!("decrypt_ipv4 {}", e)));
|
||||
}
|
||||
match net_packet.protocol() {
|
||||
Protocol::Service => {
|
||||
match service_packet::Protocol::from(net_packet.transport_protocol()) {
|
||||
service_packet::Protocol::RegistrationResponse => {
|
||||
match RegistrationResponse::parse_from_bytes(net_packet.payload()) {
|
||||
Ok(response) => {
|
||||
let device_info_list: Vec<PeerDeviceInfo> = response
|
||||
.device_info_list
|
||||
.into_iter()
|
||||
.map(|info| {
|
||||
PeerDeviceInfo::new(
|
||||
Ipv4Addr::from(info.virtual_ip),
|
||||
info.name,
|
||||
info.device_status as u8,
|
||||
info.client_secret,
|
||||
)
|
||||
})
|
||||
.collect();
|
||||
Ok(RegResponse {
|
||||
virtual_ip: Ipv4Addr::from(response.virtual_ip),
|
||||
virtual_gateway: Ipv4Addr::from(response.virtual_gateway),
|
||||
virtual_netmask: Ipv4Addr::from(response.virtual_netmask),
|
||||
epoch: response.epoch as u16,
|
||||
device_info_list,
|
||||
public_ip: Ipv4Addr::from(response.public_ip),
|
||||
public_port: response.public_port as u16,
|
||||
})
|
||||
}
|
||||
Err(_) => Err(ReqEnum::ServerError("invalid data".to_string())),
|
||||
}
|
||||
}
|
||||
_ => Err(ReqEnum::ServerError("invalid data".to_string())),
|
||||
}
|
||||
}
|
||||
Protocol::Error => {
|
||||
match InErrorPacket::new(net_packet.transport_protocol(), net_packet.payload()) {
|
||||
Ok(e) => match e {
|
||||
InErrorPacket::TokenError => Err(ReqEnum::TokenError),
|
||||
InErrorPacket::Disconnect => {
|
||||
Err(ReqEnum::ServerError("disconnect".to_string()))
|
||||
}
|
||||
InErrorPacket::AddressExhausted => Err(ReqEnum::AddressExhausted),
|
||||
InErrorPacket::OtherError(e) => match e.message() {
|
||||
Ok(str) => Err(ReqEnum::ServerError(str)),
|
||||
Err(e) => Err(ReqEnum::Other(format!("{}", e))),
|
||||
},
|
||||
InErrorPacket::IpAlreadyExists => Err(ReqEnum::IpAlreadyExists),
|
||||
InErrorPacket::InvalidIp => Err(ReqEnum::InvalidIp),
|
||||
InErrorPacket::NoKey => Err(ReqEnum::ServerError("no key".to_string())),
|
||||
},
|
||||
Err(e) => Err(ReqEnum::Other(format!("{}", e))),
|
||||
}
|
||||
}
|
||||
_ => Err(ReqEnum::ServerError("invalid data".to_string())),
|
||||
}
|
||||
}
|
||||
|
||||
fn registration_request_packet(
|
||||
server_cipher: &Cipher,
|
||||
token: String,
|
||||
device_id: String,
|
||||
name: String,
|
||||
ip: Ipv4Addr,
|
||||
is_fast: bool,
|
||||
allow_ip_change: bool,
|
||||
client_secret: bool,
|
||||
) -> crate::Result<NetPacket<Vec<u8>>> {
|
||||
let mut request = RegistrationRequest::new();
|
||||
request.token = token;
|
||||
request.device_id = device_id;
|
||||
request.name = name;
|
||||
request.virtual_ip = ip.into();
|
||||
request.allow_ip_change = allow_ip_change;
|
||||
request.is_fast = is_fast;
|
||||
request.version = crate::VNT_VERSION.to_string();
|
||||
request.client_secret = client_secret;
|
||||
let bytes = request.write_to_bytes()?;
|
||||
let buf = vec![0u8; 12 + bytes.len() + ENCRYPTION_RESERVED];
|
||||
let mut net_packet = NetPacket::new_encrypt(buf)?;
|
||||
net_packet.set_version(Version::V1);
|
||||
net_packet.set_gateway_flag(true);
|
||||
net_packet.set_protocol(Protocol::Service);
|
||||
net_packet.set_transport_protocol(service_packet::Protocol::RegistrationRequest.into());
|
||||
net_packet.first_set_ttl(MAX_TTL);
|
||||
net_packet.set_payload(&bytes)?;
|
||||
server_cipher.encrypt_ipv4(&mut net_packet)?;
|
||||
Ok(net_packet)
|
||||
}
|
||||
|
||||
pub struct Register {
|
||||
server_cipher: Cipher,
|
||||
sender: ChannelSender,
|
||||
server_address: SocketAddr,
|
||||
token: String,
|
||||
device_id: String,
|
||||
name: String,
|
||||
time: AtomicCell<Instant>,
|
||||
client_secret: bool,
|
||||
}
|
||||
|
||||
impl Register {
|
||||
pub fn new(
|
||||
server_cipher: Cipher,
|
||||
sender: ChannelSender,
|
||||
server_address: SocketAddr,
|
||||
token: String,
|
||||
device_id: String,
|
||||
name: String,
|
||||
client_secret: bool,
|
||||
) -> Self {
|
||||
Self {
|
||||
server_cipher,
|
||||
sender,
|
||||
server_address,
|
||||
token,
|
||||
device_id,
|
||||
name,
|
||||
time: AtomicCell::new(Instant::now()),
|
||||
client_secret,
|
||||
}
|
||||
}
|
||||
pub fn fast_register(&self, ip: Ipv4Addr) -> crate::Result<()> {
|
||||
let last = self.time.load();
|
||||
if last.elapsed() < Duration::from_secs(3)
|
||||
|| self.time.compare_exchange(last, Instant::now()).is_err()
|
||||
{
|
||||
//短时间不重复注册
|
||||
return Ok(());
|
||||
}
|
||||
log::info!("重新连接");
|
||||
let request_packet = registration_request_packet(
|
||||
&self.server_cipher,
|
||||
self.token.clone(),
|
||||
self.device_id.clone(),
|
||||
self.name.clone(),
|
||||
ip,
|
||||
false,
|
||||
true,
|
||||
self.client_secret,
|
||||
)?;
|
||||
let buf = request_packet.buffer();
|
||||
self.sender.send_main(buf, self.server_address)?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -1,260 +0,0 @@
|
||||
use std::sync::Arc;
|
||||
use std::{io, thread};
|
||||
|
||||
use crossbeam_utils::atomic::AtomicCell;
|
||||
|
||||
use packet::arp::arp::ArpPacket;
|
||||
use packet::ethernet;
|
||||
use packet::ethernet::packet::EthernetPacket;
|
||||
use packet::icmp::icmp::IcmpPacket;
|
||||
use packet::icmp::Kind;
|
||||
use packet::ip::ipv4;
|
||||
use packet::ip::ipv4::packet::IpV4Packet;
|
||||
|
||||
use crate::channel::sender::ChannelSender;
|
||||
use crate::cipher::Cipher;
|
||||
use crate::core::status::VntWorker;
|
||||
use crate::external_route::ExternalRoute;
|
||||
use crate::handle::tun_tap::channel_group::{buf_channel_group, BufSenderGroup};
|
||||
use crate::handle::CurrentDeviceInfo;
|
||||
use crate::igmp_server::IgmpServer;
|
||||
#[cfg(feature = "ip_proxy")]
|
||||
use crate::ip_proxy::IpProxyMap;
|
||||
use crate::tun_tap_device::{DeviceReader, DeviceWriter};
|
||||
|
||||
pub fn start(
|
||||
worker: VntWorker,
|
||||
sender: ChannelSender,
|
||||
device_reader: DeviceReader,
|
||||
device_writer: DeviceWriter,
|
||||
igmp_server: Option<IgmpServer>,
|
||||
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
|
||||
ip_route: Option<ExternalRoute>,
|
||||
#[cfg(feature = "ip_proxy")] ip_proxy_map: Option<IpProxyMap>,
|
||||
client_cipher: Cipher,
|
||||
server_cipher: Cipher,
|
||||
parallel: usize,
|
||||
) {
|
||||
if parallel == 1 {
|
||||
thread::Builder::new()
|
||||
.name("tap_handler".into())
|
||||
.spawn(move || {
|
||||
if let Err(e) = start_simple(
|
||||
&sender,
|
||||
device_reader,
|
||||
&device_writer,
|
||||
igmp_server,
|
||||
current_device,
|
||||
ip_route,
|
||||
#[cfg(feature = "ip_proxy")]
|
||||
ip_proxy_map,
|
||||
client_cipher,
|
||||
server_cipher,
|
||||
) {
|
||||
log::warn!("tap:{:?}", e);
|
||||
}
|
||||
let _ = sender.close();
|
||||
let _ = device_writer.close();
|
||||
worker.stop_all();
|
||||
})
|
||||
.unwrap();
|
||||
} else {
|
||||
let (buf_sender, buf_receiver) = buf_channel_group(parallel);
|
||||
for buf_receiver in buf_receiver.0 {
|
||||
let sender = sender.clone();
|
||||
let device_writer = device_writer.clone();
|
||||
let igmp_server = igmp_server.clone();
|
||||
let current_device = current_device.clone();
|
||||
let ip_route = ip_route.clone();
|
||||
#[cfg(feature = "ip_proxy")]
|
||||
let ip_proxy_map = ip_proxy_map.clone();
|
||||
let client_cipher = client_cipher.clone();
|
||||
let server_cipher = server_cipher.clone();
|
||||
thread::spawn(move || {
|
||||
while let Ok((mut buf, _, len)) = buf_receiver.recv() {
|
||||
match handle(
|
||||
&mut buf,
|
||||
len,
|
||||
&igmp_server,
|
||||
¤t_device,
|
||||
&device_writer,
|
||||
&sender,
|
||||
&ip_route,
|
||||
#[cfg(feature = "ip_proxy")]
|
||||
&ip_proxy_map,
|
||||
&client_cipher,
|
||||
&server_cipher,
|
||||
) {
|
||||
Ok(_) => {}
|
||||
Err(e) => {
|
||||
log::warn!("{:?}", e)
|
||||
}
|
||||
}
|
||||
}
|
||||
let _ = sender.close();
|
||||
let _ = device_writer.close();
|
||||
});
|
||||
}
|
||||
thread::Builder::new()
|
||||
.name("tap_handler".into())
|
||||
.spawn(move || {
|
||||
if let Err(e) = start_(&sender, device_reader, buf_sender) {
|
||||
log::warn!("tap:{:?}", e);
|
||||
}
|
||||
let _ = sender.close();
|
||||
let _ = device_writer.close();
|
||||
worker.stop_all();
|
||||
})
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
fn start_(
|
||||
sender: &ChannelSender,
|
||||
device_reader: DeviceReader,
|
||||
mut buf_sender: BufSenderGroup,
|
||||
) -> io::Result<()> {
|
||||
loop {
|
||||
let mut buf = vec![0; 4096];
|
||||
if sender.is_close() {
|
||||
return Ok(());
|
||||
}
|
||||
let start = 0;
|
||||
let len = device_reader.read(&mut buf)?;
|
||||
if !buf_sender.send((buf, start, len)) {
|
||||
return Err(io::Error::new(
|
||||
io::ErrorKind::Other,
|
||||
"tap buf_sender发送失败",
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn start_simple(
|
||||
sender: &ChannelSender,
|
||||
device_reader: DeviceReader,
|
||||
device_writer: &DeviceWriter,
|
||||
igmp_server: Option<IgmpServer>,
|
||||
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
|
||||
ip_route: Option<ExternalRoute>,
|
||||
#[cfg(feature = "ip_proxy")] ip_proxy_map: Option<IpProxyMap>,
|
||||
client_cipher: Cipher,
|
||||
server_cipher: Cipher,
|
||||
) -> io::Result<()> {
|
||||
let mut buf = [0; 4096];
|
||||
loop {
|
||||
if sender.is_close() {
|
||||
return Ok(());
|
||||
}
|
||||
let len = device_reader.read(&mut buf)?;
|
||||
if let Err(e) = handle(
|
||||
&mut buf,
|
||||
len,
|
||||
&igmp_server,
|
||||
¤t_device,
|
||||
device_writer,
|
||||
sender,
|
||||
&ip_route,
|
||||
#[cfg(feature = "ip_proxy")]
|
||||
&ip_proxy_map,
|
||||
&client_cipher,
|
||||
&server_cipher,
|
||||
) {
|
||||
log::warn!("tap handle{:?}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn handle(
|
||||
buf: &mut [u8],
|
||||
len: usize,
|
||||
igmp_server: &Option<IgmpServer>,
|
||||
current_device: &AtomicCell<CurrentDeviceInfo>,
|
||||
device_writer: &DeviceWriter,
|
||||
sender: &ChannelSender,
|
||||
ip_route: &Option<ExternalRoute>,
|
||||
#[cfg(feature = "ip_proxy")] proxy_map: &Option<IpProxyMap>,
|
||||
client_cipher: &Cipher,
|
||||
server_cipher: &Cipher,
|
||||
) -> crate::Result<()> {
|
||||
let mut ethernet_packet = EthernetPacket::new(&mut buf[..len])?;
|
||||
let current_device = current_device.load();
|
||||
match ethernet_packet.protocol() {
|
||||
ethernet::protocol::Protocol::Arp => {
|
||||
let mut out_ethernet_packet =
|
||||
EthernetPacket::unchecked(ethernet_packet.buffer.to_vec());
|
||||
let arp_packet = ArpPacket::unchecked(ethernet_packet.payload());
|
||||
let mut out_arp_packet = ArpPacket::unchecked(out_ethernet_packet.payload_mut());
|
||||
let sender_h = arp_packet.sender_hardware_addr();
|
||||
let sender_p = arp_packet.sender_protocol_addr();
|
||||
let target_p = arp_packet.target_protocol_addr();
|
||||
if target_p == &[0, 0, 0, 0] || sender_p == &[0, 0, 0, 0] || target_p == sender_p {
|
||||
return Ok(());
|
||||
}
|
||||
//回复一个虚假的MAC地址
|
||||
out_arp_packet.set_sender_hardware_addr(&[
|
||||
target_p[0],
|
||||
target_p[1],
|
||||
target_p[2],
|
||||
target_p[3],
|
||||
!sender_h[5],
|
||||
234,
|
||||
]);
|
||||
out_arp_packet.set_sender_protocol_addr(target_p);
|
||||
out_arp_packet.set_target_hardware_addr(sender_h);
|
||||
out_arp_packet.set_target_protocol_addr(sender_p);
|
||||
out_arp_packet.set_op_code(2);
|
||||
out_ethernet_packet.set_source(&[
|
||||
target_p[0],
|
||||
target_p[1],
|
||||
target_p[2],
|
||||
target_p[3],
|
||||
!sender_h[5],
|
||||
234,
|
||||
]);
|
||||
out_ethernet_packet.set_destination(sender_h);
|
||||
device_writer.write_ethernet_tap(&out_ethernet_packet.buffer)?;
|
||||
}
|
||||
ethernet::protocol::Protocol::Ipv4 => {
|
||||
let mut ipv4_packet = IpV4Packet::unchecked(ethernet_packet.payload_mut());
|
||||
let src_ip = ipv4_packet.source_ip();
|
||||
let dest_ip = ipv4_packet.destination_ip();
|
||||
let protocol = ipv4_packet.protocol();
|
||||
if src_ip == dest_ip {
|
||||
if protocol == ipv4::protocol::Protocol::Icmp {
|
||||
let mut icmp = IcmpPacket::new(ipv4_packet.payload_mut())?;
|
||||
if icmp.kind() == Kind::EchoRequest {
|
||||
icmp.set_kind(Kind::EchoReply);
|
||||
icmp.update_checksum();
|
||||
ipv4_packet.set_source_ip(dest_ip);
|
||||
ipv4_packet.set_destination_ip(src_ip);
|
||||
ipv4_packet.update_checksum();
|
||||
let source = ethernet_packet.source().to_vec();
|
||||
let dest = ethernet_packet.destination().to_vec();
|
||||
ethernet_packet.set_source(&dest);
|
||||
ethernet_packet.set_destination(&source);
|
||||
device_writer.write_ethernet_tap(ðernet_packet.buffer)?;
|
||||
}
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
// 以太网帧头部14字节，预留12字节
|
||||
return crate::handle::tun_tap::base_handle(
|
||||
sender,
|
||||
&mut buf[2..],
|
||||
len - 2,
|
||||
igmp_server,
|
||||
current_device,
|
||||
ip_route,
|
||||
#[cfg(feature = "ip_proxy")]
|
||||
proxy_map,
|
||||
client_cipher,
|
||||
server_cipher,
|
||||
);
|
||||
}
|
||||
_ => {
|
||||
// log::warn!("不支持的二层协议：{:?}",p)
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
Reference in New Issue
Block a user