From 66129c2a247c18a26c60cdc8bc70af2721805290 Mon Sep 17 00:00:00 2001 From: lbl8603 <49143209+lbl8603@users.noreply.github.com> Date: Sat, 15 Jun 2024 20:52:35 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=AF=E6=8C=81=E6=97=A0tun=E6=A8=A1?= =?UTF-8?q?=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- vn-link/Cargo.toml | 29 +++++ vn-link/src/config.rs | 85 +++++++++++++ vn-link/src/in_mapping/mod.rs | 2 + vn-link/src/in_mapping/tcp.rs | 46 +++++++ vn-link/src/in_mapping/udp.rs | 63 ++++++++++ vn-link/src/lib.rs | 8 ++ vn-link/src/out_mapping/mod.rs | 2 + vn-link/src/out_mapping/tcp.rs | 64 ++++++++++ vn-link/src/out_mapping/udp.rs | 137 +++++++++++++++++++++ vn-link/src/vnt_link/mod.rs | 217 +++++++++++++++++++++++++++++++++ 10 files changed, 653 insertions(+) create mode 100644 vn-link/Cargo.toml create mode 100644 vn-link/src/config.rs create mode 100644 vn-link/src/in_mapping/mod.rs create mode 100644 vn-link/src/in_mapping/tcp.rs create mode 100644 vn-link/src/in_mapping/udp.rs create mode 100644 vn-link/src/lib.rs create mode 100644 vn-link/src/out_mapping/mod.rs create mode 100644 vn-link/src/out_mapping/tcp.rs create mode 100644 vn-link/src/out_mapping/udp.rs create mode 100644 vn-link/src/vnt_link/mod.rs diff --git a/vn-link/Cargo.toml b/vn-link/Cargo.toml new file mode 100644 index 0000000..b9429a5 --- /dev/null +++ b/vn-link/Cargo.toml @@ -0,0 +1,29 @@ +[package] +name = "vn-link" +version = "1.2.10" +edition = "2021" + +[dependencies] +lwip-rs = { git = "https://github.com/lbl8603/lwip-rs" } +vnt = { path = "../vnt", package = "vnt", default-features = false } +log = "0.4.17" +anyhow = "1.0.82" +parking_lot = "0.12.1" + +tokio = { version = "1.37.0", features = ["full"] } +crossbeam-utils = "0.8" + +[features] +default = ["server_encrypt", "aes_gcm", "aes_cbc", "aes_ecb", "sm4_cbc", "chacha20_poly1305", "port_mapping", "lz4_compress"] +openssl = ["vnt/openssl"] +openssl-vendored = ["vnt/openssl-vendored"] +ring-cipher = ["vnt/ring-cipher"] +aes_cbc = ["vnt/aes_cbc"] +aes_ecb = ["vnt/aes_ecb"] +sm4_cbc = ["vnt/sm4_cbc"] +aes_gcm = ["vnt/aes_gcm"] +chacha20_poly1305 = ["vnt/chacha20_poly1305"] +server_encrypt = ["vnt/server_encrypt"] +port_mapping = ["vnt/port_mapping"] +lz4_compress = ["vnt/lz4_compress"] +zstd_compress = ["vnt/zstd_compress"] diff --git a/vn-link/src/config.rs b/vn-link/src/config.rs new file mode 100644 index 0000000..58e2c6a --- /dev/null +++ b/vn-link/src/config.rs @@ -0,0 +1,85 @@ +use anyhow::Context; +use std::net::SocketAddr; +use std::str::FromStr; + +#[derive(Clone, Debug)] +pub struct VnLinkConfig { + pub mapping: Vec, +} + +impl VnLinkConfig { + pub fn new(mapping: Vec) -> Self { + Self { mapping } + } +} + +#[derive(Copy, Clone, Eq, PartialEq, Debug)] +pub enum LinkProtocol { + Tcp, + Udp, +} + +impl LinkProtocol { + pub fn is_tcp(&self) -> bool { + self == &LinkProtocol::Tcp + } +} + +#[derive(Copy, Clone, Debug)] +pub struct LinkItem { + pub protocol: LinkProtocol, + pub src_port: u16, + pub dest: SocketAddr, +} + +impl LinkItem { + pub fn new(protocol: LinkProtocol, src_port: u16, dest: SocketAddr) -> Self { + Self { + protocol, + src_port, + dest, + } + } +} + +pub fn convert(vec: Vec) -> anyhow::Result> { + let mut rs = Vec::with_capacity(vec.len()); + for x in vec { + let string = x.trim().to_lowercase(); + if let Some(udp_mapping) = string.strip_prefix("udp:") { + let mut split = udp_mapping.split("-"); + let bind_port = split + .next() + .with_context(|| format!("vnt-mapping error {:?},eg: udp:80-10.26.0.10:8080", x))?; + let bind_port = u16::from_str(bind_port) + .with_context(|| format!("udp_mapping error {}", bind_port))?; + let dest = split + .next() + .with_context(|| format!("vnt-mapping error {:?},eg: udp:80-10.26.0.10:8080", x))?; + let dest_addr = SocketAddr::from_str(dest) + .with_context(|| format!("udp_mapping error {}", dest))?; + rs.push(LinkItem::new(LinkProtocol::Udp, bind_port, dest_addr)); + continue; + } + if let Some(tcp_mapping) = string.strip_prefix("tcp:") { + let mut split = tcp_mapping.split("-"); + let bind_port = split + .next() + .with_context(|| format!("vnt-mapping error {:?},eg: tcp:80-10.26.0.10:8080", x))?; + let bind_port = u16::from_str(bind_port) + .with_context(|| format!("tcp_mapping error {}", bind_port))?; + let dest = split + .next() + .with_context(|| format!("vnt-mapping error {:?},eg: tcp:80-10.26.0.10:8080", x))?; + let dest_addr = SocketAddr::from_str(dest) + .with_context(|| format!("tcp_mapping error {}", dest))?; + rs.push(LinkItem::new(LinkProtocol::Tcp, bind_port, dest_addr)); + continue; + } + Err(anyhow::anyhow!( + "vnt-mapping error {:?},eg: tcp:80-10.26.0.10:8080", + x + ))?; + } + Ok(rs) +} diff --git a/vn-link/src/in_mapping/mod.rs b/vn-link/src/in_mapping/mod.rs new file mode 100644 index 0000000..f7a6a76 --- /dev/null +++ b/vn-link/src/in_mapping/mod.rs @@ -0,0 +1,2 @@ +pub mod tcp; +pub mod udp; diff --git a/vn-link/src/in_mapping/tcp.rs b/vn-link/src/in_mapping/tcp.rs new file mode 100644 index 0000000..d841469 --- /dev/null +++ b/vn-link/src/in_mapping/tcp.rs @@ -0,0 +1,46 @@ +use crate::out_mapping::tcp::tcp_copy; +use crossbeam_utils::atomic::AtomicCell; +use lwip_rs::tcp_stream::TcpStream as LwIpTcpStream; +use std::net::{IpAddr, SocketAddr}; +use std::sync::Arc; +use std::time::Duration; +use tokio::net::TcpListener; +use vnt::handle::CurrentDeviceInfo; + +pub async fn tcp_mapping_listen( + tcp_listener: TcpListener, + current_device: Arc>, + dest: SocketAddr, +) { + loop { + let (stream, addr) = match tcp_listener.accept().await { + Ok((stream, addr)) => (stream, addr), + Err(e) => { + log::warn!("tcp_mapping_listen {:?} dest {}", e, dest); + continue; + } + }; + let current_info = current_device.load(); + if current_info.virtual_ip.is_unspecified() { + continue; + } + if let IpAddr::V4(ip) = dest.ip() { + if ip == current_info.virtual_ip { + //防止用错参数的 + log::warn!("目的地址不能是本地虚拟ip tcp->{}", dest); + continue; + } + } + let src = SocketAddr::new(IpAddr::V4(current_info.virtual_ip), addr.port()); + tokio::spawn(async move { + match LwIpTcpStream::connect(src, dest, Duration::from_secs(5)).await { + Ok(lw_tcp) => { + tcp_copy(lw_tcp, stream); + } + Err(e) => { + log::warn!("{} {}->{} {}", addr, src, dest, e); + } + }; + }); + } +} diff --git a/vn-link/src/in_mapping/udp.rs b/vn-link/src/in_mapping/udp.rs new file mode 100644 index 0000000..b7ac232 --- /dev/null +++ b/vn-link/src/in_mapping/udp.rs @@ -0,0 +1,63 @@ +use std::collections::HashMap; +use std::net::{IpAddr, SocketAddr}; +use std::sync::Arc; +use std::time::Instant; + +use crossbeam_utils::atomic::AtomicCell; +use parking_lot::Mutex; +use tokio::net::UdpSocket; + +use lwip_rs::udp::UdpSocketWrite; +use vnt::handle::CurrentDeviceInfo; + +pub async fn udp_mapping_start( + udp: UdpSocket, + lwip_udp_write: UdpSocketWrite, + current_device: Arc>, + in_udp_map: &Arc< + Mutex< + HashMap< + (SocketAddr, SocketAddr), + (Arc, Option, Arc>), + >, + >, + >, + + dest: SocketAddr, +) { + let udp = Arc::new(udp); + let mut buf = [0u8; 65536]; + loop { + let (len, addr) = match udp.recv_from(&mut buf).await { + Ok(rs) => rs, + Err(e) => { + log::warn!("recv_from {} {}", dest, e); + continue; + } + }; + let current_info = current_device.load(); + if current_info.virtual_ip.is_unspecified() { + continue; + } + if let IpAddr::V4(ip) = dest.ip() { + if ip == current_info.virtual_ip { + //防止用错参数的 + log::warn!("目的地址不能是本地虚拟ip udp->{}", dest); + continue; + } + } + let src = SocketAddr::new(IpAddr::V4(current_info.virtual_ip), addr.port()); + in_udp_map.lock().insert( + (dest, src), + ( + udp.clone(), + Some(addr), + Arc::new(AtomicCell::new(Instant::now())), + ), + ); + + if let Err(e) = lwip_udp_write.send(&buf[..len], &src, &dest) { + log::warn!("lwip_udp_write {}->{} {}", src, dest, e); + } + } +} diff --git a/vn-link/src/lib.rs b/vn-link/src/lib.rs new file mode 100644 index 0000000..990bdb4 --- /dev/null +++ b/vn-link/src/lib.rs @@ -0,0 +1,8 @@ +pub mod config; +mod in_mapping; +mod out_mapping; +mod vnt_link; + +pub use vnt; + +pub use vnt_link::*; diff --git a/vn-link/src/out_mapping/mod.rs b/vn-link/src/out_mapping/mod.rs new file mode 100644 index 0000000..f7a6a76 --- /dev/null +++ b/vn-link/src/out_mapping/mod.rs @@ -0,0 +1,2 @@ +pub mod tcp; +pub mod udp; diff --git a/vn-link/src/out_mapping/tcp.rs b/vn-link/src/out_mapping/tcp.rs new file mode 100644 index 0000000..7f3e9f6 --- /dev/null +++ b/vn-link/src/out_mapping/tcp.rs @@ -0,0 +1,64 @@ +use std::net::{IpAddr, Ipv4Addr}; +use std::sync::Arc; + +use crossbeam_utils::atomic::AtomicCell; +use tokio::net::TcpStream; + +use lwip_rs::tcp_listener::TcpListener; +use lwip_rs::tcp_stream::TcpStream as LwIpTcpStream; +use vnt::handle::CurrentDeviceInfo; + +pub async fn tcp_mapping_listen( + mut tcp_listener: TcpListener, + current_device: Arc>, +) { + loop { + let stream = match tcp_listener.accept().await { + Ok(stream) => stream, + Err(e) => { + log::warn!("tcp_mapping_listen err {:?}", e); + break; + } + }; + let device_info = current_device.load(); + tokio::spawn(async move { + let dest = stream.dest_addr(); + let src = stream.src_addr(); + if let Err(e) = tcp_mapping_handle(stream, device_info).await { + log::warn!("tcp_mapping_handle {}->{} {:?}", src, dest, e) + } + }); + } +} + +async fn tcp_mapping_handle( + tcp_stream: LwIpTcpStream, + device_info: CurrentDeviceInfo, +) -> anyhow::Result<()> { + let mut dest = tcp_stream.dest_addr(); + // let src = tcp_stream.src_addr(); + if let IpAddr::V4(ip) = dest.ip() { + if ip.is_unspecified() + || ip.is_broadcast() + || ip.is_multicast() + || ip == device_info.virtual_ip + || ip == device_info.broadcast_ip + { + //是自己 + dest.set_ip(IpAddr::V4(Ipv4Addr::LOCALHOST)); + } + } + let peer_stream = TcpStream::connect(dest).await?; + if dest.port() == peer_stream.local_addr()?.port() { + return Err(anyhow::anyhow!("tcp port loop")); + } + tcp_copy(tcp_stream, peer_stream); + Ok(()) +} + +pub(crate) fn tcp_copy(lw_tcp: LwIpTcpStream, tokio_tcp: TcpStream) { + let (mut write, mut read) = lw_tcp.into_split(); + let (mut peer_read, mut peer_write) = tokio_tcp.into_split(); + tokio::spawn(async move { tokio::io::copy(&mut read, &mut peer_write).await }); + tokio::spawn(async move { tokio::io::copy(&mut peer_read, &mut write).await }); +} diff --git a/vn-link/src/out_mapping/udp.rs b/vn-link/src/out_mapping/udp.rs new file mode 100644 index 0000000..04c030c --- /dev/null +++ b/vn-link/src/out_mapping/udp.rs @@ -0,0 +1,137 @@ +use crossbeam_utils::atomic::AtomicCell; +use lwip_rs::udp::{UdpSocketRead, UdpSocketWrite}; +use parking_lot::Mutex; +use std::collections::HashMap; +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::sync::Arc; +use std::time::{Duration, Instant}; +use tokio::net::UdpSocket; +use vnt::handle::CurrentDeviceInfo; + +pub async fn udp_mapping_start( + lwip_udp_write: UdpSocketWrite, + mut lwip_udp_read: UdpSocketRead, + current_device: Arc>, + in_udp_map: Arc< + Mutex< + HashMap< + (SocketAddr, SocketAddr), + (Arc, Option, Arc>), + >, + >, + >, +) { + loop { + let (buf, src, dest) = match lwip_udp_read.recv().await { + Ok(rs) => rs, + Err(e) => { + log::warn!("udp_mapping err {:?}", e); + break; + } + }; + if let Err(e) = handle( + ¤t_device, + &lwip_udp_write, + &in_udp_map, + buf, + src, + dest, + ) + .await + { + log::warn!("udp_mapping err {}->{} {:?}", src, dest, e) + } + } +} + +async fn handle( + current_device: &AtomicCell, + lwip_udp_write: &UdpSocketWrite, + map: &Arc< + Mutex< + HashMap< + (SocketAddr, SocketAddr), + (Arc, Option, Arc>), + >, + >, + >, + buf: Vec, + src: SocketAddr, + dest: SocketAddr, +) -> anyhow::Result<()> { + let option = map.lock().get(&(src, dest)).cloned(); + + if let Some((dest_udp, addr, time)) = option { + time.store(Instant::now()); + if let Some(addr) = addr { + dest_udp.send_to(&buf, addr).await?; + } else { + dest_udp.send(&buf).await?; + } + } else { + let mut real_dest = dest; + let peer_udp_socket = match UdpSocket::bind(format!("0.0.0.0:{}", src.port())).await { + Ok(udp) => udp, + Err(_) => UdpSocket::bind("0.0.0.0:0").await?, + }; + if let IpAddr::V4(ip) = dest.ip() { + let device_info = current_device.load(); + if ip.is_unspecified() + || ip.is_broadcast() + || ip.is_multicast() + || ip == device_info.virtual_ip + || ip == device_info.broadcast_ip + { + //是自己 + real_dest.set_ip(IpAddr::V4(Ipv4Addr::LOCALHOST)); + } + } + peer_udp_socket.connect(real_dest).await?; + peer_udp_socket.send(&buf).await?; + let peer_udp_socket = Arc::new(peer_udp_socket); + let time = Arc::new(AtomicCell::new(Instant::now())); + let map = map.clone(); + map.lock() + .insert((src, dest), (peer_udp_socket.clone(), None, time.clone())); + let lwip_udp_write = lwip_udp_write.clone(); + tokio::spawn(async move { + peer_udp_handle(peer_udp_socket, lwip_udp_write, src, dest, time).await; + map.lock().remove(&(src, dest)); + }); + } + Ok(()) +} + +async fn peer_udp_handle( + peer_udp_socket: Arc, + lwip_udp_write: UdpSocketWrite, + src: SocketAddr, + dest: SocketAddr, + time: Arc>, +) { + let mut buf = [0u8; 65536]; + loop { + match tokio::time::timeout(Duration::from_secs(600), peer_udp_socket.recv(&mut buf)).await { + Ok(rs) => match rs { + Ok(len) => match lwip_udp_write.send(&buf[..len], &dest, &src) { + Ok(_) => {} + Err(e) => { + log::warn!("udp proxy {}->{} {:?}", dest, src, e); + break; + } + }, + Err(e) => { + log::warn!("udp proxy {}->{} {:?}", dest, src, e); + break; + } + }, + Err(_) => { + if time.load().elapsed() > Duration::from_secs(580) { + //超时关闭 + log::warn!("udp proxy timeout {}->{}", dest, src,); + break; + } + } + } + } +} diff --git a/vn-link/src/vnt_link/mod.rs b/vn-link/src/vnt_link/mod.rs new file mode 100644 index 0000000..c009efd --- /dev/null +++ b/vn-link/src/vnt_link/mod.rs @@ -0,0 +1,217 @@ +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Instant; + +use anyhow::Context; +use crossbeam_utils::atomic::AtomicCell; +use parking_lot::Mutex; +use tokio::net::{TcpListener, UdpSocket}; +use tokio::sync::watch::{channel, Sender}; + +use lwip_rs::stack::{NetStack, NetStackWrite}; +use lwip_rs::tcp_listener::TcpListener as LwIPTcpListener; +use lwip_rs::udp::{UdpSocket as LwIpUdpSocket, UdpSocketWrite}; +use vnt::channel::BUFFER_SIZE; +use vnt::core::{Config, Vnt}; +use vnt::packet::ip::ipv4::packet::IpV4Packet; +use vnt::protocol::HEAD_LEN; +use vnt::vnt_device::DeviceWrite; +use vnt::VntCallback; + +use crate::config::{LinkItem, LinkProtocol, VnLinkConfig}; +use crate::{in_mapping, out_mapping}; + +pub struct VnLink { + vnt: Vnt, + in_udp_map: Arc< + Mutex< + HashMap< + (SocketAddr, SocketAddr), + (Arc, Option, Arc>), + >, + >, + >, + lwip_udp_write: UdpSocketWrite, + shutdown_tx: Sender, +} + +impl VnLink { + pub async fn new( + vnt_config: Config, + vn_link_config: VnLinkConfig, + callback: Call, + ) -> anyhow::Result { + let stack = NetStack::new(HEAD_LEN, 1024, vnt_config.mtu.unwrap_or(1420) as u16).await; + let udp = LwIpUdpSocket::new()?; + let tcp_listener = LwIPTcpListener::new()?; + let (shutdown_tx, shutdown_rx) = channel(false); + let (net_stack_write, mut net_stack_read) = stack.into_split(); + let vnt = Vnt::new_device(vnt_config, callback, VntDevice { net_stack_write })?; + let shutdown_tx_ = shutdown_tx.clone(); + let w = vnt.add_stop_listener("vnt-link".into(), move || { + let _ = shutdown_tx_.send(true); + })?; + let ip_sender = vnt.ipv4_packet_sender().unwrap(); + let mut shutdown_rx_ = shutdown_rx.clone(); + tokio::spawn(async move { + let mut extend = [0; BUFFER_SIZE]; + loop { + tokio::select! { + _ = shutdown_rx_.changed() => { + break; + } + rs = net_stack_read.recv_ip() => { + match rs{ + Ok((mut buf, start_index, len)) => { + let ipv4_packet = if let Ok(packet) = + IpV4Packet::new(&buf[start_index..len]) + { + packet + } else { + continue; + }; + let destination_ip = ipv4_packet.destination_ip(); + let source_ip = ipv4_packet.source_ip(); + + if let Err(e) = ip_sender.send_ip(&mut buf, len, &mut extend, destination_ip) { + log::warn!("{}->{},{}", source_ip, destination_ip, e); + } + }, + Err(e) => { + log::error!("net_stack_read {:?}", e); + break; + } + }; + } + } + } + w.stop_all(); + }); + + let (lwip_udp_write, lwip_udp_read) = udp.into_split(); + let in_udp_map: Arc< + Mutex< + HashMap< + (SocketAddr, SocketAddr), + (Arc, Option, Arc>), + >, + >, + > = Arc::new(Mutex::new(HashMap::new())); + + let current_device_info = vnt.current_device_info(); + let in_udp_map_ = in_udp_map.clone(); + let lwip_udp_write_ = lwip_udp_write.clone(); + let vnt_ = vnt.clone(); + let mut shutdown_rx_ = shutdown_rx.clone(); + tokio::spawn(async move { + tokio::select! { + _ = shutdown_rx_.changed() => {} + _ = out_mapping::udp::udp_mapping_start( + lwip_udp_write_, + lwip_udp_read, + current_device_info, + in_udp_map_, + ) => {} + } + + vnt_.stop(); + }); + let current_device_info = vnt.current_device_info(); + let vnt_ = vnt.clone(); + let mut shutdown_rx_ = shutdown_rx.clone(); + + tokio::spawn(async move { + tokio::select! { + _ = shutdown_rx_.changed() => {} + _ = out_mapping::tcp::tcp_mapping_listen(tcp_listener, current_device_info) => {} + } + vnt_.stop(); + }); + let link = Self { + vnt, + in_udp_map, + lwip_udp_write, + shutdown_tx, + }; + link.add_mapping(vn_link_config.mapping).await?; + Ok(link) + } + pub async fn add_mapping(&self, mapping: Vec) -> anyhow::Result<()> { + for item in mapping { + let current_device_info = self.vnt.current_device_info(); + if item.dest.ip().is_unspecified() { + Err(anyhow::anyhow!("dest_address {:?} is_unspecified", item))? + } + let mut shutdown_rx_ = self.shutdown_tx.subscribe(); + if *shutdown_rx_.borrow() { + Err(anyhow::anyhow!("mapping stop"))? + } + if item.protocol == LinkProtocol::Udp { + let lwip_udp_write = self.lwip_udp_write.clone(); + let in_udp_map = self.in_udp_map.clone(); + //只能本机访问，不然不同IP的相同来源端口会有问题 + let udp = UdpSocket::bind(format!("127.0.0.1:{}", item.src_port)) + .await + .with_context(|| format!("udp bind failed {}", item.src_port))?; + tokio::spawn(async move { + tokio::select! { + _ = shutdown_rx_.changed() => {} + _ = in_mapping::udp::udp_mapping_start( + udp, + lwip_udp_write, + current_device_info, + &in_udp_map, + item.dest, + ) => {} + } + }); + } else { + let listener = TcpListener::bind(format!("127.0.0.1:{}", item.src_port)) + .await + .with_context(|| format!("tcp bind failed {}", item.src_port))?; + tokio::spawn(async move { + tokio::select! { + _ = shutdown_rx_.changed() => {} + _ = in_mapping::tcp::tcp_mapping_listen( + listener, + current_device_info, + item.dest, + ) => {} + } + }); + } + } + Ok(()) + } + pub fn stop(&self) { + self.as_vnt().stop() + } + pub async fn wait(&self) { + loop { + let mut receiver = self.shutdown_tx.subscribe(); + if *receiver.borrow() { + return; + } + if receiver.changed().await.is_err() { + return; + } + } + } + + pub fn as_vnt(&self) -> &Vnt { + &self.vnt + } +} + +#[derive(Clone)] +pub struct VntDevice { + net_stack_write: NetStackWrite, +} + +impl DeviceWrite for VntDevice { + fn write(&self, buf: &[u8]) -> std::io::Result { + self.net_stack_write.send_ip(buf)?; + Ok(buf.len()) + } +}