Use system installed certificates by default and accept neo4j+ssc connections

lib/Cargo.toml

@@ -43,7 +43,8 @@ thiserror = "1.0.7"
 time = { version = "0.3.22", optional = true }
 tokio = { version = "1.5.0", features = ["full"] }
 url = "2.0.0"
-webpki-roots = "0.26.0"
+rustls-native-certs = "0.7.1"
+rustls-pemfile = "2.1.2"
 
 [dependencies.chrono]
 version = "0.4.35"

lib/src/auth/mod.rs

@@ -0,0 +1,12 @@
+mod static_client_certificate_provider;
+
+#[derive(Debug, Clone)]
+pub struct ClientCertificate {
+    pub(crate) cert_file: String,  // Path to the TLS certificate file.
+}
+
+pub trait ClientCertificateProvider {
+    fn get_certificate(&self) -> ClientCertificate;
+}
+
+pub use static_client_certificate_provider::StaticClientCertificateProvider;

lib/src/auth/static_client_certificate_provider.rs

@@ -0,0 +1,21 @@
+use crate::auth::{ClientCertificate, ClientCertificateProvider};
+
+pub struct StaticClientCertificateProvider {
+    certificate: ClientCertificate
+}
+
+impl StaticClientCertificateProvider {
+    pub fn new(cert_file: String) -> Self {
+        Self {
+            certificate: ClientCertificate {
+                cert_file,
+            }
+        }
+    }
+}
+
+impl ClientCertificateProvider for StaticClientCertificateProvider {
+    fn get_certificate(&self) -> ClientCertificate {
+        self.certificate.clone()
+    }
+}

lib/src/config.rs

@@ -1,5 +1,6 @@
 use crate::errors::{Error, Result};
 use std::{ops::Deref, sync::Arc};
+use crate::auth::{ClientCertificate, ClientCertificateProvider};
 
 const DEFAULT_DATABASE: &str = "neo4j";
 const DEFAULT_FETCH_SIZE: usize = 200;
@@ -58,6 +59,7 @@ pub struct Config {
     pub(crate) max_connections: usize,
     pub(crate) db: Database,
     pub(crate) fetch_size: usize,
+    pub(crate) client_certificate: Option<ClientCertificate>,
 }
 
 impl Config {
@@ -77,6 +79,7 @@ pub struct ConfigBuilder {
     db: Database,
     fetch_size: usize,
     max_connections: usize,
+    client_certificate_provider: Option<Box<dyn ClientCertificateProvider>>,
 }
 
 impl ConfigBuilder {
@@ -128,6 +131,11 @@ impl ConfigBuilder {
         self
     }
 
+    pub fn with_client_certificate_provider(mut self, provider: Box<dyn ClientCertificateProvider>) -> Self {
+        self.client_certificate_provider = Some(provider);
+        self
+    }
+
     pub fn build(self) -> Result<Config> {
         if let (Some(uri), Some(user), Some(password)) = (self.uri, self.user, self.password) {
             Ok(Config {
@@ -137,6 +145,7 @@ impl ConfigBuilder {
                 fetch_size: self.fetch_size,
                 max_connections: self.max_connections,
                 db: self.db,
+                client_certificate: self.client_certificate_provider.map(|p| p.get_certificate()),
             })
         } else {
             Err(Error::InvalidConfig)
@@ -153,6 +162,7 @@ impl Default for ConfigBuilder {
             db: DEFAULT_DATABASE.into(),
             max_connections: DEFAULT_MAX_CONNECTIONS,
             fetch_size: DEFAULT_FETCH_SIZE,
+            client_certificate_provider: None,
         }
     }
 }
@@ -178,6 +188,7 @@ mod tests {
         assert_eq!(&*config.db, "some_db");
         assert_eq!(config.fetch_size, 10);
         assert_eq!(config.max_connections, 5);
+        assert!(config.client_certificate.is_none());
     }
 
     #[test]
@@ -194,6 +205,7 @@ mod tests {
         assert_eq!(&*config.db, "neo4j");
         assert_eq!(config.fetch_size, 200);
         assert_eq!(config.max_connections, 16);
+        assert!(config.client_certificate.is_none());
     }
 
     #[test]

lib/src/connection.rs

@@ -7,6 +7,9 @@ use crate::{
 };
 use bytes::{Bytes, BytesMut};
 use std::{mem, sync::Arc};
+use std::fs::File;
+use std::io::BufReader;
+use log::warn;
 use stream::ConnectionStream;
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt, BufStream},
@@ -19,7 +22,9 @@ use tokio_rustls::{
     },
     TlsConnector,
 };
+use tokio_rustls::client::TlsStream;
 use url::{Host, Url};
+use crate::auth::ClientCertificate;
 
 const MAX_CHUNK_SIZE: usize = 65_535 - mem::size_of::<u16>();
 
@@ -39,7 +44,13 @@ impl Connection {
 
         match info.encryption {
             Encryption::No => Self::new_unencrypted(stream, &info.user, &info.password).await,
-            Encryption::Tls => Self::new_tls(stream, &info.host, &info.user, &info.password).await,
+            Encryption::Tls => {
+                if let Some(certificate) = info.client_certificate.as_ref() {
+                    Self::new_tls_with_certificate(stream, &info.host, &info.user, &info.password, certificate).await
+                } else {
+                    Self::new_tls(stream, &info.host, &info.user, &info.password).await
+                }
+            },
         }
     }
 
@@ -53,9 +64,47 @@ impl Connection {
         user: &str,
         password: &str,
     ) -> Result<Connection> {
+        let root_cert_store = Self::build_cert_store();
+        let stream = Self::build_stream(stream, host, root_cert_store).await?;
+
+        Self::init(user, password, stream).await
+    }
+
+    async fn new_tls_with_certificate<T: AsRef<str>>(
+        stream: TcpStream,
+        host: &Host<T>,
+        user: &str,
+        password: &str,
+        certificate: &ClientCertificate,
+    ) -> Result<Connection> {
+        let mut root_cert_store = Self::build_cert_store();
+
+        let cert_file = File::open(certificate.cert_file.as_str())?;
+        let mut reader = BufReader::new(cert_file);
+        let certs = rustls_pemfile::certs(&mut reader);
+        for certificate in certs {
+            if let Ok(cert) = certificate {
+                root_cert_store.add(cert).unwrap();
+            }
+        }
+
+        let stream = Self::build_stream(stream, host, root_cert_store).await?;
+        Self::init(user, password, stream).await
+    }
+
+    fn build_cert_store() -> RootCertStore {
         let mut root_cert_store = RootCertStore::empty();
-        root_cert_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().map(ToOwned::to_owned));
+        if let Ok(certs) = rustls_native_certs::load_native_certs() {
+            for cert in certs {
+                root_cert_store.add(cert).unwrap();
+            }
+        } else {
+            warn!("Failed to load native certificates!");
+        }
+        root_cert_store
+    }
 
