diff --git a/CHANGELOG.md b/CHANGELOG.md index 52c807c5..a2a1d42c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,31 @@ 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.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Fixed + +- **The WebSocket client transport now sends its configured headers, subprotocols and user agent** (#165). `WebSocketTransport::connect` built the upgrade request from a fixed set of headers and always offered `Sec-WebSocket-Protocol: mqtt`, so anything set with `WebSocketConfig::with_header`, `with_subprotocol`, `with_subprotocols` or `with_user_agent` never reached the server, and brokers that authenticate the upgrade request through custom headers rejected the connection. The request now carries every custom header, the configured subprotocols in order, followed by `mqtt` when the list does not include it (MQTT-6.0.0-3), and the user agent when one is set. The default user agent is now `mqtt5/` instead of `mqtt-v5/0.4.0`. +- **The upgrade request's `Host` header now includes a non-default port**, so `ws://broker:8080/mqtt` sends `Host: broker:8080` rather than `Host: broker`. + +### Changed + +- **WebSocket configurations that relied on the old behavior may connect differently or fail to connect.** Because the configuration now reaches the wire: + - A custom header that is invalid or reserved (see the next entry) used to be silently dropped. It now makes `connect` fail. + - Every WebSocket connection, including those made by `MqttClient`, now sends a `User-Agent` header. +- **`connect` rejects a WebSocket configuration it cannot send faithfully, before dialing.** It fails with `MqttError::Configuration` in these cases: + - a custom header has an invalid name or value, or repeats another custom header's name (case-insensitively); + - a custom header names one reserved for the handshake: `Host`, `Connection`, `Upgrade`, `Sec-WebSocket-Version`, `Sec-WebSocket-Key`, `Sec-WebSocket-Extensions` or `Sec-WebSocket-Accept`; + - a custom header is `Content-Length`, `Transfer-Encoding` or `Trailer`, which would describe a body the upgrade request does not have, or a hop-by-hop or expectation header: `TE`, `Keep-Alive`, `Proxy-Connection` or `Expect`; + - `Sec-WebSocket-Protocol` or `User-Agent` is set as a custom header instead of through its setter; + - a subprotocol is not a valid HTTP token, or is listed more than once (RFC 6455 §4.1). +- Custom headers are sent, and validated, in name order, so the request and the error a bad configuration reports no longer depend on `HashMap` iteration order. +- `WebSocketConfig`'s `Debug` output lists custom header names but not their values, which often carry credentials. + +### Added + +- `WebSocketConfig::build_handshake_request`, which returns the upgrade request the configuration produces. + ## [mqtt5 0.47.3] - 2026-10-09 ### Fixed diff --git a/crates/mqtt5/src/transport/websocket.rs b/crates/mqtt5/src/transport/websocket.rs index d73ec72b..7bb7fa78 100644 --- a/crates/mqtt5/src/transport/websocket.rs +++ b/crates/mqtt5/src/transport/websocket.rs @@ -51,29 +51,82 @@ use std::collections::HashMap; use std::net::SocketAddr; use tokio::net::TcpStream; use tokio_tungstenite::{ - tungstenite::{self, http::Request, protocol::Message}, + tungstenite::{ + client::IntoClientRequest, + http::{ + header::{SEC_WEBSOCKET_PROTOCOL, USER_AGENT}, + HeaderName, HeaderValue, Request, + }, + protocol::Message, + }, MaybeTlsStream, WebSocketStream, }; use tracing::{debug, error, info, instrument}; use url::Url; +/// Subprotocol always offered. MQTT-6.0.0-3 requires the client to include +/// "mqtt" in the subprotocols it offers, so it is appended when the configured +/// list does not already contain it. +const DEFAULT_SUBPROTOCOL: &str = "mqtt"; + +/// Headers a custom header may not use. The first five are set by the +/// transport, and a second copy would duplicate or override the handshake. +/// `sec-websocket-extensions` is included because the transport negotiates no +/// extensions and cannot speak one a server accepts, `sec-websocket-accept` +/// because it belongs in the server's response, and `content-length` and +/// `transfer-encoding` because the upgrade request has no body and a server +/// told otherwise would wait for one. The rest are hop-by-hop or +/// expectation headers (`te`, `trailer`, `keep-alive`, `proxy-connection`, +/// `expect`) that describe the connection or a body rather than the upgrade. +const RESERVED_HEADERS: [&str; 14] = [ + "host", + "connection", + "upgrade", + "sec-websocket-version", + "sec-websocket-key", + "sec-websocket-extensions", + "sec-websocket-accept", + "content-length", + "transfer-encoding", + "te", + "trailer", + "keep-alive", + "proxy-connection", + "expect", +]; + /// WebSocket transport configuration -#[derive(Debug)] pub struct WebSocketConfig { /// WebSocket URL (ws:// or wss://) pub url: Url, /// Connection timeout pub timeout: Duration, - /// Subprotocols to negotiate (e.g., "mqtt", "mqttv3.1", "mqttv5.0") + /// Subprotocols to offer, in order of preference (e.g., "mqtt", "mqttv5.0"). + /// "mqtt" is appended when it is not in the list. pub subprotocols: Vec, /// Custom HTTP headers for the WebSocket handshake pub headers: HashMap, - /// User agent string + /// User agent string, sent as the `User-Agent` header when set pub user_agent: Option, /// TLS configuration for secure WebSocket connections (wss://) pub tls_config: Option, } +impl std::fmt::Debug for WebSocketConfig { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + // Header values often carry credentials, so only the names are shown. + let header_names: Vec<&str> = self.headers.keys().map(String::as_str).collect(); + f.debug_struct("WebSocketConfig") + .field("url", &self.url.as_str()) + .field("timeout", &self.timeout) + .field("subprotocols", &self.subprotocols) + .field("headers", &header_names) + .field("user_agent", &self.user_agent) + .field("tls_config", &self.tls_config) + .finish() + } +} + impl WebSocketConfig { /// Creates a new WebSocket configuration /// @@ -96,9 +149,9 @@ impl WebSocketConfig { Ok(Self { url: parsed_url, timeout: Duration::from_secs(30), - subprotocols: vec!["mqtt".to_string()], + subprotocols: vec![DEFAULT_SUBPROTOCOL.to_string()], headers: HashMap::new(), - user_agent: Some("mqtt-v5/0.4.0".to_string()), + user_agent: Some(concat!("mqtt5/", env!("CARGO_PKG_VERSION")).to_string()), tls_config: None, }) } @@ -110,7 +163,13 @@ impl WebSocketConfig { self } - /// Sets the WebSocket subprotocols to negotiate + /// Sets the WebSocket subprotocols to offer, in order of preference + /// + /// "mqtt" is appended to the offer when it is not in the list, as + /// MQTT-6.0.0-3 requires; the server selects one value, so a broker that + /// wants another listed subprotocol can still choose it. Each subprotocol + /// must be an HTTP token and appear only once (RFC 6455 §4.1), or + /// [`connect`](WebSocketTransport::connect) fails. #[must_use] pub fn with_subprotocols(mut self, subprotocols: &[&str]) -> Self { self.subprotocols = subprotocols @@ -120,14 +179,31 @@ impl WebSocketConfig { self } - /// Sets a single WebSocket subprotocol + /// Sets a single WebSocket subprotocol to offer, in preference to "mqtt" + /// + /// "mqtt" is still offered after it, as MQTT-6.0.0-3 requires; see + /// [`with_subprotocols`](Self::with_subprotocols). #[must_use] pub fn with_subprotocol(mut self, subprotocol: &str) -> Self { self.subprotocols = vec![subprotocol.to_string()]; self } - /// Adds a custom HTTP header + /// Adds a custom HTTP header to the WebSocket handshake request + /// + /// The name and value are validated when the request is built, so an + /// invalid header makes [`connect`](WebSocketTransport::connect) fail + /// rather than this call. Names are matched case-insensitively and must not + /// repeat. Headers that belong to the handshake itself (`Host`, + /// `Connection`, `Upgrade`, `Sec-WebSocket-Version`, `Sec-WebSocket-Key`, + /// `Sec-WebSocket-Extensions`, `Sec-WebSocket-Accept`), would describe a + /// request body (`Content-Length`, `Transfer-Encoding`, `Trailer`), or are + /// hop-by-hop or expectation headers (`TE`, `Keep-Alive`, + /// `Proxy-Connection`, `Expect`) are rejected, as are + /// `Sec-WebSocket-Protocol` and `User-Agent`, which are set with + /// [`with_subprotocols`](Self::with_subprotocols) and + /// [`with_user_agent`](Self::with_user_agent). Custom headers are sent in + /// name order. #[must_use] pub fn with_header(mut self, name: &str, value: &str) -> Self { self.headers.insert(name.to_string(), value.to_string()); @@ -320,6 +396,102 @@ impl WebSocketConfig { pub fn take_tls_config(&mut self) -> Option { self.tls_config.take() } + + /// Builds the WebSocket handshake request this configuration produces + /// + /// The request carries the configured subprotocols followed by "mqtt" when + /// they do not include it, the user agent when one is set, and every custom + /// header. Each call generates a fresh `Sec-WebSocket-Key`. + /// + /// # Errors + /// + /// Returns an error if: + /// - A subprotocol is not a valid HTTP token or is listed more than once + /// - The user agent is not a valid header value + /// - A custom header has an invalid name or value, names a header the + /// transport sets itself, or repeats another custom header's name + pub fn build_handshake_request(&self) -> Result> { + let mut request = self.url.as_str().into_client_request().map_err(|e| { + MqttError::ConnectionError(format!("Failed to build WebSocket request: {e}")) + })?; + let headers = request.headers_mut(); + + let mut offered: Vec<&str> = Vec::with_capacity(self.subprotocols.len() + 1); + for subprotocol in &self.subprotocols { + if !is_http_token(subprotocol) { + return Err(MqttError::Configuration(format!( + "Invalid WebSocket subprotocol {subprotocol:?}: must be a non-empty HTTP token" + ))); + } + if offered.contains(&subprotocol.as_str()) { + return Err(MqttError::Configuration(format!( + "WebSocket subprotocol {subprotocol:?} is listed more than once" + ))); + } + offered.push(subprotocol); + } + if !offered.contains(&DEFAULT_SUBPROTOCOL) { + offered.push(DEFAULT_SUBPROTOCOL); + } + let subprotocols = offered.join(", "); + headers.insert( + SEC_WEBSOCKET_PROTOCOL, + header_value("Sec-WebSocket-Protocol", &subprotocols)?, + ); + + if let Some(user_agent) = &self.user_agent { + headers.insert(USER_AGENT, header_value("User-Agent", user_agent)?); + } + + // Sorted so the request, and which error a bad configuration reports, + // do not depend on HashMap iteration order. + let mut custom: Vec<(&String, &String)> = self.headers.iter().collect(); + custom.sort_unstable(); + for (name, value) in custom { + let header_name = HeaderName::from_bytes(name.as_bytes()).map_err(|e| { + MqttError::Configuration(format!("Invalid WebSocket header name {name:?}: {e}")) + })?; + if RESERVED_HEADERS.contains(&header_name.as_str()) { + return Err(MqttError::Configuration(format!( + "WebSocket header {name:?} is reserved for the handshake and cannot be set" + ))); + } + if header_name == SEC_WEBSOCKET_PROTOCOL { + return Err(MqttError::Configuration(format!( + "WebSocket header {name:?} cannot be set directly; use with_subprotocols" + ))); + } + if header_name == USER_AGENT { + return Err(MqttError::Configuration(format!( + "WebSocket header {name:?} cannot be set directly; use with_user_agent" + ))); + } + if headers.contains_key(&header_name) { + return Err(MqttError::Configuration(format!( + "WebSocket header {name:?} is configured more than once \ + (names are case-insensitive)" + ))); + } + headers.insert(header_name, header_value(name, value)?); + } + + Ok(request) + } +} + +/// Parses a handshake header value, naming the header in the error. +fn header_value(name: &str, value: &str) -> Result { + HeaderValue::from_str(value).map_err(|e| { + MqttError::Configuration(format!("Invalid value for WebSocket header {name:?}: {e}")) + }) +} + +/// Whether `s` is an HTTP token (RFC 9110 §5.6.2), as RFC 6455 §4.1 requires +/// of each offered subprotocol. +fn is_http_token(s: &str) -> bool { + !s.is_empty() + && s.bytes() + .all(|b| b.is_ascii_alphanumeric() || b"!#$%&'*+-.^_`|~".contains(&b)) } /// WebSocket transport implementation @@ -354,7 +526,9 @@ impl WebSocketTransport { &self.config.url } - /// Gets the negotiated subprotocol (if any) + /// Gets the most preferred configured subprotocol (if any) + /// + /// This is the first subprotocol offered, not the one the server selected. #[must_use] pub fn subprotocol(&self) -> Option<&str> { self.config.subprotocols.first().map(String::as_str) @@ -503,21 +677,7 @@ impl Transport for WebSocketTransport { return Err(MqttError::AlreadyConnected); } - let request = Request::builder() - .uri(self.config.url.as_str()) - .header("Host", self.config.url.host_str().unwrap_or("localhost")) - .header("Connection", "Upgrade") - .header("Upgrade", "websocket") - .header("Sec-WebSocket-Version", "13") - .header( - "Sec-WebSocket-Key", - tungstenite::handshake::client::generate_key(), - ) - .header("Sec-WebSocket-Protocol", "mqtt") - .body(()) - .map_err(|e| { - MqttError::ConnectionError(format!("Failed to build WebSocket request: {e}")) - })?; + let request = self.config.build_handshake_request()?; let ws_result = if self.config.is_secure() && self @@ -764,6 +924,261 @@ mod tests { assert_eq!(config.user_agent, Some("custom-client/1.0".to_string())); } + fn request_header<'a>(request: &'a Request<()>, name: &str) -> Option<&'a str> { + request + .headers() + .get(name) + .map(|v| v.to_str().expect("ascii header")) + } + + fn handshake_error(config: &WebSocketConfig) -> String { + config + .build_handshake_request() + .expect_err("handshake request should be rejected") + .to_string() + } + + #[test] + fn test_handshake_request_defaults() { + let config = WebSocketConfig::new("ws://localhost:8080/mqtt").unwrap(); + let request = config.build_handshake_request().unwrap(); + + assert_eq!(request.uri(), "ws://localhost:8080/mqtt"); + assert_eq!(request_header(&request, "Host"), Some("localhost:8080")); + assert_eq!(request_header(&request, "Connection"), Some("Upgrade")); + assert_eq!(request_header(&request, "Upgrade"), Some("websocket")); + assert_eq!( + request_header(&request, "Sec-WebSocket-Version"), + Some("13") + ); + assert!(request_header(&request, "Sec-WebSocket-Key").is_some()); + assert_eq!( + request_header(&request, "Sec-WebSocket-Protocol"), + Some("mqtt") + ); + assert_eq!( + request_header(&request, "User-Agent"), + Some(concat!("mqtt5/", env!("CARGO_PKG_VERSION"))) + ); + } + + #[test] + fn test_handshake_request_applies_configuration() { + let config = WebSocketConfig::new("wss://broker.example.com/mqtt") + .unwrap() + .with_subprotocols(&["mqttv5.0", "mqtt"]) + .with_header("Authorization", "Bearer token123") + .with_header("x-amz-customauthorizer-name", "authorizer") + .with_user_agent("custom-client/1.0"); + let request = config.build_handshake_request().unwrap(); + + assert_eq!( + request_header(&request, "Sec-WebSocket-Protocol"), + Some("mqttv5.0, mqtt") + ); + assert_eq!( + request_header(&request, "Authorization"), + Some("Bearer token123") + ); + assert_eq!( + request_header(&request, "x-amz-customauthorizer-name"), + Some("authorizer") + ); + assert_eq!( + request_header(&request, "User-Agent"), + Some("custom-client/1.0") + ); + } + + #[test] + fn test_handshake_request_offers_mqtt_when_no_subprotocols() { + let mut config = WebSocketConfig::new("ws://localhost/mqtt").unwrap(); + config.subprotocols.clear(); + let request = config.build_handshake_request().unwrap(); + + assert_eq!( + request_header(&request, "Sec-WebSocket-Protocol"), + Some("mqtt") + ); + } + + #[test] + fn test_handshake_request_omits_unset_user_agent() { + let mut config = WebSocketConfig::new("ws://localhost/mqtt").unwrap(); + config.user_agent = None; + let request = config.build_handshake_request().unwrap(); + + assert_eq!(request_header(&request, "User-Agent"), None); + } + + #[test] + fn test_handshake_request_host_header() { + let default_port = WebSocketConfig::new("wss://broker.example.com/mqtt").unwrap(); + let request = default_port.build_handshake_request().unwrap(); + assert_eq!(request_header(&request, "Host"), Some("broker.example.com")); + + let with_userinfo = WebSocketConfig::new("ws://user:pass@localhost:8080/mqtt").unwrap(); + let request = with_userinfo.build_handshake_request().unwrap(); + assert_eq!(request_header(&request, "Host"), Some("localhost:8080")); + + let ipv6 = WebSocketConfig::new("ws://[::1]:8080/mqtt").unwrap(); + let request = ipv6.build_handshake_request().unwrap(); + assert_eq!(request_header(&request, "Host"), Some("[::1]:8080")); + } + + #[test] + fn test_handshake_request_rejects_reserved_headers() { + for name in [ + "Host", + "connection", + "UPGRADE", + "Sec-WebSocket-Version", + "Sec-WebSocket-Key", + "Sec-WebSocket-Extensions", + "Sec-WebSocket-Accept", + "Content-Length", + "transfer-encoding", + "TE", + "Trailer", + "Keep-Alive", + "proxy-connection", + "Expect", + ] { + let config = WebSocketConfig::new("ws://localhost/mqtt") + .unwrap() + .with_header(name, "value"); + assert!( + handshake_error(&config).contains("reserved for the handshake"), + "{name} should be rejected" + ); + } + } + + #[test] + fn test_handshake_request_header_order_is_deterministic() { + // Each config gets a freshly seeded HashMap, so any dependence on + // iteration order would show up across iterations. + for _ in 0..32 { + let config = WebSocketConfig::new("ws://localhost/mqtt") + .unwrap() + .with_header("X-Charlie", "3") + .with_header("X-Alpha", "1") + .with_header("X-Bravo", "2"); + let request = config.build_handshake_request().unwrap(); + let custom: Vec<&str> = request + .headers() + .keys() + .map(HeaderName::as_str) + .filter(|name| name.starts_with("x-")) + .collect(); + assert_eq!(custom, ["x-alpha", "x-bravo", "x-charlie"]); + + let invalid = WebSocketConfig::new("ws://localhost/mqtt") + .unwrap() + .with_header("B-Header", "bad\r\nvalue") + .with_header("A Header", "value"); + assert!(handshake_error(&invalid).contains("Invalid WebSocket header name")); + } + } + + #[test] + fn test_handshake_request_rejects_headers_with_dedicated_setters() { + let config = WebSocketConfig::new("ws://localhost/mqtt") + .unwrap() + .with_header("sec-websocket-protocol", "mqttv5.0"); + assert!(handshake_error(&config).contains("use with_subprotocols")); + + let config = WebSocketConfig::new("ws://localhost/mqtt") + .unwrap() + .with_header("User-Agent", "other/1.0"); + assert!(handshake_error(&config).contains("use with_user_agent")); + } + + #[test] + fn test_handshake_request_rejects_case_insensitive_duplicates() { + let config = WebSocketConfig::new("ws://localhost/mqtt") + .unwrap() + .with_header("Authorization", "Bearer a") + .with_header("authorization", "Bearer b"); + assert!(handshake_error(&config).contains("more than once")); + } + + #[test] + fn test_handshake_request_rejects_invalid_headers() { + let config = WebSocketConfig::new("ws://localhost/mqtt") + .unwrap() + .with_header("Bad Name", "value"); + assert!(handshake_error(&config).contains("Invalid WebSocket header name")); + + let config = WebSocketConfig::new("ws://localhost/mqtt") + .unwrap() + .with_header("X-Injected", "value\r\nX-Other: smuggled"); + assert!(handshake_error(&config).contains("Invalid value")); + + let config = WebSocketConfig::new("ws://localhost/mqtt") + .unwrap() + .with_user_agent("agent\r\n"); + assert!(handshake_error(&config).contains("User-Agent")); + } + + #[test] + fn test_handshake_request_rejects_invalid_subprotocols() { + for subprotocol in ["", "mqtt, mqttv5.0", "mqtt v5", "mqtt\r\n"] { + let config = WebSocketConfig::new("ws://localhost/mqtt") + .unwrap() + .with_subprotocol(subprotocol); + assert!( + handshake_error(&config).contains("Invalid WebSocket subprotocol"), + "{subprotocol:?} should be rejected" + ); + } + } + + #[test] + fn test_handshake_request_appends_mqtt_to_other_subprotocols() { + let config = WebSocketConfig::new("ws://localhost/mqtt") + .unwrap() + .with_subprotocol("mqttv5.0"); + let request = config.build_handshake_request().unwrap(); + assert_eq!( + request_header(&request, "Sec-WebSocket-Protocol"), + Some("mqttv5.0, mqtt") + ); + + let config = WebSocketConfig::new("ws://localhost/mqtt") + .unwrap() + .with_subprotocols(&["mqtt", "mqttv5.0"]); + let request = config.build_handshake_request().unwrap(); + assert_eq!( + request_header(&request, "Sec-WebSocket-Protocol"), + Some("mqtt, mqttv5.0") + ); + } + + #[test] + fn test_handshake_request_rejects_duplicate_subprotocols() { + for list in [&["mqtt", "mqtt"][..], &["mqttv5.0", "mqtt", "mqttv5.0"][..]] { + let config = WebSocketConfig::new("ws://localhost/mqtt") + .unwrap() + .with_subprotocols(list); + assert!( + handshake_error(&config).contains("listed more than once"), + "{list:?} should be rejected" + ); + } + } + + #[test] + fn test_websocket_config_debug_redacts_header_values() { + let config = WebSocketConfig::new("ws://localhost/mqtt") + .unwrap() + .with_header("Authorization", "Bearer secret-token"); + let debug = format!("{config:?}"); + + assert!(debug.contains("Authorization")); + assert!(!debug.contains("secret-token")); + } + #[tokio::test] async fn test_websocket_transport_creation() { let config = WebSocketConfig::new("ws://localhost:8080/mqtt").unwrap(); diff --git a/crates/mqtt5/tests/websocket_config_property_tests.rs b/crates/mqtt5/tests/websocket_config_property_tests.rs index 4a1a89aa..67556094 100644 --- a/crates/mqtt5/tests/websocket_config_property_tests.rs +++ b/crates/mqtt5/tests/websocket_config_property_tests.rs @@ -81,6 +81,28 @@ fn valid_http_headers() -> impl Strategy> { .prop_map(|btree| btree.into_iter().collect()) } +// Header names a custom header may not use (lowercase): those reserved for +// the handshake, hop-by-hop and expectation headers, plus the two that have +// dedicated setters. +const RESTRICTED_HEADERS: [&str; 16] = [ + "host", + "connection", + "upgrade", + "sec-websocket-version", + "sec-websocket-key", + "sec-websocket-extensions", + "sec-websocket-accept", + "content-length", + "transfer-encoding", + "te", + "trailer", + "keep-alive", + "proxy-connection", + "expect", + "sec-websocket-protocol", + "user-agent", +]; + // Generate valid subprotocols fn valid_subprotocols() -> impl Strategy> { prop::collection::vec("[a-zA-Z][a-zA-Z0-9._-]{0,20}", 0..5) @@ -189,9 +211,58 @@ proptest! { assert_eq!(config.subprotocols, subprotocols); } - if let Some(ua) = user_agent { - assert_eq!(config.user_agent, Some(ua)); + if let Some(ua) = &user_agent { + assert_eq!(config.user_agent.as_ref(), Some(ua)); + } + + // The configuration must reach the handshake request, not just the struct. + let mut seen = std::collections::HashSet::new(); + let mut seen_protocols = std::collections::HashSet::new(); + let conflicting = headers.keys().any(|name| { + let lower = name.to_ascii_lowercase(); + RESTRICTED_HEADERS.contains(&lower.as_str()) || !seen.insert(lower) + }) || subprotocols.iter().any(|p| !seen_protocols.insert(p.as_str())); + let request = config.build_handshake_request(); + if conflicting { + prop_assert!( + request.is_err(), + "conflicting headers {headers:?} or subprotocols {subprotocols:?} were accepted" + ); + return Ok(()); + } + let request = request.expect("valid configuration builds a request"); + let sent = |name: &str| { + request + .headers() + .get(name) + .map(|v| v.to_str().expect("ascii header").to_string()) + }; + + for (name, value) in &headers { + prop_assert_eq!(sent(name), Some(value.clone())); } + // "mqtt" is always offered, after the configured subprotocols. + let mut expected_protocols = subprotocols.clone(); + if !expected_protocols.iter().any(|p| p == "mqtt") { + expected_protocols.push("mqtt".to_string()); + } + let expected_protocols = expected_protocols.join(", "); + prop_assert_eq!(sent("Sec-WebSocket-Protocol"), Some(expected_protocols)); + prop_assert_eq!(sent("User-Agent"), config.user_agent.clone()); + } + + #[test] + fn prop_handshake_request_rejects_restricted_headers( + name in prop::sample::select(RESTRICTED_HEADERS.to_vec()), + uppercase in any::(), + value in "[A-Za-z0-9]{1,20}" + ) { + let name = if uppercase { name.to_ascii_uppercase() } else { name.to_string() }; + let config = WebSocketConfig::new("ws://localhost:8080/mqtt") + .unwrap() + .with_header(&name, &value); + + prop_assert!(config.build_handshake_request().is_err(), "{name} was accepted"); } #[test] diff --git a/crates/mqtt5/tests/websocket_handshake_integration.rs b/crates/mqtt5/tests/websocket_handshake_integration.rs new file mode 100644 index 00000000..9fca053d --- /dev/null +++ b/crates/mqtt5/tests/websocket_handshake_integration.rs @@ -0,0 +1,173 @@ +#![cfg(feature = "transport-websocket")] + +//! Checks the WebSocket upgrade request the client transport actually sends, +//! as seen by a server, rather than the configuration it was built from. + +use mqtt5::transport::websocket::{WebSocketConfig, WebSocketTransport}; +use mqtt5::Transport; +use std::net::SocketAddr; +use std::sync::{Arc, Mutex}; +use std::time::Duration; +use tokio::net::TcpListener; +use tokio::time::timeout; +use tokio_tungstenite::tungstenite::handshake::server::{ + Callback, ErrorResponse, Request, Response, +}; +use tokio_tungstenite::tungstenite::http::HeaderMap; + +const WAIT: Duration = Duration::from_secs(5); + +/// Records the upgrade request's headers and selects the first offered +/// subprotocol, as a broker would. +struct CaptureRequest { + captured: Arc>>, +} + +impl Callback for CaptureRequest { + fn on_request(self, req: &Request, mut resp: Response) -> Result { + if let Some(first) = req + .headers() + .get("Sec-WebSocket-Protocol") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.split(',').next()) + { + resp.headers_mut().insert( + "Sec-WebSocket-Protocol", + first.trim().parse().expect("header"), + ); + } + *self.captured.lock().expect("lock") = Some(req.headers().clone()); + Ok(resp) + } +} + +/// Connects a transport built by `configure` to a loopback server and returns +/// the headers of the upgrade request the server received, with the address +/// the server listened on. +async fn received_headers( + configure: impl FnOnce(WebSocketConfig) -> WebSocketConfig, +) -> (HeaderMap, SocketAddr) { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("addr"); + let captured = Arc::new(Mutex::new(None)); + let callback = CaptureRequest { + captured: Arc::clone(&captured), + }; + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.expect("accept"); + tokio_tungstenite::accept_hdr_async(stream, callback) + .await + .expect("server handshake") + }); + + let config = configure(WebSocketConfig::new(&format!("ws://{addr}/mqtt")).expect("config")); + let mut transport = WebSocketTransport::new(config); + timeout(WAIT, transport.connect()) + .await + .expect("connect timed out") + .expect("client handshake"); + let _server_ws = timeout(WAIT, server) + .await + .expect("server timed out") + .expect("server task"); + transport.close().await.expect("close"); + + let headers = captured.lock().expect("lock").take(); + (headers.expect("server saw no upgrade request"), addr) +} + +fn header<'a>(headers: &'a HeaderMap, name: &str) -> Option<&'a str> { + headers.get(name).map(|v| v.to_str().expect("ascii header")) +} + +#[tokio::test] +async fn upgrade_request_carries_configured_headers_subprotocols_and_user_agent() { + let (headers, _) = received_headers(|config| { + config + .with_header("Authorization", "Bearer token123") + .with_header("X-Custom", "expected") + .with_subprotocols(&["mqttv5.0", "mqtt"]) + .with_user_agent("repro/1.0") + }) + .await; + + assert_eq!(header(&headers, "Authorization"), Some("Bearer token123")); + assert_eq!(header(&headers, "X-Custom"), Some("expected")); + assert_eq!( + header(&headers, "Sec-WebSocket-Protocol"), + Some("mqttv5.0, mqtt") + ); + assert_eq!(header(&headers, "User-Agent"), Some("repro/1.0")); +} + +#[tokio::test] +async fn upgrade_request_defaults_offer_mqtt_and_crate_user_agent() { + let (headers, addr) = received_headers(|config| config).await; + + assert_eq!(header(&headers, "Sec-WebSocket-Protocol"), Some("mqtt")); + assert_eq!( + header(&headers, "User-Agent"), + Some(concat!("mqtt5/", env!("CARGO_PKG_VERSION"))) + ); + assert_eq!(header(&headers, "Host"), Some(addr.to_string().as_str())); +} + +#[tokio::test] +async fn invalid_header_fails_connect_before_dialing() { + let config = WebSocketConfig::new("ws://127.0.0.1:9/mqtt") + .expect("config") + .with_header("Host", "spoofed.example.com"); + let mut transport = WebSocketTransport::new(config); + + let err = transport + .connect() + .await + .expect_err("reserved header accepted"); + assert!( + err.to_string().contains("reserved for the handshake"), + "unexpected error: {err}" + ); + assert!(!transport.is_connected()); +} + +/// The broker in this crate only selects a subprotocol the client offers, and +/// the client fails the handshake when the server selects none, so a client +/// that preferred `mqttv5.0` could only connect because `mqtt` is still +/// offered (MQTT-6.0.0-3). +#[cfg(feature = "broker")] +#[tokio::test] +async fn preferring_another_subprotocol_still_connects_to_the_broker() { + use mqtt5::broker::config::{ + BrokerConfig, StorageBackend, StorageConfig, WebSocketConfig as BrokerWebSocketConfig, + }; + use mqtt5::broker::MqttBroker; + + let config = BrokerConfig::default() + .with_storage(StorageConfig::new().with_backend(StorageBackend::Memory)) + .with_bind_address(([127, 0, 0, 1], 0)) + .with_websocket( + BrokerWebSocketConfig::new() + .with_bind_address(([127, 0, 0, 1], 0)) + .with_path("/mqtt"), + ); + let mut broker = MqttBroker::with_config(config).await.expect("broker"); + let addr = broker.ws_local_addr().expect("websocket listener"); + let mut ready = broker.ready_receiver(); + let broker_task = tokio::spawn(async move { broker.run().await }); + timeout(WAIT, ready.wait_for(|ready| *ready)) + .await + .expect("broker did not become ready") + .expect("broker stopped"); + + let config = WebSocketConfig::new(&format!("ws://{addr}/mqtt")) + .expect("config") + .with_subprotocol("mqttv5.0"); + let mut transport = WebSocketTransport::new(config); + timeout(WAIT, transport.connect()) + .await + .expect("connect timed out") + .expect("broker refused a client preferring mqttv5.0"); + + transport.close().await.expect("close"); + broker_task.abort(); +}