T2. Add isolated Tor connection API, but don't enable it by default (#…

…3303)

* Add arti as a zebra-network dependency

* Add a method for isolated anonymised Tor connections to a specific hostname

* Add tests for isolated tor connections

* Use a shared tor client instance for all isolated connections

* Silence a spurious tor warning in tests

* Make tor support optional, activate it via a new "tor" feature

* Extra Cargo.lock changes

* fastmod AsyncReadWrite PeerTransport zebra*

* Remove unnecessary PeerTransport generics

* Refactor common test code into a function

* Don't drop the stream until the end of the test

Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

deny.toml

@@ -68,6 +68,9 @@ skip-tree = [
 
     # wait for lots of crates in the tokio ecosystem to upgrade
     { name = "socket2", version = "=0.3.16" },
+
+    # wait for arti to stabilise
+    { name = "arti-client" },
 ]
 
 # This section is considered when running `cargo deny check sources`.

zebra-network/Cargo.toml

@@ -7,6 +7,10 @@ edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
+[features]
+default = []
+tor = ["arti-client", "tor-rtcompat"]
+
 [dependencies]
 bitflags = "1.2"
 byteorder = "1.4"
@@ -32,6 +36,10 @@ tracing = "0.1"
 tracing-futures = "0.2"
 tracing-error = { version = "0.1.2", features = ["traced-error"] }
 
+# tor dependencies
+arti-client = { version = "0.0.2", optional = true }
+tor-rtcompat  = { version = "0.0.2", optional = true }
+
 zebra-chain = { path = "../zebra-chain" }
 
 [dev-dependencies]

zebra-network/src/config.rs

@@ -101,14 +101,19 @@ impl Config {
         self.peerset_outbound_connection_limit() + self.peerset_inbound_connection_limit()
     }
 
