fix: validate subprotocol mqtt

rumqttc/src/eventloop.rs

@@ -23,7 +23,7 @@ use crate::tls;
 
 #[cfg(feature = "websocket")]
 use {
-    crate::websockets::{split_url, UrlError},
+    crate::websockets::{split_url, validate_response_headers, UrlError},
     async_tungstenite::tungstenite::client::IntoClientRequest,
     ws_stream_tungstenite::WsStream,
 };
@@ -63,6 +63,9 @@ pub enum ConnectionError {
     #[cfg(feature = "proxy")]
     #[error("Proxy Connect: {0}")]
     Proxy(#[from] ProxyError),
+    #[cfg(feature = "websocket")]
+    #[error("Websocket response validation error: ")]
+    ResponseValidation(#[from] crate::websockets::ValidationError),
 }
 
 /// Eventloop with all the state of a connection
@@ -398,7 +401,10 @@ 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?;
+
+            validate_response_headers(response)?;
 
             Network::new(WsStream::new(socket), options.max_incoming_packet_size)
         }
@@ -411,13 +417,15 @@ 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?;
 
+            validate_response_headers(response)?;
+
             Network::new(WsStream::new(socket), options.max_incoming_packet_size)
         }
     };

rumqttc/src/v5/eventloop.rs

@@ -24,7 +24,7 @@ use {std::path::Path, tokio::net::UnixStream};
 
 #[cfg(feature = "websocket")]
 use {
-    crate::websockets::{split_url, UrlError},
+    crate::websockets::{split_url, validate_response_headers, UrlError},
     async_tungstenite::tungstenite::client::IntoClientRequest,
     ws_stream_tungstenite::WsStream,
 };
@@ -62,6 +62,9 @@ pub enum ConnectionError {
     #[cfg(feature = "proxy")]
     #[error("Proxy Connect: {0}")]
     Proxy(#[from] ProxyError),
+    #[cfg(feature = "websocket")]
+    #[error("Websocket response validation error: ")]
+    ResponseValidation(#[from] crate::websockets::ValidationError),
 }
 
 /// Eventloop with all the state of a connection
@@ -336,7 +339,9 @@ async fn network_connect(options: &MqttOptions) -> Result<Network, ConnectionErr
                 request = request_modifier(request).await;
             }
 
-            let (socket, _) = async_tungstenite::tokio::client_async(request, tcp_stream).await?;
+            let (socket, response) =
+                async_tungstenite::tokio::client_async(request, tcp_stream).await?;
+            validate_response_headers(response)?;
 
             Network::new(WsStream::new(socket), max_incoming_pkt_size)
         }
@@ -353,12 +358,13 @@ async fn network_connect(options: &MqttOptions) -> Result<Network, ConnectionErr
 
             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?;
+            validate_response_headers(response)?;
 
             Network::new(WsStream::new(socket), max_incoming_pkt_size)
         }

rumqttc/src/websockets.rs

@@ -1,3 +1,5 @@
+use http::Response;
+
 #[derive(Debug, thiserror::Error)]
 pub enum UrlError {
     #[error("Invalid protocol specified inside url.")]
@@ -8,6 +10,33 @@ pub enum UrlError {
     Parse(#[from] http::uri::InvalidUri),
 }
 
+#[derive(Debug, thiserror::Error)]
+pub enum ValidationError {
+    #[error("Websocket response does not contain subprotocol header")]
+    SubprotocolHeaderMissing,
+    #[error("Websocket subprotocol header: {0}")]
+    SubprotocolMqtt(String),
+}
+
+pub(crate) fn validate_response_headers(
+    response: Response<Option<Vec<u8>>>,
+) -> Result<(), ValidationError> {
+    let val = response
+        .headers()
+        .get("Sec-WebSocket-Protocol")
+        .ok_or(ValidationError::SubprotocolHeaderMissing)?;
+
+    let sub_protocol = val
+        .to_str()
+        .map_err(|_| ValidationError::SubprotocolHeaderMissing)?;
+
+    if sub_protocol != "mqtt" {
+        return Err(ValidationError::SubprotocolMqtt(sub_protocol.to_owned()));
+    }
+
+    Ok(())
+}
+
 pub(crate) fn split_url(url: &str) -> Result<(String, u16), UrlError> {
     let uri = url.parse::<http::Uri>()?;
     let domain = domain(&uri).ok_or(UrlError::Protocol)?;