clone from github.com/sticnarf/tokio-socks

CHANGELOG.md

@@ -0,0 +1,49 @@
+# 0.5.1
+
+* Reduce dependencies on `futures` crate (#30)
+
+# 0.5.0
+
+* Upgrade tokio to 1.0 (#28)
+
+# 0.4.0
+
+* Return error if authorization is required but credentials are not present (#24)
+
+* Upgrade tokio to 0.3 (#27)
+
+# 0.3.0
+
+* Allow to take arbitrary socket instead of address to establish connections to proxy (#20)
+
+# 0.2.2
+
+* Replace failure with thiserror (#17)
+
+# 0.2.1
+
+* Remove dependency derefable (#16)
+
+# 0.2.0
+
+* Support tokio 0.2 (#10)
+
+# 0.1.3
+
+* Implement `IntoTargetAddr<'static>` for `String` (#8)
+
+# 0.1.2
+
+* Fix ConnectFuture buffer too small (#1)
+
+# 0.1.1
+
+* Support SOCKS5 `BIND` command.
+
+* Implement `std::net::ToSocketAddrs` for `TargetAddr`.
+
+# 0.1.0
+
+* Support SOCKS5 `CONNECT` command.
+
+* Support username authentication.

Cargo.toml

@@ -0,0 +1,39 @@
+[package]
+name = "tokio-socks"
+description = "Asynchronous SOCKS proxy support for Rust."
+documentation = "https://docs.rs/tokio-socks"
+homepage = "https://github.com/sticnarf/tokio-socks"
+repository = "https://github.com/sticnarf/tokio-socks"
+readme = "README.md"
+categories = ["asynchronous", "network-programming"]
+keywords = ["tokio", "async", "proxy", "socks", "socks5"]
+license = "MIT"
+version = "0.5.1"
+authors = ["Yilin Chen <sticnarf@gmail.com>"]
+edition = "2018"
+
+[badges]
+travis-ci = { repository = "sticnarf/tokio-socks" }
+
+[features]
+tor = []
+
+[[example]]
+name = "socket"
+required-features = ["tor"]
+
+[[example]]
+name = "tor"
+required-features = ["tor"]
+
+[dependencies]
+futures-util = { version = "0.3", default-features = false }
+tokio = { version = "1.0", features = ["io-util", "net"] }
+either = "1"
+thiserror = "1.0"
+
+[dev-dependencies]
+futures-executor = "0.3"
+tokio = { version = "1.0", features = ["io-util", "rt-multi-thread", "net"] }
+once_cell = "1.2.0"
+hyper = "0.14"

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2018 Yilin Chen
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -1,3 +1,26 @@
-# another-tokio-socks
+# tokio-socks
 
-Clone from: https://github.com/sticnarf/tokio-socks
+[![Build Status](https://travis-ci.org/sticnarf/tokio-socks.svg?branch=master)](https://travis-ci.org/sticnarf/tokio-socks)
+[![Crates Version](https://img.shields.io/crates/v/tokio-socks.svg)](https://crates.io/crates/tokio-socks)
+[![docs](https://docs.rs/tokio-socks/badge.svg)](https://docs.rs/tokio-socks)
+
+Asynchronous SOCKS proxy support for Rust.
+
+## Features
+
+- [x] `CONNECT` command
+- [x] `BIND` command
+- [ ] `ASSOCIATE` command
+- [x] Username/password authentication
+- [ ] GSSAPI authentication
+- [ ] Asynchronous DNS resolution
+- [X] Chain proxies ([see example](examples/chainproxy.rs))
+- [ ] SOCKS4
+
+## License
+
+This project is licensed under the MIT License - see the [LICENSE](/LICENSE) file for details.
+
+## Acknowledgments
+
+* [sfackler/rust-socks](https://github.com/sfackler/rust-socks)

examples/chainproxy.rs

@@ -0,0 +1,33 @@
+//! Test the proxy chaining capabilities
+//!
+//! This example make uses of several public proxy.
+
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt},
+    net::TcpStream,
+    runtime::Runtime,
+};
+use tokio_socks::{tcp::Socks5Stream, Error};
+
+const PROXY_ADDR: [&str; 2] = ["184.176.166.20:4145", "90.89.205.248:1080"]; // public proxies found here : http://spys.one/en/socks-proxy-list/
+const DEST_ADDR: &str = "duckduckgo.com:80";
+
+async fn connect_chained_proxy() -> Result<(), Error> {
+    let proxy_stream = TcpStream::connect(PROXY_ADDR[0]).await?;
+    let chained_proxy_stream = Socks5Stream::connect_with_socket(proxy_stream, PROXY_ADDR[1]).await?;
+    let mut stream = Socks5Stream::connect_with_socket(chained_proxy_stream, DEST_ADDR).await?;
+
+    stream.write_all(b"GET /\n\n").await?;
+
+    let mut buf = Vec::new();
+    let n = stream.read_to_end(&mut buf).await?;
+
+    println!("{} bytes read\n\n{}", n, String::from_utf8_lossy(&buf));
+
+    Ok(())
+}
+
+fn main() {
+    let rt = Runtime::new().unwrap();
+    rt.block_on(connect_chained_proxy()).unwrap();
+}

examples/socket.rs

@@ -0,0 +1,44 @@
+//! Test the tor proxy capabilities
+//!
+//! This example requires a running tor proxy.
+
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt},
+    net::{TcpStream, UnixStream},
+    runtime::Runtime,
+};
+use tokio_socks::{tcp::Socks5Stream, Error};
+
+const UNIX_PROXY_ADDR: &str = "/tmp/tor/socket.s";
+const TCP_PROXY_ADDR: &str = "127.0.0.1:9050";
+const ONION_ADDR: &str = "3g2upl4pq6kufc4m.onion:80"; // DuckDuckGo
+
+async fn connect() -> Result<(), Error> {
+    // This require Tor to listen on and Unix Domain Socket.
+    // You have to create a directory /tmp/tor owned by tor, and for which only tor
+    // has rights, and add the following line to your torrc :
+    // SocksPort unix:/tmp/tor/socket.s
+    let socket = UnixStream::connect(UNIX_PROXY_ADDR).await?;
+    let target = Socks5Stream::tor_resolve_with_socket(socket, "duckduckgo.com:0").await?;
+    eprintln!("duckduckgo.com = {:?}", target);
+    let socket = UnixStream::connect(UNIX_PROXY_ADDR).await?;
+    let target = Socks5Stream::tor_resolve_ptr_with_socket(socket, "176.34.155.23:0").await?;
+    eprintln!("176.34.155.23 = {:?}", target);
+
+    let socket = TcpStream::connect(TCP_PROXY_ADDR).await?;
+    socket.set_nodelay(true)?;
+    let mut conn = Socks5Stream::connect_with_socket(socket, ONION_ADDR).await?;
+    conn.write_all(b"GET /\n\n").await?;
+
+    let mut buf = Vec::new();
+    let n = conn.read_to_end(&mut buf).await?;
+
+    println!("{} bytes read\n\n{}", n, String::from_utf8_lossy(&buf));
+
+    Ok(())
+}
+
+fn main() {
+    let mut rt = Runtime::new().unwrap();
+    rt.block_on(connect()).unwrap();
+}

examples/tor.rs

@@ -0,0 +1,34 @@
+//! Test the tor proxy capabilities
+//!
+//! This example requires a running tor proxy.
+
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt},
+    runtime::Runtime,
+};
+use tokio_socks::{tcp::Socks5Stream, Error};
+
+const PROXY_ADDR: &str = "127.0.0.1:9050";
+const ONION_ADDR: &str = "3g2upl4pq6kufc4m.onion:80"; // DuckDuckGo
+
+async fn connect() -> Result<(), Error> {
+    let target = Socks5Stream::tor_resolve(PROXY_ADDR, "duckduckgo.com:0").await?;
+    eprintln!("duckduckgo.com = {:?}", target);
+    let target = Socks5Stream::tor_resolve_ptr(PROXY_ADDR, "176.34.155.23:0").await?;
+    eprintln!("176.34.155.23 = {:?}", target);
+
+    let mut conn = Socks5Stream::connect(PROXY_ADDR, ONION_ADDR).await?;
+    conn.write_all(b"GET /\n\n").await?;
+
+    let mut buf = Vec::new();
+    let n = conn.read_to_end(&mut buf).await?;
+
+    println!("{} bytes read\n\n{}", n, String::from_utf8_lossy(&buf));
+
+    Ok(())
+}
+
+fn main() {
+    let mut rt = Runtime::new().unwrap();
+    rt.block_on(connect()).unwrap();
+}

rustfmt.toml

@@ -0,0 +1,24 @@
+use_small_heuristics = "default"
+hard_tabs = false
+imports_layout = "HorizontalVertical"
+merge_imports = true
+match_block_trailing_comma = true
+max_width = 120
+newline_style = "Unix"
+normalize_comments = true
+reorder_imports = true
+reorder_modules = true
+reorder_impl_items = true
+report_todo = "Never"
+report_fixme = "Never"
+space_after_colon = true
+space_before_colon = false
+struct_lit_single_line = true
+use_field_init_shorthand = true
+use_try_shorthand = true
+unstable_features = true
+format_code_in_doc_comments = true
+where_single_line = true
+wrap_comments = true
+overflow_delimited_expr = true
+edition = "2018"

src/error.rs

@@ -0,0 +1,71 @@
+/// Error type of `tokio-socks`
+#[derive(thiserror::Error, Debug)]
+pub enum Error {
+    /// Failure caused by an IO error.
+    #[error("{0}")]
+    Io(#[from] std::io::Error),
+    /// Failure when parsing a `String`.
+    #[error("{0}")]
+    ParseError(#[from] std::string::ParseError),
+    /// Failure due to invalid target address. It contains the detailed error
+    /// message.
+    #[error("Target address is invalid: {0}")]
+    InvalidTargetAddress(&'static str),
+    /// Proxy server unreachable.
+    #[error("Proxy server unreachable")]
+    ProxyServerUnreachable,
+    /// Proxy server returns an invalid version number.
+    #[error("Invalid response version")]
+    InvalidResponseVersion,
+    /// No acceptable auth methods
+    #[error("No acceptable auth methods")]
+    NoAcceptableAuthMethods,
+    /// Unknown auth method
+    #[error("Unknown auth method")]
+    UnknownAuthMethod,
+    /// General SOCKS server failure
+    #[error("General SOCKS server failure")]
+    GeneralSocksServerFailure,
+    /// Connection not allowed by ruleset
+    #[error("Connection not allowed by ruleset")]
+    ConnectionNotAllowedByRuleset,
+    /// Network unreachable
+    #[error("Network unreachable")]
+    NetworkUnreachable,
+    /// Host unreachable
+    #[error("Host unreachable")]
+    HostUnreachable,
+    /// Connection refused
+    #[error("Connection refused")]
+    ConnectionRefused,
+    /// TTL expired
+    #[error("TTL expired")]
+    TtlExpired,
+    /// Command not supported
+    #[error("Command not supported")]
+    CommandNotSupported,
+    /// Address type not supported
+    #[error("Address type not supported")]
+    AddressTypeNotSupported,
+    /// Unknown error
+    #[error("Unknown error")]
+    UnknownError,
+    /// Invalid reserved byte
+    #[error("Invalid reserved byte")]
+    InvalidReservedByte,
+    /// Unknown address type
+    #[error("Unknown address type")]
+    UnknownAddressType,
+    /// Invalid authentication values. It contains the detailed error message.
+    #[error("Invalid auth values: {0}")]
+    InvalidAuthValues(&'static str),
+    /// Password auth failure
+    #[error("Password auth failure, code: {0}")]
+    PasswordAuthFailure(u8),
+
+    #[error("Authorization required")]
+    AuthorizationRequired,
+}
+
+///// Result type of `tokio-socks`
+// pub type Result<T> = std::result::Result<T, Error>;

src/lib.rs

@@ -0,0 +1,374 @@
+use either::Either;
+use futures_util::{
+    future,
+    stream::{self, Once, Stream},
+};
+use std::{
+    borrow::Cow,
+    io,
+    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6, ToSocketAddrs},
+    pin::Pin,
+    task::{Context, Poll},
+    vec,
+};
+
+pub use error::Error;
+
+pub type Result<T> = std::result::Result<T, Error>;
+
+/// A trait for objects which can be converted or resolved to one or more
+/// `SocketAddr` values, which are going to be connected as the the proxy
+/// server.
+///
+/// This trait is similar to `std::net::ToSocketAddrs` but allows asynchronous
+/// name resolution.
+pub trait ToProxyAddrs {
+    type Output: Stream<Item = Result<SocketAddr>> + Unpin;
+
+    fn to_proxy_addrs(&self) -> Self::Output;
+}
+
+macro_rules! trivial_impl_to_proxy_addrs {
+    ($t: ty) => {
+        impl ToProxyAddrs for $t {
+            type Output = Once<future::Ready<Result<SocketAddr>>>;
+
+            fn to_proxy_addrs(&self) -> Self::Output {
+                stream::once(future::ready(Ok(SocketAddr::from(*self))))
+            }
+        }
+    };
+}
+
+trivial_impl_to_proxy_addrs!(SocketAddr);
+trivial_impl_to_proxy_addrs!((IpAddr, u16));
+trivial_impl_to_proxy_addrs!((Ipv4Addr, u16));
+trivial_impl_to_proxy_addrs!((Ipv6Addr, u16));
+trivial_impl_to_proxy_addrs!(SocketAddrV4);
+trivial_impl_to_proxy_addrs!(SocketAddrV6);
+
+impl<'a> ToProxyAddrs for &'a [SocketAddr] {
+    type Output = ProxyAddrsStream;
+
+    fn to_proxy_addrs(&self) -> Self::Output {
+        ProxyAddrsStream(Some(io::Result::Ok(self.to_vec().into_iter())))
+    }
+}
+
+impl ToProxyAddrs for str {
+    type Output = ProxyAddrsStream;
+
+    fn to_proxy_addrs(&self) -> Self::Output {
+        ProxyAddrsStream(Some(self.to_socket_addrs()))
+    }
+}
+
+impl<'a> ToProxyAddrs for (&'a str, u16) {
+    type Output = ProxyAddrsStream;
+
+    fn to_proxy_addrs(&self) -> Self::Output {
+        ProxyAddrsStream(Some(self.to_socket_addrs()))
+    }
+}
+
+impl<'a, T: ToProxyAddrs + ?Sized> ToProxyAddrs for &'a T {
+    type Output = T::Output;
+
+    fn to_proxy_addrs(&self) -> Self::Output {
+        (**self).to_proxy_addrs()
+    }
+}
+
+pub struct ProxyAddrsStream(Option<io::Result<vec::IntoIter<SocketAddr>>>);
+
+impl Stream for ProxyAddrsStream {
+    type Item = Result<SocketAddr>;
+
+    fn poll_next(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        match self.0.as_mut() {
+            Some(Ok(iter)) => Poll::Ready(iter.next().map(Result::Ok)),
+            Some(Err(_)) => {
+                let err = self.0.take().unwrap().unwrap_err();
+                Poll::Ready(Some(Err(err.into())))
+            },
+            None => unreachable!(),
+        }
+    }
+}
+
+/// A SOCKS connection target.
+#[derive(Debug, PartialEq, Eq)]
+pub enum TargetAddr<'a> {
+    /// Connect to an IP address.
+    Ip(SocketAddr),
+
+    /// Connect to a fully-qualified domain name.
+    ///
+    /// The domain name will be passed along to the proxy server and DNS lookup
+    /// will happen there.
+    Domain(Cow<'a, str>, u16),
+}
+
+impl<'a> TargetAddr<'a> {
+    /// Creates owned `TargetAddr` by cloning. It is usually used to eliminate
+    /// the lifetime bound.
+    pub fn to_owned(&self) -> TargetAddr<'static> {
+        match self {
+            TargetAddr::Ip(addr) => TargetAddr::Ip(*addr),
+            TargetAddr::Domain(domain, port) => TargetAddr::Domain(String::from(domain.clone()).into(), *port),
+        }
+    }
+}
+
+impl<'a> ToSocketAddrs for TargetAddr<'a> {
+    type Iter = Either<std::option::IntoIter<SocketAddr>, std::vec::IntoIter<SocketAddr>>;
+
+    fn to_socket_addrs(&self) -> io::Result<Self::Iter> {
+        Ok(match self {
+            TargetAddr::Ip(addr) => Either::Left(addr.to_socket_addrs()?),
+            TargetAddr::Domain(domain, port) => Either::Right((&**domain, *port).to_socket_addrs()?),
+        })
+    }
+}
+
+/// A trait for objects that can be converted to `TargetAddr`.
+pub trait IntoTargetAddr<'a> {
+    /// Converts the value of self to a `TargetAddr`.
+    fn into_target_addr(self) -> Result<TargetAddr<'a>>;
+}
+
+macro_rules! trivial_impl_into_target_addr {
+    ($t: ty) => {
+        impl<'a> IntoTargetAddr<'a> for $t {
+            fn into_target_addr(self) -> Result<TargetAddr<'a>> {
+                Ok(TargetAddr::Ip(SocketAddr::from(self)))
+            }
+        }
+    };
+}
+
+trivial_impl_into_target_addr!(SocketAddr);
+trivial_impl_into_target_addr!((IpAddr, u16));
+trivial_impl_into_target_addr!((Ipv4Addr, u16));
+trivial_impl_into_target_addr!((Ipv6Addr, u16));
+trivial_impl_into_target_addr!(SocketAddrV4);
+trivial_impl_into_target_addr!(SocketAddrV6);
+
+impl<'a> IntoTargetAddr<'a> for TargetAddr<'a> {
+    fn into_target_addr(self) -> Result<TargetAddr<'a>> {
+        Ok(self)
+    }
+}
+
+impl<'a> IntoTargetAddr<'a> for (&'a str, u16) {
+    fn into_target_addr(self) -> Result<TargetAddr<'a>> {
+        // Try IP address first
+        if let Ok(addr) = self.0.parse::<IpAddr>() {
+            return (addr, self.1).into_target_addr();
+        }
+
+        // Treat as domain name
+        if self.0.len() > 255 {
+            return Err(Error::InvalidTargetAddress("overlong domain"));
+        }
+        // TODO: Should we validate the domain format here?
+
+        Ok(TargetAddr::Domain(self.0.into(), self.1))
+    }
+}
+
+impl<'a> IntoTargetAddr<'a> for &'a str {
+    fn into_target_addr(self) -> Result<TargetAddr<'a>> {
+        // Try IP address first
+        if let Ok(addr) = self.parse::<SocketAddr>() {
+            return addr.into_target_addr();
+        }
+
+        let mut parts_iter = self.rsplitn(2, ':');
+        let port: u16 = parts_iter
+            .next()
+            .and_then(|port_str| port_str.parse().ok())
+            .ok_or(Error::InvalidTargetAddress("invalid address format"))?;
+        let domain = parts_iter
+            .next()
+            .ok_or(Error::InvalidTargetAddress("invalid address format"))?;
+        if domain.len() > 255 {
+            return Err(Error::InvalidTargetAddress("overlong domain"));
+        }
+        Ok(TargetAddr::Domain(domain.into(), port))
+    }
+}
+
+impl IntoTargetAddr<'static> for String {
+    fn into_target_addr(mut self) -> Result<TargetAddr<'static>> {
+        // Try IP address first
+        if let Ok(addr) = self.parse::<SocketAddr>() {
+            return addr.into_target_addr();
+        }
+
+        let mut parts_iter = self.rsplitn(2, ':');
+        let port: u16 = parts_iter
+            .next()
+            .and_then(|port_str| port_str.parse().ok())
+            .ok_or(Error::InvalidTargetAddress("invalid address format"))?;
+        let domain_len = parts_iter
+            .next()
+            .ok_or(Error::InvalidTargetAddress("invalid address format"))?
+            .len();
+        if domain_len > 255 {
+            return Err(Error::InvalidTargetAddress("overlong domain"));
+        }
+        self.truncate(domain_len);
+        Ok(TargetAddr::Domain(self.into(), port))
+    }
+}
+
+impl IntoTargetAddr<'static> for (String, u16) {
+    fn into_target_addr(self) -> Result<TargetAddr<'static>> {
+        let addr = (self.0.as_str(), self.1).into_target_addr()?;
+        if let TargetAddr::Ip(addr) = addr {
+            Ok(TargetAddr::Ip(addr))
+        } else {
+            Ok(TargetAddr::Domain(self.0.into(), self.1))
+        }
+    }
+}
+
+impl<'a, T> IntoTargetAddr<'a> for &'a T
+where T: IntoTargetAddr<'a> + Copy
+{
+    fn into_target_addr(self) -> Result<TargetAddr<'a>> {
+        (*self).into_target_addr()
+    }
+}
+
+/// Authentication methods
+#[derive(Debug)]
+enum Authentication<'a> {
+    Password { username: &'a str, password: &'a str },
+    None,
+}
+
+impl<'a> Authentication<'a> {
+    fn id(&self) -> u8 {
+        match self {
+            Authentication::Password { .. } => 0x02,
+            Authentication::None => 0x00,
+        }
+    }
+}
+
+mod error;
+pub mod tcp;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures_executor::block_on;
+    use futures_util::StreamExt;
+
+    fn to_proxy_addrs<T: ToProxyAddrs>(t: T) -> Result<Vec<SocketAddr>> {
+        Ok(block_on(t.to_proxy_addrs().map(Result::unwrap).collect()))
+    }
+
+    #[test]
+    fn converts_socket_addr_to_proxy_addrs() -> Result<()> {
+        let addr = SocketAddr::from(([1, 1, 1, 1], 443));
+        let res = to_proxy_addrs(addr)?;
+        assert_eq!(&res[..], &[addr]);
+        Ok(())
+    }
+
+    #[test]
+    fn converts_socket_addr_ref_to_proxy_addrs() -> Result<()> {
+        let addr = SocketAddr::from(([1, 1, 1, 1], 443));
+        let res = to_proxy_addrs(&addr)?;
+        assert_eq!(&res[..], &[addr]);
+        Ok(())
+    }
+
+    #[test]
+    fn converts_socket_addrs_to_proxy_addrs() -> Result<()> {
+        let addrs = [
+            SocketAddr::from(([1, 1, 1, 1], 443)),
+            SocketAddr::from(([8, 8, 8, 8], 53)),
+        ];
+        let res = to_proxy_addrs(&addrs[..])?;
+        assert_eq!(&res[..], &addrs);
+        Ok(())
+    }
+
+    fn into_target_addr<'a, T>(t: T) -> Result<TargetAddr<'a>>
+    where T: IntoTargetAddr<'a> {
+        t.into_target_addr()
+    }
+
+    #[test]
+    fn converts_socket_addr_to_target_addr() -> Result<()> {
+        let addr = SocketAddr::from(([1, 1, 1, 1], 443));
+        let res = into_target_addr(addr)?;
+        assert_eq!(TargetAddr::Ip(addr), res);
+        Ok(())
+    }
+
+    #[test]
+    fn converts_socket_addr_ref_to_target_addr() -> Result<()> {
+        let addr = SocketAddr::from(([1, 1, 1, 1], 443));
+        let res = into_target_addr(&addr)?;
+        assert_eq!(TargetAddr::Ip(addr), res);
+        Ok(())
+    }
+
+    #[test]
+    fn converts_socket_addr_str_to_target_addr() -> Result<()> {
+        let addr = SocketAddr::from(([1, 1, 1, 1], 443));
+        let ip_str = format!("{}", addr);
+        let res = into_target_addr(ip_str.as_str())?;
+        assert_eq!(TargetAddr::Ip(addr), res);
+        Ok(())
+    }
+
+    #[test]
+    fn converts_ip_str_and_port_target_addr() -> Result<()> {
+        let addr = SocketAddr::from(([1, 1, 1, 1], 443));
+        let ip_str = format!("{}", addr.ip());
+        let res = into_target_addr((ip_str.as_str(), addr.port()))?;
+        assert_eq!(TargetAddr::Ip(addr), res);
+        Ok(())
+    }
+
+    #[test]
+    fn converts_domain_to_target_addr() -> Result<()> {
+        let domain = "www.example.com:80";
+        let res = into_target_addr(domain)?;
+        assert_eq!(TargetAddr::Domain(Cow::Borrowed("www.example.com"), 80), res);
+
+        let res = into_target_addr(domain.to_owned())?;
+        assert_eq!(TargetAddr::Domain(Cow::Owned("www.example.com".to_owned()), 80), res);
+        Ok(())
+    }
+
+    #[test]
+    fn converts_domain_and_port_to_target_addr() -> Result<()> {
+        let domain = "www.example.com";
+        let res = into_target_addr((domain, 80))?;
+        assert_eq!(TargetAddr::Domain(Cow::Borrowed("www.example.com"), 80), res);
+        Ok(())
+    }
+
+    #[test]
+    fn overlong_domain_to_target_addr_should_fail() {
+        let domain = format!("www.{:a<1$}.com:80", 'a', 300);
+        assert!(into_target_addr(domain.as_str()).is_err());
+        let domain = format!("www.{:a<1$}.com", 'a', 300);
+        assert!(into_target_addr((domain.as_str(), 80)).is_err());
+    }
+
+    #[test]
+    fn addr_with_invalid_port_to_target_addr_should_fail() {
+        let addr = "[ffff::1]:65536";
+        assert!(into_target_addr(addr).is_err());
+        let addr = "www.example.com:65536";
+        assert!(into_target_addr(addr).is_err());
+    }
+}

