feat(service): outbound UDP relay through SOCKS5-only chains

Phase 2 of the outbound proxy refactor. The new
OutboundProxyDatagram type implements DatagramSocket / DatagramSend /
DatagramReceive on top of a freshly-bound local UDP socket plus one
SOCKS5 UDP ASSOCIATE per chain hop. Each datagram on the wire carries
N nested UdpAssociateHeader frames (one per hop), the inner-most one
addressing the ss-server's UDP external endpoint.

Per-hop control connections are built by reusing the existing TCP
chain builder (connect_chain_for_udp_associate) so all of the
shadowsocks DNS resolver / ConnectOpts behaviour still applies.

Integration:

* OutboundProxyClient::associate_udp + supports_udp().
* OutboundProxyClient::supports_udp() returns true iff every hop is
  SOCKS5; HTTP/HTTPS hops cannot transport UDP and downgrade the
  whole chain to 'unsupported'.
* local::net::udp::association::create_proxied_socket dispatches
  based on supports_udp(): when true the per-association socket is
  ProxySocket<OutboundProxyDatagram> wrapping the chained datagram;
  otherwise we keep the previous direct ProxySocket<ShadowUdpSocket>.
* When the chain is configured but unable to relay UDP, sslocal /
  ssserver emit a single startup WARN ('UDP traffic will bypass the
  chain') instead of failing — runtime traffic transparently falls
  back to a direct connection to the ss-server.

OutboundProxyStream gains try_into_tcp() so SOCKS5-only chains can
recover the bare TcpStream as keep-alive control connections —
Sync-friendly storage required because the HTTP CONNECT variant
internally holds a hyper::upgrade::Upgraded which is !Sync.

Author: zonyitoo

crates/shadowsocks-service/src/local/net/udp/association.rs

@@ -17,22 +17,109 @@ use rand::{RngExt, rngs::SmallRng};
 use tokio::{sync::mpsc, task::JoinHandle, time};
 
 use shadowsocks::{
+    config::ServerConfig,
     lookup_then,
-    net::{AddrFamily, UdpSocket as ShadowUdpSocket},
+    net::{AddrFamily, ConnectOpts, TcpStream as ShadowTcpStream, UdpSocket as ShadowUdpSocket},
     relay::{
         Address,
-        udprelay::{MAXIMUM_UDP_PAYLOAD_SIZE, ProxySocket, options::UdpSocketControlData},
+        udprelay::{
+            MAXIMUM_UDP_PAYLOAD_SIZE, ProxySocket,
+            options::UdpSocketControlData,
+            proxy_socket::UdpSocketType,
+        },
     },
 };
 
 use crate::{
     local::{context::ServiceContext, loadbalancing::PingBalancer},
     net::{
-        MonProxySocket, UDP_ASSOCIATION_KEEP_ALIVE_CHANNEL_SIZE, UDP_ASSOCIATION_SEND_CHANNEL_SIZE,
+        MonProxySocket, OutboundProxyDatagram, TcpDialer,
+        UDP_ASSOCIATION_KEEP_ALIVE_CHANNEL_SIZE, UDP_ASSOCIATION_SEND_CHANNEL_SIZE,
         packet_window::PacketWindowFilter,
     },
 };
 
+/// Build the proxied socket appropriate for `svr_cfg`, transparently
+/// going through the configured outbound chain when one is present and
+/// fully SOCKS5 (otherwise the chain is bypassed for UDP).
+async fn create_proxied_socket(
+    context: &ServiceContext,
+    svr_cfg: &ServerConfig,
+    connect_opts: &ConnectOpts,
+) -> io::Result<ProxiedSocket> {
+    let use_chain = context
+        .outbound_client()
+        .map(|c| c.supports_udp())
+        .unwrap_or(false);
+
+    if use_chain {
+        let client = context
+            .outbound_client()
+            .expect("outbound_client checked above")
+            .clone();
+        let dialer = LocalTcpDialer {
+            context: Arc::new(context.clone()),
+            opts: connect_opts.clone(),
+        };
+        let target: Address = svr_cfg.udp_external_addr().into();
+        let datagram = client
+            .associate_udp(&context.context(), &dialer, connect_opts, target)
+            .await?;
+        let proxy_socket =
+            ProxySocket::from_socket(UdpSocketType::Client, context.context(), svr_cfg, datagram);
+        let mon = MonProxySocket::from_socket(proxy_socket, context.flow_stat());
+        Ok(ProxiedSocket::Chained(mon))
+    } else {
+        let socket = ProxySocket::connect_with_opts(context.context(), svr_cfg, connect_opts).await?;
+        let mon = MonProxySocket::from_socket(socket, context.flow_stat());
+        Ok(ProxiedSocket::Direct(mon))
+    }
+}
+
+/// `TcpDialer` adapter that uses the local service context's connect options.
+struct LocalTcpDialer {
+    context: Arc<ServiceContext>,
+    opts: ConnectOpts,
+}
+
+impl TcpDialer for LocalTcpDialer {
+    async fn dial(&self, addr: &Address) -> io::Result<ShadowTcpStream> {
+        ShadowTcpStream::connect_remote_with_opts(self.context.context_ref(), addr, &self.opts).await
+    }
+}
+
+/// Proxied UDP socket, either direct (single hop to ss-server) or routed
+/// through a SOCKS5-only outbound proxy chain.
+#[allow(clippy::large_enum_variant)]
+enum ProxiedSocket {
+    Direct(MonProxySocket<ShadowUdpSocket>),
+    Chained(MonProxySocket<OutboundProxyDatagram>),
+}
+
+impl ProxiedSocket {
+    async fn send_with_ctrl(
+        &self,
+        addr: &Address,
+        control: &UdpSocketControlData,
+        data: &[u8],
+    ) -> io::Result<()> {
+        match self {
+            Self::Direct(s) => s.send_with_ctrl(addr, control, data).await,
+            Self::Chained(s) => s.send_with_ctrl(addr, control, data).await,
+        }
+    }
+
+    async fn recv_with_ctrl(
+        &self,
+        buf: &mut [u8],
+    ) -> io::Result<(usize, Address, Option<UdpSocketControlData>)> {
+        match self {
+            Self::Direct(s) => s.recv_with_ctrl(buf).await,
+            Self::Chained(s) => s.recv_with_ctrl(buf).await,
+        }
+    }
+}
+
 /// Writer for sending packets back to client
 #[trait_variant::make(Send)]
 pub trait UdpInboundWrite {
@@ -213,7 +300,7 @@ where
     peer_addr: SocketAddr,
     bypassed_ipv4_socket: Option<ShadowUdpSocket>,
     bypassed_ipv6_socket: Option<ShadowUdpSocket>,
-    proxied_socket: Option<MonProxySocket<ShadowUdpSocket>>,
+    proxied_socket: Option<ProxiedSocket>,
     keepalive_tx: mpsc::Sender<SocketAddr>,
     keepalive_flag: bool,
     balancer: PingBalancer,
@@ -410,7 +497,7 @@ where
 
         #[inline]
         async fn receive_from_proxied_opt(
-            socket: &Option<MonProxySocket<ShadowUdpSocket>>,
+            socket: &Option<ProxiedSocket>,
             buf: &mut Vec<u8>,
         ) -> io::Result<(usize, Address, Option<UdpSocketControlData>)> {
             match *socket {
@@ -565,15 +652,17 @@ where
             Some(ref mut socket) => socket,
             None => {
                 // Create a new connection to proxy server
-
                 let server = self.balancer.best_udp_server();
                 let svr_cfg = server.server_config();
 
-                let socket =
-                    ProxySocket::connect_with_opts(self.context.context(), svr_cfg, server.connect_opts_ref()).await?;
-                let socket = MonProxySocket::from_socket(socket, self.context.flow_stat());
+                let proxied = create_proxied_socket(
+                    self.context.as_ref(),
+                    svr_cfg,
+                    server.connect_opts_ref(),
+                )
+                .await?;
 
-                self.proxied_socket.insert(socket)
+                self.proxied_socket.insert(proxied)
             }
         };
 

crates/shadowsocks-service/src/net/mod.rs

@@ -5,8 +5,8 @@ pub use self::{
     mon_socket::MonProxySocket,
     mon_stream::MonProxyStream,
     outbound::{
-        HttpProxyAuth, OutboundProxyClient, OutboundProxyHop, OutboundProxyKind, OutboundProxyStream, Socks5Auth,
-        Socks5Negotiator, TcpDialer,
+        HttpProxyAuth, OutboundProxyClient, OutboundProxyDatagram, OutboundProxyHop, OutboundProxyKind,
+        OutboundProxyStream, Socks5Auth, Socks5Negotiator, TcpDialer,
     },
 };
 

crates/shadowsocks-service/src/net/outbound/chain.rs

@@ -70,6 +70,39 @@ where
     Ok(stream)
 }
 
+/// Build a TCP control connection that *terminates at* `hops[hop_index]`
+/// (i.e. previous hops are traversed via SOCKS5 TcpConnect, but the last
+/// hop is left in its post-handshake "ready for an arbitrary command"
+/// state — typically `UDP ASSOCIATE`).
+///
+/// All hops in `hops[..=hop_index]` must be SOCKS5; this helper is used
+/// exclusively by the UDP outbound path which only supports
+/// SOCKS5-only chains.
+pub(crate) async fn connect_chain_for_udp_associate<D>(
+    hops: &[OutboundProxyHop],
+    hop_index: usize,
+    dialer: &D,
+) -> io::Result<OutboundProxyStream>
+where
+    D: TcpDialer + Sync,
+{
+    debug_assert!(hop_index < hops.len());
+    let prefix = &hops[..hop_index];
+    let target_hop = &hops[hop_index];
+
+    if hop_index == 0 {
+        // Direct dial.
+        let tcp = dialer.dial(&target_hop.addr).await?;
+        return OutboundProxyStream::from_tcp(tcp);
+    }
+
+    // Reuse `connect_chain`: target = hops[hop_index].addr; the chain
+    // builder will SOCKS5-TcpConnect through prefix and leave the byte
+    // stream pointed at `hops[hop_index]` ready for the caller to issue
+    // its own SOCKS5 handshake / UdpAssociate command on top.
+    connect_chain(prefix, dialer, &target_hop.addr).await
+}
+
 async fn negotiate_hop(
     mut stream: OutboundProxyStream,
     hop: &OutboundProxyHop,

crates/shadowsocks-service/src/net/outbound/mod.rs

@@ -23,7 +23,11 @@
 
 use std::{io, sync::Arc};
 
-use shadowsocks::relay::socks5::Address;
+use shadowsocks::{
+    context::SharedContext,
+    net::ConnectOpts,
+    relay::socks5::Address,
+};
 
 use crate::config::{OutboundProxy, OutboundProxyProtocol};
 
@@ -34,10 +38,12 @@ pub mod http_connect;
 pub mod socks5;
 pub mod stream;
 pub mod tls;
+pub mod udp;
 
 pub use auth::{HttpProxyAuth, Socks5Auth};
 pub use socks5::Socks5Negotiator;
 pub use stream::OutboundProxyStream;
+pub use udp::OutboundProxyDatagram;
 
 #[cfg(feature = "local-http")]
 pub use http_connect::{HttpConnectClient, HttpConnectTunnel};
@@ -141,6 +147,25 @@ impl OutboundProxyClient {
     pub fn supports_udp(&self) -> bool {
         !self.hops.is_empty() && self.hops.iter().all(OutboundProxyHop::supports_udp)
     }
+
+    /// Establish a multi-hop UDP relay through the chain.
+    ///
+    /// Requires every hop to be SOCKS5 (see [`Self::supports_udp`]).
+    /// `target` is the inner-most destination address baked into every
+    /// outgoing datagram's SOCKS5 UDP header (typically the ss-server's
+    /// UDP external address).
+    pub async fn associate_udp<D>(
+        &self,
+        context: &SharedContext,
+        dialer: &D,
+        connect_opts: &ConnectOpts,
+        target: Address,
+    ) -> io::Result<OutboundProxyDatagram>
+    where
+        D: TcpDialer + Sync,
+    {
+        OutboundProxyDatagram::associate(self, context, dialer, connect_opts, target).await
+    }
 }
 
 fn hop_from_config(proxy: &OutboundProxy) -> OutboundProxyHop {

crates/shadowsocks-service/src/net/outbound/stream.rs

@@ -79,6 +79,25 @@ impl OutboundProxyStream {
         Ok(self.local_addr)
     }
 
+    /// Try to recover the underlying [`TcpStream`].
+    ///
+    /// Succeeds only if the stream is still in the unwrapped TCP state
+    /// (i.e. no TLS or HTTP CONNECT layer has been applied). The UDP
+    /// outbound path uses this to obtain a keep-alive connection that is
+    /// `Sync`-friendly (`OutboundProxyStream` itself is intentionally not
+    /// `Sync` because the HTTP CONNECT variant wraps a
+    /// `hyper::upgrade::Upgraded` which is not).
+    #[allow(clippy::result_large_err)]
+    pub fn try_into_tcp(self) -> Result<TcpStream, Self> {
+        match self.inner {
+            OutboundProxyStreamInner::Bypassed(s) => Ok(s),
+            other => Err(Self {
+                local_addr: self.local_addr,
+                inner: other,
+            }),
+        }
+    }
+
     /// Wrap as a TLS-protected stream (used by the chain builder when the
     /// next hop is HTTPS). `local_addr` is the address recorded for the
     /// very first TCP hop and is preserved unchanged.