WIP: libp2p-tor

.cargo/config.toml

@@ -1,2 +1,5 @@
 [target.armv7-unknown-linux-gnueabihf]
 linker = "arm-linux-gnueabihf-gcc"
+
+[target.aarch64-unknown-linux-gnu]
+linker = "aarch64-linux-gnu-gcc"

Cargo.lock

@@ -1735,6 +1735,7 @@ dependencies = [
  "libp2p-dns",
  "libp2p-mplex",
  "libp2p-noise",
+ "libp2p-ping",
  "libp2p-request-response",
  "libp2p-swarm",
  "libp2p-swarm-derive",
@@ -1844,6 +1845,21 @@ dependencies = [
  "zeroize",
 ]
 
+[[package]]
+name = "libp2p-ping"
+version = "0.29.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bf4bfaffac63bf3c7ec11ed9d8879d455966ddea7e78ee14737f0b6dce0d1cd1"
+dependencies = [
+ "futures",
+ "libp2p-core",
+ "libp2p-swarm",
+ "log 0.4.14",
+ "rand 0.7.3",
+ "void",
+ "wasm-timer",
+]
+
 [[package]]
 name = "libp2p-request-response"
 version = "0.11.0"
@@ -1907,6 +1923,26 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "libp2p-tor"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "async-trait",
+ "data-encoding",
+ "futures",
+ "libp2p",
+ "rand 0.8.3",
+ "reqwest",
+ "tempfile",
+ "testcontainers 0.12.0",
+ "tokio",
+ "tokio-socks",
+ "torut",
+ "tracing",
+ "tracing-subscriber",
+]
+
 [[package]]
 name = "libp2p-websocket"
 version = "0.29.0"

Cargo.toml

@@ -1,2 +1,2 @@
 [workspace]
-members = [ "monero-harness", "monero-rpc", "swap", "monero-wallet" ]
+members = [ "monero-harness", "monero-rpc", "swap", "monero-wallet", "libp2p-tor" ]

libp2p-tor/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "libp2p-tor"
+version = "0.1.0"
+authors = ["Thomas Eizinger <thomas@eizinger.io>"]
+edition = "2018"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+anyhow = "1" # TODO: Get rid of anyhow dependency
+torut = { version = "0.1", default-features = false, features = ["v3", "control"] }
+tokio-socks = "0.5"
+libp2p = { version = "0.37", default-features = false, features = ["tcp-tokio"] }
+tokio = { version = "1", features = ["sync"] }
+futures = "0.3"
+tracing = "0.1"
+data-encoding = "2.3"
+reqwest = { version = "0.11", features = ["socks", "rustls-tls"], default-features = false }
+async-trait = "0.1"
+rand = { version = "0.8", optional = true }
+
+[dev-dependencies]
+tempfile = "3"
+tokio = { version = "1", features = ["full"] }
+rand = "0.8"
+libp2p = { version = "0.37", default-features = false, features = ["yamux", "noise", "ping"] }
+tracing-subscriber = { version = "0.2", default-features = false, features = ["fmt", "ansi", "env-filter", "chrono", "tracing-log"] }
+testcontainers = "0.12"

libp2p-tor/build.rs

@@ -0,0 +1,17 @@
+use std::process::Command;
+
+fn main() {
+    let status = Command::new("docker")
+        .arg("build")
+        .arg("-f")
+        .arg("./tor.Dockerfile")
+        .arg(".")
+        .arg("-t")
+        .arg("testcontainers-tor:latest")
+        .status()
+        .unwrap();
+
+    assert!(status.success());
+
+    println!("cargo:rerun-if-changed=./tor.Dockerfile");
+}

libp2p-tor/examples/dialer.rs

