增加端口映射功能

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_cbc = ["vnt/aes_cbc"]
-aes_ecb=["vnt/aes_ecb"]
+aes_ecb = ["vnt/aes_ecb"]
-sm4_cbc=["vnt/sm4_cbc"]
+sm4_cbc = ["vnt/sm4_cbc"]
-aes_gcm=["vnt/aes_gcm"]
+aes_gcm = ["vnt/aes_gcm"]
-server_encrypt=["vnt/server_encrypt"]
+server_encrypt = ["vnt/server_encrypt"]
-ip_proxy=["vnt/ip_proxy"]
+ip_proxy = ["vnt/ip_proxy"]
+port_mapping = ["vnt/port_mapping"]
 [build-dependencies]
 embed-manifest = "1.4.0"
 rand = "0.8.5"

vnt-cli/README.md

@@ -10,7 +10,7 @@
 ### -c
 关闭控制台交互式命令，后台运行时可以加此参数
 ### -s `<server>`
-注册和中继服务器地址，注册和转发数据
+注册和中继服务器地址，注册和转发数据，以'TXT:'开头表示解析TXT记录，TXT记录内容必须是'host:port'形式的服务器地址
 ### -e `<stun-server>`
 使用stun服务探测客户端NAT类型，不同类型有不同的打洞策略
 ### -a
@@ -72,9 +72,9 @@
 
 注意：默认情况下服务端不会对中转的数据做校验，如果要对中转的数据做校验，则需要客户端、服务端都开启此参数
 ### --punch `<punch>`
-取值ipv4/ipv6，选择只使用ipv4打洞或者只使用ipv6打洞，默认两则都会使用
+取值ipv4/ipv6，选择只使用ipv4打洞或者只使用ipv6打洞，默认两者都会使用
 ### --ports `<port1,port2>`
-指定本地监听的端口组，多个端口使用逗号分隔，多个端口可以分摊流量，增加并发，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 `<udp:0.0.0.0:80->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 `<conf>`
 指定配置文件
 配置文件采用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
 在后台运行时,查看其他设备列表

vnt-cli/src/config/mod.rs

@@ -40,6 +40,8 @@ pub struct FileConfig {
     pub device_name: Option<String>,
     pub packet_loss: Option<f64>,
     pub packet_delay: u32,
+    #[cfg(feature = "port_mapping")]
+    pub mapping: Vec<String>,
 }
 
 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))

vnt-cli/src/main.rs

@@ -73,6 +73,7 @@ fn main() {
     opts.optopt("", "packet-loss", "丢包率", "<packet-loss>");
     opts.optopt("", "packet-delay", "延迟", "<packet-delay>");
     opts.optmulti("", "dns", "dns", "<dns>");
+    opts.optmulti("", "mapping", "mapping", "<mapping>");
     opts.optopt("f", "", "配置文件", "<conf>");
     //"后台运行时,查看其他设备列表"
     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::<u32>("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 <name>           给设备一个名字,便于区分不同设备,默认使用系统版本");
     println!("  -d <id>             设备唯一标识符,不使用--ip参数时,服务端凭此参数分配虚拟ip,注意不能重复");
     println!("  -s <server>         注册和中继服务器地址,以'TXT:'开头表示解析TXT记录");
-    println!("  -e <stun-server>    stun服务器,用于探测NAT类型,可使用多个地址,如-e stun.qq.com -e stun1.l.google.com");
+    println!("  -e <stun-server>    stun服务器,用于探测NAT类型,可使用多个地址,如-e stun1.l.google.com -e stun2.l.google.com");
     println!("  -a                  使用tap模式,默认使用tun模式");
     println!("  -i <in-ip>          配置点对网(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 <host:port>   DNS服务器地址,可使用多个dns,不指定时使用系统解析");
+    #[cfg(feature = "port_mapping")]
+    println!("  --mapping <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!(

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;

vnt-jni/src/config.rs

@@ -21,6 +21,7 @@ pub fn new_config(env: &mut JNIEnv, config: JObject) -> Result<Config, Error> {
     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<Config, Error> {
         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);

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(),

vnt/src/core/mod.rs

@@ -43,6 +43,9 @@ pub struct Config {
     //控制丢包率
     pub packet_loss_rate: Option<f64>,
     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<f64>,
         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<String>,
     ) -> anyhow::Result<Self> {
         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,
         })
     }
 }

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;

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<String>) -> anyhow::Result<Vec<(bool, SocketAddr, String)>> {
+    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(())
+}

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(())
+}

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<Mutex<HashMap<SocketAddr, (Arc<UdpSocket>, Arc<AtomicCell<Instant>>)>>> =
+        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<Mutex<HashMap<SocketAddr, (Arc<UdpSocket>, Arc<AtomicCell<Instant>>)>>>,
+    udp_socket: &Arc<UdpSocket>,
+    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(())
+}