+    async fn build_stream<T: AsRef<str>>(stream: TcpStream, host: &Host<T>, root_cert_store: RootCertStore) -> Result<TlsStream<TcpStream>, Error> {
         let config = ClientConfig::builder()
             .with_root_certificates(root_cert_store)
             .with_no_client_auth();
@@ -71,8 +120,7 @@ impl Connection {
         };
 
         let stream = connector.connect(domain, stream).await?;
-
-        Self::init(user, password, stream).await
+        Ok(stream)
     }
 
     async fn init(
@@ -212,6 +260,7 @@ pub(crate) struct ConnectionInfo {
     host: Host<Arc<str>>,
     port: u16,
     encryption: Encryption,
+    client_certificate: Option<ClientCertificate>,
 }
 
 enum Encryption {
@@ -242,6 +291,13 @@ impl ConnectionInfo {
                 ));
                 Encryption::Tls
             }
+            "neo4j+ssc" => {
+                log::warn!(concat!(
+                    "This driver does not yet implement client-side routing. ",
+                    "It is possible that operations against a cluster (such as Aura) will fail."
+                ));
+                Encryption::Tls
+            }
             otherwise => return Err(Error::UnsupportedScheme(otherwise.to_owned())),
         };
 
@@ -255,8 +311,14 @@ impl ConnectionInfo {
             },
             port,
             encryption,
+            client_certificate: None
         })
     }
+
+    pub fn with_client_certificate(&mut self, certificate: &ClientCertificate) -> &Self {
+        self.client_certificate = Some(certificate.clone());
+        self
+    }
 }
 
 struct NeoUrl(Url);

lib/src/lib.rs

@@ -440,7 +440,9 @@ pub mod summary;
 mod txn;
 mod types;
 mod version;
+mod auth;
 
+pub use crate::auth::{ClientCertificate, ClientCertificateProvider, StaticClientCertificateProvider};
 pub use crate::config::{Config, ConfigBuilder, Database};
 pub use crate::errors::*;
 pub use crate::graph::{query, Graph};

lib/src/pool.rs

@@ -5,6 +5,7 @@ use crate::{
 };
 use deadpool::managed::{Manager, Metrics, Object, Pool, RecycleResult};
 use log::info;
+use crate::auth::{ClientCertificate};
 
 pub type ConnectionPool = Pool<ConnectionManager>;
 pub type ManagedConnection = Object<ConnectionManager>;
@@ -14,8 +15,11 @@ pub struct ConnectionManager {
 }
 
 impl ConnectionManager {
-    pub fn new(uri: &str, user: &str, password: &str) -> Result<Self> {
-        let info = ConnectionInfo::new(uri, user, password)?;
+    pub fn new(uri: &str, user: &str, password: &str, client_certificate: Option<&ClientCertificate>) -> Result<Self> {
+        let mut info = ConnectionInfo::new(uri, user, password)?;
+        if let Some(client_certificate) = client_certificate {
+            info.with_client_certificate(client_certificate);
+        }
         Ok(ConnectionManager { info })
     }
 }
@@ -35,7 +39,7 @@ impl Manager for ConnectionManager {
 }
 
 pub async fn create_pool(config: &Config) -> Result<ConnectionPool> {
-    let mgr = ConnectionManager::new(&config.uri, &config.user, &config.password)?;
+    let mgr = ConnectionManager::new(&config.uri, &config.user, &config.password, config.client_certificate.as_ref())?;
     info!(
         "creating connection pool with max size {}",
         config.max_connections