added second test, added grpc connector

integration-tests/Cargo.toml

@@ -13,6 +13,7 @@ nym_poc = []
 [dependencies]
 zingoproxy-testutils = { path = "../zingoproxy-testutils" }
 zingo-proxyd = { path = "../zingo-proxyd" }
+zingo-rpc = { path = "../zingo-rpc" }
 
 zcash_client_backend = { workspace = true }
 zingo-netutils = { workspace = true }

integration-tests/tests/integrations.rs

@@ -4,7 +4,7 @@
 #![forbid(unsafe_code)]
 
 use std::sync::{atomic::AtomicBool, Arc};
-use zingo_netutils::GrpcConnector;
+use zingo_rpc::walletrpc::grpc::GrpcConnector;
 use zingoproxy_testutils::{drop_test_manager, get_proxy_uri, launch_test_manager};
 
 mod proxy {
@@ -13,14 +13,12 @@ mod proxy {
     #[tokio::test]
     async fn connect_to_lwd_get_info() {
         let online = Arc::new(AtomicBool::new(true));
-
         let (conf_path, _regtest_manager, regtest_handles, _handles, proxy_port, _nym_addr) =
             launch_test_manager(online.clone()).await;
 
         let proxy_uri = get_proxy_uri(proxy_port);
         println!("Attempting to connect to GRPC server at URI: {}", proxy_uri);
 
-        // TODO: Add GrpcConnector that uses zingo-rpc's NymTxStreamerClient.
         let mut client = GrpcConnector::new(proxy_uri)
             .get_client()
             .await
@@ -35,6 +33,23 @@ mod proxy {
 
         drop_test_manager(Some(conf_path), regtest_handles, online).await
     }
+
+    #[tokio::test]
+    async fn send_over_proxy() {
+        let online = Arc::new(AtomicBool::new(true));
+        let (conf_path, _regtest_manager, regtest_handles, _handles, proxy_port, _nym_addr) =
+            launch_test_manager(online.clone()).await;
+
+        let proxy_uri = get_proxy_uri(proxy_port);
+        println!("Attempting to connect to GRPC server at URI: {}", proxy_uri);
+
+        let mut client = GrpcConnector::new(proxy_uri)
+            .get_client()
+            .await
+            .expect("Failed to create GRPC client");
+
+        drop_test_manager(Some(conf_path), regtest_handles, online).await
+    }
 }
 
 mod nym {

zingo-rpc/Cargo.toml

@@ -5,6 +5,7 @@ authors = ["[email protected]"]
 edition = "2021"
 
 [features]
+test = []
 # NOTE: Deprecated
 nym_poc = []
 
@@ -25,3 +26,9 @@ prost = { workspace = true }
 bytes = { workspace = true }
 http-body = { workspace = true }
 
+tokio-rustls = "0.23.3"
+tower = { version = "0.4" }
+hyper-rustls = { version = "0.23", features = ["http2"] }
+webpki-roots = "0.21.0"
+hyper = { version = "0.14", features = ["full"] }
+rustls-pemfile = "1.0.0"

zingo-rpc/src/walletrpc.rs

@@ -1,4 +1,5 @@
 //! Nym enabled CompactTxStreamerClient implementation.
 //! NOTE: Currently only send_transaction is implemented.
 
+pub mod grpc;
 pub mod service;

zingo-rpc/src/walletrpc/grpc.rs

@@ -0,0 +1,138 @@
+//! A Nym enabled GrpcConnector for cunsumption by zcash wallets and integration tests.
+
+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 crate::walletrpc::service::NymTxStreamerClient;
+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,
+>;
+
+/// GrpcConnector Config.
+#[derive(Clone)]
+pub struct GrpcConnector {
+    uri: http::Uri,
+}
+
+impl GrpcConnector {
+    /// Creates a new GrpcConnector.
+    pub fn new(uri: http::Uri) -> Self {
+        Self { uri }
+    }
+
+    /// Returns GrpcConnector listen uri.
+    pub fn uri(&self) -> &Uri {
+        &self.uri
+    }
+
+    /// Returns NymTxStreamerClient.
+    pub fn get_client(
+        &self,
+    ) -> impl std::future::Future<
+        Output = Result<NymTxStreamerClient<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(NymTxStreamerClient {
+                    compact_tx_streamer_client: 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()))
+                Ok(NymTxStreamerClient {
+                    compact_tx_streamer_client: 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-rpc/src/walletrpc/service.rs

@@ -1,6 +1,6 @@
-//! Wrapper implementation of LibRustZCash's CompactTXStreamerClient that also holds feature-gated, nym-enabled implementations.
+//! Wrapper implementation of LibRustZCash's CompactTXStreamerClient that also holds nym-enabled implementations.
 //!
-//! NOTE: Currently only send_transaction has been implemented.
+//! NOTE: Currently only send_transaction has been implemented over nym.
 
 use http::Uri;
 use http_body::Body;
@@ -27,7 +27,8 @@ use crate::{
 /// Wrapper struct for the Nym enabled CompactTxStreamerClient.
 #[derive(Debug, Clone)]
 pub struct NymTxStreamerClient<T> {
-    compact_tx_streamer_client: CompactTxStreamerClient<T>,
+    /// LibRustZcash's CompactTxStreamerClient.
+    pub compact_tx_streamer_client: CompactTxStreamerClient<T>,
 }
 
 impl NymTxStreamerClient<tonic::transport::Channel> {

zingoproxy-testutils/src/lib.rs

@@ -13,7 +13,14 @@ fn write_lightwalletd_yml(
     let mut file = std::fs::File::create(file_path)?;
     writeln!(file, "grpc-bind-addr: 127.0.0.1:{}", bind_addr_port)?;
     writeln!(file, "cache-size: 10")?;
+    writeln!(file, "log-file: ../logs/lwd.log")?;
     writeln!(file, "log-level: 10")?;
+    writeln!(
+        file,
+        "zcash-conf-path: ../conf/zcash.conf
+"
+    )?;
+
     Ok(())
 }
 
@@ -34,6 +41,7 @@ fn write_zcash_conf(dir: &std::path::Path, rpcport: u16) -> Result<(), Box<dyn s
     writeln!(file, "rpcpassword=xxxxxx")?;
     writeln!(file, "rpcport={}", rpcport)?;
     writeln!(file, "rpcallowip=127.0.0.1")?;
+    writeln!(file, "listen=0")?;
     writeln!(file, "minetolocalwallet=0")?;
     writeln!(file, "mineraddress=zregtestsapling1fp58yvw40ytns3qrcc4p58ga9xunqglf5al6tl49fdlq3yrc2wk99dwrnxmhcyw5nlsqqa680rq")?;
     Ok(())