Perfuse TorBackend through monero-rpc-pool

Author: nabijaczleweli

monero-rpc-pool/Cargo.toml

@@ -38,6 +38,7 @@ tracing-subscriber = { workspace = true }
 # Async runtime
 crossbeam = "0.8.4"
 tokio = { workspace = true, features = ["full"] }
+tokio-socks = "0.5"
 
 # Serialization
 chrono = { version = "0.4", features = ["serde"] }
@@ -69,6 +70,7 @@ tor-rtcompat = { workspace = true, features = ["tokio", "rustls"] }
 monero = { workspace = true }
 monero-rpc = { path = "../monero-rpc" }
 swap-serde = { path = "../swap-serde" }
+swap-tor = { path = "../swap-tor" }
 
 # Optional dependencies (for features)
 cuprate-epee-encoding = { git = "https://github.com/Cuprate/cuprate.git", optional = true }

monero-rpc-pool/src/bin/stress_test.rs

@@ -77,9 +77,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             .await
             .expect("Failed to bootstrap Tor client");
 
-        Some(client)
+        swap_tor::TorBackend::Arti(client)
     } else {
-        None
+        swap_tor::TorBackend::None
     };
 
     // Start the pool server

monero-rpc-pool/src/bin/stress_test_downloader.rs

@@ -133,9 +133,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             .await
             .expect("Failed to bootstrap Tor client");
 
-        Some(client)
+        swap_tor::TorBackend::Arti(client)
     } else {
-        None
+        swap_tor::TorBackend::None
     };
 
     // Start the pool server

monero-rpc-pool/src/config.rs

@@ -1,57 +1,44 @@
 use monero::Network;
 use std::path::PathBuf;
+use swap_tor::TorBackend;
 
-use crate::TorClientArc;
-
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct Config {
     pub host: String,
     pub port: u16,
     pub data_dir: PathBuf,
-    pub tor_client: Option<TorClientArc>,
+    pub tor_client: TorBackend,
     pub network: Network,
 }
 
-impl std::fmt::Debug for Config {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("Config")
-            .field("host", &self.host)
-            .field("port", &self.port)
-            .field("data_dir", &self.data_dir)
-            .field("tor_client", &self.tor_client.is_some())
-            .field("network", &self.network)
-            .finish()
-    }
-}
-
 impl Config {
     pub fn new_with_port(host: String, port: u16, data_dir: PathBuf, network: Network) -> Self {
-        Self::new_with_port_and_tor_client(host, port, data_dir, None, network)
+        Self::new_with_port_and_tor_client(host, port, data_dir, TorBackend::None, network)
     }
 
     pub fn new_with_port_and_tor_client(
         host: String,
         port: u16,
         data_dir: PathBuf,
-        tor_client: impl Into<Option<TorClientArc>>,
+        tor_client: TorBackend,
         network: Network,
     ) -> Self {
         Self {
             host,
             port,
             data_dir,
-            tor_client: tor_client.into(),
+            tor_client,
             network,
         }
     }
 
     pub fn new_random_port(data_dir: PathBuf, network: Network) -> Self {
-        Self::new_random_port_with_tor_client(data_dir, None, network)
+        Self::new_random_port_with_tor_client(data_dir, TorBackend::None, network)
     }
 
     pub fn new_random_port_with_tor_client(
         data_dir: PathBuf,
-        tor_client: impl Into<Option<TorClientArc>>,
+        tor_client: TorBackend,
         network: Network,
     ) -> Self {
         Self::new_with_port_and_tor_client(

monero-rpc-pool/src/lib.rs

@@ -1,25 +1,21 @@
 use std::sync::Arc;
 
 use anyhow::Result;
-use arti_client::TorClient;
 use axum::{
     routing::{any, get},
     Router,
 };
 
 use tokio::task::JoinHandle;
-use tor_rtcompat::tokio::TokioRustlsRuntime;
 use tower_http::cors::CorsLayer;
 use tracing::{error, info};
 
-/// Type alias for the Tor client used throughout the crate
-pub type TorClientArc = Arc<TorClient<TokioRustlsRuntime>>;
-
 pub mod config;
 pub mod connection_pool;
 pub mod database;
 pub mod pool;
 pub mod proxy;
+pub(crate) mod tor;
 pub mod types;
 
 use config::Config;
@@ -30,7 +26,7 @@ use proxy::{proxy_handler, stats_handler};
 #[derive(Clone)]
 pub struct AppState {
     pub node_pool: Arc<NodePool>,
-    pub tor_client: Option<TorClientArc>,
+    pub tor_client: swap_tor::TorBackend,
     pub connection_pool: crate::connection_pool::ConnectionPool,
 }
 

monero-rpc-pool/src/main.rs

@@ -74,9 +74,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             }
         });
 