@@ -0,0 +1,75 @@
+use libp2p::core::muxing::StreamMuxerBox;
+use libp2p::core::upgrade::Version;
+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;
+
+#[tokio::main]
+async fn main() {
+    let addr_to_dial = std::env::args()
+        .next()
+        .unwrap()
+        .parse::<Multiaddr>()
+        .unwrap();
+
+    let mut swarm = new_swarm();
+
+    println!("Peer-ID: {}", swarm.local_peer_id());
+    swarm.dial_addr(addr_to_dial).unwrap();
+
+    loop {
+        match swarm.next_event().await {
+            SwarmEvent::ConnectionEstablished {
+                peer_id, endpoint, ..
+            } => {
+                println!(
+                    "Connected to {} via {}",
+                    peer_id,
+                    endpoint.get_remote_address()
+                );
+            }
+            SwarmEvent::Behaviour(PingEvent { result, peer }) => match result {
+                Ok(PingSuccess::Pong) => {
+                    println!("Got pong from {}", peer);
+                }
+                Ok(PingSuccess::Ping { rtt }) => {
+                    println!("Pinged {} with rtt of {}s", peer, rtt.as_secs());
+                }
+                Err(failure) => {
+                    println!("Failed to ping {}: {}", peer, failure)
+                }
+            },
+            _ => {}
+        }
+    }
+}
+
+/// Builds a new swarm that is capable of dialling onion address.
+fn new_swarm() -> Swarm<Ping> {
+    let identity = identity::Keypair::generate_ed25519();
+
+    SwarmBuilder::new(
+        dial_only::TorConfig::new(9050)
+            .upgrade(Version::V1)
+            .authenticate(
+                noise::NoiseConfig::xx(
+                    noise::Keypair::<noise::X25519Spec>::new()
+                        .into_authentic(&identity)
+                        .unwrap(),
+                )
+                .into_authenticated(),
+            )
+            .multiplex(yamux::YamuxConfig::default())
+            .timeout(Duration::from_secs(20))
+            .map(|(peer, muxer), _| (peer, StreamMuxerBox::new(muxer)))
+            .boxed(),
+        Ping::default(),
+        identity.public().into_peer_id(),
+    )
+    .executor(Box::new(|f| {
+        tokio::spawn(f);
+    }))
+    .build()
+}

libp2p-tor/examples/listener.rs

@@ -0,0 +1,98 @@
+use libp2p::core::muxing::StreamMuxerBox;
+use libp2p::core::upgrade::Version;
+use libp2p::ping::{Ping, PingEvent, PingSuccess};
+use libp2p::swarm::{SwarmBuilder, SwarmEvent};
+use libp2p::{identity, noise, yamux, Swarm, Transport};
+use libp2p_tor::duplex;
+use libp2p_tor::torut_ext::AuthenticatedConnectionExt;
+use noise::NoiseConfig;
+use rand::Rng;
+use std::time::Duration;
+use torut::control::AuthenticatedConn;
+use torut::onion::TorSecretKeyV3;
+
+#[tokio::main]
+async fn main() {
+    let wildcard_multiaddr =
+        "/onion3/WWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWW:8080"
+            .parse()
+            .unwrap();
+
+    let mut swarm = new_swarm().await;
+
+    println!("Peer-ID: {}", swarm.local_peer_id());
+    swarm.listen_on(wildcard_multiaddr).unwrap();
+
+    loop {
+        match swarm.next_event().await {
+            SwarmEvent::NewListenAddr(addr) => {
+                println!("Listening on {}", addr);
+            }
+            SwarmEvent::ConnectionEstablished {
+                peer_id, endpoint, ..
+            } => {
+                println!(
+                    "Connected to {} via {}",
+                    peer_id,
+                    endpoint.get_remote_address()
+                );
+            }
+            SwarmEvent::Behaviour(PingEvent { result, peer }) => match result {
+                Ok(PingSuccess::Pong) => {
+                    println!("Got pong from {}", peer);
+                }
+                Ok(PingSuccess::Ping { rtt }) => {
+                    println!("Pinged {} with rtt of {}s", peer, rtt.as_secs());
+                }
+                Err(failure) => {
+                    println!("Failed to ping {}: {}", peer, failure)
+                }
+            },
+            _ => {}
+        }
+    }
+}
+
+/// Builds a new swarm that is capable of listening and dialling on the Tor
+/// network.
+///
+/// In particular, this swarm can create ephemeral hidden services on the
+/// configured Tor node.
+async fn new_swarm() -> Swarm<Ping> {
+    let identity = identity::Keypair::generate_ed25519();
+
+    SwarmBuilder::new(
+        duplex::TorConfig::new(
+            AuthenticatedConn::new(9051).await.unwrap(),
+            random_onion_identity,
+        )
+        .await
+        .unwrap()
+        .upgrade(Version::V1)
+        .authenticate(
+            NoiseConfig::xx(
+                noise::Keypair::<noise::X25519Spec>::new()
+                    .into_authentic(&identity)
+                    .unwrap(),
+            )
+            .into_authenticated(),
+        )
+        .multiplex(yamux::YamuxConfig::default())
+        .timeout(Duration::from_secs(20))
+        .map(|(peer, muxer), _| (peer, StreamMuxerBox::new(muxer)))
+        .boxed(),
+        Ping::default(),
+        identity.public().into_peer_id(),
+    )
+    .executor(Box::new(|f| {
+        tokio::spawn(f);
+    }))
+    .build()
+}
+
+fn random_onion_identity() -> TorSecretKeyV3 {
+    let mut onion_key_bytes = [0u8; 64];
+    rand::thread_rng().fill(&mut onion_key_bytes);
+
+    onion_key_bytes.into()
+}

