zcash_client_backend: Add tor::Client::connect_to_lightwalletd

Closes #1471.

Cargo.toml

@@ -145,6 +145,7 @@ rand_xorshift = "0.3"
 arti-client = { version = "0.11", default-features = false, features = ["compression", "rustls", "tokio"] }
 tokio = "1"
 tor-rtcompat = "0.9"
+tower = "0.4"
 
 # ZIP 32
 aes = "0.8"

zcash_client_backend/Cargo.toml

@@ -122,6 +122,7 @@ rayon.workspace = true
 # - Tor
 tokio = { workspace = true, optional = true, features = ["fs"] }
 tor-rtcompat = { workspace = true, optional = true }
+tower = { workspace = true, optional = true }
 
 # - HTTP through Tor
 http-body-util = { workspace = true, optional = true }
@@ -150,7 +151,7 @@ tokio = { version = "1.21.0", features = ["rt-multi-thread"] }
 
 [features]
 ## Enables the `tonic` gRPC client bindings for connecting to a `lightwalletd` server.
-lightwalletd-tonic = ["dep:tonic"]
+lightwalletd-tonic = ["dep:tonic", "hyper-util?/tokio"]
 
 ## Enables the `transport` feature of `tonic` producing a fully-featured client and server implementation
 lightwalletd-tonic-transport = ["lightwalletd-tonic", "tonic?/transport"]
@@ -188,6 +189,7 @@ tor = [
     "dep:tokio",
     "dep:tokio-rustls",
     "dep:tor-rtcompat",
+    "dep:tower",
     "dep:webpki-roots",
 ]
 

zcash_client_backend/src/tor.rs

@@ -6,9 +6,13 @@ use arti_client::{config::TorClientConfigBuilder, TorClient};
 use tor_rtcompat::PreferredRuntime;
 use tracing::debug;
 
+#[cfg(feature = "lightwalletd-tonic")]
+mod grpc;
+
 pub mod http;
 
 /// A Tor client that exposes capabilities designed for Zcash wallets.
+#[derive(Clone)]
 pub struct Client {
     inner: TorClient<PreferredRuntime>,
 }
