Repository: AlexPikalov/cdrs (language: rust)

Issue:
rustls alternative ssl feature
This would be a good alternative on non-linux platforms


Discussion / hints:
@DoumanAsh 

You are absolutely right. It would be great to start using native Rust rustls crate.
Do you think you'd like to give it a shot and prepare some PR? Let me know if you need any help with it.

Ok, do you want it as separate feature fom `ssl`?
Also, do you want ot allow both `ssl` and `rustls` to be present at the same time?
I think another feature like `native-ssl` would be a good point to start. Later on, when the implementation will be stabilised we ca deprecate existing `ssl` feature based on `open-ssl` crate in favour of `native-ssl` based on `rustls`.

To start the development you can easily check what is included into `ssl` feature and implement similar structures leveraging `rustls`.

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 473778b..dd619aa 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -15,6 +15,7 @@ license = "MIT/Apache-2.0"
 [features]
 default = ["v4"]
 ssl = ["openssl"]
+rust-tls = ["rustls", "webpki"]
 v3 = []
 v4 = []
 # enable v5 feature when it's actually implemented
@@ -34,6 +35,12 @@ rand = "0.4.1"
 snap = "0.2.3"
 time = "0.1.38"
 uuid = "0.8.1"
+webpki = { version = "0.21", optional = true }
+
+[dependencies.rustls]
+version = "0.17"
+optional = true
+default-features = false
 
 [dev-dependencies]
 env_logger = "0.4.3"
diff --git a/src/cluster/config_rustls.rs b/src/cluster/config_rustls.rs
new file mode 100644
index 0000000..2665cc9
--- /dev/null
+++ b/src/cluster/config_rustls.rs
@@ -0,0 +1,121 @@
+use core::time::Duration;
+use std::sync::Arc;
+use std::net;
+
+use crate::authenticators::Authenticator;
+
+/// Cluster configuration that holds per node SSL configs
+pub struct ClusterRustlsConfig<A: Authenticator + Sized>(pub Vec<NodeRustlsConfig<A>>);
+
+/// Single node SSL connection config.
+#[derive(Clone)]
+pub struct NodeRustlsConfig<A> {
+    pub addr: net::SocketAddr,
+    pub dns_name: webpki::DNSName,
+    pub authenticator: A,
+    pub max_size: u32,
+    pub min_idle: Option<u32>,
+    pub max_lifetime: Option<Duration>,
+    pub idle_timeout: Option<Duration>,
+    pub connection_timeout: Duration,
+    pub config: Arc<rustls::ClientConfig>,
+}
+
+/// Builder structure that helps to configure SSL connection for node.
+pub struct NodeRustlsConfigBuilder<A> {
+    addr: net::SocketAddr,
+    dns_name: webpki::DNSName,
+    authenticator: A,
+    max_size: Option<u32>,
+    min_idle: Option<u32>,
+    max_lifetime: Option<Duration>,
+    idle_timeout: Option<Duration>,
+    connection_timeout: Option<Duration>,
+    config: Arc<rustls::ClientConfig>,
+}
+
+impl<A: Authenticator + Sized> NodeRustlsConfigBuilder<A> {
+    const DEFAULT_MAX_SIZE: u32 = 10;
+    const DEFAULT_CONNECTION_TIMEOUT: Duration = Duration::from_secs(30);
+
+    /// `NodeRustlsConfigBuilder` constructor function. It receives
+    /// * node socket address
+    /// * authenticator
+    pub fn new(addr: net::SocketAddr, dns_name: webpki::DNSName, authenticator: A, config: Arc<rustls::ClientConfig>) -> Self {
+        NodeRustlsConfigBuilder {
+            addr,
+            dns_name,
+            authenticator,
+            max_size: None,
+            min_idle: None,
+            max_lifetime: None,
+            idle_timeout: None,
+            connection_timeout: None,
+            config,
+        }
+    }
+
+    /// Sets the maximum number of connections managed by the pool.
+    /// Defaults to 10.
+    pub fn max_size(mut self, size: u32) -> Self {
+        self.max_size = Some(size);
+        self
+    }
+
+    /// Sets the minimum idle connection count maintained by the pool.
+    /// If set, the pool will try to maintain at least this many idle
+    /// connections at all times, while respecting the value of `max_size`.
+    /// Defaults to None (equivalent to the value of `max_size`).
+    pub fn min_idle(mut self, min_idle: Option<u32>) -> Self {
+        self.min_idle = min_idle;
+        self
+    }
+
+    /// Sets the maximum lifetime of connections in the pool.
+    /// If set, connections will be closed after existing for at most 30 seconds beyond this duration.
+    /// If a connection reaches its maximum lifetime while checked out it will be closed when it is returned to the pool.
+    /// Defaults to 30 minutes.
+    pub fn max_lifetime(mut self, max_lifetime: Option<Duration>) -> Self {
+        self.max_lifetime = max_lifetime;
+        self
+    }
+
+    /// Sets the idle timeout used by the pool.
+    /// If set, connections will be closed after sitting idle for at most 30 seconds beyond this duration.
+    /// Defaults to 10 minutes.
+    pub fn idle_timeout(mut self, idle_timeout: Option<Duration>) -> Self {
+        self.idle_timeout = idle_timeout;
+        self
+    }
+
+    /// Sets the connection timeout used by the pool.
+    /// Defaults to 30 seconds.
+    pub fn connection_timeout(mut self, connection_timeout: Duration) -> Self {
+        self.connection_timeout = Some(connection_timeout);
+        self
+    }
+
+    /// Sets new authenticator.
+    pub fn authenticator(mut self, authenticator: A) -> Self {
+        self.authenticator = authenticator;
+        self
+    }
+
+    /// Finalizes building process and returns `NodeRustlsConfig`
+    pub fn build(self) -> NodeRustlsConfig<A> {
+        NodeRustlsConfig {
+            addr: self.addr,
+            dns_name: self.dns_name,
+            authenticator: self.authenticator,
+            config: self.config,
+
+            max_size: self.max_size.unwrap_or(Self::DEFAULT_MAX_SIZE),
+            min_idle: self.min_idle,
+            max_lifetime: self.max_lifetime,
+            idle_timeout: self.idle_timeout,
+            connection_timeout: self
+                .connection_timeout
+                .unwrap_or(Self::DEFAULT_CONNECTION_TIMEOUT),
+        }
+    }
+}
diff --git a/src/cluster/mod.rs b/src/cluster/mod.rs
index e9204ab..5d16f30 100644
--- a/src/cluster/mod.rs
+++ b/src/cluster/mod.rs
@@ -3,22 +3,32 @@ use std::cell;
 
 #[cfg(feature = "ssl")]
 mod config_ssl;
