1use crate::{Authentication, Error, IntoTargetAddr, Result, TargetAddr, ToProxyAddrs};
2use futures_util::stream::{self, Fuse, Stream, StreamExt};
3use std::{
4 borrow::Borrow,
5 io,
6 net::{Ipv4Addr, Ipv6Addr, SocketAddr},
7 ops::{Deref, DerefMut},
8 pin::Pin,
9 task::{Context, Poll},
10};
11use tokio::{
12 io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf},
13 net::TcpStream,
14};
15
16#[repr(u8)]
17#[derive(Clone, Copy)]
18enum Command {
19 Connect = 0x01,
20 Bind = 0x02,
21 #[allow(dead_code)]
22 Associate = 0x03,
23 #[cfg(feature = "tor")]
24 TorResolve = 0xF0,
25 #[cfg(feature = "tor")]
26 TorResolvePtr = 0xF1,
27}
28
29#[derive(Debug)]
33pub struct Socks5Stream<S> {
34 socket: S,
35 target: TargetAddr<'static>,
36}
37
38impl<S> Deref for Socks5Stream<S> {
39 type Target = S;
40
41 fn deref(&self) -> &Self::Target {
42 &self.socket
43 }
44}
45
46impl<S> DerefMut for Socks5Stream<S> {
47 fn deref_mut(&mut self) -> &mut Self::Target {
48 &mut self.socket
49 }
50}
51
52impl Socks5Stream<TcpStream> {
53 pub async fn connect<'t, P, T>(proxy: P, target: T) -> Result<Socks5Stream<TcpStream>>
61 where
62 P: ToProxyAddrs,
63 T: IntoTargetAddr<'t>,
64 {
65 Self::execute_command(proxy, target, Authentication::None, Command::Connect).await
66 }
67
68 pub async fn connect_with_password<'a, 't, P, T>(
76 proxy: P,
77 target: T,
78 username: &'a str,
79 password: &'a str,
80 ) -> Result<Socks5Stream<TcpStream>>
81 where
82 P: ToProxyAddrs,
83 T: IntoTargetAddr<'t>,
84 {
85 Self::execute_command(
86 proxy,
87 target,
88 Authentication::Password { username, password },
89 Command::Connect,
90 )
91 .await
92 }
93
94 pub async fn associate<'t, P, T>(proxy: P, local: T) -> Result<Socks5Stream<TcpStream>>
102 where
103 P: ToProxyAddrs,
104 T: IntoTargetAddr<'t>,
105 {
106 Self::execute_command(proxy, local, Authentication::None, Command::Associate).await
107 }
108
109 pub async fn associate_with_password<'a, 't, P, T>(
117 proxy: P,
118 local: T,
119 username: &'a str,
120 password: &'a str,
121 ) -> Result<Socks5Stream<TcpStream>>
122 where
123 P: ToProxyAddrs,
124 T: IntoTargetAddr<'t>,
125 {
126 Self::execute_command(
127 proxy,
128 local,
129 Authentication::Password { username, password },
130 Command::Associate,
131 )
132 .await
133 }
134
135 #[cfg(feature = "tor")]
136 pub async fn tor_resolve<'t, P, T>(proxy: P, target: T) -> Result<TargetAddr<'static>>
139 where
140 P: ToProxyAddrs,
141 T: IntoTargetAddr<'t>,
142 {
143 let sock = Self::execute_command(proxy, target, Authentication::None, Command::TorResolve).await?;
144
145 Ok(sock.target_addr().to_owned())
146 }
147
148 #[cfg(feature = "tor")]
149 pub async fn tor_resolve_ptr<'t, P, T>(proxy: P, target: T) -> Result<TargetAddr<'static>>
153 where
154 P: ToProxyAddrs,
155 T: IntoTargetAddr<'t>,
156 {
157 let sock = Self::execute_command(proxy, target, Authentication::None, Command::TorResolvePtr).await?;
158
159 Ok(sock.target_addr().to_owned())
160 }
161
162 async fn execute_command<'a, 't, P, T>(
163 proxy: P,
164 target: T,
165 auth: Authentication<'a>,
166 command: Command,
167 ) -> Result<Socks5Stream<TcpStream>>
168 where
169 P: ToProxyAddrs,
170 T: IntoTargetAddr<'t>,
171 {
172 Self::validate_auth(&auth)?;
173
174 let sock = SocksConnector::new(auth, command, proxy.to_proxy_addrs().fuse(), target.into_target_addr()?)
175 .execute()
176 .await?;
177
178 Ok(sock)
179 }
180}
181
182impl<S> Socks5Stream<S>
183where S: AsyncRead + AsyncWrite + Unpin
184{
185 pub async fn connect_with_socket<'t, T>(socket: S, target: T) -> Result<Socks5Stream<S>>
192 where T: IntoTargetAddr<'t> {
193 Self::execute_command_with_socket(socket, target, Authentication::None, Command::Connect).await
194 }
195
196 pub async fn connect_with_password_and_socket<'a, 't, T>(
204 socket: S,
205 target: T,
206 username: &'a str,
207 password: &'a str,
208 ) -> Result<Socks5Stream<S>>
209 where
210 T: IntoTargetAddr<'t>,
211 {
212 Self::execute_command_with_socket(
213 socket,
214 target,
215 Authentication::Password { username, password },
216 Command::Connect,
217 )
218 .await
219 }
220
221 pub async fn associate_with_socket<'t, T>(socket: S, local: T) -> Result<Socks5Stream<S>>
229 where T: IntoTargetAddr<'t> {
230 Self::execute_command_with_socket(socket, local, Authentication::None, Command::Associate).await
231 }
232
233 pub async fn associate_with_password_and_socket<'a, 't, T>(
241 socket: S,
242 local: T,
243 username: &'a str,
244 password: &'a str,
245 ) -> Result<Socks5Stream<S>>
246 where
247 T: IntoTargetAddr<'t>,
248 {
249 Self::execute_command_with_socket(
250 socket,
251 local,
252 Authentication::Password { username, password },
253 Command::Associate,
254 )
255 .await
256 }
257
258 fn validate_auth(auth: &Authentication<'_>) -> Result<()> {
259 match auth {
260 Authentication::Password { username, password } => {
261 let username_len = username.len();
262 if username_len > 255 {
263 Err(Error::InvalidAuthValues("username length should between 0 to 255"))?
264 }
265 let password_len = password.len();
266 if password_len > 255 {
267 Err(Error::InvalidAuthValues("password length should between 0 to 255"))?
268 }
269 },
270 Authentication::None => {},
271 }
272 Ok(())
273 }
274
275 #[cfg(feature = "tor")]
276 pub async fn tor_resolve_with_socket<'t, T>(socket: S, target: T) -> Result<TargetAddr<'static>>
279 where T: IntoTargetAddr<'t> {
280 let sock = Self::execute_command_with_socket(socket, target, Authentication::None, Command::TorResolve).await?;
281
282 Ok(sock.target_addr().to_owned())
283 }
284
285 #[cfg(feature = "tor")]
286 pub async fn tor_resolve_ptr_with_socket<'t, T>(socket: S, target: T) -> Result<TargetAddr<'static>>
290 where T: IntoTargetAddr<'t> {
291 let sock =
292 Self::execute_command_with_socket(socket, target, Authentication::None, Command::TorResolvePtr).await?;
293
294 Ok(sock.target_addr().to_owned())
295 }
296
297 async fn execute_command_with_socket<'a, 't, T>(
298 socket: S,
299 target: T,
300 auth: Authentication<'a>,
301 command: Command,
302 ) -> Result<Socks5Stream<S>>
303 where
304 T: IntoTargetAddr<'t>,
305 {
306 Self::validate_auth(&auth)?;
307
308 let sock = SocksConnector::new(auth, command, stream::empty().fuse(), target.into_target_addr()?)
309 .execute_with_socket(socket)
310 .await?;
311
312 Ok(sock)
313 }
314
315 pub fn into_inner(self) -> S {
317 self.socket
318 }
319
320 pub fn target_addr(&self) -> TargetAddr<'_> {
322 match &self.target {
323 TargetAddr::Ip(addr) => TargetAddr::Ip(*addr),
324 TargetAddr::Domain(domain, port) => {
325 let domain: &str = domain.borrow();
326 TargetAddr::Domain(domain.into(), *port)
327 },
328 }
329 }
330}
331
332pub struct SocksConnector<'a, 't, S> {
334 auth: Authentication<'a>,
335 command: Command,
336 proxy: Fuse<S>,
337 target: TargetAddr<'t>,
338 buf: [u8; 513],
339 ptr: usize,
340 len: usize,
341}
342
343impl<'a, 't, S> SocksConnector<'a, 't, S>
344where S: Stream<Item = Result<SocketAddr>> + Unpin
345{
346 fn new(auth: Authentication<'a>, command: Command, proxy: Fuse<S>, target: TargetAddr<'t>) -> Self {
347 SocksConnector {
348 auth,
349 command,
350 proxy,
351 target,
352 buf: [0; 513],
353 ptr: 0,
354 len: 0,
355 }
356 }
357
358 pub async fn execute(&mut self) -> Result<Socks5Stream<TcpStream>> {
371 let mut last_err: Option<Error> = None;
372 while let Some(item) = self.proxy.next().await {
373 let next_addr = match item {
374 Ok(addr) => addr,
375 Err(e) => {
376 last_err = Some(e);
377 continue;
378 },
379 };
380 match TcpStream::connect(next_addr).await {
381 Ok(tcp) => {
382 let mut stream = self.execute_with_socket(tcp).await?;
383 if let TargetAddr::Ip(SocketAddr::V4(target_addr)) = &mut stream.target
384 && let SocketAddr::V4(proxy_addr) = next_addr
385 && target_addr.ip().is_private()
386 && !proxy_addr.ip().is_private()
387 {
388 target_addr.set_ip(*proxy_addr.ip());
389 }
390 return Ok(stream);
391 },
392 Err(e) => last_err = Some(e.into()),
393 }
394 }
395 Err(last_err.unwrap_or(Error::ProxyServerUnreachable))
396 }
397
398 pub async fn execute_with_socket<T: AsyncRead + AsyncWrite + Unpin>(
399 &mut self,
400 mut socket: T,
401 ) -> Result<Socks5Stream<T>> {
402 self.authenticate(&mut socket).await?;
403
404 self.prepare_send_request();
406 socket.write_all(&self.buf[self.ptr..self.len]).await?;
407
408 let target = self.receive_reply(&mut socket).await?;
409
410 Ok(Socks5Stream { socket, target })
411 }
412
413 fn prepare_send_method_selection(&mut self) {
414 self.ptr = 0;
415 self.buf[0] = 0x05;
416 match self.auth {
417 Authentication::None => {
418 self.buf[1..3].copy_from_slice(&[1, 0x00]);
419 self.len = 3;
420 },
421 Authentication::Password { .. } => {
422 self.buf[1..4].copy_from_slice(&[2, 0x00, 0x02]);
423 self.len = 4;
424 },
425 }
426 }
427
428 fn prepare_recv_method_selection(&mut self) {
429 self.ptr = 0;
430 self.len = 2;
431 }
432
433 fn prepare_send_password_auth(&mut self) {
434 if let Authentication::Password { username, password } = self.auth {
435 self.ptr = 0;
436 self.buf[0] = 0x01;
437 let username_bytes = username.as_bytes();
438 let username_len = username_bytes.len();
439 self.buf[1] = username_len as u8;
440 self.buf[2..(2 + username_len)].copy_from_slice(username_bytes);
441 let password_bytes = password.as_bytes();
442 let password_len = password_bytes.len();
443 self.len = 3 + username_len + password_len;
444 self.buf[2 + username_len] = password_len as u8;
445 self.buf[(3 + username_len)..self.len].copy_from_slice(password_bytes);
446 } else {
447 unreachable!()
448 }
449 }
450
451 fn prepare_recv_password_auth(&mut self) {
452 self.ptr = 0;
453 self.len = 2;
454 }
455
456 fn prepare_send_request(&mut self) {
457 self.ptr = 0;
458 self.buf[..3].copy_from_slice(&[0x05, self.command as u8, 0x00]);
459 match &self.target {
460 TargetAddr::Ip(SocketAddr::V4(addr)) => {
461 self.buf[3] = 0x01;
462 self.buf[4..8].copy_from_slice(&addr.ip().octets());
463 self.buf[8..10].copy_from_slice(&addr.port().to_be_bytes());
464 self.len = 10;
465 },
466 TargetAddr::Ip(SocketAddr::V6(addr)) => {
467 self.buf[3] = 0x04;
468 self.buf[4..20].copy_from_slice(&addr.ip().octets());
469 self.buf[20..22].copy_from_slice(&addr.port().to_be_bytes());
470 self.len = 22;
471 },
472 TargetAddr::Domain(domain, port) => {
473 self.buf[3] = 0x03;
474 let domain = domain.as_bytes();
475 let len = domain.len();
476 self.buf[4] = len as u8;
477 self.buf[5..5 + len].copy_from_slice(domain);
478 self.buf[(5 + len)..(7 + len)].copy_from_slice(&port.to_be_bytes());
479 self.len = 7 + len;
480 },
481 }
482 }
483
484 fn prepare_recv_reply(&mut self) {
485 self.ptr = 0;
486 self.len = 4;
487 }
488
489 async fn password_authentication_protocol<T: AsyncRead + AsyncWrite + Unpin>(&mut self, tcp: &mut T) -> Result<()> {
490 if let Authentication::None = self.auth {
491 return Err(Error::AuthorizationRequired);
492 }
493
494 self.prepare_send_password_auth();
495 tcp.write_all(&self.buf[self.ptr..self.len]).await?;
496
497 self.prepare_recv_password_auth();
498 tcp.read_exact(&mut self.buf[self.ptr..self.len]).await?;
499
500 if self.buf[0] != 0x01 {
501 return Err(Error::InvalidResponseVersion);
502 }
503 if self.buf[1] != 0x00 {
504 return Err(Error::PasswordAuthFailure(self.buf[1]));
505 }
506
507 Ok(())
508 }
509
510 async fn authenticate<T: AsyncRead + AsyncWrite + Unpin>(&mut self, tcp: &mut T) -> Result<()> {
511 self.prepare_send_method_selection();
513 tcp.write_all(&self.buf[self.ptr..self.len]).await?;
514
515 self.prepare_recv_method_selection();
517 tcp.read_exact(&mut self.buf[self.ptr..self.len]).await?;
518 if self.buf[0] != 0x05 {
519 return Err(Error::InvalidResponseVersion);
520 }
521 match self.buf[1] {
522 0x00 => {
523 },
525 0x02 => {
526 self.password_authentication_protocol(tcp).await?;
527 },
528 0xff => {
529 return Err(Error::NoAcceptableAuthMethods);
530 },
531 _ => return Err(Error::UnknownAuthMethod),
532 }
533
534 Ok(())
535 }
536
537 async fn receive_reply<T: AsyncRead + AsyncWrite + Unpin>(&mut self, tcp: &mut T) -> Result<TargetAddr<'static>> {
538 self.prepare_recv_reply();
539 self.ptr += tcp.read_exact(&mut self.buf[self.ptr..self.len]).await?;
540 if self.buf[0] != 0x05 {
541 return Err(Error::InvalidResponseVersion);
542 }
543 if self.buf[2] != 0x00 {
544 return Err(Error::InvalidReservedByte);
545 }
546
547 match self.buf[1] {
548 0x00 => {}, 0x01 => return Err(Error::GeneralSocksServerFailure),
550 0x02 => return Err(Error::ConnectionNotAllowedByRuleset),
551 0x03 => return Err(Error::NetworkUnreachable),
552 0x04 => return Err(Error::HostUnreachable),
553 0x05 => return Err(Error::ConnectionRefused),
554 0x06 => return Err(Error::TtlExpired),
555 0x07 => return Err(Error::CommandNotSupported),
556 0x08 => return Err(Error::AddressTypeNotSupported),
557 _ => return Err(Error::UnknownError),
558 }
559
560 match self.buf[3] {
561 0x01 => {
563 self.len = 10;
564 },
565 0x04 => {
567 self.len = 22;
568 },
569 0x03 => {
571 self.len = 5;
572 self.ptr += tcp.read_exact(&mut self.buf[self.ptr..self.len]).await?;
573 self.len += self.buf[4] as usize + 2;
574 },
575 _ => Err(Error::UnknownAddressType)?,
576 }
577
578 self.ptr += tcp.read_exact(&mut self.buf[self.ptr..self.len]).await?;
579 let target: TargetAddr<'static> = match self.buf[3] {
580 0x01 => {
582 let mut ip = [0; 4];
583 ip[..].copy_from_slice(&self.buf[4..8]);
584 let ip = Ipv4Addr::from(ip);
585 let port = u16::from_be_bytes([self.buf[8], self.buf[9]]);
586 (ip, port).into_target_addr()?
587 },
588 0x04 => {
590 let mut ip = [0; 16];
591 ip[..].copy_from_slice(&self.buf[4..20]);
592 let ip = Ipv6Addr::from(ip);
593 let port = u16::from_be_bytes([self.buf[20], self.buf[21]]);
594 (ip, port).into_target_addr()?
595 },
596 0x03 => {
598 let domain_bytes = self.buf[5..(self.len - 2)].to_vec();
599 let domain = String::from_utf8(domain_bytes)
600 .map_err(|_| Error::InvalidTargetAddress("not a valid UTF-8 string"))?;
601 let port = u16::from_be_bytes([self.buf[self.len - 2], self.buf[self.len - 1]]);
602 TargetAddr::Domain(domain.into(), port)
603 },
604 _ => unreachable!(),
605 };
606
607 Ok(target)
608 }
609}
610
611pub struct Socks5Listener<S> {
618 inner: Socks5Stream<S>,
619}
620
621impl Socks5Listener<TcpStream> {
622 pub async fn bind<'t, P, T>(proxy: P, target: T) -> Result<Socks5Listener<TcpStream>>
632 where
633 P: ToProxyAddrs,
634 T: IntoTargetAddr<'t>,
635 {
636 Self::bind_with_auth(Authentication::None, proxy, target).await
637 }
638
639 pub async fn bind_with_password<'a, 't, P, T>(
650 proxy: P,
651 target: T,
652 username: &'a str,
653 password: &'a str,
654 ) -> Result<Socks5Listener<TcpStream>>
655 where
656 P: ToProxyAddrs,
657 T: IntoTargetAddr<'t>,
658 {
659 Self::bind_with_auth(Authentication::Password { username, password }, proxy, target).await
660 }
661
662 async fn bind_with_auth<'t, P, T>(
663 auth: Authentication<'_>,
664 proxy: P,
665 target: T,
666 ) -> Result<Socks5Listener<TcpStream>>
667 where
668 P: ToProxyAddrs,
669 T: IntoTargetAddr<'t>,
670 {
671 let socket = SocksConnector::new(
672 auth,
673 Command::Bind,
674 proxy.to_proxy_addrs().fuse(),
675 target.into_target_addr()?,
676 )
677 .execute()
678 .await?;
679
680 Ok(Socks5Listener { inner: socket })
681 }
682}
683
684impl<S> Socks5Listener<S>
685where S: AsyncRead + AsyncWrite + Unpin
686{
687 pub async fn bind_with_socket<'t, T>(socket: S, target: T) -> Result<Socks5Listener<S>>
698 where T: IntoTargetAddr<'t> {
699 Self::bind_with_auth_and_socket(Authentication::None, socket, target).await
700 }
701
702 pub async fn bind_with_password_and_socket<'a, 't, T>(
713 socket: S,
714 target: T,
715 username: &'a str,
716 password: &'a str,
717 ) -> Result<Socks5Listener<S>>
718 where
719 T: IntoTargetAddr<'t>,
720 {
721 Self::bind_with_auth_and_socket(Authentication::Password { username, password }, socket, target).await
722 }
723
724 async fn bind_with_auth_and_socket<'t, T>(
725 auth: Authentication<'_>,
726 socket: S,
727 target: T,
728 ) -> Result<Socks5Listener<S>>
729 where
730 T: IntoTargetAddr<'t>,
731 {
732 let socket = SocksConnector::new(auth, Command::Bind, stream::empty().fuse(), target.into_target_addr()?)
733 .execute_with_socket(socket)
734 .await?;
735
736 Ok(Socks5Listener { inner: socket })
737 }
738
739 pub fn bind_addr(&self) -> TargetAddr<'_> {
744 self.inner.target_addr()
745 }
746
747 pub async fn accept(mut self) -> Result<Socks5Stream<S>> {
753 let mut connector = SocksConnector {
754 auth: Authentication::None,
755 command: Command::Bind,
756 proxy: stream::empty().fuse(),
757 target: self.inner.target,
758 buf: [0; 513],
759 ptr: 0,
760 len: 0,
761 };
762
763 let target = connector.receive_reply(&mut self.inner.socket).await?;
764
765 Ok(Socks5Stream {
766 socket: self.inner.socket,
767 target,
768 })
769 }
770}
771
772impl<T> AsyncRead for Socks5Stream<T>
773where T: AsyncRead + Unpin
774{
775 fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
776 AsyncRead::poll_read(Pin::new(&mut self.socket), cx, buf)
777 }
778}
779
780impl<T> AsyncWrite for Socks5Stream<T>
781where T: AsyncWrite + Unpin
782{
783 fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
784 AsyncWrite::poll_write(Pin::new(&mut self.socket), cx, buf)
785 }
786
787 fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
788 AsyncWrite::poll_flush(Pin::new(&mut self.socket), cx)
789 }
790
791 fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
792 AsyncWrite::poll_shutdown(Pin::new(&mut self.socket), cx)
793 }
794}