refactor(service)!: rework outbound proxy chain into a unified client

Replace the previous parallel outbound proxy implementations
(net::outbound_proxy + local::net::tcp::outbound_proxy) with a single
self-contained module at net::outbound, exposing OutboundProxyClient,
OutboundProxyStream, Socks5Auth, HttpProxyAuth and a TcpDialer trait.

Highlights:

* OutboundProxyStream is now a closed-set enum (Bypassed / Https / Http)
  with statically-dispatched poll_* — no more Box<dyn ProxyStream>. The
  HTTPS variant boxes its inner OutboundProxyStream solely to break the
  recursive type instantiation required by the TLS libraries; this is
  not dynamic dispatch.
* HTTP CONNECT client is rebuilt on top of hyper::client::conn::http1
  + hyper::upgrade. The hand-rolled status-line parser (which could
  drop bytes that arrived in the same read() as the header terminator,
  did not handle 100-Continue, etc.) is gone.
* Auth parameters are now strongly typed:
    Socks5Auth::{None, UsernamePassword{..}}
    HttpProxyAuth::{None, Basic{..}}
  Both enums are #[non_exhaustive] for forward compatibility.
* AutoProxyClientStream's ProxiedViaChain variant is folded back into a
  single Proxied(ProxyClientStream<MonProxyStream<OutboundTransport>>)
  variant, with OutboundTransport = Direct(TcpStream) | Chained(...).
* ServiceContext (both server and local) now caches an
  Arc<OutboundProxyClient> built once at set_outbound_proxies time
  instead of re-parsing the chain on every connection.
* The standalone Socks5TcpClient / Socks5UdpClient implementations
  previously duplicated across net::socks5_client and
  local::socks::client::socks5 are unified into the latter location;
  net::outbound::socks5 only keeps the chain-friendly Socks5Negotiator.
* When UDP is enabled but the configured chain contains a non-SOCKS5
  hop, sslocal/ssserver emits a single startup warning. Phase 2 will
  add UDP relay support for SOCKS5-only chains.

BREAKING CHANGE: the following items have moved or been removed:

* crate::net::outbound_proxy — removed; use crate::net::OutboundProxyClient
* crate::net::http_connect::HttpConnectClient — moved to
  crate::net::HttpConnectClient (only available with the local-http feature)
* crate::net::Socks5TcpClient (the chain-mixing wrapper) — split: the
  pure SOCKS5 client is now crate::local::socks::client::socks5::Socks5TcpClient;
  the in-band negotiator is crate::net::Socks5Negotiator
* crate::local::net::tcp::outbound_proxy — removed
* ServiceContext::outbound_proxies() — replaced by outbound_client()
  returning Option<&Arc<OutboundProxyClient>>

Author: zonyitoo

crates/shadowsocks-service/src/local/context.rs

@@ -18,7 +18,11 @@ use tokio::sync::Mutex;
 #[cfg(feature = "local-fake-dns")]
 use tokio::sync::RwLock;
 
-use crate::{acl::AccessControl, config::{OutboundProxy, SecurityConfig}, net::FlowStat};
+use crate::{
+    acl::AccessControl,
+    config::{OutboundProxy, SecurityConfig},
+    net::{FlowStat, OutboundProxyClient},
+};
 
 #[cfg(feature = "local-fake-dns")]
 use super::fake_dns::manager::FakeDnsManager;
