Merge pull request #341 from bittcrafter/dev/0.18.0

fix: improve MQTT protocol validation and WebSocket subprotocol support #340

Author: bittcrafter

Cargo.toml

@@ -64,8 +64,8 @@ rust-version = "1.85.0"
 
 [workspace.dependencies]
 rmqtt = "0.18.0"
-rmqtt-codec = "0.2"
-rmqtt-net = "0.3"
+rmqtt-codec = "0.2.1"
+rmqtt-net = "0.3.2"
 rmqtt-conf = "0.3"
 rmqtt-macros = "0.1"
 rmqtt-utils = "0.1"

rmqtt-codec/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rmqtt-codec"
-version = "0.2.0"
+version = "0.2.1"
 description = "MQTT protocol codec implementation with multi-version support and version negotiation"
 repository = "https://github.com/rmqtt/rmqtt/tree/master/rmqtt-codec"
 edition.workspace = true

rmqtt-codec/src/version.rs

@@ -55,11 +55,22 @@ impl Decoder for VersionCodec {
                     );
 
                     // Validate protocol name matches MQTT spec
-                    ensure!(
-                        (protocol_len == 4 && &src[consumed + 2..consumed + 6] == MQTT)
-                            || (protocol_len == 6 && &src[consumed + 2..consumed + 8] == MQISDP),
-                        DecodeError::InvalidProtocol
-                    );
+                    if protocol_len == 4 {
+                        //for mqtt 3.1.1 or 5.0
+                        if &src[consumed + 2..consumed + 6] != MQTT {
+                            return Err(DecodeError::InvalidProtocol);
+                        }
+                    } else if protocol_len == 6 {
+                        //for mqtt 3.1
+                        if len <= consumed + 8 {
+                            return Ok(None);
+                        }
+                        if &src[consumed + 2..consumed + 8] != MQISDP {
+                            return Err(DecodeError::InvalidProtocol);
+                        }
+                    } else {
+                        return Err(DecodeError::InvalidProtocol);
+                    }
 
                     // Extract protocol level byte (position after protocol name)
                     match src[consumed + 2 + protocol_len as usize] {

rmqtt-net/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rmqtt-net"
-version = "0.3.1"
+version = "0.3.2"
 description = "Basic Implementation of MQTT Server"
 repository = "https://github.com/rmqtt/rmqtt/tree/master/rmqtt-net"
 edition.workspace = true

rmqtt-net/src/builder.rs

@@ -904,14 +904,26 @@ fn on_handshake(req: &Request, mut response: Response) -> std::result::Result<Re
     let mqtt_protocol = req
         .headers()
         .get("Sec-WebSocket-Protocol")
-        .ok_or_else(|| ErrorResponse::new(Some(PROTOCOL_ERROR.into())))?;
-    if mqtt_protocol != "mqtt" {
-        return Err(ErrorResponse::new(Some(PROTOCOL_ERROR.into())));
-    }
-    response.headers_mut().append(
-        "Sec-WebSocket-Protocol",
-        "mqtt".parse().map_err(|_| ErrorResponse::new(Some("InvalidHeaderValue".into())))?,
-    );
+        .ok_or_else(|| ErrorResponse::new(Some(PROTOCOL_ERROR.into())))?
+        .to_str()
+        .map_err(|_| ErrorResponse::new(Some(PROTOCOL_ERROR.into())))?;
+    match mqtt_protocol {
+        "mqtt" => {
+            response.headers_mut().append(
+                "Sec-WebSocket-Protocol",
+                "mqtt".parse().map_err(|_| ErrorResponse::new(Some("InvalidHeaderValue".into())))?,
+            );
+        }
+        "mqttv3.1" => {
+            response.headers_mut().append(
+                "Sec-WebSocket-Protocol",
+                "mqttv3.1".parse().map_err(|_| ErrorResponse::new(Some("InvalidHeaderValue".into())))?,
+            );
+        }
+        _ => {
+            return Err(ErrorResponse::new(Some(PROTOCOL_ERROR.into())));
+        }
+    }
     Ok(response)
 }