Add support for TLS connections with self-signed cert on validator gR…

…PC interface (#20238)

Client side use of TLS will be enabled-by-default in a future PR.

This has corrected TLS upgrade behavior compared to older PR #19796.

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [x] Nodes (Validators and Full nodes): Adds support for TLS on
validator gRPC interface.
- [ ] Indexer: 
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:
- [ ] REST API:

Cargo.toml

@@ -262,6 +262,7 @@ async-graphql = "=7.0.1"
 async-graphql-axum = "=7.0.1"
 async-graphql-value = "=7.0.1"
 async-recursion = "1.0.4"
+async-stream = "0.3.6"
 async-trait = "0.1.61"
 atomic_float = "0.1"
 aws-config = "0.56"

consensus/core/src/network/tonic_network.rs

@@ -24,6 +24,7 @@ use mysten_network::{
     Multiaddr,
 };
 use parking_lot::RwLock;
+use sui_tls::AllowPublicKeys;
 use tokio::{
     pin,
     task::JoinSet,
@@ -44,7 +45,6 @@ use super::{
         consensus_service_client::ConsensusServiceClient,
         consensus_service_server::ConsensusService,
     },
-    tonic_tls::create_rustls_client_config,
     BlockStream, NetworkClient, NetworkManager, NetworkService,
 };
 use crate::{
@@ -54,7 +54,7 @@ use crate::{
     error::{ConsensusError, ConsensusResult},
     network::{
         tonic_gen::consensus_service_server::ConsensusServiceServer,
-        tonic_tls::create_rustls_server_config,
+        tonic_tls::certificate_server_name,
     },
     CommitIndex, Round,
 };
@@ -381,7 +381,16 @@ impl ChannelPool {
         let address = format!("https://{address}");
         let config = &self.context.parameters.tonic;
         let buffer_size = config.connection_buffer_size;
-        let client_tls_config = create_rustls_client_config(&self.context, network_keypair, peer);
+        let client_tls_config = sui_tls::create_rustls_client_config(
+            self.context
+                .committee
+                .authority(peer)
+                .network_key
+                .clone()
+                .into_inner(),
+            certificate_server_name(&self.context),
+            Some(network_keypair.private_key().into_inner()),
+        );
         let endpoint = tonic_rustls::Channel::from_shared(address.clone())
             .unwrap()
             .connect_timeout(timeout)
@@ -728,8 +737,17 @@ impl<S: NetworkService> NetworkManager<S> for TonicManager {
             Arc::new(builder)
         };
 
-        let tls_server_config =
-            create_rustls_server_config(&self.context, self.network_keypair.clone());
+        let tls_server_config = sui_tls::create_rustls_server_config(
+            self.network_keypair.clone().private_key().into_inner(),
+            certificate_server_name(&self.context),
+            AllowPublicKeys::new(
+                self.context
+                    .committee
+                    .authorities()
+                    .map(|(_i, a)| a.network_key.clone().into_inner())
+                    .collect(),
+            ),
+        );
         let tls_acceptor = TlsAcceptor::from(Arc::new(tls_server_config));
 
         // Create listener to incoming connections.

consensus/core/src/network/tonic_tls.rs

@@ -2,63 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::context::Context;
-use consensus_config::{AuthorityIndex, NetworkKeyPair};
-use sui_tls::AllowPublicKeys;
-use tokio_rustls::rustls::{ClientConfig, ServerConfig};
 
-pub(crate) fn create_rustls_server_config(
-    context: &Context,
-    network_keypair: NetworkKeyPair,
-) -> ServerConfig {
-    let allower = AllowPublicKeys::new(
-        context
-            .committee
-            .authorities()
-            .map(|(_i, a)| a.network_key.clone().into_inner())
-            .collect(),
-    );
-    let verifier = sui_tls::ClientCertVerifier::new(allower, certificate_server_name(context));
-    // TODO: refactor to use key bytes
-    let self_signed_cert = sui_tls::SelfSignedCertificate::new(
-        network_keypair.private_key().into_inner(),
-        &certificate_server_name(context),
-    );
-    let tls_cert = self_signed_cert.rustls_certificate();
-    let tls_private_key = self_signed_cert.rustls_private_key();
-    let mut tls_config = verifier
-        .rustls_server_config(vec![tls_cert], tls_private_key)
-        .unwrap_or_else(|e| panic!("Failed to create TLS server config: {:?}", e));
-    tls_config.alpn_protocols = vec![b"h2".to_vec()];
-    tls_config
-}
-
-pub(crate) fn create_rustls_client_config(
-    context: &Context,
-    network_keypair: NetworkKeyPair,
-    target: AuthorityIndex,
-) -> ClientConfig {
-    let target_public_key = context
-        .committee
-        .authority(target)
-        .network_key
-        .clone()
-        .into_inner();
-    let self_signed_cert = sui_tls::SelfSignedCertificate::new(
-        network_keypair.private_key().into_inner(),
-        &certificate_server_name(context),
-    );
-    let tls_cert = self_signed_cert.rustls_certificate();
-    let tls_private_key = self_signed_cert.rustls_private_key();
-    let mut tls_config =
-        sui_tls::ServerCertVerifier::new(target_public_key, certificate_server_name(context))
-            .rustls_client_config(vec![tls_cert], tls_private_key)
-            .unwrap_or_else(|e| panic!("Failed to create TLS client config: {:?}", e));
-    // ServerCertVerifier sets alpn for completeness, but alpn cannot be predefined when
-    // using HttpsConnector from hyper-rustls, as in TonicManager.
-    tls_config.alpn_protocols = vec![];
-    tls_config
-}
-
-fn certificate_server_name(context: &Context) -> String {
+pub(crate) fn certificate_server_name(context: &Context) -> String {
     format!("consensus_epoch_{}", context.committee.epoch())
 }

crates/mysten-network/Cargo.toml

@@ -9,6 +9,7 @@ publish = false
 
 [dependencies]
 anemo.workspace = true
+async-stream.workspace = true
 bcs.workspace = true
 bytes.workspace = true
 eyre.workspace = true
@@ -18,8 +19,10 @@ multiaddr.workspace = true
 serde.workspace = true
 once_cell.workspace = true
 snap.workspace = true
+hyper-rustls.workspace = true
 hyper-util.workspace = true
 tokio = { workspace = true, features = ["sync", "rt", "macros"] }
+tokio-rustls.workspace = true
 tokio-stream.workspace = true
 tonic.workspace = true
 tonic-health.workspace = true

crates/mysten-network/src/client.rs

@@ -21,53 +21,67 @@ use std::{
     vec,
 };
 use tokio::task::JoinHandle;
+use tokio_rustls::rustls::ClientConfig;
 use tonic::transport::{Channel, Endpoint, Uri};
 use tower::Service;
 use tracing::{info, trace};
 
-pub async fn connect(address: &Multiaddr) -> Result<Channel> {
-    let channel = endpoint_from_multiaddr(address)?.connect().await?;
+pub async fn connect(address: &Multiaddr, tls_config: Option<ClientConfig>) -> Result<Channel> {
+    let channel = endpoint_from_multiaddr(address, tls_config)?
+        .connect()
+        .await?;
     Ok(channel)
 }
 
-pub fn connect_lazy(address: &Multiaddr) -> Result<Channel> {
-    let channel = endpoint_from_multiaddr(address)?.connect_lazy();
+pub fn connect_lazy(address: &Multiaddr, tls_config: Option<ClientConfig>) -> Result<Channel> {
+    let channel = endpoint_from_multiaddr(address, tls_config)?.connect_lazy();
     Ok(channel)
 }
 
-pub(crate) async fn connect_with_config(address: &Multiaddr, config: &Config) -> Result<Channel> {
-    let channel = endpoint_from_multiaddr(address)?
+pub(crate) async fn connect_with_config(
+    address: &Multiaddr,
+    tls_config: Option<ClientConfig>,
+    config: &Config,
+) -> Result<Channel> {
+    let channel = endpoint_from_multiaddr(address, tls_config)?
         .apply_config(config)
         .connect()
         .await?;
     Ok(channel)
 }
 
-pub(crate) fn connect_lazy_with_config(address: &Multiaddr, config: &Config) -> Result<Channel> {
-    let channel = endpoint_from_multiaddr(address)?
+pub(crate) fn connect_lazy_with_config(
+    address: &Multiaddr,
+    tls_config: Option<ClientConfig>,
+    config: &Config,
+) -> Result<Channel> {
+    let channel = endpoint_from_multiaddr(address, tls_config)?
         .apply_config(config)
         .connect_lazy();
     Ok(channel)
 }
 
-fn endpoint_from_multiaddr(addr: &Multiaddr) -> Result<MyEndpoint> {
+fn endpoint_from_multiaddr(
+    addr: &Multiaddr,
+    tls_config: Option<ClientConfig>,
+) -> Result<MyEndpoint> {
     let mut iter = addr.iter();
 
     let channel = match iter.next().ok_or_else(|| eyre!("address is empty"))? {
         Protocol::Dns(_) => {
             let (dns_name, tcp_port, http_or_https) = parse_dns(addr)?;
             let uri = format!("{http_or_https}://{dns_name}:{tcp_port}");
-            MyEndpoint::try_from_uri(uri)?
+            MyEndpoint::try_from_uri(uri, tls_config)?
         }
         Protocol::Ip4(_) => {
             let (socket_addr, http_or_https) = parse_ip4(addr)?;
             let uri = format!("{http_or_https}://{socket_addr}");
-            MyEndpoint::try_from_uri(uri)?
+            MyEndpoint::try_from_uri(uri, tls_config)?
         }
         Protocol::Ip6(_) => {
             let (socket_addr, http_or_https) = parse_ip6(addr)?;
             let uri = format!("{http_or_https}://{socket_addr}");
-            MyEndpoint::try_from_uri(uri)?
+            MyEndpoint::try_from_uri(uri, tls_config)?
         }
         unsupported => return Err(eyre!("unsupported protocol {unsupported}")),
     };
@@ -77,21 +91,25 @@ fn endpoint_from_multiaddr(addr: &Multiaddr) -> Result<MyEndpoint> {
 
 struct MyEndpoint {
     endpoint: Endpoint,
+    tls_config: Option<ClientConfig>,
 }
 
 static DISABLE_CACHING_RESOLVER: OnceCell<bool> = OnceCell::new();
 
 impl MyEndpoint {
-    fn new(endpoint: Endpoint) -> Self {
-        Self { endpoint }
+    fn new(endpoint: Endpoint, tls_config: Option<ClientConfig>) -> Self {
+        Self {
+            endpoint,
+            tls_config,
+        }
     }
 
-    fn try_from_uri(uri: String) -> Result<Self> {
+    fn try_from_uri(uri: String, tls_config: Option<ClientConfig>) -> Result<Self> {
         let uri: Uri = uri
             .parse()
             .with_context(|| format!("unable to create Uri from '{uri}'"))?;
         let endpoint = Endpoint::from(uri);
-        Ok(Self::new(endpoint))
+        Ok(Self::new(endpoint, tls_config))
     }
 
     fn apply_config(mut self, config: &Config) -> Self {
@@ -107,20 +125,51 @@ impl MyEndpoint {
         });
 
         if disable_caching_resolver {
-            self.endpoint.connect_lazy()
+            if let Some(tls_config) = self.tls_config {
+                self.endpoint.connect_with_connector_lazy(
+                    hyper_rustls::HttpsConnectorBuilder::new()
+                        .with_tls_config(tls_config)
+                        .https_only()
+                        .enable_http2()
+                        .build(),
+                )
+            } else {
+                self.endpoint.connect_lazy()
+            }
         } else {
             let mut http = HttpConnector::new_with_resolver(CachingResolver::new());
             http.enforce_http(false);
             http.set_nodelay(true);
             http.set_keepalive(None);
             http.set_connect_timeout(None);
 
-            self.endpoint.connect_with_connector_lazy(http)
+            if let Some(tls_config) = self.tls_config {
+                let https = hyper_rustls::HttpsConnectorBuilder::new()
+                    .with_tls_config(tls_config)
+                    .https_only()
+                    .enable_http1()
+                    .wrap_connector(http);
+                self.endpoint.connect_with_connector_lazy(https)
+            } else {
+                self.endpoint.connect_with_connector_lazy(http)
+            }
         }
     }
 
     async fn connect(self) -> Result<Channel> {
-        self.endpoint.connect().await.map_err(Into::into)
+        if let Some(tls_config) = self.tls_config {
+            let https_connector = hyper_rustls::HttpsConnectorBuilder::new()
+                .with_tls_config(tls_config)
+                .https_only()
+                .enable_http2()
+                .build();
+            self.endpoint
+                .connect_with_connector(https_connector)
+                .await
+                .map_err(Into::into)
+        } else {
+            self.endpoint.connect().await.map_err(Into::into)
+        }
     }
 }