Update deps and migrate to fast-socks5 v1.0.0-rc.0: Replace deprecated APIs

Cargo.toml

@@ -1,25 +1,25 @@
 [package]
 name = "socks5ws"
-version = "0.1.0"
-edition = "2021"
+version = "0.1.1"
+edition = "2024"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-anyhow = "1.0.75"
-async-stream = "0.3.3"
-clap = { version = "4.3.21", features = ["derive"] }
-ctrlc = "3.2.3"
-fast-socks5 = "0.8.1"
-flexi_logger = { version = "0.25.6", features = ["specfile_without_notification", "async"] }
-log = "0.4.20"
-serde = { version = "1.0.183", features = ["derive"] }
-serde_derive = "1.0.183"
-tokio = { version = "1.38.2", features = ["io-std", "net", "rt-multi-thread", "macros"] }
-tokio-stream = "0.1.11"
-tokio-util = "0.7.4"
-toml = "0.7.4"
-windows-service = "0.6.0"
+anyhow = "1.0.98"
+async-stream = "0.3.6"
+clap = { version = "4.5.42", features = ["derive"] }
+ctrlc = "3.4.7"
+fast-socks5 = "=1.0.0-rc.0"
+flexi_logger = { version = "0.31.2", features = ["specfile_without_notification", "async"] }
+log = "0.4.27"
+serde = { version = "1.0.219", features = ["derive"] }
+serde_derive = "1.0.219"
+tokio = { version = "1.47.0", features = ["io-std", "net", "rt-multi-thread", "macros"] }
+tokio-stream = { version = "0.1.17", features = ["net"]}
+tokio-util = "0.7.15"
+toml = "0.9.4"
+windows-service = "0.8.0"
 
 [profile.release]
 opt-level = 3

src/config.rs

@@ -3,7 +3,7 @@ use std::{
     path::{Path, PathBuf},
 };
 
-use anyhow::{Context, Result};
+use anyhow::{Context, Result, anyhow};
 use serde_derive::{Deserialize, Serialize};
 
 #[derive(Clone, Serialize, Deserialize, Debug)]
@@ -11,6 +11,9 @@ pub struct Config {
     /// Bind on address address. eg. `127.0.0.1:1080`
     #[serde(default = "default_listen_addr")]
     pub listen_addr: String,
+    /// Our external IP address to be sent in reply packets (required for UDP)
+    #[serde(default)]
+    pub public_addr: Option<std::net::IpAddr>,
     /// Request timeout
     #[serde(default = "default_timeout")]
     pub request_timeout: u64,
@@ -20,21 +23,11 @@ pub struct Config {
     /// Avoid useless roundtrips if we don't need the Authentication layer
     #[serde(default)]
     pub skip_auth: bool,
-    /// Enable dns-resolving
-    #[serde(default = "default_true")]
-    pub dns_resolve: bool,
-    /// Enable command execution
-    #[serde(default = "default_true")]
-    pub execute_command: bool,
     /// Enable UDP support
-    #[serde(default = "default_true")]
+    #[serde(default)]
     pub allow_udp: bool,
 }
 
-fn default_true() -> bool {
-    true
-}
-
 fn default_timeout() -> u64 {
     60
 }
@@ -54,12 +47,11 @@ impl Default for Config {
     fn default() -> Self {
         Config {
             listen_addr: default_listen_addr(),
+            public_addr: None,
             request_timeout: default_timeout(),
             auth: None,
             skip_auth: false,
-            dns_resolve: true,
-            execute_command: true,
-            allow_udp: true,
+            allow_udp: false,
         }
     }
 }
@@ -111,4 +103,17 @@ impl Config {
             log::info!(r#"config saved to: "{}""#, path.to_str().unwrap());
         }
     }
+
+    pub fn validate(&self) -> Result<()> {
+        if self.allow_udp && self.public_addr.is_none() {
+            return Err(anyhow!("Can't allow UDP if public-addr is not set"));
+        }
+        if self.skip_auth && self.auth.is_some() {
+            return Err(anyhow!(
+                "Can't use skip-auth flag and authentication altogether."
+            ));
+        }
+
+        Ok(())
+    }
 }

src/server.rs

@@ -1,10 +1,11 @@
-use anyhow::{anyhow, Result};
-use fast_socks5::server::{SimpleUserPassword, Socks5Server, Socks5Socket};
+use anyhow::{Context, Result, anyhow};
+use fast_socks5::server::{DnsResolveHelper, Socks5ServerProtocol, run_tcp_proxy, run_udp_proxy};
+use fast_socks5::{ReplyError, Socks5Command, SocksError};
 use std::future::Future;
-use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::net::TcpListener;
 use tokio::select;
 use tokio::task;
-use tokio_stream::{Stream, StreamExt};
+use tokio_stream::{Stream, StreamExt, wrappers::TcpListenerStream};
 use tokio_util::sync::CancellationToken;
 
 use async_stream::stream;
