Merge pull request #9 from dizda/feat-migrate-tokio

Feat migrate tokio

Cargo.toml

@@ -1,23 +1,24 @@
 [package]
 name = "fast-socks5"
-version = "0.4.3"
+version = "0.5.0"
 authors = ["Jonathan Dizdarevic <[email protected]>"]
 edition = "2018"
 license = "MIT"
 description = "Fast SOCKS5 client/server implementation written in Rust async/.await (with async-std)"
 repository = "https://github.com/dizda/fast-socks5"
 
 [dependencies]
-futures = "0.3.8"
 log = "0.4"
-async-std = { version = "1.9.0", features = ["std"] }
+tokio = { version = "1.12.0", features = ["io-util", "net", "time"] }
 anyhow = "1.0"
 thiserror = "1.0"
+tokio-stream = "0.1.7"
 
 # Dependencies for examples/
 [dev-dependencies]
 env_logger = "0.7"
 structopt = "0.3"
+tokio = { version = "1.12.0", features = ["io-util", "net", "time", "rt-multi-thread", "macros"] }
 
 [[example]]
 name = "server"

LICENSE

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2020 Jonathan Dizdarevic
+Copyright (c) 2021 Jonathan Dizdarevic
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

README.md

@@ -10,7 +10,7 @@
 
 - An `async`/`.await` [SOCKS5](https://tools.ietf.org/html/rfc1928) implementation.
 - No **unsafe** code
-- Built on-top of `async-std` library
+- Built on-top of `tokio` library
 - Ultra lightweight and scalable
 - No system dependencies
 - Cross-platform
@@ -22,11 +22,11 @@
 - `AsyncRead + AsyncWrite` traits are implemented on Socks5Stream & Socks5Socket
 - `IPv4`, `IPv6`, and `Domains` types are supported
 - Config helper for Socks5Server
-- Helpers to run a Socks5Server à la *"async-std's TcpStream"* via `incoming.next().await`
+- Helpers to run a Socks5Server à la *"std's TcpStream"* via `incoming.next().await`
 - Examples come with real cases commands scenarios
 - Can disable `DNS resolving`
-- Can skip the authentication/handshake process, which will directly handle command's request (useful to save useless round-trips in an already authenticated environment)
-- Can disable command execution (useful if you just want to forward the request to an another server)
+- Can skip the authentication/handshake process, which will directly handle command's request (useful to save useless round-trips in a current authenticated environment)
+- Can disable command execution (useful if you just want to forward the request to a different server)
 
 
 ## Install

examples/client.rs

@@ -3,11 +3,10 @@
 extern crate log;
 
 use anyhow::Context;
-use async_std::task;
 use fast_socks5::client::Config;
 use fast_socks5::{client::Socks5Stream, Result};
-use futures::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use structopt::StructOpt;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 
 /// # How to use it:
 ///
@@ -46,10 +45,11 @@ struct Opt {
     pub skip_auth: bool,
 }
 
-fn main() -> Result<()> {
+#[tokio::main]
+async fn main() -> Result<()> {
     env_logger::init();
 
-    task::block_on(spawn_socks_client())
+    spawn_socks_client().await
 }
 
 async fn spawn_socks_client() -> Result<()> {
@@ -112,7 +112,10 @@ async fn http_request<T: AsyncRead + AsyncWrite + Unpin>(
         .context("Can't read HTTP Response")?;
 
     info!("Response: {}", String::from_utf8_lossy(&result));
-    assert!(result.starts_with(b"HTTP/1.1"));
+
+    if result.starts_with(b"HTTP/1.1") {
+        info!("HTTP/1.1 Response detected!");
+    }
     //assert!(result.ends_with(b"</HTML>\r\n") || result.ends_with(b"</html>"));
 
     Ok(())

examples/server.rs

@@ -2,13 +2,15 @@
 #[macro_use]
 extern crate log;
 
-use async_std::{future::Future, stream::StreamExt, task};
 use fast_socks5::{
     server::{Config, SimpleUserPassword, Socks5Server, Socks5Socket},
     Result, SocksError,
 };
-use futures::{AsyncRead, AsyncWrite};
+use std::future::Future;
 use structopt::StructOpt;
+use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::task;
+use tokio_stream::StreamExt;
 
 /// # How to use it:
 ///
@@ -61,10 +63,11 @@ enum AuthMode {
 ///
 /// TODO: Write functional tests: https://github.com/ark0f/async-socks5/blob/master/src/lib.rs#L762
 /// TODO: Write functional tests with cURL?
-fn main() -> Result<()> {
+#[tokio::main]
+async fn main() -> Result<()> {
     env_logger::init();
 
-    task::block_on(spawn_socks_server())
+    spawn_socks_server().await
 }
 
 async fn spawn_socks_server() -> Result<()> {
@@ -79,7 +82,7 @@ async fn spawn_socks_server() -> Result<()> {
             if opt.skip_auth {
                 return Err(SocksError::ArgumentInputError(
                     "Can't use skip-auth flag and authentication altogether.",
-                ))?;
+                ));
             }
 
             config.set_authentication(SimpleUserPassword { username, password });

examples/simple_tcp_server.rs

@@ -2,15 +2,18 @@
 #[macro_use]
 extern crate log;
 
-use async_std::net::TcpListener;
-use async_std::sync::Arc;
-use async_std::{future::Future, stream::StreamExt, task};
 use fast_socks5::{
-    server::{Config, SimpleUserPassword, Socks5Server, Socks5Socket},
+    server::{Config, SimpleUserPassword, Socks5Socket},
     Result,
 };
-use futures::{AsyncRead, AsyncWrite};
+use std::future::Future;
+use std::sync::Arc;
 use structopt::StructOpt;
+use tokio::task;
+use tokio::{
+    io::{AsyncRead, AsyncWrite},
+    net::TcpListener,
+};
 
 /// # How to use it:
 ///
@@ -61,10 +64,11 @@ enum AuthMode {
 /// TODO: Write functional tests: https://github.com/ark0f/async-socks5/blob/master/src/lib.rs#L762
 /// TODO: Write functional tests with cURL?
 /// TODO: Move this to as a standalone library
-fn main() -> Result<()> {
+#[tokio::main]
+async fn main() -> Result<()> {
     env_logger::init();
 
-    task::block_on(spawn_socks_server())
+    spawn_socks_server().await
 }
 
 async fn spawn_socks_server() -> Result<()> {
@@ -82,30 +86,21 @@ async fn spawn_socks_server() -> Result<()> {
 
     let config = Arc::new(config);
 
-    let mut listener = TcpListener::bind(&opt.listen_addr).await?;
+    let listener = TcpListener::bind(&opt.listen_addr).await?;
     //    listener.set_config(config);
 
-    let mut incoming = listener.incoming();
-
     info!("Listen for socks connections @ {}", &opt.listen_addr);
 
     // Standard TCP loop
-    while let Some(socket_res) = incoming.next().await {
-        match socket_res {
-            Ok(socket) => {
+    loop {
+        match listener.accept().await {
+            Ok((socket, _addr)) => {
                 info!("Connection from {}", socket.peer_addr()?);
                 let socket = Socks5Socket::new(socket, config.clone());
 
-                //                                socket.upgrade_to_socks5().await;
                 spawn_and_log_error(socket.upgrade_to_socks5());
-                //                match socket.upgrade_to_socks5().await {
-                //                    Ok(_) => {}
-                //                    Err(e) => error!("{:#}", &e),
-                //                }
-            }
-            Err(err) => {
-                error!("accept error = {:?}", err);
             }
+            Err(err) => error!("accept error = {:?}", err),
         }
     }
 

src/client.rs

@@ -3,10 +3,12 @@ use crate::read_exact;
 use crate::util::target_addr::{read_address, TargetAddr, ToTargetAddr};
 use crate::{consts, AuthenticationMethod, ReplyError, Result, SocksError};
 use anyhow::Context;
-use async_std::net::{SocketAddr, TcpStream, ToSocketAddrs};
-use futures::{task::Poll, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use std::io;
+use std::net::SocketAddr;
 use std::pin::Pin;
+use std::task::Poll;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+use tokio::net::{TcpStream, ToSocketAddrs};
 
 const MAX_ADDR_LEN: usize = 260;
 
@@ -65,7 +67,7 @@ where
         }
 
         // Handshake Lifecycle
-        if stream.config.skip_auth == false {
+        if !stream.config.skip_auth {
             let methods = stream.send_version_and_methods(methods).await?;
             stream.which_method_accepted(methods).await?;
         } else {
@@ -171,7 +173,7 @@ where
                     .await
                     .context("Can't write that the methods are unsupported.")?;
 
-                return Err(SocksError::AuthMethodUnacceptable(vec![method]))?;
+                return Err(SocksError::AuthMethodUnacceptable(vec![method]));
             }
         }
 
@@ -287,7 +289,7 @@ where
             TargetAddr::Domain(ref domain, port) => {
                 debug!("TargetAddr::Domain");
                 if domain.len() > u8::max_value() as usize {
-                    return Err(SocksError::ExceededMaxDomainLen(domain.len()))?;
+                    return Err(SocksError::ExceededMaxDomainLen(domain.len()));
                 }
                 padding = 5 + domain.len() + 2;
 
@@ -337,7 +339,7 @@ where
         }
 
         if reply != consts::SOCKS5_REPLY_SUCCEEDED {
-            return Err(ReplyError::from_u8(reply))?; // Convert reply received into correct error
+            return Err(ReplyError::from_u8(reply).into()); // Convert reply received into correct error
         }
 
         let address = read_address(&mut self.socket, address_type).await?;
@@ -427,8 +429,8 @@ where
     fn poll_read(
         mut self: Pin<&mut Self>,
         context: &mut std::task::Context,
-        buf: &mut [u8],
-    ) -> Poll<io::Result<usize>> {
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> Poll<std::io::Result<()>> {
         Pin::new(&mut self.socket).poll_read(context, buf)
     }
 }
@@ -453,10 +455,10 @@ where
         Pin::new(&mut self.socket).poll_flush(context)
     }
 
-    fn poll_close(
+    fn poll_shutdown(
         mut self: Pin<&mut Self>,
         context: &mut std::task::Context,
     ) -> Poll<io::Result<()>> {
-        Pin::new(&mut self.socket).poll_close(context)
+        Pin::new(&mut self.socket).poll_shutdown(context)
     }
 }

src/lib.rs

@@ -91,8 +91,6 @@ impl fmt::Display for AuthenticationMethod {
 pub enum SocksError {
     #[error("i/o error: {0}")]
     Io(#[from] io::Error),
-    #[error("request timeout: {0}")]
-    FutureTimeout(#[from] async_std::future::TimeoutError),
     #[error("the data for key `{0}` is not available")]
     Redaction(String),
     #[error("invalid header (expected {expected:?}, found {found:?})")]