rebased PR streamnative#267

Author: danielmulvad

src/client.rs

@@ -511,6 +511,25 @@ impl<Exe: Executor> PulsarBuilder<Exe> {
         self
     }
 
+    /// add a certificate and private key to authenticate the client in TLS connections
+    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
+    pub fn with_identity(mut self, certificate: Vec<u8>, private_key: Vec<u8>) -> Self {
+        match &mut self.tls_options {
+            Some(tls) => {
+                tls.certificate = Some(certificate);
+                tls.private_key = Some(private_key);
+            }
+            None => {
+                self.tls_options = Some(TlsOptions {
+                    certificate: Some(certificate),
+                    private_key: Some(private_key),
+                    ..Default::default()
+                })
+            }
+        }
+        self
+    }
+
     #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
     pub fn with_allow_insecure_connection(mut self, allow: bool) -> Self {
         match &mut self.tls_options {
@@ -560,6 +579,26 @@ impl<Exe: Executor> PulsarBuilder<Exe> {
         self
     }
 
+    /// add a certificate and private key to authenticate the client in TLS connections
+    #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
+    pub fn with_identity_files<P: AsRef<std::path::Path>>(
+        self,
+        certificate_path: P,
+        private_key_path: P,
+    ) -> Result<Self, std::io::Error> {
+        use std::io::Read;
+
+        let mut file = std::fs::File::open(certificate_path)?;
+        let mut certificate = vec![];
+        file.read_to_end(&mut certificate)?;
+
+        let mut file = std::fs::File::open(private_key_path)?;
+        let mut private_key = vec![];
+        file.read_to_end(&mut private_key)?;
+
+        Ok(self.with_identity(certificate, private_key))
+    }
+
     /// creates the Pulsar client and connects it
     #[cfg_attr(feature = "telemetry", tracing::instrument(skip_all))]
     pub async fn build(self) -> Result<Pulsar<Exe>, Error> {

src/connection.rs

@@ -20,6 +20,8 @@ use futures::{
     task::{Context, Poll},
     Future, FutureExt, Sink, SinkExt, Stream, StreamExt,
 };
+#[cfg(any(feature = "tokio-runtime", feature = "async-std-runtime"))]
+use native_tls::{Certificate, Identity};
 use proto::MessageIdData;
 use rand::{seq::SliceRandom, thread_rng};
 use url::Url;
@@ -781,6 +783,9 @@ impl<Exe: Executor> Connection<Exe> {
         auth_data: Option<Arc<Mutex<Box<dyn crate::authentication::Authentication>>>>,
         proxy_to_broker_url: Option<String>,
         certificate_chain: &[Certificate],
+        #[cfg(any(feature = "tokio-runtime", feature = "async-std-runtime"))] identity: &Option<
+            Identity,
+        >,
         allow_insecure_connection: bool,
         tls_hostname_verification_enabled: bool,
         connection_timeout: Duration,
@@ -840,6 +845,8 @@ impl<Exe: Executor> Connection<Exe> {
                 auth_data.clone(),
                 proxy_to_broker_url.clone(),
                 certificate_chain,
+                #[cfg(any(feature = "tokio-runtime", feature = "async-std-runtime"))]
+                identity.clone(),
                 allow_insecure_connection,
                 tls_hostname_verification_enabled,
                 executor.clone(),
@@ -918,6 +925,9 @@ impl<Exe: Executor> Connection<Exe> {
         auth: Option<Arc<Mutex<Box<dyn crate::authentication::Authentication>>>>,
         proxy_to_broker_url: Option<String>,
         certificate_chain: &[Certificate],
+        #[cfg(any(feature = "tokio-runtime", feature = "async-std-runtime"))] identity: Option<
+            Identity,
+        >,
         allow_insecure_connection: bool,
         tls_hostname_verification_enabled: bool,
         executor: Arc<Exe>,
@@ -934,6 +944,9 @@ impl<Exe: Executor> Connection<Exe> {
                     for certificate in certificate_chain {
                         builder.add_root_certificate(certificate.clone());
                     }
+                    if let Some(identity) = identity {
+                        builder.identity(identity);
+                    }
                     builder.danger_accept_invalid_hostnames(
                         allow_insecure_connection && !tls_hostname_verification_enabled,
                     );
@@ -1035,6 +1048,9 @@ impl<Exe: Executor> Connection<Exe> {
                     for certificate in certificate_chain {
                         connector = connector.add_root_certificate(certificate.clone());
                     }
+                    if let Some(identity) = identity {
+                        connector = connector.identity(identity);
+                    }
                     connector = connector.danger_accept_invalid_hostnames(
                         allow_insecure_connection && !tls_hostname_verification_enabled,
                     );

src/connection_manager.rs

@@ -1,6 +1,8 @@
 use std::{collections::HashMap, sync::Arc, time::Duration};
 
 use futures::{channel::oneshot, lock::Mutex};
+#[cfg(any(feature = "tokio-runtime", feature = "async-std-runtime"))]
+use native_tls::{Certificate, Identity};
 use rand::Rng;
 use url::Url;
 
@@ -74,6 +76,12 @@ pub struct TlsOptions {
     /// contains a list of PEM encoded certificates
     pub certificate_chain: Option<Vec<u8>>,
 
+    /// PEM encoded X509 certificates
+    pub certificate: Option<Vec<u8>>,
+
+    /// is a PEM encoded PKCS #8 formatted private key for the leaf certificate
+    pub private_key: Option<Vec<u8>>,
+
     /// allow insecure TLS connection if set to true
     ///
     /// defaults to *false*
@@ -90,6 +98,8 @@ impl Default for TlsOptions {
     fn default() -> Self {
         Self {
             certificate_chain: None,
+            certificate: None,
+            private_key: None,
             allow_insecure_connection: false,
             tls_hostname_verification_enabled: true,
         }
@@ -116,6 +126,8 @@ pub struct ConnectionManager<Exe: Executor> {
     pub(crate) operation_retry_options: OperationRetryOptions,
     tls_options: TlsOptions,
     certificate_chain: Vec<Certificate>,
+    #[cfg(any(feature = "tokio-runtime", feature = "async-std-runtime"))]
+    identity: Option<Identity>,
     outbound_channel_size: usize,
 }
 
@@ -172,6 +184,17 @@ impl<Exe: Executor> ConnectionManager<Exe> {
             }
         };
 
+        #[cfg(any(feature = "tokio-runtime", feature = "async-std-runtime"))]
+        let identity = match (
+            tls_options.certificate.as_ref(),
+            tls_options.private_key.as_ref(),
+        ) {
+            (None, _) | (_, None) => None,
+            (Some(certificate), Some(privatekey)) => {
+                Some(native_tls::Identity::from_pkcs8(&certificate, &privatekey)?)
+            }
+        };
+
         if let Some(auth) = auth.clone() {
             auth.lock().await.initialize().await?;
         }
@@ -185,6 +208,8 @@ impl<Exe: Executor> ConnectionManager<Exe> {
             operation_retry_options,
             tls_options,
             certificate_chain,
+            #[cfg(any(feature = "tokio-runtime", feature = "async-std-runtime"))]
+            identity,
             outbound_channel_size,
         };
         let broker_address = BrokerAddress {
@@ -303,6 +328,8 @@ impl<Exe: Executor> ConnectionManager<Exe> {
                 self.auth.clone(),
                 proxy_url.clone(),
                 &self.certificate_chain,
+                #[cfg(any(feature = "tokio-runtime", feature = "async-std-runtime"))]
+                &self.identity,
                 self.tls_options.allow_insecure_connection,
                 self.tls_options.tls_hostname_verification_enabled,
                 self.connection_retry_options.connection_timeout,