Release/0.4.6 (#68)

* Add some unit tests to mqtt fetcher

* Update api doc

* update README and changelog + Bump version

* remove unneeded references

* cargo fmt

* add .vscode settings to enable all features

Author: toelo3

.gitignore

@@ -23,5 +23,3 @@ test_files/
 *.iws
 *.iml
 *.ipr
-
-.vscode/

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "rust-analyzer.cargo.features": "all"
+}

dsh_sdk/CHANGELOG.md

@@ -5,6 +5,17 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.4.6] - 2024-08-24
+### Added
+- Add token fetcher for DSH MQTT
+  - fetch token for MQTT
+  - cache tokens till expiration time
+  - support fetching tokens for multiple clients
+### Changed
+- Updated dependencies
+  - Lazy_static to 1.5
+
+
 ## [0.4.5] - 2024-07-24
 ### Added
 - Add additional/optional config for producers and consumer

dsh_sdk/Cargo.toml

@@ -9,7 +9,7 @@ license.workspace = true
 name = "dsh_sdk"
 readme = 'README.md'
 repository.workspace = true
-version = "0.4.5"
+version = "0.4.6"
 
 [package.metadata.docs.rs]
 all-features = true
@@ -21,7 +21,7 @@ dashmap = {version = "6.0", optional = true}
 http-body-util = { version = "0.1", optional = true }
 hyper = { version = "1.3", features = ["server", "http1"], optional = true }
 hyper-util = { version = "0.1", features = ["tokio"], optional = true }
-lazy_static = { version = "1.4", optional = true }
+lazy_static = { version = "1.5", optional = true }
 log = "0.4"
 pem = "3"
 prometheus = { version = "0.13", features = ["process"], optional = true }
@@ -33,10 +33,10 @@ serde_json = { version = "1.0", optional = true }
 sha2 = { version = "0.10", optional = true}
 thiserror = "1.0"
 tokio = { version = "^1.35", features = ["signal", "sync", "time", "macros"], optional = true }
-tokio-util = { version = "0.7", optional = true }
+tokio-util = { version = "0.7", default-features = false, optional = true }
 
 [features]
-default = ["bootstrap", "graceful_shutdown", "metrics", "rdkafka-ssl", "rest-token-fetcher", "mqtt-token-fetcher"]
+default = ["bootstrap", "graceful_shutdown", "metrics", "rdkafka-ssl"]
 
 bootstrap = ["rcgen", "serde_json", "reqwest"]
 metrics = ["prometheus", "hyper", "hyper-util", "http-body-util", "lazy_static", "tokio", "bytes"]

dsh_sdk/README.md

@@ -60,7 +60,8 @@ The following features are available in this library and can be enabled/disabled
 | **feature** | **default** | **Description** |
 |---|---|---|
 | `bootstrap` | &check; | Generate signed certificate and fetch datastreams info <br> Also makes certificates available, to be used as lowlevel API |
-| `rest-token-fetcher` | &check; | Fetch a token to use the DSH Rest API |
+| `rest-token-fetcher` | &cross; | Fetch tokens to use DSH Rest API |
+| `mqtt-token-fetcher` | &cross; | Fetch tokens to use DSH MQTT |
 | `metrics` | &check; | Enable (custom) metrics for your service |
 | `graceful_shutdown` | &check; | Create a signal handler for implementing a graceful shutdown |
 | `dlq` | &cross; | Dead Letter Queue implementation (experimental) |

dsh_sdk/src/dsh/mod.rs

@@ -1,4 +1,4 @@
-//! # DSH Properties
+//! # DSH
 //!
 //! This module contains logic to connect to Kafka on DSH and retreive all properties of your tenant.
 //!

dsh_sdk/src/error.rs

@@ -1,3 +1,5 @@
+//! Error types for the DSH SDK
+
 use thiserror::Error;
 
 #[derive(Error, Debug)]

dsh_sdk/src/mqtt_token_fetcher.rs

@@ -1,3 +1,6 @@
+//! # MQTT Token Fetcher
+//!
+//! `MqttTokenFetcher` is responsible for fetching and managing MQTT tokens for DSH.
 use std::{
     fmt::{Display, Formatter},
     sync::Mutex,
@@ -21,8 +24,9 @@ pub struct MqttTokenFetcher {
     tenant_name: String,
     rest_api_key: String,
     rest_token: Mutex<RestToken>,
+    rest_auth_url: String,
     mqtt_token: DashMap<String, MqttToken>, // Mapping from Client ID to MqttToken
-    platform: Platform,
+    mqtt_auth_url: String,
     //token_lifetime: Option<i32>, // TODO: Implement option of passing token lifetime to request token for specific duration
     // port: Port or connection_type: Connection // TODO: Platform provides two connection options, current implemetation only provides connecting over SSL, enable WebSocket too
 }
@@ -45,8 +49,9 @@ impl MqttTokenFetcher {
             tenant_name,
             rest_api_key,
             rest_token: Mutex::new(rest_token),
+            rest_auth_url: platform.endpoint_rest_token().to_string(),
             mqtt_token: DashMap::new(),
-            platform,
+            mqtt_auth_url: platform.endpoint_mqtt_token().to_string(),
         }
     }
     /// Retrieves an MQTT token for the specified client ID.