src/tcp.rs

@@ -0,0 +1,688 @@
+use crate::{Authentication, Error, IntoTargetAddr, Result, TargetAddr, ToProxyAddrs};
+use futures_util::stream::{self, Fuse, Stream, StreamExt};
+use std::{
+    borrow::Borrow,
+    io,
+    net::{Ipv4Addr, Ipv6Addr, SocketAddr},
+    ops::{Deref, DerefMut},
+    pin::Pin,
+    task::{Context, Poll},
+};
+use tokio::{
+    io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf},
+    net::TcpStream,
+};
+
+#[repr(u8)]
+#[derive(Clone, Copy)]
+enum Command {
+    Connect = 0x01,
+    Bind = 0x02,
+    #[allow(dead_code)]
+    Associate = 0x03,
+    #[cfg(feature = "tor")]
+    TorResolve = 0xF0,
+    #[cfg(feature = "tor")]
+    TorResolvePtr = 0xF1,
+}
+
+/// A SOCKS5 client.
+///
+/// For convenience, it can be dereferenced to it's inner socket.
+#[derive(Debug)]
+pub struct Socks5Stream<S> {
+    socket: S,
+    target: TargetAddr<'static>,
+}
+
+impl<S> Deref for Socks5Stream<S> {
+    type Target = S;
+
+    fn deref(&self) -> &Self::Target {
+        &self.socket
+    }
+}
+
+impl<S> DerefMut for Socks5Stream<S> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.socket
+    }
+}
+
+impl Socks5Stream<TcpStream> {
+    /// Connects to a target server through a SOCKS5 proxy given the proxy
+    /// address.
+    ///
+    /// # Error
+    ///
+    /// It propagates the error that occurs in the conversion from `T` to
+    /// `TargetAddr`.
+    pub async fn connect<'t, P, T>(proxy: P, target: T) -> Result<Socks5Stream<TcpStream>>
+    where
+        P: ToProxyAddrs,
+        T: IntoTargetAddr<'t>,
+    {
+        Self::execute_command(proxy, target, Authentication::None, Command::Connect).await
+    }
+
+    /// Connects to a target server through a SOCKS5 proxy using given username,
+    /// password and the address of the proxy.
+    ///
+    /// # Error
+    ///
+    /// It propagates the error that occurs in the conversion from `T` to
+    /// `TargetAddr`.
+    pub async fn connect_with_password<'a, 't, P, T>(
+        proxy: P,
+        target: T,
+        username: &'a str,
+        password: &'a str,
+    ) -> Result<Socks5Stream<TcpStream>>
+    where
+        P: ToProxyAddrs,
+        T: IntoTargetAddr<'t>,
+    {
+        Self::execute_command(
+            proxy,
+            target,
+            Authentication::Password { username, password },
+            Command::Connect,
+        )
+        .await
+    }
+
+    #[cfg(feature = "tor")]
+    /// Resolve the domain name to an ip using special Tor Resolve command, by
+    /// connecting to a Tor compatible proxy given it's address.
+    pub async fn tor_resolve<'t, P, T>(proxy: P, target: T) -> Result<TargetAddr<'static>>
+    where
+        P: ToProxyAddrs,
+        T: IntoTargetAddr<'t>,
+    {
+        let sock = Self::execute_command(proxy, target, Authentication::None, Command::TorResolve).await?;
+
+        Ok(sock.target_addr().to_owned())
+    }
+
+    #[cfg(feature = "tor")]
+    /// Perform a reverse DNS query on the given ip using special Tor Resolve
+    /// PTR command, by connecting to a Tor compatible proxy given it's
+    /// address.
+    pub async fn tor_resolve_ptr<'t, P, T>(proxy: P, target: T) -> Result<TargetAddr<'static>>
+    where
+        P: ToProxyAddrs,
+        T: IntoTargetAddr<'t>,
+    {
+        let sock = Self::execute_command(proxy, target, Authentication::None, Command::TorResolvePtr).await?;
+
+        Ok(sock.target_addr().to_owned())
+    }
+
+    async fn execute_command<'a, 't, P, T>(
+        proxy: P,
+        target: T,
+        auth: Authentication<'a>,
+        command: Command,
+    ) -> Result<Socks5Stream<TcpStream>>
+    where
+        P: ToProxyAddrs,
+        T: IntoTargetAddr<'t>,
+    {
+        Self::validate_auth(&auth)?;
+
+        let sock = SocksConnector::new(auth, command, proxy.to_proxy_addrs().fuse(), target.into_target_addr()?)
+            .execute()
+            .await?;
+
+        Ok(sock)
+    }
+}
+
+impl<S> Socks5Stream<S>
+where S: AsyncRead + AsyncWrite + Unpin
+{
+    /// Connects to a target server through a SOCKS5 proxy given a socket to it.
+    ///
+    /// # Error
+    ///
+    /// It propagates the error that occurs in the conversion from `T` to
+    /// `TargetAddr`.
+    pub async fn connect_with_socket<'t, T>(socket: S, target: T) -> Result<Socks5Stream<S>>
+    where T: IntoTargetAddr<'t> {
+        Self::execute_command_with_socket(socket, target, Authentication::None, Command::Connect).await
+    }
+
+    /// Connects to a target server through a SOCKS5 proxy using given username,
+    /// password and a socket to the proxy
+    ///
+    /// # Error
+    ///
+    /// It propagates the error that occurs in the conversion from `T` to
+    /// `TargetAddr`.
+    pub async fn connect_with_password_and_socket<'a, 't, T>(
+        socket: S,
+        target: T,
+        username: &'a str,
+        password: &'a str,
+    ) -> Result<Socks5Stream<S>>
+    where
+        T: IntoTargetAddr<'t>,
+    {
+        Self::execute_command_with_socket(
+            socket,
+            target,
+            Authentication::Password { username, password },
+            Command::Connect,
+        )
+        .await
+    }
+
+    fn validate_auth<'a>(auth: &Authentication<'a>) -> Result<()> {
+        match auth {
+            Authentication::Password { username, password } => {
+                let username_len = username.as_bytes().len();
+                if username_len < 1 || username_len > 255 {
+                    Err(Error::InvalidAuthValues("username length should between 1 to 255"))?
+                }
+                let password_len = password.as_bytes().len();
+                if password_len < 1 || password_len > 255 {
+                    Err(Error::InvalidAuthValues("password length should between 1 to 255"))?
+                }
+            },
+            Authentication::None => {},
+        }
+        Ok(())
+    }
+
+    #[cfg(feature = "tor")]
+    /// Resolve the domain name to an ip using special Tor Resolve command, by
+    /// connecting to a Tor compatible proxy given a socket to it.
+    pub async fn tor_resolve_with_socket<'t, T>(socket: S, target: T) -> Result<TargetAddr<'static>>
+    where T: IntoTargetAddr<'t> {
+        let sock = Self::execute_command_with_socket(socket, target, Authentication::None, Command::TorResolve).await?;
+
+        Ok(sock.target_addr().to_owned())
+    }
+
+    #[cfg(feature = "tor")]
+    /// Perform a reverse DNS query on the given ip using special Tor Resolve
+    /// PTR command, by connecting to a Tor compatible proxy given a socket
+    /// to it.
+    pub async fn tor_resolve_ptr_with_socket<'t, T>(socket: S, target: T) -> Result<TargetAddr<'static>>
+    where T: IntoTargetAddr<'t> {
+        let sock =
+            Self::execute_command_with_socket(socket, target, Authentication::None, Command::TorResolvePtr).await?;
+
+        Ok(sock.target_addr().to_owned())
+    }
+
+    async fn execute_command_with_socket<'a, 't, T>(
+        socket: S,
+        target: T,
+        auth: Authentication<'a>,
+        command: Command,
+    ) -> Result<Socks5Stream<S>>
+    where
+        T: IntoTargetAddr<'t>,
+    {
+        Self::validate_auth(&auth)?;
+
+        let sock = SocksConnector::new(auth, command, stream::empty().fuse(), target.into_target_addr()?)
+            .execute_with_socket(socket)
+            .await?;
+
+        Ok(sock)
+    }
+
+    /// Consumes the `Socks5Stream`, returning the inner socket.
+    pub fn into_inner(self) -> S {
+        self.socket
+    }
+
+    /// Returns the target address that the proxy server connects to.
+    pub fn target_addr(&self) -> TargetAddr<'_> {
+        match &self.target {
+            TargetAddr::Ip(addr) => TargetAddr::Ip(*addr),
+            TargetAddr::Domain(domain, port) => {
+                let domain: &str = domain.borrow();
+                TargetAddr::Domain(domain.into(), *port)
+            },
+        }
+    }
+}
+
+/// A `Future` which resolves to a socket to the target server through proxy.
+pub struct SocksConnector<'a, 't, S> {
+    auth: Authentication<'a>,
+    command: Command,
+    proxy: Fuse<S>,
+    target: TargetAddr<'t>,
+    buf: [u8; 513],
+    ptr: usize,
+    len: usize,
+}
+
+impl<'a, 't, S> SocksConnector<'a, 't, S>
+where S: Stream<Item = Result<SocketAddr>> + Unpin
+{
+    fn new(auth: Authentication<'a>, command: Command, proxy: Fuse<S>, target: TargetAddr<'t>) -> Self {
+        SocksConnector {
+            auth,
+            command,
+            proxy,
+            target,
+            buf: [0; 513],
+            ptr: 0,
+            len: 0,
+        }
+    }
+
+    /// Connect to the proxy server, authenticate and issue the SOCKS command
+    pub async fn execute(&mut self) -> Result<Socks5Stream<TcpStream>> {
+        let next_addr = self.proxy.select_next_some().await?;
+        let tcp = TcpStream::connect(next_addr)
+            .await
+            .map_err(|_| Error::ProxyServerUnreachable)?;
+
+        self.execute_with_socket(tcp).await
+    }
+
+    pub async fn execute_with_socket<T: AsyncRead + AsyncWrite + Unpin>(
+        &mut self,
+        mut socket: T,
+    ) -> Result<Socks5Stream<T>>
+    {
+        self.authenticate(&mut socket).await?;
+
+        // Send request address that should be proxied
+        self.prepare_send_request();
+        socket.write_all(&self.buf[self.ptr..self.len]).await?;
+
+        let target = self.receive_reply(&mut socket).await?;
+
+        Ok(Socks5Stream { socket, target })
+    }
+
+    fn prepare_send_method_selection(&mut self) {
+        self.ptr = 0;
+        self.buf[0] = 0x05;
+        match self.auth {
+            Authentication::None => {
+                self.buf[1..3].copy_from_slice(&[1, 0x00]);
+                self.len = 3;
+            },
+            Authentication::Password { .. } => {
+                self.buf[1..4].copy_from_slice(&[2, 0x00, 0x02]);
+                self.len = 4;
+            },
+        }
+    }
+
+    fn prepare_recv_method_selection(&mut self) {
+        self.ptr = 0;
+        self.len = 2;
+    }
+
+    fn prepare_send_password_auth(&mut self) {
+        if let Authentication::Password { username, password } = self.auth {
+            self.ptr = 0;
+            self.buf[0] = 0x01;
+            let username_bytes = username.as_bytes();
+            let username_len = username_bytes.len();
+            self.buf[1] = username_len as u8;
+            self.buf[2..(2 + username_len)].copy_from_slice(username_bytes);
+            let password_bytes = password.as_bytes();
+            let password_len = password_bytes.len();
+            self.len = 3 + username_len + password_len;
+            self.buf[(2 + username_len)] = password_len as u8;
+            self.buf[(3 + username_len)..self.len].copy_from_slice(password_bytes);
+        } else {
+            unreachable!()
+        }
+    }
+
+    fn prepare_recv_password_auth(&mut self) {
+        self.ptr = 0;
+        self.len = 2;
+    }
+
+    fn prepare_send_request(&mut self) {
+        self.ptr = 0;
+        self.buf[..3].copy_from_slice(&[0x05, self.command as u8, 0x00]);
+        match &self.target {
+            TargetAddr::Ip(SocketAddr::V4(addr)) => {
+                self.buf[3] = 0x01;
+                self.buf[4..8].copy_from_slice(&addr.ip().octets());
+                self.buf[8..10].copy_from_slice(&addr.port().to_be_bytes());
+                self.len = 10;
+            },
+            TargetAddr::Ip(SocketAddr::V6(addr)) => {
+                self.buf[3] = 0x04;
+                self.buf[4..20].copy_from_slice(&addr.ip().octets());
+                self.buf[20..22].copy_from_slice(&addr.port().to_be_bytes());
+                self.len = 22;
+            },
+            TargetAddr::Domain(domain, port) => {
+                self.buf[3] = 0x03;
+                let domain = domain.as_bytes();
+                let len = domain.len();
+                self.buf[4] = len as u8;
+                self.buf[5..5 + len].copy_from_slice(domain);
+                self.buf[(5 + len)..(7 + len)].copy_from_slice(&port.to_be_bytes());
+                self.len = 7 + len;
+            },
+        }
+    }
+
+    fn prepare_recv_reply(&mut self) {
+        self.ptr = 0;
+        self.len = 4;
+    }
+
+    async fn password_authentication_protocol<T: AsyncRead + AsyncWrite + Unpin>(&mut self, tcp: &mut T) -> Result<()> {
+        if let Authentication::None = self.auth {
+            return Err(Error::AuthorizationRequired);
+        }
+
+        self.prepare_send_password_auth();
+        tcp.write_all(&self.buf[self.ptr..self.len]).await?;
+
+        self.prepare_recv_password_auth();
+        tcp.read_exact(&mut self.buf[self.ptr..self.len]).await?;
+
+        if self.buf[0] != 0x01 {
+            return Err(Error::InvalidResponseVersion);
+        }
+        if self.buf[1] != 0x00 {
+            return Err(Error::PasswordAuthFailure(self.buf[1]));
+        }
+
+        Ok(())
+    }
+
+    async fn authenticate<T: AsyncRead + AsyncWrite + Unpin>(&mut self, tcp: &mut T) -> Result<()> {
+        // Write request to connect/authenticate
+        self.prepare_send_method_selection();
+        tcp.write_all(&self.buf[self.ptr..self.len]).await?;
+
+        // Receive authentication method
+        self.prepare_recv_method_selection();
+        tcp.read_exact(&mut self.buf[self.ptr..self.len]).await?;
+        if self.buf[0] != 0x05 {
+            return Err(Error::InvalidResponseVersion);
+        }
+        match self.buf[1] {
+            0x00 => {
+                // No auth
+            },
+            0x02 => {
+                self.password_authentication_protocol(tcp).await?;
+            },
+            0xff => {
+                return Err(Error::NoAcceptableAuthMethods);
+            },
+            m if m != self.auth.id() => return Err(Error::UnknownAuthMethod),
+            _ => unimplemented!(),
+        }
+
+        Ok(())
+    }
+
+    async fn receive_reply<T: AsyncRead + AsyncWrite + Unpin>(&mut self, tcp: &mut T) -> Result<TargetAddr<'static>> {
+        self.prepare_recv_reply();
+        self.ptr += tcp.read_exact(&mut self.buf[self.ptr..self.len]).await?;
+        if self.buf[0] != 0x05 {
+            return Err(Error::InvalidResponseVersion);
+        }
+        if self.buf[2] != 0x00 {
+            return Err(Error::InvalidReservedByte);
+        }
+
+        match self.buf[1] {
+            0x00 => {}, // succeeded
+            0x01 => Err(Error::GeneralSocksServerFailure)?,
+            0x02 => Err(Error::ConnectionNotAllowedByRuleset)?,
+            0x03 => Err(Error::NetworkUnreachable)?,
+            0x04 => Err(Error::HostUnreachable)?,
+            0x05 => Err(Error::ConnectionRefused)?,
+            0x06 => Err(Error::TtlExpired)?,
+            0x07 => Err(Error::CommandNotSupported)?,
+            0x08 => Err(Error::AddressTypeNotSupported)?,
+            _ => Err(Error::UnknownAuthMethod)?,
+        }
+
+        match self.buf[3] {
+            // IPv4
+            0x01 => {
+                self.len = 10;
+            },
+            // IPv6
+            0x04 => {
+                self.len = 22;
+            },
+            // Domain
+            0x03 => {
+                self.len = 5;
+                self.ptr += tcp.read_exact(&mut self.buf[self.ptr..self.len]).await?;
+                self.len += self.buf[4] as usize + 2;
+            },
+            _ => Err(Error::UnknownAddressType)?,
+        }
+
+        self.ptr += tcp.read_exact(&mut self.buf[self.ptr..self.len]).await?;
+        let target: TargetAddr<'static> = match self.buf[3] {
+            // IPv4
+            0x01 => {
+                let mut ip = [0; 4];
+                ip[..].copy_from_slice(&self.buf[4..8]);
+                let ip = Ipv4Addr::from(ip);
+                let port = u16::from_be_bytes([self.buf[8], self.buf[9]]);
+                (ip, port).into_target_addr()?
+            },
+            // IPv6
+            0x04 => {
+                let mut ip = [0; 16];
+                ip[..].copy_from_slice(&self.buf[4..20]);
+                let ip = Ipv6Addr::from(ip);
+                let port = u16::from_be_bytes([self.buf[20], self.buf[21]]);
+                (ip, port).into_target_addr()?
+            },
+            // Domain
+            0x03 => {
+                let domain_bytes = (&self.buf[5..(self.len - 2)]).to_vec();
+                let domain = String::from_utf8(domain_bytes)
+                    .map_err(|_| Error::InvalidTargetAddress("not a valid UTF-8 string"))?;
+                let port = u16::from_be_bytes([self.buf[self.len - 2], self.buf[self.len - 1]]);
+                TargetAddr::Domain(domain.into(), port)
+            },
+            _ => unreachable!(),
+        };
+
+        Ok(target)
+    }
+}
+
+/// A SOCKS5 BIND client.
+///
+/// Once you get an instance of `Socks5Listener`, you should send the
+/// `bind_addr` to the remote process via the primary connection. Then, call the
+/// `accept` function and wait for the other end connecting to the rendezvous
+/// address.
+pub struct Socks5Listener<S> {
+    inner: Socks5Stream<S>,
+}
+
+impl Socks5Listener<TcpStream> {
+    /// Initiates a BIND request to the specified proxy.
+    ///
+    /// The proxy will filter incoming connections based on the value of
+    /// `target`.
+    ///
+    /// # Error
+    ///
+    /// It propagates the error that occurs in the conversion from `T` to
+    /// `TargetAddr`.
+    pub async fn bind<'t, P, T>(proxy: P, target: T) -> Result<Socks5Listener<TcpStream>>
+    where
+        P: ToProxyAddrs,
+        T: IntoTargetAddr<'t>,
+    {
+        Self::bind_with_auth(Authentication::None, proxy, target).await
+    }
+
+    /// Initiates a BIND request to the specified proxy using given username
+    /// and password.
+    ///
+    /// The proxy will filter incoming connections based on the value of
+    /// `target`.
+    ///
+    /// # Error
+    ///
+    /// It propagates the error that occurs in the conversion from `T` to
+    /// `TargetAddr`.
+    pub async fn bind_with_password<'a, 't, P, T>(
+        proxy: P,
+        target: T,
+        username: &'a str,
+        password: &'a str,
+    ) -> Result<Socks5Listener<TcpStream>>
+    where
+        P: ToProxyAddrs,
+        T: IntoTargetAddr<'t>,
+    {
+        Self::bind_with_auth(Authentication::Password { username, password }, proxy, target).await
+    }
+
+    async fn bind_with_auth<'t, P, T>(
+        auth: Authentication<'_>,
+        proxy: P,
+        target: T,
+    ) -> Result<Socks5Listener<TcpStream>>
+    where
+        P: ToProxyAddrs,
+        T: IntoTargetAddr<'t>,
+    {
+        let socket = SocksConnector::new(
+            auth,
+            Command::Bind,
+            proxy.to_proxy_addrs().fuse(),
+            target.into_target_addr()?,
+        )
+        .execute()
+        .await?;
+
+        Ok(Socks5Listener { inner: socket })
+    }
+}
+
+impl<S> Socks5Listener<S>
+where S: AsyncRead + AsyncWrite + Unpin
+{
+    /// Initiates a BIND request to the specified proxy using the given socket
+    /// to it.
+    ///
+    /// The proxy will filter incoming connections based on the value of
+    /// `target`.
+    ///
+    /// # Error
+    ///
+    /// It propagates the error that occurs in the conversion from `T` to
+    /// `TargetAddr`.
+    pub async fn bind_with_socket<'t, T>(socket: S, target: T) -> Result<Socks5Listener<S>>
+    where T: IntoTargetAddr<'t> {
+        Self::bind_with_auth_and_socket(Authentication::None, socket, target).await
+    }
+
+    /// Initiates a BIND request to the specified proxy using given username,
+    /// password and socket to the proxy.
+    ///
+    /// The proxy will filter incoming connections based on the value of
+    /// `target`.
+    ///
+    /// # Error
+    ///
+    /// It propagates the error that occurs in the conversion from `T` to
+    /// `TargetAddr`.
+    pub async fn bind_with_password_and_socket<'a, 't, T>(
+        socket: S,
+        target: T,
+        username: &'a str,
+        password: &'a str,
+    ) -> Result<Socks5Listener<S>>
+    where
+        T: IntoTargetAddr<'t>,
+    {
+        Self::bind_with_auth_and_socket(Authentication::Password { username, password }, socket, target).await
+    }
+
+    async fn bind_with_auth_and_socket<'t, T>(
+        auth: Authentication<'_>,
+        socket: S,
+        target: T,
+    ) -> Result<Socks5Listener<S>>
+    where
+        T: IntoTargetAddr<'t>,
+    {
+        let socket = SocksConnector::new(auth, Command::Bind, stream::empty().fuse(), target.into_target_addr()?)
+            .execute_with_socket(socket)
+            .await?;
+
+        Ok(Socks5Listener { inner: socket })
+    }
+
+    /// Returns the address of the proxy-side TCP listener.
+    ///
+    /// This should be forwarded to the remote process, which should open a
+    /// connection to it.
+    pub fn bind_addr(&self) -> TargetAddr {
+        self.inner.target_addr()
+    }
+
+    /// Consumes this listener, returning a `Future` which resolves to the
+    /// `Socks5Stream` connected to the target server through the proxy.
+    ///
+    /// The value of `bind_addr` should be forwarded to the remote process
+    /// before this method is called.
+    pub async fn accept(mut self) -> Result<Socks5Stream<S>> {
+        let mut connector = SocksConnector {
+            auth: Authentication::None,
+            command: Command::Bind,
+            proxy: stream::empty().fuse(),
+            target: self.inner.target,
+            buf: [0; 513],
+            ptr: 0,
+            len: 0,
+        };
+
+        let target = connector.receive_reply(&mut self.inner.socket).await?;
+
+        Ok(Socks5Stream {
+            socket: self.inner.socket,
+            target,
+        })
+    }
+}
+
+impl<T> AsyncRead for Socks5Stream<T>
+where T: AsyncRead + Unpin
+{
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
+        AsyncRead::poll_read(Pin::new(&mut self.socket), cx, buf)
+    }
+}
+
+impl<T> AsyncWrite for Socks5Stream<T>
+where T: AsyncWrite + Unpin
+{
+    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
+        AsyncWrite::poll_write(Pin::new(&mut self.socket), cx, buf)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        AsyncWrite::poll_flush(Pin::new(&mut self.socket), cx)
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        AsyncWrite::poll_shutdown(Pin::new(&mut self.socket), cx)
+    }
+}

