Reload client certificates

This allows creating a client with certificate paths instead of a
preloaded certificate. When created this way, on reconnection the client
will check if the certificate files have been changed on disk and reload
them if they have.

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

Author: behos

Cargo.toml

@@ -16,7 +16,7 @@ rust-version = "1.76"
 
 [features]
 default = []
-tls = ["rustls", "rustls-pemfile", "webpki-roots", "futures-rustls"]
+tls = ["rustls", "rustls-pemfile", "webpki-roots", "futures-rustls", "async-fs"]
 sasl = ["sasl-gssapi", "sasl-digest-md5"]
 sasl-digest-md5 = ["rsasl/unstable_custom_mechanism", "md5", "linkme", "hex"]
 sasl-gssapi = ["rsasl/gssapi"]
@@ -48,6 +48,7 @@ md5 = { version = "0.7.0", optional = true }
 hex = { version = "0.4.3", optional = true }
 linkme = { version = "0.3", optional = true }
 async-io = "2.3.2"
+async-fs = { version = "2.1.2", optional = true }
 futures = "0.3.30"
 async-net = "2.0.0"
 futures-rustls = { version = "0.26.0", optional = true }
@@ -67,6 +68,7 @@ tempfile = "3.6.0"
 rcgen = { version = "0.12.1", features = ["default", "x509-parser"] }
 serial_test = "3.0.0"
 asyncs = { version = "0.3.0", features = ["test"] }
+smol = "2.0.2"
 blocking = "1.6.0"
 
 [package.metadata.cargo-all-features]
@@ -78,3 +80,8 @@ all-features = true
 [profile.dev]
 # Need this for linkme crate to work for spawns in macOS
 lto = "thin"
+
+[[example]]
+name = "tls_file_based"
+path = "examples/tls_file_based.rs"
+required-features = ["tls", "smol"]

examples/tls_file_based.rs

@@ -0,0 +1,105 @@
+use std::env;
+use std::io::{self, Write};
+use std::path::PathBuf;
+use std::time::Duration;
+
+use zookeeper_client::Error::NodeExists;
+use zookeeper_client::{Acls, Client, CreateMode, TlsOptions};
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    env_logger::init();
+    smol::block_on(run()).unwrap_or_else(|e| {
+        eprintln!("Error: {}", e);
+        std::process::exit(1);
+    });
+    Ok(())
+}
+
+async fn run() -> Result<(), Box<dyn std::error::Error>> {
+    let connect_string = env::var("ZK_CONNECT_STRING").unwrap_or_else(|_| "tcp+tls://localhost:2281".to_string());
+    let ca_cert = PathBuf::from(env::var("ZK_CA_CERT").expect("ZK_CA_CERT environment variable is required"));
+    let client_cert =
+        PathBuf::from(env::var("ZK_CLIENT_CERT").expect("ZK_CLIENT_CERT environment variable is required"));
+    let client_key = PathBuf::from(env::var("ZK_CLIENT_KEY").expect("ZK_CLIENT_KEY environment variable is required"));
+
+    println!("Connecting to ZooKeeper with file-based TLS...");
+    println!("Server: {}", connect_string);
+    println!("CA cert: {}", ca_cert.display());
+    println!("Client cert: {}", client_cert.display());
+    println!("Client key: {}", client_key.display());
+
+    let loaded_ca_cert = async_fs::read_to_string(&ca_cert).await?;
+    let tls_options = TlsOptions::default()
+        .with_pem_ca_certs(&loaded_ca_cert)?
+        .with_pem_identity_files(&client_cert, &client_key)
+        .await?;
+
+    let tls_options = unsafe { tls_options.with_no_hostname_verification() };
+
+    println!("WARNING: Hostname verification disabled!");
+
+    let client = Client::connector()
+        .connection_timeout(Duration::from_secs(10))
+        .session_timeout(Duration::from_secs(30))
+        .tls(tls_options)
+        .secure_connect(&connect_string)
+        .await?;
+
+    println!("Connected to ZooKeeper successfully!");
+
+    let path = "/tls_example";
+
+    loop {
+        print!("\nOptions:\ne. Edit key\nq. Quit\nEnter choice (e/q): ");
+        io::stdout().flush()?;
+
+        let mut input = String::new();
+        io::stdin().read_line(&mut input)?;
+
+        match input.trim() {
+            "e" => {
+                print!("Enter new data for the key: ");
+                io::stdout().flush()?;
+
+                let mut data_input = String::new();
+                io::stdin().read_line(&mut data_input)?;
+                let data = data_input.trim().as_bytes();
+
+                println!("Setting data at path: {}", path);
+                match client.create(path, data, &CreateMode::Ephemeral.with_acls(Acls::anyone_all())).await {
+                    Ok(_) => println!("ZNode created successfully"),
+                    Err(NodeExists) => {
+                        println!("ZNode already exists, updating data...");
+                        client.set_data(path, data, None).await?;
+                        println!("ZNode data updated successfully");
+                    },
+                    Err(e) => {
+                        println!("Error creating/updating ZNode: {}", e);
+                        continue;
+                    },
+                }
+
+                match client.get_data(path).await {
+                    Ok((data, _stat)) => {
+                        println!("Current data: {}", String::from_utf8_lossy(&data));
+                    },
+                    Err(e) => println!("Error reading data: {}", e),
+                }
+            },
+            "q" => {
+                println!("Cleaning up and exiting...");
+                match client.delete(path, None).await {
+                    Ok(_) => println!("ZNode deleted successfully"),
+                    Err(_) => println!("ZNode may not exist or already deleted"),
+                }
+                break;
+            },
+            _ => {
+                println!("Invalid choice. Please enter 'e' or 'q'.");
+            },
+        }
+    }
+
+    println!("Example completed successfully!");
+    Ok(())
+}

