From a3e85a127017f02e18118e08790dcd3b1fcecc31 Mon Sep 17 00:00:00 2001 From: "sijie.sun" Date: Sat, 27 Apr 2024 19:03:07 +0800 Subject: [PATCH] tunnel support ipv6 --- easytier/src/connector/udp_hole_punch.rs | 2 +- easytier/src/tunnel/mod.rs | 46 +++++++++++++++-- easytier/src/tunnel/quic.rs | 54 ++++++++++++++++++-- easytier/src/tunnel/tcp.rs | 59 +++++++++++++++++----- easytier/src/tunnel/udp.rs | 64 ++++++++++++++++++++---- easytier/src/tunnel/wireguard.rs | 61 ++++++++++++++++++++-- 6 files changed, 250 insertions(+), 36 deletions(-) diff --git a/easytier/src/connector/udp_hole_punch.rs b/easytier/src/connector/udp_hole_punch.rs index 42e1f6f7..79eeec4d 100644 --- a/easytier/src/connector/udp_hole_punch.rs +++ b/easytier/src/connector/udp_hole_punch.rs @@ -404,7 +404,7 @@ impl UdpHolePunchConnector { let socket = UdpSocket::from_std(socket2_socket.into())?; Ok(connector - .try_connect_with_socket(socket) + .try_connect_with_socket(socket, remote_mapped_addr) .await .with_context(|| "UdpTunnelConnector failed to connect remote")?) } diff --git a/easytier/src/tunnel/mod.rs b/easytier/src/tunnel/mod.rs index 1fa782fb..bfbb3fc8 100644 --- a/easytier/src/tunnel/mod.rs +++ b/easytier/src/tunnel/mod.rs @@ -55,6 +55,9 @@ pub enum TunnelError { #[error("shutdown")] Shutdown, + #[error("no dns record found")] + NoDnsRecordFound(IpVersion), + #[error("tunnel error: {0}")] TunError(String), } @@ -80,6 +83,13 @@ pub trait TunnelConnCounter: 'static + Send + Sync + Debug { fn get(&self) -> u32; } +#[derive(Debug, Clone, Copy)] +pub enum IpVersion { + V4, + V6, + Both, +} + #[async_trait] #[auto_impl::auto_impl(Box)] pub trait TunnelListener: Send { @@ -104,6 +114,7 @@ pub trait TunnelConnector: Send { async fn connect(&mut self) -> Result, TunnelError>; fn remote_url(&self) -> url::Url; fn set_bind_addrs(&mut self, _addrs: Vec) {} + fn set_ip_version(&mut self, _ip_version: IpVersion) {} } pub fn build_url_from_socket_addr(addr: &String, scheme: &str) -> url::Url { @@ -135,11 +146,26 @@ impl std::fmt::Debug for dyn TunnelListener { } pub(crate) trait FromUrl { - fn from_url(url: url::Url) -> Result + fn from_url(url: url::Url, ip_version: IpVersion) -> Result where Self: Sized; } +pub(crate) fn check_scheme_and_get_socket_addr_ext( + url: &url::Url, + scheme: &str, + ip_version: IpVersion, +) -> Result +where + T: FromUrl, +{ + if url.scheme() != scheme { + return Err(TunnelError::InvalidProtocol(url.scheme().to_string())); + } + + Ok(T::from_url(url.clone(), ip_version)?) +} + pub(crate) fn check_scheme_and_get_socket_addr( url: &url::Url, scheme: &str, @@ -151,17 +177,27 @@ where return Err(TunnelError::InvalidProtocol(url.scheme().to_string())); } - Ok(T::from_url(url.clone())?) + Ok(T::from_url(url.clone(), IpVersion::Both)?) } impl FromUrl for SocketAddr { - fn from_url(url: url::Url) -> Result { - Ok(url.socket_addrs(|| None)?.pop().unwrap()) + fn from_url(url: url::Url, ip_version: IpVersion) -> Result { + let addrs = url.socket_addrs(|| None)?; + tracing::debug!(?addrs, ?ip_version, ?url, "convert url to socket addrs"); + let mut addrs = addrs + .into_iter() + .filter(|addr| match ip_version { + IpVersion::V4 => addr.is_ipv4(), + IpVersion::V6 => addr.is_ipv6(), + IpVersion::Both => true, + }) + .collect::>(); + addrs.pop().ok_or(TunnelError::NoDnsRecordFound(ip_version)) } } impl FromUrl for uuid::Uuid { - fn from_url(url: url::Url) -> Result { + fn from_url(url: url::Url, _ip_version: IpVersion) -> Result { let o = url.host_str().unwrap(); let o = uuid::Uuid::parse_str(o).map_err(|e| TunnelError::InvalidAddr(e.to_string()))?; Ok(o) diff --git a/easytier/src/tunnel/quic.rs b/easytier/src/tunnel/quic.rs index a06bbf82..49dd455b 100644 --- a/easytier/src/tunnel/quic.rs +++ b/easytier/src/tunnel/quic.rs @@ -6,13 +6,17 @@ use std::{error::Error, net::SocketAddr, sync::Arc}; use crate::{ rpc::TunnelInfo, - tunnel::common::{FramedReader, FramedWriter, TunnelWrapper}, + tunnel::{ + check_scheme_and_get_socket_addr_ext, + common::{FramedReader, FramedWriter, TunnelWrapper}, + }, }; use anyhow::Context; use quinn::{ClientConfig, Connection, Endpoint, ServerConfig}; use super::{ - check_scheme_and_get_socket_addr, Tunnel, TunnelConnector, TunnelError, TunnelListener, + check_scheme_and_get_socket_addr, IpVersion, Tunnel, TunnelConnector, TunnelError, + TunnelListener, }; /// Dummy certificate verifier that treats any certificate as valid. @@ -153,6 +157,7 @@ impl TunnelListener for QUICTunnelListener { pub struct QUICTunnelConnector { addr: url::Url, endpoint: Option, + ip_version: IpVersion, } impl QUICTunnelConnector { @@ -160,6 +165,7 @@ impl QUICTunnelConnector { QUICTunnelConnector { addr, endpoint: None, + ip_version: IpVersion::Both, } } } @@ -167,9 +173,18 @@ impl QUICTunnelConnector { #[async_trait::async_trait] impl TunnelConnector for QUICTunnelConnector { async fn connect(&mut self) -> Result, super::TunnelError> { - let addr = check_scheme_and_get_socket_addr::(&self.addr, "quic")?; + let addr = check_scheme_and_get_socket_addr_ext::( + &self.addr, + "quic", + self.ip_version, + )?; + let local_addr = if addr.is_ipv4() { + "0.0.0.0:0" + } else { + "[::]:0" + }; - let mut endpoint = Endpoint::client("127.0.0.1:0".parse().unwrap())?; + let mut endpoint = Endpoint::client(local_addr.parse().unwrap())?; endpoint.set_default_client_config(configure_client()); // connect to server @@ -202,11 +217,18 @@ impl TunnelConnector for QUICTunnelConnector { fn remote_url(&self) -> url::Url { self.addr.clone() } + + fn set_ip_version(&mut self, ip_version: IpVersion) { + self.ip_version = ip_version; + } } #[cfg(test)] mod tests { - use crate::tunnel::common::tests::{_tunnel_bench, _tunnel_pingpong}; + use crate::tunnel::{ + common::tests::{_tunnel_bench, _tunnel_pingpong}, + IpVersion, + }; use super::*; @@ -223,4 +245,26 @@ mod tests { let connector = QUICTunnelConnector::new("quic://127.0.0.1:21012".parse().unwrap()); _tunnel_bench(listener, connector).await } + + #[tokio::test] + async fn ipv6_pingpong() { + let listener = QUICTunnelListener::new("quic://[::1]:31015".parse().unwrap()); + let connector = QUICTunnelConnector::new("quic://[::1]:31015".parse().unwrap()); + _tunnel_pingpong(listener, connector).await + } + + #[tokio::test] + async fn ipv6_domain_pingpong() { + let listener = QUICTunnelListener::new("quic://[::1]:31016".parse().unwrap()); + let mut connector = + QUICTunnelConnector::new("quic://test.kkrainbow.top:31016".parse().unwrap()); + connector.set_ip_version(IpVersion::V6); + _tunnel_pingpong(listener, connector).await; + + let listener = QUICTunnelListener::new("quic://127.0.0.1:31016".parse().unwrap()); + let mut connector = + QUICTunnelConnector::new("quic://test.kkrainbow.top:31016".parse().unwrap()); + connector.set_ip_version(IpVersion::V4); + _tunnel_pingpong(listener, connector).await; + } } diff --git a/easytier/src/tunnel/tcp.rs b/easytier/src/tunnel/tcp.rs index 55f1224a..8dddbaa9 100644 --- a/easytier/src/tunnel/tcp.rs +++ b/easytier/src/tunnel/tcp.rs @@ -7,9 +7,9 @@ use tokio::net::{TcpListener, TcpSocket, TcpStream}; use crate::{rpc::TunnelInfo, tunnel::common::setup_sokcet2}; use super::{ - check_scheme_and_get_socket_addr, + check_scheme_and_get_socket_addr, check_scheme_and_get_socket_addr_ext, common::{wait_for_connect_futures, FramedReader, FramedWriter, TunnelWrapper}, - Tunnel, TunnelError, TunnelListener, + IpVersion, Tunnel, TunnelError, TunnelListener, }; const TCP_MTU_BYTES: usize = 64 * 1024; @@ -99,6 +99,7 @@ pub struct TcpTunnelConnector { addr: url::Url, bind_addrs: Vec, + ip_version: IpVersion, } impl TcpTunnelConnector { @@ -106,33 +107,38 @@ impl TcpTunnelConnector { TcpTunnelConnector { addr, bind_addrs: vec![], + ip_version: IpVersion::Both, } } - async fn connect_with_default_bind(&mut self) -> Result, super::TunnelError> { + async fn connect_with_default_bind( + &mut self, + addr: SocketAddr, + ) -> Result, super::TunnelError> { tracing::info!(addr = ?self.addr, "connect tcp start"); - let addr = check_scheme_and_get_socket_addr::(&self.addr, "tcp")?; let stream = TcpStream::connect(addr).await?; tracing::info!(addr = ?self.addr, "connect tcp succ"); return get_tunnel_with_tcp_stream(stream, self.addr.clone().into()); } - async fn connect_with_custom_bind(&mut self) -> Result, super::TunnelError> { + async fn connect_with_custom_bind( + &mut self, + addr: SocketAddr, + ) -> Result, super::TunnelError> { let futures = FuturesUnordered::new(); - let dst_addr = check_scheme_and_get_socket_addr::(&self.addr, "tcp")?; for bind_addr in self.bind_addrs.iter() { - tracing::info!(bind_addr = ?bind_addr, ?dst_addr, "bind addr"); + tracing::info!(bind_addr = ?bind_addr, ?addr, "bind addr"); let socket2_socket = socket2::Socket::new( - socket2::Domain::for_address(dst_addr), + socket2::Domain::for_address(addr), socket2::Type::STREAM, Some(socket2::Protocol::TCP), )?; setup_sokcet2(&socket2_socket, bind_addr)?; let socket = TcpSocket::from_std_stream(socket2_socket.into()); - futures.push(socket.connect(dst_addr.clone())); + futures.push(socket.connect(addr.clone())); } let ret = wait_for_connect_futures(futures).await; @@ -143,19 +149,26 @@ impl TcpTunnelConnector { #[async_trait] impl super::TunnelConnector for TcpTunnelConnector { async fn connect(&mut self) -> Result, super::TunnelError> { - if self.bind_addrs.is_empty() { - self.connect_with_default_bind().await + let addr = + check_scheme_and_get_socket_addr_ext::(&self.addr, "tcp", self.ip_version)?; + if self.bind_addrs.is_empty() || addr.is_ipv6() { + self.connect_with_default_bind(addr).await } else { - self.connect_with_custom_bind().await + self.connect_with_custom_bind(addr).await } } fn remote_url(&self) -> url::Url { self.addr.clone() } + fn set_bind_addrs(&mut self, addrs: Vec) { self.bind_addrs = addrs; } + + fn set_ip_version(&mut self, ip_version: IpVersion) { + self.ip_version = ip_version; + } } #[cfg(test)] @@ -197,4 +210,26 @@ mod tests { connector.set_bind_addrs(vec!["10.0.0.1:0".parse().unwrap()]); _tunnel_pingpong(listener, connector).await } + + #[tokio::test] + async fn ipv6_pingpong() { + let listener = TcpTunnelListener::new("tcp://[::1]:31015".parse().unwrap()); + let connector = TcpTunnelConnector::new("tcp://[::1]:31015".parse().unwrap()); + _tunnel_pingpong(listener, connector).await + } + + #[tokio::test] + async fn ipv6_domain_pingpong() { + let listener = TcpTunnelListener::new("tcp://[::1]:31015".parse().unwrap()); + let mut connector = + TcpTunnelConnector::new("tcp://test.kkrainbow.top:31015".parse().unwrap()); + connector.set_ip_version(IpVersion::V6); + _tunnel_pingpong(listener, connector).await; + + let listener = TcpTunnelListener::new("tcp://127.0.0.1:31015".parse().unwrap()); + let mut connector = + TcpTunnelConnector::new("tcp://test.kkrainbow.top:31015".parse().unwrap()); + connector.set_ip_version(IpVersion::V4); + _tunnel_pingpong(listener, connector).await; + } } diff --git a/easytier/src/tunnel/udp.rs b/easytier/src/tunnel/udp.rs index 00bbf535..b12ab41c 100644 --- a/easytier/src/tunnel/udp.rs +++ b/easytier/src/tunnel/udp.rs @@ -29,7 +29,7 @@ use super::{ common::{setup_sokcet2, setup_sokcet2_ext, wait_for_connect_futures}, packet_def::{UDPTunnelHeader, UDP_TUNNEL_HEADER_SIZE}, ring::{RingSink, RingStream}, - Tunnel, TunnelConnCounter, TunnelError, TunnelListener, TunnelUrl, + IpVersion, Tunnel, TunnelConnCounter, TunnelError, TunnelListener, TunnelUrl, }; pub const UDP_DATA_MTU: usize = 65000; @@ -441,6 +441,7 @@ impl TunnelListener for UdpTunnelListener { pub struct UdpTunnelConnector { addr: url::Url, bind_addrs: Vec, + ip_version: IpVersion, } impl UdpTunnelConnector { @@ -448,6 +449,7 @@ impl UdpTunnelConnector { Self { addr, bind_addrs: vec![], + ip_version: IpVersion::Both, } } @@ -605,8 +607,8 @@ impl UdpTunnelConnector { pub async fn try_connect_with_socket( &self, socket: UdpSocket, + addr: SocketAddr, ) -> Result, super::TunnelError> { - let addr = super::check_scheme_and_get_socket_addr::(&self.addr, "udp")?; log::warn!("udp connect: {:?}", self.addr); #[cfg(target_os = "windows")] @@ -633,12 +635,23 @@ impl UdpTunnelConnector { self.build_tunnel(socket, addr, conn_id).await } - async fn connect_with_default_bind(&mut self) -> Result, super::TunnelError> { - let socket = UdpSocket::bind("0.0.0.0:0").await?; - return self.try_connect_with_socket(socket).await; + async fn connect_with_default_bind( + &mut self, + addr: SocketAddr, + ) -> Result, super::TunnelError> { + let socket = if addr.is_ipv4() { + UdpSocket::bind("0.0.0.0:0").await? + } else { + UdpSocket::bind("[::]:0").await? + }; + + return self.try_connect_with_socket(socket, addr).await; } - async fn connect_with_custom_bind(&mut self) -> Result, super::TunnelError> { + async fn connect_with_custom_bind( + &mut self, + addr: SocketAddr, + ) -> Result, super::TunnelError> { let futures = FuturesUnordered::new(); for bind_addr in self.bind_addrs.iter() { @@ -649,7 +662,7 @@ impl UdpTunnelConnector { )?; setup_sokcet2(&socket2_socket, &bind_addr)?; let socket = UdpSocket::from_std(socket2_socket.into())?; - futures.push(self.try_connect_with_socket(socket)); + futures.push(self.try_connect_with_socket(socket, addr)); } wait_for_connect_futures(futures).await } @@ -658,10 +671,15 @@ impl UdpTunnelConnector { #[async_trait] impl super::TunnelConnector for UdpTunnelConnector { async fn connect(&mut self) -> Result, super::TunnelError> { - if self.bind_addrs.is_empty() { - self.connect_with_default_bind().await + let addr = super::check_scheme_and_get_socket_addr_ext::( + &self.addr, + "udp", + self.ip_version, + )?; + if self.bind_addrs.is_empty() || addr.is_ipv6() { + self.connect_with_default_bind(addr).await } else { - self.connect_with_custom_bind().await + self.connect_with_custom_bind(addr).await } } @@ -672,6 +690,10 @@ impl super::TunnelConnector for UdpTunnelConnector { fn set_bind_addrs(&mut self, addrs: Vec) { self.bind_addrs = addrs; } + + fn set_ip_version(&mut self, ip_version: IpVersion) { + self.ip_version = ip_version; + } } #[cfg(test)] @@ -840,4 +862,26 @@ mod tests { setup_sokcet2_ext(&socket2_socket, &addr, bind_dev.clone()).unwrap(); } } + + #[tokio::test] + async fn ipv6_pingpong() { + let listener = UdpTunnelListener::new("udp://[::1]:31015".parse().unwrap()); + let connector = UdpTunnelConnector::new("udp://[::1]:31015".parse().unwrap()); + _tunnel_pingpong(listener, connector).await + } + + #[tokio::test] + async fn ipv6_domain_pingpong() { + let listener = UdpTunnelListener::new("udp://[::1]:31016".parse().unwrap()); + let mut connector = + UdpTunnelConnector::new("udp://test.kkrainbow.top:31016".parse().unwrap()); + connector.set_ip_version(IpVersion::V6); + _tunnel_pingpong(listener, connector).await; + + let listener = UdpTunnelListener::new("udp://127.0.0.1:31016".parse().unwrap()); + let mut connector = + UdpTunnelConnector::new("udp://test.kkrainbow.top:31016".parse().unwrap()); + connector.set_ip_version(IpVersion::V4); + _tunnel_pingpong(listener, connector).await; + } } diff --git a/easytier/src/tunnel/wireguard.rs b/easytier/src/tunnel/wireguard.rs index f753bded..ffc0475e 100644 --- a/easytier/src/tunnel/wireguard.rs +++ b/easytier/src/tunnel/wireguard.rs @@ -34,7 +34,7 @@ use super::{ generate_digest_from_str, packet_def::{ZCPacketType, PEER_MANAGER_HEADER_SIZE}, ring::create_ring_tunnel_pair, - Tunnel, TunnelError, TunnelListener, TunnelUrl, ZCPacketSink, ZCPacketStream, + IpVersion, Tunnel, TunnelError, TunnelListener, TunnelUrl, ZCPacketSink, ZCPacketStream, }; const MAX_PACKET: usize = 65500; @@ -567,6 +567,7 @@ pub struct WgTunnelConnector { udp: Option>, bind_addrs: Vec, + ip_version: IpVersion, } impl Debug for WgTunnelConnector { @@ -585,6 +586,7 @@ impl WgTunnelConnector { config, udp: None, bind_addrs: vec![], + ip_version: IpVersion::Both, } } @@ -624,8 +626,8 @@ impl WgTunnelConnector { addr_url: url::Url, config: WgConfig, udp: UdpSocket, + addr: SocketAddr, ) -> Result, super::TunnelError> { - let addr = super::check_scheme_and_get_socket_addr::(&addr_url, "wg")?; tracing::warn!("wg connect: {:?}", addr); let local_addr = udp.local_addr().unwrap().to_string(); @@ -659,19 +661,42 @@ impl WgTunnelConnector { Ok(ret) } + + async fn connect_with_ipv6( + &mut self, + addr: SocketAddr, + ) -> Result, TunnelError> { + let socket2_socket = socket2::Socket::new( + socket2::Domain::for_address(addr), + socket2::Type::DGRAM, + Some(socket2::Protocol::UDP), + )?; + setup_sokcet2_ext(&socket2_socket, &"[::]:0".parse().unwrap(), None)?; + let socket = UdpSocket::from_std(socket2_socket.into())?; + Self::connect_with_socket(self.addr.clone(), self.config.clone(), socket, addr).await + } } #[async_trait] impl super::TunnelConnector for WgTunnelConnector { #[tracing::instrument] async fn connect(&mut self) -> Result, super::TunnelError> { + let addr = super::check_scheme_and_get_socket_addr_ext::( + &self.addr, + "wg", + self.ip_version, + )?; + + if addr.is_ipv6() { + return self.connect_with_ipv6(addr).await; + } + let bind_addrs = if self.bind_addrs.is_empty() { vec!["0.0.0.0:0".parse().unwrap()] } else { self.bind_addrs.clone() }; let futures = FuturesUnordered::new(); - for bind_addr in bind_addrs.into_iter() { let socket2_socket = socket2::Socket::new( socket2::Domain::for_address(bind_addr), @@ -685,6 +710,7 @@ impl super::TunnelConnector for WgTunnelConnector { self.addr.clone(), self.config.clone(), socket, + addr, )); } @@ -698,6 +724,10 @@ impl super::TunnelConnector for WgTunnelConnector { fn set_bind_addrs(&mut self, addrs: Vec) { self.bind_addrs = addrs; } + + fn set_ip_version(&mut self, ip_version: IpVersion) { + self.ip_version = ip_version; + } } #[cfg(test)] @@ -813,4 +843,29 @@ pub mod tests { assert_eq!(0, listener.wg_peer_map.len()); } + + #[tokio::test] + async fn ipv6_pingpong() { + let (server_cfg, client_cfg) = create_wg_config(); + let listener = WgTunnelListener::new("wg://[::1]:31015".parse().unwrap(), server_cfg); + let connector = WgTunnelConnector::new("wg://[::1]:31015".parse().unwrap(), client_cfg); + _tunnel_pingpong(listener, connector).await + } + + #[tokio::test] + async fn ipv6_domain_pingpong() { + let (server_cfg, client_cfg) = create_wg_config(); + let listener = WgTunnelListener::new("wg://[::1]:31016".parse().unwrap(), server_cfg); + let mut connector = + WgTunnelConnector::new("wg://test.kkrainbow.top:31016".parse().unwrap(), client_cfg); + connector.set_ip_version(IpVersion::V6); + _tunnel_pingpong(listener, connector).await; + + let (server_cfg, client_cfg) = create_wg_config(); + let listener = WgTunnelListener::new("wg://127.0.0.1:31016".parse().unwrap(), server_cfg); + let mut connector = + WgTunnelConnector::new("wg://test.kkrainbow.top:31016".parse().unwrap(), client_cfg); + connector.set_ip_version(IpVersion::V4); + _tunnel_pingpong(listener, connector).await; + } }