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

NAT1下的tcp打洞

Browse Source
This commit is contained in:
lbl8603
2024-08-05 20:44:54 +08:00
parent 4347ed68c3
commit 0f9ac2ac87
16 changed files with 260 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vnt/proto/message.proto
+12
View File
@@ -64,11 +64,23 @@ message PunchInfo {
uint32 tcp_port = 11;
repeated uint32 udp_ports = 12;
repeated uint32 public_ports = 13;
uint32 public_tcp_port = 14;
PunchNatModel punch_model = 15;
}
enum PunchNatType {
Symmetric = 0;
Cone = 1;
}
enum PunchNatModel {
All = 0;
IPv4 = 1;
IPv6 = 2;
IPv4Tcp = 3;
IPv4Udp = 4;
IPv6Tcp = 5;
IPv6Udp = 6;
}
/// 向服务器上报客户端状态信息
message ClientStatusInfo {
fixed32 source = 1;
vnt/src/channel/context.rs
+44 -56
View File
@@ -60,6 +60,7 @@ impl ChannelContext {
up_traffic_meter,
down_traffic_meter,
default_interface,
default_route_key: AtomicCell::default(),
};
Self {
inner: Arc::new(inner),
@@ -86,7 +87,7 @@ pub struct ContextInner {
// 对称网络增加的udp socket
sub_udp_socket: RwLock<Vec<UdpSocket>>,
// tcp数据发送器
pub(crate) packet_map: RwLock<FnvHashMap<SocketAddr, PacketSender>>,
pub(crate) packet_map: RwLock<FnvHashMap<RouteKey, PacketSender>>,
// 路由信息
pub route_table: RouteTable,
// 使用什么协议连接服务器
@@ -98,6 +99,7 @@ pub struct ContextInner {
pub(crate) up_traffic_meter: Option<TrafficMeterMultiAddress>,
pub(crate) down_traffic_meter: Option<TrafficMeterMultiAddress>,
default_interface: LocalInterface,
default_route_key: AtomicCell<Option<RouteKey>>,
}
impl ContextInner {
@@ -107,6 +109,9 @@ impl ContextInner {
pub fn default_interface(&self) -> &LocalInterface {
&self.default_interface
}
pub fn set_default_route_key(&self, route_key: RouteKey) {
self.default_route_key.store(Some(route_key));
}
/// 通过sub_udp_socket是否为空来判断是否为锥形网络
pub fn is_cone(&self) -> bool {
self.sub_udp_socket.read().is_empty()
@@ -175,11 +180,14 @@ impl ContextInner {
}
Ok(ports)
}
pub fn send_tcp(&self, buf: &[u8], addr: SocketAddr) -> io::Result<()> {
if let Some(tcp) = self.packet_map.read().get(&addr) {
pub fn send_tcp(&self, buf: &[u8], route_key: &RouteKey) -> io::Result<()> {
if let Some(tcp) = self.packet_map.read().get(route_key) {
tcp.try_send(buf)
} else {
Err(io::Error::from(io::ErrorKind::NotFound))
Err(io::Error::new(
io::ErrorKind::NotFound,
format!("dest={:?}", route_key),
))
}
}
pub fn send_main_udp(&self, index: usize, buf: &[u8], addr: SocketAddr) -> io::Result<()> {
@@ -203,7 +211,14 @@ impl ContextInner {
self.send_main_udp(self.v4_len, buf.buffer(), addr)?
}
} else {
self.send_tcp(buf.buffer(), addr)?
if let Some(key) = self.default_route_key.load() {
self.send_tcp(buf.buffer(), &key)?
} else {
return Err(io::Error::new(
io::ErrorKind::NotFound,
format!("dest={:?}", addr),
));
}
}
if let Some(up_traffic_meter) = &self.up_traffic_meter {
up_traffic_meter.add_traffic(buf.destination(), buf.data_len());
@@ -300,7 +315,7 @@ impl ContextInner {
}
}
ConnectProtocol::TCP | ConnectProtocol::WS | ConnectProtocol::WSS => {
self.send_tcp(buf.buffer(), route_key.addr)?
self.send_tcp(buf.buffer(), &route_key)?
}
}
if let Some(up_traffic_meter) = &self.up_traffic_meter {
@@ -376,19 +391,11 @@ impl RouteTable {
let key = route.route_key();
if only_if_absent {
if let Some((_, list)) = self.route_table.read().get(&id) {
let mut p2p_num = 0;
for (x, _) in list {
if x.is_p2p() {
p2p_num += 1;
}
if x.route_key() == key {
return true;
}
}
if !self.first_latency && p2p_num >= self.channel_num {
// 非优先延迟的情况下，通道满了则不用再添加
return false;
}
}
}
let mut route_table = self.route_table.write();
@@ -413,61 +420,42 @@ impl RouteTable {
}
}
if exist {
// 这个排序还有待优化，因为后加入的大概率排最后，被直接淘汰的概率也大，可能导致更好的通道被移除了
list.sort_by_key(|(k, _)| k.rt);
//如果延迟都稳定了，则去除多余通道
for (route, _) in list.iter() {
if route.rt == DEFAULT_RT {
return true;
}
}
//延迟优先模式需要更多的通道探测延迟最低的路线
let limit_len = if self.first_latency {
self.channel_num + 2
} else {
self.channel_num
};
self.truncate_(list, limit_len);
} else {
if !self.first_latency {
if route.is_p2p() {
//非优先延迟的情况下 添加了直连的则排除非直连的
list.retain(|(k, _)| k.is_p2p());
}
if self.channel_num <= list.len() {
return false;
}
};
//增加路由表容量，避免波动
let limit_len = self.channel_num * 2;
list.sort_by_key(|(k, _)| k.rt);
self.truncate_(list, limit_len);
list.push((route, AtomicCell::new(Instant::now())));
}
return true;
}
fn truncate_(&self, list: &mut Vec<(Route, AtomicCell<Instant>)>, len: usize) {
if list.len() <= len {
return;
}
if self.first_latency {
//找到第一个p2p通道
if let Some(index) =
list.iter()
.enumerate()
.find_map(|(index, (route, _))| if route.is_p2p() { Some(index) } else { None })
{
if index >= len {
//保留第一个p2p通道
let route = list.remove(index);
list.truncate(len - 1);
list.push(route);
return;
}
}
}
list.truncate(len);
}
// 直接移除会导致通道不稳定，所以废弃这个方法，后面改用多余通道不发心跳包，从而让通道自动过期
// fn truncate_(&self, list: &mut Vec<(Route, AtomicCell<Instant>)>, len: usize) {
// if list.len() <= len {
// return;
// }
// if self.first_latency {
// //找到第一个p2p通道
// if let Some(index) =
// list.iter()
// .enumerate()
// .find_map(|(index, (route, _))| if route.is_p2p() { Some(index) } else { None })
// {
// if index >= len {
// //保留第一个p2p通道
// let route = list.remove(index);
// list.truncate(len - 1);
// list.push(route);
// return;
// }
// }
// }
// list.truncate(len);
// }
pub fn route(&self, id: &Ipv4Addr) -> Option<Vec<Route>> {
if let Some((_, v)) = self.route_table.read().get(id) {
Some(v.iter().map(|(i, _)| *i).collect())
vnt/src/channel/mod.rs
+1 -1
View File
@@ -179,7 +179,7 @@ pub struct RouteKey {
}
impl RouteKey {
pub(crate) fn new(protocol: ConnectProtocol, index: usize, addr: SocketAddr) -> Self {
pub(crate) const fn new(protocol: ConnectProtocol, index: usize, addr: SocketAddr) -> Self {
Self {
protocol,
index,
vnt/src/channel/punch.rs
+79 -14
View File
@@ -14,16 +14,17 @@ use crate::channel::context::ChannelContext;
use crate::channel::sender::ConnectUtil;
use crate::handle::CurrentDeviceInfo;
use crate::nat::{is_ipv4_global, NatTest};
use crate::proto::message::{PunchNatModel, PunchNatType};
#[derive(Copy, Clone, Eq, PartialEq, Debug)]
pub enum PunchModel {
All,
IPv4,
IPv6,
IPv4Tcp,
IPv4Udp,
IPv6Tcp,
IPv6Udp,
All,
}
impl PunchModel {
@@ -72,6 +73,33 @@ impl Default for PunchModel {
PunchModel::All
}
}
impl From<PunchModel> for PunchNatModel {
fn from(value: PunchModel) -> Self {
match value {
PunchModel::All => PunchNatModel::All,
PunchModel::IPv4 => PunchNatModel::IPv4,
PunchModel::IPv6 => PunchNatModel::IPv6,
PunchModel::IPv4Tcp => PunchNatModel::IPv4Tcp,
PunchModel::IPv4Udp => PunchNatModel::IPv4Udp,
PunchModel::IPv6Tcp => PunchNatModel::IPv6Tcp,
PunchModel::IPv6Udp => PunchNatModel::IPv6Udp,
}
}
}
impl Into<PunchModel> for PunchNatModel {
fn into(self) -> PunchModel {
match self {
PunchNatModel::All => PunchModel::All,
PunchNatModel::IPv4 => PunchModel::IPv4,
PunchNatModel::IPv6 => PunchModel::IPv6,
PunchNatModel::IPv4Tcp => PunchModel::IPv4Tcp,
PunchNatModel::IPv4Udp => PunchModel::IPv4Udp,
PunchNatModel::IPv6Tcp => PunchModel::IPv6Tcp,
PunchNatModel::IPv6Udp => PunchModel::IPv6Udp,
}
}
}
#[derive(Clone, Debug)]
pub struct NatInfo {
@@ -83,6 +111,8 @@ pub struct NatInfo {
pub(crate) ipv6: Option<Ipv6Addr>,
pub udp_ports: Vec<u16>,
pub tcp_port: u16,
pub public_tcp_port: u16,
pub punch_model: PunchModel,
}
#[derive(Clone, Copy, PartialEq, Eq, Debug, Hash)]
@@ -91,6 +121,29 @@ pub enum NatType {
Cone,
}
impl NatType {
pub fn is_cone(&self) -> bool {
self == &NatType::Cone
}
}
impl From<NatType> for PunchNatType {
fn from(value: NatType) -> Self {
match value {
NatType::Symmetric => PunchNatType::Symmetric,
NatType::Cone => PunchNatType::Cone,
}
}
}
impl Into<NatType> for PunchNatType {
fn into(self) -> NatType {
match self {
PunchNatType::Symmetric => NatType::Symmetric,
PunchNatType::Cone => NatType::Cone,
}
}
}
impl NatInfo {
pub fn new(
mut public_ips: Vec<Ipv4Addr>,
@@ -100,7 +153,9 @@ impl NatInfo {
mut ipv6: Option<Ipv6Addr>,
udp_ports: Vec<u16>,
tcp_port: u16,
public_tcp_port: u16,
mut nat_type: NatType,
punch_model: PunchModel,
) -> Self {
public_ips.retain(|ip| {
!ip.is_multicast()
@@ -130,7 +185,9 @@ impl NatInfo {
ipv6,
udp_ports,
tcp_port,
public_tcp_port,
nat_type,
punch_model,
}
}
pub fn update_addr(&mut self, index: usize, ip: Ipv4Addr, port: u16) -> bool {
@@ -144,7 +201,7 @@ impl NatInfo {
*public_port = port;
}
}
if crate::nat::is_ipv4_global(&ip) {
if is_ipv4_global(&ip) {
if !self.public_ips.contains(&ip) {
self.public_ips.push(ip);
updated = true;
@@ -153,6 +210,9 @@ impl NatInfo {
}
updated
}
pub fn update_tcp_port(&mut self, port: u16) {
self.public_tcp_port = port;
}
pub fn local_ipv4(&self) -> Option<Ipv4Addr> {
self.local_ipv4
}
@@ -252,7 +312,10 @@ impl Punch {
if self.nat_test.is_local_address(true, addr) {
return;
}
self.connect_util.try_connect_tcp(buf.to_vec(), addr);
if addr.ip().is_unspecified() || addr.port() == 0 {
return;
}
self.connect_util.try_connect_tcp_punch(buf.to_vec(), addr);
}
pub fn punch(
&mut self,
@@ -277,44 +340,46 @@ impl Punch {
nat_info.local_ipv4 = nat_info
.local_ipv4
.filter(|ip| device_info.not_in_network(*ip));
if punch_tcp && self.punch_model.use_tcp() && nat_info.tcp_port != 0 {
if punch_tcp && self.punch_model.use_tcp() && nat_info.punch_model.use_tcp() {
//向tcp发起连接
if self.punch_model.use_ipv6() {
if self.punch_model.use_ipv6() && nat_info.punch_model.use_ipv6() {
if let Some(ipv6_addr) = nat_info.local_tcp_ipv6addr() {
self.connect_tcp(buf, ipv6_addr)
}
}
if self.punch_model.use_ipv4() {
if self.punch_model.use_ipv4() && nat_info.punch_model.use_ipv4() {
if let Some(ipv4_addr) = nat_info.local_tcp_ipv4addr() {
self.connect_tcp(buf, ipv4_addr)
}
for ip in &nat_info.public_ips {
let addr = SocketAddr::V4(SocketAddrV4::new(*ip, nat_info.tcp_port));
self.connect_tcp(buf, addr)
self.connect_tcp(buf, addr);
}
if nat_info.nat_type.is_cone() && nat_info.public_tcp_port != 0 {
for ip in &nat_info.public_ips {
let addr = SocketAddr::V4(SocketAddrV4::new(*ip, nat_info.public_tcp_port));
self.connect_tcp(buf, addr);
}
}
}
}
if !self.punch_model.use_udp() {
if !self.punch_model.use_udp() || !nat_info.punch_model.use_udp() {
return Ok(());
}
let channel_num = self.context.channel_num();
let main_len = self.context.main_len();
if self.punch_model.use_ipv6() {
if self.punch_model.use_ipv6() && nat_info.punch_model.use_ipv6() {
for index in channel_num..main_len {
if let Some(ipv6_addr) = nat_info.local_udp_ipv6addr(index) {
if !self.nat_test.is_local_address(false, ipv6_addr) {
let rs = self.context.send_main_udp(index, buf, ipv6_addr);
log::info!("发送到ipv6地址:{:?},rs={:?} {}", ipv6_addr, rs, id);
if rs.is_ok() && self.punch_model == PunchModel::IPv6 {
return Ok(());
}
}
}
}
}
if !self.punch_model.use_ipv4() {
if !self.punch_model.use_ipv4() || !nat_info.punch_model.use_ipv4() {
return Ok(());
}
for index in 0..channel_num {
vnt/src/channel/sender.rs
+9 -3
View File
@@ -239,13 +239,13 @@ impl PacketSender {
#[derive(Clone)]
pub struct ConnectUtil {
connect_tcp: Sender<(Vec<u8>, SocketAddr)>,
connect_tcp: Sender<(Vec<u8>, Option<u16>, SocketAddr)>,
connect_ws: Sender<(Vec<u8>, String)>,
}
impl ConnectUtil {
pub fn new(
connect_tcp: Sender<(Vec<u8>, SocketAddr)>,
connect_tcp: Sender<(Vec<u8>, Option<u16>, SocketAddr)>,
connect_ws: Sender<(Vec<u8>, String)>,
) -> Self {
Self {
@@ -254,7 +254,13 @@ impl ConnectUtil {
}
}
pub fn try_connect_tcp(&self, buf: Vec<u8>, addr: SocketAddr) {
if self.connect_tcp.try_send((buf, addr)).is_err() {
if self.connect_tcp.try_send((buf, None, addr)).is_err() {
log::warn!("try_connect_tcp failed {}", addr);
}
}
pub fn try_connect_tcp_punch(&self, buf: Vec<u8>, addr: SocketAddr) {
// 打洞的连接可以绑定随机端口
if self.connect_tcp.try_send((buf, Some(0), addr)).is_err() {
log::warn!("try_connect_tcp failed {}", addr);
}
}
vnt/src/channel/socket/mod.rs
+20 -2
View File
@@ -27,15 +27,22 @@ pub struct LocalInterface {
pub async fn connect_tcp(
addr: SocketAddr,
bind_port: u16,
default_interface: &LocalInterface,
) -> anyhow::Result<tokio::net::TcpStream> {
let socket = create_tcp(addr.is_ipv4(), default_interface)?;
let socket = create_tcp0(addr.is_ipv4(), bind_port, default_interface)?;
Ok(socket.connect(addr).await?)
}
pub fn create_tcp(
v4: bool,
default_interface: &LocalInterface,
) -> anyhow::Result<tokio::net::TcpSocket> {
create_tcp0(v4, 0, default_interface)
}
pub fn create_tcp0(
v4: bool,
bind_port: u16,
default_interface: &LocalInterface,
) -> anyhow::Result<tokio::net::TcpSocket> {
let socket = if v4 {
socket2::Socket::new(
@@ -53,6 +60,17 @@ pub fn create_tcp(
if v4 {
socket.set_ip_unicast_if(default_interface)?;
}
if bind_port != 0 {
socket.set_reuse_address(true)?;
if v4 {
let addr: SocketAddr = format!("0.0.0.0:{}", bind_port).parse().unwrap();
socket.bind(&addr.into())?;
} else {
socket.set_only_v6(true)?;
let addr: SocketAddr = format!("[::]:{}", bind_port).parse().unwrap();
socket.bind(&addr.into())?;
}
}
socket.set_nonblocking(true)?;
socket.set_nodelay(true)?;
Ok(tokio::net::TcpSocket::from_std_stream(socket.into()))
vnt/src/channel/tcp_channel.rs
+33 -18
View File
@@ -1,5 +1,9 @@
use anyhow::{anyhow, Context};
use std::net::SocketAddr;
#[cfg(unix)]
use std::os::unix::io::AsRawFd;
#[cfg(windows)]
use std::os::windows::io::AsRawSocket;
use std::thread;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt};
@@ -16,7 +20,7 @@ use crate::util::StopManager;
/// 监听tcp端口，等待客户端连接
pub fn tcp_listen<H>(
tcp_server: std::net::TcpListener,
receiver: Receiver<(Vec<u8>, SocketAddr)>,
receiver: Receiver<(Vec<u8>, Option<u16>, SocketAddr)>,
recv_handler: H,
context: ChannelContext,
stop_manager: StopManager,
@@ -28,6 +32,7 @@ where
let worker = stop_manager.add_listener("tcpChannel".into(), move || {
let _ = stop_sender.send(());
})?;
let bind_port = tcp_server.local_addr()?.port();
let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
@@ -46,9 +51,9 @@ where
}
});
}
tokio::spawn(
async move { connect_tcp_handle(receiver, recv_handler, context).await },
);
tokio::spawn(async move {
connect_tcp_handle(receiver, recv_handler, context, bind_port).await
});
});
runtime.block_on(async {
let _ = stop_receiver.await;
@@ -61,17 +66,23 @@ where
}
async fn connect_tcp_handle<H>(
mut receiver: Receiver<(Vec<u8>, SocketAddr)>,
mut receiver: Receiver<(Vec<u8>, Option<u16>, SocketAddr)>,
recv_handler: H,
context: ChannelContext,
listener_bind_port: u16,
) where
H: RecvChannelHandler,
{
while let Some((data, addr)) = receiver.recv().await {
while let Some((data, bind_port, addr)) = receiver.recv().await {
let recv_handler = recv_handler.clone();
let context = context.clone();
let bind_port = if let Some(bind_port) = bind_port {
bind_port
} else {
listener_bind_port
};
tokio::spawn(async move {
if let Err(e) = connect_tcp0(data, addr, recv_handler, context).await {
if let Err(e) = connect_tcp0(data, addr, recv_handler, context, bind_port).await {
log::warn!("发送失败,链接终止:{:?},{:?}", addr, e);
}
});
@@ -83,13 +94,14 @@ async fn connect_tcp0<H>(
addr: SocketAddr,
recv_handler: H,
context: ChannelContext,
bind_port: u16,
) -> anyhow::Result<()>
where
H: RecvChannelHandler,
{
let mut stream = tokio::time::timeout(
Duration::from_secs(3),
crate::channel::socket::connect_tcp(addr, context.default_interface()),
crate::channel::socket::connect_tcp(addr, bind_port, context.default_interface()),
)
.await??;
tcp_write(&mut stream, &data).await?;
@@ -110,6 +122,7 @@ where
loop {
let (stream, addr) = tcp_server.accept().await?;
tcp_stream_handle(stream, addr, recv_handler.clone(), context.clone()).await;
}
}
@@ -123,12 +136,18 @@ pub async fn tcp_stream_handle<H>(
H: RecvChannelHandler,
{
let _ = stream.set_nodelay(true);
let local = stream.local_addr();
#[cfg(windows)]
let index = stream.as_raw_socket() as usize;
#[cfg(unix)]
let index = stream.as_raw_fd() as usize;
let route_key = RouteKey::new(ConnectProtocol::TCP, index, addr);
let (r, mut w) = stream.into_split();
let (sender, mut receiver) = channel::<Vec<u8>>(100);
context
.packet_map
.write()
.insert(addr, PacketSender::new(sender));
.insert(route_key, PacketSender::new(sender));
tokio::spawn(async move {
while let Some(data) = receiver.recv().await {
if let Err(e) = tcp_write(&mut w, &data).await {
@@ -139,10 +158,10 @@ pub async fn tcp_stream_handle<H>(
let _ = w.shutdown().await;
});
tokio::spawn(async move {
if let Err(e) = tcp_read(r, addr, &context, recv_handler).await {
log::warn!("tcp_read {:?}", e)
if let Err(e) = tcp_read(r, addr, &context, recv_handler, route_key).await {
log::warn!("tcp_read {:?} {local:?}-{addr}", e)
}
context.packet_map.write().remove(&addr);
context.packet_map.write().remove(&route_key);
});
}
@@ -162,6 +181,7 @@ async fn tcp_read<H>(
addr: SocketAddr,
context: &ChannelContext,
recv_handler: H,
route_key: RouteKey,
) -> anyhow::Result<()>
where
H: RecvChannelHandler,
@@ -179,11 +199,6 @@ where
return Err(anyhow!("tcp数据长度无效 {}", addr));
}
read.read_exact(&mut buf[..len]).await?;
recv_handler.handle(
&mut buf[..len],
&mut extend,
RouteKey::new(ConnectProtocol::TCP, 0, addr),
context,
);
recv_handler.handle(&mut buf[..len], &mut extend, route_key, context);
}
}
vnt/src/channel/ws_channel.rs
+10 -5
View File
@@ -58,14 +58,16 @@ async fn connect_ws_handle<H>(
) where
H: RecvChannelHandler,
{
let mut index = 0;
while let Some((data, url)) = receiver.recv().await {
let recv_handler = recv_handler.clone();
let context = context.clone();
tokio::spawn(async move {
if let Err(e) = connect_ws(data, url, recv_handler, context).await {
if let Err(e) = connect_ws(data, url, recv_handler, context, index).await {
log::warn!("发送失败,ws链接终止:{:?}", e);
}
});
index += 1;
}
}
const WS_ADDR: SocketAddr = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0));
@@ -75,6 +77,7 @@ async fn connect_ws<H>(
mut url: String,
recv_handler: H,
context: ChannelContext,
index: usize,
) -> anyhow::Result<()>
where
H: RecvChannelHandler,
@@ -114,10 +117,12 @@ where
ws.send(Message::Binary(data)).await?;
let (mut ws_write, ws_read) = ws.split();
let (sender, mut receiver) = channel::<Vec<u8>>(100);
let route_key = RouteKey::new(ConnectProtocol::WS, index, WS_ADDR);
context
.packet_map
.write()
.insert(WS_ADDR, PacketSender::new(sender));
.insert(route_key, PacketSender::new(sender));
tokio::spawn(async move {
while let Some(data) = receiver.recv().await {
if let Err(e) = ws_write.send(Message::Binary(data)).await {
@@ -127,22 +132,22 @@ where
}
let _ = ws_write.close().await;
});
if let Err(e) = ws_read_handle(ws_read, recv_handler, &context).await {
if let Err(e) = ws_read_handle(ws_read, recv_handler, &context, route_key).await {
log::warn!("{:?}", e);
}
context.packet_map.write().remove(&WS_ADDR);
context.packet_map.write().remove(&route_key);
Ok(())
}
async fn ws_read_handle<H>(
mut ws_read: SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>,
recv_handler: H,
context: &ChannelContext,
route_key: RouteKey,
) -> anyhow::Result<()>
where
H: RecvChannelHandler,
{
let mut extend = [0; BUFFER_SIZE];
let route_key = RouteKey::new(ConnectProtocol::WS, 0, WS_ADDR);
while let Some(msg) = ws_read.next().await {
let msg = msg.context("Error during WebSocket ")?;
match msg {
vnt/src/core/conn.rs
+1
View File
@@ -215,6 +215,7 @@ impl VntInner {
udp_ports,
tcp_port,
config.local_ipv4.is_none(),
config.punch_model,
);
// 定时器
let scheduler = Scheduler::new(stop_manager.clone())?;
vnt/src/handle/maintain/heartbeat.rs
+11 -1
View File
@@ -56,6 +56,7 @@ fn heartbeat0(
) {
let gateway_ip = current_device.virtual_gateway;
let src_ip = current_device.virtual_ip;
let channel_num = context.channel_num();
// 可能服务器ip发生变化，导致发送失败
let mut is_send_gateway = false;
match heartbeat_packet_server(device_map, server_cipher, src_ip, gateway_ip) {
@@ -78,6 +79,10 @@ fn heartbeat0(
}
heartbeat_packet_server(device_map, server_cipher, src_ip, gateway_ip)
} else {
if dest_ip < src_ip {
// 只向比自己大的发
continue;
}
heartbeat_packet_client(client_cipher, src_ip, dest_ip)
};
let net_packet = match net_packet {
@@ -87,7 +92,12 @@ fn heartbeat0(
continue;
}
};
for route in routes {
for (index, route) in routes.iter().enumerate() {
if index >= channel_num + 1 {
// 多余的通道不再发送心跳包,让它自动过期
// 这里多留一个增加稳定性
break;
}
if let Err(e) = context.send_by_key(&net_packet, route.route_key()) {
log::warn!("heartbeat err={:?}", e)
}
vnt/src/handle/maintain/punch.rs
+3 -1
View File
@@ -149,7 +149,7 @@ fn punch_start(
*v += 1;
*v
} else {
guard.insert(peer_ip, 1);
guard.insert(peer_ip, 0);
0
}
};
@@ -326,6 +326,7 @@ fn punch_packet(
punch_reply.public_port = nat_info.public_ports.get(0).map_or(0, |v| *v as u32);
punch_reply.public_ports = nat_info.public_ports.iter().map(|e| *e as u32).collect();
punch_reply.public_port_range = nat_info.public_port_range as u32;
punch_reply.public_tcp_port = nat_info.public_tcp_port as u32;
punch_reply.local_ip = u32::from(nat_info.local_ipv4().unwrap_or(Ipv4Addr::UNSPECIFIED));
punch_reply.local_port = nat_info.udp_ports[0] as u32;
punch_reply.tcp_port = nat_info.tcp_port as u32;
@@ -335,6 +336,7 @@ fn punch_packet(
punch_reply.ipv6 = ipv6.octets().to_vec();
}
punch_reply.nat_type = protobuf::EnumOrUnknown::new(PunchNatType::from(nat_info.nat_type));
punch_reply.punch_model = protobuf::EnumOrUnknown::new(nat_info.punch_model.into());
log::info!("请求打洞={:?}", punch_reply);
let bytes = punch_reply
.write_to_bytes()
vnt/src/handle/mod.rs
+3 -3
View File
@@ -223,10 +223,10 @@ impl CurrentDeviceInfo {
virtual_gateway: Ipv4Addr,
) {
let broadcast_ip = (!u32::from_be_bytes(virtual_netmask.octets()))
| u32::from_be_bytes(virtual_gateway.octets());
| u32::from_be_bytes(virtual_ip.octets());
let broadcast_ip = Ipv4Addr::from(broadcast_ip);
let virtual_network = u32::from_be_bytes(virtual_netmask.octets())
& u32::from_be_bytes(virtual_gateway.octets());
let virtual_network =
u32::from_be_bytes(virtual_netmask.octets()) & u32::from_be_bytes(virtual_ip.octets());
let virtual_network = Ipv4Addr::from(virtual_network);
self.virtual_ip = virtual_ip;
self.virtual_netmask = virtual_netmask;
vnt/src/handle/recv_data/client.rs
+14 -12
View File
@@ -216,17 +216,13 @@ impl<Device: DeviceWrite> ClientPacketHandler<Device> {
match ControlPacket::new(net_packet.transport_protocol(), net_packet.payload())? {
ControlPacket::PingPacket(_) => {
let route = Route::from_default_rt(route_key, metric);
if context.route_table.add_route_if_absent(source, route)
|| net_packet.source() < current_device.virtual_ip
{
//在路由表中，或者来源比自己小，就需要回复，注意不能调换顺序
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.send_by_key(&net_packet, route_key)?;
}
context.route_table.add_route_if_absent(source, route);
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.send_by_key(&net_packet, route_key)?;
}
ControlPacket::PongPacket(pong_packet) => {
let current_time = crate::handle::now_time() as u16;
@@ -272,7 +268,7 @@ impl<Device: DeviceWrite> ClientPacketHandler<Device> {
{
return Ok(());
}
let route = Route::from_default_rt(route_key, 1);
let route = Route::from_default_rt(route_key, metric);
context.route_table.add_route_if_absent(source, route);
}
ControlPacket::AddrRequest => match route_key.addr.ip() {
@@ -318,6 +314,7 @@ impl<Device: DeviceWrite> ClientPacketHandler<Device> {
.collect();
let local_ipv4 = Some(Ipv4Addr::from(punch_info.local_ip.to_be_bytes()));
let tcp_port = punch_info.tcp_port as u16;
let public_tcp_port = punch_info.public_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))
@@ -340,7 +337,9 @@ impl<Device: DeviceWrite> ClientPacketHandler<Device> {
ipv6,
punch_info.udp_ports.iter().map(|e| *e as u16).collect(),
tcp_port,
public_tcp_port,
punch_info.nat_type.enum_value_or_default().into(),
punch_info.punch_model.enum_value_or_default().into(),
);
{
let peer_nat_info = peer_nat_info.clone();
@@ -360,8 +359,11 @@ impl<Device: DeviceWrite> ClientPacketHandler<Device> {
nat_info.public_ports.iter().map(|e| *e as u32).collect();
punch_reply.public_port_range = nat_info.public_port_range as u32;
punch_reply.tcp_port = nat_info.tcp_port as u32;
punch_reply.public_tcp_port = nat_info.public_tcp_port as u32;
punch_reply.nat_type =
protobuf::EnumOrUnknown::new(PunchNatType::from(nat_info.nat_type));
punch_reply.punch_model =
protobuf::EnumOrUnknown::new(nat_info.punch_model.into());
punch_reply.local_ip =
u32::from(nat_info.local_ipv4().unwrap_or(Ipv4Addr::UNSPECIFIED));
punch_reply.local_port = nat_info.udp_ports[0] as u32;
vnt/src/handle/recv_data/server.rs
+11 -2
View File
@@ -151,6 +151,8 @@ impl<Call: VntCallback, Device: DeviceWrite> PacketHandler for ServerPacketHandl
let response = HandshakeResponse::parse_from_bytes(net_packet.payload())
.map_err(|e| anyhow!("HandshakeResponse {:?}", e))?;
log::info!("握手响应:{:?},{}", route_key, response);
//设置为默认通道
context.set_default_route_key(route_key);
//如果开启了加密，则发送加密握手请求
#[cfg(feature = "server_encrypt")]
if let Some(key) = self.server_cipher.key() {
@@ -211,7 +213,7 @@ impl<Call: VntCallback, Device: DeviceWrite> PacketHandler for ServerPacketHandl
let handshake_info = HandshakeInfo::new_no_secret(response.version);
if self.callback.handshake(handshake_info) {
//没有加密，则发送注册请求
self.register(current_device, context)?;
self.register(current_device, context, route_key)?;
}
return Ok(());
@@ -295,6 +297,10 @@ impl<Call: VntCallback, Device: DeviceWrite> ServerPacketHandler<Call, Device> {
let public_port = response.public_port as u16;
self.nat_test
.update_addr(route_key.index(), public_ip, public_port);
if route_key.protocol().is_tcp() {
log::info!("更新公网tcp端口 {public_port}");
self.nat_test.update_tcp_port(public_port);
}
let old = current_device;
let mut cur = *current_device;
loop {
@@ -421,7 +427,7 @@ impl<Call: VntCallback, Device: DeviceWrite> ServerPacketHandler<Call, Device> {
service_packet::Protocol::SecretHandshakeResponse => {
log::info!("SecretHandshakeResponse");
//加密握手结束，发送注册数据
self.register(current_device, context)?;
self.register(current_device, context, route_key)?;
}
_ => {
log::warn!(
@@ -466,11 +472,14 @@ impl<Call: VntCallback, Device: DeviceWrite> ServerPacketHandler<Call, Device> {
&self,
current_device: &CurrentDeviceInfo,
context: &ChannelContext,
route_key: RouteKey,
) -> anyhow::Result<()> {
if current_device.status.online() {
log::info!("已连接的不需要注册，{:?}", self.config_info);
return Ok(());
}
//设置为默认通道
context.set_default_route_key(route_key);
let token = self.config_info.token.clone();
let device_id = self.config_info.device_id.clone();
let name = self.config_info.name.clone();
vnt/src/nat/mod.rs
+8 -20
View File
@@ -10,9 +10,8 @@ use parking_lot::Mutex;
use rand::prelude::SliceRandom;
use rand::Rng;
use crate::channel::punch::{NatInfo, NatType};
use crate::channel::punch::{NatInfo, NatType, PunchModel};
use crate::channel::socket::LocalInterface;
use crate::proto::message::PunchNatType;
#[cfg(feature = "upnp")]
use crate::util::UPnP;
@@ -120,24 +119,6 @@ pub struct NatTest {
pub(crate) update_local_ipv4: bool,
}
impl From<NatType> for PunchNatType {
fn from(value: NatType) -> Self {
match value {
NatType::Symmetric => PunchNatType::Symmetric,
NatType::Cone => PunchNatType::Cone,
}
}
}
impl Into<NatType> for PunchNatType {
fn into(self) -> NatType {
match self {
PunchNatType::Symmetric => NatType::Symmetric,
PunchNatType::Cone => NatType::Cone,
}
}
}
impl NatTest {
pub fn new(
_channel_num: usize,
@@ -147,6 +128,7 @@ impl NatTest {
udp_ports: Vec<u16>,
tcp_port: u16,
update_local_ipv4: bool,
punch_model: PunchModel,
) -> NatTest {
let ports = vec![0; udp_ports.len()];
let nat_info = NatInfo::new(
@@ -157,7 +139,9 @@ impl NatTest {
ipv6,
udp_ports.clone(),
tcp_port,
0,
NatType::Cone,
punch_model,
);
let info = Arc::new(Mutex::new(nat_info));
#[cfg(feature = "upnp")]
@@ -257,6 +241,10 @@ impl NatTest {
let mut guard = self.info.lock();
guard.update_addr(index, ip, port)
}
pub fn update_tcp_port(&self, port: u16) {
let mut guard = self.info.lock();
guard.update_tcp_port(port)
}
pub fn re_test(
&self,
local_ipv4: Option<Ipv4Addr>,
vnt/tun/src/windows/tun/mod.rs
+1
View File
@@ -154,6 +154,7 @@ impl Device {
}
fn hash_guid(input: &str) -> [u8; 16] {
let mut hasher = sha2::Sha256::new();
hasher.update(input.as_bytes());
hasher.update(b"VNT");
hasher.update(input.as_bytes());
hasher.update(b"2024");