@@ -66,18 +71,20 @@ impl MqttTokenFetcher {
         client_id: &str,
         claims: Option<Vec<Claims>>,
     ) -> Result<MqttToken, DshError> {
-        let mut mqtt_token = self
-            .mqtt_token
-            .entry(client_id.to_string())
-            .or_insert(self.fetch_new_mqtt_token(client_id, claims.clone()).await?);
-
-        if !mqtt_token.is_valid() {
-            *mqtt_token = self
-                .fetch_new_mqtt_token(client_id, claims.clone())
-                .await
-                .unwrap()
-        };
-        Ok(mqtt_token.clone())
+        match self.mqtt_token.entry(client_id.to_string()) {
+            dashmap::Entry::Occupied(mut entry) => {
+                let mqtt_token = entry.get_mut();
+                if !mqtt_token.is_valid() {
+                    *mqtt_token = self.fetch_new_mqtt_token(client_id, claims).await?;
+                };
+                Ok(mqtt_token.clone())
+            }
+            dashmap::Entry::Vacant(entry) => {
+                let mqtt_token = self.fetch_new_mqtt_token(client_id, claims).await?;
+                entry.insert(mqtt_token.clone());
+                Ok(mqtt_token)
+            }
+        }
     }
     /// Fetches a new MQTT token from the platform.
     ///