@@ -50,6 +54,9 @@ impl Client {
 pub enum Error {
     /// The directory passed to [`Client::create`] does not exist.
     MissingTorDirectory,
+    #[cfg(feature = "lightwalletd-tonic")]
+    /// An error occurred while using gRPC-over-Tor.
+    Grpc(self::grpc::GrpcError),
     /// An error occurred while using HTTP-over-Tor.
     Http(self::http::HttpError),
     /// An IO error occurred while interacting with the filesystem.
@@ -62,6 +69,8 @@ impl fmt::Display for Error {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Error::MissingTorDirectory => write!(f, "Tor directory is missing"),
+            #[cfg(feature = "lightwalletd-tonic")]
+            Error::Grpc(e) => write!(f, "gRPC-over-Tor error: {}", e),
             Error::Http(e) => write!(f, "HTTP-over-Tor error: {}", e),
             Error::Io(e) => write!(f, "IO error: {}", e),
             Error::Tor(e) => write!(f, "Tor error: {}", e),
@@ -73,13 +82,22 @@ impl std::error::Error for Error {
     fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
         match self {
             Error::MissingTorDirectory => None,
+            #[cfg(feature = "lightwalletd-tonic")]
+            Error::Grpc(e) => Some(e),
             Error::Http(e) => Some(e),
             Error::Io(e) => Some(e),
             Error::Tor(e) => Some(e),
         }
     }
 }
 
+#[cfg(feature = "lightwalletd-tonic")]
+impl From<self::grpc::GrpcError> for Error {
+    fn from(e: self::grpc::GrpcError) -> Self {
+        Error::Grpc(e)
+    }
+}
+
 impl From<self::http::HttpError> for Error {
     fn from(e: self::http::HttpError) -> Self {
         Error::Http(e)

zcash_client_backend/src/tor/grpc.rs

@@ -0,0 +1,106 @@
+use std::{
+    fmt,
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use arti_client::DataStream;
+use hyper_util::rt::TokioIo;
+use tonic::transport::{Channel, ClientTlsConfig, Endpoint, Uri};
+use tower::Service;
+use tracing::debug;
+
+use super::{http, Client, Error};
+use crate::proto::service::compact_tx_streamer_client::CompactTxStreamerClient;
+
+impl Client {
+    /// Connects to the `lightwalletd` server at the given endpoint.
+    pub async fn connect_to_lightwalletd(
+        &self,
+        endpoint: Uri,
+    ) -> Result<CompactTxStreamerClient<Channel>, Error> {
+        let is_https = http::url_is_https(&endpoint)?;
+
+        let channel = Endpoint::from(endpoint);
+        let channel = if is_https {
+            channel
+                .tls_config(ClientTlsConfig::new().with_webpki_roots())
+                .map_err(GrpcError::Tonic)?
+        } else {
+            channel
+        };
+
+        let conn = channel
+            .connect_with_connector(self.http_tcp_connector())
+            .await
+            .map_err(GrpcError::Tonic)?;
+
+        Ok(CompactTxStreamerClient::new(conn))
+    }
+
+    fn http_tcp_connector(&self) -> HttpTcpConnector {
+        HttpTcpConnector {
+            client: self.clone(),
+        }
+    }
+}
+
+struct HttpTcpConnector {
+    client: Client,
+}
+
+impl Service<Uri> for HttpTcpConnector {
+    type Response = TokioIo<DataStream>;
+    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
+
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, endpoint: Uri) -> Self::Future {
+        let parsed = http::parse_url(&endpoint);
+        let client = self.client.clone();
+
+        let fut = async move {
+            let (_, host, port) = parsed?;
+
+            debug!("Connecting through Tor to {}:{}", host, port);
+            let stream = client.inner.connect((host.as_str(), port)).await?;
+
+            Ok(TokioIo::new(stream))
+        };
+
+        Box::pin(fut)
+    }
+}
+
+/// Errors that can occurr while using HTTP-over-Tor.
+#[derive(Debug)]
+pub enum GrpcError {
+    /// A [`tonic`] error.
+    Tonic(tonic::transport::Error),
+}
+
+impl fmt::Display for GrpcError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            GrpcError::Tonic(e) => write!(f, "Hyper error: {}", e),
+        }
+    }
+}
+
+impl std::error::Error for GrpcError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            GrpcError::Tonic(e) => Some(e),
+        }
+    }
+}
+
+impl From<tonic::transport::Error> for GrpcError {
+    fn from(e: tonic::transport::Error) -> Self {
+        GrpcError::Tonic(e)
+    }
+}

zcash_client_backend/src/tor/http.rs

@@ -24,6 +24,24 @@ use super::{Client, Error};
 
 pub mod cryptex;
 
+pub(super) fn url_is_https(url: &Uri) -> Result<bool, HttpError> {
+    Ok(url.scheme().ok_or_else(|| HttpError::NonHttpUrl)? == &Scheme::HTTPS)
+}
+
+pub(super) fn parse_url(url: &Uri) -> Result<(bool, String, u16), Error> {
+    let is_https = url_is_https(url)?;
+
+    let host = url.host().ok_or_else(|| HttpError::NonHttpUrl)?.to_string();
+
+    let port = match url.port_u16() {
+        Some(port) => port,
+        None if is_https => 443,
+        None => 80,
+    };
+
+    Ok((is_https, host, port))
+}
+
 impl Client {
     #[tracing::instrument(skip(self, h, f))]
     async fn get<T, F: Future<Output = Result<T, Error>>>(
@@ -32,15 +50,7 @@ impl Client {
         h: impl FnOnce(Builder) -> Builder,
         f: impl FnOnce(Incoming) -> F,
     ) -> Result<Response<T>, Error> {
-        let is_https = url.scheme().ok_or_else(|| HttpError::NonHttpUrl)? == &Scheme::HTTPS;
-
-        let host = url.host().ok_or_else(|| HttpError::NonHttpUrl)?.to_string();
-
-        let port = match url.port_u16() {
-            Some(port) => port,
-            None if is_https => 443,
-            None => 80,
-        };
+        let (is_https, host, port) = parse_url(&url)?;
 
         // Connect to the server.
         debug!("Connecting through Tor to {}:{}", host, port);