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

支持wg

Browse Source
This commit is contained in:
lbl8603
2024-07-20 10:38:18 +08:00
parent 14ee9c4632
commit 7ebdffcdd3
22 changed files with 282 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/src/cli.rs
+4
View File
@@ -75,6 +75,7 @@ pub fn parse_args_config() -> anyhow::Result<Option<(Config, Vec<String>, bool)>
opts.optopt("f", "", "配置文件", "<conf>");
opts.optopt("", "compressor", "压缩算法", "<lz4>");
opts.optflag("", "disable-stats", "关闭流量统计");
opts.optflag("", "allow-wg", "允许接入WireGuard");
//"后台运行时,查看其他设备列表"
opts.optflag("", "add", "后台运行时,添加地址");
opts.optflag("", "list", "后台运行时,查看其他设备列表");
@@ -281,6 +282,7 @@ pub fn parse_args_config() -> anyhow::Result<Option<(Config, Vec<String>, bool)>
let port_mapping_list = matches.opt_strs("mapping");
let vnt_mapping_list = matches.opt_strs("vnt-mapping");
let disable_stats = matches.opt_present("disable-stats");
let allow_wire_guard = matches.opt_present("allow-wg");
let compressor = if let Some(compressor) = matches.opt_str("compressor").as_ref() {
Compressor::from_str(compressor)
.map_err(|e| anyhow!("{}", e))
@@ -321,6 +323,7 @@ pub fn parse_args_config() -> anyhow::Result<Option<(Config, Vec<String>, bool)>
port_mapping_list,
compressor,
!disable_stats,
allow_wire_guard,
) {
Ok(config) => config,
Err(e) => {
@@ -435,6 +438,7 @@ fn print_usage(program: &str, _opts: Options) {
)
);
println!(" --disable-stats 关闭流量统计");
println!(" --allow-wg 允许接入WireGuard客户端");
println!();
#[cfg(feature = "command")]
{
common/src/command/entity.rs
+1
View File
@@ -45,6 +45,7 @@ pub struct DeviceItem {
pub client_secret_hash: Vec<u8>,
pub current_client_secret: bool,
pub current_client_secret_hash: Vec<u8>,
pub wire_guard: bool,
}
#[derive(Serialize, Deserialize, Debug, Default)]
common/src/command/mod.rs
+1
View File
@@ -219,6 +219,7 @@ pub fn command_list(vnt: &Vnt) -> Vec<DeviceItem> {
client_secret_hash: peer.client_secret_hash,
current_client_secret,
current_client_secret_hash: client_encrypt_hash.to_vec(),
wire_guard: peer.wireguard,
};
list.push(item);
}
common/src/config/file_config.rs
+4
View File
@@ -46,6 +46,8 @@ pub struct FileConfig {
pub compressor: Option<String>,
pub vnt_mapping: Vec<String>,
pub disable_stats: bool,
// 允许传递wg流量
pub allow_wire_guard: bool,
}
impl Default for FileConfig {
@@ -90,6 +92,7 @@ impl Default for FileConfig {
compressor: None,
vnt_mapping: vec![],
disable_stats: false,
allow_wire_guard: false,
}
}
}
@@ -177,6 +180,7 @@ pub fn read_config(file_path: &str) -> anyhow::Result<(Config, Vec<String>, bool
file_conf.mapping,
compressor,
!file_conf.disable_stats,
file_conf.allow_wire_guard,
)?;
Ok((config, file_conf.vnt_mapping, file_conf.cmd))
common/src/console_out/mod.rs
+14 -8
View File
@@ -132,15 +132,21 @@ pub fn console_device_list(mut list: Vec<DeviceItem>) {
("Rt".to_string(), Style::new()),
]);
for item in list {
let name = if item.wire_guard {
format!("{}(wg)", item.name)
} else {
item.name
};
if &item.status == "Online" {
if item.client_secret != item.current_client_secret
|| (!item.current_client_secret_hash.is_empty()
&& !item.client_secret_hash.is_empty()
&& item.current_client_secret_hash != item.client_secret_hash)
if !item.wire_guard
&& (item.client_secret != item.current_client_secret
|| (!item.current_client_secret_hash.is_empty()
&& !item.client_secret_hash.is_empty()
&& item.current_client_secret_hash != item.client_secret_hash))
{
//加密状态不一致，无法通信的
out_list.push(vec![
(item.name, Style::new().red()),
(name, Style::new().red()),
(item.virtual_ip, Style::new().red()),
(item.status, Style::new().red()),
("Mismatch".to_string(), Style::new().red()),
@@ -149,7 +155,7 @@ pub fn console_device_list(mut list: Vec<DeviceItem>) {
} else {
if item.nat_traversal_type.contains("p2p") {
out_list.push(vec![
(item.name, Style::new().green()),
(name, Style::new().green()),
(item.virtual_ip, Style::new().green()),
(item.status, Style::new().green()),
(item.nat_traversal_type, Style::new().green()),
@@ -157,7 +163,7 @@ pub fn console_device_list(mut list: Vec<DeviceItem>) {
]);
} else {
out_list.push(vec![
(item.name, Style::new().yellow()),
(name, Style::new().yellow()),
(item.virtual_ip, Style::new().yellow()),
(item.status, Style::new().yellow()),
(item.nat_traversal_type, Style::new().yellow()),
@@ -167,7 +173,7 @@ pub fn console_device_list(mut list: Vec<DeviceItem>) {
}
} else {
out_list.push(vec![
(item.name, Style::new().color256(102)),
(name, Style::new().color256(102)),
(item.virtual_ip, Style::new().color256(102)),
(item.status, Style::new().color256(102)),
("".to_string(), Style::new().color256(102)),
vn-link-cli/Cargo.toml
+1 -1
View File
@@ -11,7 +11,7 @@ log = "0.4.17"
[features]
default = ["default-feature"]
default-feature = ["server_encrypt", "aes_gcm", "aes_cbc", "aes_ecb", "sm4_cbc", "chacha20_poly1305", "port_mapping", "log", "command", "file_config", "lz4"]
default-feature = ["server_encrypt", "aes_gcm", "aes_cbc", "aes_ecb", "sm4_cbc", "chacha20_poly1305", "port_mapping", "log", "command", "file_config", "lz4", "ws"]
openssl = ["vn-link/openssl", "common/openssl"]
openssl-vendored = ["vn-link/openssl-vendored", "common/openssl-vendored"]
vnt/proto/message.proto
+1
View File
@@ -43,6 +43,7 @@ message DeviceInfo {
uint32 device_status = 3;
bool client_secret = 4;
bytes client_secret_hash = 5;
bool wireguard = 6;
}
message DeviceList {
vnt/src/channel/punch.rs
+2 -3
View File
@@ -264,9 +264,8 @@ impl Punch {
if let Some(ipv4_addr) = nat_info.local_tcp_ipv4addr() {
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));
for ip in &nat_info.public_ips {
let addr = SocketAddr::V4(SocketAddrV4::new(*ip, nat_info.tcp_port));
self.connect_tcp(buf, addr)
}
}
vnt/src/channel/sender.rs
+85 -5
View File
@@ -1,9 +1,11 @@
use std::collections::HashMap;
use std::io;
use std::net::{Ipv4Addr, SocketAddr};
use std::sync::mpsc::{SyncSender, TrySendError};
use std::sync::Arc;
use crossbeam_utils::atomic::AtomicCell;
use parking_lot::Mutex;
use tokio::sync::mpsc::Sender;
use crate::channel::context::ChannelContext;
@@ -11,7 +13,7 @@ use crate::channel::notify::AcceptNotify;
use crate::cipher::Cipher;
use crate::compression::Compressor;
use crate::external_route::ExternalRoute;
use crate::handle::CurrentDeviceInfo;
use crate::handle::{CurrentDeviceInfo, PeerDeviceInfo};
use crate::protocol;
use crate::protocol::{ip_turn_packet, NetPacket};
@@ -21,7 +23,10 @@ pub struct IpPacketSender {
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
compressor: Compressor,
client_cipher: Cipher,
server_cipher: Cipher,
ip_route: ExternalRoute,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
allow_wire_guard: bool,
}
impl IpPacketSender {
@@ -30,14 +35,20 @@ impl IpPacketSender {
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
compressor: Compressor,
client_cipher: Cipher,
server_cipher: Cipher,
ip_route: ExternalRoute,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
allow_wire_guard: bool,
) -> Self {
Self {
context,
current_device,
compressor,
client_cipher,
server_cipher,
ip_route,
device_map,
allow_wire_guard,
}
}
pub fn self_virtual_ip(&self) -> Ipv4Addr {
@@ -58,19 +69,54 @@ impl IpPacketSender {
if let Some(v) = self.ip_route.route(&dest_ip) {
dest_ip = v;
}
if dest_ip.is_multicast() || dest_ip.is_broadcast() || dest_ip == device_info.broadcast_ip {
if dest_ip.is_multicast() {
//广播
dest_ip = Ipv4Addr::BROADCAST;
}
let mut net_packet = NetPacket::new0(data_len, buf)?;
let mut auxiliary = NetPacket::new(auxiliary_buf)?;
net_packet.set_default_version();
net_packet.set_protocol(protocol::Protocol::IpTurn);
net_packet.set_transport_protocol(ip_turn_packet::Protocol::Ipv4.into());
net_packet.first_set_ttl(6);
net_packet.set_source(src_ip);
net_packet.set_destination(dest_ip);
if self.allow_wire_guard {
if dest_ip.is_broadcast() || dest_ip == device_info.broadcast_ip {
let exists_wg = self
.device_map
.lock()
.1
.values()
.any(|v| v.status.is_online() && v.wireguard);
if exists_wg {
send_to_wg_broadcast(
&self.context,
&net_packet,
&self.server_cipher,
&device_info,
)?;
}
} else {
let guard = self.device_map.lock();
if let Some(peer_info) = guard.1.get(&dest_ip) {
if peer_info.wireguard {
if peer_info.status.is_offline() {
return Ok(());
}
drop(guard);
send_to_wg(
&self.context,
&mut net_packet,
&self.server_cipher,
&device_info,
)?;
return Ok(());
}
}
}
}
let mut auxiliary = NetPacket::new(auxiliary_buf)?;
let mut net_packet = if self.compressor.compress(&net_packet, &mut auxiliary)? {
auxiliary.set_default_version();
@@ -84,7 +130,7 @@ impl IpPacketSender {
net_packet
};
self.client_cipher.encrypt_ipv4(&mut net_packet)?;
if dest_ip.is_broadcast() {
if dest_ip.is_broadcast() || dest_ip == device_info.broadcast_ip {
//走服务端广播
self.context
.send_default(&net_packet, device_info.connect_server)?;
@@ -105,6 +151,40 @@ impl IpPacketSender {
}
}
pub fn send_to_wg_broadcast(
sender: &ChannelContext,
net_packet: &NetPacket<&mut [u8]>,
server_cipher: &Cipher,
current_device: &CurrentDeviceInfo,
) -> anyhow::Result<()> {
let mut copy_packet = NetPacket::new0(net_packet.data_len(), [0; 65536])?;
copy_packet.set_default_version();
copy_packet.set_protocol(protocol::Protocol::IpTurn);
copy_packet.set_transport_protocol(ip_turn_packet::Protocol::WGIpv4.into());
copy_packet.first_set_ttl(6);
copy_packet.set_source(net_packet.source());
copy_packet.set_destination(net_packet.destination());
copy_packet.set_gateway_flag(true);
copy_packet.set_payload(net_packet.payload())?;
server_cipher.encrypt_ipv4(&mut copy_packet)?;
sender.send_default(&copy_packet, current_device.connect_server)?;
Ok(())
}
pub fn send_to_wg(
sender: &ChannelContext,
net_packet: &mut NetPacket<&mut [u8]>,
server_cipher: &Cipher,
current_device: &CurrentDeviceInfo,
) -> anyhow::Result<()> {
net_packet.set_transport_protocol(ip_turn_packet::Protocol::WGIpv4.into());
net_packet.set_gateway_flag(true);
server_cipher.encrypt_ipv4(net_packet)?;
sender.send_default(&net_packet, current_device.connect_server)?;
Ok(())
}
pub struct AcceptSocketSender<T> {
sender: SyncSender<T>,
notify: AcceptNotify,
vnt/src/core/conn.rs
+21 -14
View File
@@ -66,12 +66,13 @@ pub struct VntInner {
config: Config,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
nat_test: NatTest,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
context: Arc<Mutex<Option<ChannelContext>>>,
peer_nat_info_map: Arc<RwLock<HashMap<Ipv4Addr, NatInfo>>>,
client_secret_hash: Option<[u8; 16]>,
compressor: Compressor,
client_cipher: Cipher,
server_cipher: Cipher,
external_route: ExternalRoute,
up_traffic_meter: Option<TrafficMeterMultiAddress>,
down_traffic_meter: Option<TrafficMeterMultiAddress>,
@@ -128,8 +129,8 @@ impl VntInner {
config.server_address,
)));
//设备列表
let device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>> =
Arc::new(Mutex::new((0, Vec::with_capacity(16))));
let device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>> =
Arc::new(Mutex::new((0, HashMap::with_capacity(16))));
//基础信息
let config_info = BaseConfigInfo::new(
config.name.clone(),
@@ -147,6 +148,7 @@ impl VntInner {
#[cfg(feature = "integrated_tun")]
#[cfg(any(target_os = "windows", target_os = "linux", target_os = "macos"))]
config.device_name.clone(),
config.allow_wire_guard,
);
// 服务停止管理器
let stop_manager = {
@@ -228,7 +230,7 @@ impl VntInner {
proxy_map.clone(),
client_cipher.clone(),
server_cipher.clone(),
device_list.clone(),
device_map.clone(),
config.compressor,
device.clone().into_device_adapter(),
)
@@ -241,7 +243,7 @@ impl VntInner {
client_cipher.clone(),
current_device.clone(),
device,
device_list.clone(),
device_map.clone(),
config_info.clone(),
nat_test.clone(),
callback.clone(),
@@ -287,7 +289,7 @@ impl VntInner {
{
let context = context.clone();
let nat_test = nat_test.clone();
let device_list = device_list.clone();
let device_map = device_map.clone();
let config_info = config_info.clone();
let current_device = current_device.clone();
if !config.use_channel_type.is_only_relay() {
@@ -300,13 +302,14 @@ impl VntInner {
);
}
let client_cipher = client_cipher.clone();
let server_cipher = server_cipher.clone();
//延迟启动
scheduler.timeout(Duration::from_secs(3), move |scheduler| {
start(
scheduler,
context,
nat_test,
device_list,
device_map,
current_device,
client_cipher,
server_cipher,
@@ -323,12 +326,13 @@ impl VntInner {
config,
current_device,
nat_test,
device_list,
device_map,
context: Arc::new(Mutex::new(Some(context))),
peer_nat_info_map,
client_secret_hash: config_info.client_secret_hash,
compressor,
client_cipher,
server_cipher,
external_route,
up_traffic_meter,
down_traffic_meter,
@@ -340,7 +344,7 @@ pub fn start<Call: VntCallback>(
scheduler: &Scheduler,
context: ChannelContext,
nat_test: NatTest,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
client_cipher: Cipher,
server_cipher: Cipher,
@@ -354,7 +358,7 @@ pub fn start<Call: VntCallback>(
&scheduler,
context.clone(),
current_device.clone(),
device_list.clone(),
device_map.clone(),
client_cipher.clone(),
server_cipher.clone(),
);
@@ -374,7 +378,7 @@ pub fn start<Call: VntCallback>(
&scheduler,
context.clone(),
current_device.clone(),
device_list.clone(),
device_map.clone(),
client_cipher.clone(),
);
}
@@ -393,7 +397,7 @@ pub fn start<Call: VntCallback>(
&scheduler,
context.clone(),
nat_test.clone(),
device_list.clone(),
device_map.clone(),
current_device.clone(),
client_cipher.clone(),
punch_receiver,
@@ -432,10 +436,10 @@ impl VntInner {
self.nat_test.nat_info()
}
pub fn device_list(&self) -> Vec<PeerDeviceInfo> {
let device_list_lock = self.device_list.lock();
let device_list_lock = self.device_map.lock();
let (_epoch, device_list) = device_list_lock.clone();
drop(device_list_lock);
device_list
device_list.into_values().collect()
}
pub fn route(&self, ip: &Ipv4Addr) -> Option<Route> {
self.context.lock().as_ref()?.route_table.route_one(ip)
@@ -507,7 +511,10 @@ impl VntInner {
self.current_device.clone(),
self.compressor.clone(),
self.client_cipher.clone(),
self.server_cipher.clone(),
self.external_route.clone(),
self.device_map.clone(),
self.config.allow_wire_guard,
))
} else {
None
vnt/src/core/mod.rs
+4
View File
@@ -51,6 +51,7 @@ pub struct Config {
pub port_mapping_list: Vec<(bool, SocketAddr, String)>,
pub compressor: Compressor,
pub enable_traffic: bool,
pub allow_wire_guard: bool,
}
impl Config {
@@ -88,6 +89,8 @@ impl Config {
#[cfg(feature = "port_mapping")] port_mapping_list: Vec<String>,
compressor: Compressor,
enable_traffic: bool,
// 允许传递wg流量
allow_wire_guard: bool,
) -> anyhow::Result<Self> {
for x in stun_server.iter_mut() {
if !x.contains(":") {
@@ -180,6 +183,7 @@ impl Config {
port_mapping_list,
compressor,
enable_traffic,
allow_wire_guard,
})
}
}
vnt/src/handle/maintain/heartbeat.rs
+24 -20
View File
@@ -1,3 +1,4 @@
use std::collections::HashMap;
use std::net::Ipv4Addr;
use std::sync::Arc;
use std::time::Duration;
@@ -19,14 +20,14 @@ pub fn heartbeat(
scheduler: &Scheduler,
context: ChannelContext,
current_device_info: Arc<AtomicCell<CurrentDeviceInfo>>,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
client_cipher: Cipher,
server_cipher: Cipher,
) {
heartbeat0(
&context,
&current_device_info.load(),
&device_list,
&device_map,
&client_cipher,
&server_cipher,
);
@@ -36,7 +37,7 @@ pub fn heartbeat(
s,
context,
current_device_info,
device_list,
device_map,
client_cipher,
server_cipher,
)
@@ -49,7 +50,7 @@ pub fn heartbeat(
fn heartbeat0(
context: &ChannelContext,
current_device: &CurrentDeviceInfo,
device_list: &Mutex<(u16, Vec<PeerDeviceInfo>)>,
device_map: &Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>,
client_cipher: &Cipher,
server_cipher: &Cipher,
) {
@@ -57,7 +58,7 @@ fn heartbeat0(
let src_ip = current_device.virtual_ip;
// 可能服务器ip发生变化，导致发送失败
let mut is_send_gateway = false;
match heartbeat_packet_server(device_list, server_cipher, src_ip, gateway_ip) {
match heartbeat_packet_server(device_map, server_cipher, src_ip, gateway_ip) {
Ok(net_packet) => {
if let Err(e) = context.send_default(&net_packet, current_device.connect_server) {
log::warn!("heartbeat err={:?}", e)
@@ -75,7 +76,7 @@ fn heartbeat0(
if is_send_gateway {
continue;
}
heartbeat_packet_server(device_list, server_cipher, src_ip, gateway_ip)
heartbeat_packet_server(device_map, server_cipher, src_ip, gateway_ip)
} else {
heartbeat_packet_client(client_cipher, src_ip, dest_ip)
};
@@ -92,9 +93,9 @@ fn heartbeat0(
}
}
}
let peer_list = { device_list.lock().1.clone() };
for peer in &peer_list {
if !peer.status.is_online() {
let peer_list = { device_map.lock().1.clone() };
for peer in peer_list.values() {
if !peer.status.is_online() || peer.wireguard {
continue;
}
if current_device.is_gateway(&peer.virtual_ip) {
@@ -124,11 +125,11 @@ pub fn client_relay(
scheduler: &Scheduler,
context: ChannelContext,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
client_cipher: Cipher,
) {
let rs = scheduler.timeout(Duration::from_secs(30), move |s| {
client_relay_(s, context, current_device, device_list, client_cipher)
client_relay_(s, context, current_device, device_map, client_cipher)
});
if !rs {
log::info!("定时任务停止");
@@ -140,19 +141,19 @@ fn client_relay_(
scheduler: &Scheduler,
context: ChannelContext,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
client_cipher: Cipher,
) {
if let Err(e) = client_relay0(
&context,
&current_device.load(),
&device_list,
&device_map,
&client_cipher,
) {
log::error!("{:?}", e);
}
let rs = scheduler.timeout(Duration::from_secs(30), move |s| {
client_relay_(s, context, current_device, device_list, client_cipher)
client_relay_(s, context, current_device, device_map, client_cipher)
});
if !rs {
log::info!("定时任务停止");
@@ -162,17 +163,20 @@ fn client_relay_(
fn client_relay0(
context: &ChannelContext,
current_device: &CurrentDeviceInfo,
device_list: &Mutex<(u16, Vec<PeerDeviceInfo>)>,
device_map: &Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>,
client_cipher: &Cipher,
) -> anyhow::Result<()> {
// 离线了不再探测
if current_device.status.offline() {
return Ok(());
}
let peer_list = { device_list.lock().1.clone() };
let peer_list = { device_map.lock().1.clone() };
let mut routes = context.route_table.route_table_p2p();
for peer in &peer_list {
if !peer.status.is_online() || peer.virtual_ip == current_device.virtual_ip {
for peer in peer_list.values() {
if peer.wireguard
|| !peer.status.is_online()
|| peer.virtual_ip == current_device.virtual_ip
{
continue;
}
if context
@@ -232,14 +236,14 @@ fn heartbeat_packet_client(
}
fn heartbeat_packet_server(
device_list: &Mutex<(u16, Vec<PeerDeviceInfo>)>,
device_map: &Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>,
server_cipher: &Cipher,
src: Ipv4Addr,
dest: Ipv4Addr,
) -> anyhow::Result<NetPacket<[u8; 12 + 4 + ENCRYPTION_RESERVED]>> {
let mut net_packet = heartbeat_packet(src, dest)?;
let mut ping = PingPacket::new(net_packet.payload_mut())?;
ping.set_epoch(device_list.lock().0);
ping.set_epoch(device_map.lock().0);
net_packet.set_gateway_flag(true);
server_cipher.encrypt_ipv4(&mut net_packet)?;
Ok(net_packet)
vnt/src/handle/maintain/punch.rs
+9 -9
View File
@@ -90,7 +90,7 @@ pub fn punch(
scheduler: &Scheduler,
context: ChannelContext,
nat_test: NatTest,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
client_cipher: Cipher,
receiver: PunchReceiver,
@@ -102,7 +102,7 @@ pub fn punch(
scheduler,
context,
nat_test,
device_list,
device_map,
current_device.clone(),
client_cipher.clone(),
0,
@@ -170,7 +170,7 @@ fn punch_request(
scheduler: &Scheduler,
context: ChannelContext,
nat_test: NatTest,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
client_cipher: Cipher,
count: usize,
@@ -182,7 +182,7 @@ fn punch_request(
if let Err(e) = punch0(
&context,
&nat_test,
&device_list,
&device_map,
curr,
&client_cipher,
&punch_record,
@@ -201,7 +201,7 @@ fn punch_request(
s,
context,
nat_test,
device_list,
device_map,
current_device,
client_cipher,
count + 1,
@@ -218,7 +218,7 @@ fn punch_request(
fn punch0(
context: &ChannelContext,
nat_test: &NatTest,
device_list: &Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: &Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
current_device: CurrentDeviceInfo,
client_cipher: &Cipher,
punch_record: &Mutex<HashMap<Ipv4Addr, usize>>,
@@ -237,11 +237,11 @@ fn punch0(
return Ok(());
}
let current_ip = current_device.virtual_ip;
let mut list: Vec<PeerDeviceInfo> = device_list
let mut list: Vec<PeerDeviceInfo> = device_map
.lock()
.1
.iter()
.filter(|info| info.status.is_online() && info.virtual_ip > current_ip)
.values()
.filter(|info| !info.wireguard && info.status.is_online() && info.virtual_ip > current_ip)
.cloned()
.collect();
list.shuffle(&mut rand::thread_rng());
vnt/src/handle/mod.rs
+9
View File
@@ -29,6 +29,7 @@ pub struct PeerDeviceInfo {
pub status: PeerDeviceStatus,
pub client_secret: bool,
pub client_secret_hash: Vec<u8>,
pub wireguard: bool,
}
impl PeerDeviceInfo {
@@ -38,6 +39,7 @@ impl PeerDeviceInfo {
status: u8,
client_secret: bool,
client_secret_hash: Vec<u8>,
wireguard: bool,
) -> Self {
Self {
virtual_ip,
@@ -45,6 +47,7 @@ impl PeerDeviceInfo {
status: PeerDeviceStatus::from(status),
client_secret,
client_secret_hash,
wireguard,
}
}
}
@@ -66,6 +69,7 @@ pub struct BaseConfigInfo {
#[cfg(feature = "integrated_tun")]
#[cfg(any(target_os = "windows", target_os = "linux", target_os = "macos"))]
pub device_name: Option<String>,
pub allow_wire_guard: bool,
}
impl BaseConfigInfo {
@@ -85,6 +89,7 @@ impl BaseConfigInfo {
#[cfg(feature = "integrated_tun")]
#[cfg(any(target_os = "windows", target_os = "linux", target_os = "macos"))]
device_name: Option<String>,
allow_wire_guard: bool,
) -> Self {
Self {
name,
@@ -102,6 +107,7 @@ impl BaseConfigInfo {
#[cfg(feature = "integrated_tun")]
#[cfg(any(target_os = "windows", target_os = "linux", target_os = "macos"))]
device_name,
allow_wire_guard,
}
}
}
@@ -116,6 +122,9 @@ impl PeerDeviceStatus {
pub fn is_online(&self) -> bool {
self == &PeerDeviceStatus::Online
}
pub fn is_offline(&self) -> bool {
self == &PeerDeviceStatus::Offline
}
}
impl Into<u8> for PeerDeviceStatus {
vnt/src/handle/recv_data/client.rs
+3
View File
@@ -194,6 +194,9 @@ impl<Device: DeviceWrite> ClientPacketHandler<Device> {
}
self.device.write(net_packet.payload())?;
}
ip_turn_packet::Protocol::WGIpv4 => {
// WG客户端的数据不会直接发过来，不用处理
}
ip_turn_packet::Protocol::Ipv4Broadcast => {
//客户端不帮忙转发广播包，所以不会出现这种类型的数据
}
vnt/src/handle/recv_data/mod.rs
+2 -2
View File
@@ -80,7 +80,7 @@ impl<Call: VntCallback, Device: DeviceWrite> RecvDataHandler<Call, Device> {
client_cipher: Cipher,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
device: Device,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
config_info: BaseConfigInfo,
nat_test: NatTest,
callback: Call,
@@ -101,7 +101,7 @@ impl<Call: VntCallback, Device: DeviceWrite> RecvDataHandler<Call, Device> {
server_cipher,
current_device.clone(),
device.clone(),
device_list,
device_map,
config_info,
nat_test.clone(),
callback,
vnt/src/handle/recv_data/server.rs
+19 -8
View File
@@ -1,4 +1,5 @@
use anyhow::anyhow;
use std::collections::HashMap;
use std::io;
use std::net::Ipv4Addr;
use std::sync::Arc;
@@ -42,7 +43,7 @@ pub struct ServerPacketHandler<Call, Device> {
server_cipher: Cipher,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
device: Device,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
config_info: BaseConfigInfo,
nat_test: NatTest,
callback: Call,
@@ -60,7 +61,7 @@ impl<Call, Device> ServerPacketHandler<Call, Device> {
server_cipher: Cipher,
current_device: Arc<AtomicCell<CurrentDeviceInfo>>,
device: Device,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
config_info: BaseConfigInfo,
nat_test: NatTest,
callback: Call,
@@ -75,7 +76,7 @@ impl<Call, Device> ServerPacketHandler<Call, Device> {
server_cipher,
current_device,
device,
device_list,
device_map,
config_info,
nat_test,
callback,
@@ -246,6 +247,11 @@ impl<Call: VntCallback, Device: DeviceWrite> PacketHandler for ServerPacketHandl
_ => {}
}
}
ip_turn_packet::Protocol::WGIpv4 => {
if self.config_info.allow_wire_guard {
self.device.write(net_packet.payload())?;
}
}
ip_turn_packet::Protocol::Ipv4Broadcast => {}
ip_turn_packet::Protocol::Unknown(_) => {}
}
@@ -353,7 +359,8 @@ impl<Call: VntCallback, Device: DeviceWrite> ServerPacketHandler<Call, Device> {
);
log::info!("tun信息{:?}", tun_info);
self.callback.create_tun(tun_info);
self.tun_device_helper.start(device)?;
self.tun_device_helper
.start(device, self.config_info.allow_wire_guard)?;
}
Err(e) => {
log::error!("{:?}", e);
@@ -435,14 +442,18 @@ impl<Call: VntCallback, Device: DeviceWrite> ServerPacketHandler<Call, Device> {
info.device_status as u8,
info.client_secret,
info.client_secret_hash,
info.wireguard,
)
})
.collect();
{
let mut dev = self.device_list.lock();
let mut dev = self.device_map.lock();
//这里可能会收到旧的消息，但是随着时间推移总会收到新的
dev.0 = epoch;
dev.1 = ip_list.clone();
dev.1.clear();
for info in ip_list.clone() {
dev.1.insert(info.virtual_ip, info);
}
}
self.callback.peer_client_list(
ip_list
@@ -506,7 +517,7 @@ impl<Call: VntCallback, Device: DeviceWrite> ServerPacketHandler<Call, Device> {
self.callback.error(err);
//掉线epoch要归零
{
let mut dev = self.device_list.lock();
let mut dev = self.device_map.lock();
dev.0 = 0;
drop(dev);
}
@@ -554,7 +565,7 @@ impl<Call: VntCallback, Device: DeviceWrite> ServerPacketHandler<Call, Device> {
let rt = (current_time - pong_packet.time()) as i64;
let route = Route::from(route_key, metric, rt);
context.route_table.add_route(net_packet.source(), route);
let epoch = self.device_list.lock().0;
let epoch = self.device_map.lock().0;
if pong_packet.epoch() != epoch {
//纪元不一致，可能有新客户端连接，向服务端拉取客户端列表
let mut poll_device = NetPacket::new_encrypt([0; 12 + ENCRYPTION_RESERVED])?;
vnt/src/handle/tun_tap/tun_handler.rs
+42 -12
View File
@@ -1,10 +1,10 @@
use crossbeam_utils::atomic::AtomicCell;
use parking_lot::Mutex;
use std::collections::HashMap;
use std::net::Ipv4Addr;
use std::sync::Arc;
use std::{io, thread};
use crossbeam_utils::atomic::AtomicCell;
use parking_lot::Mutex;
use packet::icmp::icmp::IcmpPacket;
use packet::icmp::Kind;
use packet::ip::ipv4::packet::IpV4Packet;
@@ -13,6 +13,7 @@ use tun::device::IFace;
use tun::Device;
use crate::channel::context::ChannelContext;
use crate::channel::sender::{send_to_wg, send_to_wg_broadcast};
use crate::cipher::Cipher;
use crate::compression::Compressor;
use crate::external_route::ExternalRoute;
@@ -27,7 +28,6 @@ use crate::protocol::body::ENCRYPTION_RESERVED;
use crate::protocol::ip_turn_packet::BroadcastPacket;
use crate::protocol::{ip_turn_packet, NetPacket, MAX_TTL};
use crate::util::StopManager;
fn icmp(device_writer: &Device, mut ipv4_packet: IpV4Packet<&mut [u8]>) -> anyhow::Result<()> {
if ipv4_packet.protocol() == Protocol::Icmp {
let mut icmp = IcmpPacket::new(ipv4_packet.payload_mut())?;
@@ -53,9 +53,10 @@ pub fn start(
#[cfg(feature = "ip_proxy")] ip_proxy_map: Option<IpProxyMap>,
client_cipher: Cipher,
server_cipher: Cipher,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
compressor: Compressor,
device_stop: DeviceStop,
allow_wire_guard: bool,
) -> io::Result<()> {
thread::Builder::new()
.name("tunHandlerS".into())
@@ -70,9 +71,10 @@ pub fn start(
ip_proxy_map,
client_cipher,
server_cipher,
device_list,
device_map,
compressor,
device_stop,
allow_wire_guard,
) {
log::warn!("stop:{}", e);
}
@@ -86,13 +88,13 @@ fn broadcast(
sender: &ChannelContext,
net_packet: &mut NetPacket<&mut [u8]>,
current_device: &CurrentDeviceInfo,
device_list: &Mutex<(u16, Vec<PeerDeviceInfo>)>,
device_map: &Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>,
) -> anyhow::Result<()> {
let list: Vec<Ipv4Addr> = device_list
let list: Vec<Ipv4Addr> = device_map
.lock()
.1
.iter()
.filter(|info| info.status.is_online())
.values()
.filter(|info| !info.wireguard && info.status.is_online())
.map(|info| info.virtual_ip)
.collect();
const MAX_COUNT: usize = 8;
@@ -177,8 +179,9 @@ pub(crate) fn handle(
#[cfg(feature = "ip_proxy")] proxy_map: &Option<IpProxyMap>,
client_cipher: &Cipher,
server_cipher: &Cipher,
device_list: &Mutex<(u16, Vec<PeerDeviceInfo>)>,
device_map: &Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>,
compressor: &Compressor,
allow_wire_guard: bool,
) -> anyhow::Result<()> {
//忽略掉结构不对的情况（ipv6数据、win tap会读到空数据），不然日志打印太多了
let ipv4_packet = match IpV4Packet::new(&mut buf[12..data_len]) {
@@ -237,6 +240,33 @@ pub(crate) fn handle(
dest_ip = Ipv4Addr::BROADCAST;
net_packet.set_destination(Ipv4Addr::BROADCAST);
}
let is_broadcast = dest_ip.is_broadcast() || current_device.broadcast_ip == dest_ip;
if allow_wire_guard {
if is_broadcast {
// wg客户端和vnt客户端分开广播
let exists_wg = device_map
.lock()
.1
.values()
.any(|v| v.status.is_online() && v.wireguard);
if exists_wg {
send_to_wg_broadcast(context, &net_packet, server_cipher, &current_device)?;
}
} else {
// 如果是wg客户端则发到vnts转发
let guard = device_map.lock();
if let Some(peer_info) = guard.1.get(&dest_ip) {
if peer_info.wireguard {
if peer_info.status.is_offline() {
return Ok(());
}
drop(guard);
send_to_wg(context, &mut net_packet, server_cipher, &current_device)?;
return Ok(());
}
}
}
}
let mut net_packet = if compressor.compress(&net_packet, &mut out)? {
out.set_default_version();
@@ -257,7 +287,7 @@ pub(crate) fn handle(
context,
&mut net_packet,
&current_device,
device_list,
device_map,
)?;
return Ok(());
}
vnt/src/handle/tun_tap/unix.rs
+10 -4
View File
@@ -13,7 +13,9 @@ use mio::event::Source;
use mio::unix::SourceFd;
use mio::{Events, Interest, Poll, Token, Waker};
use parking_lot::Mutex;
use std::collections::HashMap;
use std::io;
use std::net::Ipv4Addr;
use std::os::fd::AsRawFd;
use std::sync::Arc;
use tun::Device;
@@ -30,9 +32,10 @@ pub(crate) fn start_simple(
#[cfg(feature = "ip_proxy")] ip_proxy_map: Option<IpProxyMap>,
client_cipher: Cipher,
server_cipher: Cipher,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
compressor: Compressor,
device_stop: DeviceStop,
allow_wire_guard: bool,
) -> anyhow::Result<()> {
let poll = Poll::new()?;
let waker = Arc::new(Waker::new(poll.registry(), STOP)?);
@@ -61,8 +64,9 @@ pub(crate) fn start_simple(
ip_proxy_map,
client_cipher,
server_cipher,
device_list,
device_map,
compressor,
allow_wire_guard,
) {
log::error!("{:?}", e);
};
@@ -83,8 +87,9 @@ fn start_simple0(
#[cfg(feature = "ip_proxy")] ip_proxy_map: Option<IpProxyMap>,
client_cipher: Cipher,
server_cipher: Cipher,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
compressor: Compressor,
allow_wire_guard: bool,
) -> anyhow::Result<()> {
let mut buf = [0; BUFFER_SIZE];
let mut extend = [0; BUFFER_SIZE];
@@ -134,8 +139,9 @@ fn start_simple0(
&ip_proxy_map,
&client_cipher,
&server_cipher,
&device_list,
&device_map,
&compressor,
allow_wire_guard,
) {
Ok(_) => {}
Err(e) => {
vnt/src/handle/tun_tap/windows.rs
+10 -4
View File
@@ -10,6 +10,8 @@ use crate::ip_proxy::IpProxyMap;
use crate::util::StopManager;
use crossbeam_utils::atomic::AtomicCell;
use parking_lot::Mutex;
use std::collections::HashMap;
use std::net::Ipv4Addr;
use std::sync::Arc;
use tun::device::IFace;
use tun::Device;
@@ -23,9 +25,10 @@ pub(crate) fn start_simple(
#[cfg(feature = "ip_proxy")] ip_proxy_map: Option<IpProxyMap>,
client_cipher: Cipher,
server_cipher: Cipher,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
compressor: Compressor,
device_stop: DeviceStop,
allow_wire_guard: bool,
) -> anyhow::Result<()> {
let worker = {
let device = device.clone();
@@ -54,8 +57,9 @@ pub(crate) fn start_simple(
ip_proxy_map,
client_cipher,
server_cipher,
device_list,
device_map,
compressor,
allow_wire_guard,
) {
log::error!("{:?}", e);
}
@@ -74,8 +78,9 @@ fn start_simple0(
#[cfg(feature = "ip_proxy")] ip_proxy_map: Option<IpProxyMap>,
client_cipher: Cipher,
server_cipher: Cipher,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
compressor: Compressor,
allow_wire_guard: bool,
) -> anyhow::Result<()> {
let mut buf = [0; BUFFER_SIZE];
let mut extend = [0; BUFFER_SIZE];
@@ -96,8 +101,9 @@ fn start_simple0(
&ip_proxy_map,
&client_cipher,
&server_cipher,
&device_list,
&device_map,
&compressor,
allow_wire_guard,
) {
Ok(_) => {}
Err(e) => {
vnt/src/protocol/ip_turn_packet.rs
+8 -3
View File
@@ -1,9 +1,12 @@
#![allow(dead_code)]
use std::io;
use std::net::Ipv4Addr;
#[derive(Copy, Clone, Eq, PartialEq, Debug)]
pub enum Protocol {
Ipv4,
WGIpv4,
Ipv4Broadcast,
Unknown(u8),
}
@@ -12,16 +15,18 @@ impl From<u8> for Protocol {
fn from(value: u8) -> Self {
match value {
4 => Protocol::Ipv4,
5 => Protocol::WGIpv4,
201 => Protocol::Ipv4Broadcast,
val => Protocol::Unknown(val),
}
}
}
impl Into<u8> for Protocol {
fn into(self) -> u8 {
match self {
impl From<Protocol> for u8 {
fn from(val: Protocol) -> Self {
match val {
Protocol::Ipv4 => 4,
Protocol::WGIpv4 => 5,
Protocol::Ipv4Broadcast => 201,
Protocol::Unknown(val) => val,
}
vnt/src/tun_tap_device/tun_create_helper.rs
+8 -5
View File
@@ -1,4 +1,6 @@
use std::collections::HashMap;
use std::io;
use std::net::Ipv4Addr;
use std::sync::Arc;
use crossbeam_utils::atomic::AtomicCell;
@@ -67,7 +69,7 @@ struct TunDeviceHelperInner {
ip_proxy_map: Option<IpProxyMap>,
client_cipher: Cipher,
server_cipher: Cipher,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
compressor: Compressor,
}
@@ -80,7 +82,7 @@ impl TunDeviceHelper {
#[cfg(feature = "ip_proxy")] ip_proxy_map: Option<IpProxyMap>,
client_cipher: Cipher,
server_cipher: Cipher,
device_list: Arc<Mutex<(u16, Vec<PeerDeviceInfo>)>>,
device_map: Arc<Mutex<(u16, HashMap<Ipv4Addr, PeerDeviceInfo>)>>,
compressor: Compressor,
device_adapter: DeviceAdapter,
) -> Self {
@@ -93,7 +95,7 @@ impl TunDeviceHelper {
ip_proxy_map,
client_cipher,
server_cipher,
device_list,
device_map,
compressor,
};
Self {
@@ -117,7 +119,7 @@ impl TunDeviceHelper {
}
}
/// 要保证先stop 再start
pub fn start(&self, device: Arc<Device>) -> io::Result<()> {
pub fn start(&self, device: Arc<Device>, allow_wire_guard: bool) -> io::Result<()> {
self.device_adapter.insert(device.clone());
let device_stop = DeviceStop::default();
let s = self.device_stop.lock().replace(device_stop.clone());
@@ -133,9 +135,10 @@ impl TunDeviceHelper {
inner.ip_proxy_map,
inner.client_cipher,
inner.server_cipher,
inner.device_list,
inner.device_map,
inner.compressor,
device_stop,
allow_wire_guard,
)
}
}