tests/common.rs

@@ -0,0 +1,74 @@
+use once_cell::sync::OnceCell;
+use std::{
+    io::{Read, Write},
+    net::{SocketAddr, TcpStream as StdTcpStream},
+    sync::Mutex,
+};
+use tokio::{
+    io::{copy, split, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt},
+    net::{TcpListener, UnixStream},
+    runtime::Runtime,
+};
+use tokio_socks::{
+    tcp::{Socks5Listener, Socks5Stream},
+    Error,
+    Result,
+};
+
+pub const UNIX_PROXY_ADDR: &'static str = "/tmp/proxy.s";
+pub const PROXY_ADDR: &'static str = "127.0.0.1:41080";
+pub const ECHO_SERVER_ADDR: &'static str = "localhost:10007";
+pub const MSG: &[u8] = b"hello";
+
+pub async fn echo_server() -> Result<()> {
+    let listener = TcpListener::bind(&SocketAddr::from(([0, 0, 0, 0], 10007))).await?;
+    loop {
+        let (mut stream, _) = listener.accept().await?;
+        tokio::spawn(async move {
+            let (mut reader, mut writer) = stream.split();
+            copy(&mut reader, &mut writer).await.unwrap();
+        });
+    }
+}
+
+pub async fn reply_response<S: AsyncRead + AsyncWrite + Unpin>(mut socket: Socks5Stream<S>) -> Result<[u8; 5]> {
+    socket.write_all(MSG).await?;
+    let mut buf = [0; 5];
+    socket.read_exact(&mut buf).await?;
+    Ok(buf)
+}
+
+pub async fn test_connect<S: AsyncRead + AsyncWrite + Unpin>(socket: Socks5Stream<S>) -> Result<()> {
+    let res = reply_response(socket).await?;
+    assert_eq!(&res[..], MSG);
+    Ok(())
+}
+
+pub fn test_bind<S: 'static + AsyncRead + AsyncWrite + Unpin + Send>(listener: Socks5Listener<S>) -> Result<()> {
+    let bind_addr = listener.bind_addr().to_owned();
+    runtime().lock().unwrap().spawn(async move {
+        let stream = listener.accept().await.unwrap();
+        let (mut reader, mut writer) = split(stream);
+        copy(&mut reader, &mut writer).await.unwrap();
+    });
+
+    let mut tcp = StdTcpStream::connect(bind_addr)?;
+    tcp.write_all(MSG)?;
+    let mut buf = [0; 5];
+    tcp.read_exact(&mut buf[..])?;
+    assert_eq!(&buf[..], MSG);
+    Ok(())
+}
+
+pub async fn connect_unix() -> Result<UnixStream> {
+    UnixStream::connect(UNIX_PROXY_ADDR).await.map_err(Error::Io)
+}
+
+pub fn runtime() -> &'static Mutex<Runtime> {
+    static RUNTIME: OnceCell<Mutex<Runtime>> = OnceCell::new();
+    RUNTIME.get_or_init(|| {
+        let runtime = Runtime::new().expect("Unable to create runtime");
+        runtime.spawn(async { echo_server().await.expect("Unable to bind") });
+        Mutex::new(runtime)
+    })
+}