libp2p-tor/src/dial_only.rs

@@ -0,0 +1,57 @@
+use crate::torut_ext::AuthenticatedConnectionExt;
+use crate::{fmt_as_tor_compatible_address, Error};
+use anyhow::Result;
+use fmt_as_tor_compatible_address::fmt_as_tor_compatible_address;
+use futures::future::BoxFuture;
+use futures::prelude::*;
+use libp2p::core::multiaddr::Multiaddr;
+use libp2p::core::transport::{ListenerEvent, TransportError};
+use libp2p::core::Transport;
+use libp2p::futures::stream::BoxStream;
+use libp2p::tcp::tokio::TcpStream;
+use torut::control::AuthenticatedConn;
+
+#[derive(Clone)]
+pub struct TorConfig {
+    socks_port: u16,
+}
+
+impl TorConfig {
+    pub fn new(socks_port: u16) -> Self {
+        Self { socks_port }
+    }
+
+    pub async fn from_control_port(control_port: u16) -> Result<Self, Error> {
+        let mut client = AuthenticatedConn::new(control_port).await?;
+        let socks_port = client.get_socks_port().await?;
+
+        Ok(Self::new(socks_port))
+    }
+}
+
+impl Transport for TorConfig {
+    type Output = TcpStream;
+    type Error = Error;
+    #[allow(clippy::type_complexity)]
+    type Listener =
+        BoxStream<'static, Result<ListenerEvent<Self::ListenerUpgrade, Self::Error>, Self::Error>>;
+    type ListenerUpgrade = BoxFuture<'static, Result<Self::Output, Self::Error>>;
+    type Dial = BoxFuture<'static, Result<Self::Output, Self::Error>>;
+
+    fn listen_on(self, addr: Multiaddr) -> Result<Self::Listener, TransportError<Self::Error>> {
+        Err(TransportError::MultiaddrNotSupported(addr))
+    }
+
+    fn dial(self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
+        tracing::debug!("Connecting through Tor proxy to address {}", addr);
+
+        let address = fmt_as_tor_compatible_address(addr.clone())
+            .ok_or(TransportError::MultiaddrNotSupported(addr))?;
+
+        Ok(crate::dial_via_tor(address, self.socks_port).boxed())
+    }
+
+    fn address_translation(&self, _: &Multiaddr, _: &Multiaddr) -> Option<Multiaddr> {
+        None // address translation for tor doesn't make any sense :)
+    }
+}

libp2p-tor/src/duplex.rs