src/session/connection.rs

@@ -1,5 +1,7 @@
 use std::io::{Error, ErrorKind, IoSlice, Result};
 use std::pin::Pin;
+#[cfg(feature = "tls")]
+use std::sync::Arc;
 use std::task::{Context, Poll};
 use std::time::Duration;
 
@@ -15,18 +17,18 @@ 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::TlsOptions;
 
 #[derive(Debug)]
 pub enum Connection {
@@ -170,15 +172,15 @@ impl Connection {
 #[derive(Clone)]
 pub struct Connector {
     #[cfg(feature = "tls")]
-    tls: Option<TlsConnector>,
+    tls_options: Option<TlsOptions>,
     timeout: Duration,
 }
 
 impl Connector {
     #[cfg(feature = "tls")]
     #[allow(dead_code)]
     pub fn new() -> Self {
-        Self { tls: None, timeout: Duration::from_secs(10) }
+        Self { tls_options: None, timeout: Duration::from_secs(10) }
     }
 
     #[cfg(not(feature = "tls"))]
@@ -187,14 +189,27 @@ 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) }
+    pub fn with_tls_options(tls_options: TlsOptions) -> Self {
+        Self { tls_options: Some(tls_options), timeout: Duration::from_secs(10) }
+    }
+
+    #[cfg(feature = "tls")]
+    async fn get_current_tls_connector(&self) -> Result<TlsConnector> {
+        let Some(ref tls_opts) = self.tls_options else {
+            return Err(Error::new(ErrorKind::InvalidInput, "no TLS configuration"));
+        };
+        let config = tls_opts
+            .to_config()
+            .await
+            .map_err(|e| Error::new(ErrorKind::InvalidData, format!("TLS config creation failed: {}", e)))?;
+        Ok(TlsConnector::from(Arc::new(config)))
     }
 
     #[cfg(feature = "tls")]
     async fn connect_tls(&self, stream: TcpStream, host: &str) -> Result<Connection> {
+        let tls_connector = self.get_current_tls_connector().await?;
         let domain = ServerName::try_from(host).unwrap().to_owned();
-        let stream = self.tls.as_ref().unwrap().connect(domain, stream).await?;
+        let stream = tls_connector.connect(domain, stream).await?;
         Ok(Connection::new_tls(stream))
     }
 
@@ -209,7 +224,7 @@ impl Connector {
     pub async fn connect(&self, endpoint: EndpointRef<'_>, deadline: &mut Deadline) -> Result<Connection> {
         if endpoint.tls {
             #[cfg(feature = "tls")]
-            if self.tls.is_none() {
+            if self.tls_options.is_none() {
                 return Err(Error::new(ErrorKind::Unsupported, "tls not supported"));
             }
             #[cfg(not(feature = "tls"))]
@@ -288,4 +303,12 @@ mod tests {
         let err = connector.connect(endpoint, &mut Deadline::never()).await.unwrap_err();
         assert_eq!(err.kind(), ErrorKind::Unsupported);
     }
+
+    #[cfg(feature = "tls")]
+    #[test]
+    fn test_with_tls_options() {
+        let tls_options = crate::TlsOptions::default();
+        let connector = Connector::with_tls_options(tls_options);
+        assert!(connector.tls_options.is_some());
+    }
 }

src/session/mod.rs

@@ -130,7 +130,7 @@ impl Builder {
             return Err(Error::BadArguments(&"connection timeout must not be negative"));
         }
         #[cfg(feature = "tls")]
-        let connector = Connector::with_tls(self.tls.unwrap_or_default().into_config()?);
+        let connector = Connector::with_tls_options(self.tls.unwrap_or_default());
         #[cfg(not(feature = "tls"))]
         let connector = Connector::new();
         let (state_sender, state_receiver) = asyncs::sync::watch::channel(SessionState::Disconnected);