diff --git a/Cargo.toml b/Cargo.toml index 3e9bdaa9..aec15a0e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -83,6 +83,7 @@ socket2 = "0.3" cfg-if = "0.1" bloomfilter = "^1.0.2" spin = "0.5" +mio = "0.6" [target.'cfg(windows)'.dependencies] winapi = { version = "0.3", features = ["mswsock", "winsock2"] } diff --git a/src/relay/tcprelay/redir_local.rs b/src/relay/tcprelay/redir_local.rs index 126a759c..41df8af4 100644 --- a/src/relay/tcprelay/redir_local.rs +++ b/src/relay/tcprelay/redir_local.rs @@ -30,9 +30,10 @@ cfg_if! { use std::{ io::Error, mem, - net::{SocketAddrV4, SocketAddrV6}, os::unix::io::AsRawFd, }; + use crate::relay::utils::sockaddr_to_std; + let fd = s.as_raw_fd(); unsafe { let mut target_addr: libc::sockaddr_storage = mem::zeroed(); @@ -71,21 +72,9 @@ cfg_if! { return Err(Error::last_os_error()); } } + // Convert sockaddr_storage to SocketAddr - match target_addr.ss_family as libc::c_int { - libc::AF_INET => { - let addr: SocketAddrV4 = mem::transmute_copy(&target_addr); - Ok(SocketAddr::V4(addr)) - } - libc::AF_INET6 => { - let addr: SocketAddrV6 = mem::transmute_copy(&target_addr); - Ok(SocketAddr::V6(addr)) - } - _ => { - let err = Error::new(ErrorKind::InvalidData, "family must be either AF_INET or AF_INET6"); - return Err(err); - } - } + sockaddr_to_std(&target_addr) } } } else if #[cfg(any(target_os = "macos", target_os = "freebsd", target_os = "openbsd", target_os = "netbsd", target_os = "dragonfly"))] { @@ -96,7 +85,7 @@ cfg_if! { } } else { fn get_original_destination_addr(_: &mut TcpStream) -> io::Result { - unimplemented!("Transparent Proxy (redir) doesn't work on this platform"); + unimplemented!("TCP Transparent Proxy (redir) is not supported on this platform"); } } } diff --git a/src/relay/udprelay/client.rs b/src/relay/udprelay/client.rs index 55973e4a..21d10c93 100644 --- a/src/relay/udprelay/client.rs +++ b/src/relay/udprelay/client.rs @@ -19,7 +19,7 @@ use crate::{ use super::{ crypto_io::{decrypt_payload, encrypt_payload}, - utils::create_socket, + sys::create_socket, DEFAULT_TIMEOUT, MAXIMUM_UDP_PAYLOAD_SIZE, }; diff --git a/src/relay/udprelay/mod.rs b/src/relay/udprelay/mod.rs index f90d5855..06c8f73e 100644 --- a/src/relay/udprelay/mod.rs +++ b/src/relay/udprelay/mod.rs @@ -54,6 +54,8 @@ pub mod local; mod redir_local; pub mod server; mod socks5_local; +mod sys; +mod tproxy_socket; mod tunnel_local; mod utils; diff --git a/src/relay/udprelay/redir_local.rs b/src/relay/udprelay/redir_local.rs index 3bb3c84d..c5562e8b 100644 --- a/src/relay/udprelay/redir_local.rs +++ b/src/relay/udprelay/redir_local.rs @@ -1,8 +1,362 @@ -use std::io; +//! UDP relay local server -use crate::context::SharedContext; +use std::{ + io::{self, Cursor, Read}, + net::{IpAddr, Ipv4Addr, SocketAddr}, + sync::{ + atomic::{AtomicU64, Ordering}, + Arc, + }, + time::Duration, +}; + +use bytes::BytesMut; +use futures::{future, FutureExt}; +use log::{debug, error, info, trace}; +use lru_time_cache::{Entry, LruCache}; +use tokio::{ + self, + net::udp::{RecvHalf, SendHalf}, + sync::{mpsc, oneshot, Mutex}, + time, +}; + +use crate::{ + config::{ServerAddr, ServerConfig}, + context::{Context, SharedContext}, + relay::{ + loadbalancing::server::{LoadBalancer, PingBalancer, PingServer, PingServerType}, + socks5::Address, + utils::try_timeout, + }, +}; + +use super::{ + crypto_io::{decrypt_payload, encrypt_payload}, + sys::create_socket, + tproxy_socket::TProxyUdpSocket, + DEFAULT_TIMEOUT, + MAXIMUM_UDP_PAYLOAD_SIZE, +}; + +fn cache_key(src: &SocketAddr, dst: &SocketAddr) -> String { + format!("{}-{}", src, dst) +} + +// Drop the oneshot::Sender<()> will trigger local <- remote task to finish +struct UdpAssociationWatcher(oneshot::Sender<()>); + +// Represent a UDP association +#[derive(Clone)] +struct UdpAssociation { + // local -> remote Queue + // Drops tx, will close local -> remote task + tx: mpsc::Sender>, + + // local <- remote task life watcher + watcher: Arc, +} + +impl UdpAssociation { + /// Create an association with addr + async fn associate( + context: SharedContext, + svr_cfg: Arc, + src_addr: SocketAddr, + dst_addr: SocketAddr, + assoc_map: SharedAssocMap, + ) -> io::Result { + // Create a socket for receiving packets + let local_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0); + let remote_udp = create_socket(&local_addr).await?; + + let local_addr = remote_udp.local_addr().expect("Could not determine port bound to"); + debug!( + "Created UDP Association for {} from {} -> {}", + src_addr, local_addr, dst_addr + ); + + // Create a channel for sending packets to remote + // FIXME: Channel size 1024? + let (tx, mut rx) = mpsc::channel::>(1024); + + // Create a watcher for local <- remote task + let (watcher_tx, watcher_rx) = oneshot::channel::<()>(); + + let close_flag = Arc::new(UdpAssociationWatcher(watcher_tx)); + + // Splits socket into sender and receiver + let (mut receiver, mut sender) = remote_udp.split(); + + // Create a socket for sending packets back + let mut local_udp = TProxyUdpSocket::bind(&dst_addr)?; + + let timeout = context.config().udp_timeout.unwrap_or(DEFAULT_TIMEOUT); + + // local -> remote + let c_svr_cfg = svr_cfg.clone(); + let c_context = context.clone(); + tokio::spawn(async move { + let svr_cfg = c_svr_cfg.server_config(); + + while let Some(pkt) = rx.recv().await { + // pkt is already a raw packet, so just send it + if let Err(err) = + UdpAssociation::relay_l2r(&*c_context, src_addr, &mut sender, &pkt[..], timeout, svr_cfg).await + { + error!("failed to send packet {} -> {}, error: {}", src_addr, dst_addr, err); + + // FIXME: Ignore? Or how to deal with it? + } + } + + debug!("UDP REDIR {} -> {} finished", src_addr, dst_addr); + }); + + // local <- remote + tokio::spawn(async move { + let svr_cfg = svr_cfg.server_config(); + + let transfer_fut = async move { + loop { + // Read and send back to source + match UdpAssociation::relay_r2l(&*context, src_addr, &mut receiver, &mut local_udp, svr_cfg).await { + Ok(..) => {} + Err(err) => { + error!("failed to receive packet, {} <- {}, error: {}", src_addr, dst_addr, err); + + // FIXME: Don't break, or if you can find a way to drop the UdpAssociation + // break; + } + } + + let cache_key = cache_key(&src_addr, &dst_addr); + { + let mut amap = assoc_map.lock().await; + + // Check or update expire time + let _ = amap.get(&cache_key); + } + } + }; + + // Resolved only if watcher_rx resolved + let _ = future::select(transfer_fut.boxed(), watcher_rx.boxed()).await; + + debug!("UDP REDIR {} <- {} finished", src_addr, dst_addr); + }); + + Ok(UdpAssociation { + tx, + watcher: close_flag, + }) + } + + /// Relay packets from local to remote + async fn relay_l2r( + context: &Context, + src: SocketAddr, + remote_udp: &mut SendHalf, + payload: &[u8], + timeout: Duration, + svr_cfg: &ServerConfig, + ) -> io::Result<()> { + let addr = context.config().forward.as_ref().unwrap(); + + debug!("UDP REDIR {} -> {}, payload length {} bytes", src, addr, payload.len()); + + // CLIENT -> SERVER protocol: ADDRESS + PAYLOAD + let mut send_buf = Vec::new(); + addr.write_to_buf(&mut send_buf); + send_buf.extend_from_slice(payload); + + let mut encrypt_buf = BytesMut::new(); + encrypt_payload(context, svr_cfg.method(), svr_cfg.key(), &send_buf, &mut encrypt_buf)?; + + let send_len = match svr_cfg.addr() { + ServerAddr::SocketAddr(ref remote_addr) => { + try_timeout(remote_udp.send_to(&encrypt_buf[..], remote_addr), Some(timeout)).await? + } + ServerAddr::DomainName(ref dname, port) => lookup_then!(context, dname, *port, false, |addr| { + try_timeout(remote_udp.send_to(&encrypt_buf[..], &addr), Some(timeout)).await + }) + .map(|(_, l)| l)?, + }; + + assert_eq!(encrypt_buf.len(), send_len); + + Ok(()) + } + + /// Relay packets from remote to local + async fn relay_r2l( + context: &Context, + src_addr: SocketAddr, + remote_udp: &mut RecvHalf, + local_udp: &mut TProxyUdpSocket, + svr_cfg: &ServerConfig, + ) -> io::Result<()> { + // Waiting for response from server SERVER -> CLIENT + // Packet length is limited by MAXIMUM_UDP_PAYLOAD_SIZE, excess bytes will be discarded. + let mut recv_buf = [0u8; MAXIMUM_UDP_PAYLOAD_SIZE]; + + let (recv_n, remote_addr) = remote_udp.recv_from(&mut recv_buf).await?; + + let decrypt_buf = match decrypt_payload(context, svr_cfg.method(), svr_cfg.key(), &recv_buf[..recv_n])? { + None => { + error!("UDP packet too short, received length {}", recv_n); + let err = io::Error::new(io::ErrorKind::InvalidData, "packet too short"); + return Err(err); + } + Some(b) => b, + }; + // SERVER -> CLIENT protocol: ADDRESS + PAYLOAD + let mut cur = Cursor::new(decrypt_buf); + // FIXME: Address is ignored. Maybe useful in the future if we uses one common UdpSocket for communicate with remote server + let _ = Address::read_from(&mut cur).await?; + + let mut payload = Vec::new(); + cur.read_to_end(&mut payload)?; + + debug!( + "UDP REDIR {} <- {}, payload length {} bytes", + src_addr, + remote_addr, + payload.len() + ); + + // Send back to src_addr + local_udp.send_to(&payload, &src_addr).await.map(|_| ()) + } + + // Send packet to remote + // + // Return `Err` if receiver have been closed + async fn send(&mut self, pkt: Vec) { + if let Err(..) = self.tx.send(pkt).await { + // SHOULDn't HAPPEN + unreachable!("UDP Association local -> remote Queue closed unexpectly"); + } + } +} + +struct ServerScore { + svr_cfg: ServerConfig, + score: AtomicU64, +} + +impl ServerScore { + fn new(config: &ServerConfig) -> Arc { + let s = ServerScore { + svr_cfg: config.clone(), + score: AtomicU64::new(0), + }; + Arc::new(s) + } +} + +impl PingServer for ServerScore { + fn server_config(&self) -> &ServerConfig { + &self.svr_cfg + } + + fn score(&self) -> u64 { + self.score.load(Ordering::Acquire) + } + + fn set_score(&self, score: u64) { + self.score.store(score, Ordering::Release); + } +} + +type AssocMap = LruCache; +type SharedAssocMap = Arc>; /// Starts a UDP local server -pub async fn run(_: SharedContext) -> io::Result<()> { - unimplemented!("UDP Transparent Proxy (redir) isn't implemented yet"); +pub async fn run(context: SharedContext) -> io::Result<()> { + if let Err(err) = super::sys::check_support_tproxy() { + panic!("{}", err); + } + + let local_addr = context.config().local.as_ref().expect("Missing local config"); + let bind_addr = local_addr.bind_addr(&*context).await?; + + // let l = create_socket(&bind_addr).await?; + let mut l = TProxyUdpSocket::bind(&bind_addr)?; + let local_addr = l.local_addr().expect("Could not determine port bound to"); + + let servers = context.config().server.iter().map(ServerScore::new).collect(); + let mut balancer = PingBalancer::new(context.clone(), servers, PingServerType::Udp).await; + + info!("ShadowSocks UDP Redir listening on {}", local_addr); + + // NOTE: Associations are only eliminated by expire time + // So it may exhaust all available file descriptors + let timeout = context.config().udp_timeout.unwrap_or(DEFAULT_TIMEOUT); + let assoc_map: SharedAssocMap = Arc::new(Mutex::new(LruCache::with_expiry_duration(timeout))); + + let mut pkt_buf = [0u8; MAXIMUM_UDP_PAYLOAD_SIZE]; + + loop { + let (recv_len, src, dst) = match time::timeout(timeout, l.recv_from(&mut pkt_buf)).await { + Ok(r) => r?, + Err(..) => { + // Cleanup expired association + // Do not consume this iterator, it will updates expire time of items that traversed + let mut assoc_map = assoc_map.lock().await; + let _ = assoc_map.iter(); + continue; + } + }; + + // Packet length is limited by MAXIMUM_UDP_PAYLOAD_SIZE, excess bytes will be discarded. + // Copy bytes, because udp_associate runs in another tokio Task + let pkt = &pkt_buf[..recv_len]; + + trace!( + "received UDP packet from {}, destination {}, length {} bytes", + src, + dst, + recv_len + ); + + if recv_len == 0 { + // For windows, it will generate a ICMP Port Unreachable Message + // https://docs.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-recvfrom + // Which will result in recv_from return 0. + // + // It cannot be solved here, because `WSAGetLastError` is already set. + // + // See `relay::udprelay::utils::create_socket` for more detail. + continue; + } + + // Check or (re)create an association + let mut assoc = { + // Locks the whole association map + let mut ref_assoc_map = assoc_map.lock().await; + + // Get or create an association + let assoc = match ref_assoc_map.entry(cache_key(&src, &dst)) { + Entry::Occupied(oc) => oc.into_mut(), + Entry::Vacant(vc) => { + // Pick a server + let svr_cfg = balancer.pick_server(); + + vc.insert( + UdpAssociation::associate(context.clone(), svr_cfg.clone(), src, dst, assoc_map.clone()) + .await + .expect("Failed to create udp association"), + ) + } + }; + + // Clone the handle and release the lock. + // Make sure we keep the critical section small + assoc.clone() + }; + + // Send to local -> remote task + assoc.send(pkt.to_vec()).await; + } } diff --git a/src/relay/udprelay/server.rs b/src/relay/udprelay/server.rs index 1b445854..bcc9376a 100644 --- a/src/relay/udprelay/server.rs +++ b/src/relay/udprelay/server.rs @@ -26,7 +26,7 @@ use crate::{ use super::{ crypto_io::{decrypt_payload, encrypt_payload}, - utils::create_socket, + sys::create_socket, DEFAULT_TIMEOUT, MAXIMUM_UDP_PAYLOAD_SIZE, }; diff --git a/src/relay/udprelay/socks5_local.rs b/src/relay/udprelay/socks5_local.rs index eb58c6e6..3b6f5310 100644 --- a/src/relay/udprelay/socks5_local.rs +++ b/src/relay/udprelay/socks5_local.rs @@ -33,7 +33,7 @@ use crate::{ use super::{ crypto_io::{decrypt_payload, encrypt_payload}, - utils::create_socket, + sys::create_socket, DEFAULT_TIMEOUT, MAXIMUM_UDP_PAYLOAD_SIZE, }; diff --git a/src/relay/udprelay/sys/mod.rs b/src/relay/udprelay/sys/mod.rs new file mode 100644 index 00000000..3d4584fd --- /dev/null +++ b/src/relay/udprelay/sys/mod.rs @@ -0,0 +1,104 @@ +use std::{ + io::{self, ErrorKind}, + net::SocketAddr, + task::{Context, Poll}, +}; + +use cfg_if::cfg_if; +use futures::{future::poll_fn, ready}; +use socket2::{Domain, Protocol, SockAddr, Socket, Type}; +use tokio::io::PollEvented; + +cfg_if! { + if #[cfg(unix)] { + mod unix; + pub use self::unix::*; + } else if #[cfg(windows)] { + mod windows; + pub use self::windows::*; + } else { + compile_error!("UDP Relay is not supported in current platform"); + } +} + +/// A socket interface for transparent proxy +/// +/// It has basically the same APIs like `tokio::net::UdpSocket`, +/// but `recv_from` will return destination address of UDP packet +pub struct TProxyUdpSocket { + io: PollEvented, +} + +impl TProxyUdpSocket { + /// Create a new UDP socket binded to `addr` + /// + /// This will allow binding to `addr` that is not in local host + pub fn bind(addr: &SocketAddr) -> io::Result { + // Check if current plaform supports TPROXY (UDP) + // This is a runtime error. + check_support_tproxy()?; + + let domain = match *addr { + SocketAddr::V4(..) => Domain::ipv4(), + SocketAddr::V6(..) => Domain::ipv6(), + }; + let socket = Socket::new(domain, Type::dgram(), Some(Protocol::udp()))?; + set_socket_before_bind(addr, &socket)?; + + socket.set_nonblocking(true)?; + socket.set_reuse_address(true)?; + + socket.bind(&SockAddr::from(*addr))?; + + let msock = mio::net::UdpSocket::from_socket(socket.into_udp_socket())?; + let io = PollEvented::new(msock)?; + Ok(TProxyUdpSocket { io }) + } + + /// Send data to the socket to the given target address + pub async fn send_to(&mut self, buf: &[u8], target: &SocketAddr) -> io::Result { + poll_fn(|cx| self.poll_send_to(cx, buf, target)).await + } + + fn poll_send_to(&self, cx: &mut Context<'_>, buf: &[u8], target: &SocketAddr) -> Poll> { + ready!(self.io.poll_write_ready(cx))?; + + match self.io.get_ref().send_to(buf, target) { + Err(ref e) if e.kind() == ErrorKind::WouldBlock => { + self.io.clear_write_ready(cx)?; + Poll::Pending + } + x => Poll::Ready(x), + } + } + + /// Returns the local address that this socket is bound to. + pub fn local_addr(&self) -> io::Result { + self.io.get_ref().local_addr() + } + + /// Receive a single datagram from the socket. + /// + /// On success, the future resolves to the number of bytes read and the origin, target address + /// + /// `(bytes read, origin address, target address)` + pub async fn recv_from(&mut self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr, SocketAddr)> { + poll_fn(|cx| self.poll_recv_from(cx, buf)).await + } + + fn poll_recv_from( + &self, + cx: &mut Context<'_>, + buf: &mut [u8], + ) -> Poll> { + ready!(self.io.poll_read_ready(cx, mio::Ready::readable()))?; + + match recv_from_with_destination(self.io.get_ref(), buf) { + Err(ref e) if e.kind() == ErrorKind::WouldBlock => { + self.io.clear_read_ready(cx, mio::Ready::readable())?; + Poll::Pending + } + x => Poll::Ready(x), + } + } +} diff --git a/src/relay/udprelay/sys/unix/freebsd.rs b/src/relay/udprelay/sys/unix/freebsd.rs new file mode 100644 index 00000000..42c91164 --- /dev/null +++ b/src/relay/udprelay/sys/unix/freebsd.rs @@ -0,0 +1,145 @@ +use std::{ + io::{self, Error, ErrorKind}, + mem, + net::SocketAddr, + os::unix::io::AsRawFd, + ptr, +}; + +use mio::net::UdpSocket; +use socket2::Socket; + +use crate::relay::utils::sockaddr_to_std; + +pub fn check_support_tproxy() -> io::Result<()> { + Ok(()) +} + +fn set_bindany(level: libc::c_int, socket: &Socket) -> io::Result<()> { + let fd = socket.as_raw_fd(); + + let enable: libc::c_int = 1; + + // https://www.freebsd.org/cgi/man.cgi?query=ip&sektion=4&manpath=FreeBSD+9.0-RELEASE + let ret = libc::setsockopt( + fd, + level, + libc::IP_BINDANY, + &enable as *const _ as *const _, + mem::size_of_val(&enable) as libc::socklen_t, + ); + if ret != 0 { + return Err(Error::last_os_error()); + } + + Ok(()) +} + +pub fn set_socket_before_bind(addr: &SocketAddr, socket: &Socket) -> io::Result<()> { + let fd = socket.as_raw_fd(); + + let enable: libc::c_int = 1; + + unsafe { + // https://www.freebsd.org/cgi/man.cgi?query=ip&sektion=4&manpath=FreeBSD+9.0-RELEASE + let (level, opt) = match *addr { + SocketAddr::V4(..) => (libc::IPPROTO_IP, libc::IP_ORIGDSTADDR), + SocketAddr::V6(..) => (libc::IPPROTO_IPV6, libc::IPV6_ORIGDSTADDR), + }; + + // 1. BINDANY + set_bindany(level, socket)?; + + // 2. set ORIGDSTADDR for retrieving original destination address + let ret = libc::setsockopt( + fd, + level, + opt, + &enable as *const _ as *const _, + mem::size_of_val(&enable) as libc::socklen_t, + ); + if ret != 0 { + return Err(Error::last_os_error()); + } + } + + Ok(()) +} + +fn get_destination_addr(msg: &libc::msghdr) -> Option { + // https://www.freebsd.org/cgi/man.cgi?ip(4) + // + // Called `recvmsg` with `IP_ORIGDSTADDR` set + + unsafe { + let mut cmsg: *mut libc::cmsghdr = libc::CMSG_FIRSTHDR(msg); + while !cmsg.is_null() { + let rcmsg = &*cmsg; + match (rcmsg.cmsg_level, rcmsg.cmsg_type) { + (libc::IPPROTO_IP, libc::IP_ORIGDSTADDR) => { + let mut dst_addr: libc::sockaddr_storage = mem::zeroed(); + + ptr::copy( + libc::CMSG_DATA(cmsg), + &mut dst_addr as *mut _ as *mut _, + mem::size_of::(), + ); + + return Some(dst_addr); + } + (libc::IPPROTO_IPV6, libc::IPV6_ORIGDSTADDR) => { + let mut dst_addr: libc::sockaddr_storage = mem::zeroed(); + + ptr::copy( + libc::CMSG_DATA(cmsg), + &mut dst_addr as *mut _ as *mut _, + mem::size_of::(), + ); + + return Some(dst_addr); + } + _ => {} + } + cmsg = libc::CMSG_NXTHDR(msg, cmsg); + } + } + + None +} + +pub fn recv_from_with_destination(socket: &UdpSocket, buf: &mut [u8]) -> io::Result<(usize, SocketAddr, SocketAddr)> { + unsafe { + let mut control_buf = [0u8; 64]; + let mut src_addr: libc::sockaddr_storage = mem::zeroed(); + + let mut msg: libc::msghdr = mem::zeroed(); + msg.msg_name = &mut src_addr as *mut _ as *mut _; + msg.msg_namelen = mem::size_of_val(&src_addr) as libc::socklen_t; + + let mut iov = libc::iovec { + iov_base: buf.as_mut_ptr() as *mut _, + iov_len: buf.len() as libc::size_t, + }; + msg.msg_iov = &mut iov; + msg.msg_iovlen = 1; + + msg.msg_control = control_buf.as_mut_ptr() as *mut _; + msg.msg_controllen = control_buf.len() as libc::socklen_t; + + let fd = socket.as_raw_fd(); + let ret = libc::recvmsg(fd, &mut msg, 0); + if ret < 0 { + return Err(Error::last_os_error()); + } + + let dst_addr = match get_destination_addr(&msg) { + None => { + let err = Error::new(ErrorKind::InvalidData, "missing destination address in msghdr"); + return Err(err); + } + Some(d) => d, + }; + + Ok((ret as usize, sockaddr_to_std(&src_addr)?, sockaddr_to_std(&dst_addr)?)) + } +} diff --git a/src/relay/udprelay/sys/unix/linux.rs b/src/relay/udprelay/sys/unix/linux.rs new file mode 100644 index 00000000..217a738c --- /dev/null +++ b/src/relay/udprelay/sys/unix/linux.rs @@ -0,0 +1,132 @@ +use std::{ + io::{self, Error, ErrorKind}, + mem, + net::SocketAddr, + os::unix::io::AsRawFd, + ptr, +}; + +use mio::net::UdpSocket; +use socket2::Socket; + +use crate::relay::utils::sockaddr_to_std; + +pub fn check_support_tproxy() -> io::Result<()> { + Ok(()) +} + +pub fn set_socket_before_bind(addr: &SocketAddr, socket: &Socket) -> io::Result<()> { + let fd = socket.as_raw_fd(); + + let enable: libc::c_int = 1; + unsafe { + // 1. Set IP_TRANSPARENT to allow binding to non-local addresses + let ret = libc::setsockopt( + fd, + libc::SOL_IP, + libc::IP_TRANSPARENT, + &enable as *const _ as *const _, + mem::size_of_val(&enable) as libc::socklen_t, + ); + if ret != 0 { + return Err(Error::last_os_error()); + } + + // 2. Set IP_RECVORIGDSTADDR, IPV6_RECVORIGDSTADDR + let ret = match *addr { + SocketAddr::V4(..) => libc::setsockopt( + fd, + libc::SOL_IP, + libc::IP_RECVORIGDSTADDR, + &enable as *const _ as *const _, + mem::size_of_val(&enable) as libc::socklen_t, + ), + SocketAddr::V6(..) => libc::setsockopt( + fd, + libc::SOL_IPV6, + libc::IPV6_RECVORIGDSTADDR, + &enable as *const _ as *const _, + mem::size_of_val(&enable) as libc::socklen_t, + ), + }; + if ret != 0 { + return Err(Error::last_os_error()); + } + } + + Ok(()) +} + +fn get_destination_addr(msg: &libc::msghdr) -> Option { + unsafe { + let mut cmsg: *mut libc::cmsghdr = libc::CMSG_FIRSTHDR(msg); + while !cmsg.is_null() { + let rcmsg = &*cmsg; + match (rcmsg.cmsg_level, rcmsg.cmsg_type) { + (libc::SOL_IP, libc::IP_RECVORIGDSTADDR) => { + let mut dst_addr: libc::sockaddr_storage = mem::zeroed(); + + ptr::copy( + libc::CMSG_DATA(cmsg), + &mut dst_addr as *mut _ as *mut _, + mem::size_of::(), + ); + + return Some(dst_addr); + } + (libc::SOL_IPV6, libc::IPV6_RECVORIGDSTADDR) => { + let mut dst_addr: libc::sockaddr_storage = mem::zeroed(); + + ptr::copy( + libc::CMSG_DATA(cmsg), + &mut dst_addr as *mut _ as *mut _, + mem::size_of::(), + ); + + return Some(dst_addr); + } + _ => {} + } + cmsg = libc::CMSG_NXTHDR(msg, cmsg); + } + } + + None +} + +pub fn recv_from_with_destination(socket: &UdpSocket, buf: &mut [u8]) -> io::Result<(usize, SocketAddr, SocketAddr)> { + unsafe { + let mut control_buf = [0u8; 64]; + let mut src_addr: libc::sockaddr_storage = mem::zeroed(); + + let mut msg: libc::msghdr = mem::zeroed(); + msg.msg_name = &mut src_addr as *mut _ as *mut _; + msg.msg_namelen = mem::size_of_val(&src_addr) as libc::socklen_t; + + let mut iov = libc::iovec { + iov_base: buf.as_mut_ptr() as *mut _, + iov_len: buf.len() as libc::size_t, + }; + msg.msg_iov = &mut iov; + msg.msg_iovlen = 1; + + msg.msg_control = control_buf.as_mut_ptr() as *mut _; + msg.msg_controllen = control_buf.len() as libc::socklen_t; + + let fd = socket.as_raw_fd(); + let ret = libc::recvmsg(fd, &mut msg, 0); + if ret < 0 { + return Err(Error::last_os_error()); + } + + let dst_addr = match get_destination_addr(&msg) { + None => { + let err = Error::new(ErrorKind::InvalidData, "missing destination address in msghdr"); + return Err(err); + } + Some(d) => d, + }; + + Ok((ret as usize, sockaddr_to_std(&src_addr)?, sockaddr_to_std(&dst_addr)?)) + } +} diff --git a/src/relay/udprelay/sys/unix/macos.rs b/src/relay/udprelay/sys/unix/macos.rs new file mode 100644 index 00000000..c51b82b8 --- /dev/null +++ b/src/relay/udprelay/sys/unix/macos.rs @@ -0,0 +1,25 @@ +use std::{ + io::{self, Error, ErrorKind}, + net::SocketAddr, +}; + +use mio::net::UdpSocket; +use socket2::Socket; + +pub fn check_support_tproxy() -> io::Result<()> { + // FIXME: I can't find any reference about how to set *_BINDANY option on Mac OS X + // + // It may result in `bind()` returns `EADDRNOTAVAIL` + // https://developer.apple.com/library/archive/documentation/System/Conceptual/ManPages_iPhoneOS/man2/bind.2.html#//apple_ref/doc/man/2/bind + + let err = Error::new(ErrorKind::Other, "Mac OS X doesn't support UDP transparent proxy"); + Err(err) +} + +pub fn set_socket_before_bind(_addr: &SocketAddr, _socket: &Socket) -> io::Result<()> { + unimplemented!("Mac OS X doesn't support UDP transparent proxy"); +} + +pub fn recv_from_with_destination(_socket: &UdpSocket, _buf: &mut [u8]) -> io::Result<(usize, SocketAddr, SocketAddr)> { + unimplemented!("Mac OS X doesn't support UDP transparent proxy"); +} diff --git a/src/relay/udprelay/sys/unix/mod.rs b/src/relay/udprelay/sys/unix/mod.rs new file mode 100644 index 00000000..de6f11fb --- /dev/null +++ b/src/relay/udprelay/sys/unix/mod.rs @@ -0,0 +1,26 @@ +use std::{io, net::SocketAddr}; + +use cfg_if::cfg_if; +use tokio::net::UdpSocket; + +cfg_if! { + if #[cfg(any(target_os = "linux", target_os = "android"))] { + mod linux; + pub use self::linux::*; + } else if #[cfg(target_os = "macos")] { + mod macos; + pub use self::macos::*; + } else if #[cfg(target_os = "freebsd")] { + mod freebsd; + pub use self::freebsd::*; + } else { + mod not_supported; + pub use self::not_supported::*; + } +} + +/// Create a `UdpSocket` binded to `addr` +#[inline(always)] +pub async fn create_socket(addr: &SocketAddr) -> io::Result { + UdpSocket::bind(addr).await +} diff --git a/src/relay/udprelay/sys/unix/not_supported.rs b/src/relay/udprelay/sys/unix/not_supported.rs new file mode 100644 index 00000000..b2448a67 --- /dev/null +++ b/src/relay/udprelay/sys/unix/not_supported.rs @@ -0,0 +1,23 @@ +use std::{ + io::{self, ErrorKind}, + net::SocketAddr, +}; + +use mio::net::UdpSocket; +use socket2::Socket; + +pub fn check_support_tproxy() -> io::Result<()> { + let err = Error::new( + ErrorKind::Other, + "Current Platform doesn't support UDP transparent proxy", + ); + Err(err) +} + +pub fn set_socket_before_bind(_addr: &SocketAddr, _socket: &Socket) -> io::Result<()> { + unimplemented!("Current Platform doesn't support UDP transparent proxy"); +} + +pub fn recv_from_with_destination(_socket: &UdpSocket, _buf: &mut [u8]) -> io::Result<(usize, SocketAddr, SocketAddr)> { + unimplemented!("Current Platform doesn't support UDP transparent proxy"); +} diff --git a/src/relay/udprelay/sys/windows/mod.rs b/src/relay/udprelay/sys/windows/mod.rs new file mode 100644 index 00000000..dc830468 --- /dev/null +++ b/src/relay/udprelay/sys/windows/mod.rs @@ -0,0 +1,82 @@ +use std::{ + io, + io::{self, Error, ErrorKind}, + mem, + net::SocketAddr, + os::windows::io::AsRawSocket, + ptr, +}; + +use mio::net::UdpSocket; +use socket2::Socket; +use tokio::net::UdpSocket; +use winapi::{ + shared::minwindef::{BOOL, DWORD, FALSE, LPDWORD, LPVOID}, + um::{ + mswsock::SIO_UDP_CONNRESET, + winsock2::{WSAGetLastError, WSAIoctl, SOCKET, SOCKET_ERROR}, + }, +}; + +/// Create a `UdpSocket` binded to `addr` +/// +/// It also disables `WSAECONNRESET` for UDP socket +pub async fn create_socket(addr: &SocketAddr) -> io::Result { + // FIXME: Temporary solution. Should be replaced by tokio's UdpSocket::as_raw_socket + // https://github.com/tokio-rs/tokio/issues/2017 + + let socket = ::std::net::UdpSocket::bind(addr)?; + let handle = socket.as_raw_socket() as SOCKET; + + unsafe { + // Ignoring UdpSocket's WSAECONNRESET error + // https://github.com/shadowsocks/shadowsocks-rust/issues/179 + // https://stackoverflow.com/questions/30749423/is-winsock-error-10054-wsaeconnreset-normal-with-udp-to-from-localhost + // + // This is because `UdpSocket::recv_from` may return WSAECONNRESET + // if you called `UdpSocket::send_to` a destination that is not existed (may be closed). + // + // It is not an error. Could be ignored completely. + // We have to ignore it here because it will crash the server. + + let mut bytes_returned: DWORD = 0; + let mut enable: BOOL = FALSE; + + let ret = WSAIoctl( + handle, + SIO_UDP_CONNRESET, + &mut enable as *mut _ as LPVOID, + mem::size_of_val(&enable) as DWORD, + ptr::null_mut(), + 0, + &mut bytes_returned as *mut _ as LPDWORD, + ptr::null_mut(), + None, + ); + + if ret == SOCKET_ERROR { + use std::io::Error; + + // Error occurs + let err_code = WSAGetLastError(); + return Err(Error::from_raw_os_error(err_code)); + } + } + + UdpSocket::from_std(socket) +} + +pub fn check_support_tproxy() -> io::Result<()> { + // Windows seems to support transparent proxy, but I haven't found any useful document about it + + let err = Error::new(ErrorKind::Other, "Windows doesn't support UDP transparent proxy"); + Err(err) +} + +pub fn set_socket_before_bind(_addr: &SocketAddr, _socket: &Socket) -> io::Result<()> { + unimplemented!("Windows doesn't support UDP transparent proxy"); +} + +pub fn recv_from_with_destination(_socket: &UdpSocket, _buf: &mut [u8]) -> io::Result<(usize, SocketAddr, SocketAddr)> { + unimplemented!("Windows doesn't support UDP transparent proxy"); +} diff --git a/src/relay/udprelay/tproxy_socket.rs b/src/relay/udprelay/tproxy_socket.rs new file mode 100644 index 00000000..8fe13bf1 --- /dev/null +++ b/src/relay/udprelay/tproxy_socket.rs @@ -0,0 +1,3 @@ +//! Socket for supporting TPROXY + +pub use super::sys::TProxyUdpSocket; diff --git a/src/relay/udprelay/tunnel_local.rs b/src/relay/udprelay/tunnel_local.rs index e7209808..a0ad0975 100644 --- a/src/relay/udprelay/tunnel_local.rs +++ b/src/relay/udprelay/tunnel_local.rs @@ -33,7 +33,7 @@ use crate::{ use super::{ crypto_io::{decrypt_payload, encrypt_payload}, - utils::create_socket, + sys::create_socket, DEFAULT_TIMEOUT, MAXIMUM_UDP_PAYLOAD_SIZE, }; diff --git a/src/relay/udprelay/utils.rs b/src/relay/udprelay/utils.rs index dc987c61..de1df2bd 100644 --- a/src/relay/udprelay/utils.rs +++ b/src/relay/udprelay/utils.rs @@ -1,66 +1 @@ //! Utilities for UDP relay - -use std::{io, net::SocketAddr}; - -use tokio::net::UdpSocket; - -#[cfg(not(windows))] -#[inline(always)] -pub async fn create_socket(addr: &SocketAddr) -> io::Result { - UdpSocket::bind(addr).await -} - -#[cfg(windows)] -pub async fn create_socket(addr: &SocketAddr) -> io::Result { - use std::{mem, os::windows::io::AsRawSocket, ptr}; - use winapi::{ - shared::minwindef::{BOOL, DWORD, FALSE, LPDWORD, LPVOID}, - um::{ - mswsock::SIO_UDP_CONNRESET, - winsock2::{WSAGetLastError, WSAIoctl, SOCKET, SOCKET_ERROR}, - }, - }; - - // FIXME: Temporary solution. Should be replaced by tokio's UdpSocket::as_raw_socket - // https://github.com/tokio-rs/tokio/issues/2017 - - let socket = ::std::net::UdpSocket::bind(addr)?; - let handle = socket.as_raw_socket() as SOCKET; - - unsafe { - // Ignoring UdpSocket's WSAECONNRESET error - // https://github.com/shadowsocks/shadowsocks-rust/issues/179 - // https://stackoverflow.com/questions/30749423/is-winsock-error-10054-wsaeconnreset-normal-with-udp-to-from-localhost - // - // This is because `UdpSocket::recv_from` may return WSAECONNRESET - // if you called `UdpSocket::send_to` a destination that is not existed (may be closed). - // - // It is not an error. Could be ignored completely. - // We have to ignore it here because it will crash the server. - - let mut bytes_returned: DWORD = 0; - let mut enable: BOOL = FALSE; - - let ret = WSAIoctl( - handle, - SIO_UDP_CONNRESET, - &mut enable as *mut _ as LPVOID, - mem::size_of_val(&enable) as DWORD, - ptr::null_mut(), - 0, - &mut bytes_returned as *mut _ as LPDWORD, - ptr::null_mut(), - None, - ); - - if ret == SOCKET_ERROR { - use std::io::Error; - - // Error occurs - let err_code = WSAGetLastError(); - return Err(Error::from_raw_os_error(err_code)); - } - } - - UdpSocket::from_std(socket) -} diff --git a/src/relay/utils.rs b/src/relay/utils.rs index 2d75f0ea..08a0c9e7 100644 --- a/src/relay/utils.rs +++ b/src/relay/utils.rs @@ -1,6 +1,8 @@ use std::{ future::Future, - io::{self, Error}, + io::{self, Error, ErrorKind}, + mem, + net::{SocketAddr, SocketAddrV4, SocketAddrV6}, time::Duration, }; @@ -40,3 +42,22 @@ pub fn set_nofile(_nofile: u64) -> io::Result<()> { // set_rlimit only works on *nix systems Ok(()) } + +/// Convert `sockaddr_storage` to `SocketAddr` +#[allow(dead_code)] +pub fn sockaddr_to_std(saddr: &libc::sockaddr_storage) -> io::Result { + match saddr.ss_family as libc::c_int { + libc::AF_INET => unsafe { + let addr: SocketAddrV4 = mem::transmute_copy(&saddr); + Ok(SocketAddr::V4(addr)) + }, + libc::AF_INET6 => unsafe { + let addr: SocketAddrV6 = mem::transmute_copy(&saddr); + Ok(SocketAddr::V6(addr)) + }, + _ => { + let err = Error::new(ErrorKind::InvalidData, "family must be either AF_INET or AF_INET6"); + Err(err) + } + } +}