Merge pull request #794 from AloeareV/cratify_netutils

move some rpc code to new zingo-netutils crate, for consumption by zingo-proxy

Cargo.toml

@@ -11,6 +11,7 @@ members = [
     "zingolib",
     "zingoconfig",
     "zingo-testvectors",
+    "zingo-netutils",
     "zingo-memo",
 ]
 resolver = "2"
@@ -41,6 +42,7 @@ tonic = {version = "0.10.0", features = ["tls", "tls-roots", "tls-webpki-roots"]
 prost = "0.12.0"
 tower = { version = "0.4" }
 hex = "0.4"
+tokio-rustls = "0.23"
 
 [profile.release]
 debug = false

darkside-tests/src/utils.rs

@@ -281,7 +281,7 @@ pub async fn update_tree_states_for_transaction(
     raw_tx: RawTransaction,
     height: u64,
 ) -> TreeState {
-    let trees = zingolib::grpc_connector::GrpcConnector::get_trees(server_id.clone(), height - 1)
+    let trees = zingolib::grpc_connector::get_trees(server_id.clone(), height - 1)
         .await
         .unwrap();
     let mut sapling_tree: sapling_crypto::CommitmentTree = read_commitment_tree(

integration-tests/tests/integrations.rs

@@ -2587,7 +2587,7 @@ mod slow {
             .witness_tree_orchard
             .max_leaf_position(0)
             .unwrap();
-        let server_trees = zingolib::grpc_connector::GrpcConnector::get_trees(
+        let server_trees = zingolib::grpc_connector::get_trees(
             recipient.get_server_uri(),
             recipient.wallet.last_synced_height().await,
         )

zingo-netutils/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "zingo-netutils"
+version = "0.1.0"
+authors = ["[email protected]"]
+edition = "2021"
+
+[dependencies]
+http.workspace = true
+tokio-rustls.workspace = true
+zcash_client_backend.workspace = true
+tower.workspace = true
+hyper-rustls.workspace = true
+webpki-roots = "0.21.0"
+hyper.workspace = true
+http-body.workspace = true
+tonic.workspace = true
+prost.workspace = true
+rustls-pemfile = "1.0.0"
+
+[features]
+test = []
+
+

zingo-netutils/src/lib.rs

@@ -0,0 +1,125 @@
+use std::sync::Arc;
+use tower::ServiceExt;
+
+use http::Uri;
+use http_body::combinators::UnsyncBoxBody;
+use hyper::client::HttpConnector;
+use tokio_rustls::rustls::{ClientConfig, RootCertStore};
+use tonic::Status;
+use tower::util::BoxCloneService;
+use zcash_client_backend::proto::service::compact_tx_streamer_client::CompactTxStreamerClient;
+
+type UnderlyingService = BoxCloneService<
+    http::Request<UnsyncBoxBody<prost::bytes::Bytes, Status>>,
+    http::Response<hyper::Body>,
+    hyper::Error,
+>;
+
+#[derive(Clone)]
+pub struct GrpcConnector {
+    uri: http::Uri,
+}
+
+impl GrpcConnector {
+    pub fn new(uri: http::Uri) -> Self {
+        Self { uri }
+    }
+
+    pub fn uri(&self) -> &Uri {
+        &self.uri
+    }
+
+    pub fn get_client(
+        &self,
+    ) -> impl std::future::Future<
+        Output = Result<CompactTxStreamerClient<UnderlyingService>, Box<dyn std::error::Error>>,
+    > {
+        let uri = Arc::new(self.uri.clone());
+        async move {
+            let mut http_connector = HttpConnector::new();
+            http_connector.enforce_http(false);
+            if uri.scheme_str() == Some("https") {
+                let mut roots = RootCertStore::empty();
+                roots.add_server_trust_anchors(webpki_roots::TLS_SERVER_ROOTS.0.iter().map(
+                    |anchor_ref| {
+                        tokio_rustls::rustls::OwnedTrustAnchor::from_subject_spki_name_constraints(
+                            anchor_ref.subject,
+                            anchor_ref.spki,
+                            anchor_ref.name_constraints,
+                        )
+                    },
+                ));
+
+                #[cfg(test)]
+                add_test_cert_to_roots(&mut roots);
+
+                let tls = ClientConfig::builder()
+                    .with_safe_defaults()
+                    .with_root_certificates(roots)
+                    .with_no_client_auth();
+                let connector = tower::ServiceBuilder::new()
+                    .layer_fn(move |s| {
+                        let tls = tls.clone();
+
+                        hyper_rustls::HttpsConnectorBuilder::new()
+                            .with_tls_config(tls)
+                            .https_or_http()
+                            .enable_http2()
+                            .wrap_connector(s)
+                    })
+                    .service(http_connector);
+                let client = Box::new(hyper::Client::builder().build(connector));
+                let uri = uri.clone();
+                let svc = tower::ServiceBuilder::new()
+                    //Here, we take all the pieces of our uri, and add in the path from the Requests's uri
+                    .map_request(move |mut req: http::Request<tonic::body::BoxBody>| {
+                        let uri = Uri::builder()
+                            .scheme(uri.scheme().unwrap().clone())
+                            .authority(uri.authority().unwrap().clone())
+                            //here. The Request's uri contains the path to the GRPC sever and
+                            //the method being called
+                            .path_and_query(req.uri().path_and_query().unwrap().clone())
+                            .build()
+                            .unwrap();
+
+                        *req.uri_mut() = uri;
+                        req
+                    })
+                    .service(client);
+
+                Ok(CompactTxStreamerClient::new(svc.boxed_clone()))
+            } else {
+                let connector = tower::ServiceBuilder::new().service(http_connector);
+                let client = Box::new(hyper::Client::builder().http2_only(true).build(connector));
+                let uri = uri.clone();
+                let svc = tower::ServiceBuilder::new()
+                    //Here, we take all the pieces of our uri, and add in the path from the Requests's uri
+                    .map_request(move |mut req: http::Request<tonic::body::BoxBody>| {
+                        let uri = Uri::builder()
+                            .scheme(uri.scheme().unwrap().clone())
+                            .authority(uri.authority().unwrap().clone())
+                            //here. The Request's uri contains the path to the GRPC sever and
+                            //the method being called
+                            .path_and_query(req.uri().path_and_query().unwrap().clone())
+                            .build()
+                            .unwrap();
+
+                        *req.uri_mut() = uri;
+                        req
+                    })
+                    .service(client);
+
+                Ok(CompactTxStreamerClient::new(svc.boxed_clone()))
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+fn add_test_cert_to_roots(roots: &mut RootCertStore) {
+    const TEST_PEMFILE_PATH: &str = "test-data/localhost.pem";
+    let fd = std::fs::File::open(TEST_PEMFILE_PATH).unwrap();
+    let mut buf = std::io::BufReader::new(&fd);
+    let certs = rustls_pemfile::certs(&mut buf).unwrap();
+    roots.add_parsable_certificates(&certs);
+}

zingo-testutils/Cargo.toml

@@ -12,6 +12,7 @@ default = ["grpc-proxy"]
 [dependencies]
 zingoconfig = { path = "../zingoconfig" }
 zingolib = { path = "../zingolib", features = ["test"] }
+zingo-netutils = { path = "../zingo-netutils", features = ["test"] }
 
 zcash_client_backend = { workspace = true }
 zcash_primitives = { workspace = true }

zingo-testutils/src/grpc_proxy.rs

@@ -46,7 +46,7 @@ macro_rules! define_grpc_passthrough {
                 }
 
                 println!("Proxy passing through {rpc_name} call");
-                ::zingolib::grpc_connector::GrpcConnector::new($self.lightwalletd_uri.clone())
+                ::zingo_netutils::GrpcConnector::new($self.lightwalletd_uri.clone())
                     .get_client()
                     .await
                     .expect("Proxy server failed to create client")

zingocli/Cargo.toml

@@ -19,6 +19,6 @@ futures = "0.3.15"
 rustls-pemfile = "1.0.0"
 tokio =  { version = "1.24.2", features = ["full"] }
 tokio-stream = "0.1.6"
-tokio-rustls = "0.23.3"
+tokio-rustls = { workspace = true }
 webpki-roots = "0.21.0"
 json = "0.12.4"

zingolib/Cargo.toml

@@ -20,6 +20,7 @@ zingoconfig = { path = "../zingoconfig" }
 zingo-memo = { path = "../zingo-memo" }
 zingo-status = { path = "../zingo-status" }
 zingo-testvectors = { path = "../zingo-testvectors", optional = true }
+zingo-netutils = { path = "../zingo-netutils" }
 
 http-body = { workspace = true }
 hyper = { workspace = true }
@@ -48,15 +49,12 @@ rand = "0.8.5"
 serde_json = "1.0.82"
 tokio =  { version = "1.24.2", features = ["full"] }
 tokio-stream = "0.1.6"
-tokio-rustls = "0.23.3"
 reqwest = { version = "0.11", features = ["json"] }
-rustls-pemfile = "1.0.0"
 tower-http = { version = "0.2", features = ["add-extension"] }
 futures = { workspace = true }
 hex = { workspace = true }
 ring = "0.17.0"
 json = "0.12.4"
-webpki-roots = "0.21.0"
 lazy_static = "1.4.0"
 secp256k1 = "=0.26.0"
 ripemd160 = "0.9.1"

zingolib/src/blaze/block_management_reorg_detection.rs

@@ -1,12 +1,9 @@
 use crate::wallet::traits::FromCommitment;
-use crate::{
-    grpc_connector::GrpcConnector,
-    wallet::{
-        data::{BlockData, PoolNullifier},
-        notes::ShieldedNoteInterface,
-        traits::DomainWalletExt,
-        transactions::TransactionMetadataSet,
-    },
+use crate::wallet::{
+    data::{BlockData, PoolNullifier},
+    notes::ShieldedNoteInterface,
+    traits::DomainWalletExt,
+    transactions::TransactionMetadataSet,
 };
 use incrementalmerkletree::frontier::CommitmentTree;
 use incrementalmerkletree::{frontier, witness::IncrementalWitness, Hashable};
@@ -523,7 +520,7 @@ impl BlockManagementData {
             let tree = if prev_height < activation_height {
                 frontier::CommitmentTree::<<D::WalletNote as ShieldedNoteInterface>::Node, 32>::empty()
             } else {
-                let tree_state = GrpcConnector::get_trees(uri, prev_height).await?;
+                let tree_state = crate::grpc_connector::get_trees(uri, prev_height).await?;
                 let tree = hex::decode(D::get_tree(&tree_state)).unwrap();
                 self.unverified_treestates.write().await.push(tree_state);
                 read_commitment_tree(&tree[..]).map_err(|e| format!("{}", e))?

zingolib/src/blaze/fetch_compact_blocks.rs

@@ -35,7 +35,7 @@ impl FetchCompactBlocks {
 
             debug!("Fetching blocks {}-{}", start, end);
 
-            grpc_client.get_block_range(start, end, senders).await?;
+            crate::grpc_connector::get_block_range(&grpc_client, start, end, senders).await?;
         }
 
         Ok(())