Support tls certificates reload and crls

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 reload refreshed certificates stored somewhere in
client side.

This commit also adds support for crls in cert verifier, so it will
reject revoked server certs.

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::{TlsCa, TlsCerts, TlsCertsBuilder, TlsCertsOptions, TlsDynamicCerts, TlsIdentity, 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,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"))]