Support client side tls certificates reload

This allows creating a client with dynamic tls certificates. When
created this way, on reconnection the client will use latest tls
certificates.

This allows us to have auto-reloading of refreshed certificates stored
anywhere in client side.

Resolves #59.

Author: behos

Cargo.toml

@@ -69,6 +69,10 @@ rcgen = { version = "0.14.1", features = ["default", "x509-parser"] }
 serial_test = "3.0.0"
 asyncs = { version = "0.4.0", features = ["test"] }
 blocking = "1.6.0"
+rustls-pki-types = "1.12.0"
+x509-parser = "0.17.0"
+atomic-write-file = "0.2.3"
+notify = "7.0.0"
 
 [package.metadata.cargo-all-features]
 skip_optional_dependencies = true

src/lib.rs

@@ -45,7 +45,7 @@ mod util;
 pub use self::acl::{Acl, Acls, AuthId, AuthUser, Permission};
 pub use self::error::Error;
 #[cfg(feature = "tls")]
-pub use self::tls::TlsOptions;
+pub use self::tls::{TlsCerts, TlsCertsOptions, TlsDynamicCerts, TlsOptions};
 pub use crate::client::*;
 #[cfg(feature = "sasl-digest-md5")]
 pub use crate::sasl::DigestMd5SaslOptions;

src/session/connection.rs

@@ -10,23 +10,15 @@ use bytes::buf::BufMut;
 use futures::io::BufReader;
 use futures::prelude::*;
 use futures_lite::AsyncReadExt;
+#[cfg(feature = "tls")]
+pub use futures_rustls::client::TlsStream;
 use ignore_result::Ignore;
 use tracing::{debug, trace};
 
-#[cfg(feature = "tls")]
-mod tls {
-    pub use std::sync::Arc;
-
-    pub use futures_rustls::client::TlsStream;
-    pub use futures_rustls::TlsConnector;
-    pub use rustls::pki_types::ServerName;
-    pub use rustls::ClientConfig;
-}
-#[cfg(feature = "tls")]
-use tls::*;
-
 use crate::deadline::Deadline;
 use crate::endpoint::{EndpointRef, IterableEndpoints};
