Add ability to use rustls over native-tls

.github/workflows/test.yml

@@ -10,21 +10,74 @@ on:
 
 env:
   CARGO_TERM_COLOR: always
+  RUST_BACKTRACE: 1
 
 jobs:
   build:
-
     runs-on: ubuntu-latest
-
     services:
       clickhouse:
-        image: yandex/clickhouse-server
+        image: clickhouse/clickhouse-server
         ports:
           - 9000:9000
-
     steps:
       - uses: actions/checkout@v3
       - name: Build
         run: cargo build --verbose
       - name: Run tests
         run: cargo test --verbose
+
+  build-native-tls:
+    runs-on: ubuntu-latest
+    env:
+      # NOTE: not all tests "secure" aware, so let's define DATABASE_URL explicitly
+      # NOTE: sometimes for native-tls default connection_timeout (500ms) is not enough, interestingly that for rustls it is OK.
+      DATABASE_URL: "tcp://localhost:9440?compression=lz4&ping_timeout=2s&retry_timeout=3s&secure=true&skip_verify=true&connection_timeout=5s"
+    steps:
+      - uses: actions/checkout@v3
+      # NOTE:
+      # - we cannot use "services" because they are executed before the steps, i.e. repository checkout.
+      # - "job.container.network" is empty, hence "host"
+      # - github actions does not support YAML anchors (sigh)
+      - name: Run clickhouse-server
+        run: docker run
+          -v ./extras/ci/generate_certs.sh:/docker-entrypoint-initdb.d/generate_certs.sh
+          -v ./extras/ci/overrides.xml:/etc/clickhouse-server/config.d/overrides.xml
+          -e CH_SSL_CERTIFICATE=/etc/clickhouse-server/config.d/server.crt
+          -e CH_SSL_PRIVATE_KEY=/etc/clickhouse-server/config.d/server.key
+          --network host
+          --rm
+          --detach
+          --publish 9440:9440
+          clickhouse/clickhouse-server
+      - name: Build
+        run: cargo build --features tls-native-tls --verbose
+      - name: Run tests
+        run: cargo test --features tls-native-tls --verbose
+
+  build-rustls:
+    runs-on: ubuntu-latest
+    env:
+      # NOTE: not all tests "secure" aware, so let's define DATABASE_URL explicitly
+      DATABASE_URL: "tcp://localhost:9440?compression=lz4&ping_timeout=2s&retry_timeout=3s&secure=true&skip_verify=true"
+    steps:
+      - uses: actions/checkout@v3
+      # NOTE:
+      # - we cannot use "services" because they are executed before the steps, i.e. repository checkout.
+      # - "job.container.network" is empty, hence "host"
+      # - github actions does not support YAML anchors (sigh)
+      - name: Run clickhouse-server
+        run: docker run
+          -v ./extras/ci/generate_certs.sh:/docker-entrypoint-initdb.d/generate_certs.sh
+          -v ./extras/ci/overrides.xml:/etc/clickhouse-server/config.d/overrides.xml
+          -e CH_SSL_CERTIFICATE=/etc/clickhouse-server/config.d/server.crt
+          -e CH_SSL_PRIVATE_KEY=/etc/clickhouse-server/config.d/server.key
+          --network host
+          --rm
+          --detach
+          --publish 9440:9440
+          clickhouse/clickhouse-server
+      - name: Build
+        run: cargo build --features tls-rustls --verbose
+      - name: Run tests
+        run: cargo test --features tls-rustls --verbose

Cargo.toml

@@ -15,7 +15,10 @@ exclude = ["tests/*", "examples/*"]
 
 [features]
 default = ["tokio_io"]
-tls = ["tokio-native-tls", "native-tls"]
+_tls = [] # meta feature for the clickhouse-rs generic TLS code
+tls = ["tls-native-tls"] # backward compatibility
+tls-native-tls = ["tokio-native-tls", "native-tls", "_tls"]
+tls-rustls = ["tokio-rustls", "rustls", "rustls-pemfile", "webpki-roots", "_tls"]
 async_std = ["async-std"]
 tokio_io = ["tokio"]
 
@@ -67,6 +70,22 @@ optional = true
 version = "^0.3"
 optional = true
 
+[dependencies.rustls]
+version = "0.22.1"
+optional = true
+
+[dependencies.rustls-pemfile]
+version = "2.0"
+optional = true
+
+[dependencies.tokio-rustls]
+version = "0.25.0"
+optional = true
+
+[dependencies.webpki-roots]
+version = "*"
+optional = true
+
 [dependencies.chrono]
 version = "^0.4"
 default-features = false

examples/simple.rs

@@ -38,15 +38,15 @@ async fn execute(database_url: String) -> Result<(), Box<dyn Error>> {
     Ok(())
 }
 
