net: fix IPv6

leaf/src/app/dispatcher.rs

@@ -6,7 +6,7 @@ use std::time::Duration;
 use async_recursion::async_recursion;
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio::sync::RwLock;
-use tracing::{debug, info, trace, warn};
+use tracing::{debug, info, warn};
 
 use crate::{
     app::SyncDnsClient,
@@ -132,7 +132,7 @@ impl Dispatcher {
                     tag.to_owned()
                 }
                 Err(err) => {
-                    trace!("pick route failed: {}", err);
+                    debug!("pick route failed: {}", err);
                     if let Some(tag) = self.outbound_manager.read().await.default_handler() {
                         debug!(
                             "picked default route [{}] for {} -> {}",
@@ -267,7 +267,7 @@ impl Dispatcher {
                     tag.to_owned()
                 }
                 Err(err) => {
-                    trace!("pick route failed: {}", err);
+                    debug!("pick route failed: {}", err);
                     if let Some(tag) = self.outbound_manager.read().await.default_handler() {
                         debug!(
                             "picked default route [{}] for {} -> {}",

leaf/src/app/dns_client.rs

@@ -1,5 +1,5 @@
 use std::collections::HashMap;
-use std::net::{IpAddr, SocketAddr};
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
 use std::num::NonZeroUsize;
 use std::str::FromStr;
 use std::sync::{Arc, Weak};
@@ -177,12 +177,21 @@ impl DnsClient {
         server: &SocketAddr,
     ) -> Result<CacheEntry> {
         let socket = if is_direct {
+            debug!("direct lookup");
             let socket = self.new_udp_socket(server).await?;
             Box::new(StdOutboundDatagram::new(socket))
         } else {
+            debug!("dispatched lookup");
             if let Some(dispatcher_weak) = self.dispatcher.as_ref() {
+                // The source address will be used to determine which address the
+                // underlying socket will bind.
+                let source = match server {
+                    SocketAddr::V4(_) => SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0),
+                    SocketAddr::V6(_) => SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0),
+                };
                 let sess = Session {
                     network: Network::Udp,
+                    source,
                     destination: SocksAddr::from(server),
                     inbound_tag: "internal".to_string(),
                     ..Default::default()
@@ -265,7 +274,7 @@ impl DnsClient {
                                         break;
                                     };
                                     let entry = CacheEntry { ips, deadline };
-                                    trace!("ips for {}:\n{:#?}", host, &entry);
+                                    debug!("ips for {}: {:#?}", host, &entry);
                                     return Ok(entry);
                                 } else {
                                     // response with 0 records

leaf/src/app/nat_manager.rs

@@ -83,10 +83,9 @@ impl NatManager {
                 let n_removed = n_total - n_remaining;
                 drop(sessions); // release the lock
                 if n_removed > 0 {
-                    trace!(
+                    debug!(
                         "removed {} nat sessions, remaining {} sessions",
-                        n_removed,
-                        n_remaining
+                        n_removed, n_remaining
                     );
                 }
                 tokio::time::sleep(Duration::from_secs(

leaf/src/app/router.rs

@@ -526,7 +526,7 @@ impl Router {
             if !ips.is_empty() {
                 let mut new_sess = sess.clone();
                 new_sess.destination = SocksAddr::from((ips[0], sess.destination.port()));
-                trace!(
+                debug!(
                     "re-matching with resolved ip [{}] for [{}]",
                     ips[0],
                     sess.destination.host()

leaf/src/proxy/datagram.rs

@@ -68,28 +68,31 @@ impl OutboundDatagramSendHalf for StdOutboundDatagramSendHalf {
     }
 }
 
-/// An outbound datagram simply wraps a UDP socket.
-pub struct SimpleOutboundDatagram {
+/// An outbound datagram that sends to a domain target.
+pub struct DomainAssociatedOutboundDatagram {
     inner: UdpSocket,
-    destination: Option<SocksAddr>,
+    source: SocketAddr,
+    destination: SocksAddr,
     dns_client: SyncDnsClient,
 }
 
-impl SimpleOutboundDatagram {
+impl DomainAssociatedOutboundDatagram {
     pub fn new(
         inner: UdpSocket,
-        destination: Option<SocksAddr>,
+        source: SocketAddr,
+        destination: SocksAddr,
         dns_client: SyncDnsClient,
     ) -> Self {
-        SimpleOutboundDatagram {
+        DomainAssociatedOutboundDatagram {
             inner,
+            source,
             destination,
             dns_client,
         }
     }
 }
 
-impl OutboundDatagram for SimpleOutboundDatagram {
+impl OutboundDatagram for DomainAssociatedOutboundDatagram {
     fn split(
         self: Box<Self>,
     ) -> (
@@ -99,8 +102,15 @@ impl OutboundDatagram for SimpleOutboundDatagram {
         let r = Arc::new(self.inner);
         let s = r.clone();
         (
-            Box::new(SimpleOutboundDatagramRecvHalf(r, self.destination)),
-            Box::new(SimpleOutboundDatagramSendHalf(s, self.dns_client)),
+            Box::new(DomainAssociatedOutboundDatagramRecvHalf(
+                r,
+                self.destination,
+            )),
+            Box::new(DomainAssociatedOutboundDatagramSendHalf(
+                s,
+                self.source,
+                self.dns_client,
+            )),
         )
     }
 }
@@ -117,33 +127,27 @@ fn unmapped_ipv4(addr: SocketAddr) -> SocketAddr {
     addr
 }
 
-pub struct SimpleOutboundDatagramRecvHalf(Arc<UdpSocket>, Option<SocksAddr>);
+pub struct DomainAssociatedOutboundDatagramRecvHalf(Arc<UdpSocket>, SocksAddr);
 
 #[async_trait]
-impl OutboundDatagramRecvHalf for SimpleOutboundDatagramRecvHalf {
+impl OutboundDatagramRecvHalf for DomainAssociatedOutboundDatagramRecvHalf {
     async fn recv_from(&mut self, buf: &mut [u8]) -> io::Result<(usize, SocksAddr)> {
         match self.0.recv_from(buf).await {
-            Ok((n, a)) => {
-                if self.1.is_some() {
-                    Ok((n, self.1.as_ref().unwrap().clone()))
-                } else {
-                    Ok((n, SocksAddr::Ip(unmapped_ipv4(a))))
-                }
-            }
+            Ok((n, a)) => Ok((n, self.1.clone())),
             Err(e) => Err(e),
         }
     }
 }
 
-pub struct SimpleOutboundDatagramSendHalf(Arc<UdpSocket>, SyncDnsClient);
+pub struct DomainAssociatedOutboundDatagramSendHalf(Arc<UdpSocket>, SocketAddr, SyncDnsClient);
 
 #[async_trait]
-impl OutboundDatagramSendHalf for SimpleOutboundDatagramSendHalf {
+impl OutboundDatagramSendHalf for DomainAssociatedOutboundDatagramSendHalf {
     async fn send_to(&mut self, buf: &[u8], target: &SocksAddr) -> io::Result<usize> {
         let addr = match target {
             SocksAddr::Domain(domain, port) => {
                 let ips = {
-                    self.1
+                    self.2
                         .read()
                         .await
                         .lookup(domain)
@@ -155,13 +159,20 @@ impl OutboundDatagramSendHalf for SimpleOutboundDatagramSendHalf {
                         })
                         .await?
                 };
-                if ips.is_empty() {
+                // FIXME Since FakeDns returns IPv4 address only, it's always bound
+                // to IPv4 address if FakeDns is used.
+                //
+                // If the socket was bound to an IPv4 address, we need an IPv4
+                // address for sending, and vice versa for IPv6.
+                let needs_ipv4 = self.1.is_ipv4();
+                if let Some(ip) = ips.into_iter().find(|x| x.is_ipv4() == needs_ipv4) {
+                    SocketAddr::new(ip, port.to_owned())
+                } else {
                     return Err(io::Error::new(
                         io::ErrorKind::InvalidInput,
                         "could not resolve to any address",
                     ));
                 }
-                SocketAddr::new(ips[0], port.to_owned())
             }
             SocksAddr::Ip(a) => a.to_owned(),
         };

leaf/src/proxy/mod.rs

@@ -13,7 +13,7 @@ use thiserror::Error;
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio::net::{TcpSocket, TcpStream, UdpSocket};
 use tokio::time::timeout;
-use tracing::trace;
+use tracing::{debug, trace};
 
 #[cfg(unix)]
 use std::os::unix::io::{AsFd, AsRawFd};
@@ -84,9 +84,9 @@ pub mod vmess;
 pub mod ws;
 
 pub use datagram::{
-    SimpleInboundDatagram, SimpleInboundDatagramRecvHalf, SimpleInboundDatagramSendHalf,
-    SimpleOutboundDatagram, SimpleOutboundDatagramRecvHalf, SimpleOutboundDatagramSendHalf,
-    StdOutboundDatagram,
+    DomainAssociatedOutboundDatagram, DomainAssociatedOutboundDatagramRecvHalf,
+    DomainAssociatedOutboundDatagramSendHalf, SimpleInboundDatagram, SimpleInboundDatagramRecvHalf,
+    SimpleInboundDatagramSendHalf, StdOutboundDatagram,
 };
 
 #[derive(Error, Debug)]
@@ -207,12 +207,12 @@ async fn bind_socket<T: BindSocket>(socket: &T, indicator: &SocketAddr) -> io::R
     match indicator.ip() {
         IpAddr::V4(v4) if v4.is_loopback() => {
             socket.bind(&SocketAddrV4::new(Ipv4Addr::new(127, 0, 0, 1), 0).into())?;
-            trace!("socket bind loopback v4");
+            debug!("socket bind loopback v4");
             return Ok(());
         }
         IpAddr::V6(v6) if v6.is_loopback() => {
             socket.bind(&SocketAddrV6::new("::1".parse().unwrap(), 0, 0, 0).into())?;
-            trace!("socket bind loopback v6");
+            debug!("socket bind loopback v6");
             return Ok(());
         }
         _ => {}
@@ -250,7 +250,7 @@ async fn bind_socket<T: BindSocket>(socket: &T, indicator: &SocketAddr) -> io::R
                         last_err = Some(io::Error::last_os_error());
                         continue;
                     }
-                    trace!("socket bind {}", iface);
+                    debug!("socket bind {}", iface);
                     return Ok(());
                 }
                 #[cfg(target_os = "linux")]
@@ -267,7 +267,7 @@ async fn bind_socket<T: BindSocket>(socket: &T, indicator: &SocketAddr) -> io::R
                         last_err = Some(io::Error::last_os_error());
                         continue;
                     }
-                    trace!("socket bind {}", iface);
+                    debug!("socket bind {}", iface);
                     return Ok(());
                 }
                 #[cfg(not(any(target_os = "macos", target_os = "linux")))]
@@ -286,7 +286,7 @@ async fn bind_socket<T: BindSocket>(socket: &T, indicator: &SocketAddr) -> io::R
                         last_err = Some(e);
                         continue;
                     }
-                    trace!("socket bind {}", addr);
+                    debug!("socket bind {}", addr);
                     return Ok(());
                 }
             }
@@ -303,26 +303,12 @@ async fn bind_socket<T: BindSocket>(socket: &T, indicator: &SocketAddr) -> io::R
 // New UDP socket.
 pub async fn new_udp_socket(indicator: &SocketAddr) -> io::Result<UdpSocket> {
     use socket2::{Domain, Socket, Type};
-    let socket = if *option::ENABLE_IPV6 {
-        // Dual-stack socket.
-        // FIXME Windows IPV6_V6ONLY?
-        Socket::new(Domain::IPV6, Type::DGRAM, None)?
-    } else {
-        match indicator {
-            SocketAddr::V4(..) => Socket::new(Domain::IPV4, Type::DGRAM, None)?,
-            SocketAddr::V6(..) => Socket::new(Domain::IPV6, Type::DGRAM, None)?,
-        }
+    let socket = match indicator {
+        SocketAddr::V4(..) => Socket::new(Domain::IPV4, Type::DGRAM, None)?,
+        SocketAddr::V6(..) => Socket::new(Domain::IPV6, Type::DGRAM, None)?,
     };
     socket.set_nonblocking(true)?;
 
-    // If the proxy request is coming from an inbound listens on the loopback,
-    // the indicator could be a loopback address, we must ignore it.
-    if indicator.ip().is_loopback() || *option::ENABLE_IPV6 {
-        bind_socket(&socket, &*option::UNSPECIFIED_BIND_ADDR).await?;
-    } else {
-        bind_socket(&socket, indicator).await?;
-    }
-
     #[cfg(target_os = "android")]
     protect_socket(socket.as_raw_fd()).await?;
 
@@ -369,7 +355,7 @@ async fn tcp_dial_task(dial_addr: SocketAddr) -> io::Result<DialResult> {
     #[cfg(target_os = "android")]
     protect_socket(socket.as_raw_fd()).await?;
 
-    trace!("tcp dialing {}", &dial_addr);
+    debug!("tcp dialing {}", &dial_addr);
     let start = tokio::time::Instant::now();
     let stream = timeout(
         Duration::from_secs(*option::OUTBOUND_DIAL_TIMEOUT),
@@ -380,7 +366,7 @@ async fn tcp_dial_task(dial_addr: SocketAddr) -> io::Result<DialResult> {
 
     apply_socket_opts(&stream)?;
 
-    trace!(
+    debug!(
         "tcp {} <-> {} connected in {}ms",
         stream.local_addr()?,
         &dial_addr,
@@ -423,26 +409,33 @@ pub async fn connect_datagram_outbound(
             Network::Udp => {
                 let socket = new_udp_socket(&sess.source).await?;
                 Ok(Some(OutboundTransport::Datagram(Box::new(
-                    SimpleOutboundDatagram::new(socket, None, dns_client.clone()),
+                    StdOutboundDatagram::new(socket),
                 ))))
             }
             Network::Tcp => {
                 let stream = new_tcp_stream(dns_client.clone(), &addr, &port).await?;
                 Ok(Some(OutboundTransport::Stream(stream)))
             }
         },
-        OutboundConnect::Direct => {
-            let socket = new_udp_socket(&sess.source).await?;
-            let dest = match &sess.destination {
-                SocksAddr::Domain(domain, port) => {
-                    Some(SocksAddr::Domain(domain.to_owned(), port.to_owned()))
-                }
-                _ => None,
-            };
-            Ok(Some(OutboundTransport::Datagram(Box::new(
-                SimpleOutboundDatagram::new(socket, dest, dns_client.clone()),
-            ))))
-        }
+        OutboundConnect::Direct => match &sess.destination {
+            SocksAddr::Domain(domain, port) => {
+                let socket = new_udp_socket(&sess.source).await?;
+                Ok(Some(OutboundTransport::Datagram(Box::new(
+                    DomainAssociatedOutboundDatagram::new(
+                        socket,
+                        sess.source.clone(),
+                        SocksAddr::Domain(domain.to_owned(), port.to_owned()),
+                        dns_client.clone(),
+                    ),
+                ))))
+            }
+            SocksAddr::Ip(_) => {
+                let socket = new_udp_socket(&sess.source).await?;
+                Ok(Some(OutboundTransport::Datagram(Box::new(
+                    StdOutboundDatagram::new(socket),
+                ))))
+            }
+        },
         _ => Ok(None),
     }
 }