@@ -36,8 +40,8 @@ pub struct ServiceContext {
     // Flow statistic report
     flow_stat: Arc<FlowStat>,
 
-    // Outbound proxy chain (sslocal → ss-server connection goes through these proxies)
-    outbound_proxies: Vec<OutboundProxy>,
+    // Outbound proxy chain (sslocal -> ss-server connection routes through these proxies)
+    outbound_client: Option<Arc<OutboundProxyClient>>,
 
     // For DNS relay's ACL domain name reverse lookup -- whether the IP shall be forwarded
     #[cfg(feature = "local-dns")]
@@ -62,7 +66,7 @@ impl ServiceContext {
             accept_opts: AcceptOpts::default(),
             acl: None,
             flow_stat: Arc::new(FlowStat::new()),
-            outbound_proxies: Vec::new(),
+            outbound_client: None,
             #[cfg(feature = "local-dns")]
             reverse_lookup_cache: Arc::new(Mutex::new(LruCache::with_expiry_duration_and_capacity(
                 Duration::from_secs(3 * 24 * 60 * 60),
@@ -115,12 +119,16 @@ impl ServiceContext {
 
     /// Set outbound proxy chain (connection to SS server routes through these proxies)
     pub fn set_outbound_proxies(&mut self, proxies: Vec<OutboundProxy>) {
-        self.outbound_proxies = proxies;
+        self.outbound_client = if proxies.is_empty() {
+            None
+        } else {
+            Some(Arc::new(OutboundProxyClient::from_config(&proxies)))
+        };
     }
 
-    /// Get outbound proxy chain
-    pub fn outbound_proxies(&self) -> &[OutboundProxy] {
-        &self.outbound_proxies
+    /// Get the outbound proxy client (if a chain is configured).
+    pub fn outbound_client(&self) -> Option<&Arc<OutboundProxyClient>> {
+        self.outbound_client.as_ref()
     }
 
     /// Get cloned flow statistic

crates/shadowsocks-service/src/local/http/mod.rs

@@ -11,5 +11,5 @@ pub mod config;
 mod http_client;
 mod http_service;
 pub mod server;
-mod tokio_rt;
+pub(crate) mod tokio_rt;
 mod utils;

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

@@ -3,7 +3,7 @@
 use std::{io, net::SocketAddr, sync::Arc, time::Duration};
 
 use futures::future;
-use log::{info, trace};
+use log::trace;
 use shadowsocks::{
     config::Mode,
     net::{AcceptOpts, ConnectOpts},
@@ -183,9 +183,13 @@ impl Server {
         if !config.outbound_proxy.is_empty() {
             let has_udp = config.local.iter().any(|local| local.config.mode.enable_udp());
             if has_udp {
-                info!(
-                    "outbound proxy chain only supports TCP; UDP traffic may not be proxied and may behave unexpectedly"
-                );
+                let preview = crate::net::OutboundProxyClient::from_config(&config.outbound_proxy);
+                if !preview.supports_udp() {
+                    log::warn!(
+                        "outbound proxy chain contains non-SOCKS5 hop(s); UDP traffic will bypass the chain. \
+                         Configure a SOCKS5-only chain to enable UDP relay."
+                    );
+                }
             }
             context.set_outbound_proxies(config.outbound_proxy);
         }

crates/shadowsocks-service/src/local/net/tcp/auto_proxy_stream.rs

@@ -18,17 +18,112 @@ use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 
 use crate::{
     local::{context::ServiceContext, loadbalancing::ServerIdent},
-    net::MonProxyStream,
+    net::{MonProxyStream, OutboundProxyStream, TcpDialer},
 };
 
-use super::{auto_proxy_io::AutoProxyIo, outbound_proxy::OutboundProxyStream};
+use super::auto_proxy_io::AutoProxyIo;
+
+/// Outbound transport used by [`AutoProxyClientStream`]: either a direct
+/// TCP connection or a tunnel through the configured outbound proxy chain.
+#[allow(clippy::large_enum_variant)]
+#[pin_project(project = OutboundTransportProj)]
+pub enum OutboundTransport {
+    /// Direct TCP, no outbound chain configured.
+    Direct(#[pin] TcpStream),
+    /// Tunnel produced by `OutboundProxyClient::connect_tcp`.
+    Chained(#[pin] OutboundProxyStream),
+}
+
+impl OutboundTransport {
+    fn local_addr(&self) -> io::Result<SocketAddr> {
+        match self {
+            Self::Direct(s) => s.local_addr(),
+            Self::Chained(s) => s.local_addr(),
+        }
+    }
+
+    fn set_nodelay(&self, nodelay: bool) -> io::Result<()> {
+        match self {
+            Self::Direct(s) => s.set_nodelay(nodelay),
+            // For tunnels we can only forward the request to the underlying
+            // TCP socket if it is exposed; the unified enum has no such
+            // accessor today, so the call is a no-op.
+            Self::Chained(_) => Ok(()),
+        }
+    }
+}
+
+impl AsyncRead for OutboundTransport {
+    fn poll_read(self: Pin<&mut Self>, cx: &mut task::Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
+        match self.project() {
+            OutboundTransportProj::Direct(s) => s.poll_read(cx, buf),
+            OutboundTransportProj::Chained(s) => s.poll_read(cx, buf),
+        }
+    }
+}
+
+impl AsyncWrite for OutboundTransport {
+    fn poll_write(self: Pin<&mut Self>, cx: &mut task::Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
+        match self.project() {
+            OutboundTransportProj::Direct(s) => s.poll_write(cx, buf),
+            OutboundTransportProj::Chained(s) => s.poll_write(cx, buf),
+        }
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<io::Result<()>> {
+        match self.project() {
+            OutboundTransportProj::Direct(s) => s.poll_flush(cx),
+            OutboundTransportProj::Chained(s) => s.poll_flush(cx),
+        }
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<io::Result<()>> {
+        match self.project() {
+            OutboundTransportProj::Direct(s) => s.poll_shutdown(cx),
+            OutboundTransportProj::Chained(s) => s.poll_shutdown(cx),
+        }
+    }
+
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut task::Context<'_>,
+        bufs: &[IoSlice<'_>],
+    ) -> Poll<io::Result<usize>> {
+        match self.project() {
+            OutboundTransportProj::Direct(s) => s.poll_write_vectored(cx, bufs),
+            OutboundTransportProj::Chained(s) => s.poll_write_vectored(cx, bufs),
+        }
+    }
+
+    fn is_write_vectored(&self) -> bool {
+        match self {
+            Self::Direct(s) => s.is_write_vectored(),
+            Self::Chained(s) => s.is_write_vectored(),
+        }
+    }
+}
+
+/// `TcpDialer` adapter that dials directly via the shadowsocks
+/// infrastructure (DNS resolver, connect options).
+struct DirectTcpDialer<'a> {
+    context: &'a ServiceContext,
+    opts: &'a ConnectOpts,
+}
+
+impl<'a> TcpDialer for DirectTcpDialer<'a> {
+    async fn dial(&self, addr: &Address) -> io::Result<TcpStream> {
+        TcpStream::connect_remote_with_opts(self.context.context_ref(), addr, self.opts).await
+    }
+}
 
 /// Unified stream for bypassed and proxied connections
 #[allow(clippy::large_enum_variant)]
 #[pin_project(project = AutoProxyClientStreamProj)]
 pub enum AutoProxyClientStream {
-    Proxied(#[pin] ProxyClientStream<MonProxyStream<TcpStream>>),
-    ProxiedViaChain(#[pin] ProxyClientStream<MonProxyStream<OutboundProxyStream>>),
+    /// Tunnel through the shadowsocks server (optionally over the outbound
+    /// proxy chain).
+    Proxied(#[pin] ProxyClientStream<MonProxyStream<OutboundTransport>>),
+    /// Direct TCP, bypassing the shadowsocks server.
     Bypassed(#[pin] TcpStream),
 }
 
@@ -83,7 +178,6 @@ impl AutoProxyClientStream {
     where
         A: Into<Address>,
     {
-        // Connect directly.
         #[cfg_attr(not(feature = "local-fake-dns"), allow(unused_mut))]
         let mut addr = addr.into();
         #[cfg(feature = "local-fake-dns")]
@@ -143,74 +237,63 @@ impl AutoProxyClientStream {
         A: Into<Address>,
     {
         let flow_stat = context.flow_stat();
-        match context.outbound_proxies() {
-            [] => {
-                let stream = match ProxyClientStream::connect_with_opts_map(
-                    context.context(),
-                    server.server_config(),
-                    addr,
-                    connect_opts,
-                    |stream| MonProxyStream::from_stream(stream, flow_stat),
-                )
+        let target_addr: Address = addr.into();
+        let ss_addr: Address = server.server_config().tcp_external_addr().into();
+
+        let dial_result = match context.outbound_client() {
+            None => TcpStream::connect_remote_with_opts(context.context_ref(), &ss_addr, connect_opts)
                 .await
-                {
-                    Ok(s) => s,
-                    Err(err) => {
-                        server.tcp_score().report_failure().await;
-                        return Err(err);
-                    }
+                .map(OutboundTransport::Direct),
+            Some(client) => {
+                let dialer = DirectTcpDialer {
+                    context: context.as_ref(),
+                    opts: connect_opts,
                 };
-                Ok(Self::Proxied(stream))
+                client
+                    .connect_tcp(&dialer, &ss_addr)
+                    .await
+                    .map(OutboundTransport::Chained)
             }
-            proxies => {
-                use super::outbound_proxy::connect_outbound_proxy_chain;
-
-                let addr = addr.into();
-                let ss_addr = server.server_config().tcp_external_addr().into();
-                let proxy_stream =
-                    match connect_outbound_proxy_chain(context.clone(), ss_addr, proxies, connect_opts).await {
-                        Ok(s) => s,
-                        Err(err) => {
-                            server.tcp_score().report_failure().await;
-                            return Err(err);
-                        }
-                    };
-                let mon_stream = MonProxyStream::from_stream(proxy_stream, flow_stat);
-                let stream =
-                    ProxyClientStream::from_stream(context.context(), mon_stream, server.server_config(), addr);
-                Ok(Self::ProxiedViaChain(stream))
+        };
+
+        let transport = match dial_result {
+            Ok(t) => t,
+            Err(err) => {
+                server.tcp_score().report_failure().await;
+                return Err(err);
             }
-        }
+        };
+
+        let mon = MonProxyStream::from_stream(transport, flow_stat);
+        let stream = ProxyClientStream::from_stream(context.context(), mon, server.server_config(), target_addr);
+        Ok(Self::Proxied(stream))
     }
 
     pub fn local_addr(&self) -> io::Result<SocketAddr> {
         match *self {
             Self::Proxied(ref s) => s.get_ref().get_ref().local_addr(),
-            Self::ProxiedViaChain(ref s) => s.get_ref().get_ref().local_addr(),
             Self::Bypassed(ref s) => s.local_addr(),
         }
     }
 
     pub fn set_nodelay(&self, nodelay: bool) -> io::Result<()> {
         match *self {
             Self::Proxied(ref s) => s.get_ref().get_ref().set_nodelay(nodelay),
-            Self::ProxiedViaChain(..) => Ok(()),
             Self::Bypassed(ref s) => s.set_nodelay(nodelay),
         }
     }
 }
 
 impl AutoProxyIo for AutoProxyClientStream {
     fn is_proxied(&self) -> bool {
-        matches!(*self, Self::Proxied(..) | Self::ProxiedViaChain(..))
+        matches!(*self, Self::Proxied(..))
     }
 }
 
 impl AsyncRead for AutoProxyClientStream {
     fn poll_read(self: Pin<&mut Self>, cx: &mut task::Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
         match self.project() {
             AutoProxyClientStreamProj::Proxied(s) => s.poll_read(cx, buf),
-            AutoProxyClientStreamProj::ProxiedViaChain(s) => s.poll_read(cx, buf),
             AutoProxyClientStreamProj::Bypassed(s) => s.poll_read(cx, buf),
         }
     }
@@ -220,23 +303,20 @@ impl AsyncWrite for AutoProxyClientStream {
     fn poll_write(self: Pin<&mut Self>, cx: &mut task::Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
         match self.project() {
             AutoProxyClientStreamProj::Proxied(s) => s.poll_write(cx, buf),
-            AutoProxyClientStreamProj::ProxiedViaChain(s) => s.poll_write(cx, buf),
             AutoProxyClientStreamProj::Bypassed(s) => s.poll_write(cx, buf),
         }
     }
 
     fn poll_flush(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<io::Result<()>> {
         match self.project() {
             AutoProxyClientStreamProj::Proxied(s) => s.poll_flush(cx),
-            AutoProxyClientStreamProj::ProxiedViaChain(s) => s.poll_flush(cx),
             AutoProxyClientStreamProj::Bypassed(s) => s.poll_flush(cx),
         }
     }
 
     fn poll_shutdown(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<io::Result<()>> {
         match self.project() {
             AutoProxyClientStreamProj::Proxied(s) => s.poll_shutdown(cx),
-            AutoProxyClientStreamProj::ProxiedViaChain(s) => s.poll_shutdown(cx),
             AutoProxyClientStreamProj::Bypassed(s) => s.poll_shutdown(cx),
         }
     }
@@ -248,14 +328,7 @@ impl AsyncWrite for AutoProxyClientStream {
     ) -> Poll<io::Result<usize>> {
         match self.project() {
             AutoProxyClientStreamProj::Proxied(s) => s.poll_write_vectored(cx, bufs),
-            AutoProxyClientStreamProj::ProxiedViaChain(s) => s.poll_write_vectored(cx, bufs),
             AutoProxyClientStreamProj::Bypassed(s) => s.poll_write_vectored(cx, bufs),
         }
     }
 }
-
-impl From<ProxyClientStream<MonProxyStream<TcpStream>>> for AutoProxyClientStream {
-    fn from(s: ProxyClientStream<MonProxyStream<TcpStream>>) -> Self {
-        Self::Proxied(s)
-    }
-}

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

@@ -1,4 +1,3 @@
 pub mod auto_proxy_io;
 pub mod auto_proxy_stream;
 pub mod listener;
-pub(crate) mod outbound_proxy;