-#[cfg(all(feature = "tokio_io", not(feature = "tls")))]
+#[cfg(all(feature = "tokio_io", not(feature = "_tls")))]
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
     let database_url =
         env::var("DATABASE_URL").unwrap_or_else(|_| "tcp://localhost:9000?compression=lz4".into());
     execute(database_url).await
 }
 
-#[cfg(all(feature = "tokio_io", feature = "tls"))]
+#[cfg(all(feature = "tokio_io", feature = "_tls"))]
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
     let database_url = env::var("DATABASE_URL")

extras/ci/generate_certs.sh

@@ -0,0 +1,7 @@
+#!/usr/bin/env bash
+
+crt=$CH_SSL_CERTIFICATE
+key=$CH_SSL_PRIVATE_KEY
+
+openssl req -subj "/CN=localhost" -new -newkey rsa:2048 -days 365 -nodes -x509 -keyout "$key" -out "$crt"
+chown clickhouse:clickhouse "$crt" "$key"

extras/ci/overrides.xml

@@ -0,0 +1,18 @@
+<clickhouse>
+    <openSSL>
+        <server>
+            <certificateFile from_env="CH_SSL_CERTIFICATE" replace="1"></certificateFile>
+            <privateKeyFile from_env="CH_SSL_PRIVATE_KEY" replace="1"></privateKeyFile>
+            <verificationMode>none</verificationMode>
+            <loadDefaultCAFile>true</loadDefaultCAFile>
+            <cacheSessions>true</cacheSessions>
+            <disableProtocols>sslv2,sslv3</disableProtocols>
+            <preferServerCiphers>true</preferServerCiphers>
+        </server>
+    </openSSL>
+    <tcp_port_secure>9440</tcp_port_secure>
+
+    <logger>
+        <console>1</console>
+    </logger>
+</clickhouse>

src/connecting_stream.rs

@@ -6,22 +6,35 @@ use std::{
 };
 
 use futures_util::future::{select_ok, BoxFuture, SelectOk, TryFutureExt};
-#[cfg(feature = "tls")]
+#[cfg(feature = "_tls")]
 use futures_util::FutureExt;
 
 #[cfg(feature = "async_std")]
 use async_std::net::TcpStream;
-#[cfg(feature = "tls")]
+#[cfg(feature = "tls-native-tls")]
 use native_tls::TlsConnector;
 #[cfg(feature = "tokio_io")]
 use tokio::net::TcpStream;
+#[cfg(feature = "tls-rustls")]
+use {
+    rustls::{
+        client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier},
+        crypto::{verify_tls12_signature, verify_tls13_signature},
+        pki_types::{CertificateDer, ServerName, UnixTime},
+        ClientConfig, DigitallySignedStruct, Error as TlsError, RootCertStore,
+    },
+    std::sync::Arc,
+    tokio_rustls::TlsConnector,
+};
 
 use pin_project::pin_project;
 use url::Url;
 
 use crate::{errors::ConnectionError, io::Stream as InnerStream, Options};
-#[cfg(feature = "tls")]
+#[cfg(feature = "tls-native-tls")]
 use tokio_native_tls::TlsStream;
+#[cfg(feature = "tls-rustls")]
+use tokio_rustls::client::TlsStream;
 
 type Result<T> = std::result::Result<T, ConnectionError>;
 
@@ -33,7 +46,7 @@ enum TcpState {
     Fail(Option<ConnectionError>),
 }
 
