rumqttd: adding optional native-tls support.

By using the `use-native-tls` feature, this crate can now use tokio-native-tls vs tokio-rustls.

Changed:

* CI tests to incorporate disparate features (rustls vs native-tls)
* Made certain rustls includes to be conditional in rumqttd
* How errors are handled in main loop. Otherwise process loop exits silently.

Added:

* Notes to Readme about adding native-tls
* Added separate tls() function in rumqttd for native-tls
* Added use of tokio-native-tls

Author: jaredwolff

.github/workflows/features.yml

@@ -9,8 +9,8 @@ on:
 name: features
 
 jobs:
-  test:
-    name: Rumqtt build and test
+  test-rustls:
+    name: Rumqtt build and test 
     # runs-on: [self-hosted, linux, X64, rumqtt]
     runs-on: ubuntu-18.04
     steps:
@@ -21,6 +21,9 @@ jobs:
       - uses: actions-rs/cargo@v1
         with:
           command: test
-          args: --release --all-features
-
-
+          args: --release
+      # Test rumqttd native-tls support
+      - uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --release --features use-native-tls --manifest-path rumqttd/Cargo.toml --no-default-features

rumqttd/Cargo.toml

@@ -18,13 +18,15 @@ name = "rumqttd"
 path = "src/bin.rs"
 
 [features]
+default = ["use-rustls"]
 prof = ["pprof"]
+use-rustls = ["tokio-rustls"]
+use-native-tls = ["tokio-native-tls"]
 
 [dependencies]
 rumqttlog = { path = "../rumqttlog", version = "0.5"}
 mqttbytes = { path = "../mqttbytes", version = "0.2" }
 tokio = { version = "1.0", features = ["full"] }
-tokio-rustls = "0.22"
 serde = { version = "1", features = ["derive"] }
 log = "0.4"
 thiserror = "1"
@@ -36,5 +38,9 @@ warp = "0.3"
 futures-util = "0.3.8"
 pprof = { version = "0.4", features = ["flamegraph", "protobuf"], optional = true }
 
+# Optional
+tokio-rustls = { version = "0.22", optional = true }
+tokio-native-tls = { version = "0.3", optional = true }
+
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
 jemallocator = "0.3"

rumqttd/README.md

