feat(proxy-chain): refactor outbound proxy architecture

- Move HTTP CONNECT logic to shared outbound_proxy helper module
- Simplify Socks5TcpClient to only handle SOCKS5 protocol, remove OutboundProxyAuth dependency
- Extract proxy chain utilities into separate net/outbound_proxy.rs with HTTP/HTTPS support
- Add UDP mode warnings in local and server startup when proxy chain is configured
- Share ProxyStream trait between modules for better code organization
- Gate HTTPS proxy support with feature flags (local-http-native-tls or local-http-rustls)
This commit is contained in:
zonyitoo
2026-04-12 01:12:21 +08:00
parent c18abd4062
commit 4d7b9e1203
9 changed files with 252 additions and 221 deletions
+7 -1
View File
@@ -3,7 +3,7 @@
use std::{io, net::SocketAddr, sync::Arc, time::Duration};
use futures::future;
use log::trace;
use log::{info, trace};
use shadowsocks::{
config::Mode,
net::{AcceptOpts, ConnectOpts},
@@ -181,6 +181,12 @@ impl Server {
context.set_security_config(&config.security);
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"
);
}
context.set_outbound_proxies(config.outbound_proxy);
}
@@ -163,19 +163,21 @@ impl AutoProxyClientStream {
Ok(Self::Proxied(stream))
}
proxies => {
use super::outbound_proxy::connect_outbound_chain;
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_chain(context.clone(), ss_addr, proxies, connect_opts).await {
Ok(s) => s,
Err(err) => {
server.tcp_score().report_failure().await;
return Err(err);
}
};
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);
let stream =
ProxyClientStream::from_stream(context.context(), mon_stream, server.server_config(), addr);
Ok(Self::ProxiedViaChain(stream))
}
}
@@ -22,7 +22,8 @@ use crate::{
local::context::ServiceContext,
net::{
http_stream::ProxyHttpStream,
socks5_client::{http_connect, socks5_command, socks5_handshake},
outbound_proxy::connect_via_proxy,
socks5_client::{socks5_command, socks5_handshake},
},
};
@@ -82,14 +83,17 @@ impl AsyncWrite for OutboundProxyStream {
/// which uses the shadowsocks infrastructure (DNS, connect opts, etc.).
/// HTTPS proxies use the shared `ProxyHttpStream::connect_https()` helper,
/// the same transport used by the local HTTP client for TLS connections.
pub async fn connect_outbound_chain(
pub async fn connect_outbound_proxy_chain(
context: Arc<ServiceContext>,
target: Address,
proxies: &[OutboundProxy],
connect_opts: &ConnectOpts,
) -> io::Result<OutboundProxyStream> {
let Some(first_proxy) = proxies.first() else {
return Err(io::Error::new(io::ErrorKind::InvalidInput, "empty outbound proxy chain"));
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"empty outbound proxy chain",
));
};
let first_addr = first_proxy.address();
@@ -102,8 +106,19 @@ pub async fn connect_outbound_chain(
// For the first proxy: if it's HTTPS, TLS-wrap with the shared helper that is
// also used by the local HTTP client. Otherwise box directly.
let mut stream: OutboundProxyStream = if first_proxy.protocol == OutboundProxyProtocol::Https {
let tls = ProxyHttpStream::connect_https(initial, &first_proxy.host).await?;
OutboundProxyStream::new(tls, local_addr)
#[cfg(any(feature = "local-http-native-tls", feature = "local-http-rustls"))]
{
let tls = ProxyHttpStream::connect_https(initial, &first_proxy.host).await?;
OutboundProxyStream::new(tls, local_addr)
}
#[cfg(not(any(feature = "local-http-native-tls", feature = "local-http-rustls")))]
{
return Err(io::Error::new(
io::ErrorKind::Unsupported,
"HTTPS outbound proxy requires either local-http-native-tls or local-http-rustls feature",
));
}
} else {
OutboundProxyStream::new(initial, local_addr)
};
@@ -111,9 +126,20 @@ pub async fn connect_outbound_chain(
// Process each proxy hop in order.
for (idx, proxy) in proxies.iter().enumerate() {
if idx > 0 && proxy.protocol == OutboundProxyProtocol::Https {
let local_addr = stream.local_addr()?;
let tls = ProxyHttpStream::connect_https(stream, &proxy.host).await?;
stream = OutboundProxyStream::new(tls, local_addr);
#[cfg(any(feature = "local-http-native-tls", feature = "local-http-rustls"))]
{
let local_addr = stream.local_addr()?;
let tls = ProxyHttpStream::connect_https(stream, &proxy.host).await?;
stream = OutboundProxyStream::new(tls, local_addr);
}
#[cfg(not(any(feature = "local-http-native-tls", feature = "local-http-rustls")))]
{
return Err(io::Error::new(
io::ErrorKind::Unsupported,
"HTTPS outbound proxy requires either local-http-native-tls or local-http-rustls feature",
));
}
}
let next_target = proxies
@@ -121,22 +147,7 @@ pub async fn connect_outbound_chain(
.map(OutboundProxy::address)
.unwrap_or_else(|| target.clone());
match proxy.protocol {
OutboundProxyProtocol::Socks5 => {
socks5_handshake(&mut stream, proxy.auth.as_ref())
.await
.map_err(io::Error::other)?;
socks5_command(&mut stream, Command::TcpConnect, next_target)
.await
.map_err(io::Error::other)?;
}
OutboundProxyProtocol::Http => {
http_connect(&mut stream, proxy, &next_target).await?;
}
OutboundProxyProtocol::Https => {
http_connect(&mut stream, proxy, &next_target).await?;
}
}
connect_via_proxy(&mut stream, proxy, &next_target).await?;
}
Ok(stream)
@@ -13,7 +13,7 @@ use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
#[pin_project(project = ProxyHttpStreamProj)]
pub enum ProxyHttpStream<S> {
Http(#[pin] S),
#[cfg(not(feature = "local-http-rustls"))]
#[cfg(all(feature = "local-http-native-tls", not(feature = "local-http-rustls")))]
Https(#[pin] tokio_native_tls::TlsStream<S>, bool),
#[cfg(feature = "local-http-rustls")]
Https(#[pin] tokio_rustls::client::TlsStream<S>, bool),
@@ -27,7 +27,7 @@ where
Self::Http(stream)
}
#[cfg(not(feature = "local-http-rustls"))]
#[cfg(all(feature = "local-http-native-tls", not(feature = "local-http-rustls")))]
pub async fn connect_https(stream: S, domain: &str) -> io::Result<Self> {
use native_tls::TlsConnector;
@@ -119,6 +119,7 @@ where
pub fn negotiated_http2(&self) -> bool {
match *self {
Self::Http(..) => false,
#[cfg(any(feature = "local-http-native-tls", feature = "local-http-rustls"))]
Self::Https(_, n) => n,
}
}
@@ -128,6 +129,7 @@ macro_rules! forward_call {
($self:expr, $method:ident $(, $param:expr)*) => {
match $self.as_mut().project() {
ProxyHttpStreamProj::Http(stream) => stream.$method($($param),*),
#[cfg(any(feature = "local-http-native-tls", feature = "local-http-rustls"))]
ProxyHttpStreamProj::Https(stream, ..) => stream.$method($($param),*),
}
};
@@ -12,6 +12,7 @@ pub mod mon_socket;
pub mod mon_stream;
pub mod packet_window;
pub mod socks5_client;
pub mod outbound_proxy;
pub mod utils;
/// Packet size for all UDP associations' send queue
@@ -0,0 +1,159 @@
//! Shared outbound proxy chain utilities.
use std::io;
use crate::config::{OutboundProxy, OutboundProxyProtocol};
use crate::net::{
http_stream::ProxyHttpStream,
socks5_client::{BoxProxyStream, Socks5TcpClient, socks5_command, socks5_handshake},
};
use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD};
use shadowsocks::{
context::Context,
net::{ConnectOpts, TcpStream as OutboundTcpStream},
relay::socks5::{Address, Command},
};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
pub(crate) fn address_authority(addr: &Address) -> String {
match addr {
Address::SocketAddress(sa) => {
if sa.is_ipv6() {
format!("[{}]:{}", sa.ip(), sa.port())
} else {
sa.to_string()
}
}
Address::DomainNameAddress(host, port) => format!("{host}:{port}"),
}
}
#[cfg(any(feature = "local-http-native-tls", feature = "local-http-rustls"))]
async fn tls_wrap(stream: BoxProxyStream, host: &str) -> io::Result<BoxProxyStream> {
let stream = ProxyHttpStream::connect_https(stream, host).await?;
Ok(Box::new(stream))
}
#[cfg(not(any(feature = "local-http-native-tls", feature = "local-http-rustls")))]
async fn tls_wrap(_stream: BoxProxyStream, _host: &str) -> io::Result<BoxProxyStream> {
Err(io::Error::new(
io::ErrorKind::Unsupported,
"HTTPS outbound proxy requires either local-http-native-tls or local-http-rustls feature",
))
}
pub(crate) async fn http_connect<S>(stream: &mut S, proxy: &OutboundProxy, target: &Address) -> io::Result<()>
where
S: AsyncRead + AsyncWrite + Unpin,
{
let authority = address_authority(target);
let mut request = format!("CONNECT {authority} HTTP/1.1\r\nHost: {authority}\r\nProxy-Connection: Keep-Alive\r\n");
if let Some(ref auth) = proxy.auth {
let encoded = BASE64_STANDARD.encode(format!("{}:{}", auth.username, auth.password));
request.push_str(&format!("Proxy-Authorization: Basic {encoded}\r\n"));
}
request.push_str("\r\n");
stream.write_all(request.as_bytes()).await?;
let mut response = Vec::with_capacity(1024);
let mut buf = [0u8; 1024];
let header_end = loop {
if response.len() > 16 * 1024 {
return Err(io::Error::other("HTTP CONNECT response header is too large"));
}
let n = stream.read(&mut buf).await?;
if n == 0 {
return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
"unexpected EOF while reading HTTP CONNECT response",
));
}
response.extend_from_slice(&buf[..n]);
if let Some(pos) = response.windows(4).position(|w| w == b"\r\n\r\n") {
break pos + 4;
}
};
let header = String::from_utf8_lossy(&response[..header_end]);
let status_line = header.lines().next().unwrap_or_default();
let status_code = status_line
.split_whitespace()
.nth(1)
.and_then(|code| code.parse::<u16>().ok())
.ok_or_else(|| io::Error::other(format!("invalid HTTP CONNECT response: {status_line}")))?;
if status_code != 200 {
return Err(io::Error::other(format!(
"HTTP CONNECT proxy rejected tunnel with status {status_code}: {status_line}"
)));
}
Ok(())
}
pub(crate) async fn connect_via_proxy<S>(stream: &mut S, proxy: &OutboundProxy, target: &Address) -> io::Result<()>
where
S: AsyncRead + AsyncWrite + Unpin,
{
match proxy.protocol {
OutboundProxyProtocol::Socks5 => {
let auth = proxy
.auth
.as_ref()
.map(|auth| (auth.username.as_bytes(), auth.password.as_bytes()));
socks5_handshake(stream, auth).await.map_err(io::Error::other)?;
socks5_command(stream, Command::TcpConnect, target.clone())
.await
.map_err(io::Error::other)?;
Ok(())
}
OutboundProxyProtocol::Http | OutboundProxyProtocol::Https => http_connect(stream, proxy, target).await,
}
}
pub(crate) async fn connect_chain_with_opts<A>(
context: &Context,
addr: A,
proxies: &[OutboundProxy],
opts: &ConnectOpts,
) -> Result<Socks5TcpClient, io::Error>
where
A: Into<Address>,
{
let Some(first_proxy) = proxies.first() else {
return Err(io::Error::new(
io::ErrorKind::InvalidInput,
"empty outbound proxy chain",
));
};
let target_addr = addr.into();
let first_proxy_addr = first_proxy.address();
let first_stream = OutboundTcpStream::connect_remote_with_opts(context, &first_proxy_addr, opts).await?;
let local_addr = first_stream.local_addr()?;
let mut stream: BoxProxyStream = Box::new(first_stream);
if first_proxy.protocol == OutboundProxyProtocol::Https {
stream = tls_wrap(stream, &first_proxy.host).await?;
}
for (idx, proxy) in proxies.iter().enumerate() {
if idx > 0 && proxy.protocol == OutboundProxyProtocol::Https {
stream = tls_wrap(stream, &proxy.host).await?;
}
let next_target = proxies
.get(idx + 1)
.map(OutboundProxy::address)
.unwrap_or_else(|| target_addr.clone());
connect_via_proxy(&mut stream, proxy, &next_target).await?;
}
Ok(Socks5TcpClient::from_stream(stream, local_addr))
}
@@ -1,4 +1,4 @@
//! TCP client for outbound proxy connections
//! TCP client for SOCKS5 outbound connections
use std::{
io,
@@ -7,51 +7,23 @@ use std::{
task::{self, Poll},
};
use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD};
use log::trace;
use shadowsocks::{
context::Context,
net::{ConnectOpts, TcpStream as OutboundTcpStream},
relay::socks5::{
self, Address, Command, Error, HandshakeRequest, HandshakeResponse, PasswdAuthRequest, PasswdAuthResponse, Reply,
TcpRequestHeader, TcpResponseHeader,
},
use shadowsocks::relay::socks5::{
self, Address, Command, Error, HandshakeRequest, HandshakeResponse, PasswdAuthRequest, PasswdAuthResponse, Reply,
TcpRequestHeader, TcpResponseHeader,
};
use tokio::{
io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf},
io::{AsyncRead, AsyncWrite, ReadBuf},
net::{TcpStream, ToSocketAddrs},
};
use crate::{
config::{OutboundProxy, OutboundProxyAuth, OutboundProxyProtocol},
net::http_stream::ProxyHttpStream,
};
trait ProxyStream: AsyncRead + AsyncWrite + Send + Unpin {}
pub(crate) trait ProxyStream: AsyncRead + AsyncWrite + Send + Unpin {}
impl<T> ProxyStream for T where T: AsyncRead + AsyncWrite + Send + Unpin {}
type BoxProxyStream = Box<dyn ProxyStream>;
pub(crate) type BoxProxyStream = Box<dyn ProxyStream>;
pub(crate) fn address_authority(addr: &Address) -> String {
match addr {
Address::SocketAddress(sa) => {
if sa.is_ipv6() {
format!("[{}]:{}", sa.ip(), sa.port())
} else {
sa.to_string()
}
}
Address::DomainNameAddress(host, port) => format!("{host}:{port}"),
}
}
async fn tls_wrap(stream: BoxProxyStream, host: &str) -> io::Result<BoxProxyStream> {
let stream = ProxyHttpStream::connect_https(stream, host).await?;
Ok(Box::new(stream))
}
pub(crate) async fn socks5_handshake<S>(stream: &mut S, auth: Option<&OutboundProxyAuth>) -> Result<(), Error>
pub(crate) async fn socks5_handshake<S>(stream: &mut S, auth: Option<(&[u8], &[u8])>) -> Result<(), Error>
where
S: AsyncRead + AsyncWrite + Unpin,
{
@@ -70,13 +42,13 @@ where
match hsp.chosen_method {
socks5::SOCKS5_AUTH_METHOD_NONE => Ok(()),
socks5::SOCKS5_AUTH_METHOD_PASSWORD => {
let auth = auth.ok_or_else(|| {
let (username, password) = auth.ok_or_else(|| {
io::Error::other(
"SOCKS5 proxy requires username/password authentication, but no credentials were provided",
)
})?;
let req = PasswdAuthRequest::new(auth.username.as_bytes(), auth.password.as_bytes());
let req = PasswdAuthRequest::new(username, password);
req.write_to(stream).await?;
let rsp = PasswdAuthResponse::read_from(stream).await?;
@@ -114,76 +86,7 @@ where
}
}
pub(crate) async fn http_connect<S>(stream: &mut S, proxy: &OutboundProxy, target: &Address) -> io::Result<()>
where
S: AsyncRead + AsyncWrite + Unpin,
{
let authority = address_authority(target);
let mut request = format!("CONNECT {authority} HTTP/1.1\r\nHost: {authority}\r\nProxy-Connection: Keep-Alive\r\n");
if let Some(ref auth) = proxy.auth {
let encoded = BASE64_STANDARD.encode(format!("{}:{}", auth.username, auth.password));
request.push_str(&format!("Proxy-Authorization: Basic {encoded}\r\n"));
}
request.push_str("\r\n");
stream.write_all(request.as_bytes()).await?;
let mut response = Vec::with_capacity(1024);
let mut buf = [0u8; 1024];
let header_end = loop {
if response.len() > 16 * 1024 {
return Err(io::Error::other("HTTP CONNECT response header is too large"));
}
let n = stream.read(&mut buf).await?;
if n == 0 {
return Err(io::Error::new(
io::ErrorKind::UnexpectedEof,
"unexpected EOF while reading HTTP CONNECT response",
));
}
response.extend_from_slice(&buf[..n]);
if let Some(pos) = response.windows(4).position(|w| w == b"\r\n\r\n") {
break pos + 4;
}
};
let header = String::from_utf8_lossy(&response[..header_end]);
let status_line = header.lines().next().unwrap_or_default();
let status_code = status_line
.split_whitespace()
.nth(1)
.and_then(|code| code.parse::<u16>().ok())
.ok_or_else(|| io::Error::other(format!("invalid HTTP CONNECT response: {status_line}")))?;
if status_code != 200 {
return Err(io::Error::other(format!(
"HTTP CONNECT proxy rejected tunnel with status {status_code}: {status_line}"
)));
}
Ok(())
}
async fn connect_via_proxy(stream: &mut BoxProxyStream, proxy: &OutboundProxy, target: &Address) -> io::Result<()> {
match proxy.protocol {
OutboundProxyProtocol::Socks5 => {
socks5_handshake(stream, proxy.auth.as_ref())
.await
.map_err(io::Error::other)?;
socks5_command(stream, Command::TcpConnect, target.clone())
.await
.map_err(io::Error::other)?;
Ok(())
}
OutboundProxyProtocol::Http | OutboundProxyProtocol::Https => http_connect(stream, proxy, target).await,
}
}
/// TCP client for outbound proxy chains
/// TCP client for SOCKS5 outbound connections.
pub struct Socks5TcpClient {
stream: BoxProxyStream,
local_addr: SocketAddr,
@@ -207,75 +110,8 @@ impl Socks5TcpClient {
})
}
/// Connects to `addr` via an outbound proxy chain
pub async fn connect_chain<A>(addr: A, proxies: &[OutboundProxy]) -> Result<Self, Error>
where
A: Into<Address>,
{
let Some(first_proxy) = proxies.first() else {
return Err(io::Error::new(io::ErrorKind::InvalidInput, "empty outbound proxy chain").into());
};
let target_addr = addr.into();
let first_stream = TcpStream::connect((first_proxy.host.as_str(), first_proxy.port)).await?;
let local_addr = first_stream.local_addr()?;
let mut stream: BoxProxyStream = Box::new(first_stream);
for (idx, proxy) in proxies.iter().enumerate() {
let next_target = proxies
.get(idx + 1)
.map(OutboundProxy::address)
.unwrap_or_else(|| target_addr.clone());
if proxy.protocol == OutboundProxyProtocol::Https {
stream = tls_wrap(stream, &proxy.host).await.map_err(Error::from)?;
}
connect_via_proxy(&mut stream, proxy, &next_target)
.await
.map_err(Error::from)?;
}
Ok(Self { stream, local_addr })
}
/// Connects to `addr` via an outbound proxy chain, using shadowsocks connect options
/// for the first outbound hop.
pub async fn connect_chain_with_opts<A>(
context: &Context,
addr: A,
proxies: &[OutboundProxy],
opts: &ConnectOpts,
) -> Result<Self, Error>
where
A: Into<Address>,
{
let Some(first_proxy) = proxies.first() else {
return Err(io::Error::new(io::ErrorKind::InvalidInput, "empty outbound proxy chain").into());
};
let target_addr = addr.into();
let first_proxy_addr = first_proxy.address();
let first_stream = OutboundTcpStream::connect_remote_with_opts(context, &first_proxy_addr, opts).await?;
let local_addr = first_stream.local_addr()?;
let mut stream: BoxProxyStream = Box::new(first_stream);
for (idx, proxy) in proxies.iter().enumerate() {
let next_target = proxies
.get(idx + 1)
.map(OutboundProxy::address)
.unwrap_or_else(|| target_addr.clone());
if proxy.protocol == OutboundProxyProtocol::Https {
stream = tls_wrap(stream, &proxy.host).await.map_err(Error::from)?;
}
connect_via_proxy(&mut stream, proxy, &next_target)
.await
.map_err(Error::from)?;
}
Ok(Self { stream, local_addr })
pub(crate) fn from_stream(stream: BoxProxyStream, local_addr: SocketAddr) -> Self {
Self { stream, local_addr }
}
/// UDP Associate `addr` via SOCKS5 `proxy`
@@ -291,7 +127,13 @@ impl Socks5TcpClient {
socks5_handshake(&mut s, None).await?;
let hp = socks5_command(&mut s, Command::UdpAssociate, addr).await?;
Ok((Self { stream: Box::new(s), local_addr }, hp.address))
Ok((
Self {
stream: Box::new(s),
local_addr,
},
hp.address,
))
}
/// Returns the local socket address of the underlying connection
+13 -5
View File
@@ -3,7 +3,7 @@
use std::{io, net::SocketAddr, sync::Arc, time::Duration};
use futures::future;
use log::trace;
use log::{info, trace};
use shadowsocks::net::{AcceptOpts, ConnectOpts, UdpSocketOpts};
use crate::{
@@ -110,6 +110,7 @@ pub async fn run(config: Config) -> io::Result<()> {
for inst in config.server {
let svr_cfg = inst.config;
let svr_enable_udp = svr_cfg.mode().enable_udp();
let mut server_builder = ServerBuilder::new(svr_cfg);
if let Some(ref r) = resolver {
@@ -141,10 +142,17 @@ pub async fn run(config: Config) -> io::Result<()> {
connect_opts.udp.allow_fragmentation = udp_allow_fragmentation;
}
if !inst.outbound_proxy.is_empty() {
server_builder.set_outbound_proxies(inst.outbound_proxy.clone());
} else if !config.outbound_proxy.is_empty() {
server_builder.set_outbound_proxies(config.outbound_proxy.clone());
let has_outbound_proxies = !inst.outbound_proxy.is_empty() || !config.outbound_proxy.is_empty();
if has_outbound_proxies {
server_builder.set_outbound_proxies(if !inst.outbound_proxy.is_empty() {
inst.outbound_proxy.clone()
} else {
config.outbound_proxy.clone()
});
}
if has_outbound_proxies && svr_enable_udp {
info!("outbound proxy chain only supports TCP; UDP traffic may not be proxied and may behave unexpectedly");
}
server_builder.set_connect_opts(connect_opts);
@@ -23,7 +23,7 @@ use tokio::{
time,
};
use crate::net::{MonProxyStream, Socks5TcpClient, utils::ignore_until_end};
use crate::net::{outbound_proxy::connect_chain_with_opts, MonProxyStream, Socks5TcpClient, utils::ignore_until_end};
use super::context::ServiceContext;
@@ -253,7 +253,7 @@ impl TcpServerClient {
)
.await
.map(RemoteStream::Direct),
proxies => Socks5TcpClient::connect_chain_with_opts(
proxies => connect_chain_with_opts(
self.context.context_ref(),
target_addr.clone(),
proxies,