-#[cfg(feature = "tls")]
+#[cfg(feature = "_tls")]
 #[pin_project(project = TlsStateProj)]
 enum TlsState {
     Wait(#[pin] ConnectingFuture<TlsStream<TcpStream>>),
@@ -43,7 +56,7 @@ enum TlsState {
 #[pin_project(project = StateProj)]
 enum State {
     Tcp(#[pin] TcpState),
-    #[cfg(feature = "tls")]
+    #[cfg(feature = "_tls")]
     Tls(#[pin] TlsState),
 }
 
@@ -60,7 +73,7 @@ impl TcpState {
     }
 }
 
-#[cfg(feature = "tls")]
+#[cfg(feature = "_tls")]
 impl TlsState {
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<InnerStream>> {
         match self.project() {
@@ -81,7 +94,7 @@ impl State {
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<InnerStream>> {
         match self.project() {
             StateProj::Tcp(inner) => inner.poll(cx),
-            #[cfg(feature = "tls")]
+            #[cfg(feature = "_tls")]
             StateProj::Tls(inner) => inner.poll(cx),
         }
     }
@@ -91,7 +104,7 @@ impl State {
         State::Tcp(TcpState::Fail(Some(conn_error)))
     }
 
-    #[cfg(feature = "tls")]
+    #[cfg(feature = "_tls")]
     fn tls_host_err() -> Self {
         State::Tls(TlsState::Fail(Some(ConnectionError::TlsHostNotProvided)))
     }
@@ -100,7 +113,7 @@ impl State {
         State::Tcp(TcpState::Wait(socket))
     }
 
-    #[cfg(feature = "tls")]
+    #[cfg(feature = "_tls")]
     fn tls_wait(s: ConnectingFuture<TlsStream<TcpStream>>) -> Self {
         State::Tls(TlsState::Wait(s))
     }
@@ -112,6 +125,57 @@ pub(crate) struct ConnectingStream {
     state: State,
 }
 
+#[derive(Debug)]
+struct DummyTlsVerifier;
+
+#[cfg(feature = "tls-rustls")]
+impl ServerCertVerifier for DummyTlsVerifier {
+    fn verify_server_cert(
+        &self,
+        _end_entity: &CertificateDer<'_>,
+        _intermediates: &[CertificateDer<'_>],
+        _server_name: &ServerName<'_>,
+        _ocsp_response: &[u8],
+        _now: UnixTime,
+    ) -> std::result::Result<ServerCertVerified, TlsError> {
+        Ok(ServerCertVerified::assertion())
+    }
+
+    fn verify_tls12_signature(
+        &self,
+        message: &[u8],
+        cert: &CertificateDer<'_>,
+        dss: &DigitallySignedStruct,
+    ) -> std::result::Result<HandshakeSignatureValid, TlsError> {
+        verify_tls12_signature(
+            message,
+            cert,
+            dss,
+            &rustls::crypto::ring::default_provider().signature_verification_algorithms,
+        )
+    }
+
+    fn verify_tls13_signature(
+        &self,
+        message: &[u8],
+        cert: &CertificateDer<'_>,
+        dss: &DigitallySignedStruct,
+    ) -> std::result::Result<HandshakeSignatureValid, TlsError> {
+        verify_tls13_signature(
+            message,
+            cert,
+            dss,
+            &rustls::crypto::ring::default_provider().signature_verification_algorithms,
+        )
+    }
+
+    fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
+        rustls::crypto::ring::default_provider()
+            .signature_verification_algorithms
+            .supported_schemes()
+    }
+}
+
 impl ConnectingStream {
     #[allow(unused_variables)]
     pub(crate) fn new(addr: &Url, options: &Options) -> Self {
@@ -137,7 +201,7 @@ impl ConnectingStream {
 
                 let socket = select_ok(streams);
 
-                #[cfg(feature = "tls")]
+                #[cfg(feature = "_tls")]
                 {
                     if options.secure {
                         return ConnectingStream::new_tls_connection(addr, socket, options);
@@ -154,7 +218,7 @@ impl ConnectingStream {
         }
     }
 
-    #[cfg(feature = "tls")]
+    #[cfg(feature = "tls-native-tls")]
     fn new_tls_connection(
         addr: &Url,
         socket: SelectOk<ConnectingFuture<TcpStream>>,
@@ -185,6 +249,67 @@ impl ConnectingStream {
             }
         }
     }
+
+    #[cfg(feature = "tls-rustls")]
+    fn new_tls_connection(
+        addr: &Url,
+        socket: SelectOk<ConnectingFuture<TcpStream>>,
+        options: &Options,
+    ) -> Self {
+        match addr.host_str().map(|host| host.to_owned()) {
+            None => Self {
+                state: State::tls_host_err(),
+            },
+            Some(host) => {
+                let config = if options.skip_verify {
+                    ClientConfig::builder()
+                        .dangerous()
+                        .with_custom_certificate_verifier(Arc::new(DummyTlsVerifier))
+                        .with_no_client_auth()
+                } else {
+                    let mut cert_store = RootCertStore::empty();
+                    cert_store.extend(
+                        webpki_roots::TLS_SERVER_ROOTS
+                            .iter()
+                            .cloned()
+                    );
+                    if let Some(certificates) = options.certificate.clone() {
+                        for certificate in
+                            Into::<Vec<rustls::pki_types::CertificateDer<'static>>>::into(
+                                certificates,
+                            )
+                        {
+                            match cert_store.add(certificate) {
+                                Ok(_) => {},
+                                Err(err) => {
+                                    let err = io::Error::new(
+                                        io::ErrorKind::InvalidInput,
+                                        format!("Could not load certificate: {}.", err),
+                                    );
+                                    return Self { state: State::tcp_err(err) };
+                                },
+                            }
+                        }
+                    }
+                    ClientConfig::builder()
+                        .with_root_certificates(cert_store)
+                        .with_no_client_auth()
+                };
+                Self {
+                    state: State::tls_wait(Box::pin(async move {
+                        let (s, _) = socket.await?;
+                        let cx = TlsConnector::from(Arc::new(config));
+                        let host = ServerName::try_from(host)
+                            .map_err(|_| ConnectionError::TlsHostNotProvided)?;
+                        Ok(cx
+                            .connect(host, s)
+                            .await
+                            .map_err(|e| ConnectionError::IoError(e))?)
+                    })),
+                }
+            }
+        }
+    }
 }
 
 impl Future for ConnectingStream {