Formatting and debugging changes

Cargo.lock

@@ -1881,6 +1881,7 @@ dependencies = [
  "futures",
  "lazy_static",
  "libp2p-core",
+ "libp2p-mplex",
  "libp2p-noise 0.30.0",
  "libp2p-ping",
  "libp2p-swarm",

libp2p-tor/Cargo.toml

@@ -21,8 +21,8 @@ tracing = "0.1"
 [dev-dependencies]
 libp2p = { version = "0.37", default-features = false, features = [ "yamux", "noise", "ping", "mplex" ] }
 rand = "0.8"
+reqwest = { version = "0.11", features = [ "rustls-tls", "stream", "socks" ], default-features = false }
 tempfile = "3"
 testcontainers = "0.12"
 tokio = { version = "1", features = [ "full" ] }
 tracing-subscriber = { version = "0.2", default-features = false, features = [ "fmt", "ansi", "env-filter", "chrono", "tracing-log" ] }
-reqwest = { version = "0.11", features = [ "rustls-tls", "stream", "socks" ], default-features = false }

libp2p-tor/examples/dialer.rs

@@ -1,17 +1,16 @@
+use anyhow::{anyhow, bail, Result};
 use libp2p::core::muxing::StreamMuxerBox;
-use libp2p::core::upgrade::{Version, SelectUpgrade};
+use libp2p::core::upgrade::{SelectUpgrade, Version};
+use libp2p::mplex::MplexConfig;
 use libp2p::ping::{Ping, PingEvent, PingSuccess};
 use libp2p::swarm::{SwarmBuilder, SwarmEvent};
 use libp2p::{identity, noise, yamux, Multiaddr, Swarm, Transport};
 use libp2p_tor::dial_only;
 use std::time::Duration;
-use libp2p::mplex::MplexConfig;
-use anyhow::{anyhow, bail, Result};
 use tracing_subscriber::util::SubscriberInitExt;
 
-#[tokio::main]
+#[tokio::main(flavor = "multi_thread")]
 async fn main() -> Result<()> {
-
     let _guard = tracing_subscriber::fmt()
         .with_env_filter("debug,libp2p_tor=debug") // add `reqwest::connect::verbose=trace` if you want to logs of the RPC clients
         .with_test_writer()
@@ -28,7 +27,6 @@ async fn main() -> Result<()> {
         bail!("Tor is currently not running")
     }
 
-
     let addr_to_dial = "/onion3/jpclybnowuibjexya3qggzvzkoeruuav4nyjlxpnkrosldsvykfbn6qd:7654/p2p/12D3KooWHKqGyK4hVtf5BQY8GpbY6fSGKDZ8eBXMQ3H2RsdnKVzC"
         .parse::<Multiaddr>()
         .unwrap();

libp2p-tor/examples/listener.rs

@@ -1,22 +1,21 @@
 use libp2p::core::muxing::StreamMuxerBox;
-use libp2p::core::upgrade::{Version, SelectUpgrade};
+use libp2p::core::upgrade::{SelectUpgrade, Version};
+use libp2p::identity::Keypair;
+use libp2p::mplex::MplexConfig;
 use libp2p::ping::{Ping, PingEvent, PingSuccess};
 use libp2p::swarm::{SwarmBuilder, SwarmEvent};
-use libp2p::{identity, noise, yamux, Swarm, Transport, Multiaddr};
+use libp2p::{identity, noise, yamux, Multiaddr, Swarm, Transport};
 use libp2p_tor::duplex;
 use libp2p_tor::torut_ext::AuthenticatedConnectionExt;
 use noise::NoiseConfig;
+use std::str::FromStr;
 use std::time::Duration;
 use torut::control::AuthenticatedConn;
 use torut::onion::TorSecretKeyV3;
-use std::str::FromStr;
-use libp2p::mplex::MplexConfig;
-use libp2p::identity::Keypair;
 use tracing_subscriber::util::SubscriberInitExt;
 
-#[tokio::main]
+#[tokio::main(flavor = "multi_thread")]
 async fn main() {
-
     let _guard = tracing_subscriber::fmt()
         .with_env_filter("debug,libp2p_tor=debug") // add `reqwest::connect::verbose=trace` if you want to logs of the RPC clients
         .with_test_writer()
@@ -26,21 +25,38 @@ async fn main() {
 
     let key = fixed_onion_identity();
 
-    let onion_address = key.public().get_onion_address().get_address_without_dot_onion();
+    let onion_address = key
+        .public()
+        .get_onion_address()
+        .get_address_without_dot_onion();
     let onion_port = 7654;
 
-    let mut swarm = new_swarm(key).await;
+    let mut swarm = new_swarm().await;
     let peer_id = *swarm.local_peer_id();
 
     println!("Peer-ID: {}", peer_id);
-    // TODO: Figure out what to with the port, we could also set it to 0 and then imply it from the assigned port
+    // TODO: Figure out what to with the port, we could also set it to 0 and then
-    swarm.listen_on(Multiaddr::from_str(format!("/onion3/{}:{}", onion_address, onion_port).as_str()).unwrap()).unwrap();
+    // imply it from the assigned port swarm.listen_on(Multiaddr::
+    // from_str(format!("/onion3/{}:{}", onion_address,
+    // onion_port).as_str()).unwrap()).unwrap();
+    swarm
+        .listen_on(
+            Multiaddr::from_str(format!("/ip4/127.0.0.1/tcp/{}", onion_port).as_str()).unwrap(),
+        )
+        .unwrap();
 
     loop {
         match swarm.next_event().await {
             SwarmEvent::NewListenAddr(addr) => {
                 println!("Listening on {}", addr);
-                println!("Connection string: {}/p2p/{}", addr, peer_id)
+                println!("Connection string: {}/p2p/{}", addr, peer_id);
+
+                AuthenticatedConn::new(9051)
+                    .await
+                    .unwrap()
+                    .add_ephemeral_service(&key, onion_port, onion_port)
+                    .await
+                    .unwrap();
             }
             SwarmEvent::ConnectionEstablished {
                 peer_id, endpoint, ..
@@ -74,32 +90,28 @@ async fn main() {
 ///
 /// In particular, this swarm can create ephemeral hidden services on the
 /// configured Tor node.
-async fn new_swarm(key: TorSecretKeyV3) -> Swarm<Ping> {
+async fn new_swarm() -> Swarm<Ping> {
     let identity = fixed_libp2p_identity();
 
     SwarmBuilder::new(
-        duplex::TorConfig::new(
+        libp2p::tcp::TokioTcpConfig::new()
-            AuthenticatedConn::new(9051).await.unwrap(),
+            .boxed()
-            key,
+            .upgrade(Version::V1)
-        )
+            .authenticate(
-        .await
+                NoiseConfig::xx(
-        .unwrap()
+                    noise::Keypair::<noise::X25519Spec>::new()
-        .upgrade(Version::V1)
+                        .into_authentic(&identity)
-        .authenticate(
+                        .unwrap(),
-            NoiseConfig::xx(
+                )
-                noise::Keypair::<noise::X25519Spec>::new()
+                .into_authenticated(),
-                    .into_authentic(&identity)
-                    .unwrap(),
             )
-            .into_authenticated(),
+            .multiplex(SelectUpgrade::new(
-        )
+                yamux::YamuxConfig::default(),
-        .multiplex(SelectUpgrade::new(
+                MplexConfig::new(),
-            yamux::YamuxConfig::default(),
+            ))
-            MplexConfig::new(),
+            .timeout(Duration::from_secs(20))
-        ))
+            .map(|(peer, muxer), _| (peer, StreamMuxerBox::new(muxer)))
-        .timeout(Duration::from_secs(20))
+            .boxed(),
-        .map(|(peer, muxer), _| (peer, StreamMuxerBox::new(muxer)))
-        .boxed(),
         Ping::default(),
         identity.public().into_peer_id(),
     )
@@ -110,15 +122,26 @@ async fn new_swarm(key: TorSecretKeyV3) -> Swarm<Ping> {
 }
 
 fn fixed_onion_identity() -> TorSecretKeyV3 {
-    // randomly generated bytes, corresponding onion address: jpclybnowuibjexya3qggzvzkoeruuav4nyjlxpnkrosldsvykfbn6qd
+    // randomly generated bytes, corresponding onion address:
-    let fixed_onion_bytes = [7, 164, 217, 80, 139, 239, 11, 110, 37, 77, 191, 158, 206, 252, 178, 188, 147, 98, 54, 13, 35, 183, 114, 231, 202, 38, 30, 29, 245, 8, 118, 153, 55, 141, 228, 109, 78, 189, 120, 28, 172, 131, 198, 55, 113, 47, 131, 135, 139, 117, 182, 195, 46, 34, 234, 169, 85, 96, 203, 215, 7, 155, 209, 211];
+    // jpclybnowuibjexya3qggzvzkoeruuav4nyjlxpnkrosldsvykfbn6qd
+    let fixed_onion_bytes = [
+        7, 164, 217, 80, 139, 239, 11, 110, 37, 77, 191, 158, 206, 252, 178, 188, 147, 98, 54, 13,
+        35, 183, 114, 231, 202, 38, 30, 29, 245, 8, 118, 153, 55, 141, 228, 109, 78, 189, 120, 28,
+        172, 131, 198, 55, 113, 47, 131, 135, 139, 117, 182, 195, 46, 34, 234, 169, 85, 96, 203,
+        215, 7, 155, 209, 211,
+    ];
     fixed_onion_bytes.into()
 }
 
 fn fixed_libp2p_identity() -> Keypair {
-    // randomly venerated bytes, corresponding peer-id: 12D3KooWHKqGyK4hVtf5BQY8GpbY6fSGKDZ8eBXMQ3H2RsdnKVzC
+    // randomly venerated bytes, corresponding peer-id:
-    let fixed_identity = [75,146,26,107,50,252,71,2,238,224,92,112,216,238,131,57,84,9,218,120,195,9,129,102,42,206,165,102,32,238,158,248];
+    // 12D3KooWHKqGyK4hVtf5BQY8GpbY6fSGKDZ8eBXMQ3H2RsdnKVzC
+    let fixed_identity = [
+        75, 146, 26, 107, 50, 252, 71, 2, 238, 224, 92, 112, 216, 238, 131, 57, 84, 9, 218, 120,
+        195, 9, 129, 102, 42, 206, 165, 102, 32, 238, 158, 248,
+    ];
 
-    let key = identity::ed25519::SecretKey::from_bytes(fixed_identity).expect("we always pass 32 bytes");
+    let key =
+        identity::ed25519::SecretKey::from_bytes(fixed_identity).expect("we always pass 32 bytes");
     identity::Keypair::Ed25519(key.into())
 }

libp2p-tor/src/duplex.rs

@@ -5,7 +5,7 @@ use futures::future::BoxFuture;
 use futures::prelude::*;
 use libp2p::core::multiaddr::{Multiaddr, Protocol};
 use libp2p::core::transport::map_err::MapErr;
-use libp2p::core::transport::{ListenerEvent, TransportError};
+use libp2p::core::transport::{Boxed, ListenerEvent, TransportError};
 use libp2p::core::Transport;
 use libp2p::futures::stream::BoxStream;
 use libp2p::futures::{StreamExt, TryStreamExt};
@@ -37,17 +37,14 @@ impl TorConfig {
         let socks_port = client.get_socks_port().await?;
 
         Ok(Self {
-            inner: TokioTcpConfig::new().map_err(Error::InnerTransprot),
+            inner: TokioTcpConfig::new().map_err(Error::InnerTransport),
             tor_client: Arc::new(Mutex::new(client)),
             key,
             socks_port,
         })
     }
 
-    pub async fn from_control_port(
+    pub async fn from_control_port(control_port: u16, key: TorSecretKeyV3) -> Result<Self, Error> {
-        control_port: u16,
-        key: TorSecretKeyV3,
-    ) -> Result<Self, Error> {
         let client = AuthenticatedConn::new(control_port).await?;
 
         Self::new(client, key).await
@@ -82,7 +79,8 @@ impl Transport for TorConfig {
             return Err(TransportError::MultiaddrNotSupported(addr));
         }
 
-        let localhost_tcp_random_port_addr = format!("/ip4/127.0.0.1/tcp/{}", onion_port).as_str()
+        let localhost_tcp_random_port_addr = format!("/ip4/127.0.0.1/tcp/{}", onion_port)
+            .as_str()
             .parse()
             .expect("always a valid multiaddr");
 
@@ -109,7 +107,9 @@ impl Transport for TorConfig {
                                     })
                                     .expect("TODO: Error handling");
 
-                                // TODO: Don't fully understand this part, why would we have two different multiaddresses here? the actual onion address and the multiaddress would make more sense...?
+                                // TODO: Don't fully understand this part, why would we have two
+                                // different multiaddresses here? the actual onion address and the
+                                // multiaddress would make more sense...?
                                 tracing::debug!(
                                     "Setting up hidden service at {} to forward to {}",
                                     onion_multiaddress,
@@ -120,7 +120,9 @@ impl Transport for TorConfig {
                                     .clone()
                                     .lock()
                                     .await
-                                    // TODO: Potentially simplify this, in our setup the onion port is always equal to the local port. Otherwise we would have the user provide an additional port for the oion service.
+                                    // TODO: Potentially simplify this, in our setup the onion port
+                                    // is always equal to the local port. Otherwise we would have
+                                    // the user provide an additional port for the oion service.
                                     .add_ephemeral_service(&key, onion_port, local_port)
                                     .await
                                 {
@@ -139,8 +141,9 @@ impl Transport for TorConfig {
                             },
                             // TODO: why was the constructed multiaddr used here?
                             ListenerEvent::AddressExpired(adr) => {
-                                // TODO: even if so, why would we ignore it? Far more logical to just use it...
+                                // TODO: even if so, why would we ignore it? Far more logical to
-                                // can ignore address because we only ever listened on one and we
+                                // just use it... can ignore address
+                                // because we only ever listened on one and we
                                 // know which one that was
 
                                 let onion_address_without_dot_onion = key
@@ -148,7 +151,10 @@ impl Transport for TorConfig {
                                     .get_onion_address()
                                     .get_address_without_dot_onion();
 
-                                tracing::debug!("Listening expired, removing onion {}", onion_address_without_dot_onion);
+                                tracing::debug!(
+                                    "Listening expired, removing onion {}",
+                                    onion_address_without_dot_onion
+                                );
 
                                 match tor_client
                                     .lock()
@@ -181,7 +187,7 @@ impl Transport for TorConfig {
         Ok(crate::dial_via_tor(address, self.socks_port).boxed())
     }
 
-    fn address_translation(&self, _: &Multiaddr, _: &Multiaddr) -> Option<Multiaddr> {
+    fn address_translation(&self, listen: &Multiaddr, observed: &Multiaddr) -> Option<Multiaddr> {
-        None // address translation for tor doesn't make any sense :)
+        self.inner.address_translation(listen, observed)
     }
 }

libp2p-tor/src/lib.rs

@@ -24,7 +24,7 @@ pub enum Error {
     OnlyWildcardAllowed,
     Torut(torut_ext::Error),
     UnreachableProxy(tokio_socks::Error),
-    InnerTransprot(io::Error),
+    InnerTransport(io::Error),
 }
 
 impl std::error::Error for Error {}

libp2p-tor/src/torut_ext.rs

@@ -1,6 +1,6 @@
 use std::borrow::Cow;
 use std::future::Future;
-use std::net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4};
+use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 use std::num::ParseIntError;
 use std::{io, iter};
 use torut::control::{AsyncEvent, AuthenticatedConn, TorAuthData, UnauthenticatedConn};
@@ -83,7 +83,10 @@ impl AuthenticatedConnectionExt for AuthenticatedConn<tokio::net::TcpStream, Asy
         onion_port: u16,
         local_port: u16,
     ) -> Result<(), Error> {
-        println!("Adding ephemeral service, onion port {}, local port {}", onion_port, local_port);
+        println!(
+            "Adding ephemeral service, onion port {}, local port {}",
+            onion_port, local_port
+        );
 
         self.add_onion_v3(
             &key,