Repository: bytebeamio/rumqtt (language: rust)

Issue:
rumqttd: external auth should be async

currently, the external authentication function is synchronous:

https://github.com/bytebeamio/rumqtt/blob/0266b85bd5986f556b3eaedc806c964e906232b8/rumqttd/src/lib.rs#L46  

### Issue
this means we have to use _blocking_ functions, so lets say we have to sleep for some duration, we have to use `std::thread::sleep()`. Issue with this is that it will _block_ the whole thread, thus no client will be able to make any progress!

for examples:
c1 -> running fine, doing its job
c2 -> tries to connect, but auth handler gets blocked
c1 -> won't recv any incoming messages as server is blocked! and will have no clue why it isn't receiving anyting!
c2 -> on client side, it will get network timeout, but on server, it will still be blocked.

### Solution

Make external authentication asynchronous! e.g. 

```rust
pub type AuthHandler = Arc<
    dyn Fn(ClientId, AuthUser, AuthPass) -> Pin<Box<dyn std::future::Future<Output = bool> + Send>>
        + Send
        + Sync
        + 'static,
>;
```
 this will allow users to set async fns as auth handler and give them privilege to use non-blocking code for auth.

 note: as the type is complex, it would be nice to have a helper method like `set_auth_handler(_)` which would take the fn and wrap it in `Pin<Box<_>>` [ method can be implemented on `ConnectionSettings` / `ServerSettings` or both :) ], similar to how we implement `request_modifier` in rumqttc.
 
 ps: if someone still uses blocking code in async fn, the issue will still arise.


Discussion / hints:

Fix patch:
diff --git a/rumqttd/CHANGELOG.md b/rumqttd/CHANGELOG.md
index e429b0a72..bb77d2e7d 100644
--- a/rumqttd/CHANGELOG.md
+++ b/rumqttd/CHANGELOG.md
@@ -12,6 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Changed
 - Public re-export `Strategy` for shared subscriptions
 - Peer initiated disconnects logged as info rather than error.
+- External authentication function must be async
 - Update `tokio-rustls` to `0.25.0`, `rustls-webpki` to `0.102.1`, `tokio-native-tls` to `0.3.1` and
   `rust-pemfile` to `2.0.0`.
 
diff --git a/rumqttd/examples/external_auth.rs b/rumqttd/examples/external_auth.rs
index 66edeaf6c..f5d7cdfcd 100644
--- a/rumqttd/examples/external_auth.rs
+++ b/rumqttd/examples/external_auth.rs
@@ -1,7 +1,5 @@
 use rumqttd::{Broker, Config};
 
