Add proxy and onion support for tentacle

Signed-off-by: Eval EXEC <execvy@gmail.com>

Author: eval-exec

multiaddr/src/onion_addr.rs

@@ -1,5 +1,7 @@
 use std::{borrow::Cow, fmt};
 
+use data_encoding::BASE32;
+
 /// Represents an Onion v3 address
 #[derive(Clone)]
 pub struct Onion3Addr<'a>(Cow<'a, [u8; 35]>, u16);
@@ -19,6 +21,11 @@ impl Onion3Addr<'_> {
     pub fn acquire<'b>(self) -> Onion3Addr<'b> {
         Onion3Addr(Cow::Owned(self.0.into_owned()), self.1)
     }
+
+    pub fn hash_string(&self) -> String {
+        let s = BASE32.encode(self.hash());
+        s.to_lowercase()
+    }
 }
 
 impl PartialEq for Onion3Addr<'_> {

multiaddr/src/protocol.rs

@@ -273,8 +273,7 @@ impl fmt::Display for Protocol<'_> {
             Wss => write!(f, "/wss"),
             Memory(port) => write!(f, "/memory/{}", port),
             Onion3(addr) => {
-                let s = BASE32.encode(addr.hash());
-                write!(f, "/onion3/{}:{}", s.to_lowercase(), addr.port())
+                write!(f, "/onion3/{}:{}", addr.hash_string(), addr.port())
             }
         }
     }

tentacle/Cargo.toml

@@ -36,6 +36,7 @@ httparse = { version = "1.9", optional = true }
 futures-timer = { version = "3.0.2", optional = true }
 
 multiaddr = { path = "../multiaddr", package = "tentacle-multiaddr", version = "0.3.4" }
+url = "2.5.4"
 molecule = "0.8.0"
 
 # upnp
@@ -48,6 +49,7 @@ tokio-rustls = { version = "0.26.0", optional = true }
 # rand 0.8 not support wasm32
 rand = "0.8"
 socket2 = { version = "0.5.0", features = ["all"] }
+fast-socks5 = "0.10.0"
 
 [target.'cfg(target_family = "wasm")'.dependencies]
 js-sys = "0.3"

tentacle/src/builder.rs

@@ -1,5 +1,6 @@
 use std::{io, sync::Arc, time::Duration};
 
+use crate::service::config::TransformerContext;
 use nohash_hasher::IntMap;
 use tokio_util::codec::LengthDelimitedCodec;
 
@@ -217,9 +218,49 @@ where
     #[cfg(not(target_family = "wasm"))]
     pub fn tcp_config<F>(mut self, f: F) -> Self
     where
-        F: Fn(TcpSocket) -> Result<TcpSocket, std::io::Error> + Send + Sync + 'static,
+        F: Fn(TcpSocket, TransformerContext) -> Result<TcpSocket, std::io::Error>
+            + Send
+            + Sync
+            + 'static,
     {
-        self.config.tcp_config.tcp = Arc::new(f);
+        self.config.tcp_config.tcp.socket_transformer = Arc::new(f);
+        self
+    }
+
+    fn must_parse_proxy_url(url: String) -> url::Url {
+        let url = url::Url::parse(&url)
+            .unwrap_or_else(|err| panic!("parse {} to url::Url failed: {}", url, err));
+        if url.scheme().ne("socks5") {
+            panic!("tentacle only support socks5 proxy");
+        }
+        url
+    }
+
+    /// Proxy config for tcp
+    ///
+    /// Example: socks5://127.0.0.1:9050
+    /// Example: socks5://username:password@127.0.0.1:9050
+    #[cfg(not(target_family = "wasm"))]
+    pub fn tcp_proxy_config(mut self, proxy_url: String) -> Self {
+        let proxy_url: url::Url = Self::must_parse_proxy_url(proxy_url);
+        self.config.tcp_config.tcp.proxy_url = Some(proxy_url);
+        self
+    }
+
+    /// Onion config for tcp
+    ///
+    /// Example: socks5://127.0.0.1:9050
+    #[cfg(not(target_family = "wasm"))]
+    pub fn tcp_onion_config(mut self, onion_url: String) -> Self {
+        let onion_url: url::Url = Self::must_parse_proxy_url(onion_url);
+        self.config.tcp_config.tcp.onion_url = Some(onion_url);
+        self
+    }
+
+    /// Onion config for tcp
+    #[cfg(not(target_family = "wasm"))]
+    pub fn tcp_onion_random_socks_auth(mut self, onion_random_socks_auth: bool) -> Self {
+        self.config.tcp_config.tcp.onion_random_socks_auth = onion_random_socks_auth;
         self
     }
 
@@ -228,9 +269,12 @@ where
     #[cfg_attr(docsrs, doc(cfg(feature = "ws")))]
     pub fn tcp_config_on_ws<F>(mut self, f: F) -> Self
     where
