From 607c8c2e9f52fb48d0642bbb6875cd80a3deb64f Mon Sep 17 00:00:00 2001 From: lbl8603 <49143209+lbl8603@users.noreply.github.com> Date: Wed, 8 May 2024 21:31:21 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E7=AB=AF=E5=8F=A3=E6=98=A0?= =?UTF-8?q?=E5=B0=84=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- vnt-cli/Cargo.toml | 17 +-- vnt-cli/README.md | 27 ++++- vnt-cli/src/config/mod.rs | 8 +- vnt-cli/src/main.rs | 19 +++- .../java/top/wherewego/vnt/jni/Config.java | 11 ++ vnt-jni/src/config.rs | 4 +- vnt/src/core/conn.rs | 16 ++- vnt/src/core/mod.rs | 9 ++ vnt/src/lib.rs | 2 + vnt/src/port_mapping/mod.rs | 97 ++++++++++++++++ vnt/src/port_mapping/tcp_mapping.rs | 49 ++++++++ vnt/src/port_mapping/udp_mapping.rs | 107 ++++++++++++++++++ 12 files changed, 342 insertions(+), 24 deletions(-) create mode 100644 vnt/src/port_mapping/mod.rs create mode 100644 vnt/src/port_mapping/tcp_mapping.rs create mode 100644 vnt/src/port_mapping/udp_mapping.rs diff --git a/vnt-cli/Cargo.toml b/vnt-cli/Cargo.toml index e6d1193..80eede1 100644 --- a/vnt-cli/Cargo.toml +++ b/vnt-cli/Cargo.toml @@ -6,7 +6,7 @@ edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] -vnt = { path = "../vnt", package = "vnt",default-features = false } +vnt = { path = "../vnt", package = "vnt", default-features = false } common = { path = "../common" } getopts = "0.2.21" console = "0.15.2" @@ -28,16 +28,17 @@ sudo = "0.6.0" winapi = { version = "0.3.9", features = ["handleapi", "processthreadsapi", "winnt", "securitybaseapi", "impl-default"] } [features] -default = ["server_encrypt","aes_gcm","aes_cbc","aes_ecb","sm4_cbc","ip_proxy"] +default = ["server_encrypt", "aes_gcm", "aes_cbc", "aes_ecb", "sm4_cbc", "ip_proxy", "port_mapping"] 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"] -server_encrypt=["vnt/server_encrypt"] -ip_proxy=["vnt/ip_proxy"] +aes_cbc = ["vnt/aes_cbc"] +aes_ecb = ["vnt/aes_ecb"] +sm4_cbc = ["vnt/sm4_cbc"] +aes_gcm = ["vnt/aes_gcm"] +server_encrypt = ["vnt/server_encrypt"] +ip_proxy = ["vnt/ip_proxy"] +port_mapping = ["vnt/port_mapping"] [build-dependencies] embed-manifest = "1.4.0" rand = "0.8.5" diff --git a/vnt-cli/README.md b/vnt-cli/README.md index 026bac9..52da0e0 100644 --- a/vnt-cli/README.md +++ b/vnt-cli/README.md @@ -10,7 +10,7 @@ ### -c 关闭控制台交互式命令，后台运行时可以加此参数 ### -s `` -注册和中继服务器地址，注册和转发数据 +注册和中继服务器地址，注册和转发数据，以'TXT:'开头表示解析TXT记录，TXT记录内容必须是'host:port'形式的服务器地址 ### -e `` 使用stun服务探测客户端NAT类型，不同类型有不同的打洞策略 ### -a @@ -72,9 +72,9 @@ 注意：默认情况下服务端不会对中转的数据做校验，如果要对中转的数据做校验，则需要客户端、服务端都开启此参数 ### --punch `` -取值ipv4/ipv6，选择只使用ipv4打洞或者只使用ipv6打洞，默认两则都会使用 +取值ipv4/ipv6，选择只使用ipv4打洞或者只使用ipv6打洞，默认两者都会使用 ### --ports `` -指定本地监听的端口组，多个端口使用逗号分隔，多个端口可以分摊流量，增加并发，tcp会监听端口组的第一个端口，用于tcp直连 +指定本地监听的端口组，多个端口使用逗号分隔，多个端口可以分摊流量，增加并发、减缓流量限制，tcp会监听端口组的第一个端口，用于tcp直连 - 例1：‘--ports 12345,12346,12347’ 表示udp监听12345、12346、12347这三个端口，tcp监听12345端口 - 例2：‘--ports 0,0’ 表示udp监听两个未使用的端口，tcp监听一个未使用的端口 @@ -85,6 +85,14 @@ ### --no-proxy 关闭内置的ip代理，内置的代理较为简单，而且一般来说直接使用网卡NAT转发性能会更高， 有需要可以自行配置NAT转发，[可参考‘编译’小节中的NAT配置](https://github.com/lbl8603/vnt#%E7%BC%96%E8%AF%91) +### --dns `<223.5.5.5>` +设置域名解析服务器地址，可以设置多个。如果使用TXT记录的域名，则dns默认使用223.5.5.5和114.114.114.114，端口省略值为53 + +当地址解析失败时，会依次尝试后面的dns，直到有A记录、AAAA记录(或TXT记录)的解析结果 + +### --mapping `10.26.0.10:80>` +端口映射,可以设置多个映射地址，例如 '--mapping udp:0.0.0.0:80->10.26.0.10:80 --mapping tcp:0.0.0.0:80->10.26.0.11:81' +表示将本地udp 80端口的数据转发到10.26.0.10:80，将本地tcp 80端口的数据转发到10.26.0.11:81，转发的目的地址可以使用域名+端口 ### -f `` 指定配置文件 配置文件采用yaml格式，可参考： @@ -111,7 +119,7 @@ server_encrypt: true #服务端加密 parallel: 1 #任务并行度 cipher_model: aes_gcm #客户端加密算法 finger: false #关闭数据指纹 -punch_model: ipv4 #打洞模式 +punch_model: ipv4 #打洞模式，表示只使用ipv4地址打洞，默认会同时使用v6和v4 ports: - 0 #使用随机端口，tcp监听此端口 - 0 @@ -122,7 +130,12 @@ device_name: vnt-tun #网卡名称 packet_loss: 0 #指定丢包率 取值0~1之间的数 用于模拟弱网 packet_delay: 0 #指定延迟 单位毫秒 用于模拟弱网 dns: - - 8.8.8.8:53 + - 223.5.5.5 # 首选dns + - 8.8.8.8 # 备选dns +mapping: + - udp:0.0.0.0:80->10.26.0.10:80 # 映射udp数据 + - tcp:0.0.0.0:80->10.26.0.10:81 # 映射tcp数据 + - tcp:0.0.0.0:82->localhost:83 # 映射tcp数据 ``` 或者需要哪个配置就加哪个，当然token是必须的 @@ -134,7 +147,9 @@ token: xxx #组网token - relay:仅中继模式，会禁止打洞/p2p直连，只使用服务器转发 - p2p:仅直连模式，会禁止网络数据从服务器/客户端转发，只会使用服务器转发控制包 ### --packet-loss `<0>` -模拟丢包，取值0~1之间的小数，程序会按设定的概率主动丢包。在模拟弱网环境会有帮助。 +模拟丢包，取值0~1之间的小数，程序会按设定的概率主动丢包。在模拟弱网环境时会有帮助。 +### --packet-delay `<0>` +模拟延迟,整数,单位毫秒(ms),程序会按设定的值延迟发包,可用于模拟弱网 ### --list 在后台运行时,查看其他设备列表 diff --git a/vnt-cli/src/config/mod.rs b/vnt-cli/src/config/mod.rs index 4c42695..3d9dfff 100644 --- a/vnt-cli/src/config/mod.rs +++ b/vnt-cli/src/config/mod.rs @@ -40,6 +40,8 @@ pub struct FileConfig { pub device_name: Option, pub packet_loss: Option, pub packet_delay: u32, + #[cfg(feature = "port_mapping")] + pub mapping: Vec, } impl Default for FileConfig { @@ -54,7 +56,7 @@ impl Default for FileConfig { stun_server: vec![ "stun1.l.google.com:19302".to_string(), "stun2.l.google.com:19302".to_string(), - "stun.qq.com:3478".to_string(), + "stun.miwifi.com:3478".to_string(), ], dns: vec![], in_ips: vec![], @@ -77,6 +79,8 @@ impl Default for FileConfig { device_name: None, packet_loss: None, packet_delay: 0, + #[cfg(feature = "port_mapping")] + mapping: vec![], } } } @@ -157,6 +161,8 @@ pub fn read_config(file_path: &str) -> io::Result<(Config, bool)> { use_channel_type, file_conf.packet_loss, file_conf.packet_delay, + #[cfg(feature = "port_mapping")] + file_conf.mapping, ) .unwrap(); Ok((config, file_conf.cmd)) diff --git a/vnt-cli/src/main.rs b/vnt-cli/src/main.rs index 7b42ef3..3f1c0c4 100644 --- a/vnt-cli/src/main.rs +++ b/vnt-cli/src/main.rs @@ -73,6 +73,7 @@ fn main() { opts.optopt("", "packet-loss", "丢包率", ""); opts.optopt("", "packet-delay", "延迟", ""); opts.optmulti("", "dns", "dns", ""); + opts.optmulti("", "mapping", "mapping", ""); opts.optopt("f", "", "配置文件", ""); //"后台运行时,查看其他设备列表" opts.optflag("", "list", "后台运行时,查看其他设备列表"); @@ -156,7 +157,7 @@ fn main() { if stun_server.is_empty() { stun_server.push("stun1.l.google.com:19302".to_string()); stun_server.push("stun2.l.google.com:19302".to_string()); - stun_server.push("stun.qq.com:3478".to_string()); + stun_server.push("stun.miwifi.com:3478".to_string()); } let dns = matches.opt_strs("dns"); let in_ip = matches.opt_strs("i"); @@ -287,6 +288,8 @@ fn main() { .opt_get::("packet-delay") .expect("--packet-delay") .unwrap_or(0); + #[cfg(feature = "port_mapping")] + let port_mapping_list = matches.opt_strs("mapping"); let config = match Config::new( #[cfg(any(target_os = "windows", target_os = "linux"))] tap, @@ -315,6 +318,8 @@ fn main() { use_channel_type, packet_loss, packet_delay, + #[cfg(feature = "port_mapping")] + port_mapping_list, ) { Ok(config) => config, Err(e) => { @@ -338,6 +343,14 @@ fn main() { mod callback; fn main0(config: Config, show_cmd: bool) { + #[cfg(feature = "port_mapping")] + for (is_tcp, addr, dest) in config.port_mapping_list.iter() { + if *is_tcp { + println!("TCP port mapping {}->{}", addr, dest) + } else { + println!("UDP port mapping {}->{}", addr, dest) + } + } let vnt_util = Vnt::new(config, callback::VntHandler {}).unwrap(); let vnt_c = vnt_util.clone(); thread::Builder::new() @@ -412,7 +425,7 @@ fn print_usage(program: &str, _opts: Options) { println!(" -n 给设备一个名字,便于区分不同设备,默认使用系统版本"); println!(" -d 设备唯一标识符,不使用--ip参数时,服务端凭此参数分配虚拟ip,注意不能重复"); println!(" -s 注册和中继服务器地址,以'TXT:'开头表示解析TXT记录"); - println!(" -e stun服务器,用于探测NAT类型,可使用多个地址,如-e stun.qq.com -e stun1.l.google.com"); + println!(" -e stun服务器,用于探测NAT类型,可使用多个地址,如-e stun1.l.google.com -e stun2.l.google.com"); println!(" -a 使用tap模式,默认使用tun模式"); println!(" -i 配置点对网(IP代理)时使用,-i 192.168.0.0/24,10.26.0.3表示允许接收网段192.168.0.0/24的数据"); println!(" 并转发到10.26.0.3,可指定多个网段"); @@ -475,6 +488,8 @@ fn print_usage(program: &str, _opts: Options) { " --packet-delay <0> 模拟延迟,整数,单位毫秒(ms),程序会按设定的值延迟发包,可用于模拟弱网" ); println!(" --dns DNS服务器地址,可使用多个dns,不指定时使用系统解析"); + #[cfg(feature = "port_mapping")] + println!(" --mapping 端口映射,例如 --mapping udp:0.0.0.0:80->10.26.0.10:80 --mapping tcp:0.0.0.0:80->10.26.0.10:80"); println!(); println!( diff --git a/vnt-jni/java/top/wherewego/vnt/jni/Config.java b/vnt-jni/java/top/wherewego/vnt/jni/Config.java index 4fd3abc..342127d 100644 --- a/vnt-jni/java/top/wherewego/vnt/jni/Config.java +++ b/vnt-jni/java/top/wherewego/vnt/jni/Config.java @@ -54,6 +54,10 @@ public class Config { * dns地址 */ private String[] dns; + /** + * 端口映射 + */ + private String[] portMapping; /** * stun服务地址 */ @@ -201,6 +205,13 @@ public class Config { public void setDns(String[] dns) { this.dns = dns; } + public String[] getPortMapping() { + return portMapping; + } + + public void setPortMapping(String[] portMapping) { + this.portMapping = portMapping; + } public String[] getStunServer() { return stunServer; diff --git a/vnt-jni/src/config.rs b/vnt-jni/src/config.rs index 3afd5fe..eb1db2d 100644 --- a/vnt-jni/src/config.rs +++ b/vnt-jni/src/config.rs @@ -21,6 +21,7 @@ pub fn new_config(env: &mut JNIEnv, config: JObject) -> Result { let server_address_str = to_string_not_null(env, &config, "server")?; let stun_server = to_string_array_not_null(env, &config, "stunServer")?; let dns = to_string_array(env, &config, "dns")?.unwrap_or_else(|| vec![]); + let port_mapping = to_string_array(env, &config, "portMapping")?.unwrap_or_else(|| vec![]); let cipher_model = to_string_not_null(env, &config, "cipherModel")?; let punch_model = to_string(env, &config, "punchModel")?; let mtu = to_integer(env, &config, "mtu")?.map(|v| v as u32); @@ -116,12 +117,13 @@ pub fn new_config(env: &mut JNIEnv, config: JObject) -> Result { UseChannelType::from_str(&use_channel.unwrap_or_default()).unwrap_or_default(), packet_loss_rate, packet_delay, + port_mapping, ) { Ok(config) => config, Err(e) => { env.throw_new( "java/lang/RuntimeException", - format!("vnt start error {}", e), + format!("vnt start error {:?}", e), ) .expect("throw"); return Err(Error::JavaException); diff --git a/vnt/src/core/conn.rs b/vnt/src/core/conn.rs index 9bd965c..e51600a 100644 --- a/vnt/src/core/conn.rs +++ b/vnt/src/core/conn.rs @@ -93,6 +93,16 @@ impl Vnt { config.server_address_str.clone(), config.name_servers.clone(), ); + // 服务停止管理器 + let stop_manager = { + let callback = callback.clone(); + StopManager::new(move || callback.stop()) + }; + #[cfg(feature = "port_mapping")] + crate::port_mapping::start_port_mapping( + stop_manager.clone(), + config.port_mapping_list.clone(), + )?; let ports = config.ports.as_ref().map_or(vec![0, 0], |v| { if v.is_empty() { vec![0, 0] @@ -131,11 +141,6 @@ impl Vnt { callback.create_tun(tun_info); device }; - // 服务停止管理器 - let stop_manager = { - let callback = callback.clone(); - StopManager::new(move || callback.stop()) - }; // 定时器 let scheduler = Scheduler::new(stop_manager.clone())?; let external_route = ExternalRoute::new(config.in_ips.clone()); @@ -145,7 +150,6 @@ impl Vnt { let proxy_map = if !config.out_ips.is_empty() && !config.no_proxy { Some(crate::ip_proxy::init_proxy( context.clone(), - scheduler.clone(), stop_manager.clone(), current_device.clone(), client_cipher.clone(), diff --git a/vnt/src/core/mod.rs b/vnt/src/core/mod.rs index d45e446..3959897 100644 --- a/vnt/src/core/mod.rs +++ b/vnt/src/core/mod.rs @@ -43,6 +43,9 @@ pub struct Config { //控制丢包率 pub packet_loss_rate: Option, pub packet_delay: u32, + // 端口映射 + #[cfg(feature = "port_mapping")] + pub port_mapping_list: Vec<(bool, SocketAddr, String)>, } impl Config { @@ -72,6 +75,8 @@ impl Config { use_channel_type: UseChannelType, packet_loss_rate: Option, packet_delay: u32, + // 例如 [udp:127.0.0.1:80->10.26.0.10:8080,tcp:127.0.0.1:80->10.26.0.10:8080] + #[cfg(feature = "port_mapping")] port_mapping_list: Vec, ) -> anyhow::Result { for x in stun_server.iter_mut() { if !x.contains(":") { @@ -96,6 +101,8 @@ impl Config { } let 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)?; Ok(Self { #[cfg(any(target_os = "windows", target_os = "linux"))] tap, @@ -126,6 +133,8 @@ impl Config { use_channel_type, packet_loss_rate, packet_delay, + #[cfg(feature = "port_mapping")] + port_mapping_list, }) } } diff --git a/vnt/src/lib.rs b/vnt/src/lib.rs index 92ae669..a159e14 100644 --- a/vnt/src/lib.rs +++ b/vnt/src/lib.rs @@ -8,6 +8,8 @@ pub mod handle; #[cfg(feature = "ip_proxy")] pub mod ip_proxy; pub mod nat; +#[cfg(feature = "port_mapping")] +pub mod port_mapping; pub mod proto; pub mod protocol; pub mod tun_tap_device; diff --git a/vnt/src/port_mapping/mod.rs b/vnt/src/port_mapping/mod.rs new file mode 100644 index 0000000..9573311 --- /dev/null +++ b/vnt/src/port_mapping/mod.rs @@ -0,0 +1,97 @@ +use std::net::SocketAddr; +use std::str::FromStr; +use std::thread; + +use anyhow::Context; + +use crate::util::StopManager; + +mod tcp_mapping; + +mod udp_mapping; + +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_addr = split.next().with_context(|| { + format!( + "udp_mapping error {:?},eg: udp:127.0.0.1:80->10.26.0.10:8080", + x + ) + })?; + let bind_addr = SocketAddr::from_str(bind_addr) + .with_context(|| format!("udp_mapping error {}", bind_addr))?; + let dest = split.next().with_context(|| { + format!( + "udp_mapping error {:?},eg: udp:127.0.0.1:80->10.26.0.10:8080", + x + ) + })?; + rs.push((false, bind_addr, dest.to_string())); + continue; + } + if let Some(tcp_mapping) = string.strip_prefix("tcp:") { + let mut split = tcp_mapping.split("->"); + let bind_addr = split.next().with_context(|| { + format!( + "tcp_mapping error {:?},eg: tcp:127.0.0.1:80->10.26.0.10:8080", + x + ) + })?; + let bind_addr = SocketAddr::from_str(bind_addr) + .with_context(|| format!("udp_mapping error {}", bind_addr))?; + let dest = split.next().with_context(|| { + format!( + "tcp_mapping error {:?},eg: tcp:127.0.0.1:80->10.26.0.10:8080", + x + ) + })?; + rs.push((true, bind_addr, dest.to_string())); + continue; + } + Err(anyhow::anyhow!( + "port_mapping error {:?},eg: tcp:127.0.0.1:80->10.26.0.10:8080", + x + ))?; + } + Ok(rs) +} +pub fn start_port_mapping( + stop_manager: StopManager, + vec: Vec<(bool, SocketAddr, String)>, +) -> anyhow::Result<()> { + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .thread_name("portMapping") + .build()?; + runtime.block_on(start_port_mapping0(vec))?; + let (sender, receiver) = tokio::sync::oneshot::channel::<()>(); + let worker = stop_manager.add_listener("portMapping".into(), move || { + let _ = sender.send(()); + })?; + thread::Builder::new() + .name("portMapping".into()) + .spawn(move || { + runtime.block_on(async { + let _ = receiver.await; + }); + runtime.shutdown_background(); + drop(worker); + })?; + + Ok(()) +} + +async fn start_port_mapping0(vec: Vec<(bool, SocketAddr, String)>) -> anyhow::Result<()> { + for (is_tcp, bind_addr, destination) in vec { + if is_tcp { + tcp_mapping::tcp_mapping(bind_addr, destination).await?; + } else { + udp_mapping::udp_mapping(bind_addr, destination).await?; + } + } + Ok(()) +} diff --git a/vnt/src/port_mapping/tcp_mapping.rs b/vnt/src/port_mapping/tcp_mapping.rs new file mode 100644 index 0000000..32a717b --- /dev/null +++ b/vnt/src/port_mapping/tcp_mapping.rs @@ -0,0 +1,49 @@ +use anyhow::Context; +use std::net::SocketAddr; +use tokio::net::{TcpListener, TcpStream}; + +pub async fn tcp_mapping(bind_addr: SocketAddr, destination: String) -> anyhow::Result<()> { + let tcp_listener = TcpListener::bind(bind_addr) + .await + .with_context(|| format!("TCP binding {:?} failed", bind_addr))?; + tokio::spawn(tcp_mapping_(bind_addr, tcp_listener, destination)); + Ok(()) +} + +async fn tcp_mapping_( + bind_addr: SocketAddr, + tcp_listener: TcpListener, + destination: String, +) -> anyhow::Result<()> { + loop { + let (tcp_stream, _) = tcp_listener.accept().await?; + + let destination = destination.clone(); + tokio::spawn(async move { + if let Err(e) = copy(tcp_stream, &destination).await { + log::warn!("tcp port mapping {}->{} {:?}", bind_addr, destination, e); + } + }); + } +} + +async fn copy(source_tcp: TcpStream, destination: &String) -> anyhow::Result<()> { + let dest_tcp = TcpStream::connect(destination) + .await + .with_context(|| format!("TCP connection target failed {:?}", destination))?; + let _ = source_tcp.set_nodelay(true); + let _ = dest_tcp.set_nodelay(true); + + let destination = dest_tcp.peer_addr()?; + let (mut client_read, mut client_write) = source_tcp.into_split(); + let (mut server_read, mut server_write) = dest_tcp.into_split(); + tokio::spawn(async move { + if let Err(e) = tokio::io::copy(&mut client_read, &mut server_write).await { + log::warn!("client tcp proxy ->{:},{:?}", destination, e); + } + }); + if let Err(e) = tokio::io::copy(&mut server_read, &mut client_write).await { + log::warn!("server tcp proxy ->{:?},{:?}", destination, e); + } + Ok(()) +} diff --git a/vnt/src/port_mapping/udp_mapping.rs b/vnt/src/port_mapping/udp_mapping.rs new file mode 100644 index 0000000..56d8893 --- /dev/null +++ b/vnt/src/port_mapping/udp_mapping.rs @@ -0,0 +1,107 @@ +use anyhow::Context; +use crossbeam_utils::atomic::AtomicCell; +use parking_lot::Mutex; +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::{Duration, Instant}; +use tokio::net::UdpSocket; + +pub async fn udp_mapping(bind_addr: SocketAddr, destination: String) -> anyhow::Result<()> { + let udp = UdpSocket::bind(bind_addr) + .await + .with_context(|| format!("port proxy UDP binding {:?} failed", bind_addr))?; + let udp = Arc::new(udp); + + let inner_map: Arc, Arc>)>>> = + Arc::new(Mutex::new(HashMap::with_capacity(64))); + + tokio::spawn(async move { + let mut buf = [0; 65536]; + loop { + match udp.recv_from(&mut buf).await { + Ok((len, src_addr)) => { + if let Err(e) = + udp_mapping0(&buf[..len], src_addr, &inner_map, &udp, &destination).await + { + log::warn!("udp port mapping {}->{} {:?}", src_addr, destination, e); + } + } + Err(e) => { + log::warn!("port proxy UDP {:?}", e); + } + } + } + }); + Ok(()) +} + +async fn udp_mapping0( + buf: &[u8], + src_addr: SocketAddr, + inner_map: &Arc, Arc>)>>>, + udp_socket: &Arc, + destination: &String, +) -> anyhow::Result<()> { + let option = inner_map.lock().get(&src_addr).cloned(); + if let Some((udp, time)) = option { + time.store(Instant::now()); + udp.send(buf).await?; + } else { + let dest_udp = UdpSocket::bind("0.0.0.0:0").await?; + dest_udp.connect(destination).await?; + dest_udp.send(buf).await?; + let destination_addr = dest_udp.peer_addr()?; + let udp_socket = udp_socket.clone(); + let inner_map = inner_map.clone(); + let dest_udp = Arc::new(dest_udp); + let time = Arc::new(AtomicCell::new(Instant::now())); + inner_map + .lock() + .insert(src_addr, (dest_udp.clone(), time.clone())); + tokio::spawn(async move { + let mut buf = [0u8; 65536]; + loop { + match tokio::time::timeout(Duration::from_secs(600), dest_udp.recv(&mut buf)).await + { + Ok(rs) => match rs { + Ok(len) => match udp_socket.send_to(&buf[..len], src_addr).await { + Ok(_) => {} + Err(e) => { + log::warn!( + "udp port mapping {}->{} {:?}", + src_addr, + destination_addr, + e + ); + break; + } + }, + Err(e) => { + log::warn!( + "udp port mapping {}->{} {:?}", + src_addr, + destination_addr, + e + ); + break; + } + }, + Err(_) => { + if time.load().elapsed() > Duration::from_secs(580) { + //超时关闭 + log::warn!( + "udp port mapping timeout {}->{} ", + src_addr, + destination_addr + ); + break; + } + } + } + } + inner_map.lock().remove(&src_addr); + }); + } + Ok(()) +}