@@ -20,24 +21,11 @@ pub fn server_executor(cfg: Config, token: CancellationToken) -> Result<()> {
 }
 
 pub async fn spawn_socks5_server(cfg: Config, token: CancellationToken) -> Result<()> {
-    let mut server_config = fast_socks5::server::Config::default();
-    server_config.set_request_timeout(cfg.request_timeout);
-    server_config.set_skip_auth(cfg.skip_auth);
-    server_config.set_dns_resolve(cfg.dns_resolve);
-    server_config.set_execute_command(cfg.execute_command);
-    server_config.set_udp_support(cfg.allow_udp);
+    cfg.validate()?;
 
-    if let Some(PasswordAuth { username, password }) = cfg.auth {
-        server_config.set_authentication(SimpleUserPassword { username, password });
-        log::info!("Simple auth system has been set.");
-    } else {
-        log::warn!("No authentication has been set!");
-    }
+    let listener = TcpListener::bind(&cfg.listen_addr).await?;
 
-    let mut listener = Socks5Server::bind(&cfg.listen_addr).await?;
-    listener.set_config(server_config);
-
-    let incoming = stream_with_cancellation(listener.incoming(), &token);
+    let incoming = stream_with_cancellation(TcpListenerStream::new(listener), &token);
     tokio::pin!(incoming);
 
     log::info!("Listen for socks connections @ {}", &cfg.listen_addr);
@@ -46,10 +34,10 @@ pub async fn spawn_socks5_server(cfg: Config, token: CancellationToken) -> Resul
         match socket_res {
             Ok(socket) => {
                 let child_token = token.child_token();
-                spawn_and_log_error(socket.upgrade_to_socks5(), child_token);
+                spawn_and_log_error(serve_socks5(cfg.clone(), socket), child_token);
             }
             Err(err) => {
-                log::error!("accept error: {}", err);
+                log::error!("accept error: {err}");
             }
         }
     }
@@ -57,6 +45,39 @@ pub async fn spawn_socks5_server(cfg: Config, token: CancellationToken) -> Resul
     Ok(())
 }
 
+async fn serve_socks5(cfg: Config, socket: tokio::net::TcpStream) -> Result<(), SocksError> {
+    let (proto, cmd, target_addr) = match &cfg.auth {
+        None if cfg.skip_auth => Socks5ServerProtocol::skip_auth_this_is_not_rfc_compliant(socket),
+        None => Socks5ServerProtocol::accept_no_auth(socket).await?,
+        Some(PasswordAuth { username, password }) => {
+            Socks5ServerProtocol::accept_password_auth(socket, |user, pass| {
+                user == *username && pass == *password
+            })
+            .await?
+            .0
+        }
+    }
+    .read_command()
+    .await?
+    .resolve_dns()
+    .await?;
+
+    match cmd {
+        Socks5Command::TCPConnect => {
+            run_tcp_proxy(proto, &target_addr, cfg.request_timeout, false).await?;
+        }
+        Socks5Command::UDPAssociate if cfg.allow_udp => {
+            let reply_ip = cfg.public_addr.context("invalid reply ip")?;
+            run_udp_proxy(proto, &target_addr, None, reply_ip, None).await?;
+        }
+        _ => {
+            proto.reply_error(&ReplyError::CommandNotSupported).await?;
+            return Err(ReplyError::CommandNotSupported.into());
+        }
+    };
+    Ok(())
+}
+
 fn stream_with_cancellation<'a, S>(
     mut inner: S,
     token: &'a CancellationToken,
@@ -89,10 +110,9 @@ where
     }
 }
 
-fn spawn_and_log_error<F, T>(future: F, token: CancellationToken) -> task::JoinHandle<()>
+fn spawn_and_log_error<F>(future: F, token: CancellationToken) -> task::JoinHandle<()>
 where
-    F: Future<Output = fast_socks5::Result<Socks5Socket<T>>> + Send + 'static,
-    T: AsyncRead + AsyncWrite + Unpin,
+    F: Future<Output = Result<(), SocksError>> + Send + 'static,
 {
     tokio::spawn(async move {
         let result = select! {

src/service.rs

@@ -1,6 +1,6 @@
 use tokio_util::sync::CancellationToken;
 
-use anyhow::{anyhow, Result};
+use anyhow::{Result, anyhow};
 use std::{ffi::OsString, thread, time::Duration};
 use windows_service::{
     define_windows_service,
@@ -150,7 +150,7 @@ define_windows_service!(ffi_service_main, my_service_main);
 // output to file if needed.
 pub fn my_service_main(_arguments: Vec<OsString>) {
     if let Err(e) = run_service() {
-        log::error!("error: {}", e);
+        log::error!("error: {e}");
     }
 }
 
@@ -185,7 +185,7 @@ pub fn run_service() -> Result<()> {
     status_handle.set_service_status(ServiceStatus::running())?;
 
     let cfg = Config::get();
-    log::info!("start with config: {:#?}", cfg);
+    log::info!("start with config: {cfg:#?}");
 
     let result = std::thread::spawn(move || server_executor(cfg, server_token)).join();
 
@@ -193,14 +193,14 @@ pub fn run_service() -> Result<()> {
 
     // join() => Err(), when thread panic
     if let Err(e) = result {
-        log::error!("server panic: {:#?}", e);
+        log::error!("server panic: {e:#?}");
         status_handle.set_service_status(ServiceStatus::stopped_with_error(1))?;
         return Err(anyhow!("server panic"));
     }
 
     // join() => Ok(Err()), when server executor error
     if let Err(e) = result.unwrap() {
-        log::error!("server error: {:#?}", e);
+        log::error!("server error: {e:#?}");
         status_handle.set_service_status(ServiceStatus::stopped_with_error(2))?;
         return Err(anyhow!("server error"));
     }