@@ -0,0 +1,187 @@
+use crate::torut_ext::AuthenticatedConnectionExt;
+use crate::{fmt_as_tor_compatible_address, torut_ext, Error};
+use fmt_as_tor_compatible_address::fmt_as_tor_compatible_address;
+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;
+use libp2p::futures::stream::BoxStream;
+use libp2p::futures::{StreamExt, TryStreamExt};
+use libp2p::tcp::{GenTcpConfig, TokioTcpConfig};
+use std::sync::Arc;
+use tokio::sync::Mutex;
+use torut::control::{AsyncEvent, AuthenticatedConn};
+use torut::onion::TorSecretKeyV3;
+
+/// This is the hash of
+/// `/onion3/WWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWW`.
+const WILDCARD_ONION_ADDR_HASH: [u8; 35] = [
+    181, 173, 107, 90, 214, 181, 173, 107, 90, 214, 181, 173, 107, 90, 214, 181, 173, 107, 90, 214,
+    181, 173, 107, 90, 214, 181, 173, 107, 90, 214, 181, 173, 107, 90, 214,
+];
+
+type TorutAsyncEventHandler =
+    fn(
+        AsyncEvent<'_>,
+    ) -> Box<dyn Future<Output = Result<(), torut::control::ConnError>> + Unpin + Send>;
+
+#[derive(Clone)]
+pub struct TorConfig {
+    inner: MapErr<GenTcpConfig<libp2p::tcp::tokio::Tcp>, fn(std::io::Error) -> Error>, /* TODO: Make generic over async-std / tokio */
+    tor_client: Arc<Mutex<AuthenticatedConn<tokio::net::TcpStream, TorutAsyncEventHandler>>>,
+    onion_key_generator: Arc<dyn (Fn() -> TorSecretKeyV3) + Send + Sync>,
+    socks_port: u16,
+}
+
+impl TorConfig {
+    pub async fn new(
+        mut client: AuthenticatedConn<tokio::net::TcpStream, TorutAsyncEventHandler>,
+        onion_key_generator: impl (Fn() -> TorSecretKeyV3) + Send + Sync + 'static,
+    ) -> Result<Self, Error> {
+        let socks_port = client.get_socks_port().await?;
+
+        Ok(Self {
+            inner: TokioTcpConfig::new().map_err(Error::InnerTransprot),
+            tor_client: Arc::new(Mutex::new(client)),
+            onion_key_generator: Arc::new(onion_key_generator),
+            socks_port,
+        })
+    }
+
+    pub async fn from_control_port(
+        control_port: u16,
+        key_generator: impl (Fn() -> TorSecretKeyV3) + Send + Sync + 'static,
+    ) -> Result<Self, Error> {
+        let client = AuthenticatedConn::new(control_port).await?;
+
+        Self::new(client, key_generator).await
+    }
+}
+
+impl Transport for TorConfig {
+    type Output = libp2p::tcp::tokio::TcpStream;
+    type Error = Error;
+    #[allow(clippy::type_complexity)]
+    type Listener =
+        BoxStream<'static, Result<ListenerEvent<Self::ListenerUpgrade, Self::Error>, Self::Error>>;
+    type ListenerUpgrade = BoxFuture<'static, Result<Self::Output, Self::Error>>;
+    type Dial = BoxFuture<'static, Result<Self::Output, Self::Error>>;
+
+    fn listen_on(self, addr: Multiaddr) -> Result<Self::Listener, TransportError<Self::Error>> {
+        let mut protocols = addr.iter();
+        let onion = if let Protocol::Onion3(onion) = protocols
+            .next()
+            .ok_or_else(|| TransportError::MultiaddrNotSupported(addr.clone()))?
+        {
+            onion
+        } else {
+            return Err(TransportError::MultiaddrNotSupported(addr));
+        };
+
+        if onion.hash() != &WILDCARD_ONION_ADDR_HASH {
+            return Err(TransportError::Other(Error::OnlyWildcardAllowed));
+        }
+
+        let localhost_tcp_random_port_addr = "/ip4/127.0.0.1/tcp/0"
+            .parse()
+            .expect("always a valid multiaddr");
+
+        let listener = self.inner.listen_on(localhost_tcp_random_port_addr)?;
+
+        let key: TorSecretKeyV3 = (self.onion_key_generator)();
+        let onion_bytes = key.public().get_onion_address().get_raw_bytes();
+        let onion_port = onion.port();
+
+        let tor_client = self.tor_client;
+
+        let listener = listener
+            .and_then({
+                move |event| {
+                    let tor_client = tor_client.clone();
+                    let key = key.clone();
+                    let onion_multiaddress =
+                        Multiaddr::empty().with(Protocol::Onion3((onion_bytes, onion_port).into()));
+
+                    async move {
+                        Ok(match event {
+                            ListenerEvent::NewAddress(address) => {
+                                let local_port = address
+                                    .iter()
+                                    .find_map(|p| match p {
+                                        Protocol::Tcp(port) => Some(port),
+                                        _ => None,
+                                    })
+                                    .expect("TODO: Error handling");
+
+                                tracing::debug!(
+                                    "Setting up hidden service at {} to forward to {}",
+                                    onion_multiaddress,
+                                    address
+                                );
+
+                                match tor_client
+                                    .clone()
+                                    .lock()
+                                    .await
+                                    .add_ephemeral_service(&key, onion_port, local_port)
+                                    .await
+                                {
+                                    Ok(()) => ListenerEvent::NewAddress(onion_multiaddress.clone()),
+                                    Err(e) => ListenerEvent::Error(Error::Torut(e)),
+                                }
+                            }
+                            ListenerEvent::Upgrade {
+                                upgrade,
+                                local_addr,
+                                remote_addr,
+                            } => ListenerEvent::Upgrade {
+                                upgrade: upgrade.boxed(),
+                                local_addr,
+                                remote_addr,
+                            },
+                            ListenerEvent::AddressExpired(_) => {
+                                // can ignore address because we only ever listened on one and we
+                                // know which one that was
+
+                                let onion_address_without_dot_onion = key
+                                    .public()
+                                    .get_onion_address()
+                                    .get_address_without_dot_onion();
+
+                                match tor_client
+                                    .lock()
+                                    .await
+                                    .del_onion(&onion_address_without_dot_onion)
+                                    .await
+                                {
+                                    Ok(()) => ListenerEvent::AddressExpired(onion_multiaddress),
+                                    Err(e) => ListenerEvent::Error(Error::Torut(
+                                        torut_ext::Error::Connection(e),
+                                    )),
+                                }
+                            }
+                            ListenerEvent::Error(e) => ListenerEvent::Error(e),
+                        })
+                    }
+                }
+            })
+            .boxed();
+
+        Ok(listener)
+    }
+
+    fn dial(self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
+        tracing::debug!("Connecting through Tor proxy to address {}", addr);
+
+        let address = fmt_as_tor_compatible_address(addr.clone())
+            .ok_or(TransportError::MultiaddrNotSupported(addr))?;
+
+        Ok(crate::dial_via_tor(address, self.socks_port).boxed())
+    }
+
+    fn address_translation(&self, _: &Multiaddr, _: &Multiaddr) -> Option<Multiaddr> {
+        None // address translation for tor doesn't make any sense :)
+    }
+}

