zcash_client_backend/tor/
grpc.rs

1use std::{
2    error::Error as _,
3    fmt,
4    future::Future,
5    pin::Pin,
6    task::{Context, Poll},
7};
8
9use arti_client::{DataStream, StreamPrefs, config::BoolOrAuto};
10use hyper_util::rt::TokioIo;
11use tonic::transport::{Channel, ClientTlsConfig, Endpoint, Uri};
12use tower::Service;
13use tracing::debug;
14
15use super::{Client, Error, http};
16use crate::proto::service::compact_tx_streamer_client::CompactTxStreamerClient;
17
18impl Client {
19    /// Connects to the `lightwalletd` server at the given endpoint.
20    ///
21    /// If `allow_onion_services` is `true`, the connection will be permitted to reach
22    /// Tor hidden services (`.onion` addresses). The caller is responsible for deciding
23    /// whether onion connections are appropriate for the given endpoint; this crate
24    /// does not infer that from the endpoint host.
25    ///
26    /// The returned client applies this `Client`'s [`Timeouts`] to every request it makes.
27    /// Note that the request deadline bounds the wait for a response's headers, not the
28    /// duration of the response itself, so long-running streaming methods such as
29    /// `GetBlockRange` are not capped by it; a peer that stalls partway through a stream
30    /// is instead detected by the HTTP/2 keep-alive.
31    ///
32    /// [`Timeouts`]: super::Timeouts
33    pub async fn connect_to_lightwalletd(
34        &self,
35        endpoint: Uri,
36        allow_onion_services: bool,
37    ) -> Result<CompactTxStreamerClient<Channel>, Error> {
38        self.ensure_bootstrapped().await?;
39
40        let is_https = http::url_is_https(&endpoint)?;
41
42        let connector = if allow_onion_services {
43            HttpTcpConnector::with_onion_services(self.clone())
44        } else {
45            HttpTcpConnector::new(self.clone())
46        };
47
48        let channel = Endpoint::from(endpoint)
49            .connect_timeout(self.timeouts.connect)
50            .timeout(self.timeouts.request)
51            .http2_keep_alive_interval(self.timeouts.grpc_keepalive_interval)
52            .keep_alive_timeout(self.timeouts.grpc_keepalive_timeout)
53            .keep_alive_while_idle(false);
54        let channel = if is_https {
55            channel
56                .tls_config(ClientTlsConfig::new().with_webpki_roots())
57                .map_err(GrpcError::Tonic)?
58        } else {
59            channel
60        };
61
62        let conn = channel
63            .connect_with_connector(connector)
64            .await
65            .map_err(GrpcError::Tonic)?;
66
67        Ok(CompactTxStreamerClient::new(conn))
68    }
69}
70
71struct HttpTcpConnector {
72    client: Client,
73    prefs: StreamPrefs,
74}
75
76impl HttpTcpConnector {
77    /// Creates a new `HttpTcpConnector` with default [`StreamPrefs`].
78    ///
79    /// Connections made through this connector will not attempt to connect to `.onion`
80    /// services.
81    fn new(client: Client) -> Self {
82        HttpTcpConnector {
83            client,
84            prefs: StreamPrefs::new(),
85        }
86    }
87
88    /// Creates a new `HttpTcpConnector` that enables connections to `.onion` services.
89    ///
90    /// Use this constructor when the endpoint host is a Tor hidden service (`.onion`
91    /// address). For regular clearnet endpoints, use [`HttpTcpConnector::new`] instead.
92    fn with_onion_services(client: Client) -> Self {
93        let mut prefs = StreamPrefs::new();
94        prefs.connect_to_onion_services(BoolOrAuto::Explicit(true));
95        HttpTcpConnector { client, prefs }
96    }
97}
98
99impl Service<Uri> for HttpTcpConnector {
100    type Response = TokioIo<DataStream>;
101    type Error = Error;
102    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
103
104    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
105        Poll::Ready(Ok(()))
106    }
107
108    fn call(&mut self, endpoint: Uri) -> Self::Future {
109        let parsed = http::parse_url(&endpoint);
110        let client = self.client.clone();
111        let prefs = self.prefs.clone();
112
113        let fut = async move {
114            let (_, host, port) = parsed?;
115
116            debug!("Connecting through Tor to {}:{}", host, port);
117            let stream = client
118                .inner
119                .connect_with_prefs((host.as_str(), port), &prefs)
120                .await?;
121
122            Ok(TokioIo::new(stream))
123        };
124
125        Box::pin(fut)
126    }
127}
128
129/// Errors that can occurr while using HTTP-over-Tor.
130#[derive(Debug)]
131#[non_exhaustive]
132pub enum GrpcError {
133    /// A [`tonic`] error.
134    Tonic(tonic::transport::Error),
135}
136
137impl fmt::Display for GrpcError {
138    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
139        match self {
140            GrpcError::Tonic(e) => {
141                if let Some(source) = e.source() {
142                    // Tonic doesn't include the source error in its `Display` impl;
143                    // add it manually for the benefit of our downstreams.
144                    write!(f, "Tonic error: {e}: {source}")
145                } else {
146                    write!(f, "Tonic error: {e}")
147                }
148            }
149        }
150    }
151}
152
153impl std::error::Error for GrpcError {
154    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
155        match self {
156            GrpcError::Tonic(e) => Some(e),
157        }
158    }
159}
160
161impl From<tonic::transport::Error> for GrpcError {
162    fn from(e: tonic::transport::Error) -> Self {
163        GrpcError::Tonic(e)
164    }
165}
zcash_client_backend/tor/grpc.rs

zcash_client_backend/tor/
grpc.rs
