fix(comms): correctly initialize hidden service (#6124)

Description
---
Correctly sets up a tor hidden service in the HiddenServiceTransport
Tor control client will not error if a value in a key-value query is
empty, as this can be valid
Disables signal handlers for libtor

Motivation and Context
---

This PR correctly initializes a tor hidden service in the
HiddenServiceTransport. A task is spawned by `create_hidden_service`
that monitors the tor control port connection and automatically tries to
reestablish it if it disconnects. I also fix the incorrect setting of
the proxied address to be the listening port of the Tari node rather
than tor's SOCKS port.

The minor change to the control port client is minor and allows for an
empty value to be returned, which is valid when querying key-value
pairs. In practice, this never happens in our current usage but I
encountered it when debugging and it prevented the real problem from
coming through.

Ref PR #6092 

There is an additional existing problem where libtor handles interrupt
signals and exits e.g. when pressing ctrl+c in the base node to type a
command. I fixed this by [disabling signal
handlers](https://github.com/torproject/torspec/blob/8961bb4d83fccb2b987f9899ca83aa430f84ab0c/control-spec.txt#L3946)
in libtor.

How Has This Been Tested?
---
Manually

What process can a PR reviewer use to test or verify this change?
---

<!-- Checklist -->
<!-- 1. Is the title of your PR in the form that would make nice release
notes? The title, excluding the conventional commit
tag, will be included exactly as is in the CHANGELOG, so please think
about it carefully. -->
Nodes should receive inbound and outbound tor connections.

Breaking Changes
---

- [x] None
- [ ] Requires data directory on base node to be deleted
- [ ] Requires hard fork
- [ ] Other - Please specify

<!-- Does this include a breaking change? If so, include this line as a
footer -->
<!-- BREAKING CHANGE: Description what the user should do, e.g. delete a
database, resync the chain -->

comms/core/src/tor/control_client/client.rs

@@ -125,9 +125,6 @@ impl TorControlPortClient {
     pub async fn get_info(&mut self, key_name: &'static str) -> Result<Vec<Cow<'_, str>>, TorClientError> {
         let command = commands::get_info(key_name);
         let response = self.request_response(command).await?;
-        if response.is_empty() {
-            return Err(TorClientError::ServerNoResponse);
-        }
         Ok(response)
     }
 
@@ -202,7 +199,6 @@ impl TorControlPortClient {
         let cmd_str = command.to_command_string().map_err(Into::into)?;
         self.send_line(cmd_str).await?;
         let responses = self.recv_next_responses().await?;
-        trace!(target: LOG_TARGET, "Response from tor: {:?}", responses);
         if responses.is_empty() {
             return Err(TorClientError::ServerNoResponse);
         }

comms/core/src/tor/control_client/monitor.rs

@@ -53,7 +53,7 @@ where
             match either {
                 // Received a command to send to the control server
                 Either::Left(Some(line)) => {
-                    trace!(target: LOG_TARGET, "Writing command of length '{}'", line.len());
+                    trace!(target: LOG_TARGET, "Tor send: {}", line);
                     if let Err(err) = sink.send(line).await {
                         error!(
                             target: LOG_TARGET,
@@ -64,7 +64,7 @@ where
                 },
                 // Command stream ended
                 Either::Left(None) => {
-                    debug!(
+                    warn!(
                         target: LOG_TARGET,
                         "Tor control server command receiver closed. Monitor is exiting."
                     );
@@ -73,7 +73,7 @@ where
 
                 // Received a line from the control server
                 Either::Right(Some(Ok(line))) => {
-                    trace!(target: LOG_TARGET, "Read line of length '{}'", line.len());
+                    trace!(target: LOG_TARGET, "Tor recv: {}", line);
                     match parsers::response_line(&line) {
                         Ok(mut line) => {
                             if line.is_multiline {
@@ -116,7 +116,7 @@ where
                 // The control server disconnected
                 Either::Right(None) => {
                     cmd_rx.close();
-                    debug!(
+                    warn!(
                         target: LOG_TARGET,
                         "Connection to tor control port closed. Monitor is exiting."
                     );

comms/core/src/tor/hidden_service/controller.rs

@@ -24,7 +24,6 @@ use std::{fs, io, net::SocketAddr, sync::Arc, time::Duration};
 
 use futures::{future, future::Either, pin_mut, StreamExt};
 use log::*;
-use multiaddr::{multiaddr, Protocol};
 use tari_shutdown::OptionalShutdownSignal;
 use tari_utilities::hex::Hex;
 use thiserror::Error;
@@ -127,16 +126,6 @@ impl HiddenServiceController {
     pub async fn initialize_transport(&mut self) -> Result<SocksTransport, HiddenServiceControllerError> {
         self.connect_and_auth().await?;
 
-        let socks_addr = self.get_socks_address().await?;
-        let mut proxied_addr = self.proxied_address();
-        if proxied_addr.ends_with(&multiaddr!(Tcp(0u16))) {
-            if let Some(Protocol::Tcp(port)) = socks_addr.iter().last() {
-                proxied_addr.pop();
-                proxied_addr.push(Protocol::Tcp(port));
-            }
-            self.set_proxied_addr(&proxied_addr);
-        }
-        self.create_hidden_service_from_identity().await?;
         let socks_addr = self.get_socks_address().await?;
         Ok(SocksTransport::new(SocksConfig {
             proxy_address: socks_addr,

comms/core/src/transports/hidden_service_transport.rs

@@ -23,12 +23,12 @@
 use std::{io, io::ErrorKind, sync::Arc};
 
 use log::*;
-use multiaddr::Multiaddr;
+use multiaddr::{multiaddr, Multiaddr, Protocol};
 use tokio::sync::RwLock;
 
 use crate::{
     tor::{HiddenServiceController, TorIdentity},
-    transports::{SocksTransport, Transport},
+    transports::{tcp::TcpInbound, SocksTransport, Transport},
 };
 
 const LOG_TARGET: &str = "comms::transports::hidden_service_transport";
@@ -43,7 +43,7 @@ pub enum HiddenServiceTransportError {
 
 struct HiddenServiceTransportInner {
     socks_transport: Option<SocksTransport>,
-    hidden_service_ctl: HiddenServiceController,
+    hidden_service_ctl: Option<HiddenServiceController>,
 }
 
 #[derive(Clone)]
@@ -57,37 +57,55 @@ impl<F: Fn(TorIdentity)> HiddenServiceTransport<F> {
         Self {
             inner: Arc::new(RwLock::new(HiddenServiceTransportInner {
                 socks_transport: None,
-                hidden_service_ctl,
+                hidden_service_ctl: Some(hidden_service_ctl),
             })),
             after_init,
         }
     }
 
-    async fn ensure_initialized(&self) -> Result<(), io::Error> {
-        let inner = self.inner.read().await;
-        if inner.socks_transport.is_none() {
-            drop(inner);
-            let mut mut_inner = self.inner.write().await;
-            if mut_inner.socks_transport.is_none() {
-                let transport = mut_inner.hidden_service_ctl.initialize_transport().await.map_err(|e| {
-                    error!(
-                        target: LOG_TARGET,
-                        "Error initializing hidden transport service stack{}",
-                        e
-                    );
-                    io::Error::new(ErrorKind::Other, e.to_string())
-                })?;
-                (self.after_init)(
-                    mut_inner
-                        .hidden_service_ctl
-                        .identity
-                        .clone()
-                        .ok_or(io::Error::new(ErrorKind::Other, "Missing tor identity".to_string()))?,
-                );
-                mut_inner.socks_transport = Some(transport);
+    async fn is_initialized(&self) -> bool {
+        self.inner.read().await.socks_transport.is_some()
+    }
+
+    async fn initialize(&self, listen_addr: &Multiaddr) -> Result<(TcpInbound, Multiaddr), io::Error> {
+        let mut inner_mut = self.inner.write().await;
+        let mut hs_ctl = inner_mut.hidden_service_ctl.take().ok_or(io::Error::new(
+            ErrorKind::Other,
+            "BUG: Hidden service controller not set in transport".to_string(),
+        ))?;
+
+        let transport = hs_ctl.initialize_transport().await.map_err(|e| {
+            error!(
+                target: LOG_TARGET,
+                "Error initializing hidden transport service stack{}",
+                e
+            );
+            io::Error::new(ErrorKind::Other, e.to_string())
+        })?;
+        let (inbound, listen_addr) = transport.listen(listen_addr).await?;
+        inner_mut.socks_transport = Some(transport);
+
+        // Set the proxied address to the port we just listened on
+        let mut proxied_addr = hs_ctl.proxied_address();
+        if proxied_addr.ends_with(&multiaddr!(Tcp(0u16))) {
+            if let Some(Protocol::Tcp(port)) = listen_addr.iter().last() {
+                proxied_addr.pop();
+                proxied_addr.push(Protocol::Tcp(port));
             }
+            hs_ctl.set_proxied_addr(&proxied_addr);
         }
-        Ok(())
+
+        let hidden_service = hs_ctl.create_hidden_service().await.map_err(|err| {
+            error!(
+                target: LOG_TARGET,
+                "Error creating hidden service: {}",
+                err
+            );
+            io::Error::new(ErrorKind::Other, err.to_string())
+        })?;
+
+        (self.after_init)(hidden_service.tor_identity().clone());
+        Ok((inbound, listen_addr))
     }
 }
 #[crate::async_trait]
@@ -97,15 +115,27 @@ impl<F: Fn(TorIdentity) + Send + Sync> Transport for HiddenServiceTransport<F> {
     type Output = <SocksTransport as Transport>::Output;
 
     async fn listen(&self, addr: &Multiaddr) -> Result<(Self::Listener, Multiaddr), Self::Error> {
-        self.ensure_initialized().await?;
-        let inner = self.inner.read().await;
-
-        Ok(inner.socks_transport.as_ref().unwrap().listen(addr).await?)
+        if self.is_initialized().await {
+            // For now, we only can listen on a single Tor hidden service. This behaviour is not technically correct as
+            // per the Transport trait, but we only ever call listen once in practice. The fix for this is to
+            // improve the tor client implementation to allow for multiple hidden services.
+            return Err(io::Error::new(
+                ErrorKind::Other,
+                "BUG: Hidden service transport already initialized".to_string(),
+            ));
+        }
+        let (listener, addr) = self.initialize(addr).await?;
+        Ok((listener, addr))
     }
 
     async fn dial(&self, addr: &Multiaddr) -> Result<Self::Output, Self::Error> {
-        self.ensure_initialized().await?;
         let inner = self.inner.read().await;
-        Ok(inner.socks_transport.as_ref().unwrap().dial(addr).await?)
+        let transport = inner.socks_transport.as_ref().ok_or_else(|| {
+            io::Error::new(
+                ErrorKind::Other,
+                "BUG: Hidden service transport not initialized before dialling".to_string(),
+            )
+        })?;
+        transport.dial(addr).await
     }
 }

infrastructure/libtor/Cargo.toml

@@ -7,7 +7,7 @@ license = "BSD-3-Clause"
 [dependencies]
 tari_common = { path = "../../common" }
 tari_p2p = { path = "../../base_layer/p2p" }
-tari_shutdown = {  path = "../shutdown"}
+tari_shutdown = { path = "../shutdown" }
 
 derivative = "2.2.0"
 log = "0.4.8"
@@ -16,7 +16,7 @@ tempfile = "3.1.0"
 tor-hash-passwd = "1.0.1"
 
 [target.'cfg(unix)'.dependencies]
-libtor = { version="46.9.0"}
+libtor = { version = "46.9.0" }
 openssl = { version = "0.10.61", features = ["vendored"] }
 
 [package.metadata.cargo-machete]

infrastructure/libtor/src/tor.rs

@@ -147,6 +147,9 @@ impl Tor {
         tor.flag(TorFlag::DataDirectory(data_dir.clone()))
             .flag(TorFlag::SocksPort(socks_port))
             .flag(TorFlag::ControlPort(control_port))
+            // Disable signal handlers so that ctrl+c can be handled by our application
+            // https://github.com/torproject/torspec/blob/8961bb4d83fccb2b987f9899ca83aa430f84ab0c/control-spec.txt#L3946
+            .flag(TorFlag::Custom("__DisableSignalHandlers 1".to_string()))
             .flag(TorFlag::Hush())
             .flag(TorFlag::LogTo(log_level, LogDestination::File(log_destination)));