tests/integration_tests.sh

@@ -0,0 +1,35 @@
+#!/usr/bin/env bash
+set -x
+
+dir="$(dirname "$(which "$0")")"
+SOCK="/tmp/proxy.s"
+PROXY_HOST="127.0.0.1:41080"
+
+
+#socat tcp-listen:10007,fork exec:cat &
+#echo $! > /tmp/socat-test.pid
+
+if test -z "$@"; then
+    list="no_auth username_auth long_username_password_auth"
+else
+    list="$@"
+fi
+
+socat UNIX-LISTEN:${SOCK},reuseaddr,fork TCP:${PROXY_HOST} &
+
+for test in ${list}; do
+    3proxy ${dir}/${test}.cfg
+    sleep 1
+    cargo test --test ${test} -- --test-threads 1
+    test_exit_code=$?
+
+    pkill -F /tmp/3proxy-test.pid
+
+    if test "$test_exit_code" -ne 0; then
+        break
+    fi
+done
+
+
+#pkill -F /tmp/socat-test.pid
+exit ${test_exit_code}

tests/long_username_password_auth.cfg

@@ -0,0 +1,6 @@
+daemon
+users mylonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglogin:CL:longlonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglongpassword
+pidfile /tmp/3proxy-test.pid
+auth strong
+allow mylonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglogin
+socks -p41080

