rumqttc: Enable native tls TlsConnector support.

rumqttc/CHANGELOG.md

@@ -23,6 +23,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * use `Login` to store credentials
 * Made `DisconnectProperties` struct public.
 * Replace `Vec<Option<u16>>` with `FixedBitSet` for managing packet ids of released QoS 2 publishes and incoming QoS 2 publishes in `MqttState`.
+* Accept `native_tls::TlsConnector` as input for `Transport::tls_with_config` and `Transport::wss_with_config`.
 
 ### Deprecated
 

rumqttc/Cargo.toml

@@ -64,6 +64,11 @@ name = "tls"
 path = "examples/tls.rs"
 required-features = ["use-rustls"]
 
+[[example]]
+name = "tls_native"
+path = "examples/tls_native.rs"
+required-features = ["use-native-tls"]
+
 [[example]]
 name = "tls2"
 path = "examples/tls2.rs"

rumqttc/examples/tls.rs

@@ -1,4 +1,4 @@
-//! Example of how to configure rumqttd to connect to a server using TLS and authentication.
+//! Example of how to configure rumqttc to connect to a server using TLS and authentication.
 use std::error::Error;
 
 use rumqttc::{AsyncClient, Event, Incoming, MqttOptions, Transport};

rumqttc/examples/tls_native.rs

@@ -0,0 +1,42 @@
+//! Example of how to configure rumqttc to connect to a server using TLS and authentication.
+use std::error::Error;
+
+use rumqttc::{AsyncClient, Event, Incoming, MqttOptions, Transport};
+use tokio_native_tls::native_tls;
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> Result<(), Box<dyn Error>> {
+    pretty_env_logger::init();
+    color_backtrace::install();
+
+    let mut mqttoptions = MqttOptions::new("test-1", "test.mosquitto.org", 8883);
+    mqttoptions.set_keep_alive(std::time::Duration::from_secs(5));
+
+    // Use native-tls to load root certificates from the operating system.
+    let mut builder = native_tls::TlsConnector::builder();
+    let pem = vec![1, 2, 3];
+    // let pem = include_bytes!("native-tls-cert.pem");
+    let cert = native_tls::Certificate::from_pem(&pem)?;
+    builder.add_root_certificate(cert);
+    let connector = builder.build()?;
+
+    mqttoptions.set_transport(Transport::tls_with_config(connector.into()));
+
+    let (_client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
+
+    loop {
+        match eventloop.poll().await {
+            Ok(Event::Incoming(Incoming::Publish(p))) => {
+                println!("Topic: {}, Payload: {:?}", p.topic, p.payload);
+            }
+            Ok(Event::Incoming(i)) => {
+                println!("Incoming = {i:?}");
+            }
+            Ok(Event::Outgoing(o)) => println!("Outgoing = {o:?}"),
+            Err(e) => {
+                println!("Error = {e:?}");
+                return Ok(());
+            }
+        }
+    }
+}

rumqttc/src/lib.rs

@@ -100,7 +100,10 @@ extern crate log;
 
 use std::fmt::{self, Debug, Formatter};
 
-#[cfg(any(feature = "use-rustls", feature = "websocket"))]
+#[cfg(any(
+    feature = "use-rustls",
+    feature = "websocket"
+))]
 use std::sync::Arc;
 
 use std::time::Duration;
@@ -145,6 +148,10 @@ use rustls_native_certs::load_native_certs;
 pub use state::{MqttState, StateError};
 #[cfg(any(feature = "use-rustls", feature = "use-native-tls"))]
 pub use tls::Error as TlsError;
+#[cfg(feature = "use-native-tls")]
+pub use tokio_native_tls;
+#[cfg(feature = "use-native-tls")]
+use tokio_native_tls::native_tls::TlsConnector;
 #[cfg(feature = "use-rustls")]
 pub use tokio_rustls;
 #[cfg(feature = "use-rustls")]
@@ -229,8 +236,8 @@ pub enum Transport {
     #[cfg(feature = "websocket")]
     #[cfg_attr(docsrs, doc(cfg(feature = "websocket")))]
     Ws,
-    #[cfg(all(feature = "use-rustls", feature = "websocket"))]
-    #[cfg_attr(docsrs, doc(cfg(all(feature = "use-rustls", feature = "websocket"))))]
+    #[cfg(all(any(feature = "use-rustls", feature = "use-native-tls"), feature = "websocket"))]
+    #[cfg_attr(docsrs, doc(cfg(all(any(feature = "use-rustls", feature = "use-native-tls"), feature = "websocket"))))]
     Wss(TlsConfiguration),
 }
 
@@ -301,8 +308,8 @@ impl Transport {
         Self::wss_with_config(config)
     }
 
-    #[cfg(all(feature = "use-rustls", feature = "websocket"))]
-    #[cfg_attr(docsrs, doc(cfg(all(feature = "use-rustls", feature = "websocket"))))]
+    #[cfg(all(any(feature = "use-rustls", feature = "use-native-tls"), feature = "websocket"))]
+    #[cfg_attr(docsrs, doc(cfg(all(any(feature = "use-rustls", feature = "use-native-tls"), feature = "websocket"))))]
     pub fn wss_with_config(tls_config: TlsConfiguration) -> Self {
         Self::Wss(tls_config)
     }
@@ -339,7 +346,11 @@ pub enum TlsConfiguration {
     /// Injected rustls ClientConfig for TLS, to allow more customisation.
     Rustls(Arc<ClientConfig>),
     #[cfg(feature = "use-native-tls")]
+    /// Use default native-tls configuration
     Native,
+    #[cfg(feature = "use-native-tls")]
+    /// Injected native-tls TlsConnector for TLS, to allow more customisation.
+    NativeConnector(TlsConnector),
 }
 
 #[cfg(feature = "use-rustls")]
@@ -364,6 +375,13 @@ impl From<ClientConfig> for TlsConfiguration {
     }
 }
 
+#[cfg(feature = "use-native-tls")]
+impl From<TlsConnector> for TlsConfiguration {
+    fn from(connector: TlsConnector) -> Self {
+        TlsConfiguration::NativeConnector(connector)
+    }
+}
+
 /// Provides a way to configure low level network connection configurations
 #[derive(Clone, Default)]
 pub struct NetworkOptions {

rumqttc/src/tls.rs

@@ -155,6 +155,7 @@ pub async fn native_tls_connector(
             connector_builder.build()?
         }
         TlsConfiguration::Native => native_tls::TlsConnector::new()?,
+        TlsConfiguration::NativeConnector(connector) => connector.to_owned(),
         #[allow(unreachable_patterns)]
         _ => unreachable!("This cannot be called for other TLS backends than Native TLS"),
     };
@@ -176,7 +177,9 @@ pub async fn tls_connect(
             Box::new(connector.connect(domain, tcp).await?)
         }
         #[cfg(feature = "use-native-tls")]
-        TlsConfiguration::Native | TlsConfiguration::SimpleNative { .. } => {
+        TlsConfiguration::Native
+        | TlsConfiguration::NativeConnector(_)
+        | TlsConfiguration::SimpleNative { .. } => {
             let connector = native_tls_connector(tls_config).await?;
             Box::new(connector.connect(addr, tcp).await?)
         }