Support both openssl and rustls

.github/workflows/tls.yml

@@ -29,4 +29,4 @@ jobs:
       - name: Run tests
         env:
           FAKTORY_URL_SECURE: tcp://localhost:17419
-        run: cargo test --locked --features tls --test tls
+        run: cargo test --locked --features openssl,rustls --test tls

Cargo.toml

@@ -15,7 +15,9 @@ exclude = [".github", "docker", ".gitignore", "Makefile"]
 
 [features]
 default = []
-tls = ["dep:tokio-native-tls", "dep:pin-project"]
+tls = ["dep:pin-project"]
+openssl = ["tls", "dep:tokio-native-tls"]
+rustls = ["tls", "dep:tokio-rustls"]
 binaries = ["dep:clap", "tokio/macros"]
 ent = []
 
@@ -44,6 +46,7 @@ tokio = { version = "1.35.1", features = [
     "time",
 ] }
 tokio-native-tls = { version = "0.3.1", optional = true }
+tokio-rustls = { version = "0.25.0", optional = true }
 url = "2"
 
 [dev-dependencies]
@@ -53,7 +56,7 @@ x509-parser = "0.15.1"
 
 # to make -Zminimal-versions work
 [target.'cfg(any())'.dependencies]
-openssl = { version = "0.10.60", optional = true }
+openssl-crate = { package = "openssl", version = "0.10.60", optional = true }
 native-tls = { version = "0.2.4", optional = true }
 num-bigint = "0.4.2"
 oid-registry = "0.6.1"

Makefile

@@ -51,7 +51,7 @@ test/e2e:
 .PHONY: test/e2e/tls
 test/e2e/tls:
 	FAKTORY_URL_SECURE=tcp://${FAKTORY_HOST}:${FAKTORY_PORT_SECURE} \
-	cargo test --locked --features tls --test tls
+	cargo test --locked --features openssl,rustls --test tls
 
 .PHONY: test/load
 test/load:

src/lib.rs

@@ -91,5 +91,4 @@ pub mod ent {
 mod tls;
 
 #[cfg(feature = "tls")]
-#[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
-pub use tls::TlsStream;
+pub use tls::*;

src/tls/mod.rs

@@ -0,0 +1,12 @@
+#[cfg(feature = "openssl")]
+#[cfg_attr(docsrs, doc(cfg(feature = "openssl")))]
+/// Namespace for OpenSSL-powered [`TlsStream`](crate::openssl::TlsStream).
+///
+/// The underlying crate (`native-tls`) will use _SChannel_ on Windows,
+/// _SecureTransport_ on OSX, and _OpenSSL_ on other platforms.
+pub mod openssl;
+
+#[cfg(feature = "rustls")]
+#[cfg_attr(docsrs, doc(cfg(feature = "rustls")))]
+/// Namespace for Rustls-powered [`TlsStream`](crate::rustls::TlsStream).
+pub mod rustls;

src/tls.rs → src/tls/openssl.rs

@@ -19,7 +19,8 @@ use tokio_native_tls::{native_tls::TlsConnector, TlsConnector as AsyncTlsConnect
 ///
 /// ```no_run
 /// # tokio_test::block_on(async {
-/// use faktory::{Client, TlsStream};
+/// use faktory::Client;
+/// use faktory::openssl::TlsStream;
 /// let tls = TlsStream::connect(None).await.unwrap();
 /// let cl = Client::connect_with(tls, None).await.unwrap();
 /// # drop(cl);
@@ -31,7 +32,7 @@ pub struct TlsStream<S> {
     connector: AsyncTlsConnector,
     hostname: String,
     #[pin]
-    stream: NativeTlsStream<S>,
+    pub(crate) stream: NativeTlsStream<S>,
 }
 
 impl TlsStream<TokioTcpStream> {

src/tls/rustls.rs

@@ -0,0 +1,182 @@
+#[cfg(doc)]
+use crate::{Client, WorkerBuilder};
+
+use crate::{proto::utils, Error, Reconnect};
+use std::fmt::Debug;
+use std::io;
+use std::ops::{Deref, DerefMut};
+use std::sync::Arc;
+use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::net::TcpStream as TokioTcpStream;
+use tokio_rustls::client::TlsStream as RustlsStream;
+use tokio_rustls::rustls::{ClientConfig, RootCertStore};
+use tokio_rustls::TlsConnector;
+
+/// A reconnectable stream encrypted with TLS.
+///
+/// This can be used as an argument to [`WorkerBuilder::connect_with`] and [`Client::connect_with`] to
+/// connect to a TLS-secured Faktory server.
+///
+/// # Examples
+///
+/// ```no_run
+/// # tokio_test::block_on(async {
+/// use faktory::Client;
+/// use faktory::rustls::TlsStream;
+/// let tls = TlsStream::connect(None).await.unwrap();
+/// let cl = Client::connect_with(tls, None).await.unwrap();
+/// # drop(cl);
+/// # });
+/// ```
+///
+#[pin_project::pin_project]
+pub struct TlsStream<S> {
+    connector: TlsConnector,
+    hostname: &'static str,
+    #[pin]
+    pub(crate) stream: RustlsStream<S>,
+}
+
+impl TlsStream<TokioTcpStream> {
+    /// Create a new TLS connection over TCP.
+    ///
+    /// If `url` is not given, will use the standard Faktory environment variables. Specifically,
+    /// `FAKTORY_PROVIDER` is read to get the name of the environment variable to get the address
+    /// from (defaults to `FAKTORY_URL`), and then that environment variable is read to get the
+    /// server address. If the latter environment variable is not defined, the connection will be
+    /// made to
+    ///
+    /// ```text
+    /// tcp://localhost:7419
+    /// ```
+    ///
+    /// If `url` is given, but does not specify a port, it defaults to 7419.
+    ///
+    /// Internally creates a `ClientConfig` with an empty root certificates store and no client
+    /// authentication. Use [`with_client_config`](TlsStream::with_client_config)
+    /// or [`with_connector`](TlsStream::with_connector) for customized
+    /// `ClientConfig` and `TlsConnector` accordingly.
+    pub async fn connect(url: Option<&str>) -> Result<Self, Error> {
+        let conf = ClientConfig::builder()
+            .with_root_certificates(RootCertStore::empty())
+            .with_no_client_auth();
+        let con = TlsConnector::from(Arc::new(conf));
+        TlsStream::with_connector(con, url).await
+    }
+
+    /// Create a new asynchronous TLS connection over TCP using a non-default TLS configuration.
+    ///
+    /// See `connect` for details about the `url` parameter.
+    pub async fn with_client_config(conf: ClientConfig, url: Option<&str>) -> Result<Self, Error> {
+        let con = TlsConnector::from(Arc::new(conf));
+        TlsStream::with_connector(con, url).await
+    }
+
+    /// Create a new asynchronous TLS connection over TCP using a connector with a non-default TLS configuration.
+    ///
+    /// See `connect` for details about the `url` parameter.
+    pub async fn with_connector(connector: TlsConnector, url: Option<&str>) -> Result<Self, Error> {
+        let url = match url {
+            Some(url) => utils::url_parse(url),
+            None => utils::url_parse(&utils::get_env_url()),
+        }?;
+        let hostname = utils::host_from_url(&url);
+        let tcp_stream = TokioTcpStream::connect(&hostname).await?;
+        let hostname: &'static str = url.host_str().unwrap().to_string().leak();
+        Ok(TlsStream::new(tcp_stream, connector, hostname).await?)
+    }
+}
+
+impl<S> TlsStream<S>
+where
+    S: AsyncRead + AsyncWrite + Send + Unpin + Reconnect + Debug + 'static,
+{
+    /// Create a new asynchronous TLS connection on an existing stream.
+    ///
+    /// Internally creates a `ClientConfig` with an empty root certificates store and no client
+    /// authentication. Use [`new`](TlsStream::new) for a customized `TlsConnector`.
+    pub async fn default(stream: S, hostname: &'static str) -> io::Result<Self> {
+        let conf = ClientConfig::builder()
+            .with_root_certificates(RootCertStore::empty())
+            .with_no_client_auth();
+
+        Self::new(stream, TlsConnector::from(Arc::new(conf)), hostname).await
+    }
+
+    /// Create a new asynchronous TLS connection on an existing stream with a non-default TLS configuration.
+    pub async fn new(
+        stream: S,
+        connector: TlsConnector,
+        hostname: &'static str,
+    ) -> io::Result<Self> {
+        // let hostname: &'static str = hostname.to_string().leak();
+        let domain = hostname.try_into().expect("a valid DNS name or IP address");
+        let tls_stream = connector
+            .connect(domain, stream)
+            .await
+            .map_err(|e| io::Error::new(io::ErrorKind::ConnectionAborted, e))?;
+        Ok(TlsStream {
+            connector,
+            hostname,
+            stream: tls_stream,
+        })
+    }
+}
+
+#[async_trait::async_trait]
+impl<S> Reconnect for TlsStream<S>
+where
+    S: AsyncRead + AsyncWrite + Send + Unpin + Reconnect + Debug + 'static + Sync,
+{
+    async fn reconnect(&mut self) -> io::Result<Self> {
+        let stream = self.stream.get_mut().0.reconnect().await?;
+        Self::new(stream, self.connector.clone(), &self.hostname).await
+    }
+}
+
+impl<S> Deref for TlsStream<S> {
+    type Target = RustlsStream<S>;
+    fn deref(&self) -> &Self::Target {
+        &self.stream
+    }
+}
+
+impl<S> DerefMut for TlsStream<S> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.stream
+    }
+}
+
+impl<S: AsyncRead + AsyncWrite + Unpin> AsyncRead for TlsStream<S> {
+    fn poll_read(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> std::task::Poll<io::Result<()>> {
+        self.project().stream.poll_read(cx, buf)
+    }
+}
+
+impl<S: AsyncRead + AsyncWrite + Unpin> AsyncWrite for TlsStream<S> {
+    fn poll_write(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &[u8],
+    ) -> std::task::Poll<Result<usize, io::Error>> {
+        self.project().stream.poll_write(cx, buf)
+    }
+
+    fn poll_flush(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), io::Error>> {
+        self.project().stream.poll_flush(cx)
+    }
+
+    fn poll_shutdown(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), io::Error>> {
+        self.project().stream.poll_shutdown(cx)
+    }
+}