tests/long_username_password_auth.rs

@@ -0,0 +1,55 @@
+mod common;
+
+use common::{connect_unix, runtime, test_bind, test_connect, ECHO_SERVER_ADDR, PROXY_ADDR};
+use tokio_socks::{
+    tcp::{Socks5Listener, Socks5Stream},
+    Result,
+};
+
+#[test]
+fn connect_long_username_password() -> Result<()> {
+    let runtime = runtime().lock().unwrap();
+    let conn = runtime.block_on(Socks5Stream::connect_with_password(
+        PROXY_ADDR, ECHO_SERVER_ADDR, "mylonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglogin",
+                                                                    "longlonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglongpassword"))?;
+    runtime.block_on(test_connect(conn))
+}
+
+#[test]
+fn bind_long_username_password() -> Result<()> {
+    let bind = {
+        let runtime = runtime().lock().unwrap();
+        runtime.block_on(Socks5Listener::bind_with_password(
+            PROXY_ADDR,
+            ECHO_SERVER_ADDR,
+            "mylonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglogin",
+            "longlonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglongpassword"
+        ))
+    }?;
+    test_bind(bind)
+}
+
+#[test]
+fn connect_with_socket_long_username_password() -> Result<()> {
+    let runtime = runtime().lock().unwrap();
+    let socket = runtime.block_on(connect_unix())?;
+    let conn = runtime.block_on(Socks5Stream::connect_with_password_and_socket(
+        socket, ECHO_SERVER_ADDR, "mylonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglogin",
+                                                                    "longlonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglongpassword"))?;
+    runtime.block_on(test_connect(conn))
+}
+
+#[test]
+fn bind_with_socket_long_username_password() -> Result<()> {
+    let bind = {
+        let runtime = runtime().lock().unwrap();
+        let socket = runtime.block_on(connect_unix())?;
+        runtime.block_on(Socks5Listener::bind_with_password_and_socket(
+            socket,
+            ECHO_SERVER_ADDR,
+            "mylonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglogin",
+            "longlonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglonglongpassword"
+        ))
+    }?;
+    test_bind(bind)
+}