-        F: Fn(TcpSocket) -> Result<TcpSocket, std::io::Error> + Send + Sync + 'static,
+        F: Fn(TcpSocket, TransformerContext) -> Result<TcpSocket, std::io::Error>
+            + Send
+            + Sync
+            + 'static,
     {
-        self.config.tcp_config.ws = Arc::new(f);
+        self.config.tcp_config.ws.socket_transformer = Arc::new(f);
         self
     }
 
@@ -252,9 +296,12 @@ where
     #[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
     pub fn tcp_config_on_tls<F>(mut self, f: F) -> Self
     where
-        F: Fn(TcpSocket) -> Result<TcpSocket, std::io::Error> + Send + Sync + 'static,
+        F: Fn(TcpSocket, TransformerContext) -> Result<TcpSocket, std::io::Error>
+            + Send
+            + Sync
+            + 'static,
     {
-        self.config.tcp_config.tls = Arc::new(f);
+        self.config.tcp_config.tls.socket_transformer = Arc::new(f);
         self
     }
 }

tentacle/src/error.rs

@@ -9,6 +9,9 @@ pub enum TransportErrorKind {
     /// IO error
     #[error("transport io error: `{0:?}`")]
     Io(#[from] IOError),
+    /// Proxy server error
+    #[error("proxy error: `{0:?}`")]
+    ProxyError(IOError),
     /// Protocol not support
     #[error("multiaddr `{0:?}` is not supported")]
     NotSupported(Multiaddr),

tentacle/src/runtime/mod.rs

@@ -14,6 +14,7 @@
     all(target_family = "wasm", feature = "wasm-timer")
 ))]
 mod generic_timer;
+pub(crate) mod proxy;
 #[cfg(all(not(target_family = "wasm"), feature = "tokio-runtime"))]
 mod tokio_runtime;
 #[cfg(target_family = "wasm")]

tentacle/src/runtime/proxy/mod.rs

@@ -0,0 +1,4 @@
+#[cfg(not(target_family = "wasm"))]
+pub(crate) mod socks5;
+#[cfg(not(target_family = "wasm"))]
+pub(crate) mod socks5_config;

tentacle/src/runtime/proxy/socks5.rs

@@ -0,0 +1,41 @@
+use fast_socks5::{
+    AuthenticationMethod, Socks5Command,
+    client::{Config as ConnectConfig, Socks5Stream},
+};
+use tokio::net::TcpStream;
+
+pub async fn connect(
+    socks_server: url::Url,
+    target_addr: String,
+    target_port: u16,
+) -> Result<TcpStream, fast_socks5::SocksError> {
+    let auth = {
+        if socks_server.username().is_empty() {
+            AuthenticationMethod::None
+        } else {
+            AuthenticationMethod::Password {
+                username: socks_server.username().to_string(),
+                password: socks_server.password().unwrap_or_default().to_string(),
+            }
+        }
+    };
+    let socks_server_str = format!(
+        "{}:{}",
+        socks_server.host_str().ok_or_else(|| {
+            fast_socks5::SocksError::ArgumentInputError("socks_server should have host")
+        })?,
+        socks_server.port().ok_or_else(|| {
+            fast_socks5::SocksError::ArgumentInputError("socks_server should have port")
+        })?
+    );
+    Socks5Stream::connect_raw(
+        Socks5Command::TCPConnect,
+        socks_server_str,
+        target_addr,
+        target_port,
+        Some(auth),
+        ConnectConfig::default(),
+    )
+    .await
+    .map(|socket| socket.get_socket())
+}

tentacle/src/runtime/proxy/socks5_config.rs

@@ -0,0 +1,15 @@
+use rand::Rng;
+
+pub(crate) fn random_auth() -> (String, String) {
+    let username = rand::thread_rng()
+        .sample_iter(&rand::distributions::Alphanumeric)
+        .take(8)
+        .map(char::from)
+        .collect();
+    let password = rand::thread_rng()
+        .sample_iter(&rand::distributions::Alphanumeric)
+        .take(16)
+        .map(char::from)
+        .collect();
+    (username, password)
+}

tentacle/src/runtime/tokio_runtime.rs tentacle/src/runtime/tokio_runtime/mod.rstentacle/src/runtime/tokio_runtime.rs renamed to tentacle/src/runtime/tokio_runtime/mod.rs

@@ -1,10 +1,14 @@
+use super::proxy::{socks5, socks5_config::random_auth};
+use multiaddr::{MultiAddr, Protocol};
 pub use tokio::{
     net::{TcpListener, TcpStream},
     spawn,
     task::{JoinHandle, block_in_place, spawn_blocking, yield_now},
 };
 
