Repository: mintlayer/mintlayer-core
Pull request #740 (2023-03-04 13:29:32)

Add socks5 support

pavel-kokolemin: > I'm very proud we made that adapter design, which made adding this proxy very easy.

Right, it worked well. Thank you for the clean design.

> What if the user chose multiple proxys? Do we want to allow that?

Good question. I just checked and the node fails to start if multiple proxies are specified on the command line:
```
error: the argument '--p2p-socks5-proxy <P2P_SOCKS5_PROXY>' cannot be used multiple times
```
Supporting multiple proxies or proxy chains should not be very difficult, but I don't think we should bother with that (Bitcoin Core does not support it).

> This lacks tests.
... Adding TODOs or creating an issue is good enough for now.

TODO added.
And I tested the node with socks5 server (`ssh -D`) and it worked as expected.

I also switched from `SocketAddr` to `String` so users can use host names (for example, `localhost:9050`)

Patch:
diff --git a/Cargo.lock b/Cargo.lock
index 2ff26418a..765aa3db2 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2214,6 +2214,7 @@ dependencies = [
  "test-utils",
  "thiserror",
  "tokio",
+ "tokio-socks",
  "tokio-util",
  "utils",
  "void",
@@ -3622,6 +3623,18 @@ dependencies = [
  "webpki",
 ]
 
+[[package]]
+name = "tokio-socks"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "51165dfa029d2a65969413a6cc96f354b86b464498702f174a4efa13608fd8c0"
+dependencies = [
+ "either",
+ "futures-util",
+ "thiserror",
+ "tokio",
+]
+
 [[package]]
 name = "tokio-stream"
 version = "0.1.12"
diff --git a/dns_server/src/main.rs b/dns_server/src/main.rs
index f115a99b0..508212807 100644
--- a/dns_server/src/main.rs
+++ b/dns_server/src/main.rs
@@ -41,6 +41,7 @@ async fn run(config: Arc<DnsServerConfig>) -> Result<void::Void, error::DnsServe
 
     let p2p_config = Arc::new(P2pConfig {
         bind_addresses: Vec::new(),
+        socks5_proxy: None,
         boot_nodes: Vec::new(),
         reserved_nodes: Vec::new(),
         max_inbound_connections: Default::default(),
diff --git a/node/src/config_files/mod.rs b/node/src/config_files/mod.rs
index 59d62ab8f..3fe0b82d4 100644
--- a/node/src/config_files/mod.rs
+++ b/node/src/config_files/mod.rs
@@ -126,6 +126,7 @@ fn chainstate_config(
 fn p2p_config(config: P2pConfigFile, options: &RunOptions) -> P2pConfigFile {
     let P2pConfigFile {
         bind_addresses,
+        socks5_proxy: proxy,
         boot_nodes,
         reserved_nodes,
         max_inbound_connections,
@@ -138,6 +139,7 @@ fn p2p_config(config: P2pConfigFile, options: &RunOptions) -> P2pConfigFile {
     } = config;
 
     let bind_addresses = options.p2p_addr.clone().or(bind_addresses);
+    let socks5_proxy = options.p2p_socks5_proxy.clone().or(proxy);
     let boot_nodes = options.p2p_boot_node.clone().or(boot_nodes);
     let reserved_nodes = options.p2p_reserved_node.clone().or(reserved_nodes);
     let max_inbound_connections = options.p2p_max_inbound_connections.or(max_inbound_connections);
@@ -150,6 +152,7 @@ fn p2p_config(config: P2pConfigFile, options: &RunOptions) -> P2pConfigFile {
 
     P2pConfigFile {
         bind_addresses,
+        socks5_proxy,
         boot_nodes,
         reserved_nodes,
         max_inbound_connections,
diff --git a/node/src/config_files/p2p.rs b/node/src/config_files/p2p.rs
index ec8b1cf72..c2fe1d4d6 100644
--- a/node/src/config_files/p2p.rs
+++ b/node/src/config_files/p2p.rs
@@ -54,6 +54,8 @@ impl FromStr for NodeTypeConfigFile {
 pub struct P2pConfigFile {
     /// Address to bind P2P to.
     pub bind_addresses: Option<Vec<String>>,
+    /// SOCKS5 proxy.
+    pub socks5_proxy: Option<String>,
     /// Optional list of boot node addresses to connect.
     pub boot_nodes: Option<Vec<String>>,
     /// Optional list of reserved node addresses to connect.
@@ -78,6 +80,7 @@ impl From<P2pConfigFile> for P2pConfig {
     fn from(c: P2pConfigFile) -> Self {
         P2pConfig {
             bind_addresses: c.bind_addresses.clone().unwrap_or_default(),
+            socks5_proxy: c.socks5_proxy.clone(),
             boot_nodes: c.boot_nodes.clone().unwrap_or_default(),
             reserved_nodes: c.reserved_nodes.clone().unwrap_or_default(),
             max_inbound_connections: c.max_inbound_connections.into(),
diff --git a/node/src/options.rs b/node/src/options.rs
index b99afce34..02f5740dd 100644
--- a/node/src/options.rs
+++ b/node/src/options.rs
@@ -76,6 +76,10 @@ pub struct RunOptions {
     #[clap(long, value_name = "ADDR")]
     pub p2p_addr: Option<Vec<String>>,
 
+    /// Connect through SOCKS5 proxy.
+    #[clap(long)]
+    pub p2p_socks5_proxy: Option<String>,
+
     /// Optional list of boot node addresses to connect.
     #[clap(long, value_name = "NODE")]
     pub p2p_boot_node: Option<Vec<String>>,
diff --git a/p2p/Cargo.toml b/p2p/Cargo.toml
index 0e0a5a660..8c9ae5b74 100644
--- a/p2p/Cargo.toml
+++ b/p2p/Cargo.toml
@@ -38,6 +38,7 @@ tokio-util = {version = "0.7", default-features = false, features = ["codec"] }
 snowstorm = "0.4.0"
 serde.workspace = true
 socket2 = { version = "0.5", features = ["all"] }
+tokio-socks = { version = "0.5.1" }
 
 [dev-dependencies]
 chainstate-test-framework = { path = "../chainstate/test-framework" }
diff --git a/p2p/src/config.rs b/p2p/src/config.rs
index 8680acc85..2a06324b1 100644
--- a/p2p/src/config.rs
+++ b/p2p/src/config.rs
@@ -66,6 +66,8 @@ impl From<NodeType> for BTreeSet<PubSubTopic> {
 pub struct P2pConfig {
     /// Address to bind P2P to.
     pub bind_addresses: Vec<String>,
+    /// SOCKS5 proxy.
+    pub socks5_proxy: Option<String>,
     /// Optional list of initial node addresses.
     /// Boot node addresses are added to PeerDb as regular discovered addresses.
     pub boot_nodes: Vec<String>,
diff --git a/p2p/src/error.rs b/p2p/src/error.rs
index be1100dff..5795efe2b 100644
--- a/p2p/src/error.rs
+++ b/p2p/src/error.rs
@@ -80,6 +80,8 @@ pub enum DialError {
     ConnectionRefusedOrTimedOut,
     #[error("I/O error: `{0:?}`")]
     IoError(std::io::ErrorKind),
+    #[error("Proxy error: {0}")]
+    ProxyError(String),
 }
 
 /// Conversion errors
@@ -119,6 +121,12 @@ pub enum P2pError {
     InvalidStorageState(String),
 }
 
+impl From<DialError> for P2pError {
+    fn from(e: DialError) -> P2pError {
+        P2pError::DialError(e)
+    }
+}
+
 impl From<std::io::Error> for P2pError {
     fn from(e: std::io::Error) -> P2pError {
         P2pError::DialError(DialError::IoError(e.kind()))
diff --git a/p2p/src/lib.rs b/p2p/src/lib.rs
index 72e95b0fe..21d2f5941 100644
--- a/p2p/src/lib.rs
+++ b/p2p/src/lib.rs
@@ -34,6 +34,7 @@ use std::{
 };
 
 use interface::p2p_interface::P2pInterface;
+use net::default_backend::transport::NoiseSocks5Transport;
 use peer_manager::peerdb::storage::PeerDbStorage;
 use tap::TapFallible;
 use tokio::sync::mpsc;
@@ -152,6 +153,7 @@ impl subsystem::Subsystem for Box<dyn P2pInterface> {}
 pub type P2pHandle = subsystem::Handle<Box<dyn P2pInterface>>;
 
 pub type P2pNetworkingService = DefaultNetworkingService<NoiseTcpTransport>;
+pub type P2pNetworkingServiceSocks5Proxy = DefaultNetworkingService<NoiseSocks5Transport>;
 
 pub fn make_p2p_transport() -> NoiseTcpTransport {
     let stream_adapter = NoiseEncryptionAdapter::gen_new();
@@ -159,9 +161,16 @@ pub fn make_p2p_transport() -> NoiseTcpTransport {
     NoiseTcpTransport::new(stream_adapter, base_transport)
 }
 
+pub fn make_p2p_transport_socks5_proxy(proxy: &str) -> NoiseSocks5Transport {
+    let stream_adapter = NoiseEncryptionAdapter::gen_new();
+    let base_transport = net::default_backend::transport::Socks5TransportSocket::new(proxy);
+    NoiseSocks5Transport::new(stream_adapter, base_transport)
+}
+
 fn get_p2p_bind_addresses<S: AsRef<str>>(
     bind_addresses: &[S],
     p2p_port: u16,
+    proxy_used: bool,
 ) -> Result<Vec<SocketAddr>> {
     if !bind_addresses.is_empty() {
         bind_addresses
@@ -177,12 +186,14 @@ fn get_p2p_bind_addresses<S: AsRef<str>>(
                     })
             })
             .collect::<Result<Vec<_>>>()
-    } else {
+    } else if !proxy_used {
         // Bind to default addresses if none are specified by the user
         Ok(vec![
             SocketAddr::new(Ipv4Addr::UNSPECIFIED.into(), p2p_port),
             SocketAddr::new(Ipv6Addr::UNSPECIFIED.into(), p2p_port),
         ])
+    } else {
+        Ok(Vec::new())
     }
 }
 
@@ -194,22 +205,43 @@ pub async fn make_p2p<S: PeerDbStorage + 'static>(
     time_getter: TimeGetter,
     peerdb_storage: S,
 ) -> Result<Box<dyn P2pInterface>> {
-    let transport = make_p2p_transport();
-
-    let bind_addresses =
-        get_p2p_bind_addresses(&p2p_config.bind_addresses, chain_config.p2p_port())?;
-
-    let p2p = P2p::<P2pNetworkingService>::new(
-        transport,
-        bind_addresses,
-        chain_config,
-        p2p_config,
-        chainstate_handle,
-        mempool_handle,
-        time_getter,
-        peerdb_storage,
-    )
-    .await?;
-
-    Ok(Box::new(p2p))
+    let bind_addresses = get_p2p_bind_addresses(
+        &p2p_config.bind_addresses,
+        chain_config.p2p_port(),
+        p2p_config.socks5_proxy.is_some(),
+    )?;
+
+    if let Some(socks5_proxy) = &p2p_config.socks5_proxy {
+        let transport = make_p2p_transport_socks5_proxy(socks5_proxy);
+
+        let p2p = P2p::<P2pNetworkingServiceSocks5Proxy>::new(
+            transport,
+            bind_addresses,
+            chain_config,
+            p2p_config,
+            chainstate_handle,
+            mempool_handle,
+            time_getter,
+            peerdb_storage,
+        )
+        .await?;
+
+        Ok(Box::new(p2p))
+    } else {
+        let transport = make_p2p_transport();
+
+        let p2p = P2p::<P2pNetworkingService>::new(
+            transport,
+            bind_addresses,
+            chain_config,
+            p2p_config,
+            chainstate_handle,
+            mempool_handle,
+            time_getter,
+            peerdb_storage,
+        )
+        .await?;
+
+        Ok(Box::new(p2p))
+    }
 }
diff --git a/p2p/src/net/default_backend/backend.rs b/p2p/src/net/default_backend/backend.rs
index 4288be484..a4df7d279 100644
--- a/p2p/src/net/default_backend/backend.rs
+++ b/p2p/src/net/default_backend/backend.rs
@@ -296,7 +296,12 @@ where
     ) -> crate::Result<()> {
         let (tx, rx) = mpsc::unbounded_channel();
 
-        let receiver_address = Some(address.as_peer_address());
+        // Sending the remote socket address makes no sense and can leak private information when using a proxy
+        let receiver_address = if self.p2p_config.socks5_proxy.is_some() {
+            None
+        } else {
+            Some(address.as_peer_address())
+        };
 
         self.pending.insert(
             remote_peer_id,
diff --git a/p2p/src/net/default_backend/transport/impls/mod.rs b/p2p/src/net/default_backend/transport/impls/mod.rs
index e426fc2d5..c5ba8691b 100644
--- a/p2p/src/net/default_backend/transport/impls/mod.rs
+++ b/p2p/src/net/default_backend/transport/impls/mod.rs
@@ -14,5 +14,6 @@
 // limitations under the License.
 
 pub mod channel;
+pub mod socks5;
 pub mod stream_adapter;
 pub mod tcp;
diff --git a/p2p/src/net/default_backend/transport/impls/socks5.rs b/p2p/src/net/default_backend/transport/impls/socks5.rs
new file mode 100644
index 000000000..4a7f804d7
--- /dev/null
+++ b/p2p/src/net/default_backend/transport/impls/socks5.rs
@@ -0,0 +1,102 @@
+// Copyright (c) 2023 RBB S.r.l
+// opensource@mintlayer.org
+// SPDX-License-Identifier: MIT
+// Licensed under the MIT License;
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://github.com/mintlayer/mintlayer-core/blob/master/LICENSE
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::{
+    net::{IpAddr, SocketAddr},
+    sync::Arc,
+};
+
+use async_trait::async_trait;
+use futures::future::BoxFuture;
+use tokio::net::TcpStream;
+use tokio_socks::tcp::Socks5Stream;
+
+use crate::{
+    error::{DialError, P2pError},
+    net::default_backend::transport::{PeerStream, TransportListener, TransportSocket},
+    Result,
+};
+
+// TODO: Add tests. A typical way to test this is to create a forwarding proxy with a socks interface
+// that can test that things sent through the proxy are delivered to the other end.
+
+#[derive(Debug)]
+pub struct Socks5TransportSocket {
+    proxy: Arc<String>,
+}
+
+impl Socks5TransportSocket {
+    pub fn new(proxy: &str) -> Self {
+        Self {
+            proxy: Arc::new(proxy.to_owned()),
+        }
+    }
+}
+
+#[async_trait]
+impl TransportSocket for Socks5TransportSocket {
+    type Address = SocketAddr;
+    type BannableAddress = IpAddr;
+    type Listener = Socks5TransportListener;
+    type Stream = Socks5TransportStream;
+
+    async fn bind(&self, addresses: Vec<Self::Address>) -> Result<Self::Listener> {
+        Socks5TransportListener::new(addresses)
+    }
+
+    fn connect(&self, address: Self::Address) -> BoxFuture<'static, crate::Result<Self::Stream>> {
+        let proxy = Arc::clone(&self.proxy);
+        Box::pin(async move {
+            let socket = TcpStream::connect(proxy.as_str()).await.map_err(|e| {
+                DialError::ProxyError(format!("Connection to the SOCKS5 proxy failed: {e}"))
+            })?;
+
+            let stream = Socks5Stream::connect_with_socket(socket, address)
+                .await
+                .map_err(|e| DialError::ProxyError(format!("Unexpected SOCKS5 error: {e}")))?;
+
+            Ok(stream)
+        })
+    }
+}
+
+pub struct Socks5TransportListener {}
+
+impl Socks5TransportListener {
+    fn new(addresses: Vec<SocketAddr>) -> Result<Self> {
+        utils::ensure!(
+            addresses.is_empty(),
+            P2pError::InvalidConfigurationValue(
+                "Listening with socks5 proxy not implemented".to_owned()
+            ),
+        );
+        Ok(Self {})
+    }
+}
+
+#[async_trait]
+impl TransportListener<Socks5TransportStream, SocketAddr> for Socks5TransportListener {
+    async fn accept(&mut self) -> Result<(Socks5TransportStream, SocketAddr)> {
+        std::future::pending().await
+    }
+
+    fn local_addresses(&self) -> Result<Vec<SocketAddr>> {
+        Ok(Vec::new())
+    }
+}
+
+pub type Socks5TransportStream = Socks5Stream<TcpStream>;
+
+impl PeerStream for Socks5TransportStream {}
diff --git a/p2p/src/net/default_backend/transport/mod.rs b/p2p/src/net/default_backend/transport/mod.rs
index cf7adc38a..39f3840ed 100644
--- a/p2p/src/net/default_backend/transport/mod.rs
+++ b/p2p/src/net/default_backend/transport/mod.rs
@@ -17,11 +17,12 @@ mod impls;
 mod message_codec;
 mod traits;
 
-use impls::{channel, stream_adapter, tcp};
+use impls::{channel, socks5, stream_adapter, tcp};
 
 pub use self::{
     channel::{ChannelListener, ChannelStream, MpscChannelTransport},
     message_codec::BufferedTranscoder,
+    socks5::Socks5TransportSocket,
     stream_adapter::{
         identity::IdentityStreamAdapter, noise::NoiseEncryptionAdapter,
         wrapped_transport::wrapped_socket::WrappedTransportSocket,
@@ -31,3 +32,5 @@ pub use self::{
 };
 
 pub type NoiseTcpTransport = WrappedTransportSocket<NoiseEncryptionAdapter, TcpTransportSocket>;
+pub type NoiseSocks5Transport =
+    WrappedTransportSocket<NoiseEncryptionAdapter, Socks5TransportSocket>;


Test patch:
diff --git a/node/tests/cli.rs b/node/tests/cli.rs
index 7e9fc9760..d0dc028b8 100644
--- a/node/tests/cli.rs
+++ b/node/tests/cli.rs
@@ -98,6 +98,7 @@ fn read_config_override_values() {
     let max_db_commit_attempts = 1;
     let max_orphan_blocks = 2;
     let p2p_addr = "address";
+    let p2p_socks5_proxy = "socks5_proxy";
     let p2p_boot_node = "boot_node";
     let p2p_reserved_node = "reserved_node";
     let p2p_max_inbound_connections = 123;
@@ -118,6 +119,7 @@ fn read_config_override_values() {
         max_orphan_blocks: Some(max_orphan_blocks),
         tx_index_enabled: Some(false),
         p2p_addr: Some(vec![p2p_addr.to_owned()]),
+        p2p_socks5_proxy: Some(p2p_socks5_proxy.to_owned()),
         p2p_boot_node: Some(vec![p2p_boot_node.to_owned()]),
         p2p_reserved_node: Some(vec![p2p_reserved_node.to_owned()]),
         p2p_max_inbound_connections: Some(p2p_max_inbound_connections),
@@ -154,6 +156,10 @@ fn read_config_override_values() {
         config.p2p.clone().unwrap().bind_addresses,
         Some(vec!(p2p_addr.to_owned()))
     );
+    assert_eq!(
+        config.p2p.clone().unwrap().socks5_proxy,
+        Some(p2p_socks5_proxy.to_owned())
+    );
     assert_eq!(
         config.p2p.clone().unwrap().boot_nodes,
         Some(vec!(p2p_boot_node.to_owned()))
diff --git a/p2p/backend-test-suite/src/block_announcement.rs b/p2p/backend-test-suite/src/block_announcement.rs
index 8a67e2ba3..f1f5f2ab8 100644
--- a/p2p/backend-test-suite/src/block_announcement.rs
+++ b/p2p/backend-test-suite/src/block_announcement.rs
@@ -133,6 +133,7 @@ where
     let chain_config = Arc::new(common::chain::config::create_mainnet());
     let p2p_config = Arc::new(P2pConfig {
         bind_addresses: Vec::new(),
+        socks5_proxy: None,
         boot_nodes: Vec::new(),
         reserved_nodes: Vec::new(),
         max_inbound_connections: Default::default(),
diff --git a/p2p/src/peer_manager/peerdb/tests.rs b/p2p/src/peer_manager/peerdb/tests.rs
index dc5668e9c..3467650c0 100644
--- a/p2p/src/peer_manager/peerdb/tests.rs
+++ b/p2p/src/peer_manager/peerdb/tests.rs
@@ -34,6 +34,7 @@ fn unban_peer() {
     let mut peerdb = PeerDb::new(
         Arc::new(P2pConfig {
             bind_addresses: Default::default(),
+            socks5_proxy: None,
             boot_nodes: Default::default(),
             reserved_nodes: Default::default(),
             max_inbound_connections: Default::default(),
diff --git a/p2p/src/peer_manager/tests/connections.rs b/p2p/src/peer_manager/tests/connections.rs
index 25c3d8f36..52eb7a868 100644
--- a/p2p/src/peer_manager/tests/connections.rs
+++ b/p2p/src/peer_manager/tests/connections.rs
@@ -673,6 +673,7 @@ where
     // Start first peer manager
     let p2p_config_1 = Arc::new(P2pConfig {
         bind_addresses: Default::default(),
+        socks5_proxy: None,
         boot_nodes: Default::default(),
         reserved_nodes: Default::default(),
         max_inbound_connections: Default::default(),
@@ -705,6 +706,7 @@ where
     // Start second peer manager and let it know about first manager via reserved
     let p2p_config_2 = Arc::new(P2pConfig {
         bind_addresses: Default::default(),
+        socks5_proxy: None,
         boot_nodes: Default::default(),
         reserved_nodes: bind_addresses,
         max_inbound_connections: Default::default(),
@@ -779,6 +781,7 @@ where
     // Start the first peer manager
     let p2p_config_1 = Arc::new(P2pConfig {
         bind_addresses: Default::default(),
+        socks5_proxy: None,
         boot_nodes: Default::default(),
         reserved_nodes: Default::default(),
         max_inbound_connections: Default::default(),
@@ -812,6 +815,7 @@ where
     // Start the second peer manager and let it know about the first peer using reserved
     let p2p_config_2 = Arc::new(P2pConfig {
         bind_addresses: Default::default(),
+        socks5_proxy: None,
         boot_nodes: Default::default(),
         reserved_nodes: bind_addresses.clone(),
         max_inbound_connections: Default::default(),
@@ -838,6 +842,7 @@ where
     // Start the third peer manager and let it know about the first peer using reserved
     let p2p_config_3 = Arc::new(P2pConfig {
         bind_addresses: Default::default(),
+        socks5_proxy: None,
         boot_nodes: Default::default(),
         reserved_nodes: bind_addresses,
         max_inbound_connections: Default::default(),
diff --git a/p2p/src/peer_manager/tests/ping.rs b/p2p/src/peer_manager/tests/ping.rs
index a5f6d053d..635af6c2a 100644
--- a/p2p/src/peer_manager/tests/ping.rs
+++ b/p2p/src/peer_manager/tests/ping.rs
@@ -41,6 +41,7 @@ async fn ping_timeout() {
     let chain_config = Arc::new(config::create_mainnet());
     let p2p_config: Arc<P2pConfig> = Arc::new(P2pConfig {
         bind_addresses: Default::default(),
+        socks5_proxy: None,
         boot_nodes: Default::default(),
         reserved_nodes: Default::default(),
         max_inbound_connections: Default::default(),