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

支持websocket协议

Browse Source
This commit is contained in:
lbl8603
2024-06-30 22:16:30 +08:00
parent 728598ba8e
commit 17d1ad3716
27 changed files with 720 additions and 657 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vnt/Cargo.toml
+8 -5
View File
@@ -1,6 +1,6 @@
[package]
name = "vnt"
version = "1.2.10"
version = "1.2.11"
edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@@ -37,13 +37,15 @@ crossbeam-queue = "0.3.11"
anyhow = "1.0.82"
dns-parser = "0.8.0"
tokio = { version = "1.37.0", features = ["full"], optional = true }
tokio = { version = "1.37.0", features = ["full"] }
lz4_flex = { version = "0.11", default-features = false, optional = true }
zstd = { version = "0.13.1", optional = true }
fnv = "1.0.7"
igd = { version = "0.12.1", optional = true }
tokio-tungstenite = { version = "0.23.1", optional = true }
futures-util = "0.3.30"
[target.'cfg(target_os = "windows")'.dependencies]
libloading = "0.8.0"
@@ -54,7 +56,7 @@ protoc-bin-vendored = "3.0.0"
cfg_aliases = "0.2.1"
[features]
default = ["server_encrypt", "aes_gcm", "aes_cbc", "aes_ecb", "sm4_cbc", "chacha20_poly1305", "ip_proxy", "port_mapping", "lz4_compress", "zstd_compress", "integrated_tun"]
default = ["websocket", "server_encrypt", "aes_gcm", "aes_cbc", "aes_ecb", "sm4_cbc", "chacha20_poly1305", "ip_proxy", "port_mapping", "lz4_compress", "zstd_compress", "integrated_tun"]
openssl = ["openssl-sys"]
# 从源码编译
openssl-vendored = ["openssl-sys/vendored"]
@@ -65,9 +67,10 @@ sm4_cbc = ["libsm"]
aes_gcm = ["aes-gcm"]
chacha20_poly1305 = ["chacha20poly1305", "chacha20"]
server_encrypt = ["aes-gcm", "rsa", "spki"]
ip_proxy = ["tokio"]
port_mapping = ["tokio"]
ip_proxy = []
port_mapping = []
lz4_compress = ["lz4_flex"]
zstd_compress = ["zstd"]
integrated_tun = ["tun"]
upnp = ["igd"]
websocket = ["tokio-tungstenite"]
vnt/src/channel/context.rs
+31 -29
View File
@@ -12,7 +12,7 @@ use rand::Rng;
use crate::channel::punch::NatType;
use crate::channel::sender::{AcceptSocketSender, PacketSender};
use crate::channel::{Route, RouteKey, UseChannelType, DEFAULT_RT};
use crate::channel::{ConnectProtocol, Route, RouteKey, UseChannelType, DEFAULT_RT};
/// 传输通道上下文，持有udp socket、tcp socket和路由信息
#[derive(Clone)]
@@ -25,7 +25,7 @@ impl ChannelContext {
main_udp_socket: Vec<UdpSocket>,
use_channel_type: UseChannelType,
first_latency: bool,
is_tcp: bool,
protocol: ConnectProtocol,
packet_loss_rate: Option<f64>,
packet_delay: u32,
use_ipv6: bool,
@@ -45,9 +45,9 @@ impl ChannelContext {
let inner = ContextInner {
main_udp_socket,
sub_udp_socket: RwLock::new(Vec::new()),
tcp_map: RwLock::new(FnvHashMap::default()),
packet_map: RwLock::new(FnvHashMap::default()),
route_table: RouteTable::new(use_channel_type, first_latency, channel_num),
is_tcp,
protocol,
packet_loss_rate,
packet_delay,
main_index: AtomicUsize::new(0),
@@ -77,11 +77,11 @@ pub struct ContextInner {
// 对称网络增加的udp socket
sub_udp_socket: RwLock<Vec<UdpSocket>>,
// tcp数据发送器
pub(crate) tcp_map: RwLock<FnvHashMap<SocketAddr, PacketSender>>,
pub(crate) packet_map: RwLock<FnvHashMap<SocketAddr, PacketSender>>,
// 路由信息
pub route_table: RouteTable,
// 是否使用tcp连接服务器
is_tcp: bool,
// 使用什么协议连接服务器
protocol: ConnectProtocol,
//控制丢包率，取值v=[0,100_0000] 丢包率r=v/100_0000
packet_loss_rate: u32,
//控制延迟
@@ -98,11 +98,11 @@ impl ContextInner {
pub fn is_cone(&self) -> bool {
self.sub_udp_socket.read().is_empty()
}
pub fn is_main_tcp(&self) -> bool {
self.is_tcp
pub fn main_protocol(&self) -> ConnectProtocol {
self.protocol
}
pub fn is_udp_main(&self, route_key: &RouteKey) -> bool {
!route_key.is_tcp() && route_key.index < self.main_udp_socket.len()
route_key.protocol().is_udp() && route_key.index < self.main_udp_socket.len()
}
pub fn first_latency(&self) -> bool {
self.route_table.first_latency
@@ -157,7 +157,7 @@ impl ContextInner {
Ok(ports)
}
pub fn send_tcp(&self, buf: &[u8], addr: SocketAddr) -> io::Result<()> {
if let Some(tcp) = self.tcp_map.read().get(&addr) {
if let Some(tcp) = self.packet_map.read().get(&addr) {
tcp.try_send(buf)
} else {
Err(io::Error::from(io::ErrorKind::NotFound))
@@ -180,11 +180,10 @@ impl ContextInner {
}
/// 将数据发送到默认通道，一般发往服务器才用此方法
pub fn send_default(&self, buf: &[u8], addr: SocketAddr) -> io::Result<()> {
if self.is_tcp {
//服务端地址只在重连时检测变化
self.send_tcp(buf, addr)
} else {
if self.protocol.is_udp() {
self.send_main_udp(self.main_index.load(Ordering::Relaxed), buf, addr)
} else {
self.send_tcp(buf, addr)
}
}
@@ -259,23 +258,26 @@ impl ContextInner {
}
/// 将数据发到指定路由
pub fn send_by_key(&self, buf: &[u8], route_key: RouteKey) -> io::Result<()> {
if route_key.is_tcp {
self.send_tcp(buf, route_key.addr)
} else {
if let Some(main_udp) = self.main_udp_socket.get(route_key.index) {
main_udp.send_to(buf, route_key.addr)?;
} else {
if let Some(udp) = self
.sub_udp_socket
.read()
.get(route_key.index - self.main_udp_socket.len())
{
udp.send_to(buf, route_key.addr)?;
match route_key.protocol() {
ConnectProtocol::UDP => {
if let Some(main_udp) = self.main_udp_socket.get(route_key.index) {
main_udp.send_to(buf, route_key.addr)?;
} else {
Err(io::Error::from(io::ErrorKind::NotFound))?
if let Some(udp) = self
.sub_udp_socket
.read()
.get(route_key.index - self.main_udp_socket.len())
{
udp.send_to(buf, route_key.addr)?;
} else {
Err(io::Error::from(io::ErrorKind::NotFound))?
}
}
Ok(())
}
ConnectProtocol::TCP | ConnectProtocol::WS | ConnectProtocol::WSS => {
self.send_tcp(buf, route_key.addr)
}
Ok(())
}
}
pub fn remove_route(&self, ip: &Ipv4Addr, route_key: RouteKey) {
vnt/src/channel/handler.rs
+1 -1
View File
@@ -3,7 +3,7 @@ use crate::channel::RouteKey;
pub trait RecvChannelHandler: Clone + Send + 'static {
fn handle(
&mut self,
&self,
buf: &mut [u8],
extend: &mut [u8],
route_key: RouteKey,
vnt/src/channel/mod.rs
+83 -23
View File
@@ -1,12 +1,15 @@
use anyhow::Context;
use std::net::{SocketAddr, UdpSocket};
use std::str::FromStr;
use tokio::sync::mpsc::channel;
use crate::channel::context::ChannelContext;
use crate::channel::handler::RecvChannelHandler;
use crate::channel::sender::AcceptSocketSender;
use crate::channel::sender::{AcceptSocketSender, ConnectUtil};
use crate::channel::tcp_channel::tcp_listen;
use crate::channel::udp_channel::udp_listen;
#[cfg(feature = "websocket")]
use crate::channel::ws_channel::ws_connect_accept;
use crate::util::StopManager;
pub mod context;
@@ -17,14 +20,21 @@ pub mod punch;
pub mod sender;
pub mod tcp_channel;
pub mod udp_channel;
#[cfg(feature = "websocket")]
pub mod ws_channel;
pub const BUFFER_SIZE: usize = 1024 * 64;
// 这里留个坑，tcp是支持_TCP_MAX_PACKET_SIZE长度的，
// 但是缓存只用BUFFER_SIZE，会导致多余的数据接收不了
const TCP_MAX_PACKET_SIZE: usize = (1 << 24) - 1;
pub const BUFFER_SIZE: usize = 1024 * 16;
#[derive(Debug, Copy, Clone, Eq, PartialEq)]
pub enum UseChannelType {
Relay,
P2p,
All,
}
impl UseChannelType {
pub fn is_only_relay(&self) -> bool {
self == &UseChannelType::Relay
@@ -36,6 +46,7 @@ impl UseChannelType {
self == &UseChannelType::All
}
}
impl FromStr for UseChannelType {
type Err = String;
@@ -48,15 +59,49 @@ impl FromStr for UseChannelType {
}
}
}
impl Default for UseChannelType {
fn default() -> Self {
UseChannelType::All
}
}
#[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
pub enum ConnectProtocol {
UDP,
TCP,
WS,
WSS,
}
impl ConnectProtocol {
#[inline]
pub fn is_tcp(&self) -> bool {
self == &ConnectProtocol::TCP
}
#[inline]
pub fn is_udp(&self) -> bool {
self == &ConnectProtocol::UDP
}
#[inline]
pub fn is_ws(&self) -> bool {
self == &ConnectProtocol::WS
}
#[inline]
pub fn is_wss(&self) -> bool {
self == &ConnectProtocol::WSS
}
pub fn is_transport(&self) -> bool {
self.is_tcp() || self.is_udp()
}
pub fn is_base_tcp(&self) -> bool {
self.is_tcp() || self.is_ws() || self.is_wss()
}
}
#[derive(Copy, Clone, Debug)]
pub struct Route {
pub is_tcp: bool,
pub protocol: ConnectProtocol,
index: usize,
pub addr: SocketAddr,
pub metric: u8,
@@ -68,11 +113,19 @@ pub struct RouteSortKey {
pub metric: u8,
pub rt: i64,
}
const DEFAULT_RT: i64 = 9999;
impl Route {
pub fn new(is_tcp: bool, index: usize, addr: SocketAddr, metric: u8, rt: i64) -> Self {
pub fn new(
protocol: ConnectProtocol,
index: usize,
addr: SocketAddr,
metric: u8,
rt: i64,
) -> Self {
Self {
is_tcp,
protocol,
index,
addr,
metric,
@@ -81,7 +134,7 @@ impl Route {
}
pub fn from(route_key: RouteKey, metric: u8, rt: i64) -> Self {
Self {
is_tcp: route_key.is_tcp,
protocol: route_key.protocol,
index: route_key.index,
addr: route_key.addr,
metric,
@@ -90,7 +143,7 @@ impl Route {
}
pub fn from_default_rt(route_key: RouteKey, metric: u8) -> Self {
Self {
is_tcp: route_key.is_tcp,
protocol: route_key.protocol,
index: route_key.index,
addr: route_key.addr,
metric,
@@ -99,7 +152,7 @@ impl Route {
}
pub fn route_key(&self) -> RouteKey {
RouteKey {
is_tcp: self.is_tcp,
protocol: self.protocol,
index: self.index,
addr: self.addr,
}
@@ -117,22 +170,24 @@ impl Route {
#[derive(Copy, Clone, Ord, PartialOrd, Eq, PartialEq, Hash, Debug)]
pub struct RouteKey {
is_tcp: bool,
protocol: ConnectProtocol,
index: usize,
pub addr: SocketAddr,
}
impl RouteKey {
pub(crate) fn new(is_tcp: bool, index: usize, addr: SocketAddr) -> Self {
pub(crate) fn new(protocol: ConnectProtocol, index: usize, addr: SocketAddr) -> Self {
Self {
is_tcp,
protocol,
index,
addr,
}
}
pub fn is_tcp(&self) -> bool {
self.is_tcp
#[inline]
pub fn protocol(&self) -> ConnectProtocol {
self.protocol
}
#[inline]
pub fn index(&self) -> usize {
self.index
}
@@ -142,10 +197,10 @@ pub(crate) fn init_context(
ports: Vec<u16>,
use_channel_type: UseChannelType,
first_latency: bool,
is_tcp: bool,
protocol: ConnectProtocol,
packet_loss_rate: Option<f64>,
packet_delay: u32,
) -> anyhow::Result<(ChannelContext, mio::net::TcpListener)> {
) -> anyhow::Result<(ChannelContext, std::net::TcpListener)> {
assert!(!ports.is_empty(), "not channel");
let mut udps = Vec::with_capacity(ports.len());
//检查系统是否支持ipv6
@@ -188,7 +243,7 @@ pub(crate) fn init_context(
udps,
use_channel_type,
first_latency,
is_tcp,
protocol,
packet_loss_rate,
packet_delay,
use_ipv6,
@@ -229,32 +284,37 @@ pub(crate) fn init_context(
socket.listen(128)?;
socket.set_nonblocking(true)?;
socket.set_nodelay(false)?;
let tcp_listener = mio::net::TcpListener::from_std(socket.into());
Ok((context, tcp_listener))
Ok((context, socket.into()))
}
pub(crate) fn init_channel<H>(
tcp_listener: mio::net::TcpListener,
tcp_listener: std::net::TcpListener,
context: ChannelContext,
stop_manager: StopManager,
recv_handler: H,
) -> anyhow::Result<(
AcceptSocketSender<Option<Vec<mio::net::UdpSocket>>>,
AcceptSocketSender<(mio::net::TcpStream, SocketAddr, Option<Vec<u8>>)>,
ConnectUtil,
)>
where
H: RecvChannelHandler,
{
let (tcp_connect_s, tcp_connect_r) = channel(16);
let (ws_connect_s, _ws_connect_r) = channel(16);
let connect_util = ConnectUtil::new(tcp_connect_s, ws_connect_s);
// udp监听，udp_socket_sender 用于NAT类型切换
let udp_socket_sender =
udp_listen(stop_manager.clone(), recv_handler.clone(), context.clone())?;
// 建立tcp监听，tcp_socket_sender 用于tcp 直连
let tcp_socket_sender = tcp_listen(
tcp_listen(
tcp_listener,
stop_manager.clone(),
tcp_connect_r,
recv_handler.clone(),
context.clone(),
stop_manager.clone(),
)?;
#[cfg(feature = "websocket")]
ws_connect_accept(_ws_connect_r, recv_handler, context.clone(), stop_manager)?;
Ok((udp_socket_sender, tcp_socket_sender))
Ok((udp_socket_sender, connect_util))
}
vnt/src/channel/punch.rs
+11 -33
View File
@@ -7,12 +7,11 @@ use std::sync::Arc;
use std::time::Duration;
use std::{io, thread};
use mio::net::TcpStream;
use rand::prelude::SliceRandom;
use rand::Rng;
use crate::channel::context::ChannelContext;
use crate::channel::sender::AcceptSocketSender;
use crate::channel::sender::ConnectUtil;
use crate::external_route::ExternalRoute;
use crate::handle::CurrentDeviceInfo;
use crate::nat::NatTest;
@@ -51,7 +50,7 @@ pub struct NatInfo {
pub nat_type: NatType,
pub(crate) local_ipv4: Option<Ipv4Addr>,
pub(crate) ipv6: Option<Ipv6Addr>,
pub(crate) udp_ports: Vec<u16>,
pub udp_ports: Vec<u16>,
pub tcp_port: u16,
}
@@ -189,7 +188,7 @@ pub struct Punch {
port_index: HashMap<Ipv4Addr, usize>,
punch_model: PunchModel,
is_tcp: bool,
tcp_socket_sender: AcceptSocketSender<(TcpStream, SocketAddr, Option<Vec<u8>>)>,
connect_util: ConnectUtil,
external_route: ExternalRoute,
nat_test: NatTest,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
@@ -200,7 +199,7 @@ impl Punch {
context: ChannelContext,
punch_model: PunchModel,
is_tcp: bool,
tcp_socket_sender: AcceptSocketSender<(TcpStream, SocketAddr, Option<Vec<u8>>)>,
connect_util: ConnectUtil,
external_route: ExternalRoute,
nat_test: NatTest,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
@@ -215,7 +214,7 @@ impl Punch {
port_index: HashMap::new(),
punch_model,
is_tcp,
tcp_socket_sender,
connect_util,
external_route,
nat_test,
current_device,
@@ -224,26 +223,11 @@ impl Punch {
}
impl Punch {
fn connect_tcp(&self, buf: &[u8], addr: SocketAddr) -> bool {
fn connect_tcp(&self, buf: &[u8], addr: SocketAddr) {
if self.nat_test.is_local_address(true, addr) {
return false;
return;
}
// mio是非阻塞的，不能立马判断是否能连接成功，所以用标准库的tcp
match std::net::TcpStream::connect_timeout(&addr, Duration::from_millis(100)) {
Ok(tcp_stream) => {
if tcp_stream.set_nonblocking(true).is_err() {
return false;
}
return self
.tcp_socket_sender
.try_add_socket((TcpStream::from_std(tcp_stream), addr, Some(buf.to_vec())))
.is_ok();
}
Err(e) => {
log::warn!("连接到tcp失败,addr={},err={}", addr, e);
}
}
false
self.connect_util.try_connect_tcp(buf.to_vec(), addr);
}
pub fn punch(
&mut self,
@@ -274,22 +258,16 @@ impl Punch {
if punch_tcp && self.is_tcp && nat_info.tcp_port != 0 {
//向tcp发起连接
if let Some(ipv6_addr) = nat_info.local_tcp_ipv6addr() {
if self.connect_tcp(buf, ipv6_addr) {
// return Ok(());
}
self.connect_tcp(buf, ipv6_addr)
}
//向tcp发起连接
if let Some(ipv4_addr) = nat_info.local_tcp_ipv4addr() {
if self.connect_tcp(buf, ipv4_addr) {
// return Ok(());
}
self.connect_tcp(buf, ipv4_addr)
}
if nat_info.nat_type == NatType::Cone && nat_info.public_ips.len() == 1 {
let addr =
SocketAddr::V4(SocketAddrV4::new(nat_info.public_ips[0], nat_info.tcp_port));
if self.connect_tcp(buf, addr) {
// return Ok(());
}
self.connect_tcp(buf, addr)
}
}
let channel_num = self.context.channel_num();
vnt/src/channel/sender.rs
+43 -47
View File
@@ -1,13 +1,13 @@
use std::io;
use std::net::Ipv4Addr;
use std::net::{Ipv4Addr, SocketAddr};
use std::sync::mpsc::{SyncSender, TrySendError};
use std::sync::Arc;
use crossbeam_utils::atomic::AtomicCell;
use mio::Token;
use tokio::sync::mpsc::Sender;
use crate::channel::context::ChannelContext;
use crate::channel::notify::{AcceptNotify, WritableNotify};
use crate::channel::notify::AcceptNotify;
use crate::cipher::Cipher;
use crate::compression::Compressor;
use crate::external_route::ExternalRoute;
@@ -133,58 +133,54 @@ impl<T> AcceptSocketSender<T> {
}
}
}
#[derive(Clone)]
pub struct PacketSender {
inner: Arc<PacketSenderInner>,
sender: Sender<Vec<u8>>,
}
impl PacketSender {
pub fn new(notify: WritableNotify, buffer: SyncSender<Vec<u8>>, token: Token) -> Self {
Self {
inner: Arc::new(PacketSenderInner {
token,
notify,
buffer,
}),
}
pub fn new(sender: Sender<Vec<u8>>) -> Self {
Self { sender }
}
#[inline]
pub fn try_send(&self, buf: &[u8]) -> io::Result<()> {
self.inner.try_send(buf)
}
pub fn shutdown(&self) -> io::Result<()> {
self.inner.shutdown()
}
}
pub struct PacketSenderInner {
token: Token,
notify: WritableNotify,
buffer: SyncSender<Vec<u8>>,
}
impl PacketSenderInner {
#[inline]
fn try_send(&self, buf: &[u8]) -> io::Result<()> {
let len = buf.len();
let mut buf_vec = Vec::with_capacity(buf.len() + 4);
buf_vec.extend_from_slice(&[
(len >> 24) as u8,
(len >> 16) as u8,
(len >> 8) as u8,
len as u8,
]);
buf_vec.extend_from_slice(buf);
match self.buffer.try_send(buf_vec) {
Ok(_) => self.notify.notify(self.token, true),
Err(e) => match e {
TrySendError::Disconnected(_) => Err(io::Error::from(io::ErrorKind::WriteZero)),
TrySendError::Full(_) => Err(io::Error::from(io::ErrorKind::WouldBlock)),
},
match self.sender.try_send(buf.to_vec()) {
Ok(_) => Ok(()),
Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => Err(io::Error::new(
io::ErrorKind::WouldBlock,
"通道已满，发生丢包",
)),
Err(_) => Err(io::Error::new(
io::ErrorKind::ConnectionRefused,
"通道关闭，发生丢包",
)),
}
}
fn shutdown(&self) -> io::Result<()> {
self.notify.notify(self.token, false)
}
#[derive(Clone)]
pub struct ConnectUtil {
connect_tcp: Sender<(Vec<u8>, SocketAddr)>,
connect_ws: Sender<(Vec<u8>, String)>,
}
impl ConnectUtil {
pub fn new(
connect_tcp: Sender<(Vec<u8>, SocketAddr)>,
connect_ws: Sender<(Vec<u8>, String)>,
) -> Self {
Self {
connect_tcp,
connect_ws,
}
}
pub fn try_connect_tcp(&self, buf: Vec<u8>, addr: SocketAddr) {
if self.connect_tcp.try_send((buf, addr)).is_err() {
log::warn!("try_connect_tcp failed {}", addr);
}
}
pub fn try_connect_ws(&self, buf: Vec<u8>, addr: String) {
if self.connect_ws.try_send((buf, addr)).is_err() {
log::warn!("try_connect_ws failed");
}
}
}
vnt/src/channel/tcp_channel.rs
+148 -434
View File
@@ -1,472 +1,186 @@
use std::collections::HashMap;
use std::io::{Read, Write};
use std::net::{Shutdown, SocketAddr};
#[cfg(any(unix))]
use std::os::fd::FromRawFd;
#[cfg(any(unix))]
use std::os::fd::IntoRawFd;
#[cfg(windows)]
use std::os::windows::io::FromRawSocket;
#[cfg(windows)]
use std::os::windows::io::IntoRawSocket;
use std::sync::mpsc::{sync_channel, Receiver, SyncSender, TryRecvError, TrySendError};
use std::{io, thread};
use mio::net::{TcpListener, TcpStream};
use mio::{Events, Interest, Poll, Registry, Token, Waker};
use anyhow::{anyhow, Context};
use std::net::SocketAddr;
use std::thread;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::net::tcp::OwnedReadHalf;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc::{channel, Receiver};
use crate::channel::context::ChannelContext;
use crate::channel::handler::RecvChannelHandler;
use crate::channel::notify::{AcceptNotify, WritableNotify};
use crate::channel::sender::{AcceptSocketSender, PacketSender};
use crate::channel::{RouteKey, BUFFER_SIZE};
use crate::channel::sender::PacketSender;
use crate::channel::{ConnectProtocol, RouteKey, BUFFER_SIZE, TCP_MAX_PACKET_SIZE};
use crate::util::StopManager;
const SERVER: Token = Token(0);
const NOTIFY: Token = Token(1);
/// 监听tcp端口，等待客户端连接
pub fn tcp_listen<H>(
tcp_server: TcpListener,
stop_manager: StopManager,
tcp_server: std::net::TcpListener,
receiver: Receiver<(Vec<u8>, SocketAddr)>,
recv_handler: H,
context: ChannelContext,
) -> anyhow::Result<AcceptSocketSender<(TcpStream, SocketAddr, Option<Vec<u8>>)>>
stop_manager: StopManager,
) -> anyhow::Result<()>
where
H: RecvChannelHandler,
{
let (tcp_sender, tcp_receiver) = sync_channel(64);
let poll = Poll::new()?;
let waker = AcceptNotify::new(Waker::new(poll.registry(), NOTIFY)?);
let accept = AcceptSocketSender::new(waker.clone(), tcp_sender);
let worker = {
let waker = waker.clone();
stop_manager.add_listener("tcp_listen".into(), move || {
if let Err(e) = waker.stop() {
log::error!("{:?}", e);
}
})?
};
let (stop_sender, stop_receiver) = tokio::sync::oneshot::channel::<()>();
let worker = stop_manager.add_listener("tcpChannel".into(), move || {
let _ = stop_sender.send(());
})?;
let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.context("tcp tokio runtime build failed")?;
thread::Builder::new()
.name("tcpRead".into())
.name("tcpChannel".into())
.spawn(move || {
if let Err(e) = tcp_listen0(
poll,
tcp_server,
&stop_manager,
waker,
tcp_receiver,
recv_handler,
context,
) {
log::error!("{:?}", e);
}
runtime.spawn(async move {
{
let recv_handler = recv_handler.clone();
let context = context.clone();
tokio::spawn(async move {
if let Err(e) = tcp_accept(tcp_server, recv_handler, context).await {
log::warn!("tcp_listen {:?}", e);
}
});
}
tokio::spawn(
async move { connect_tcp_handle(receiver, recv_handler, context).await },
);
});
runtime.block_on(async {
let _ = stop_receiver.await;
});
runtime.shutdown_background();
worker.stop_all();
})?;
Ok(accept)
})
.context("tcp thread build failed")?;
Ok(())
}
fn tcp_listen0<H>(
mut poll: Poll,
mut tcp_server: TcpListener,
stop_manager: &StopManager,
accept_notify: AcceptNotify,
accept_tcp_receiver: Receiver<(TcpStream, SocketAddr, Option<Vec<u8>>)>,
mut recv_handler: H,
async fn connect_tcp_handle<H>(
mut receiver: Receiver<(Vec<u8>, SocketAddr)>,
recv_handler: H,
context: ChannelContext,
) where
H: RecvChannelHandler,
{
while let Some((data, addr)) = receiver.recv().await {
let recv_handler = recv_handler.clone();
let context = context.clone();
tokio::spawn(async move {
if let Err(e) = connect_tcp0(data, addr, recv_handler, context).await {
log::warn!("发送失败,链接终止:{:?},{:?}", addr, e);
}
});
}
}
async fn connect_tcp0<H>(
data: Vec<u8>,
addr: SocketAddr,
recv_handler: H,
context: ChannelContext,
) -> anyhow::Result<()>
where
H: RecvChannelHandler,
{
let (tcp_sender, tcp_receiver) = sync_channel(64);
let write_waker = init_writable_handler(tcp_receiver, stop_manager.clone(), context.clone())?;
poll.registry()
.register(&mut tcp_server, SERVER, Interest::READABLE)?;
let mut events = Events::with_capacity(1024);
let mut stream =
tokio::time::timeout(Duration::from_secs(3), TcpStream::connect(addr)).await??;
tcp_write(&mut stream, &data).await?;
let mut read_map: HashMap<Token, (RouteKey, TcpStream, Box<[u8; BUFFER_SIZE]>, usize)> =
HashMap::with_capacity(32);
let mut extend = [0; BUFFER_SIZE];
loop {
if let Err(e) = poll.poll(&mut events, None) {
crate::ignore_io_interrupted(e)?;
continue;
}
for event in events.iter() {
match event.token() {
SERVER => loop {
match tcp_server.accept() {
Ok((stream, addr)) => {
accept_handle(
stream,
addr,
None,
&write_waker,
&mut read_map,
&tcp_sender,
poll.registry(),
)?;
}
Err(e) => {
if e.kind() == io::ErrorKind::WouldBlock {
break;
}
return Err(e)?;
}
}
},
NOTIFY => {
if accept_notify.is_stop() {
return Ok(());
}
if accept_notify.is_add_socket() {
while let Ok((stream, addr, init_buf)) = accept_tcp_receiver.try_recv() {
accept_handle(
stream,
addr,
init_buf,
&write_waker,
&mut read_map,
&tcp_sender,
poll.registry(),
)?;
}
}
}
token => {
if event.is_readable() {
if let Err(e) = readable_handle(
&token,
&mut read_map,
&mut recv_handler,
&context,
&mut extend,
) {
closed_handle_r(&token, &mut read_map);
log::warn!("{:?}", e);
if let Err(e) = write_waker.notify(token, false) {
log::warn!("{:?}", e);
}
}
} else {
closed_handle_r(&token, &mut read_map);
if let Err(e) = write_waker.notify(token, false) {
log::warn!("{:?}", e);
}
}
}
}
}
}
}
/// 处理写事件
fn init_writable_handler(
receiver: Receiver<(TcpStream, Token, SocketAddr, Option<Vec<u8>>)>,
stop_manager: StopManager,
context: ChannelContext,
) -> anyhow::Result<WritableNotify> {
let poll = Poll::new()?;
let writable_notify = WritableNotify::new(Waker::new(poll.registry(), NOTIFY)?);
let worker = {
let writable_notify = writable_notify.clone();
stop_manager.add_listener("tcp_writable_handler".into(), move || {
if let Err(e) = writable_notify.stop() {
log::error!("{:?}", e);
}
})?
};
{
let writable_notify = writable_notify.clone();
thread::Builder::new()
.name("tcpWriteableListen".into())
.spawn(move || {
if let Err(e) = tcp_writable_listen(receiver, poll, writable_notify, &context) {
log::error!("{:?}", e);
}
worker.stop_all();
})?;
}
Ok(writable_notify)
}
/// 处理写事件
fn tcp_writable_listen(
receiver: Receiver<(TcpStream, Token, SocketAddr, Option<Vec<u8>>)>,
mut poll: Poll,
writable_notify: WritableNotify,
context: &ChannelContext,
) -> io::Result<()> {
let mut events = Events::with_capacity(1024);
let mut write_map: HashMap<
Token,
(
TcpStream,
SocketAddr,
Receiver<Vec<u8>>,
Option<(Vec<u8>, usize)>,
),
> = HashMap::with_capacity(32);
loop {
if let Err(e) = poll.poll(&mut events, None) {
crate::ignore_io_interrupted(e)?;
continue;
}
for event in events.iter() {
match event.token() {
NOTIFY => {
if writable_notify.is_stop() {
//服务停止
return Ok(());
}
if writable_notify.is_need_write() {
// 需要写入数据
if let Some(tokens) = writable_notify.take_all() {
for (token, state) in tokens {
if !state {
closed_handle_w(&token, &mut write_map, &context);
continue;
}
if let Err(e) = writable_handle(&token, &mut write_map) {
closed_handle_w(&token, &mut write_map, &context);
log::warn!("{:?}", e);
}
}
}
}
if writable_notify.is_add_socket() {
//添加tcp连接，并监听写事件
while let Ok((mut stream, token, addr, init_buf)) = receiver.try_recv() {
if let Err(e) = stream.set_nodelay(true) {
log::warn!("set_nodelay err={:?}", e);
}
if let Err(e) =
poll.registry()
.register(&mut stream, token, Interest::WRITABLE)
{
log::warn!("registry err={:?}", e);
continue;
}
let (sender, receiver) = sync_channel(128);
let packet_sender =
PacketSender::new(writable_notify.clone(), sender, token);
if let Some(init_buf) = init_buf {
packet_sender.try_send(&init_buf)?;
}
context.tcp_map.write().insert(addr, packet_sender);
write_map.insert(token, (stream, addr, receiver, None));
}
}
}
token => {
if event.is_writable() {
if let Err(e) = writable_handle(&token, &mut write_map) {
closed_handle_w(&token, &mut write_map, &context);
log::warn!("{:?}", e);
}
} else {
closed_handle_w(&token, &mut write_map, &context);
}
}
}
}
}
}
fn accept_handle(
stream: TcpStream,
addr: SocketAddr,
init_buf: Option<Vec<u8>>,
write_waker: &WritableNotify,
read_map: &mut HashMap<Token, (RouteKey, TcpStream, Box<[u8; BUFFER_SIZE]>, usize)>,
tcp_sender: &SyncSender<(TcpStream, Token, SocketAddr, Option<Vec<u8>>)>,
registry: &Registry,
) -> io::Result<()> {
#[cfg(windows)]
let (tcp_stream, index) = unsafe {
let fd = stream.into_raw_socket();
(std::net::TcpStream::from_raw_socket(fd), fd as usize)
};
#[cfg(any(unix))]
let (tcp_stream, index) = unsafe {
let fd = stream.into_raw_fd();
(std::net::TcpStream::from_raw_fd(fd), fd as usize)
};
if index == 0 || index == 1 {
log::error!("index err={:?}", addr);
return Ok(());
}
let token = Token(index);
match tcp_stream.try_clone() {
Ok(tcp_writer) => {
match tcp_sender.try_send((TcpStream::from_std(tcp_writer), token, addr, init_buf)) {
Ok(_) => {
if let Err(e) = write_waker.add_socket() {
log::error!("write_waker,err={:?},addr={:?}", e, addr);
return Ok(());
}
}
Err(e) => {
return match e {
TrySendError::Full(_) => {
log::error!("Full,addr={:?}", addr);
Ok(())
}
TrySendError::Disconnected(_) => {
Err(io::Error::new(io::ErrorKind::Other, "write thread exit"))
}
};
}
}
}
Err(e) => {
log::error!("try_clone err={:?},addr={:?}", e, addr);
return Ok(());
}
}
let mut stream = TcpStream::from_std(tcp_stream);
if let Err(e) = registry.register(&mut stream, token, Interest::READABLE) {
log::error!("registry err={:?},addr={:?}", e, addr);
return Ok(());
}
read_map.insert(
token,
(
RouteKey::new(true, index, addr),
stream,
Box::new([0; BUFFER_SIZE]),
0,
),
);
tcp_stream_handle(stream, addr, recv_handler, context).await;
Ok(())
}
fn readable_handle<H>(
token: &Token,
map: &mut HashMap<Token, (RouteKey, TcpStream, Box<[u8; BUFFER_SIZE]>, usize)>,
recv_handler: &mut H,
context: &ChannelContext,
extend: &mut [u8],
) -> io::Result<()>
async fn tcp_accept<H>(
tcp_server: std::net::TcpListener,
recv_handler: H,
context: ChannelContext,
) -> anyhow::Result<()>
where
H: RecvChannelHandler,
{
if let Some((route_key, stream, buf, begin)) = map.get_mut(token) {
loop {
let end = if *begin >= 4 {
let len = ((buf[0] as usize) << 24)
| ((buf[1] as usize) << 16)
| ((buf[2] as usize) << 8)
| buf[3] as usize;
4 + len
} else {
4
};
if end > BUFFER_SIZE {
return Err(io::Error::from(io::ErrorKind::InvalidData));
}
match stream.read(&mut buf[*begin..end]) {
Ok(len) => {
if len == 0 {
return Err(io::Error::from(io::ErrorKind::UnexpectedEof));
}
*begin += len;
if end > 4 && *begin == end {
recv_handler.handle(&mut buf[4..end], extend, *route_key, context);
*begin = 0;
}
}
Err(e) => {
if e.kind() == io::ErrorKind::WouldBlock {
break;
}
return Err(e);
}
let tcp_server = TcpListener::from_std(tcp_server)?;
loop {
let (stream, addr) = tcp_server.accept().await?;
tcp_stream_handle(stream, addr, recv_handler.clone(), context.clone()).await;
}
}
pub async fn tcp_stream_handle<H>(
stream: TcpStream,
addr: SocketAddr,
recv_handler: H,
context: ChannelContext,
) where
H: RecvChannelHandler,
{
let _ = stream.set_nodelay(true);
let (r, mut w) = stream.into_split();
let (sender, mut receiver) = channel::<Vec<u8>>(100);
context
.packet_map
.write()
.insert(addr, PacketSender::new(sender));
tokio::spawn(async move {
while let Some(data) = receiver.recv().await {
if let Err(e) = tcp_write(&mut w, &data).await {
log::info!("发送失败,tcp链接终止:{:?},{:?}", addr, e);
break;
}
}
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)
}
context.packet_map.write().remove(&addr);
});
}
async fn tcp_write<W: AsyncWrite + Unpin>(w: &mut W, buf: &[u8]) -> anyhow::Result<()> {
let len = buf.len();
if len > TCP_MAX_PACKET_SIZE {
return Err(anyhow!("超过了tcp的最大长度传输"));
}
w.write_all(&[0, (len >> 16) as u8, (len >> 8) as u8, len as u8])
.await?;
w.write_all(&buf).await?;
Ok(())
}
fn writable_handle(
token: &Token,
map: &mut HashMap<
Token,
(
TcpStream,
SocketAddr,
Receiver<Vec<u8>>,
Option<(Vec<u8>, usize)>,
),
>,
) -> io::Result<()> {
if let Some((stream, _, receiver, last)) = map.get_mut(token) {
loop {
if let Some((buf, begin)) = last {
match stream.write(&buf[*begin..]) {
Ok(len) => {
if len == 0 {
return Err(io::Error::from(io::ErrorKind::WriteZero));
}
if len + *begin == buf.len() {
*last = None;
} else {
*begin += len;
continue;
}
}
Err(e) => {
if e.kind() == io::ErrorKind::WouldBlock {
break;
}
return Err(e);
}
}
}
match receiver.try_recv() {
Ok(buf) => *last = Some((buf, 0)),
Err(e) => match e {
TryRecvError::Empty => {
break;
}
TryRecvError::Disconnected => {
return Err(io::Error::from(io::ErrorKind::Other));
}
},
}
}
}
Ok(())
}
fn closed_handle_r(
token: &Token,
map: &mut HashMap<Token, (RouteKey, TcpStream, Box<[u8; BUFFER_SIZE]>, usize)>,
) {
if let Some((_, tcp, _, _)) = map.remove(token) {
let _ = tcp.shutdown(Shutdown::Both);
}
}
fn closed_handle_w(
token: &Token,
map: &mut HashMap<
Token,
(
TcpStream,
SocketAddr,
Receiver<Vec<u8>>,
Option<(Vec<u8>, usize)>,
),
>,
async fn tcp_read<H>(
mut read: OwnedReadHalf,
addr: SocketAddr,
context: &ChannelContext,
) {
if let Some((tcp, addr, _, _)) = map.remove(token) {
context.tcp_map.write().remove(&addr);
let _ = tcp.shutdown(Shutdown::Both);
recv_handler: H,
) -> anyhow::Result<()>
where
H: RecvChannelHandler,
{
let mut head = [0; 4];
let mut buf = [0; BUFFER_SIZE];
let mut extend = [0; BUFFER_SIZE];
loop {
read.read_exact(&mut head).await?;
if head[0] != 0 {
return Err(anyhow!("tcp数据流错误 {}", addr));
}
let len = ((head[1] as usize) << 16) | ((head[2] as usize) << 8) | head[3] as usize;
if len < 12 || len > buf.len() {
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,
);
}
}
vnt/src/channel/udp_channel.rs
+5 -5
View File
@@ -10,7 +10,7 @@ use crate::channel::context::ChannelContext;
use crate::channel::handler::RecvChannelHandler;
use crate::channel::notify::AcceptNotify;
use crate::channel::sender::AcceptSocketSender;
use crate::channel::{RouteKey, BUFFER_SIZE};
use crate::channel::{ConnectProtocol, RouteKey, BUFFER_SIZE};
use crate::util::StopManager;
pub fn udp_listen<H>(
@@ -60,7 +60,7 @@ where
fn sub_udp_listen0<H>(
mut poll: Poll,
mut recv_handler: H,
recv_handler: H,
context: ChannelContext,
accept_notify: AcceptNotify,
accept_receiver: Receiver<Option<Vec<UdpSocket>>>,
@@ -120,7 +120,7 @@ where
recv_handler.handle(
&mut buf[..len],
&mut extend,
RouteKey::new(false, token.0, addr),
RouteKey::new(ConnectProtocol::UDP, token.0, addr),
&context,
);
}
@@ -238,7 +238,7 @@ where
pub fn main_udp_listen0<H>(
mut poll: Poll,
mut recv_handler: H,
recv_handler: H,
context: ChannelContext,
) -> io::Result<()>
where
@@ -280,7 +280,7 @@ where
recv_handler.handle(
&mut buf[..len],
&mut extend,
RouteKey::new(false, index, addr),
RouteKey::new(ConnectProtocol::UDP, index, addr),
&context,
);
}
vnt/src/channel/ws_channel.rs
+130
View File
@@ -0,0 +1,130 @@
use crate::channel::{ConnectProtocol, RouteKey, BUFFER_SIZE};
use anyhow::Context;
use futures_util::stream::SplitStream;
use futures_util::{SinkExt, StreamExt};
use std::convert::Into;
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::thread;
use std::time::Duration;
use tokio::net::TcpStream;
use tokio::sync::mpsc::{channel, Receiver};
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
use crate::channel::context::ChannelContext;
use crate::channel::handler::RecvChannelHandler;
use crate::channel::sender::PacketSender;
use crate::util::StopManager;
/// ws协议，
/// 暂时只允许用ws连服务端，不能用ws打洞/连客户端
pub fn ws_connect_accept<H>(
receiver: Receiver<(Vec<u8>, String)>,
recv_handler: H,
context: ChannelContext,
stop_manager: StopManager,
) -> anyhow::Result<()>
where
H: RecvChannelHandler,
{
let (stop_sender, stop_receiver) = tokio::sync::oneshot::channel::<()>();
let worker = stop_manager.add_listener("wsChannel".into(), move || {
let _ = stop_sender.send(());
})?;
let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.context("ws tokio runtime build failed")?;
thread::Builder::new()
.name("wsChannel".into())
.spawn(move || {
runtime.spawn(async move { connect_ws_handle(receiver, recv_handler, context).await });
runtime.block_on(async {
let _ = stop_receiver.await;
});
runtime.shutdown_background();
worker.stop_all();
})
.context("ws thread build failed")?;
Ok(())
}
async fn connect_ws_handle<H>(
mut receiver: Receiver<(Vec<u8>, String)>,
recv_handler: H,
context: ChannelContext,
) where
H: RecvChannelHandler,
{
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 {
log::warn!("发送失败,ws链接终止:{:?}", e);
}
});
}
}
const WS_ADDR: SocketAddr = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0));
async fn connect_ws<H>(
data: Vec<u8>,
url: String,
recv_handler: H,
context: ChannelContext,
) -> anyhow::Result<()>
where
H: RecvChannelHandler,
{
println!("ws协议 {}", url);
let (mut ws, response) =
tokio::time::timeout(Duration::from_secs(3), connect_async(url)).await??;
println!("ws协议 {:?}", response);
ws.send(Message::Binary(data)).await?;
let (mut ws_write, ws_read) = ws.split();
let (sender, mut receiver) = channel::<Vec<u8>>(100);
context
.packet_map
.write()
.insert(WS_ADDR, 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 {
log::warn!("websocket err {:?}", e);
break;
}
}
let _ = ws_write.close().await;
});
if let Err(e) = ws_read_handle(ws_read, recv_handler, &context).await {
log::warn!("{:?}", e);
}
context.packet_map.write().remove(&WS_ADDR);
Ok(())
}
async fn ws_read_handle<H>(
mut ws_read: SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>,
recv_handler: H,
context: &ChannelContext,
) -> 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 {
Message::Text(txt) => log::info!("Received text message: {}", txt),
Message::Binary(mut data) => {
recv_handler.handle(&mut data, &mut extend, route_key, context);
}
Message::Ping(_) | Message::Pong(_) => (),
Message::Close(_) => break,
_ => {}
}
}
Ok(())
}
vnt/src/core/conn.rs
+5 -5
View File
@@ -164,7 +164,7 @@ impl VntInner {
ports,
config.use_channel_type,
config.first_latency,
config.tcp,
config.protocol,
config.packet_loss_rate,
config.packet_delay,
)?;
@@ -253,14 +253,14 @@ impl VntInner {
);
//初始化网络数据通道
let (udp_socket_sender, tcp_socket_sender) =
let (udp_socket_sender, connect_util) =
init_channel(tcp_listener, context.clone(), stop_manager.clone(), handler)?;
// 打洞逻辑
let punch = Punch::new(
context.clone(),
config.punch_model,
config.tcp,
tcp_socket_sender.clone(),
config.protocol.is_base_tcp(),
connect_util.clone(),
external_route.clone(),
nat_test.clone(),
current_device.clone(),
@@ -274,7 +274,7 @@ impl VntInner {
context.clone(),
current_device.clone(),
config_info.clone(),
tcp_socket_sender.clone(),
connect_util.clone(),
callback.clone(),
0,
handshake,
vnt/src/core/mod.rs
+29 -6
View File
@@ -5,7 +5,7 @@ use std::str::FromStr;
pub use conn::Vnt;
use crate::channel::punch::PunchModel;
use crate::channel::UseChannelType;
use crate::channel::{ConnectProtocol, UseChannelType};
use crate::cipher::CipherModel;
use crate::compression::Compressor;
use crate::util::{address_choose, dns_query_all};
@@ -28,7 +28,7 @@ pub struct Config {
pub out_ips: Vec<(u32, u32)>,
pub password: Option<String>,
pub mtu: Option<u32>,
pub tcp: bool,
pub protocol: ConnectProtocol,
pub ip: Option<Ipv4Addr>,
#[cfg(feature = "ip_proxy")]
#[cfg(feature = "integrated_tun")]
@@ -67,7 +67,6 @@ impl Config {
out_ips: Vec<(u32, u32)>,
password: Option<String>,
mtu: Option<u32>,
tcp: bool,
ip: Option<Ipv4Addr>,
#[cfg(feature = "integrated_tun")]
#[cfg(feature = "ip_proxy")]
@@ -109,8 +108,32 @@ impl Config {
if name.is_empty() || name.len() > 128 {
return Err(anyhow!("name too long"));
}
let server_address =
address_choose(dns_query_all(&server_address_str, name_servers.clone())?)?;
let mut server_address_str = server_address_str.to_lowercase();
let mut _query_dns = true;
let mut protocol = ConnectProtocol::UDP;
#[cfg(feature = "websocket")]
{
if server_address_str.starts_with("ws://") {
protocol = ConnectProtocol::WS;
_query_dns = false;
}
if server_address_str.starts_with("wss://") {
protocol = ConnectProtocol::WSS;
_query_dns = false;
}
}
let mut server_address = "0.0.0.0:0".parse().unwrap();
if _query_dns {
if let Some(s) = server_address_str.strip_prefix("udp://") {
server_address_str = s.to_string();
} else if let Some(s) = server_address_str.strip_prefix("tcp://") {
server_address_str = s.to_string();
protocol = ConnectProtocol::TCP;
}
server_address =
address_choose(dns_query_all(&server_address_str, name_servers.clone())?)?;
}
#[cfg(feature = "port_mapping")]
let port_mapping_list = crate::port_mapping::convert(port_mapping_list)?;
@@ -133,7 +156,7 @@ impl Config {
out_ips,
password,
mtu,
tcp,
protocol,
ip,
#[cfg(feature = "ip_proxy")]
#[cfg(feature = "integrated_tun")]
vnt/src/handle/maintain/addr_request.rs
+1 -1
View File
@@ -92,7 +92,7 @@ fn addr_request0(
return Ok(());
}
if current_dev.connect_server.is_ipv4() && !context.is_main_tcp() {
if current_dev.connect_server.is_ipv4() && !context.main_protocol().is_base_tcp() {
// 如果连接的是ipv4服务，则探测公网端口
let gateway_ip = current_dev.virtual_gateway;
let src_ip = current_dev.virtual_ip;
vnt/src/handle/maintain/idle.rs
+25 -25
View File
@@ -1,14 +1,13 @@
use std::io;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use crossbeam_utils::atomic::AtomicCell;
use mio::net::TcpStream;
use crate::channel::context::ChannelContext;
use crate::channel::idle::{Idle, IdleType};
use crate::channel::sender::AcceptSocketSender;
use crate::channel::sender::ConnectUtil;
use crate::channel::ConnectProtocol;
use crate::handle::callback::{ConnectInfo, ErrorType};
use crate::handle::handshaker::Handshake;
use crate::handle::{BaseConfigInfo, ConnectStatus, CurrentDeviceInfo};
@@ -36,7 +35,7 @@ pub fn idle_gateway<Call: VntCallback>(
context: ChannelContext,
current_device_info: Arc<AtomicCell<CurrentDeviceInfo>>,
config: BaseConfigInfo,
tcp_socket_sender: AcceptSocketSender<(TcpStream, SocketAddr, Option<Vec<u8>>)>,
connect_util: ConnectUtil,
call: Call,
mut connect_count: usize,
handshake: Handshake,
@@ -45,7 +44,7 @@ pub fn idle_gateway<Call: VntCallback>(
&context,
&current_device_info,
&config,
&tcp_socket_sender,
&connect_util,
&call,
&mut connect_count,
&handshake,
@@ -56,7 +55,7 @@ pub fn idle_gateway<Call: VntCallback>(
context,
current_device_info,
config,
tcp_socket_sender,
connect_util,
call,
connect_count,
handshake,
@@ -71,7 +70,7 @@ fn idle_gateway0<Call: VntCallback>(
context: &ChannelContext,
current_device: &AtomicCell<CurrentDeviceInfo>,
config: &BaseConfigInfo,
tcp_socket_sender: &AcceptSocketSender<(TcpStream, SocketAddr, Option<Vec<u8>>)>,
connect_util: &ConnectUtil,
call: &Call,
connect_count: &mut usize,
handshake: &Handshake,
@@ -80,7 +79,7 @@ fn idle_gateway0<Call: VntCallback>(
context,
current_device,
config,
tcp_socket_sender,
connect_util,
call,
connect_count,
handshake,
@@ -120,7 +119,7 @@ fn check_gateway_channel<Call: VntCallback>(
context: &ChannelContext,
current_device_info: &AtomicCell<CurrentDeviceInfo>,
config: &BaseConfigInfo,
tcp_socket_sender: &AcceptSocketSender<(TcpStream, SocketAddr, Option<Vec<u8>>)>,
connect_util: &ConnectUtil,
call: &Call,
count: &mut usize,
handshake: &Handshake,
@@ -128,28 +127,29 @@ fn check_gateway_channel<Call: VntCallback>(
let mut current_device = current_device_info.load();
if current_device.status.offline() {
*count += 1;
// 探测服务器地址
current_device = domain_request0(current_device_info, config);
let connect_protocol = context.main_protocol();
if connect_protocol.is_transport() {
// 传输层的协议需要探测服务器地址
current_device = domain_request0(current_device_info, config);
}
//需要重连
call.connect(ConnectInfo::new(*count, current_device.connect_server));
log::info!("发送握手请求,{:?}", config);
if let Err(e) = handshake.send(context, config.server_secret, current_device.connect_server)
{
log::warn!("{:?}", e);
if context.is_main_tcp() {
let request_packet = handshake.handshake_request_packet(config.server_secret)?;
//tcp需要重连
let tcp_stream = std::net::TcpStream::connect_timeout(
&current_device.connect_server,
Duration::from_secs(5),
)?;
tcp_stream.set_nonblocking(true)?;
if let Err(e) = tcp_socket_sender.try_add_socket((
TcpStream::from_std(tcp_stream),
current_device.connect_server,
Some(request_packet.into_buffer()),
)) {
log::warn!("{:?}", e)
let request_packet = handshake.handshake_request_packet(config.server_secret)?;
match connect_protocol {
ConnectProtocol::UDP => {}
ConnectProtocol::TCP => {
connect_util.try_connect_tcp(
request_packet.into_buffer(),
current_device.connect_server,
);
}
ConnectProtocol::WS | ConnectProtocol::WSS => {
connect_util
.try_connect_ws(request_packet.into_buffer(), config.server_addr.clone());
}
}
}
vnt/src/handle/recv_data/client.rs
+2 -2
View File
@@ -238,7 +238,7 @@ impl<Device: DeviceWrite> ClientPacketHandler<Device> {
//忽略掉来源于自己的包
if self
.nat_test
.is_local_address(route_key.is_tcp(), route_key.addr)
.is_local_address(route_key.protocol().is_base_tcp(), route_key.addr)
{
return Ok(());
}
@@ -261,7 +261,7 @@ impl<Device: DeviceWrite> ClientPacketHandler<Device> {
}
if self
.nat_test
.is_local_address(route_key.is_tcp(), route_key.addr)
.is_local_address(route_key.protocol().is_base_tcp(), route_key.addr)
{
return Ok(());
}
vnt/src/handle/recv_data/mod.rs
+3 -3
View File
@@ -44,7 +44,7 @@ pub struct RecvDataHandler<Call, Device> {
impl<Call: VntCallback, Device: DeviceWrite> RecvChannelHandler for RecvDataHandler<Call, Device> {
fn handle(
&mut self,
&self,
buf: &mut [u8],
extend: &mut [u8],
route_key: RouteKey,
@@ -54,7 +54,7 @@ impl<Call: VntCallback, Device: DeviceWrite> RecvChannelHandler for RecvDataHand
return;
}
//判断stun响应包
if !route_key.is_tcp() {
if route_key.protocol().is_udp() {
if let Ok(rs) = self
.nat_test
.recv_data(route_key.index(), route_key.addr, buf)
@@ -135,7 +135,7 @@ impl<Call: VntCallback, Device: DeviceWrite> RecvDataHandler<Call, Device> {
}
}
fn handle0(
&mut self,
&self,
buf: &mut [u8],
extend: &mut [u8],
route_key: RouteKey,
vnt/src/handle/recv_data/server.rs
+1 -3
View File
@@ -24,9 +24,7 @@ use crate::handle::callback::{ErrorInfo, ErrorType, HandshakeInfo, RegisterInfo,
use crate::handle::handshaker;
use crate::handle::handshaker::Handshake;
use crate::handle::recv_data::PacketHandler;
use crate::handle::{
registrar, BaseConfigInfo, ConnectStatus, CurrentDeviceInfo, PeerDeviceInfo, GATEWAY_IP,
};
use crate::handle::{registrar, BaseConfigInfo, ConnectStatus, CurrentDeviceInfo, PeerDeviceInfo};
use crate::nat::NatTest;
use crate::proto::message::{DeviceList, HandshakeResponse, RegistrationResponse};
use crate::protocol::body::ENCRYPTION_RESERVED;
vnt/src/util/counter/adder.rs
+1 -1
View File
@@ -90,7 +90,7 @@ impl U64Adder {
index: 0,
}
}
pub fn add(&mut self, num: u64) {
pub fn add(&self, num: u64) {
self.inner.base[self.index].add(num);
}
pub fn get(&self) -> u64 {