[DEVEX-223] Add support for changing the root certificate

eventstore/src/grpc.rs

@@ -361,6 +361,7 @@ pub struct ClientSettings {
     )]
     pub(crate) default_deadline: Option<Duration>,
     pub(crate) connection_name: Option<String>,
+    pub(crate) tls_ca_file: Option<String>,
     pub(crate) user_cert_file: Option<String>,
     pub(crate) user_key_file: Option<String>,
 }
@@ -416,6 +417,10 @@ impl ClientSettings {
             .zip(self.user_key_file.as_ref())
     }
 
+    pub fn tls_ca_file(&self) -> Option<&String> {
+        self.tls_ca_file.as_ref()
+    }
+
     pub(crate) fn to_hyper_uri(&self, endpoint: &Endpoint) -> hyper::Uri {
         let scheme = if self.secure { "https" } else { "http" };
 
@@ -631,6 +636,10 @@ fn parse_from_url(
                 result.user_key_file = Some(value.to_string());
             }
 
+            "tlscafile" => {
+                result.tls_ca_file = Some(value.to_string());
+            }
+
             ignored => {
                 warn!("Ignored connection string parameter: {}", ignored);
                 continue;
@@ -645,6 +654,10 @@ fn parse_from_url(
         });
     }
 
+    if !result.secure && result.tls_ca_file.is_some() {
+        warn!("tlsCAFile passed to insecure connection. Will be ignored.");
+    }
+
     Ok(result)
 }
 
@@ -763,6 +776,7 @@ impl Default for ClientSettings {
             connection_name: None,
             user_cert_file: None,
             user_key_file: None,
+            tls_ca_file: None,
         }
     }
 }
@@ -812,17 +826,27 @@ impl NodeConnection {
     fn new(settings: ClientSettings) -> Self {
         let mut roots = rustls::RootCertStore::empty();
 
-        for cert in rustls_native_certs::load_native_certs().expect("could not load platform certs")
-        {
-            roots.add(cert).unwrap();
-        }
-
         RUSTLS_INIT.call_once(|| {
             rustls::crypto::aws_lc_rs::default_provider()
                 .install_default()
                 .expect("failed to install rustls crypto provider");
         });
 
+        if let Some(cert) = settings.tls_ca_file() {
+            let cert_chain: Result<Vec<CertificateDer<'_>>, _> =
+                CertificateDer::pem_file_iter(cert).unwrap().collect();
+
+            for cert in cert_chain.unwrap() {
+                roots.add(cert).unwrap();
+            }
+        } else {
+            for cert in
+                rustls_native_certs::load_native_certs().expect("could not load platform certs")
+            {
+                roots.add(cert).unwrap();
+            }
+        }
+
         let tls = tokio_rustls::rustls::ClientConfig::builder().with_root_certificates(roots);
 
         let mut tls = if let Some((cert, key)) = settings.user_certificate() {

eventstore/tests/fixtures/connection_string/mockups.toml

@@ -428,3 +428,20 @@ user_key_file = "bar"
 host = "localhost"
 port = 2_113
 
+[[mockups]]
+string = "esdb://localhost?tlsCaFile=foo"
+[mockups.expected]
+dns_discover = false
+max_discover_attempts = 3
+discovery_interval = 500
+gossip_timeout = 3_000
+preference = "Leader"
+secure = true
+tls_verify_cert = true
+tls_ca_file = "foo"
+keep_alive_interval = 10_000
+keep_alive_timeout = 10_000
+[[mockups.expected.hosts]]
+host = "localhost"
+port = 2_113
+

eventstore/tests/integration.rs

@@ -1,6 +1,7 @@
 mod api;
 mod common;
 mod images;
+mod misc;
 mod plugins;
 
 use crate::common::{fresh_stream_id, generate_events};
@@ -160,6 +161,7 @@ async fn wait_for_admin_to_be_available(client: &Client) -> eventstore::Result<(
 enum Tests {
     Api(ApiTests),
     Plugins(PluginTests),
+    Misc(MiscTests),
 }
 
 impl Tests {
@@ -191,6 +193,16 @@ impl From<PluginTests> for Tests {
     }
 }
 
+enum MiscTests {
+    RootCertificates,
+}
+
+impl From<MiscTests> for Tests {
+    fn from(test: MiscTests) -> Self {
+        Tests::Misc(test)
+    }
+}
+
 enum Topologies {
     SingleNode,
     Cluster,
@@ -266,6 +278,10 @@ async fn run_test(test: impl Into<Tests>, topology: Topologies) -> eyre::Result<
                 plugins::user_certificates::tests(container_port).await
             }
         },
+
+        Tests::Misc(test) => match test {
+            MiscTests::RootCertificates => misc::root_certificates::tests(container_port).await,
+        },
     };
 
     result?;
@@ -297,6 +313,11 @@ async fn plugin_usercertificates() -> eyre::Result<()> {
     run_test(PluginTests::UserCertificates, Topologies::SingleNode).await
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn root_certificates() -> eyre::Result<()> {
+    run_test(MiscTests::RootCertificates, Topologies::SingleNode).await
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn cluster_streams() -> eyre::Result<()> {
     run_test(ApiTests::Streams, Topologies::Cluster).await

eventstore/tests/misc/mod.rs

@@ -0,0 +1 @@
+pub mod root_certificates;

eventstore/tests/misc/root_certificates.rs

@@ -0,0 +1,51 @@
+use tracing::debug;
+
+async fn test_with_valid_root_certificate(port: u16) -> eyre::Result<()> {
+    let root_cert = "certs/ca/ca.crt";
+
+    let setts = format!(
+        "esdb://admin:changeit@localhost:{}?tlsVerifyCert=true&tls=true&tlsCaFile={}",
+        port, root_cert
+    )
+    .parse()?;
+    let client = eventstore::Client::new(setts)?;
+
+    let mut streams = client.read_all(&Default::default()).await?;
+
+    streams.next().await?;
+
+    Ok(())
+}
+
+async fn test_with_invalid_certificate(port: u16) -> eyre::Result<()> {
+    // invalid root certificate
+    let root_cert = "certs/node1/node.crt";
+
+    let setts = format!(
+        "esdb://admin:changeit@localhost:{}?tlsVerifyCert=true&tls=true&tlsCaFile={}",
+        port, root_cert
+    )
+    .parse()?;
+    let client = eventstore::Client::new(setts)?;
+
+    let result = client.read_all(&Default::default()).await;
+
+    assert!(
+        result.is_err(),
+        "Expected an error due to invalid certificate"
+    );
+
+    Ok(())
+}
+
+pub async fn tests(port: u16) -> eyre::Result<()> {
+    debug!("Before test_with_valid_root_certificate…");
+    test_with_valid_root_certificate(port).await?;
+    debug!("Complete");
+
+    debug!("Before test_with_invalid_certificate…");
+    test_with_invalid_certificate(port).await?;
+    debug!("Complete");
+
+    Ok(())
+}