-    /// Get the initial seed peers based on the configured network.
-    pub async fn initial_peers(&self) -> HashSet<SocketAddr> {
+    /// Returns the initial seed peer hostnames for the configured network.
+    pub fn initial_peer_hostnames(&self) -> &HashSet<String> {
         match self.network {
-            Network::Mainnet => Config::resolve_peers(&self.initial_mainnet_peers).await,
-            Network::Testnet => Config::resolve_peers(&self.initial_testnet_peers).await,
+            Network::Mainnet => &self.initial_mainnet_peers,
+            Network::Testnet => &self.initial_testnet_peers,
         }
     }
 
+    /// Resolve initial seed peer IP addresses, based on the configured network.
+    pub async fn initial_peers(&self) -> HashSet<SocketAddr> {
+        Config::resolve_peers(self.initial_peer_hostnames()).await
+    }
+
     /// Concurrently resolves `peers` into zero or more IP addresses, with a
     /// timeout of a few seconds on each DNS request.
     ///

zebra-network/src/isolated.rs

@@ -17,6 +17,9 @@ use crate::{
     BoxError, Config, Request, Response,
 };
 
+#[cfg(feature = "tor")]
+pub(crate) mod tor;
+
 #[cfg(test)]
 mod tests;
 
@@ -44,13 +47,13 @@ mod tests;
 ///                  or a Tor client [`DataStream`].
 ///
 /// - `user_agent`: a valid BIP14 user-agent, e.g., the empty string.
-pub fn connect_isolated<AsyncReadWrite>(
+pub fn connect_isolated<PeerTransport>(
     network: Network,
-    data_stream: AsyncReadWrite,
+    data_stream: PeerTransport,
     user_agent: String,
 ) -> impl Future<Output = Result<BoxService<Request, Response, BoxError>, BoxError>>
 where
-    AsyncReadWrite: AsyncRead + AsyncWrite + Unpin + Send + 'static,
+    PeerTransport: AsyncRead + AsyncWrite + Unpin + Send + 'static,
 {
     let config = Config {
         network,
@@ -91,6 +94,8 @@ where
 ///
 /// Transactions sent over this connection can be linked to the sending and receiving IP address
 /// by passive internet observers.
+///
+/// Prefer [`connect_isolated_run_tor`](tor::connect_isolated_run_tor) if available.
 pub fn connect_isolated_tcp_direct(
     network: Network,
     addr: SocketAddr,

zebra-network/src/isolated/tests/vectors.rs

@@ -49,58 +49,7 @@ async fn connect_isolated_sends_anonymised_version_message_tcp_net(network: Netw
     let mut inbound_stream =
         Framed::new(inbound_conn, Codec::builder().for_network(network).finish());
 
-    // We don't need to send any bytes to get a version message.
-    if let Message::Version {
-        version,
-        services,
-        timestamp,
-        address_recv,
-        address_from,
-        nonce: _,
-        user_agent,
-        start_height,
-        relay,
-    } = inbound_stream
-        .next()
-        .await
-        .expect("stream item")
-        .expect("item is Ok(msg)")
-    {
-        // Check that the version message sent by connect_isolated
-        // anonymises all the fields that it possibly can.
-        //
-        // The version field needs to be accurate, because it controls protocol features.
-        // The nonce must be randomised for security.
-        //
-        // SECURITY TODO: check if the timestamp field can be zeroed, to remove another distinguisher (#3300)
-
-        let mut fixed_isolated_addr: SocketAddr = "0.0.0.0:0".parse().unwrap();
-        fixed_isolated_addr.set_port(network.default_port());
-
-        // Required fields should be accurate and match most other peers.
-        // (We can't test nonce randomness here.)
-        assert_eq!(version, CURRENT_NETWORK_PROTOCOL_VERSION);
-        assert_eq!(timestamp.timestamp() % (5 * 60), 0);
-
-        // Other fields should be empty or zeroed.
-        assert_eq!(services, PeerServices::empty());
-        assert_eq!(
-            address_recv,
-            // Since we're connecting to the peer, we expect it to have the node flag.
-            //
-            // SECURITY TODO: should this just be zeroed anyway? (#3300)
-            AddrInVersion::new(fixed_isolated_addr, PeerServices::NODE_NETWORK),
-        );
-        assert_eq!(
-            address_from,
-            AddrInVersion::new(fixed_isolated_addr, PeerServices::empty()),
-        );
-        assert_eq!(user_agent, "");
-        assert_eq!(start_height.0, 0);
-        assert!(!relay);
-    } else {
-        panic!("handshake did not send version message");
-    }
+    check_version_message(network, &mut inbound_stream).await;
 
     // Let the spawned task run for a short time.
     tokio::time::sleep(Duration::from_secs(1)).await;
@@ -125,7 +74,7 @@ async fn connect_isolated_sends_anonymised_version_message_tcp_net(network: Netw
 /// when sent in-memory.
 ///
 /// This test also:
-/// - checks `AsyncReadWrite` support, and
+/// - checks `PeerTransport` support, and
 /// - runs even if network tests are disabled.
 #[tokio::test]
 async fn connect_isolated_sends_anonymised_version_message_mem() {
@@ -147,6 +96,36 @@ async fn connect_isolated_sends_anonymised_version_message_mem_net(network: Netw
         Codec::builder().for_network(network).finish(),
     );
 
+    check_version_message(network, &mut inbound_stream).await;
+
+    // Let the spawned task run for a short time.
+    tokio::time::sleep(Duration::from_secs(1)).await;
+
+    // Make sure that the isolated connection did not:
+    // - panic, or
+    // - return a service.
+    //
+    // This test doesn't send a version message on `inbound_conn`,
+    // so providing a service is incorrect behaviour.
+    // (But a timeout error would be acceptable.)
+    let outbound_result = futures::poll!(&mut outbound_join_handle);
+    assert!(matches!(
+        outbound_result,
+        Poll::Pending | Poll::Ready(Ok(Err(_)))
+    ));
+
+    outbound_join_handle.abort();
+}
+
+/// Wait to receive a version message on `inbound_stream`,
+/// then check that it is correctly anonymised.
+#[track_caller]
+async fn check_version_message<PeerTransport>(
+    network: Network,
+    inbound_stream: &mut Framed<PeerTransport, Codec>,
+) where
+    PeerTransport: AsyncRead + Unpin,
+{
     // We don't need to send any bytes to get a version message.
     if let Message::Version {
         version,
@@ -199,22 +178,4 @@ async fn connect_isolated_sends_anonymised_version_message_mem_net(network: Netw
     } else {
         panic!("handshake did not send version message");
     }
-
-    // Let the spawned task run for a short time.
-    tokio::time::sleep(Duration::from_secs(1)).await;
-
-    // Make sure that the isolated connection did not:
-    // - panic, or
-    // - return a service.
-    //
-    // This test doesn't send a version message on `inbound_conn`,
-    // so providing a service is incorrect behaviour.
-    // (But a timeout error would be acceptable.)
-    let outbound_result = futures::poll!(&mut outbound_join_handle);
-    assert!(matches!(
-        outbound_result,
-        Poll::Pending | Poll::Ready(Ok(Err(_)))
-    ));
-
-    outbound_join_handle.abort();
 }

zebra-network/src/isolated/tor.rs

@@ -0,0 +1,91 @@
+//! Uses tor to create isolated and anonymised connections to specific peers.
+
+use std::sync::{Arc, Mutex};
+
+use arti_client::{TorAddr, TorClient, TorClientConfig};
+use tor_rtcompat::tokio::TokioRuntimeHandle;
+use tower::util::BoxService;
+
+use zebra_chain::parameters::Network;
+
+use crate::{connect_isolated, BoxError, Request, Response};
+
+#[cfg(test)]
+mod tests;
+
+lazy_static::lazy_static! {
+    /// The shared isolated [`TorClient`] instance.
+    ///
+    /// TODO: turn this into a tower service that takes a hostname, and returns an `arti_client::DataStream`
+    ///       (or a task that updates a watch channel when it's done?)
+    pub static ref SHARED_TOR_CLIENT: Arc<Mutex<Option<TorClient<TokioRuntimeHandle>>>> =
+        Arc::new(Mutex::new(None));
+}
+
+/// Creates a Zcash peer connection to `hostname` via Tor.
+/// This connection is completely isolated from all other node state.
+///
+/// See [`connect_isolated`] for details.
+///
+/// # Privacy
+///
+/// The sender IP address is anonymised using Tor.
+/// But transactions sent over this connection can still be linked to the receiving IP address
+/// by passive internet observers.
+/// This happens because the Zcash network protocol uses unencrypted TCP connections.
+///
+/// `hostname` should be a DNS name for the Tor exit to look up, or a hard-coded IP address.
+/// If the application does a local DNS lookup on a hostname, and passes the IP address to this function,
+/// passive internet observers can link the hostname to the sender's IP address.
+///
+/// For details, see
+/// [`TorAddr`](https://tpo.pages.torproject.net/core/doc/rust/arti_client/struct.TorAddr.html).
+pub async fn connect_isolated_tor(
+    network: Network,
+    hostname: String,
+    user_agent: String,
+) -> Result<BoxService<Request, Response, BoxError>, BoxError> {
+    let addr = TorAddr::from(hostname)?;
+
+    // Initialize or clone the shared tor client instance
+    let tor_client = match cloned_tor_client() {
+        Some(tor_client) => tor_client,
+        None => new_tor_client().await?,
+    };
+
+    let tor_stream = tor_client.connect(addr, None).await?;
+
+    connect_isolated(network, tor_stream, user_agent).await
+}
+
+/// Returns a new tor client instance, and updates [`SHARED_TOR_CLIENT`].
+///
+/// If there is a bootstrap error, [`SHARED_TOR_CLIENT`] is not modified.
+async fn new_tor_client() -> Result<TorClient<TokioRuntimeHandle>, BoxError> {
+    let runtime = tokio::runtime::Handle::current();
+    let runtime = TokioRuntimeHandle::new(runtime);
+    let tor_client = TorClient::bootstrap(runtime, TorClientConfig::default()).await?;
+
+    // # Correctness
+    //
+    // It is ok for multiple tasks to race, because all tor clients have identical configs.
+    // And all connections are isolated, regardless of whether they use a new or cloned client.
+    // (Any replaced clients will be dropped.)
+    let mut shared_tor_client = SHARED_TOR_CLIENT
+        .lock()
+        .expect("panic in shared tor client mutex guard");
+    *shared_tor_client = Some(tor_client.isolated_client());
+
+    Ok(tor_client)
+}
+
+/// Returns an isolated tor client instance by cloning [`SHARED_TOR_CLIENT`].
+///
+/// If [`new_tor_client`] has not run successfully yet, returns `None`.
+fn cloned_tor_client() -> Option<TorClient<TokioRuntimeHandle>> {
+    SHARED_TOR_CLIENT
+        .lock()
+        .expect("panic in shared tor client mutex guard")
+        .as_ref()
+        .map(TorClient::isolated_client)
+}

zebra-network/src/isolated/tor/tests.rs

@@ -0,0 +1,3 @@
+//! Tests for isolated Tor connections.
+
+mod vectors;