+#[cfg(feature = "rust-tls")]
+mod config_rustls;
 mod config_tcp;
 mod generic_connection_pool;
 mod pager;
 pub mod session;
 #[cfg(feature = "ssl")]
 mod ssl_connection_pool;
+#[cfg(feature = "rust-tls")]
+mod rustls_connection_pool;
 mod tcp_connection_pool;
 
 #[cfg(feature = "ssl")]
 pub use crate::cluster::config_ssl::{ClusterSslConfig, NodeSslConfig, NodeSslConfigBuilder};
+#[cfg(feature = "rust-tls")]
+pub use crate::cluster::config_rustls::{ClusterRustlsConfig, NodeRustlsConfig, NodeRustlsConfigBuilder};
 pub use crate::cluster::config_tcp::{ClusterTcpConfig, NodeTcpConfig, NodeTcpConfigBuilder};
 pub use crate::cluster::pager::{PagerState, QueryPager, SessionPager};
 #[cfg(feature = "ssl")]
 pub use crate::cluster::ssl_connection_pool::{
     new_ssl_pool, SslConnectionPool, SslConnectionsManager,
 };
+#[cfg(feature = "rust-tls")]
+pub use crate::cluster::rustls_connection_pool::{
+    new_rustls_pool, RustlsConnectionPool, RustlsConnectionsManager,
+};
 pub use crate::cluster::tcp_connection_pool::{
     new_tcp_pool, startup, TcpConnectionPool, TcpConnectionsManager,
 };