libp2p-tor/src/fmt_as_tor_compatible_address.rs

@@ -0,0 +1,60 @@
+use data_encoding::BASE32;
+use libp2p::multiaddr::Protocol;
+use libp2p::Multiaddr;
+
+/// Tor expects an address format of ADDR:PORT.
+/// This helper function tries to convert the provided multi-address into this
+/// format. None is returned if an unsupported protocol was provided.
+pub fn fmt_as_tor_compatible_address(multi: Multiaddr) -> Option<String> {
+    let mut protocols = multi.iter();
+    let address_string = match protocols.next()? {
+        // if it is an Onion address, we have all we need and can return
+        Protocol::Onion3(addr) => {
+            return Some(format!(
+                "{}.onion:{}",
+                BASE32.encode(addr.hash()).to_lowercase(),
+                addr.port()
+            ));
+        }
+        // Deal with non-onion addresses
+        Protocol::Ip4(addr) => format!("{}", addr),
+        Protocol::Ip6(addr) => format!("{}", addr),
+        Protocol::Dns(addr) => format!("{}", addr),
+        Protocol::Dns4(addr) => format!("{}", addr),
+        _ => return None,
+    };
+
+    let port = match protocols.next()? {
+        Protocol::Tcp(port) => port,
+        Protocol::Udp(port) => port,
+        _ => return None,
+    };
+
+    Some(format!("{}:{}", address_string, port))
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn test_fmt_as_tor_compatible_address() {
+        let test_cases = &[
+            ("/onion3/oarchy4tamydxcitaki6bc2v4leza6v35iezmu2chg2bap63sv6f2did:1024/p2p/12D3KooWPD4uHN74SHotLN7VCH7Fm8zZgaNVymYcpeF1fpD2guc9", Some("oarchy4tamydxcitaki6bc2v4leza6v35iezmu2chg2bap63sv6f2did.onion:1024")),
+                ("/ip4/127.0.0.1/tcp/7777", Some("127.0.0.1:7777")),
+                ("/ip6/2001:db8:85a3:8d3:1319:8a2e:370:7348/tcp/7777", Some("2001:db8:85a3:8d3:1319:8a2e:370:7348:7777")),
+                ("/ip4/127.0.0.1/udp/7777", Some("127.0.0.1:7777")),
+                ("/ip4/127.0.0.1/tcp/7777/ws", Some("127.0.0.1:7777")),
+                ("/dns4/randomdomain.com/tcp/7777", Some("randomdomain.com:7777")),
+                ("/dns/randomdomain.com/tcp/7777", Some("randomdomain.com:7777")),
+                ("/dnsaddr/randomdomain.com", None),
+        ];
+
+        for (multiaddress, expected_address) in test_cases {
+            let actual_address =
+                fmt_as_tor_compatible_address(multiaddress.parse().expect("a valid multi-address"));
+
+            assert_eq!(&actual_address.as_deref(), expected_address)
+        }
+    }
+}

