feat(shadowsocks): DatagramTransport & DatagramTransportExt

ProxySocket and MonProxySocket accepts socket as a generic type that
implements DatagramTransport.
This commit is contained in:
zonyitoo
2024-09-21 00:03:50 +08:00
parent 4e295818e5
commit 60d0576b26
8 changed files with 250 additions and 149 deletions
@@ -61,7 +61,7 @@ pub enum DnsClient {
stream: ProxyClientStream<MonProxyStream<ShadowTcpStream>>,
},
UdpRemote {
socket: MonProxySocket,
socket: MonProxySocket<ShadowUdpSocket>,
ns: Address,
control: UdpSocketControlData,
server_windows: LruCache<u64, PacketWindowFilter>,
@@ -217,7 +217,7 @@ where
peer_addr: SocketAddr,
bypassed_ipv4_socket: Option<ShadowUdpSocket>,
bypassed_ipv6_socket: Option<ShadowUdpSocket>,
proxied_socket: Option<MonProxySocket>,
proxied_socket: Option<MonProxySocket<ShadowUdpSocket>>,
keepalive_tx: mpsc::Sender<SocketAddr>,
keepalive_flag: bool,
balancer: PingBalancer,
@@ -409,7 +409,7 @@ where
#[inline]
async fn receive_from_proxied_opt(
socket: &Option<MonProxySocket>,
socket: &Option<MonProxySocket<ShadowUdpSocket>>,
buf: &mut Vec<u8>,
) -> io::Result<(usize, Address, Option<UdpSocketControlData>)> {
match *socket {
@@ -3,24 +3,44 @@
use std::{io, net::SocketAddr, sync::Arc};
use shadowsocks::{
relay::{socks5::Address, udprelay::options::UdpSocketControlData},
relay::{
socks5::Address,
udprelay::{options::UdpSocketControlData, DatagramTransport},
},
ProxySocket,
};
use super::flow::FlowStat;
/// Monitored `ProxySocket`
pub struct MonProxySocket {
socket: ProxySocket,
pub struct MonProxySocket<S> {
socket: ProxySocket<S>,
flow_stat: Arc<FlowStat>,
}
impl MonProxySocket {
impl<S> MonProxySocket<S> {
/// Create a new socket with flow monitor
pub fn from_socket(socket: ProxySocket, flow_stat: Arc<FlowStat>) -> MonProxySocket {
pub fn from_socket(socket: ProxySocket<S>, flow_stat: Arc<FlowStat>) -> MonProxySocket<S> {
MonProxySocket { socket, flow_stat }
}
/// Get the underlying `ProxySocket<S>` immutable reference
#[inline]
pub fn get_ref(&self) -> &ProxySocket<S> {
&self.socket
}
/// Get the flow statistic data
#[inline]
pub fn flow_stat(&self) -> &FlowStat {
&self.flow_stat
}
}
impl<S> MonProxySocket<S>
where
S: DatagramTransport,
{
/// Send a UDP packet to addr through proxy
#[inline]
pub async fn send(&self, addr: &Address, payload: &[u8]) -> io::Result<()> {
@@ -125,14 +145,4 @@ impl MonProxySocket {
Ok((n, peer_addr, addr, control))
}
#[inline]
pub fn get_ref(&self) -> &ProxySocket {
&self.socket
}
#[inline]
pub fn flow_stat(&self) -> &FlowStat {
&self.flow_stat
}
}
@@ -17,7 +17,10 @@ use shadowsocks::{
config::ServerUser,
crypto::CipherCategory,
lookup_then,
net::{get_ip_stack_capabilities, AcceptOpts, AddrFamily, UdpSocket as OutboundUdpSocket},
net::{
get_ip_stack_capabilities, AcceptOpts, AddrFamily, UdpSocket as OutboundUdpSocket,
UdpSocket as InboundUdpSocket,
},
relay::{
socks5::Address,
udprelay::{options::UdpSocketControlData, ProxySocket, MAXIMUM_UDP_PAYLOAD_SIZE},
@@ -87,7 +90,7 @@ pub struct UdpServer {
keepalive_tx: mpsc::Sender<NatKey>,
keepalive_rx: mpsc::Receiver<NatKey>,
time_to_live: Duration,
listener: Arc<MonProxySocket>,
listener: Arc<MonProxySocket<InboundUdpSocket>>,
svr_cfg: ServerConfig,
}
@@ -276,7 +279,7 @@ impl UdpServer {
async fn recv_one_packet(
context: &ServiceContext,
l: &MonProxySocket,
l: &MonProxySocket<InboundUdpSocket>,
buffer: &mut [u8],
) -> Option<(usize, SocketAddr, Address, Option<UdpSocketControlData>)> {
let (n, peer_addr, target_addr, control) = match l.recv_from_with_ctrl(buffer).await {
@@ -316,7 +319,7 @@ impl UdpServer {
async fn send_packet(
&mut self,
listener: &Arc<MonProxySocket>,
listener: &Arc<MonProxySocket<InboundUdpSocket>>,
peer_addr: SocketAddr,
target_addr: Address,
control: Option<UdpSocketControlData>,
@@ -394,7 +397,7 @@ impl Drop for UdpAssociation {
impl UdpAssociation {
fn new_association(
context: Arc<ServiceContext>,
inbound: Arc<MonProxySocket>,
inbound: Arc<MonProxySocket<InboundUdpSocket>>,
peer_addr: SocketAddr,
keepalive_tx: mpsc::Sender<NatKey>,
) -> UdpAssociation {
@@ -405,7 +408,7 @@ impl UdpAssociation {
#[cfg(feature = "aead-cipher-2022")]
fn new_session(
context: Arc<ServiceContext>,
inbound: Arc<MonProxySocket>,
inbound: Arc<MonProxySocket<InboundUdpSocket>>,
peer_addr: SocketAddr,
keepalive_tx: mpsc::Sender<NatKey>,
client_session_id: u64,
@@ -447,7 +450,7 @@ struct UdpAssociationContext {
outbound_ipv6_socket: Option<OutboundUdpSocket>,
keepalive_tx: mpsc::Sender<NatKey>,
keepalive_flag: bool,
inbound: Arc<MonProxySocket>,
inbound: Arc<MonProxySocket<InboundUdpSocket>>,
// AEAD 2022
client_session: Option<ClientSessionContext>,
server_session_id: u64,
@@ -472,7 +475,7 @@ fn generate_server_session_id() -> u64 {
impl UdpAssociationContext {
fn create(
context: Arc<ServiceContext>,
inbound: Arc<MonProxySocket>,
inbound: Arc<MonProxySocket<InboundUdpSocket>>,
peer_addr: SocketAddr,
keepalive_tx: mpsc::Sender<NatKey>,
client_session_id: Option<u64>,
+155 -49
View File
@@ -1,88 +1,194 @@
use async_trait::async_trait;
use std::{
io::Result,
future::Future,
io,
net::SocketAddr,
ops::Deref,
pin::Pin,
task::{Context, Poll},
};
use futures::ready;
use pin_project::pin_project;
use tokio::io::ReadBuf;
use crate::net::UdpSocket;
/// a trait for datagram transport that wraps around a tokio `UdpSocket`
#[async_trait]
pub trait DatagramTransport: Send + Sync + std::fmt::Debug {
async fn recv(&self, buf: &mut [u8]) -> Result<usize>;
async fn recv_from(&self, buf: &mut [u8]) -> Result<(usize, SocketAddr)>;
pub trait DatagramTransport {
/// Local binded address
fn local_addr(&self) -> io::Result<SocketAddr>;
async fn send(&self, buf: &[u8]) -> Result<usize>;
async fn send_to(&self, buf: &[u8], target: SocketAddr) -> Result<usize>;
/// `recv` data into `buf`
fn poll_recv(&self, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>>;
/// `recv` data into `buf` with source address
fn poll_recv_from(&self, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<SocketAddr>>;
/// Check if the underlying I/O object is ready for `recv`
fn poll_recv_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>>;
fn poll_recv(&self, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<Result<()>>;
fn poll_recv_from(&self, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<Result<SocketAddr>>;
fn poll_recv_ready(&self, cx: &mut Context<'_>) -> Poll<Result<()>>;
fn poll_send(&self, cx: &mut Context<'_>, buf: &[u8]) -> Poll<Result<usize>>;
fn poll_send_to(&self, cx: &mut Context<'_>, buf: &[u8], target: SocketAddr) -> Poll<Result<usize>>;
fn poll_send_ready(&self, cx: &mut Context<'_>) -> Poll<Result<()>>;
fn local_addr(&self) -> Result<SocketAddr>;
#[cfg(unix)]
fn as_raw_fd(&self) -> std::os::fd::RawFd;
/// `send` data with `buf`, returning the sent bytes
fn poll_send(&self, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>>;
/// `send` data with `buf` to `target`, returning the sent bytes
fn poll_send_to(&self, cx: &mut Context<'_>, buf: &[u8], target: SocketAddr) -> Poll<io::Result<usize>>;
/// Check if the underlying I/O object is ready for `send`
fn poll_send_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>>;
}
#[async_trait]
impl DatagramTransport for UdpSocket {
async fn recv(&self, buf: &mut [u8]) -> Result<usize> {
UdpSocket::recv(self, buf).await
}
async fn recv_from(&self, buf: &mut [u8]) -> Result<(usize, SocketAddr)> {
UdpSocket::recv_from(self, buf).await
}
async fn send(&self, buf: &[u8]) -> Result<usize> {
UdpSocket::send(self, buf).await
}
async fn send_to(&self, buf: &[u8], target: SocketAddr) -> Result<usize> {
UdpSocket::send_to(self, buf, target).await
}
fn poll_recv(&self, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<Result<()>> {
fn poll_recv(&self, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
UdpSocket::poll_recv(self, cx, buf)
}
fn poll_recv_from(&self, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<Result<SocketAddr>> {
fn poll_recv_from(&self, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<SocketAddr>> {
UdpSocket::poll_recv_from(self, cx, buf)
}
fn poll_recv_ready(&self, cx: &mut Context<'_>) -> Poll<Result<()>> {
fn poll_recv_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
self.deref().poll_recv_ready(cx)
}
fn poll_send(&self, cx: &mut Context<'_>, buf: &[u8]) -> Poll<Result<usize>> {
fn poll_send(&self, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
UdpSocket::poll_send(self, cx, buf)
}
fn poll_send_to(&self, cx: &mut Context<'_>, buf: &[u8], target: SocketAddr) -> Poll<Result<usize>> {
fn poll_send_to(&self, cx: &mut Context<'_>, buf: &[u8], target: SocketAddr) -> Poll<io::Result<usize>> {
UdpSocket::poll_send_to(self, cx, buf, target)
}
fn poll_send_ready(&self, cx: &mut Context<'_>) -> Poll<Result<()>> {
fn poll_send_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
self.deref().poll_send_ready(cx)
}
fn local_addr(&self) -> Result<SocketAddr> {
fn local_addr(&self) -> io::Result<SocketAddr> {
self.deref().local_addr()
}
}
#[cfg(unix)]
fn as_raw_fd(&self) -> std::os::fd::RawFd {
use std::ops::Deref;
use std::os::fd::AsRawFd;
/// Future for `recv`
#[pin_project]
pub struct RecvFut<'a, S: DatagramTransport + ?Sized> {
#[pin]
io: &'a S,
buf: &'a mut [u8],
}
self.deref().as_raw_fd()
impl<'a, S: DatagramTransport + ?Sized> Future for RecvFut<'a, S> {
type Output = io::Result<usize>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
let mut read_buf = ReadBuf::new(this.buf);
ready!(this.io.poll_recv(cx, &mut read_buf))?;
Ok(read_buf.filled().len()).into()
}
}
/// Future for `recv_from`
#[pin_project]
pub struct RecvFromFut<'a, S: DatagramTransport + ?Sized> {
#[pin]
io: &'a S,
buf: &'a mut [u8],
}
impl<'a, S: DatagramTransport + ?Sized> Future for RecvFromFut<'a, S> {
type Output = io::Result<(usize, SocketAddr)>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
let mut read_buf = ReadBuf::new(this.buf);
let src_addr = ready!(this.io.poll_recv_from(cx, &mut read_buf))?;
Ok((read_buf.filled().len(), src_addr)).into()
}
}
/// Future for `recv_ready`
pub struct RecvReadyFut<'a, S: DatagramTransport + ?Sized> {
io: &'a S,
}
impl<'a, S: DatagramTransport + ?Sized> Future for RecvReadyFut<'a, S> {
type Output = io::Result<()>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.io.poll_recv_ready(cx)
}
}
/// Future for `send`
pub struct SendFut<'a, S: DatagramTransport + ?Sized> {
io: &'a S,
buf: &'a [u8],
}
impl<'a, S: DatagramTransport + ?Sized> Future for SendFut<'a, S> {
type Output = io::Result<usize>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.io.poll_send(cx, self.buf)
}
}
/// Future for `send_to`
pub struct SendToFut<'a, S: DatagramTransport + ?Sized> {
io: &'a S,
target: SocketAddr,
buf: &'a [u8],
}
impl<'a, S: DatagramTransport + ?Sized> Future for SendToFut<'a, S> {
type Output = io::Result<usize>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.io.poll_send_to(cx, self.buf, self.target)
}
}
/// Future for `recv_ready`
pub struct SendReadyFut<'a, S: DatagramTransport + ?Sized> {
io: &'a S,
}
impl<'a, S: DatagramTransport + ?Sized> Future for SendReadyFut<'a, S> {
type Output = io::Result<()>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.io.poll_recv_ready(cx)
}
}
/// Extension methods for `DatagramTransport`
pub trait DatagramTransportExt: DatagramTransport {
/// Async method for `poll_recv`
fn recv<'a, 'b>(&'a self, buf: &'a mut [u8]) -> RecvFut<'a, Self> {
RecvFut { io: self, buf }
}
/// Async method for `poll_recv_from`
fn recv_from<'a, 'b>(&'a self, buf: &'a mut [u8]) -> RecvFromFut<'a, Self> {
RecvFromFut { io: self, buf }
}
/// Async method for `poll_recv_ready`
fn recv_ready<'a>(&'a self) -> RecvReadyFut<'a, Self> {
RecvReadyFut { io: self }
}
/// Async method for `poll_send`
fn send<'a>(&'a self, buf: &'a [u8]) -> SendFut<'a, Self> {
SendFut { io: self, buf }
}
/// Async method for `poll_send_to`
fn send_to<'a>(&'a self, buf: &'a [u8], target: SocketAddr) -> SendToFut<'a, Self> {
SendToFut { io: self, target, buf }
}
/// Async method for `poll_send_ready`
fn send_ready<'a>(&'a self) -> SendReadyFut<'a, Self> {
SendReadyFut { io: self }
}
}
impl<S: DatagramTransport> DatagramTransportExt for S {}
+1 -1
View File
@@ -50,7 +50,7 @@
use std::time::Duration;
pub use self::proxy_socket::ProxySocket;
pub use compat::DatagramTransport;
pub use compat::{DatagramTransport, DatagramTransportExt};
mod aead;
#[cfg(feature = "aead-cipher-2022")]
@@ -23,7 +23,7 @@ use crate::{
};
use super::{
compat::DatagramTransport,
compat::{DatagramTransport, DatagramTransportExt},
crypto_io::{
decrypt_client_payload, decrypt_server_payload, encrypt_client_payload, encrypt_server_payload, ProtocolError,
ProtocolResult,
@@ -73,9 +73,9 @@ pub type ProxySocketResult<T> = Result<T, ProxySocketError>;
/// UDP client for communicating with ShadowSocks' server
#[derive(Debug)]
pub struct ProxySocket {
pub struct ProxySocket<S> {
socket_type: UdpSocketType,
io: Box<dyn DatagramTransport>,
io: S,
method: CipherKind,
key: Box<[u8]>,
send_timeout: Option<Duration>,
@@ -85,9 +85,12 @@ pub struct ProxySocket {
user_manager: Option<Arc<ServerUserManager>>,
}
impl ProxySocket {
impl ProxySocket<ShadowUdpSocket> {
/// Create a client to communicate with Shadowsocks' UDP server (outbound)
pub async fn connect(context: SharedContext, svr_cfg: &ServerConfig) -> ProxySocketResult<ProxySocket> {
pub async fn connect(
context: SharedContext,
svr_cfg: &ServerConfig,
) -> ProxySocketResult<ProxySocket<ShadowUdpSocket>> {
ProxySocket::connect_with_opts(context, svr_cfg, &DEFAULT_CONNECT_OPTS)
.await
.map_err(Into::into)
@@ -98,7 +101,7 @@ impl ProxySocket {
context: SharedContext,
svr_cfg: &ServerConfig,
opts: &ConnectOpts,
) -> ProxySocketResult<ProxySocket> {
) -> ProxySocketResult<ProxySocket<ShadowUdpSocket>> {
// Note: Plugins doesn't support UDP relay
let socket = ShadowUdpSocket::connect_server_with_opts(&context, svr_cfg.udp_external_addr(), opts).await?;
@@ -118,71 +121,11 @@ impl ProxySocket {
))
}
/// Create a `ProxySocket` from a `UdpSocket`
pub fn from_socket<S>(
socket_type: UdpSocketType,
context: SharedContext,
svr_cfg: &ServerConfig,
socket: S,
) -> ProxySocket
where
S: Into<ShadowUdpSocket>,
{
let key = svr_cfg.key().to_vec().into_boxed_slice();
let method = svr_cfg.method();
// NOTE: svr_cfg.timeout() is not for this socket, but for associations.
ProxySocket {
socket_type,
io: Box::new(socket.into()),
method,
key,
send_timeout: None,
recv_timeout: None,
context,
identity_keys: match socket_type {
UdpSocketType::Client => svr_cfg.clone_identity_keys(),
UdpSocketType::Server => Arc::new(Vec::new()),
},
user_manager: match socket_type {
UdpSocketType::Client => None,
UdpSocketType::Server => svr_cfg.clone_user_manager(),
},
}
}
pub fn from_io(
socket_type: UdpSocketType,
context: SharedContext,
svr_cfg: &ServerConfig,
io: Box<dyn DatagramTransport>,
) -> ProxySocket {
let key = svr_cfg.key().to_vec().into_boxed_slice();
let method = svr_cfg.method();
// NOTE: svr_cfg.timeout() is not for this socket, but for associations.
ProxySocket {
socket_type,
io,
method,
key,
send_timeout: None,
recv_timeout: None,
context,
identity_keys: match socket_type {
UdpSocketType::Client => svr_cfg.clone_identity_keys(),
UdpSocketType::Server => Arc::new(Vec::new()),
},
user_manager: match socket_type {
UdpSocketType::Client => None,
UdpSocketType::Server => svr_cfg.clone_user_manager(),
},
}
}
/// Create a `ProxySocket` binding to a specific address (inbound)
pub async fn bind(context: SharedContext, svr_cfg: &ServerConfig) -> ProxySocketResult<ProxySocket> {
pub async fn bind(
context: SharedContext,
svr_cfg: &ServerConfig,
) -> ProxySocketResult<ProxySocket<ShadowUdpSocket>> {
ProxySocket::bind_with_opts(context, svr_cfg, AcceptOpts::default())
.await
.map_err(Into::into)
@@ -193,7 +136,7 @@ impl ProxySocket {
context: SharedContext,
svr_cfg: &ServerConfig,
opts: AcceptOpts,
) -> ProxySocketResult<ProxySocket> {
) -> ProxySocketResult<ProxySocket<ShadowUdpSocket>> {
// Plugins doesn't support UDP
let socket = match svr_cfg.udp_external_addr() {
ServerAddr::SocketAddr(sa) => ShadowUdpSocket::listen_with_opts(sa, opts).await?,
@@ -211,6 +154,41 @@ impl ProxySocket {
socket,
))
}
}
impl<S> ProxySocket<S>
where
S: DatagramTransport,
{
/// Create a `ProxySocket` from a I/O object that impls `DatagramTransport`
pub fn from_socket(
socket_type: UdpSocketType,
context: SharedContext,
svr_cfg: &ServerConfig,
socket: S,
) -> ProxySocket<S> {
let key = svr_cfg.key().to_vec().into_boxed_slice();
let method = svr_cfg.method();
// NOTE: svr_cfg.timeout() is not for this socket, but for associations.
ProxySocket {
socket_type,
io: socket,
method,
key,
send_timeout: None,
recv_timeout: None,
context,
identity_keys: match socket_type {
UdpSocketType::Client => svr_cfg.clone_identity_keys(),
UdpSocketType::Server => Arc::new(Vec::new()),
},
user_manager: match socket_type {
UdpSocketType::Client => None,
UdpSocketType::Server => svr_cfg.clone_user_manager(),
},
}
}
fn encrypt_send_buffer(
&self,
@@ -627,7 +605,10 @@ impl ProxySocket {
}
#[cfg(unix)]
impl AsRawFd for ProxySocket {
impl<S> AsRawFd for ProxySocket<S>
where
S: AsRawFd,
{
/// Retrieve raw fd of the outbound socket
fn as_raw_fd(&self) -> RawFd {
self.io.as_raw_fd()
+2 -1
View File
@@ -8,6 +8,7 @@ use shadowsocks::{
config::{ServerConfig, ServerType},
context::{Context, SharedContext},
crypto::CipherKind,
net::UdpSocket as ShadowUdpSocket,
relay::{socks5::Address, udprelay::ProxySocket},
};
@@ -15,7 +16,7 @@ async fn handle_udp_server_client(
peer_addr: SocketAddr,
remote_addr: Address,
payload: &[u8],
socket: &ProxySocket,
socket: &ProxySocket<ShadowUdpSocket>,
) -> io::Result<()> {
let remote_socket = UdpSocket::bind("0.0.0.0:0").await?;