support use ipv6

easytier/src/connector/direct.rs

@@ -1,6 +1,6 @@
 // try connect peers directly, with either its public ip or lan ip
 
-use std::sync::Arc;
+use std::{net::SocketAddr, sync::Arc};
 
 use crate::{
     common::{error::Error, global_ctx::ArcGlobalCtx, PeerId},
@@ -10,6 +10,7 @@ use crate::{
 use crate::rpc::{peer::GetIpListResponse, PeerConnInfo};
 use tokio::{task::JoinSet, time::timeout};
 use tracing::Instrument;
+use url::Host;
 
 use super::create_connector_by_url;
 
@@ -230,17 +231,19 @@ impl DirectConnectorManager {
         dst_peer_id: PeerId,
         ip_list: GetIpListResponse,
     ) -> Result<(), Error> {
+        let enable_ipv6 = data.global_ctx.get_flags().enable_ipv6;
         let available_listeners = ip_list
             .listeners
             .iter()
             .filter_map(|l| if l.scheme() != "ring" { Some(l) } else { None })
-            .filter(|l| l.port().is_some())
+            .filter(|l| l.port().is_some() && l.host().is_some())
             .filter(|l| {
                 !data.dst_sceme_blacklist.contains(&DstSchemeBlackListItem(
                     dst_peer_id.clone(),
                     l.scheme().to_string(),
                 ))
             })
+            .filter(|l| enable_ipv6 || !matches!(l.host().unwrap().to_owned(), Host::Ipv6(_)))
             .collect::<Vec<_>>();
 
         let mut listener = available_listeners.get(0).ok_or(anyhow::anyhow!(
@@ -255,31 +258,58 @@ impl DirectConnectorManager {
             .unwrap_or(listener);
 
         let mut tasks = JoinSet::new();
-        ip_list.interface_ipv4s.iter().for_each(|ip| {
-            let addr = format!(
-                "{}://{}:{}",
-                listener.scheme(),
-                ip,
-                listener.port().unwrap_or(11010)
-            );
-            tasks.spawn(Self::try_connect_to_ip(
-                data.clone(),
-                dst_peer_id.clone(),
-                addr,
-            ));
-        });
 
-        let addr = format!(
-            "{}://{}:{}",
-            listener.scheme(),
-            ip_list.public_ipv4.clone(),
-            listener.port().unwrap_or(11010)
-        );
-        tasks.spawn(Self::try_connect_to_ip(
-            data.clone(),
-            dst_peer_id.clone(),
-            addr,
-        ));
+        let listener_host = listener.socket_addrs(|| None).unwrap().pop();
+        match listener_host {
+            Some(SocketAddr::V4(_)) => {
+                ip_list.interface_ipv4s.iter().for_each(|ip| {
+                    let mut addr = (*listener).clone();
+                    if addr.set_host(Some(ip.as_str())).is_ok() {
+                        tasks.spawn(Self::try_connect_to_ip(
+                            data.clone(),
+                            dst_peer_id.clone(),
+                            addr.to_string(),
+                        ));
+                    }
+                });
+
+                let mut addr = (*listener).clone();
+                if addr.set_host(Some(ip_list.public_ipv4.as_str())).is_ok() {
+                    tasks.spawn(Self::try_connect_to_ip(
+                        data.clone(),
+                        dst_peer_id.clone(),
+                        addr.to_string(),
+                    ));
+                }
+            }
+            Some(SocketAddr::V6(_)) => {
+                ip_list.interface_ipv6s.iter().for_each(|ip| {
+                    let mut addr = (*listener).clone();
+                    if addr.set_host(Some(format!("[{}]", ip).as_str())).is_ok() {
+                        tasks.spawn(Self::try_connect_to_ip(
+                            data.clone(),
+                            dst_peer_id.clone(),
+                            addr.to_string(),
+                        ));
+                    }
+                });
+
+                let mut addr = (*listener).clone();
+                if addr
+                    .set_host(Some(format!("[{}]", ip_list.public_ipv6).as_str()))
+                    .is_ok()
+                {
+                    tasks.spawn(Self::try_connect_to_ip(
+                        data.clone(),
+                        dst_peer_id.clone(),
+                        addr.to_string(),
+                    ));
+                }
+            }
+            p => {
+                tracing::error!(?p, ?listener, "failed to parse ip version from listener");
+            }
+        }
 
         let mut has_succ = false;
         while let Some(ret) = tasks.join_next().await {
@@ -351,7 +381,14 @@ mod tests {
 
     #[rstest::rstest]
     #[tokio::test]
-    async fn direct_connector_basic_test(#[values("tcp", "udp", "wg")] proto: &str) {
+    async fn direct_connector_basic_test(
+        #[values("tcp", "udp", "wg")] proto: &str,
+        #[values("true", "false")] ipv6: bool,
+    ) {
+        if ipv6 && proto != "udp" {
+            return;
+        }
+
         let p_a = create_mock_peer_manager().await;
         let p_b = create_mock_peer_manager().await;
         let p_c = create_mock_peer_manager().await;
@@ -366,12 +403,18 @@ mod tests {
         dm_a.run_as_client();
         dm_c.run_as_server();
 
-        let port = if proto == "wg" { 11040 } else { 11041 };
-        p_c.get_global_ctx()
-            .config
-            .set_listeners(vec![format!("{}://0.0.0.0:{}", proto, port)
-                .parse()
-                .unwrap()]);
+        if !ipv6 {
+            let port = if proto == "wg" { 11040 } else { 11041 };
+            p_c.get_global_ctx().config.set_listeners(vec![format!(
+                "{}://0.0.0.0:{}",
+                proto, port
+            )
+            .parse()
+            .unwrap()]);
+        }
+        let mut f = p_c.get_global_ctx().config.get_flags();
+        f.enable_ipv6 = ipv6;
+        p_c.get_global_ctx().config.set_flags(f);
         let mut lis_c = ListenerManager::new(p_c.get_global_ctx(), p_c.clone());
         lis_c.prepare_listeners().await.unwrap();
 

easytier/src/connector/manual.rs

@@ -1,5 +1,6 @@
 use std::{collections::BTreeSet, sync::Arc};
 
+use anyhow::Context;
 use dashmap::{DashMap, DashSet};
 use tokio::{
     sync::{broadcast::Receiver, mpsc, Mutex},
@@ -8,7 +9,10 @@ use tokio::{
 };
 
 use crate::{
-    common::PeerId, peers::peer_conn::PeerConnId, rpc as easytier_rpc, tunnel::TunnelConnector,
+    common::PeerId,
+    peers::peer_conn::PeerConnId,
+    rpc as easytier_rpc,
+    tunnel::{IpVersion, TunnelConnector},
 };
 
 use crate::{
@@ -254,66 +258,109 @@ impl ManualConnectorManager {
         &all_urls - &curr_alive
     }
 
+    async fn conn_reconnect_with_ip_version(
+        data: Arc<ConnectorManagerData>,
+        dead_url: String,
+        connector: MutexConnector,
+        ip_version: IpVersion,
+    ) -> Result<ReconnResult, Error> {
+        let ip_collector = data.global_ctx.get_ip_collector();
+        let net_ns = data.net_ns.clone();
+
+        connector.lock().await.set_ip_version(ip_version);
+
+        set_bind_addr_for_peer_connector(
+            connector.lock().await.as_mut(),
+            ip_version == IpVersion::V4,
+            &ip_collector,
+        )
+        .await;
+
+        data.global_ctx.issue_event(GlobalCtxEvent::Connecting(
+            connector.lock().await.remote_url().clone(),
+        ));
+
+        let _g = net_ns.guard();
+        log::info!("reconnect try connect... conn: {:?}", connector);
+        let tunnel = connector.lock().await.connect().await?;
+        log::info!("reconnect get tunnel succ: {:?}", tunnel);
+        assert_eq!(
+            dead_url,
+            tunnel.info().unwrap().remote_addr,
+            "info: {:?}",
+            tunnel.info()
+        );
+        let (peer_id, conn_id) = data.peer_manager.add_client_tunnel(tunnel).await?;
+        log::info!("reconnect succ: {} {} {}", peer_id, conn_id, dead_url);
+        Ok(ReconnResult {
+            dead_url,
+            peer_id,
+            conn_id,
+        })
+    }
+
     async fn conn_reconnect(
         data: Arc<ConnectorManagerData>,
         dead_url: String,
         connector: MutexConnector,
     ) -> Result<ReconnResult, Error> {
-        let connector = Arc::new(Mutex::new(Some(connector)));
-        let net_ns = data.net_ns.clone();
-
         log::info!("reconnect: {}", dead_url);
 
-        let connector_clone = connector.clone();
-        let data_clone = data.clone();
-        let url_clone = dead_url.clone();
-        let ip_collector = data.global_ctx.get_ip_collector();
-        let reconn_task = async move {
-            let mut locked = connector_clone.lock().await;
-            let conn = locked.as_mut().unwrap();
-            // TODO: should support set v6 here, use url in connector array
-            set_bind_addr_for_peer_connector(conn.lock().await.as_mut(), true, &ip_collector).await;
-
-            data_clone
-                .global_ctx
-                .issue_event(GlobalCtxEvent::Connecting(
-                    conn.lock().await.remote_url().clone(),
-                ));
-
-            let _g = net_ns.guard();
-            log::info!("reconnect try connect... conn: {:?}", conn);
-            let tunnel = conn.lock().await.connect().await?;
-            log::info!("reconnect get tunnel succ: {:?}", tunnel);
-            assert_eq!(
-                url_clone,
-                tunnel.info().unwrap().remote_addr,
-                "info: {:?}",
-                tunnel.info()
-            );
-            let (peer_id, conn_id) = data_clone.peer_manager.add_client_tunnel(tunnel).await?;
-            log::info!("reconnect succ: {} {} {}", peer_id, conn_id, url_clone);
-            Ok(ReconnResult {
-                dead_url: url_clone,
-                peer_id,
-                conn_id,
-            })
-        };
-
-        let ret = timeout(std::time::Duration::from_secs(1), reconn_task).await;
-        log::info!("reconnect: {} done, ret: {:?}", dead_url, ret);
-
-        if ret.is_err() || ret.as_ref().unwrap().is_err() {
-            data.global_ctx.issue_event(GlobalCtxEvent::ConnectError(
-                dead_url.clone(),
-                format!("{:?}", ret),
-            ));
+        let mut ip_versions = vec![];
+        let u = url::Url::parse(&dead_url)
+            .with_context(|| format!("failed to parse connector url {:?}", dead_url))?;
+        if u.scheme() == "ring" {
+            ip_versions.push(IpVersion::Both);
+        } else {
+            let addrs = u.socket_addrs(|| Some(1000))?;
+            let mut has_ipv4 = false;
+            let mut has_ipv6 = false;
+            for addr in addrs {
+                if addr.is_ipv4() {
+                    if !has_ipv4 {
+                        ip_versions.insert(0, IpVersion::V4);
+                    }
+                    has_ipv4 = true;
+                } else if addr.is_ipv6() {
+                    if !has_ipv6 {
+                        ip_versions.push(IpVersion::V6);
+                    }
+                    has_ipv6 = true;
+                }
+            }
         }
 
-        let conn = connector.lock().await.take().unwrap();
-        data.reconnecting.remove(&dead_url).unwrap();
-        data.connectors.insert(dead_url.clone(), conn);
+        let mut reconn_ret = Err(Error::AnyhowError(anyhow::anyhow!(
+            "cannot get ip from url"
+        )));
+        for ip_version in ip_versions {
+            let ret = timeout(
+                std::time::Duration::from_secs(1),
+                Self::conn_reconnect_with_ip_version(
+                    data.clone(),
+                    dead_url.clone(),
+                    connector.clone(),
+                    ip_version,
+                ),
+            )
+            .await;
+            log::info!("reconnect: {} done, ret: {:?}", dead_url, ret);
 
-        ret?
+            if ret.is_ok() && ret.as_ref().unwrap().is_ok() {
+                reconn_ret = ret.unwrap();
+                break;
+            } else {
+                if ret.is_err() {
+                    reconn_ret = Err(ret.unwrap_err().into());
+                } else if ret.as_ref().unwrap().is_err() {
+                    reconn_ret = Err(ret.unwrap().unwrap_err());
+                }
+            }
+        }
+        data.reconnecting.remove(&dead_url).unwrap();
+        data.connectors.insert(dead_url.clone(), connector);
+
+        reconn_ret
     }
 }