libp2p-tor/src/lib.rs

@@ -0,0 +1,42 @@
+use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
+use std::{fmt, io};
+
+use libp2p::tcp::tokio::TcpStream;
+use tokio_socks::tcp::Socks5Stream;
+
+pub mod dial_only;
+pub mod duplex;
+mod fmt_as_tor_compatible_address;
+pub mod torut_ext;
+
+async fn dial_via_tor(onion_address: String, socks_port: u16) -> anyhow::Result<TcpStream, Error> {
+    let sock = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, socks_port));
+    let stream = Socks5Stream::connect(sock, onion_address)
+        .await
+        .map_err(Error::UnreachableProxy)?;
+    let stream = TcpStream(stream.into_inner());
+
+    Ok(stream)
+}
+
+#[derive(Debug)]
+pub enum Error {
+    OnlyWildcardAllowed,
+    Torut(torut_ext::Error),
+    UnreachableProxy(tokio_socks::Error),
+    InnerTransprot(io::Error),
+}
+
+impl std::error::Error for Error {}
+
+impl fmt::Display for Error {
+    fn fmt(&self, _: &mut fmt::Formatter<'_>) -> fmt::Result {
+        todo!()
+    }
+}
+
+impl From<torut_ext::Error> for Error {
+    fn from(e: torut_ext::Error) -> Self {
+        Error::Torut(e)
+    }
+}

libp2p-tor/src/torut_ext.rs

@@ -0,0 +1,116 @@
+use std::borrow::Cow;
+use std::future::Future;
+use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
+use std::num::ParseIntError;
+use std::{io, iter};
+use torut::control::{AsyncEvent, AuthenticatedConn, TorAuthData, UnauthenticatedConn};
+use torut::onion::TorSecretKeyV3;
+
+pub type AsyncEventHandler =
+    fn(
+        AsyncEvent<'_>,
+    ) -> Box<dyn Future<Output = Result<(), torut::control::ConnError>> + Unpin + Send>;
+
+#[derive(Debug)]
+pub enum Error {
+    FailedToConnect(io::Error),
+    NoAuthData(Option<io::Error>),
+    Connection(torut::control::ConnError),
+    FailedToAddHiddenService(torut::control::ConnError),
+    FailedToParsePort(ParseIntError),
+}
+
+impl From<torut::control::ConnError> for Error {
+    fn from(e: torut::control::ConnError) -> Self {
+        Error::Connection(e)
+    }
+}
+
+impl From<ParseIntError> for Error {
+    fn from(e: ParseIntError) -> Self {
+        Error::FailedToParsePort(e)
+    }
+}
+
+#[async_trait::async_trait]
+pub trait AuthenticatedConnectionExt: Sized {
+    async fn new(control_port: u16) -> Result<Self, Error>;
+    async fn with_password(control_port: u16, password: &str) -> Result<Self, Error>;
+    async fn add_ephemeral_service(
+        &mut self,
+        key: &TorSecretKeyV3,
+        onion_port: u16,
+        local_port: u16,
+    ) -> Result<(), Error>;
+    async fn get_socks_port(&mut self) -> Result<u16, Error>;
+}
+
+#[async_trait::async_trait]
+impl AuthenticatedConnectionExt for AuthenticatedConn<tokio::net::TcpStream, AsyncEventHandler> {
+    async fn new(control_port: u16) -> Result<Self, Error> {
+        let stream = tokio::net::TcpStream::connect(format!("127.0.0.1:{}", control_port))
+            .await
+            .map_err(Error::FailedToConnect)?;
+        let mut uac = UnauthenticatedConn::new(stream);
+
+        let tor_info = uac.load_protocol_info().await?;
+
+        let tor_auth_data = tor_info
+            .make_auth_data()
+            .map_err(|e| Error::NoAuthData(Some(e)))?
+            .ok_or(Error::NoAuthData(None))?;
+
+        uac.authenticate(&tor_auth_data).await?;
+
+        Ok(uac.into_authenticated().await)
+    }
+
+    async fn with_password(control_port: u16, password: &str) -> Result<Self, Error> {
+        let stream = tokio::net::TcpStream::connect(format!("127.0.0.1:{}", control_port))
+            .await
+            .map_err(Error::FailedToConnect)?;
+        let mut uac = UnauthenticatedConn::new(stream);
+
+        uac.authenticate(&TorAuthData::HashedPassword(Cow::Borrowed(password)))
+            .await?;
+
+        Ok(uac.into_authenticated().await)
+    }
+
+    async fn add_ephemeral_service(
+        &mut self,
+        key: &TorSecretKeyV3,
+        onion_port: u16,
+        local_port: u16,
+    ) -> Result<(), Error> {
+        self.add_onion_v3(
+            &key,
+            false,
+            false,
+            false,
+            None,
+            &mut iter::once(&(
+                onion_port,
+                SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(127, 0, 0, 1), local_port)),
+            )),
+        )
+        .await
+        .map_err(Error::FailedToAddHiddenService)
+    }
+
+    async fn get_socks_port(&mut self) -> Result<u16, Error> {
+        const DEFAULT_SOCKS_PORT: u16 = 9050;
+
+        let mut vec = self
+            .get_conf("SocksPort")
+            .await
+            .map_err(Error::Connection)?;
+
+        let first_element = vec
+            .pop()
+            .expect("exactly one element because we requested one config option");
+        let port = first_element.map_or(Ok(DEFAULT_SOCKS_PORT), |port| port.parse())?; // if config is empty, we are listing on the default port
+
+        Ok(port)
+    }
+}

