[#190] Supports UDP transparent proxy on Linux, BSD-like untested

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"] }

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<SocketAddr> {
-            unimplemented!("Transparent Proxy (redir) doesn't work on this platform");
+            unimplemented!("TCP Transparent Proxy (redir) is not supported on this platform");
         }
     }
 }

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,
 };

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;
 

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<Vec<u8>>,
+
+    // local <- remote task life watcher
+    watcher: Arc<UdpAssociationWatcher>,
+}
+
+impl UdpAssociation {
+    /// Create an association with addr
+    async fn associate(
+        context: SharedContext,
+        svr_cfg: Arc<ServerScore>,
+        src_addr: SocketAddr,
+        dst_addr: SocketAddr,
+        assoc_map: SharedAssocMap,
+    ) -> io::Result<UdpAssociation> {
+        // 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::<Vec<u8>>(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<u8>) {
+        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<ServerScore> {
+        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<String, UdpAssociation>;
+type SharedAssocMap = Arc<Mutex<AssocMap>>;
 
 /// 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;
+    }
 }

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,
 };

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,
 };

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<mio::net::UdpSocket>,
+}
+
+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<TProxyUdpSocket> {
+        // 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<usize> {
+        poll_fn(|cx| self.poll_send_to(cx, buf, target)).await
+    }
+
+    fn poll_send_to(&self, cx: &mut Context<'_>, buf: &[u8], target: &SocketAddr) -> Poll<io::Result<usize>> {
+        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<SocketAddr> {
+        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<io::Result<(usize, SocketAddr, SocketAddr)>> {
+        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),
+        }
+    }
+}

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<libc::sockaddr_storage> {
+    // 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::<libc::sockaddr_in>(),
+                    );
+
+                    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::<libc::sockaddr_in6>(),
+                    );
+
+                    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)?))
+    }
+}

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<libc::sockaddr_storage> {
+    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::<libc::sockaddr_in>(),
+                    );
+
+                    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::<libc::sockaddr_in6>(),
+                    );
+
+                    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)?))
+    }
+}

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");
+}

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> {
+    UdpSocket::bind(addr).await
+}

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");
+}

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<UdpSocket> {
+    // 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");
+}

src/relay/udprelay/tproxy_socket.rs

@@ -0,0 +1,3 @@
+//! Socket for supporting TPROXY
+
+pub use super::sys::TProxyUdpSocket;

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,
 };

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> {
-    UdpSocket::bind(addr).await
-}
-
-#[cfg(windows)]
-pub async fn create_socket(addr: &SocketAddr) -> io::Result<UdpSocket> {
-    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)
-}

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<SocketAddr> {
+    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)
+        }
+    }
+}