+#[cfg(feature = "tls")]
+use crate::tls::TlsClient;
 
 #[derive(Debug)]
 pub enum Connection {
@@ -170,7 +162,7 @@ impl Connection {
 #[derive(Clone)]
 pub struct Connector {
     #[cfg(feature = "tls")]
-    tls: Option<TlsConnector>,
+    tls: Option<TlsClient>,
     timeout: Duration,
 }
 
@@ -186,15 +178,8 @@ impl Connector {
     }
 
     #[cfg(feature = "tls")]
-    pub fn with_tls(config: ClientConfig) -> Self {
-        Self { tls: Some(TlsConnector::from(Arc::new(config))), timeout: Duration::from_secs(10) }
-    }
-
-    #[cfg(feature = "tls")]
-    async fn connect_tls(&self, stream: TcpStream, host: &str) -> Result<Connection> {
-        let domain = ServerName::try_from(host).unwrap().to_owned();
-        let stream = self.tls.as_ref().unwrap().connect(domain, stream).await?;
-        Ok(Connection::new_tls(stream))
+    pub fn with_tls(client: TlsClient) -> Self {
+        Self { tls: Some(client), timeout: Duration::from_secs(10) }
     }
 
     pub fn timeout(&self) -> Duration {
@@ -205,34 +190,25 @@ impl Connector {
         self.timeout = timeout;
     }
 
-    pub async fn connect(&self, endpoint: EndpointRef<'_>, deadline: &mut Deadline) -> Result<Connection> {
+    async fn connect_endpoint(&self, endpoint: EndpointRef<'_>) -> Result<Connection> {
         if endpoint.tls {
             #[cfg(feature = "tls")]
-            if self.tls.is_none() {
-                return Err(Error::new(ErrorKind::Unsupported, "tls not configured"));
-            }
+            return match self.tls.as_ref() {
+                None => return Err(Error::new(ErrorKind::Unsupported, "tls not configured")),
+                Some(client) => client.connect(endpoint.host, endpoint.port).await.map(Connection::new_tls),
+            };
             #[cfg(not(feature = "tls"))]
             return Err(Error::new(ErrorKind::Unsupported, "tls not supported"));
         }
+        TcpStream::connect((endpoint.host, endpoint.port)).await.map(Connection::new_raw)
+    }
+
+    pub async fn connect(&self, endpoint: EndpointRef<'_>, deadline: &mut Deadline) -> Result<Connection> {
         select! {
+            biased;
+            r = self.connect_endpoint(endpoint) => r,
             _ = unsafe { Pin::new_unchecked(deadline) } => Err(Error::new(ErrorKind::TimedOut, "deadline exceed")),
             _ = Timer::after(self.timeout) => Err(Error::new(ErrorKind::TimedOut, format!("connection timeout{:?} exceed", self.timeout))),
-            r = TcpStream::connect((endpoint.host, endpoint.port)) => {
-                match r {
-                    Err(err) => Err(err),
-                    Ok(sock) => {
-                        let connection = if endpoint.tls {
-                            #[cfg(not(feature = "tls"))]
-                            unreachable!("tls not supported");
-                            #[cfg(feature = "tls")]
-                            self.connect_tls(sock, endpoint.host).await?
-                        } else {
-                            Connection::new_raw(sock)
-                        };
-                        Ok(connection)
-                    },
-                }
-            },
         }
     }
 

src/session/mod.rs

@@ -131,7 +131,7 @@ impl Builder {
         }
         #[cfg(feature = "tls")]
         let connector = match self.tls {
-            Some(options) => Connector::with_tls(options.into_config()?),
+            Some(options) => Connector::with_tls(options.into_client()?),
             None => Connector::new(),
         };
         #[cfg(not(feature = "tls"))]

src/tls.rs

@@ -0,0 +1,67 @@
+use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
+use rustls::crypto::{CryptoProvider, WebPkiSupportedAlgorithms};
+use rustls::pki_types::{CertificateDer, ServerName, UnixTime};
+use rustls::server::ParsedCertificate;
+use rustls::{DigitallySignedStruct, Error as TlsError, RootCertStore, SignatureScheme};
+
+use crate::client::Result;
+
+// Rustls tends to make disable of hostname verification verbose since it exposes man-in-the-middle
+// attacks. Though, there are still attempts to disable hostname verification in rustls, but no got
+// merged until now.
+// * Allow disabling Hostname Verification: https://github.com/rustls/rustls/issues/578
+// * Dangerous verifiers API proposal: https://github.com/rustls/rustls/pull/1197
+#[derive(Debug)]
+pub(super) struct NoHostnameVerificationServerCertVerifier {
+    roots: RootCertStore,
+    supported: WebPkiSupportedAlgorithms,
+}
+
+impl NoHostnameVerificationServerCertVerifier {
+    pub unsafe fn new(roots: RootCertStore) -> Self {
+        Self { roots, supported: CryptoProvider::get_default().unwrap().signature_verification_algorithms }
+    }
+}
+
+impl ServerCertVerifier for NoHostnameVerificationServerCertVerifier {
+    fn verify_server_cert(
+        &self,
+        end_entity: &CertificateDer<'_>,
+        intermediates: &[CertificateDer<'_>],
+        _server_name: &ServerName<'_>,
+        _ocsp_response: &[u8],
+        now: UnixTime,
+    ) -> Result<ServerCertVerified, TlsError> {
+        let cert = ParsedCertificate::try_from(end_entity)?;
+        rustls::client::verify_server_cert_signed_by_trust_anchor(
+            &cert,
+            &self.roots,
+            intermediates,
+            now,
+            self.supported.all,
+        )?;
+        Ok(ServerCertVerified::assertion())
+    }
+
+    fn verify_tls12_signature(
+        &self,
+        message: &[u8],
+        cert: &CertificateDer<'_>,
+        dss: &DigitallySignedStruct,
+    ) -> Result<HandshakeSignatureValid, TlsError> {
+        rustls::crypto::verify_tls12_signature(message, cert, dss, &self.supported)
+    }
+
+    fn verify_tls13_signature(
+        &self,
+        message: &[u8],
+        cert: &CertificateDer<'_>,
+        dss: &DigitallySignedStruct,
+    ) -> Result<HandshakeSignatureValid, TlsError> {
+        rustls::crypto::verify_tls12_signature(message, cert, dss, &self.supported)
+    }
+
+    fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
+        self.supported.supported_schemes()
+    }
+}