-use crate::service::config::{TcpSocket, TcpSocketConfig};
+use crate::service::config::{
+    TcpSocket, TcpSocketConfig, TcpSocketTransformer, TransformerContext,
+};
 use socket2::{Domain, Protocol as SocketProtocol, Socket, Type};
 #[cfg(unix)]
 use std::os::unix::io::{FromRawFd, IntoRawFd};
@@ -88,7 +92,8 @@ pub(crate) fn listen(addr: SocketAddr, tcp_config: TcpSocketConfig) -> io::Resul
         // user can disable it on tcp_config
         #[cfg(not(windows))]
         socket.set_reuse_address(true)?;
-        let t = tcp_config(TcpSocket { inner: socket })?;
+        let transformer_context = TransformerContext::new_listen(addr);
+        let t = (tcp_config.socket_transformer)(TcpSocket { inner: socket }, transformer_context)?;
         t.inner.set_nonblocking(true)?;
         // safety: fd convert by socket2
         unsafe {
@@ -113,15 +118,16 @@ pub(crate) fn listen(addr: SocketAddr, tcp_config: TcpSocketConfig) -> io::Resul
     socket.listen(1024)
 }
 
-pub(crate) async fn connect(
+async fn connect_direct(
     addr: SocketAddr,
-    tcp_config: TcpSocketConfig,
+    socket_transformer: TcpSocketTransformer,
 ) -> io::Result<TcpStream> {
     let domain = Domain::for_address(addr);
     let socket = Socket::new(domain, Type::STREAM, Some(SocketProtocol::TCP))?;
 
     let socket = {
-        let t = tcp_config(TcpSocket { inner: socket })?;
+        let transformer_context = TransformerContext::new_dial(addr);
+        let t = socket_transformer(TcpSocket { inner: socket }, transformer_context)?;
         t.inner.set_nonblocking(true)?;
         // safety: fd convert by socket2
         unsafe {
@@ -135,3 +141,93 @@ pub(crate) async fn connect(
 
     socket.connect(addr).await
 }
+
+async fn connect_by_proxy(
+    target_addr: String,
+    target_port: u16,
+    proxy_url: url::Url,
+) -> io::Result<TcpStream> {
+    socks5::connect(proxy_url.clone(), target_addr.clone(), target_port)
+        .await
+        .map_err(|err| {
+            io::Error::new(
+                io::ErrorKind::Other,
+                format!(
+                    "socks5_connect to target_addr: {}, target_port: {} by proxy_server: {} failed, err: {}",
+                    target_addr, target_port, proxy_url, err
+                ),
+            )
+        })
+}
+
+pub(crate) async fn connect(
+    target_addr: SocketAddr,
+    tcp_config: TcpSocketConfig,
+) -> io::Result<TcpStream> {
+    let TcpSocketConfig {
+        socket_transformer,
+        proxy_url,
+        onion_url: _,
+        onion_random_socks_auth: _,
+    } = tcp_config;
+
+    match proxy_url {
+        Some(proxy_url) => connect_by_proxy(
+            target_addr.ip().to_string(),
+            target_addr.port(),
+            proxy_url.clone(),
+        )
+        .await
+        .map_err(|err| {
+            io::Error::new(
+                io::ErrorKind::Other,
+                format!("connect_by_proxy: {}, error: {}", proxy_url, err),
+            )
+        }),
+        None => connect_direct(target_addr, socket_transformer).await,
+    }
+}
+
+pub(crate) async fn connect_onion(
+    onion_addr: MultiAddr,
+    tcp_config: TcpSocketConfig,
+) -> io::Result<TcpStream> {
+    let TcpSocketConfig {
+        socket_transformer: _,
+        proxy_url,
+        onion_url,
+        onion_random_socks_auth,
+    } = tcp_config;
+    let mut tor_server_url = onion_url.or(proxy_url).ok_or(io::Error::other(
+        "need tor proxy server to connect to onion address",
+    ))?;
+
+    if onion_random_socks_auth {
+        let (random_username, random_passwd) = random_auth();
+        tor_server_url
+            .set_username(&random_username)
+            .map_err(|_| io::Error::other("failed to set username"))?;
+        tor_server_url
+            .set_password(Some(&random_passwd))
+            .map_err(|_| io::Error::other("failed to set password"))?;
+    }
+
+    let onion_protocol = onion_addr
+        .iter()
+        .find_map(|protocol| {
+            if let Protocol::Onion3(onion_address) = protocol {
+                Some(onion_address)
+            } else {
+                None
+            }
+        })
+        .ok_or(io::Error::other(format!(
+            "No Onion3 address found. in {}",
+            onion_addr
+        )))?;
+
+    let onion_str = onion_protocol.hash_string() + ".onion";
+    let onion_port = onion_protocol.port();
+
+    connect_by_proxy(onion_str, onion_port, tor_server_url).await
+}