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

@@ -16,7 +16,7 @@ rust-version = "1.76"
 
 [features]
 default = []
-tls = ["rustls", "rustls-pemfile", "futures-rustls"]
+tls = ["rustls", "rustls-webpki", "rustls-pemfile", "futures-rustls"]
 sasl = ["sasl-gssapi", "sasl-digest-md5"]
 sasl-digest-md5 = ["rsasl/unstable_custom_mechanism", "md5", "linkme", "hex"]
 sasl-gssapi = ["rsasl/gssapi"]
@@ -38,6 +38,7 @@ hashlink = "0.8.0"
 either = "1.9.0"
 uuid = { version = "1.4.1", features = ["v4"] }
 rustls = { version =  "0.23.2", optional = true }
+rustls-webpki = { version = "0.103.4", optional = true }
 rustls-pemfile = { version = "2", optional = true }
 derive-where = "1.2.7"
 fastrand = "2.0.2"
@@ -67,6 +68,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::*;
 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,31 +162,22 @@ impl Connection {
 #[derive(Clone)]
 pub struct Connector {
     #[cfg(feature = "tls")]
-    tls: Option<TlsConnector>,
+    tls: Option<TlsClient>,
     timeout: Duration,
 }
 
 impl Connector {
-    #[cfg(feature = "tls")]
-    pub fn new() -> Self {
-        Self { tls: None, timeout: Duration::from_secs(10) }
-    }
-
-    #[cfg(not(feature = "tls"))]
     pub fn new() -> Self {
-        Self { timeout: Duration::from_secs(10) }
-    }
-
-    #[cfg(feature = "tls")]
-    pub fn with_tls(config: ClientConfig) -> Self {
-        Self { tls: Some(TlsConnector::from(Arc::new(config))), timeout: Duration::from_secs(10) }
+        Self {
+            #[cfg(feature = "tls")]
+            tls: None,
+            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 +188,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"))]