-use std::sync::Arc;
-
 fn main() {
     let builder = tracing_subscriber::fmt()
         .pretty()
@@ -23,18 +21,23 @@ fn main() {
     // for e.g. if you want it for [v4.1] server, you can do something like
     let server = config.v4.as_mut().and_then(|v4| v4.get_mut("1")).unwrap();
 
-    // set the external_auth field in ConnectionSettings
     // external_auth function / closure signature must be:
-    // Fn(ClientId, AuthUser, AuthPass) -> bool
+    // async fn(ClientId, AuthUser, AuthPass) -> bool
     // type for ClientId, AuthUser and AuthPass is String
-    server.connections.external_auth = Some(Arc::new(auth));
+    server.set_auth_handler(auth);
+
+    // or you can pass closure
+    // server.set_auth_handler(|_client_id, _username, _password| async {
+    //     // perform auth
+    //     true
+    // });
 
     let mut broker = Broker::new(config);
 
     broker.start().unwrap();
 }
 
-fn auth(_client_id: String, _username: String, _password: String) -> bool {
+async fn auth(_client_id: String, _username: String, _password: String) -> bool {
     // users can fetch data from DB or tokens and use them!
     // do the verification and return true if verified, else false
     true
diff --git a/rumqttd/src/lib.rs b/rumqttd/src/lib.rs
index ac5d4a872..3763bab57 100644
--- a/rumqttd/src/lib.rs
+++ b/rumqttd/src/lib.rs
@@ -1,6 +1,8 @@
 use std::fmt;
+use std::future::IntoFuture;
 use std::net::SocketAddr;
 use std::path::PathBuf;
+use std::pin::Pin;
 use std::sync::Arc;
 use std::{collections::HashMap, path::Path};
 
@@ -43,7 +45,11 @@ pub type Cursor = (u64, u64);
 pub type ClientId = String;
 pub type AuthUser = String;
 pub type AuthPass = String;
-pub type AuthHandler = Arc<dyn Fn(ClientId, AuthUser, AuthPass) -> bool + Send + Sync + 'static>;
+pub type AuthHandler = Arc<
+    dyn Fn(ClientId, AuthUser, AuthPass) -> Pin<Box<dyn std::future::Future<Output = bool> + Send>>
+        + Send
+        + Sync,
+>;
 
 #[derive(Debug, Default, Serialize, Deserialize, Clone)]
 pub struct Config {
@@ -112,6 +118,17 @@ pub struct ServerSettings {
     pub connections: ConnectionSettings,
 }
 
+impl ServerSettings {
+    pub fn set_auth_handler<F, O>(&mut self, auth_fn: F)
+    where
+        F: Fn(ClientId, AuthUser, AuthPass) -> O + Send + Sync + 'static,
+        O: IntoFuture<Output = bool> + 'static,
+        O::IntoFuture: Send,
+    {
+        self.connections.set_auth_handler(auth_fn)
+    }
+}
+
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct BridgeConfig {
     pub name: String,
@@ -132,11 +149,25 @@ pub struct ConnectionSettings {
     pub max_inflight_count: usize,
     pub auth: Option<HashMap<String, String>>,
     #[serde(skip)]
-    pub external_auth: Option<AuthHandler>,
+    external_auth: Option<AuthHandler>,
     #[serde(default)]
     pub dynamic_filters: bool,
 }
 
+impl ConnectionSettings {
+    pub fn set_auth_handler<F, O>(&mut self, auth_fn: F)
+    where
+        F: Fn(ClientId, AuthUser, AuthPass) -> O + Send + Sync + 'static,
+        O: IntoFuture<Output = bool> + 'static,
+        O::IntoFuture: Send,
+    {
+        self.external_auth = Some(Arc::new(move |client_id, username, password| {
+            let auth = auth_fn(client_id, username, password).into_future();
+            Box::pin(auth)
+        }));
+    }
+}
+
 impl fmt::Debug for ConnectionSettings {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("ConnectionSettings")
diff --git a/rumqttd/src/link/remote.rs b/rumqttd/src/link/remote.rs
index 56c75f5cd..7ecbeeab2 100644
--- a/rumqttd/src/link/remote.rs
+++ b/rumqttd/src/link/remote.rs
@@ -188,7 +188,7 @@ where
 
     Span::current().record("client_id", &connect.client_id);
 
-    handle_auth(config.clone(), login.as_ref(), &connect.client_id)?;
+    handle_auth(config.clone(), login.as_ref(), &connect.client_id).await?;
 
     // When keep_alive feature is disabled client can live forever, which is not good in
     // distributed broker context so currenlty we don't allow it.
@@ -213,7 +213,7 @@ where
     Ok(packet)
 }
 
-fn handle_auth(
+async fn handle_auth(
     config: Arc<ConnectionSettings>,
     login: Option<&Login>,
     client_id: &str,
@@ -236,7 +236,9 @@ fn handle_auth(
             client_id.to_owned(),
             username.to_owned(),
             password.to_owned(),
-        ) {
+        )
+        .await
+        {
             return Err(Error::InvalidAuth);
         }
 
@@ -284,23 +286,23 @@ mod tests {
         }
     }
 
-    #[test]
-    fn no_login_no_auth() {
+    #[tokio::test]
+    async fn no_login_no_auth() {
         let cfg = Arc::new(config());
-        let r = handle_auth(cfg, None, "");
+        let r = handle_auth(cfg, None, "").await;
         assert!(r.is_ok());
     }
 
-    #[test]
-    fn some_login_no_auth() {
+    #[tokio::test]
+    async fn some_login_no_auth() {
         let cfg = Arc::new(config());
         let login = login();
-        let r = handle_auth(cfg, Some(&login), "");
+        let r = handle_auth(cfg, Some(&login), "").await;
         assert!(r.is_ok());
     }
 
-    #[test]
-    fn login_matches_static_auth() {
+    #[tokio::test]
+    async fn login_matches_static_auth() {
         let login = login();
         let mut map = HashMap::<String, String>::new();
         map.insert(login.username.clone(), login.password.clone());
@@ -308,12 +310,12 @@ mod tests {
         let mut cfg = config();
         cfg.auth = Some(map);
 
-        let r = handle_auth(Arc::new(cfg), Some(&login), "");
+        let r = handle_auth(Arc::new(cfg), Some(&login), "").await;
         assert!(r.is_ok());
     }
 
-    #[test]
-    fn login_fails_static_no_external() {
+    #[tokio::test]
+    async fn login_fails_static_no_external() {
         let login = login();
         let mut map = HashMap::<String, String>::new();
         map.insert("wrong".to_owned(), "wrong".to_owned());
@@ -321,53 +323,53 @@ mod tests {
         let mut cfg = config();
         cfg.auth = Some(map);
 
-        let r = handle_auth(Arc::new(cfg), Some(&login), "");
+        let r = handle_auth(Arc::new(cfg), Some(&login), "").await;
         assert!(r.is_err());
     }
 
-    #[test]
-    fn login_fails_static_matches_external() {
+    #[tokio::test]
+    async fn login_fails_static_matches_external() {
         let login = login();
 
         let mut map = HashMap::<String, String>::new();
         map.insert("wrong".to_owned(), "wrong".to_owned());
 
-        let dynamic = |_: String, _: String, _: String| -> bool { true };
+        let dynamic = |_: String, _: String, _: String| async { true };
 
         let mut cfg = config();
         cfg.auth = Some(map);
-        cfg.external_auth = Some(Arc::new(dynamic));
+        cfg.set_auth_handler(dynamic);
 
-        let r = handle_auth(Arc::new(cfg), Some(&login), "");
+        let r = handle_auth(Arc::new(cfg), Some(&login), "").await;
         assert!(r.is_ok());
     }
 
-    #[test]
-    fn login_fails_static_fails_external() {
+    #[tokio::test]
+    async fn login_fails_static_fails_external() {
         let login = login();
 
         let mut map = HashMap::<String, String>::new();
         map.insert("wrong".to_owned(), "wrong".to_owned());
 
-        let dynamic = |_: String, _: String, _: String| -> bool { false };
+        let dynamic = |_: String, _: String, _: String| async { false };
 
         let mut cfg = config();
         cfg.auth = Some(map);
-        cfg.external_auth = Some(Arc::new(dynamic));
+        cfg.set_auth_handler(dynamic);
 
-        let r = handle_auth(Arc::new(cfg), Some(&login), "");
+        let r = handle_auth(Arc::new(cfg), Some(&login), "").await;
         assert!(r.is_err());
     }
 
-    #[test]
-    fn external_auth_clousre_or_fnptr_type_check_or_fail_compile() {
-        let closure = |_: String, _: String, _: String| -> bool { false };
-        fn fnptr(_: String, _: String, _: String) -> bool {
+    #[tokio::test]
+    async fn external_auth_clousre_or_fnptr_type_check_or_fail_compile() {
+        let closure = |_: String, _: String, _: String| async { false };
+        async fn fnptr(_: String, _: String, _: String) -> bool {
             true
         }
 
         let mut cfg = config();
-        cfg.external_auth = Some(Arc::new(closure));
-        cfg.external_auth = Some(Arc::new(fnptr));
+        cfg.set_auth_handler(closure);
+        cfg.set_auth_handler(fnptr);
     }
 }


Test patch: