diff --git a/crates/shadowsocks-service/src/local/net/udp/association.rs b/crates/shadowsocks-service/src/local/net/udp/association.rs index e93f99c7..c070145c 100644 --- a/crates/shadowsocks-service/src/local/net/udp/association.rs +++ b/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 { + 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, + opts: ConnectOpts, +} + +impl TcpDialer for LocalTcpDialer { + async fn dial(&self, addr: &Address) -> io::Result { + 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), + Chained(MonProxySocket), +} + +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)> { + 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, bypassed_ipv6_socket: Option, - proxied_socket: Option>, + proxied_socket: Option, keepalive_tx: mpsc::Sender, keepalive_flag: bool, balancer: PingBalancer, @@ -410,7 +497,7 @@ where #[inline] async fn receive_from_proxied_opt( - socket: &Option>, + socket: &Option, buf: &mut Vec, ) -> io::Result<(usize, Address, Option)> { 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) } }; diff --git a/crates/shadowsocks-service/src/net/mod.rs b/crates/shadowsocks-service/src/net/mod.rs index 0fdf336c..3daace6e 100644 --- a/crates/shadowsocks-service/src/net/mod.rs +++ b/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, }, }; diff --git a/crates/shadowsocks-service/src/net/outbound/chain.rs b/crates/shadowsocks-service/src/net/outbound/chain.rs index a3f0fda5..f3c465de 100644 --- a/crates/shadowsocks-service/src/net/outbound/chain.rs +++ b/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( + hops: &[OutboundProxyHop], + hop_index: usize, + dialer: &D, +) -> io::Result +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, diff --git a/crates/shadowsocks-service/src/net/outbound/mod.rs b/crates/shadowsocks-service/src/net/outbound/mod.rs index 5b64c83f..8e45a0ed 100644 --- a/crates/shadowsocks-service/src/net/outbound/mod.rs +++ b/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( + &self, + context: &SharedContext, + dialer: &D, + connect_opts: &ConnectOpts, + target: Address, + ) -> io::Result + where + D: TcpDialer + Sync, + { + OutboundProxyDatagram::associate(self, context, dialer, connect_opts, target).await + } } fn hop_from_config(proxy: &OutboundProxy) -> OutboundProxyHop { diff --git a/crates/shadowsocks-service/src/net/outbound/stream.rs b/crates/shadowsocks-service/src/net/outbound/stream.rs index 054a3b5a..20d25a65 100644 --- a/crates/shadowsocks-service/src/net/outbound/stream.rs +++ b/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 { + 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. diff --git a/crates/shadowsocks-service/src/net/outbound/udp.rs b/crates/shadowsocks-service/src/net/outbound/udp.rs new file mode 100644 index 00000000..b2ba3290 --- /dev/null +++ b/crates/shadowsocks-service/src/net/outbound/udp.rs @@ -0,0 +1,343 @@ +//! Outbound UDP relay through a SOCKS5-only proxy chain. +//! +//! Establishes a SOCKS5 `UDP ASSOCIATE` against every hop in the +//! configured chain, then transports each datagram by nesting one +//! [`UdpAssociateHeader`] per hop. The resulting [`OutboundProxyDatagram`] +//! implements [`DatagramSocket`] / [`DatagramSend`] / [`DatagramReceive`] +//! so it can be plugged directly into [`ProxySocket::from_socket`] in +//! place of a raw UDP socket. +//! +//! Supported only when **every** hop is SOCKS5 (see +//! [`OutboundProxyClient::supports_udp`]); the dispatcher in +//! `local::net::udp::association` is responsible for picking this path +//! vs. the direct one. + +use std::{ + io::{self, Cursor}, + net::SocketAddr, + sync::Mutex, + task::{Context, Poll, ready}, +}; + +use bytes::{BufMut, BytesMut}; +use log::{debug, trace}; +use shadowsocks::{ + context::SharedContext, + lookup_then, + net::{AddrFamily, ConnectOpts, TcpStream as ShadowTcpStream, UdpSocket as ShadowUdpSocket}, + relay::{ + socks5::{Address, UdpAssociateHeader}, + udprelay::{DatagramReceive, DatagramSend, DatagramSocket}, + }, +}; +use tokio::io::ReadBuf; + +use super::{ + OutboundProxyClient, OutboundProxyKind, TcpDialer, chain::connect_chain_for_udp_associate, + socks5::Socks5Negotiator, +}; + +/// One SOCKS5 UDP relay hop. The `assoc_tcp` keep-alive connection must +/// outlive every datagram exchanged through this hop (RFC 1928). +struct Socks5UdpRelay { + /// Keep-alive TCP control connection. We hold the bare [`ShadowTcpStream`] + /// rather than an [`super::OutboundProxyStream`] so the relay (and + /// therefore [`OutboundProxyDatagram`]) is `Sync` — the chain builder + /// never produces a non-TCP variant for SOCKS5-only chains, so this + /// downcast always succeeds. + #[allow(dead_code)] + assoc_tcp: ShadowTcpStream, + relay_addr: Address, + relay_socket_addr: Option, +} + +impl Socks5UdpRelay { + fn relay_addr(&self) -> &Address { + &self.relay_addr + } +} + +/// Multi-hop SOCKS5 UDP relay implementing the +/// [`DatagramSocket`] / [`DatagramSend`] / [`DatagramReceive`] triple. +/// +/// The chain is built once at association time. Each `send` then nests +/// `hops` SOCKS5 UDP headers around the upper-layer payload (the inner- +/// most header carries the **final target** address — for sslocal / +/// ssserver this is always the configured `target` baked in at +/// construction time, namely the ss-server's UDP external address). +pub struct OutboundProxyDatagram { + socket: ShadowUdpSocket, + relays: Vec, + /// Inner-most target address — recorded once at construction time + /// and used as the SOCKS5 UDP innermost destination on every send. + /// For sslocal / ssserver outbound chains this is the ss-server's + /// UDP external address. + target: Address, + /// Address the local UDP socket is `connect()`ed to (the first hop's + /// relay address). Reported to upper layers as the "peer" / source + /// from which `recv_from` returns datagrams. + first_hop_addr: SocketAddr, + /// Reusable receive scratch buffer (header stripping requires reading + /// the full datagram before we know the payload offset). + recv_scratch: Mutex>, +} + +impl OutboundProxyDatagram { + /// Establish UDP relays through every hop of `client`. + /// + /// * `target` is the inner-most destination (typically the ss-server's + /// UDP external address). + /// * `dialer` is used to dial all TCP control connections. + /// * `connect_opts` configures the local UDP socket. + pub async fn associate( + client: &OutboundProxyClient, + context: &SharedContext, + dialer: &D, + connect_opts: &ConnectOpts, + target: Address, + ) -> io::Result + where + D: TcpDialer + Sync, + { + let hops = client.hops(); + if hops.is_empty() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "empty outbound proxy chain", + )); + } + for hop in hops { + if !matches!(hop.kind, OutboundProxyKind::Socks5 { .. }) { + return Err(io::Error::new( + io::ErrorKind::Unsupported, + "outbound UDP relay requires every hop to be SOCKS5", + )); + } + } + + // Bind the local UDP socket using the shadowsocks helper so + // `ConnectOpts` (bind address, fwmark, ...) is honoured. + let socket = ShadowUdpSocket::connect_any_with_opts(AddrFamily::Ipv4, connect_opts).await?; + + let local_udp_addr = socket.local_addr()?; + trace!("outbound udp local socket bound to {}", local_udp_addr); + + let mut relays: Vec = Vec::with_capacity(hops.len()); + + // The "announce" address tells the next hop the source it should + // expect datagrams from. For the first hop this is the local UDP + // socket; for later hops it is the relay address granted by the + // previous hop. + let mut announce: Address = Address::from(local_udp_addr); + + for (idx, hop) in hops.iter().enumerate() { + // Build a TCP control connection that *terminates* at `hop` + // (i.e. previous hops are traversed via SOCKS5 TcpConnect). + let mut tcp = connect_chain_for_udp_associate(hops, idx, dialer).await?; + + let auth = match &hop.kind { + OutboundProxyKind::Socks5 { auth } => auth, + _ => unreachable!("guarded above"), + }; + + let resp = Socks5Negotiator::establish_udp_associate(&mut tcp, announce.clone(), auth) + .await + .map_err(io::Error::other)?; + + // For a SOCKS5-only chain the byte stream is always still raw + // TCP after the handshake (no TLS / HTTP layers were + // applied). Recover the underlying TcpStream so the relay + // struct stays `Sync`. + let tcp = tcp.try_into_tcp().map_err(|_| { + io::Error::other( + "internal error: outbound UDP relay produced a non-TCP keep-alive stream", + ) + })?; + + let relay_addr = resp.address; + let relay_socket_addr = match relay_addr { + Address::SocketAddress(sa) => Some(sa), + Address::DomainNameAddress(ref name, port) => { + // Resolve domain-name relay addresses via the + // shadowsocks shared resolver — matches what a direct + // ProxySocket::connect_with_opts does. + let (sa, _) = lookup_then!(context, name, port, |sa| { + Ok::(sa) + })?; + Some(sa) + } + }; + + trace!( + "outbound udp hop {}: relay_addr={} (resolved={:?}), announce={}", + idx, relay_addr, relay_socket_addr, announce + ); + + // Next hop's announce = this hop's relay address. + announce = match relay_socket_addr { + Some(sa) => Address::from(sa), + None => relay_addr.clone(), + }; + + relays.push(Socks5UdpRelay { + assoc_tcp: tcp, + relay_addr, + relay_socket_addr, + }); + } + + let first_hop_addr = relays[0] + .relay_socket_addr + .ok_or_else(|| io::Error::other("first SOCKS5 UDP relay returned an unresolved domain"))?; + + // Connect the local UDP socket to the first hop so subsequent + // sends can use `send` instead of `send_to`. + socket.connect(first_hop_addr).await?; + + debug!( + "outbound udp chain established: {} hop(s), first_hop={}, target={}", + relays.len(), + first_hop_addr, + target, + ); + + Ok(Self { + socket, + relays, + target, + first_hop_addr, + recv_scratch: Mutex::new(Vec::new()), + }) + } + + /// Number of SOCKS5 hops the chain consists of. + pub fn hop_count(&self) -> usize { + self.relays.len() + } + + /// Wrap `payload` in N nested SOCKS5 UDP headers, where N is the + /// number of hops. The inner-most header carries `self.target`. + fn encode(&self, payload: &[u8]) -> BytesMut { + let n = self.relays.len(); + + // Header[i] addr = relays[i+1].relay_addr() for i < n-1, or + // self.target for i == n-1. + let mut total = payload.len(); + for i in 0..n { + let addr = if i + 1 < n { + self.relays[i + 1].relay_addr() + } else { + &self.target + }; + total += UdpAssociateHeader::new(0, addr.clone()).serialized_len(); + } + + let mut buf = BytesMut::with_capacity(total); + for i in 0..n { + let addr = if i + 1 < n { + self.relays[i + 1].relay_addr().clone() + } else { + self.target.clone() + }; + UdpAssociateHeader::new(0, addr).write_to_buf(&mut buf); + } + buf.put_slice(payload); + buf + } + + /// Strip N nested SOCKS5 UDP headers from a freshly received datagram. + /// Returns the inner-most reported source address and the payload + /// offset. + fn decode(&self, recv_buf: &[u8]) -> io::Result<(Address, usize)> { + let n = self.relays.len(); + let mut cur = Cursor::new(recv_buf); + let mut last_addr: Option
= None; + + for _ in 0..n { + // `UdpAssociateHeader::read_from` is async over an + // `AsyncRead`. Cursor's `AsyncRead` impl yields synchronously, + // so polling the future once always completes. + let fut = std::pin::pin!(UdpAssociateHeader::read_from(&mut cur)); + match futures::FutureExt::now_or_never(fut) { + Some(Ok(h)) => last_addr = Some(h.address), + Some(Err(err)) => { + return Err(io::Error::other(format!( + "outbound udp: failed to parse SOCKS5 UDP header: {err}" + ))); + } + None => unreachable!("Cursor read_from should be synchronous"), + } + } + + let pos = cur.position() as usize; + Ok((last_addr.expect("relays is non-empty"), pos)) + } +} + +impl DatagramSocket for OutboundProxyDatagram { + fn local_addr(&self) -> io::Result { + self.socket.local_addr() + } +} + +impl DatagramSend for OutboundProxyDatagram { + fn poll_send(&self, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + let wire = self.encode(buf); + let sent = ready!(self.socket.poll_send(cx, &wire))?; + if sent == wire.len() { + // Report the upper-layer payload size (matches what + // ProxySocket expects from a direct UDP send). + Poll::Ready(Ok(buf.len())) + } else { + Poll::Ready(Err(io::Error::from(io::ErrorKind::WriteZero))) + } + } + + fn poll_send_to(&self, cx: &mut Context<'_>, buf: &[u8], _target: SocketAddr) -> Poll> { + // The first-hop relay is what we actually send to on the wire; + // the upper layer's `target` (typically the ss-server) is already + // baked into the inner-most SOCKS5 UDP header via `self.target`. + let _ = _target; + self.poll_send(cx, buf) + } + + fn poll_send_ready(&self, cx: &mut Context<'_>) -> Poll> { + self.socket.poll_send_ready(cx) + } +} + +impl DatagramReceive for OutboundProxyDatagram { + fn poll_recv(&self, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + // Receive into our scratch, strip headers, then copy payload into + // the caller's buffer. + let mut scratch = self.recv_scratch.lock().expect("recv scratch poisoned"); + let needed = buf.remaining() + 1024; + if scratch.len() < needed { + scratch.resize(needed, 0); + } + let mut tmp = ReadBuf::new(&mut scratch); + ready!(self.socket.poll_recv(cx, &mut tmp))?; + let n = tmp.filled().len(); + let (_inner_src, pos) = self.decode(&scratch[..n])?; + let payload_len = n - pos; + if payload_len > buf.remaining() { + return Poll::Ready(Err(io::Error::from(io::ErrorKind::InvalidData))); + } + buf.put_slice(&scratch[pos..n]); + Poll::Ready(Ok(())) + } + + fn poll_recv_from(&self, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + // The upper layer wants the source address. We report the first- + // hop relay we actually received from (the only socket-level + // address available); the inner-most SOCKS5 source is discarded + // because the upper layer will decrypt and parse the ss-server's + // own packet header anyway. + ready!(self.poll_recv(cx, buf))?; + Poll::Ready(Ok(self.first_hop_addr)) + } + + fn poll_recv_ready(&self, cx: &mut Context<'_>) -> Poll> { + self.socket.poll_recv_ready(cx) + } +}