diff --git a/src/cluster/rustls_connection_pool.rs b/src/cluster/rustls_connection_pool.rs
new file mode 100644
index 0000000..a916323
--- /dev/null
+++ b/src/cluster/rustls_connection_pool.rs
@@ -0,0 +1,84 @@
+use r2d2::{Builder, ManageConnection};
+
+use std::net;
+use std::sync::Arc;
+use std::io::Write;
+use std::error::Error;
+use core::cell::RefCell;
+
+use crate::cluster::{startup, NodeRustlsConfig};
+use crate::authenticators::Authenticator;
+use crate::cluster::ConnectionPool;
+use crate::compression::Compression;
+use crate::frame::parser::parse_frame;
+use crate::frame::{Frame, IntoBytes};
+use crate::transport::{CDRSTransport, TransportRustls};
+use crate::error;
+
+pub type RustlsConnectionPool<A> = ConnectionPool<RustlsConnectionsManager<A>>;
+
+/// `r2d2::Pool` of SSL-based CDRS connections.
+///
+/// Used internally for SSL Session for holding connections to a specific Cassandra node.
+pub fn new_rustls_pool<A: Authenticator + Send + Sync + 'static>(node_config: NodeRustlsConfig<A>) -> error::Result<RustlsConnectionPool<A>> {
+    let manager = RustlsConnectionsManager::new(
+        node_config.addr,
+        node_config.dns_name,
+        node_config.config,
+        node_config.authenticator,
+    );
+
+    let pool = Builder::new()
+        .max_size(node_config.max_size)
+        .min_idle(node_config.min_idle)
+        .max_lifetime(node_config.max_lifetime)
+        .idle_timeout(node_config.idle_timeout)
+        .connection_timeout(node_config.connection_timeout)
+        .build(manager)
+        .map_err(|err| error::Error::from(err.description()))?;
+
+    Ok(RustlsConnectionPool::new(pool, node_config.addr))
+}
+
+/// `r2d2` connection manager.
+pub struct RustlsConnectionsManager<A> {
+    addr: net::SocketAddr,
+    dns_name: webpki::DNSName,
+    config: Arc<rustls::ClientConfig>,
+    auth: A,
+}
+
+impl<A> RustlsConnectionsManager<A> {
+    #[inline]
+    pub fn new(addr: net::SocketAddr, dns_name: webpki::DNSName, config: Arc<rustls::ClientConfig>, auth: A) -> Self {
+        Self {
+            addr,
+            dns_name,
+            config,
+            auth,
+        }
+    }
+}
+
+impl<A: Authenticator + 'static + Send + Sync> ManageConnection for RustlsConnectionsManager<A> {
+    type Connection = RefCell<TransportRustls>;
+    type Error = error::Error;
+
+    fn connect(&self) -> Result<Self::Connection, Self::Error> {
+        let transport = RefCell::new(TransportRustls::new(self.addr, self.dns_name.clone(), self.config.clone())?);
+        startup(&transport, &self.auth)?;
+
+        Ok(transport)
+    }
+
+    fn is_valid(&self, conn: &mut Self::Connection) -> Result<(), Self::Error> {
+        let options_frame = Frame::new_req_options().into_cbytes();
+        conn.borrow_mut().write(options_frame.as_slice())?;
+
+        parse_frame(conn, &Compression::None {}).map(|_| ())
+    }
+
+    fn has_broken(&self, conn: &mut Self::Connection) -> bool {
+        !conn.borrow().is_alive()
+    }
+}
diff --git a/src/lib.rs b/src/lib.rs
index 0f0074d..d5186ad 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -10,6 +10,8 @@ extern crate log;
 extern crate lz4_compress;
 #[cfg(feature = "ssl")]
 extern crate openssl;
+#[cfg(feature = "rust-tls")]
+extern crate rustls;
 extern crate r2d2;
 extern crate rand;
 extern crate time;
diff --git a/src/transport.rs b/src/transport.rs
index 10f044b..f0264b7 100644
--- a/src/transport.rs
+++ b/src/transport.rs
@@ -18,6 +18,7 @@ use std::io::{Read, Write};
 use std::net;
 use std::net::TcpStream;
 use std::time::Duration;
+use std::sync::Arc;
 
 // TODO [v 2.x.x]: CDRSTransport: ... + BufReader + ButWriter + ...
 ///General CDRS transport trait. Both [`TranportTcp`][transportTcp]
@@ -108,6 +109,72 @@ impl CDRSTransport for TransportTcp {
     }
 }
 
+#[cfg(feature = "rust-tls")]
+pub struct TransportRustls {
+    inner: rustls::StreamOwned<rustls::ClientSession, net::TcpStream>,
+    config: Arc<rustls::ClientConfig>,
+    addr: net::SocketAddr,
+    dns_name: webpki::DNSName,
+}
+
+#[cfg(feature = "rust-tls")]
+impl TransportRustls {
+    ///Creates new instance with provided configuration
+    pub fn new(addr: net::SocketAddr, dns_name: webpki::DNSName, config: Arc<rustls::ClientConfig>) -> io::Result<Self> {
+        let socket = std::net::TcpStream::connect(addr)?;
+        let session = rustls::ClientSession::new(&config, dns_name.as_ref());
+
+        Ok(Self {
+            inner: rustls::StreamOwned::new(session, socket),
+            config,
+            addr,
+            dns_name,
+        })
+    }
+}
+
+#[cfg(feature = "rust-tls")]
+impl Read for TransportRustls {
+    #[inline]
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        self.inner.read(buf)
+    }
+}
+
+#[cfg(feature = "rust-tls")]
+impl Write for TransportRustls {
+    #[inline]
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.inner.write(buf)
+    }
+
+    #[inline]
+    fn flush(&mut self) -> io::Result<()> {
+        self.inner.flush()
+    }
+}
+
+#[cfg(feature = "rust-tls")]
+impl CDRSTransport for TransportRustls {
+    #[inline]
+    fn try_clone(&self) -> io::Result<Self> {
+        Self::new(self.addr, self.dns_name.clone(), self.config.clone())
+    }
+
+    fn close(&mut self, close: net::Shutdown) -> io::Result<()> {
+        self.inner.get_mut().shutdown(close)
+    }
+
+    fn set_timeout(&mut self, dur: Option<Duration>) -> io::Result<()> {
+        self.inner.get_mut().set_read_timeout(dur)?;
+        self.inner.get_mut().set_write_timeout(dur)
+    }
+
+    fn is_alive(&self) -> bool {
+        self.inner.get_ref().peer_addr().is_ok()
+    }
+}
+
 /// ***********************************
 #[cfg(feature = "ssl")]
 pub struct TransportTls {


Test patch: