feat: Added AUTH packet support for enhanced authentication.

rumqttc/CHANGELOG.md

@@ -15,6 +15,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * `set_session_expiry_interval` and `session_expiry_interval` methods on `MqttOptions`.
 * `Auth` packet as per MQTT5 standards
 * Allow configuring  the `nodelay` property of underlying TCP client with the `tcp_nodelay` field in `NetworkOptions`
+* `MqttOptions::set_auth_manager` that allows users to set their own authentication manager that implements the `AuthManager` trait.
+* `Client::reauth` that enables users to send `AUTH` packet for re-authentication purposes.
+
 
 ### Changed
 

rumqttc/Cargo.toml

@@ -21,6 +21,7 @@ use-rustls = ["dep:tokio-rustls", "dep:rustls-webpki", "dep:rustls-pemfile", "de
 use-native-tls = ["dep:tokio-native-tls", "dep:native-tls"]
 websocket = ["dep:async-tungstenite", "dep:ws_stream_tungstenite", "dep:http"]
 proxy = ["dep:async-http-proxy"]
+auth-scram = ["dep:scram"]
 
 [dependencies]
 futures-util = { version = "0.3", default-features = false, features = ["std", "sink"] }
@@ -50,6 +51,8 @@ url = { version = "2", default-features = false, optional = true }
 async-http-proxy = { version = "1.2.5", features = ["runtime-tokio", "basic-auth"], optional = true }
 tokio-stream = "0.1.15"
 fixedbitset = "0.5.7"
+#auth
+scram = { version = "0.6.0", optional = true }
 
 [dev-dependencies]
 bincode = "1.3.3"

rumqttc/examples/async_auth_oauth.rs

@@ -0,0 +1,64 @@
+use rumqttc::v5::mqttbytes::v5::AuthProperties;
+use rumqttc::v5::{mqttbytes::QoS, AsyncClient, MqttOptions};
+use rumqttc::{TlsConfiguration, Transport};
+use std::error::Error;
+use std::sync::Arc;
+use tokio::task;
+use tokio_rustls::rustls::ClientConfig;
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let pubsub_access_token = "";
+
+    let mut mqttoptions = MqttOptions::new("client1-session1", "MQTT hostname", 8883);
+    mqttoptions.set_authentication_method(Some("OAUTH2-JWT".to_string()));
+    mqttoptions.set_authentication_data(Some(pubsub_access_token.into()));
+
+    // Use rustls-native-certs to load root certificates from the operating system.
+    let mut root_cert_store = tokio_rustls::rustls::RootCertStore::empty();
+    root_cert_store.add_parsable_certificates(
+        rustls_native_certs::load_native_certs().expect("could not load platform certs"),
+    );
+
+    let client_config = ClientConfig::builder()
+        .with_root_certificates(root_cert_store)
+        .with_no_client_auth();
+
+    let transport = Transport::Tls(TlsConfiguration::Rustls(Arc::new(client_config.into())));
+
+    mqttoptions.set_transport(transport);
+
+    let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
+
+    task::spawn(async move {
+        client.subscribe("topic1", QoS::AtLeastOnce).await.unwrap();
+        client
+            .publish("topic1", QoS::AtLeastOnce, false, "hello world")
+            .await
+            .unwrap();
+
+        // Re-authentication test.
+        let props = AuthProperties {
+            method: Some("OAUTH2-JWT".to_string()),
+            data: Some(pubsub_access_token.into()),
+            reason: None,
+            user_properties: Vec::new(),
+        };
+
+        client.reauth(Some(props)).await.unwrap();
+    });
+
+    loop {
+        let notification = eventloop.poll().await;
+
+        match notification {
+            Ok(event) => println!("{:?}", event),
+            Err(e) => {
+                println!("Error = {:?}", e);
+                break;
+            }
+        }
+    }
+
+    Ok(())
+}

rumqttc/examples/async_auth_scram.rs

@@ -0,0 +1,143 @@
+use bytes::Bytes;
+use flume::bounded;
+use rumqttc::v5::mqttbytes::{v5::AuthProperties, QoS};
+use rumqttc::v5::{AsyncClient, AuthManager, MqttOptions};
+#[cfg(feature = "auth-scram")]
+use scram::client::ServerFirst;
+#[cfg(feature = "auth-scram")]
+use scram::ScramClient;
+use std::error::Error;
+use std::sync::{Arc, Mutex};
+use tokio::task;
+
+#[derive(Debug)]
+struct ScramAuthManager<'a> {
+    user: &'a str,
+    password: &'a str,
+    #[cfg(feature = "auth-scram")]
+    scram: Option<ServerFirst<'a>>,
+}
+
+impl<'a> ScramAuthManager<'a> {
+    fn new(user: &'a str, password: &'a str) -> ScramAuthManager<'a> {
+        ScramAuthManager {
+            user,
+            password,
+            #[cfg(feature = "auth-scram")]
+            scram: None,
+        }
+    }
+
+    fn auth_start(&mut self) -> Result<Option<Bytes>, String> {
+        #[cfg(feature = "auth-scram")]
+        {
+            let scram = ScramClient::new(self.user, self.password, None);
+            let (scram, client_first) = scram.client_first();
+            self.scram = Some(scram);
+
+            Ok(Some(client_first.into()))
+        }
+
+        #[cfg(not(feature = "auth-scram"))]
+        Ok(Some("client first message".into()))
+    }
+}
+
+impl<'a> AuthManager for ScramAuthManager<'a> {
+    fn auth_continue(
+        &mut self,
+        auth_method: Option<String>,
+        auth_data: Option<Bytes>,
+    ) -> Result<Option<Bytes>, String> {
+        #[cfg(feature = "auth-scram")]
+        {
+            // Check if the authentication method is SCRAM-SHA-256
+            if auth_method.unwrap() != "SCRAM-SHA-256" {
+                return Err("Invalid authentication method".to_string());
+            }
+
+            if self.scram.is_none() {
+                return Err("Invalid state".to_string());
+            }
+
+            let scram = self.scram.take().unwrap();
+
+            let auth_data = String::from_utf8(auth_data.unwrap().to_vec()).unwrap();
+
+            // Process the server first message and reassign the SCRAM state.
+            let scram = match scram.handle_server_first(&auth_data) {
+                Ok(scram) => scram,
+                Err(e) => return Err(e.to_string()),
+            };
+
+            // Get the client final message and reassign the SCRAM state.
+            let (_, client_final) = scram.client_final();
+
+            Ok(Some(client_final.into()))
+        }
+
+        #[cfg(not(feature = "auth-scram"))]
+        Ok(Some("client final message".into()))
+    }
+}
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let mut authmanager = ScramAuthManager::new("user1", "123456");
+    let client_first = authmanager.auth_start().unwrap();
+    let authmanager = Arc::new(Mutex::new(authmanager));
+
+    let mut mqttoptions = MqttOptions::new("auth_test", "127.0.0.1", 1883);
+    mqttoptions.set_authentication_method(Some("SCRAM-SHA-256".to_string()));
+    mqttoptions.set_authentication_data(client_first);
+    mqttoptions.set_auth_manager(authmanager.clone());
+    let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
+
+    let (tx, rx) = bounded(1);
+
+    task::spawn(async move {
+        client
+            .subscribe("rumqtt_auth/topic", QoS::AtLeastOnce)
+            .await
+            .unwrap();
+        client
+            .publish("rumqtt_auth/topic", QoS::AtLeastOnce, false, "hello world")
+            .await
+            .unwrap();
+
+        // Wait for the connection to be established.
+        rx.recv_async().await.unwrap();
+
+        // Reauthenticate using SCRAM-SHA-256
+        let client_first = authmanager.clone().lock().unwrap().auth_start().unwrap();
+        let properties = AuthProperties {
+            method: Some("SCRAM-SHA-256".to_string()),
+            data: client_first,
+            reason: None,
+            user_properties: Vec::new(),
+        };
+        client.reauth(Some(properties)).await.unwrap();
+    });
+
+    loop {
+        let notification = eventloop.poll().await;
+
+        match notification {
+            Ok(event) => {
+                println!("Event = {:?}", event);
+                match event {
+                    rumqttc::v5::Event::Incoming(rumqttc::v5::Incoming::ConnAck(_)) => {
+                        tx.send_async("Connected").await.unwrap();
+                    }
+                    _ => {}
+                }
+            }
+            Err(e) => {
+                println!("Error = {:?}", e);
+                break;
+            }
+        }
+    }
+
+    Ok(())
+}