tests/no_auth.cfg

@@ -0,0 +1,4 @@
+daemon
+pidfile /tmp/3proxy-test.pid
+auth none
+socks -p41080

tests/no_auth.rs

@@ -0,0 +1,42 @@
+mod common;
+
+use crate::common::{runtime, test_bind};
+use common::{connect_unix, test_connect, ECHO_SERVER_ADDR, PROXY_ADDR};
+use tokio_socks::{
+    tcp::{Socks5Listener, Socks5Stream},
+    Result,
+};
+
+#[test]
+fn connect_no_auth() -> Result<()> {
+    let runtime = runtime().lock().unwrap();
+    let conn = runtime.block_on(Socks5Stream::connect(PROXY_ADDR, ECHO_SERVER_ADDR))?;
+    runtime.block_on(test_connect(conn))
+}
+
+#[test]
+fn bind_no_auth() -> Result<()> {
+    let bind = {
+        let runtime = runtime().lock().unwrap();
+        runtime.block_on(Socks5Listener::bind(PROXY_ADDR, ECHO_SERVER_ADDR))
+    }?;
+    test_bind(bind)
+}
+
+#[test]
+fn connect_with_socket_no_auth() -> Result<()> {
+    let runtime = runtime().lock().unwrap();
+    let socket = runtime.block_on(connect_unix())?;
+    let conn = runtime.block_on(Socks5Stream::connect_with_socket(socket, ECHO_SERVER_ADDR))?;
+    runtime.block_on(test_connect(conn))
+}
+
+#[test]
+fn bind_with_socket_no_auth() -> Result<()> {
+    let bind = {
+        let runtime = runtime().lock().unwrap();
+        let socket = runtime.block_on(connect_unix())?;
+        runtime.block_on(Socks5Listener::bind_with_socket(socket, ECHO_SERVER_ADDR))
+    }?;
+    test_bind(bind)
+}