-        Some(client)
+        swap_tor::TorBackend::Arti(client)
     } else {
-        None
+        swap_tor::TorBackend::None
     };
 
     let config = Config::new_with_port_and_tor_client(

monero-rpc-pool/src/proxy.rs

@@ -10,10 +10,7 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::net::TcpStream;
-use tokio::{
-    io::{AsyncRead, AsyncWrite},
-    time::timeout,
-};
+use tokio::time::timeout;
 
 use tokio_rustls::rustls::{
     client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier},
@@ -22,6 +19,7 @@ use tokio_rustls::rustls::{
 };
 use tracing::{error, info_span, Instrument};
 
+use crate::tor::*;
 use crate::AppState;
 
 /// wallet2.h has a default timeout of 3 minutes + 30 seconds.
@@ -32,10 +30,6 @@ static TIMEOUT: Duration = Duration::from_secs(3 * 60 + 30).checked_div(2).unwra
 /// If the main node does not finish within this period, we start a hedged request.
 static SOFT_TIMEOUT: Duration = TIMEOUT.checked_div(2).unwrap();
 
-/// Trait alias for a stream that can be used with hyper
-trait HyperStream: AsyncRead + AsyncWrite + Unpin + Send {}
-impl<T: AsyncRead + AsyncWrite + Unpin + Send> HyperStream for T {}
-
 #[derive(Debug)]
 struct NoCertificateVerification;
 
@@ -149,14 +143,7 @@ async fn proxy_to_multiple_nodes(
 
     // Sort nodes to prioritize those with available connections
     // Check if we're using Tor for this request
-    let use_tor = match &state.tor_client {
-        Some(tc)
-            if tc.bootstrap_status().ready_for_traffic() && !request.clearnet_whitelisted() =>
-        {
-            true
-        }
-        _ => false,
-    };
+    let use_tor = state.tor_client.ready_for_traffic() && !request.clearnet_whitelisted();
 
     // Create a vector of (node, has_connection) pairs
     let mut nodes_with_availability = Vec::new();
@@ -321,7 +308,7 @@ async fn proxy_to_multiple_nodes(
 
 /// Wraps a stream with TLS if HTTPS is being used
 async fn maybe_wrap_with_tls(
-    stream: impl AsyncRead + AsyncWrite + Unpin + Send + 'static,
+    stream: impl HyperStream + 'static,
     scheme: &str,
     host: &str,
 ) -> Result<Box<dyn HyperStream>, SingleRequestError> {
@@ -465,14 +452,7 @@ async fn proxy_to_single_node(
         tracing::trace!("Request is whitelisted, sending over clearnet");
     }
 
-    let use_tor = match &state.tor_client {
-        Some(tc)
-            if tc.bootstrap_status().ready_for_traffic() && !request.clearnet_whitelisted() =>
-        {
-            true
-        }
-        _ => false,
-    };
+    let use_tor = state.tor_client.ready_for_traffic() && !request.clearnet_whitelisted();
 
     let key = (node.0.clone(), node.1.clone(), node.2, use_tor);
 
@@ -484,24 +464,14 @@ async fn proxy_to_single_node(
         let address = (node.1.as_str(), node.2);
 
         let maybe_tls_stream = timeout(TIMEOUT, async {
-            let no_tls_stream: Box<dyn HyperStream> = if use_tor {
-                let tor_client = state.tor_client.as_ref().ok_or_else(|| {
-                    SingleRequestError::ConnectionError("Tor requested but client missing".into())
-                })?;
-
-                let stream = tor_client
-                    .connect(address)
-                    .await
-                    .map_err(|e| SingleRequestError::ConnectionError(format!("{:?}", e)))?;
-
-                Box::new(stream)
-            } else {
-                let stream = TcpStream::connect(address)
+            let no_tls_stream = match use_tor {
+                true => state.tor_client.connect(address).await,
+                false => TcpStream::connect(address)
                     .await
-                    .map_err(|e| SingleRequestError::ConnectionError(format!("{:?}", e)))?;
-
-                Box::new(stream)
-            };
+                    .map(|s| Box::new(s) as Box<dyn HyperStream>)
+                    .map_err(|e| e.into()),
+            }
+            .map_err(|e| SingleRequestError::ConnectionError(format!("{:?}", e)))?;
 
             maybe_wrap_with_tls(no_tls_stream, &node.0, &node.1).await
         })

monero-rpc-pool/src/tor.rs

@@ -0,0 +1,80 @@
+use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
+use swap_tor::TorBackend;
+use tokio::io::{AsyncRead, AsyncWrite};
+use tokio_socks::TargetAddr;
+
+/// Trait alias for a stream that can be used with hyper
+pub trait HyperStream: AsyncRead + AsyncWrite + Unpin + Send {}
+impl<T: AsyncRead + AsyncWrite + Unpin + Send> HyperStream for T {}
+
+pub trait TorBackendRpc {
+    fn is_some(&self) -> bool;
+    fn ready_for_traffic(&self) -> bool;
+    async fn connect(&self, address: (&str, u16)) -> anyhow::Result<Box<dyn HyperStream>>;
+}
+impl TorBackendRpc for TorBackend {
+    fn is_some(&self) -> bool {
+        !matches!(self, TorBackend::None)
+    }
+
+    fn ready_for_traffic(&self) -> bool {
+        match self {
+            TorBackend::Arti(arti) => arti.bootstrap_status().ready_for_traffic(),
+            TorBackend::Socks(..) => true,
+            TorBackend::None => false,
+        }
+    }
+
+    async fn connect(&self, address: (&str, u16)) -> anyhow::Result<Box<dyn HyperStream>> {
+        match self {
+            TorBackend::Arti(tor_client) => Ok(Box::new(tor_client.connect(address).await?)),
+            TorBackend::Socks(proxy) => Ok(Box::new(proxy.proxy(pair_to_socks(address)).await?)),
+            TorBackend::None => Ok(Box::new(tokio::net::TcpStream::connect(address).await?)),
+        }
+    }
+}
+
+// Parse order matches tokio::net::ToSocketAddrs
+fn pair_to_socks((host, port): (&str, u16)) -> TargetAddr {
+    if let Ok(addr) = host.parse::<Ipv4Addr>() {
+        TargetAddr::Ip(SocketAddr::new(addr.into(), 10))
+    } else if let Ok(addr) = host.parse::<Ipv6Addr>() {
+        TargetAddr::Ip(SocketAddr::new(addr.into(), 10))
+    } else {
+        TargetAddr::Domain(host.into(), port)
+    }
+}
+#[cfg(test)]
+mod tests {
+    use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
+    use tokio_socks::TargetAddr;
+
+    #[test]
+    fn pair_to_socks() {
+        assert_eq!(
+            [
+                ("ip.tld", 10),
+                ("dns.ip4.tld", 11),
+                ("dns.ip6.tld", 12),
+                (
+                    "cebulka7uxchnbpvmqapg5pfos4ngaxglsktzvha7a5rigndghvadeyd.onion",
+                    13
+                ),
+                ("127.0.0.1", 10),
+                ("::1", 10),
+            ]
+            .map(super::pair_to_socks),
+            [
+                TargetAddr::Domain("ip.tld".into(), 10),
+                TargetAddr::Domain("dns.ip4.tld".into(), 11),
+                TargetAddr::Domain("dns.ip6.tld".into(), 12),
+                TargetAddr::Domain(
+                    "cebulka7uxchnbpvmqapg5pfos4ngaxglsktzvha7a5rigndghvadeyd.onion".into(),
+                    13,
+                ),
+                TargetAddr::Ip(SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 10)),
+                TargetAddr::Ip(SocketAddr::new(Ipv6Addr::LOCALHOST.into(), 10)),
+            ],
+        );
+    }
+}

swap-tor/Cargo.toml

@@ -9,7 +9,7 @@ description = "Arti/SOCKS5 Tor back-end."
 anyhow = { workspace = true }
 arti-client = { workspace = true, features = ["static-sqlite", "tokio", "rustls", "onion-service-service"] }
 data-encoding = "2.6"
-libp2p = { workspace = true, features = ["tcp", "yamux", "dns", "noise", "request-response", "ping", "rendezvous", "identify", "macros", "cbor", "json", "tokio", "serde", "rsa"] }
+libp2p = { workspace = true }
 
 # Tokio
 tokio = { workspace = true, features = ["net"] }

swap-tor/src/lib.rs

@@ -159,9 +159,7 @@ impl Transport for Socks5Transport {
 
         Ok(Box::pin(async move {
             Ok(tokio_util::compat::TokioAsyncReadCompatExt::compat(
-                Socks5Stream::connect_with_socket(proxy.connect().await?, target)
-                    .await?
-                    .into_inner(),
+                proxy.proxy(target).await?,
             ))
         }))
     }
@@ -208,6 +206,15 @@ impl SocksServerAddress {
         }
     }
 
+    pub async fn proxy(
+        &self,
+        target: TargetAddr<'_>,
+    ) -> Result<TcpOrUnixStream, tokio_socks::Error> {
+        Socks5Stream::connect_with_socket(self.connect().await?, target)
+            .await
+            .map(Socks5Stream::into_inner)
+    }
+
     /// Consult `$TOR_SOCKS_{IPC_PATH,HOST+PORT}`
     ///
     /// `$TOR_SOCKS_IPC_PATH` is ignored if `cfg(not(unix))`, and takes precedence if `cfg(unix)`.
@@ -234,3 +241,13 @@ pub enum TorBackend {
     Socks(Arc<SocksServerAddress>),
     None,
 }
+
+impl std::fmt::Debug for TorBackend {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> Result<(), std::fmt::Error> {
+        f.write_str(match self {
+            TorBackend::Arti(..) => "Arti",
+            TorBackend::Socks(..) => "Socks",
+            TorBackend::None => "None",
+        })
+    }
+}