@@ -94,7 +101,7 @@ impl MqttTokenFetcher {
 
         if !rest_token.is_valid() {
             *rest_token =
-                RestToken::get(&self.tenant_name, &self.rest_api_key, &self.platform).await?
+                RestToken::get(&self.tenant_name, &self.rest_api_key, &self.rest_auth_url).await?
         }
 
         let authorization_header = format!("Bearer {}", rest_token.raw_token);
@@ -103,7 +110,7 @@ impl MqttTokenFetcher {
         let payload = serde_json::to_value(&mqtt_token_request)?;
 
         let response = mqtt_token_request
-            .send(&self.platform, &authorization_header, &payload)
+            .send(&self.mqtt_auth_url, &authorization_header, &payload)
             .await?;
 
         MqttToken::new(response)
@@ -202,7 +209,7 @@ impl MqttTokenRequest {
 
     async fn send(
         &self,
-        platform: &Platform,
+        mqtt_auth_url: &str,
         authorization_header: &str,
         payload: &serde_json::Value,
     ) -> Result<String, DshError> {
@@ -215,7 +222,7 @@ impl MqttTokenRequest {
             .expect("Failed to build reqwest client");
 
         let response = reqwest_client
-            .post(platform.endpoint_mqtt_token())
+            .post(mqtt_auth_url)
             .header("Authorization", authorization_header)
             .json(payload)
             .send()
@@ -225,7 +232,7 @@ impl MqttTokenRequest {
             Ok(response.text().await?)
         } else {
             Err(DshError::DshCallError {
-                url: platform.endpoint_mqtt_token().to_string(),
+                url: mqtt_auth_url.to_string(),
                 status_code: response.status(),
                 error_body: response.text().await?,
             })
@@ -283,7 +290,7 @@ impl MqttToken {
             .duration_since(UNIX_EPOCH)
             .expect("SystemTime before UNIX EPOCH!")
             .as_secs() as i32;
-        self.exp >= current_unixtime - 5
+        self.exp >= current_unixtime + 5
     }
 }
 
@@ -292,10 +299,10 @@ impl MqttToken {
 #[serde(rename_all = "kebab-case")]
 struct RestTokenAttributes {
     gen: i64,
-    pub endpoint: String,
+    endpoint: String,
     iss: String,
-    pub claims: RestClaims,
-    pub exp: i64,
+    claims: RestClaims,
+    exp: i32,
     tenant_id: String,
 }
 
@@ -312,7 +319,7 @@ struct DatastreamsData {}
 #[derive(Serialize, Deserialize, Debug)]
 struct RestToken {
     raw_token: String,
-    exp: i64,
+    exp: i32,
 }
 
 impl RestToken {
@@ -327,8 +334,8 @@ impl RestToken {
     /// # Returns
     ///
     /// A Result containing the created `RestToken` or a `DshError`.
-    async fn get(tenant: &str, api_key: &str, env: &Platform) -> Result<RestToken, DshError> {
-        let raw_token = Self::fetch_token(tenant, api_key, env).await.unwrap();
+    async fn get(tenant: &str, api_key: &str, auth_url: &str) -> Result<RestToken, DshError> {
+        let raw_token = Self::fetch_token(tenant, api_key, auth_url).await.unwrap();
 
         let header_payload = extract_header_and_payload(&raw_token)?;
 
@@ -347,11 +354,11 @@ impl RestToken {
         let current_unixtime = SystemTime::now()
             .duration_since(UNIX_EPOCH)
             .expect("SystemTime before UNIX EPOCH!")
-            .as_secs() as i64;
-        self.exp >= current_unixtime - 5
+            .as_secs() as i32;
+        self.exp >= current_unixtime + 5
     }
 
-    async fn fetch_token(tenant: &str, api_key: &str, env: &Platform) -> Result<String, DshError> {
+    async fn fetch_token(tenant: &str, api_key: &str, auth_url: &str) -> Result<String, DshError> {
         let json_body = json!({"tenant": tenant});
 
         const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
@@ -364,7 +371,7 @@ impl RestToken {
 
         let rest_client = reqwest_client;
         let response = rest_client
-            .post(env.endpoint_rest_token())
+            .post(auth_url)
             .header("apikey", api_key)
             .json(&json_body)
             .send()
@@ -375,7 +382,7 @@ impl RestToken {
         match status {
             reqwest::StatusCode::OK => Ok(body_text),
             _ => Err(DshError::DshCallError {
-                url: env.endpoint_rest_api().to_string(),
+                url: auth_url.to_string(),
                 status_code: status,
                 error_body: body_text,
             }),
@@ -437,6 +444,33 @@ fn decode_base64(payload: &str) -> Result<Vec<u8>, DshError> {
 mod tests {
     use super::*;
 
+    fn create_valid_fetcher() -> MqttTokenFetcher {
+        let exp_time = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_secs() as i32
+            + 3600;
+        println!("exp_time: {}", exp_time);
+        let rest_token: RestToken = RestToken {
+            exp: exp_time as i32,
+            raw_token: "valid.token.payload".to_string(),
+        };
+        let mqtt_token = MqttToken {
+            exp: exp_time,
+            raw_token: "valid.token.payload".to_string(),
+        };
+        let mqtt_token_map = DashMap::new();
+        mqtt_token_map.insert("test_client".to_string(), mqtt_token.clone());
+        MqttTokenFetcher {
+            tenant_name: "test_tenant".to_string(),
+            rest_api_key: "test_api_key".to_string(),
+            rest_token: Mutex::new(rest_token),
+            rest_auth_url: "test_auth_url".to_string(),
+            mqtt_token: mqtt_token_map,
+            mqtt_auth_url: "test_auth_url".to_string(),
+        }
+    }
+
     #[tokio::test]
     async fn test_mqtt_token_fetcher_new() {
         let tenant_name = "test_tenant".to_string();
@@ -448,6 +482,13 @@ mod tests {
         assert!(fetcher.mqtt_token.is_empty());
     }
 
+    #[tokio::test]
+    async fn test_mqtt_token_fetcher_get_token() {
+        let fetcher = create_valid_fetcher();
+        let token = fetcher.get_token("test_client", None).await.unwrap();
+        assert_eq!(token.raw_token, "valid.token.payload");
+    }
+
     #[test]
     fn test_claims_new() {
         let resource = Resource::new(
@@ -492,18 +533,66 @@ mod tests {
 
         assert!(token.is_valid());
     }
+    #[test]
+    fn test_mqtt_token_is_invalid() {
+        let raw_token = "valid.token.payload".to_string();
+        let token = MqttToken {
+            exp: SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap()
+                .as_secs() as i32,
+            raw_token,
+        };
+
+        assert!(!token.is_valid());
+    }
 
     #[test]
     fn test_rest_token_is_valid() {
         let token = RestToken {
             exp: SystemTime::now()
                 .duration_since(UNIX_EPOCH)
                 .unwrap()
-                .as_secs() as i64
+                .as_secs() as i32
                 + 3600,
             raw_token: "valid.token.payload".to_string(),
         };
 
         assert!(token.is_valid());
     }
+
+    #[test]
+    fn test_rest_token_is_invalid() {
+        let token = RestToken {
+            exp: SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap()
+                .as_secs() as i32,
+            raw_token: "valid.token.payload".to_string(),
+        };
+
+        assert!(!token.is_valid());
+    }
+
+    #[test]
+    fn test_rest_token_default_is_invalid() {
+        let token = RestToken::default();
+
+        assert!(!token.is_valid());
+    }
+
+    #[test]
+    fn test_extract_header_and_payload() {
+        let raw = "header.payload.signature";
+        let result = extract_header_and_payload(raw).unwrap();
+        assert_eq!(result, "payload");
+
+        let raw = "header.payload";
+        let result = extract_header_and_payload(raw).unwrap();
+        assert_eq!(result, "payload");
+
+        let raw = "header";
+        let result = extract_header_and_payload(raw);
+        assert!(result.is_err());
+    }
 }

dsh_sdk/src/utils.rs

@@ -92,6 +92,8 @@ impl Platform {
 
     /// Get the endpoint for fetching DSH Rest Authentication Token
     ///
+    /// With this token you can authenticate for the mqtt token endpoint
+    ///
     /// It will return the endpoint for DSH Rest authentication token based on the platform
     pub fn endpoint_rest_token(&self) -> &str {
         match self {