fix: validate subprotocol mqtt

mosquitto.conf

@@ -0,0 +1,3 @@
+listener 9001
+protocol websockets
+allow_anonymous true

rumqttc/examples/websocket.rs

@@ -11,11 +11,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
     pretty_env_logger::init();
 
     // port parameter is ignored when scheme is websocket
-    let mut mqttoptions = MqttOptions::new(
-        "clientId-aSziq39Bp3",
-        "ws://broker.mqttdashboard.com:8000/mqtt",
-        8000,
-    );
+    let mut mqttoptions: MqttOptions = MqttOptions::new("clientId-1", "ws://localhost:9001", 9001);
     mqttoptions.set_transport(Transport::Ws);
     mqttoptions.set_keep_alive(Duration::from_secs(60));
 

rumqttc/src/eventloop.rs

@@ -63,6 +63,9 @@ pub enum ConnectionError {
     #[cfg(feature = "proxy")]
     #[error("Proxy Connect: {0}")]
     Proxy(#[from] ProxyError),
+    #[cfg(feature = "websocket")]
+    #[error("Websocket Connect Subprotocol")]
+    WebsocketSubprotocol,
 }
 
 /// Eventloop with all the state of a connection
@@ -398,7 +401,14 @@ async fn network_connect(
                 .headers_mut()
                 .insert("Sec-WebSocket-Protocol", "mqtt".parse().unwrap());
 
-            let (socket, _) = async_tungstenite::tokio::client_async(request, tcp_stream).await?;
+            let (socket, response) =
+                async_tungstenite::tokio::client_async(request, tcp_stream).await?;
+
+            response
+                .headers()
+                .get("Sec-WebSocket-Protocol")
+                .filter(|x| x == "mqtt")
+                .ok_or(Err(ConnectionError::WebsocketSubprotocol));
 
             Network::new(WsStream::new(socket), options.max_incoming_packet_size)
         }
@@ -411,13 +421,19 @@ async fn network_connect(
 
             let connector = tls::rustls_connector(&tls_config).await?;
 
-            let (socket, _) = async_tungstenite::tokio::client_async_tls_with_connector(
+            let (socket, response) = async_tungstenite::tokio::client_async_tls_with_connector(
                 request,
                 tcp_stream,
                 Some(connector),
             )
             .await?;
 
+            response
+                .headers()
+                .get("Sec-WebSocket-Protocol")
+                .filter(|x| x == "mqtt")
+                .ok_or(Err(ConnectionError::WebsocketSubprotocol));
+
             Network::new(WsStream::new(socket), options.max_incoming_packet_size)
         }
     };