@@ -2,3 +2,32 @@
 
 [![crates.io page](https://img.shields.io/crates/v/rumqttd.svg)](https://crates.io/crates/rumqttd)
 [![docs.rs page](https://docs.rs/rumqttd/badge.svg)](https://docs.rs/rumqttd)
+
+## `native-tls` support
+
+This crate, by default uses the `tokio-rustls` crate. There's also support for the `tokio-native-tls` crate.
+Add it to your Cargo.toml like so:
+
+```
+rumqttd = { version = "0.5", default-features = false, features = ["use-native-tls"] }
+```
+
+Then in your config file make sure that you use the `pkcs12` options for your cert instead of `cert_path`, `key_path`, etc.
+
+```toml
+[rumqtt.servers.1]
+port = 8883
+pkcs12_path = "/root/identity.pfx"
+pkcs12_pass = ""
+# cert_path = "/root/issued/test.crt"
+# key_path = "/root/private/test.key"
+# ca_path = "/root/ca.crt"
+```
+
+You can generate the `.p12`/`.pfx` file using `openssl`:
+
+```
+openssl pkcs12 -export -out identity.pfx -inkey ~/pki/private/test.key -in ~/pki/issued/test.crt -certfile ~/pki/ca.crt
+```
+
+Make sure if you use a password it matches the entry in `pkcs12_pass`. If no password, use an empty string `""`.

rumqttd/src/lib.rs

@@ -2,9 +2,9 @@
 extern crate log;
 
 use serde::{Deserialize, Serialize};
-use std::{net::SocketAddr, sync::Arc};
 use std::time::Duration;
 use std::{io, thread};
+use std::{net::SocketAddr, sync::Arc};
 
 use mqttbytes::v4::Packet;
 use rumqttlog::*;
@@ -15,12 +15,25 @@ use crate::remotelink::RemoteLink;
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio::net::TcpListener;
 use tokio::{task, time};
+
+// All requirements for `rustls`
+#[cfg(feature = "use-rustls")]
 use tokio_rustls::rustls::internal::pemfile::{certs, rsa_private_keys};
+#[cfg(feature = "use-rustls")]
 use tokio_rustls::rustls::{
     AllowAnyAuthenticatedClient, NoClientAuth, RootCertStore, ServerConfig, TLSError,
 };
+#[cfg(feature = "use-rustls")]
 use tokio_rustls::TlsAcceptor;
 
+// All requirements for `native-tls`
+#[cfg(feature = "use-native-tls")]
+use std::io::Read;
+#[cfg(feature = "use-native-tls")]
+use tokio_native_tls::native_tls::Error as TLSError;
+#[cfg(feature = "use-native-tls")]
+use tokio_native_tls::{native_tls, TlsAcceptor};
+
 pub mod async_locallink;
 mod consolelink;
 mod locallink;
@@ -31,9 +44,11 @@ mod state;
 use crate::consolelink::ConsoleLink;
 pub use crate::locallink::{LinkError, LinkRx, LinkTx};
 use crate::network::Network;
+#[cfg(feature = "use-rustls")]
 use crate::Error::ServerKeyNotFound;
 use std::collections::HashMap;
 use std::fs::File;
+#[cfg(feature = "use-rustls")]
 use std::io::BufReader;
 
 #[derive(Debug, thiserror::Error)]
@@ -87,6 +102,10 @@ pub struct Config {
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct ServerSettings {
     pub listen: SocketAddr,
+    /// Used only for native-tls implementation
+    pub pkcs12_path: Option<String>,
+    /// Used only for native-tls implementation
+    pub pkcs12_pass: Option<String>,
     pub ca_path: Option<String>,
     pub cert_path: Option<String>,
     pub key_path: Option<String>,
@@ -215,6 +234,40 @@ impl Server {
         }
     }
 
+    #[cfg(feature = "use-native-tls")]
+    fn tls(&self) -> Result<Option<TlsAcceptor>, Error> {
+        match (
+            self.config.pkcs12_path.clone(),
+            self.config.pkcs12_pass.clone(),
+        ) {
+            (Some(cert), Some(password)) => {
+                // Get certificates
+                let cert_file = File::open(&cert);
+                let mut cert_file =
+                    cert_file.map_err(|_| Error::ServerCertNotFound(cert.clone()))?;
+
+                // Read cert into memory
+                let mut buf = Vec::new();
+                cert_file
+                    .read_to_end(&mut buf)
+                    .map_err(|_| Error::InvalidServerCert(cert.clone()))?;
+
+                // Get the identity
+                let identity = native_tls::Identity::from_pkcs12(&buf, &password)
+                    .map_err(|_| Error::InvalidServerCert(cert.clone()))?;
+
+                // Builder
+                let builder = native_tls::TlsAcceptor::builder(identity).build()?;
+
+                // Create acceptor
+                let acceptor = TlsAcceptor::from(builder);
+                Ok(Some(acceptor))
+            }
+            _ => Ok(None),
+        }
+    }
+
+    #[cfg(feature = "use-rustls")]
     fn tls(&self) -> Result<Option<TlsAcceptor>, Error> {
         let (certs, key) = match self.config.cert_path.clone() {
             Some(cert) => {
@@ -271,13 +324,35 @@ impl Server {
         let acceptor = self.tls()?;
         let max_incoming_size = config.max_payload_size;
 
-        info!("Waiting for connections on {}. Server = {}", self.config.listen, self.id);
+        info!(
+            "Waiting for connections on {}. Server = {}",
+            self.config.listen, self.id
+        );
         loop {
-            let (stream, addr) = listener.accept().await?;
+            // Await new network connection.
+            let (stream, addr) = match listener.accept().await {
+                Ok((s, r)) => (s, r),
+                Err(_e) => {
+                    error!("Unable to accept socket.");
+                    continue;
+                }
+            };
+
+            // Depending on TLS or not create a new Network
             let network = match &acceptor {
                 Some(acceptor) => {
                     info!("{}. Accepting TLS connection from: {}", count, addr);
-                    Network::new(acceptor.accept(stream).await?, max_incoming_size)
+                    let sock = match acceptor.accept(stream).await {
+                        Ok(s) => s,
+                        Err(_e) => {
+                            error!(
+                                "{} Unable to accept incoming TLS connection from {}",
+                                count, addr
+                            );
+                            continue;
+                        }
+                    };
+                    Network::new(sock, max_incoming_size)
                 }
                 None => {
                     info!("{}. Accepting TCP connection from: {}", count, addr);
@@ -289,6 +364,8 @@ impl Server {
 
             let config = config.clone();
             let router_tx = self.router_tx.clone();
+
+            // Spawn a new thread to handle this connection.
             task::spawn(async {
                 let connector = Connector::new(config, router_tx);
                 if let Err(e) = connector.new_connection(network).await {