tests/username_auth.cfg

@@ -0,0 +1,6 @@
+daemon
+users mylogin:CL:mypassword
+pidfile /tmp/3proxy-test.pid
+auth strong
+allow mylogin
+socks -p41080

tests/username_auth.rs

@@ -0,0 +1,61 @@
+mod common;
+
+use common::{connect_unix, runtime, test_bind, test_connect, ECHO_SERVER_ADDR, PROXY_ADDR};
+use tokio_socks::{
+    tcp::{Socks5Listener, Socks5Stream},
+    Result,
+};
+
+#[test]
+fn connect_username_auth() -> Result<()> {
+    let runtime = runtime().lock().unwrap();
+    let conn = runtime.block_on(Socks5Stream::connect_with_password(
+        PROXY_ADDR,
+        ECHO_SERVER_ADDR,
+        "mylogin",
+        "mypassword",
+    ))?;
+    runtime.block_on(test_connect(conn))
+}
+
+#[test]
+fn bind_username_auth() -> Result<()> {
+    let bind = {
+        let runtime = runtime().lock().unwrap();
+        runtime.block_on(Socks5Listener::bind_with_password(
+            PROXY_ADDR,
+            ECHO_SERVER_ADDR,
+            "mylogin",
+            "mypassword",
+        ))
+    }?;
+    test_bind(bind)
+}
+
+#[test]
+fn connect_with_socket_username_auth() -> Result<()> {
+    let runtime = runtime().lock().unwrap();
+    let socket = runtime.block_on(connect_unix())?;
+    let conn = runtime.block_on(Socks5Stream::connect_with_password_and_socket(
+        socket,
+        ECHO_SERVER_ADDR,
+        "mylogin",
+        "mypassword",
+    ))?;
+    runtime.block_on(test_connect(conn))
+}
+
+#[test]
+fn bind_with_socket_username_auth() -> Result<()> {
+    let bind = {
+        let runtime = runtime().lock().unwrap();
+        let socket = runtime.block_on(connect_unix())?;
+        runtime.block_on(Socks5Listener::bind_with_password_and_socket(
+            socket,
+            ECHO_SERVER_ADDR,
+            "mylogin",
+            "mypassword",
+        ))
+    }?;
+    test_bind(bind)
+}