libp2p-tor/tests/integration_test.rs

@@ -0,0 +1,203 @@
+use libp2p::core::muxing::StreamMuxerBox;
+use libp2p::core::transport;
+use libp2p::core::upgrade::Version;
+use libp2p::ping::{Ping, PingEvent};
+use libp2p::swarm::{SwarmBuilder, SwarmEvent};
+use libp2p::tcp::tokio::TcpStream;
+use libp2p::{noise, yamux, Swarm, Transport};
+use libp2p_tor::torut_ext::AuthenticatedConnectionExt;
+use libp2p_tor::{dial_only, duplex};
+use rand::Rng;
+use std::collections::HashMap;
+use std::convert::Infallible;
+use std::future::Future;
+use std::time::Duration;
+use testcontainers::{Container, Docker, Image, WaitForMessage};
+use torut::control::AuthenticatedConn;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn create_ephemeral_service() {
+    tracing_subscriber::fmt().with_env_filter("debug").init();
+    let wildcard_multiaddr =
+        "/onion3/WWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWWW:8080"
+            .parse()
+            .unwrap();
+
+    // let docker = Cli::default();
+    //
+    // let tor1 = docker.run(TorImage::default().with_args(TorArgs {
+    //     control_port: Some(9051),
+    //     socks_port: None
+    // }));
+    // let tor2 = docker.run(TorImage::default());
+    //
+    // let tor1_control_port = tor1.get_host_port(9051).unwrap();
+    // let tor2_socks_port = tor2.get_host_port(9050).unwrap();
+
+    let mut listen_swarm = make_swarm(async move {
+        let mut onion_key_bytes = [0u8; 64];
+        rand::thread_rng().fill(&mut onion_key_bytes);
+
+        duplex::TorConfig::new(
+            AuthenticatedConn::with_password(9051, "supersecret")
+                .await
+                .unwrap(),
+            move || onion_key_bytes.into(),
+        )
+        .await
+        .unwrap()
+        .boxed()
+    })
+    .await;
+    let mut dial_swarm = make_swarm(async { dial_only::TorConfig::new(9050).boxed() }).await;
+
+    listen_swarm.listen_on(wildcard_multiaddr).unwrap();
+
+    let onion_listen_addr = loop {
+        let event = listen_swarm.next_event().await;
+
+        tracing::info!("{:?}", event);
+
+        if let SwarmEvent::NewListenAddr(addr) = event {
+            break addr;
+        }
+    };
+
+    dial_swarm.dial_addr(onion_listen_addr).unwrap();
+
+    loop {
+        tokio::select! {
+            event = listen_swarm.next_event() => {
+                tracing::info!("{:?}", event);
+            },
+            event = dial_swarm.next_event() => {
+                tracing::info!("{:?}", event);
+            }
+        }
+    }
+}
+
+async fn make_swarm(
+    transport_future: impl Future<Output = transport::Boxed<TcpStream>>,
+) -> Swarm<Behaviour> {
+    let identity = libp2p::identity::Keypair::generate_ed25519();
+
+    let dh_keys = noise::Keypair::<noise::X25519Spec>::new()
+        .into_authentic(&identity)
+        .unwrap();
+    let noise = noise::NoiseConfig::xx(dh_keys).into_authenticated();
+
+    let transport = transport_future
+        .await
+        .upgrade(Version::V1)
+        .authenticate(noise)
+        .multiplex(yamux::YamuxConfig::default())
+        .timeout(Duration::from_secs(20))
+        .map(|(peer, muxer), _| (peer, StreamMuxerBox::new(muxer)))
+        .boxed();
+
+    SwarmBuilder::new(
+        transport,
+        Behaviour::default(),
+        identity.public().into_peer_id(),
+    )
+    .executor(Box::new(|f| {
+        tokio::spawn(f);
+    }))
+    .build()
+}
+
+#[derive(Debug)]
+enum OutEvent {
+    Ping(PingEvent),
+}
+
+impl From<PingEvent> for OutEvent {
+    fn from(e: PingEvent) -> Self {
+        OutEvent::Ping(e)
+    }
+}
+
+#[derive(libp2p::NetworkBehaviour, Default)]
+#[behaviour(event_process = false, out_event = "OutEvent")]
+struct Behaviour {
+    ping: Ping,
+}
+
+#[derive(Default)]
+struct TorImage {
+    args: TorArgs,
+}
+
+impl TorImage {
+    // fn control_port_password(&self) -> String {
+    //     "supersecret".to_owned()
+    // }
+}
+
+#[derive(Default, Copy, Clone)]
+struct TorArgs {
+    control_port: Option<u16>,
+    socks_port: Option<u16>,
+}
+
+impl IntoIterator for TorArgs {
+    type Item = String;
+    type IntoIter = std::vec::IntoIter<String>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        let mut args = Vec::new();
+
+        if let Some(port) = self.socks_port {
+            args.push(format!("SocksPort"));
+            args.push(format!("0.0.0.0:{}", port));
+        }
+
+        if let Some(port) = self.control_port {
+            args.push(format!("ControlPort"));
+            args.push(format!("0.0.0.0:{}", port));
+            args.push(format!("HashedControlPassword"));
+            args.push(format!(
+                "16:436B425404AA332A60B4F341C2023146C4B3A80548D757F0BB10DE81B4"
+            ))
+        }
+
+        args.into_iter()
+    }
+}
+
+impl Image for TorImage {
+    type Args = TorArgs;
+    type EnvVars = HashMap<String, String>;
+    type Volumes = HashMap<String, String>;
+    type EntryPoint = Infallible;
+
+    fn descriptor(&self) -> String {
+        "testcontainers-tor:latest".to_owned() // this is build locally using
+                                               // the buildscript
+    }
+
+    fn wait_until_ready<D: Docker>(&self, container: &Container<'_, D, Self>) {
+        container
+            .logs()
+            .stdout
+            .wait_for_message("Bootstrapped 100% (done): Done")
+            .unwrap();
+    }
+
+    fn args(&self) -> Self::Args {
+        self.args.clone()
+    }
+
+    fn env_vars(&self) -> Self::EnvVars {
+        HashMap::new()
+    }
+
+    fn volumes(&self) -> Self::Volumes {
+        HashMap::new()
+    }
+
+    fn with_args(self, args: Self::Args) -> Self {
+        Self { args }
+    }
+}

libp2p-tor/tor.Dockerfile

@@ -0,0 +1,13 @@
+# set alpine as the base image of the Dockerfile
+FROM alpine:latest
+
+# update the package repository and install Tor
+RUN apk update && apk add tor
+# Set `tor` as the default user during the container runtime
+USER tor
+
+EXPOSE 9050
+EXPOSE 9051
+
+# Set `tor` as the entrypoint for the image
+ENTRYPOINT ["tor"]