net/transport: Add experimental Tor hidden service support

Cargo.lock

@@ -2032,8 +2032,11 @@ dependencies = [
  "thiserror",
  "tinyjson",
  "toml 0.8.13",
+ "tor-cell",
  "tor-error",
  "tor-hscrypto",
+ "tor-hsservice",
+ "tor-proto",
  "tor-rtcompat",
  "url",
  "wasmer",

Cargo.toml

@@ -72,6 +72,9 @@ arti-client = {version = "0.18.0", default-features = false, features = ["async-
 tor-error = {version = "0.18.0", optional = true}
 tor-rtcompat = {version = "0.18.0", features = ["async-std", "rustls"], optional = true}
 tor-hscrypto = {version = "0.18.0", optional = true}
+tor-hsservice = {version = "0.18.0", optional = true}
+tor-proto = {version = "0.18.0", optional = true}
+tor-cell = {version = "0.18.0", optional = true}
 
 # TLS cert utilities
 ed25519-compact = {version = "2.1.1", optional = true}
@@ -216,10 +219,22 @@ event-graph = [
 ]
 
 p2p-unix = []
-p2p-tcp = ["socket2"]
-p2p-tor = ["arti-client", "tor-hscrypto", "tor-error", "tor-rtcompat", "libsqlite3-sys"]
+
 p2p-nym = []
 
+p2p-tcp = ["socket2"]
+
+p2p-tor = [
+    "arti-client",
+    "tor-hsservice",
+    "tor-hscrypto",
+    "tor-error",
+    "tor-rtcompat",
+    "tor-proto",
+    "tor-cell",
+    "libsqlite3-sys",
+]
+
 net = [
     "async-trait",
     "ed25519-compact",

src/net/protocol/protocol_address.rs

@@ -44,19 +44,19 @@ use crate::{Error, Result};
 /// The node selection logic for creating an AddrMessage is as follows:
 ///
 /// 1. First select nodes matching the requested transports from the
-/// anchorlist. These nodes have the highest guarantee of being reachable, so we
-/// prioritize them first.
+///    anchorlist. These nodes have the highest guarantee of being reachable,
+///    so we prioritize them first.
 ///
 /// 2. Then select nodes matching the requested transports from the
-/// whitelist.
+///    whitelist.
 ///
 /// 3. Next select whitelist nodes that don't match our transports. We do
-/// this so that nodes share and propagate nodes of different transports,
-/// even if they can't connect to them themselves.
+///    this so that nodes share and propagate nodes of different transports,
+///    even if they can't connect to them themselves.
 ///
 /// 4. Finally, if there's still space available, fill the remaining vector
-/// space with darklist entries. This is necessary to propagate transports
-/// that neither this node nor the receiving node support.
+///    space with darklist entries. This is necessary to propagate transports
+///    that neither this node nor the receiving node support.
 pub struct ProtocolAddress {
     channel: ChannelPtr,
     addrs_sub: MessageSubscription<AddrsMessage>,

src/net/transport/mod.rs

@@ -88,6 +88,10 @@ pub enum ListenerVariant {
     /// TCP with TLS
     TcpTls(tcp::TcpListener),
 
+    #[cfg(feature = "p2p-tor")]
+    /// Tor
+    Tor(tor::TorListener),
+
     #[cfg(feature = "p2p-unix")]
     /// Unix socket
     Unix(unix::UnixListener),
@@ -305,6 +309,15 @@ impl Listener {
                 Ok(Self { endpoint, variant })
             }
 
+            #[cfg(feature = "p2p-tor")]
+            "tor" => {
+                // Build a Tor Hidden Service listener
+                enforce_hostport!(endpoint);
+                let variant = tor::TorListener::new().await?;
+                let variant = ListenerVariant::Tor(variant);
+                Ok(Self { endpoint, variant })
+            }
+
             #[cfg(feature = "p2p-unix")]
             "unix" => {
                 enforce_abspath!(endpoint);
@@ -340,6 +353,13 @@ impl Listener {
                 Ok(Box::new(l))
             }
 
+            #[cfg(feature = "p2p-tor")]
+            ListenerVariant::Tor(listener) => {
+                let port = self.endpoint.port().unwrap();
+                let l = listener.do_listen(port).await?;
+                Ok(Box::new(l))
+            }
+
             #[cfg(feature = "p2p-unix")]
             ListenerVariant::Unix(listener) => {
                 let path = match self.endpoint.to_file_path() {
@@ -380,6 +400,6 @@ impl PtStream for smol::net::unix::UnixStream {}
 
 /// Wrapper trait for async listeners
 #[async_trait]
-pub trait PtListener: Send + Sync + Unpin {
+pub trait PtListener: Send + Unpin {
     async fn next(&self) -> io::Result<(Box<dyn PtStream>, Url)>;
 }

src/net/transport/tor.rs

@@ -18,18 +18,35 @@
 
 use std::{
     io::{self, ErrorKind},
+    pin::Pin,
+    sync::Arc,
     time::Duration,
 };
 
-use arti_client::{config::BoolOrAuto, DataStream, StreamPrefs, TorClient};
+use arti_client::{
+    config::{onion_service::OnionServiceConfigBuilder, BoolOrAuto},
+    DataStream, StreamPrefs, TorClient,
+};
+use async_trait::async_trait;
 use futures::{
     future::{select, Either},
     pin_mut,
+    stream::StreamExt,
+    Stream,
+};
+use log::{debug, error, info, warn};
+use smol::{
+    lock::{Mutex, OnceCell},
+    Timer,
 };
-use log::{debug, warn};
-use smol::{lock::OnceCell, Timer};
+use tor_cell::relaycell::msg::Connected;
 use tor_error::ErrorReport;
+use tor_hsservice::{HsNickname, RendRequest, RunningOnionService};
+use tor_proto::stream::IncomingStreamRequest;
 use tor_rtcompat::PreferredRuntime;
+use url::Url;
+
+use super::{PtListener, PtStream};
 
 /// A static for `TorClient` reusability
 static TOR_CLIENT: OnceCell<TorClient<PreferredRuntime>> = OnceCell::new();
@@ -119,3 +136,139 @@ impl TorDialer {
         }
     }
 }
+
+/// Tor Listener implementation
+#[derive(Clone, Debug)]
+pub struct TorListener;
+
+impl TorListener {
+    /// Instantiate a new [`TorListener`]
+    pub async fn new() -> io::Result<Self> {
+        Ok(Self {})
+    }
+
+    /// Internal listen function
+    pub(crate) async fn do_listen(&self, port: u16) -> io::Result<TorListenerIntern> {
+        // Initialize or fetch the static TOR_CLIENT that should be reused in
+        // the Tor dialer
+        let client = match TOR_CLIENT
+            .get_or_try_init(|| async {
+                debug!(target: "net::tor::do_dial", "Bootstrapping...");
+                TorClient::builder().create_bootstrapped().await
+            })
+            .await
+        {
+            Ok(client) => client,
+            Err(e) => {
+                warn!("{}", e.report());
+                return Err(io::Error::new(
+                    ErrorKind::Other,
+                    "Internal Tor error, see logged warning",
+                ))
+            }
+        };
+
+        let hs_nick = HsNickname::new("darkfi_tor".to_string()).unwrap();
+
+        let hs_config = match OnionServiceConfigBuilder::default().nickname(hs_nick).build() {
+            Ok(v) => v,
+            Err(e) => {
+                error!(
+                    target: "net::tor::do_listen",
+                    "[P2P] Failed to create OnionServiceConfig: {}", e,
+                );
+                return Err(io::Error::new(ErrorKind::Other, "Internal Tor error"))
+            }
+        };
+
+        let (onion_service, rendreq_stream) = match client.launch_onion_service(hs_config) {
+            Ok(v) => v,
+            Err(e) => {
+                error!(
+                    target: "net::tor::do_listen",
+                    "[P2P] Failed to launch Onion Service: {}", e,
+                );
+                return Err(io::Error::new(ErrorKind::Other, "Internal Tor error"))
+            }
+        };
+
+        info!(
+            target: "net::tor::do_listen",
+            "[P2P] Established Tor listener on tor://{}:{}",
+            onion_service.onion_name().unwrap(), port,
+        );
+
+        Ok(TorListenerIntern {
+            port,
+            _onion_service: onion_service,
+            rendreq_stream: Mutex::new(Box::pin(rendreq_stream)),
+        })
+    }
+}
+
+/*
+/// Internal Tor Listener implementation, used with `PtListener`
+pub struct TorListenerIntern<'a> {
+    port: u16,
+    _onion_service: Arc<RunningOnionService>,
+    rendreq_stream: Mutex<BoxStream<'a, RendRequest>>,
+}
+
+unsafe impl Sync for TorListenerIntern<'_> {}
+*/
+
+pub struct TorListenerIntern {
+    port: u16,
+    _onion_service: Arc<RunningOnionService>,
+    //rendreq_stream: Mutex<BoxStream<'a, RendRequest>>,
+    rendreq_stream: Mutex<Pin<Box<dyn Stream<Item = RendRequest> + Send>>>,
+}
+
+unsafe impl Sync for TorListenerIntern {}
+
+#[async_trait]
+impl PtListener for TorListenerIntern {
+    async fn next(&self) -> io::Result<(Box<dyn PtStream>, Url)> {
+        let mut rendreq_stream = self.rendreq_stream.lock().await;
+
+        let Some(rendrequest) = rendreq_stream.next().await else {
+            todo!();
+        };
+
+        let mut streamreq_stream = match rendrequest.accept().await {
+            Ok(v) => v,
+            Err(e) => {
+                error!(
+                    target: "net::tor::PtListener::next",
+                    "[P2P] Failed accepting Tor RendRequest: {}", e,
+                );
+                return Err(io::Error::new(ErrorKind::Other, "Internal Tor error"))
+            }
+        };
+
+        let Some(streamrequest) = streamreq_stream.next().await else { todo!() };
+
+        // Validate port correctness
+        match streamrequest.request() {
+            IncomingStreamRequest::Begin(begin) => {
+                if begin.port() != self.port {
+                    return Err(io::Error::new(ErrorKind::ConnectionAborted, "Connection Aborted"))
+                }
+            }
+            &_ => return Err(io::Error::new(ErrorKind::ConnectionAborted, "Connection Aborted")),
+        }
+
+        let stream = match streamrequest.accept(Connected::new_empty()).await {
+            Ok(v) => v,
+            Err(e) => {
+                error!(
+                    target: "net::tor::PtListener::next",
+                    "[P2P] Failed accepting Tor StreamRequest: {}", e,
+                );
+                return Err(io::Error::new(ErrorKind::Other, "Internal Tor error"))
+            }
+        };
+
+        Ok((Box::new(stream), Url::parse(&format!("tor://127.0.0.1:{}", self.port)).unwrap()))
+    }
+}

src/validator/consensus.rs

@@ -221,6 +221,7 @@ impl Consensus {
     /// - If the current best fork has reached greater length than the security threshold,
     ///   and no other fork exist with same rank, first proposal(s) in that fork can be
     ///   appended to canonical blockchain (finalize).
+    ///
     /// When best fork can be finalized, first block(s) should be appended to canonical,
     /// and forks should be rebuilt.
     pub async fn finalization(&self) -> Result<Option<usize>> {