Repository: bytebeamio/rumqtt (language: rust)

Issue:
Rumqttc: Validate response header
# Expected Behavior

Rumqttc should validate that the broker that it connects to has the correct websocket-protocol header set. As per the [spec](https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Using_WebSocket_as):
` The WebSocket Subprotocol name selected and returned by the Server MUST be “mqtt”`

# Current Behavior

Currently it ignores it and continues not matter the response.


Should be a very simple fix. [The discarded parameter here](https://github.com/bytebeamio/rumqtt/blob/714f71c9bc3369fa186353201dc2e8197d07594c/rumqttc/src/eventloop.rs#L389) includes the response headers, just needs to validate that it is there, otherwise return a error. 



Discussion / hints:

Fix patch:
diff --git a/rumqttc/src/eventloop.rs b/rumqttc/src/eventloop.rs
index f0bf7644d..e6d96861a 100644
--- a/rumqttc/src/eventloop.rs
+++ b/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,9 @@ 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,12 +416,13 @@ 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)
         }
diff --git a/rumqttc/src/v5/eventloop.rs b/rumqttc/src/v5/eventloop.rs
index d7973af21..fc15fa0d9 100644
--- a/rumqttc/src/v5/eventloop.rs
+++ b/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)
         }
diff --git a/rumqttc/src/websockets.rs b/rumqttc/src/websockets.rs
index 3ee67c3fe..4caa75802 100644
--- a/rumqttc/src/websockets.rs
+++ b/rumqttc/src/websockets.rs
@@ -1,3 +1,5 @@
+use http::{header::ToStrError, Response};
+
 #[derive(Debug, thiserror::Error)]
 pub enum UrlError {
     #[error("Invalid protocol specified inside url.")]
@@ -8,6 +10,35 @@ 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("MQTT not in subprotocol header: {0}")]
+    SubprotocolMqttMissing(String),
+    #[error("Subprotocol header couldn't be converted into string representation")]
+    HeaderToStr(#[from] ToStrError),
+}
+
+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()?;
+
+    if !sub_protocol.contains("mqtt") {
+        return Err(ValidationError::SubprotocolMqttMissing(
+            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)?;


Test patch: