hyper_util/client/legacy/connect/proxy/socks/
mod.rs

1mod v5;
2pub use v5::{SocksV5, SocksV5Error};
3
4mod v4;
5pub use v4::{SocksV4, SocksV4Error};
6
7use pin_project_lite::pin_project;
8use std::pin::Pin;
9use std::task::{Context, Poll};
10
11use bytes::{Buf, BytesMut};
12
13use hyper::rt::Read;
14
15#[derive(Debug)]
16pub enum SocksError<C> {
17    Inner(C),
18    Io(std::io::Error),
19
20    DnsFailure,
21    MissingHost,
22    MissingPort,
23
24    V4(SocksV4Error),
25    V5(SocksV5Error),
26
27    Parsing(ParsingError),
28    Serialize(SerializeError),
29}
30
31#[derive(Debug)]
32pub enum ParsingError {
33    Incomplete,
34    WouldOverflow,
35    Other,
36}
37
38#[derive(Debug)]
39pub enum SerializeError {
40    WouldOverflow,
41}
42
43async fn read_message<T, M, C>(mut conn: &mut T, buf: &mut BytesMut) -> Result<M, SocksError<C>>
44where
45    T: Read + Unpin,
46    M: for<'a, 'b> TryFrom<&'a mut &'b [u8], Error = ParsingError>,
47{
48    let mut tmp = [0; 513];
49
50    loop {
51        let mut view = &buf[..];
52        match M::try_from(&mut view) {
53            Err(ParsingError::Incomplete) => {
54                let n = crate::rt::read(&mut conn, &mut tmp).await?;
55
56                if n == 0 {
57                    if buf.spare_capacity_mut().is_empty() {
58                        return Err(SocksError::Parsing(ParsingError::WouldOverflow));
59                    } else {
60                        return Err(std::io::Error::new(
61                            std::io::ErrorKind::UnexpectedEof,
62                            "unexpected eof",
63                        )
64                        .into());
65                    }
66                }
67
68                buf.extend_from_slice(&tmp[..n]);
69            }
70            Err(err) => return Err(err.into()),
71            Ok(res) => {
72                let consumed = buf.len() - view.len();
73                buf.advance(consumed);
74                return Ok(res);
75            }
76        }
77    }
78}
79
80impl<C> std::fmt::Display for SocksError<C> {
81    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
82        f.write_str("SOCKS error: ")?;
83
84        match self {
85            Self::Inner(_) => f.write_str("failed to create underlying connection"),
86            Self::Io(_) => f.write_str("io error during SOCKS handshake"),
87
88            Self::DnsFailure => f.write_str("could not resolve to acceptable address type"),
89            Self::MissingHost => f.write_str("missing destination host"),
90            Self::MissingPort => f.write_str("missing destination port"),
91
92            Self::Parsing(_) => f.write_str("failed parsing server response"),
93            Self::Serialize(_) => f.write_str("failed serialize request"),
94
95            Self::V4(e) => e.fmt(f),
96            Self::V5(e) => e.fmt(f),
97        }
98    }
99}
100
101impl<C: std::fmt::Debug + std::fmt::Display> std::error::Error for SocksError<C> {}
102
103impl<C> From<std::io::Error> for SocksError<C> {
104    fn from(err: std::io::Error) -> Self {
105        Self::Io(err)
106    }
107}
108
109impl<C> From<ParsingError> for SocksError<C> {
110    fn from(err: ParsingError) -> Self {
111        Self::Parsing(err)
112    }
113}
114
115impl<C> From<SerializeError> for SocksError<C> {
116    fn from(err: SerializeError) -> Self {
117        Self::Serialize(err)
118    }
119}
120
121impl<C> From<SocksV4Error> for SocksError<C> {
122    fn from(err: SocksV4Error) -> Self {
123        Self::V4(err)
124    }
125}
126
127impl<C> From<SocksV5Error> for SocksError<C> {
128    fn from(err: SocksV5Error) -> Self {
129        Self::V5(err)
130    }
131}
132
133pin_project! {
134    // Not publicly exported (so missing_docs doesn't trigger).
135    //
136    // We return this `Future` instead of the `Pin<Box<dyn Future>>` directly
137    // so that users don't rely on it fitting in a `Pin<Box<dyn Future>>` slot
138    // (and thus we can change the type in the future).
139    #[must_use = "futures do nothing unless polled"]
140    #[allow(missing_debug_implementations)]
141    pub struct Handshaking<F, T, E> {
142        #[pin]
143        fut: BoxHandshaking<T, E>,
144        _marker: std::marker::PhantomData<F>
145    }
146}
147
148type BoxHandshaking<T, E> = Pin<Box<dyn Future<Output = Result<T, SocksError<E>>> + Send>>;
149
150impl<F, T, E> Future for Handshaking<F, T, E>
151where
152    F: Future<Output = Result<T, E>>,
153{
154    type Output = Result<T, SocksError<E>>;
155
156    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
157        self.project().fut.poll(cx)
158    }
159}
160
161#[cfg(all(test, feature = "tokio"))]
162mod test {
163    use bytes::BytesMut;
164    use tokio::io::AsyncWriteExt;
165
166    use super::v5::messages::{AuthMethod, AuthenticationRes, NegotiationRes, ProxyRes, Status};
167    use super::{SocksError, read_message};
168    use crate::rt::TokioIo;
169
170    // A SOCKS5 ProxyRes message. Successful, bound to 127.0.0.1:8080.
171    const SEG1: [u8; 4] = [0x05, 0x00, 0x00, 0x01];
172    const SEG2: [u8; 6] = [0x7F, 0x00, 0x00, 0x01, 0x1F, 0x90];
173
174    // A SOCKS5 NegotiationRes message: username/password method selected.
175    const NEG_RES: [u8; 2] = [0x05, 0x02];
176    // A SOCKS5 AuthenticationRes message: success.
177    const AUTH_RES: [u8; 2] = [0x01, 0x00];
178
179    #[tokio::test]
180    async fn it_works_in_one_read() {
181        let (client, mut server) = tokio::io::duplex(SEG1.len() + SEG2.len());
182        server.write_all(&SEG1).await.unwrap();
183        server.write_all(&SEG2).await.unwrap();
184
185        let mut conn = TokioIo::new(client);
186        let mut buf = BytesMut::new();
187
188        let m: Result<ProxyRes, SocksError<()>> = read_message(&mut conn, &mut buf).await;
189        assert!(m.is_ok());
190        assert_eq!(m.unwrap(), ProxyRes(Status::Success));
191    }
192
193    #[tokio::test]
194    async fn it_works_in_multiple_reads() {
195        // Bounded stream ensures message arrives in two reads
196        let (client, mut server) = tokio::io::duplex(SEG1.len());
197        let _writer = tokio::spawn(async move {
198            server.write_all(&SEG1).await.unwrap();
199            server.write_all(&SEG2).await.unwrap();
200        });
201
202        let mut conn = TokioIo::new(client);
203        let mut buf = BytesMut::new();
204
205        let m: Result<ProxyRes, SocksError<()>> = read_message(&mut conn, &mut buf).await;
206        assert!(m.is_ok());
207        assert_eq!(m.unwrap(), ProxyRes(Status::Success));
208
209        _writer.await.unwrap();
210    }
211
212    #[tokio::test(start_paused = true)]
213    async fn optimistic_sending_works_in_single_read() {
214        // Messages will arrive in a single read
215        let message = [&NEG_RES[..], &AUTH_RES[..], &SEG1[..], &SEG2[..]].concat();
216        let (client, mut server) = tokio::io::duplex(message.len());
217        server.write_all(&message).await.unwrap();
218
219        let mut conn = TokioIo::new(client);
220        let mut buf = BytesMut::new();
221
222        let m: Result<NegotiationRes, SocksError<()>> = read_message(&mut conn, &mut buf).await;
223        assert_eq!(m.unwrap(), NegotiationRes(AuthMethod::UserPass));
224
225        let m: Result<AuthenticationRes, SocksError<()>> = tokio::time::timeout(
226            std::time::Duration::from_secs(1),
227            read_message(&mut conn, &mut buf),
228        )
229        .await
230        .expect("second message should be parsed from the buffer, not read from the socket");
231        assert_eq!(m.unwrap(), AuthenticationRes(true));
232
233        let m: Result<ProxyRes, SocksError<()>> = tokio::time::timeout(
234            std::time::Duration::from_secs(1),
235            read_message(&mut conn, &mut buf),
236        )
237        .await
238        .expect("third message should be parsed from the buffer, not read from the socket");
239        assert_eq!(m.unwrap(), ProxyRes(Status::Success));
240
241        assert!(buf.is_empty(), "all handshake bytes should be consumed");
242        drop(server);
243    }
244
245    #[tokio::test]
246    async fn optimistic_sending_works_in_multiple_reads() {
247        // Bounded stream ensures message arrive in multiple reads
248        let message = [&NEG_RES[..], &AUTH_RES[..], &SEG1[..], &SEG2[..]].concat();
249        let (client, mut server) = tokio::io::duplex(message.len() / 4);
250        let _writer = tokio::spawn(async move {
251            server.write_all(&message).await.unwrap();
252            server
253        });
254
255        let mut conn = TokioIo::new(client);
256        let mut buf = BytesMut::new();
257
258        let m: Result<NegotiationRes, SocksError<()>> = read_message(&mut conn, &mut buf).await;
259        assert_eq!(m.unwrap(), NegotiationRes(AuthMethod::UserPass));
260
261        let m: Result<AuthenticationRes, SocksError<()>> = read_message(&mut conn, &mut buf).await;
262        assert_eq!(m.unwrap(), AuthenticationRes(true));
263
264        let m: Result<ProxyRes, SocksError<()>> = read_message(&mut conn, &mut buf).await;
265        assert_eq!(m.unwrap(), ProxyRes(Status::Success));
266
267        assert!(buf.is_empty(), "all handshake bytes should be consumed");
268
269        let server = _writer.await.unwrap();
270        drop(server);
271    }
272}
hyper_util/client/legacy/connect/proxy/socks/mod.rs

hyper_util/client/legacy/connect/proxy/socks/
mod.rs
