From fa67ae375dae8f7114fa1903c84936f267b0ce3c Mon Sep 17 00:00:00 2001 From: Bryan Ashby Date: Fri, 2 Oct 2026 10:01:36 -0600 Subject: [PATCH 1/3] send WebSocketConfig headers, subprotocols and user agent in the upgrade request (#165) WebSocketTransport::connect built the upgrade request from a fixed set of headers and always offered Sec-WebSocket-Protocol: mqtt, so with_header, with_subprotocol(s) and with_user_agent never reached the server. Request construction moves to WebSocketConfig::build_handshake_request, which starts from tungstenite's into_client_request (fixing a Host header that dropped non-default ports) and adds the configured subprotocols, user agent and custom headers. Invalid, reserved or case-duplicated headers and non-token subprotocols are rejected before dialing. WebSocketConfig's Debug no longer prints header values. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 16 + crates/mqtt5/src/transport/websocket.rs | 362 ++++++++++++++++-- .../tests/websocket_config_property_tests.rs | 61 ++- .../tests/websocket_handshake_integration.rs | 127 ++++++ 4 files changed, 539 insertions(+), 27 deletions(-) create mode 100644 crates/mqtt5/tests/websocket_handshake_integration.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 60e8bfd5..a03486a2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,22 @@ 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 (or `mqtt` when none are configured), 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 + +- **`connect` rejects a WebSocket configuration it cannot send faithfully, before dialing.** It fails with `MqttError::Configuration` when a custom header has an invalid name or value, repeats another custom header's name (case-insensitively), or names a header the transport sets itself (`Host`, `Connection`, `Upgrade`, `Sec-WebSocket-Version`, `Sec-WebSocket-Key`, `Sec-WebSocket-Extensions`); when `Sec-WebSocket-Protocol` or `User-Agent` is set as a custom header instead of through its setter; or when a subprotocol is not a valid HTTP token. +- `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.45.1] - 2026-10-01 ### Fixed diff --git a/crates/mqtt5/src/transport/websocket.rs b/crates/mqtt5/src/transport/websocket.rs index d73ec72b..664fe826 100644 --- a/crates/mqtt5/src/transport/websocket.rs +++ b/crates/mqtt5/src/transport/websocket.rs @@ -51,29 +51,68 @@ 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 offered when none is configured. MQTT-6.0.0-3 requires the +/// client to include "mqtt" in the subprotocols it offers. +const DEFAULT_SUBPROTOCOL: &str = "mqtt"; + +/// Handshake headers the transport sets itself. A custom header with one of +/// these names would duplicate or override the handshake, so it is rejected. +/// `sec-websocket-extensions` is included because the transport negotiates no +/// extensions and cannot speak one a server accepts. +const RESERVED_HEADERS: [&str; 6] = [ + "host", + "connection", + "upgrade", + "sec-websocket-version", + "sec-websocket-key", + "sec-websocket-extensions", +]; + /// 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 offered when this is empty. 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 +135,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 +149,10 @@ impl WebSocketConfig { self } - /// Sets the WebSocket subprotocols to negotiate + /// Sets the WebSocket subprotocols to offer, in order of preference + /// + /// The list is sent exactly as given. MQTT-6.0.0-3 requires the client to + /// offer "mqtt", so keep it in the list unless the broker requires otherwise. #[must_use] pub fn with_subprotocols(mut self, subprotocols: &[&str]) -> Self { self.subprotocols = subprotocols @@ -120,14 +162,26 @@ impl WebSocketConfig { self } - /// Sets a single WebSocket subprotocol + /// Sets a single WebSocket subprotocol to offer + /// + /// MQTT-6.0.0-3 requires the client to offer "mqtt"; see + /// [`with_subprotocols`](Self::with_subprotocols) to offer it alongside another. #[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. The handshake headers (`Host`, `Connection`, `Upgrade`, + /// `Sec-WebSocket-Version`, `Sec-WebSocket-Key`, `Sec-WebSocket-Extensions`) + /// 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). #[must_use] pub fn with_header(mut self, name: &str, value: &str) -> Self { self.headers.insert(name.to_string(), value.to_string()); @@ -320,6 +374,90 @@ 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 (or "mqtt" when none + /// are configured), 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 + /// - 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 subprotocols = if self.subprotocols.is_empty() { + DEFAULT_SUBPROTOCOL.to_string() + } else { + if let Some(invalid) = self.subprotocols.iter().find(|p| !is_http_token(p)) { + return Err(MqttError::Configuration(format!( + "Invalid WebSocket subprotocol {invalid:?}: must be a non-empty HTTP token" + ))); + } + self.subprotocols.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)?); + } + + for (name, value) in &self.headers { + 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 set by the transport and cannot be overridden" + ))); + } + 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 +492,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 +643,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 +890,192 @@ 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", + ] { + let config = WebSocketConfig::new("ws://localhost/mqtt") + .unwrap() + .with_header(name, "value"); + assert!( + handshake_error(&config).contains("set by the transport"), + "{name} should be rejected" + ); + } + } + + #[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_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..e8811fae 100644 --- a/crates/mqtt5/tests/websocket_config_property_tests.rs +++ b/crates/mqtt5/tests/websocket_config_property_tests.rs @@ -81,6 +81,19 @@ fn valid_http_headers() -> impl Strategy> { .prop_map(|btree| btree.into_iter().collect()) } +// Header names a custom header may not use (lowercase): the handshake headers +// the transport sets itself, plus the two that have dedicated setters. +const RESTRICTED_HEADERS: [&str; 8] = [ + "host", + "connection", + "upgrade", + "sec-websocket-version", + "sec-websocket-key", + "sec-websocket-extensions", + "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 +202,53 @@ 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 conflicting = headers.keys().any(|name| { + let lower = name.to_ascii_lowercase(); + RESTRICTED_HEADERS.contains(&lower.as_str()) || !seen.insert(lower) + }); + let request = config.build_handshake_request(); + if conflicting { + prop_assert!(request.is_err(), "conflicting headers {headers:?} 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())); + } + let expected_protocols = if subprotocols.is_empty() { + "mqtt".to_string() + } else { + subprotocols.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..8154c329 --- /dev/null +++ b/crates/mqtt5/tests/websocket_handshake_integration.rs @@ -0,0 +1,127 @@ +#![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::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. +async fn received_headers(configure: impl FnOnce(WebSocketConfig) -> WebSocketConfig) -> HeaderMap { + 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") +} + +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 = 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!(header(&headers, "Host").is_some_and(|host| host.starts_with("127.0.0.1:"))); +} + +#[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("set by the transport"), + "unexpected error: {err}" + ); + assert!(!transport.is_connected()); +} From 4eaa525b84ebabb86756b2c0061447695ea5d5ab Mon Sep 17 00:00:00 2001 From: Bryan Ashby Date: Fri, 2 Oct 2026 10:30:02 -0600 Subject: [PATCH 2/3] reserve body headers, order custom headers, and note behavior changes (#165) Content-Length, Transfer-Encoding and Sec-WebSocket-Accept are now rejected as custom handshake headers alongside the existing reserved names. Custom headers are validated and sent in name order, so the request and the reported error do not depend on HashMap iteration. The integration test checks Host against the exact bound address, and the CHANGELOG calls out the changes that can affect existing configurations. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 12 +++- crates/mqtt5/src/transport/websocket.rs | 67 +++++++++++++++---- .../tests/websocket_config_property_tests.rs | 9 ++- .../tests/websocket_handshake_integration.rs | 18 +++-- 4 files changed, 83 insertions(+), 23 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a03486a2..5b4d7be0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,7 +14,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed -- **`connect` rejects a WebSocket configuration it cannot send faithfully, before dialing.** It fails with `MqttError::Configuration` when a custom header has an invalid name or value, repeats another custom header's name (case-insensitively), or names a header the transport sets itself (`Host`, `Connection`, `Upgrade`, `Sec-WebSocket-Version`, `Sec-WebSocket-Key`, `Sec-WebSocket-Extensions`); when `Sec-WebSocket-Protocol` or `User-Agent` is set as a custom header instead of through its setter; or when a subprotocol is not a valid HTTP token. +- **WebSocket configurations that relied on the old behavior may connect differently or fail to connect.** Because the configuration now reaches the wire: + - `with_subprotocol("mqttv5.0")` or a `with_subprotocols` list without `mqtt` now offers only what is listed. Previously `mqtt` was always offered instead, so a broker that accepts only `mqtt` will now refuse the upgrade. Include `mqtt` in the list (MQTT-6.0.0-3 requires it) to keep connecting. + - 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` or `Transfer-Encoding`, which would describe a body the upgrade request does not have; + - `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. +- 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 diff --git a/crates/mqtt5/src/transport/websocket.rs b/crates/mqtt5/src/transport/websocket.rs index 664fe826..5632e31e 100644 --- a/crates/mqtt5/src/transport/websocket.rs +++ b/crates/mqtt5/src/transport/websocket.rs @@ -68,17 +68,23 @@ use url::Url; /// client to include "mqtt" in the subprotocols it offers. const DEFAULT_SUBPROTOCOL: &str = "mqtt"; -/// Handshake headers the transport sets itself. A custom header with one of -/// these names would duplicate or override the handshake, so it is rejected. +/// 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. -const RESERVED_HEADERS: [&str; 6] = [ +/// 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. +const RESERVED_HEADERS: [&str; 9] = [ "host", "connection", "upgrade", "sec-websocket-version", "sec-websocket-key", "sec-websocket-extensions", + "sec-websocket-accept", + "content-length", + "transfer-encoding", ]; /// WebSocket transport configuration @@ -177,11 +183,14 @@ impl WebSocketConfig { /// 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. The handshake headers (`Host`, `Connection`, `Upgrade`, - /// `Sec-WebSocket-Version`, `Sec-WebSocket-Key`, `Sec-WebSocket-Extensions`) - /// 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). + /// repeat. Headers that belong to the handshake itself (`Host`, + /// `Connection`, `Upgrade`, `Sec-WebSocket-Version`, `Sec-WebSocket-Key`, + /// `Sec-WebSocket-Extensions`, `Sec-WebSocket-Accept`) or would describe a + /// request body (`Content-Length`, `Transfer-Encoding`) 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()); @@ -413,13 +422,17 @@ impl WebSocketConfig { headers.insert(USER_AGENT, header_value("User-Agent", user_agent)?); } - for (name, value) in &self.headers { + // 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 set by the transport and cannot be overridden" + "WebSocket header {name:?} is reserved for the handshake and cannot be set" ))); } if header_name == SEC_WEBSOCKET_PROTOCOL { @@ -1001,17 +1014,47 @@ mod tests { "Sec-WebSocket-Version", "Sec-WebSocket-Key", "Sec-WebSocket-Extensions", + "Sec-WebSocket-Accept", + "Content-Length", + "transfer-encoding", ] { let config = WebSocketConfig::new("ws://localhost/mqtt") .unwrap() .with_header(name, "value"); assert!( - handshake_error(&config).contains("set by the transport"), + 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") diff --git a/crates/mqtt5/tests/websocket_config_property_tests.rs b/crates/mqtt5/tests/websocket_config_property_tests.rs index e8811fae..7f76f892 100644 --- a/crates/mqtt5/tests/websocket_config_property_tests.rs +++ b/crates/mqtt5/tests/websocket_config_property_tests.rs @@ -81,15 +81,18 @@ fn valid_http_headers() -> impl Strategy> { .prop_map(|btree| btree.into_iter().collect()) } -// Header names a custom header may not use (lowercase): the handshake headers -// the transport sets itself, plus the two that have dedicated setters. -const RESTRICTED_HEADERS: [&str; 8] = [ +// Header names a custom header may not use (lowercase): those reserved for +// the handshake, plus the two that have dedicated setters. +const RESTRICTED_HEADERS: [&str; 11] = [ "host", "connection", "upgrade", "sec-websocket-version", "sec-websocket-key", "sec-websocket-extensions", + "sec-websocket-accept", + "content-length", + "transfer-encoding", "sec-websocket-protocol", "user-agent", ]; diff --git a/crates/mqtt5/tests/websocket_handshake_integration.rs b/crates/mqtt5/tests/websocket_handshake_integration.rs index 8154c329..01d1e1b8 100644 --- a/crates/mqtt5/tests/websocket_handshake_integration.rs +++ b/crates/mqtt5/tests/websocket_handshake_integration.rs @@ -5,6 +5,7 @@ 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; @@ -41,8 +42,11 @@ impl Callback for CaptureRequest { } /// Connects a transport built by `configure` to a loopback server and returns -/// the headers of the upgrade request the server received. -async fn received_headers(configure: impl FnOnce(WebSocketConfig) -> WebSocketConfig) -> HeaderMap { +/// 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)); @@ -69,7 +73,7 @@ async fn received_headers(configure: impl FnOnce(WebSocketConfig) -> WebSocketCo transport.close().await.expect("close"); let headers = captured.lock().expect("lock").take(); - headers.expect("server saw no upgrade request") + (headers.expect("server saw no upgrade request"), addr) } fn header<'a>(headers: &'a HeaderMap, name: &str) -> Option<&'a str> { @@ -78,7 +82,7 @@ fn header<'a>(headers: &'a HeaderMap, name: &str) -> Option<&'a str> { #[tokio::test] async fn upgrade_request_carries_configured_headers_subprotocols_and_user_agent() { - let headers = received_headers(|config| { + let (headers, _) = received_headers(|config| { config .with_header("Authorization", "Bearer token123") .with_header("X-Custom", "expected") @@ -98,14 +102,14 @@ async fn upgrade_request_carries_configured_headers_subprotocols_and_user_agent( #[tokio::test] async fn upgrade_request_defaults_offer_mqtt_and_crate_user_agent() { - let headers = received_headers(|config| config).await; + 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!(header(&headers, "Host").is_some_and(|host| host.starts_with("127.0.0.1:"))); + assert_eq!(header(&headers, "Host"), Some(addr.to_string().as_str())); } #[tokio::test] @@ -120,7 +124,7 @@ async fn invalid_header_fails_connect_before_dialing() { .await .expect_err("reserved header accepted"); assert!( - err.to_string().contains("set by the transport"), + err.to_string().contains("reserved for the handshake"), "unexpected error: {err}" ); assert!(!transport.is_connected()); From 0f6b48faa164aa2e3a12bd5d249b6d902aac7446 Mon Sep 17 00:00:00 2001 From: Bryan Ashby Date: Thu, 8 Oct 2026 16:11:30 -0600 Subject: [PATCH 3/3] always offer mqtt, reject duplicate subprotocols, reserve hop-by-hop headers (#165) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-up. The configured subprotocols are now followed by "mqtt" when the list does not include it, as MQTT-6.0.0-3 requires: the broker in this crate only selects a subprotocol the client offers, so a config preferring mqttv5.0 failed with "Server sent no subprotocol". A subprotocol listed twice is rejected (RFC 6455 §4.1), and TE, Trailer, Keep-Alive, Proxy-Connection and Expect join the reserved headers. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 7 +- crates/mqtt5/src/transport/websocket.rs | 104 ++++++++++++++---- .../tests/websocket_config_property_tests.rs | 29 +++-- .../tests/websocket_handshake_integration.rs | 42 +++++++ 4 files changed, 147 insertions(+), 35 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b4d7be0..66266a66 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,21 +9,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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 (or `mqtt` when none are configured), and the user agent when one is set. The default user agent is now `mqtt5/` instead of `mqtt-v5/0.4.0`. +- **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: - - `with_subprotocol("mqttv5.0")` or a `with_subprotocols` list without `mqtt` now offers only what is listed. Previously `mqtt` was always offered instead, so a broker that accepts only `mqtt` will now refuse the upgrade. Include `mqtt` in the list (MQTT-6.0.0-3 requires it) to keep connecting. - 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` or `Transfer-Encoding`, which would describe a body the upgrade request does not have; + - 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. + - 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. diff --git a/crates/mqtt5/src/transport/websocket.rs b/crates/mqtt5/src/transport/websocket.rs index 5632e31e..7bb7fa78 100644 --- a/crates/mqtt5/src/transport/websocket.rs +++ b/crates/mqtt5/src/transport/websocket.rs @@ -64,8 +64,9 @@ use tokio_tungstenite::{ use tracing::{debug, error, info, instrument}; use url::Url; -/// Subprotocol offered when none is configured. MQTT-6.0.0-3 requires the -/// client to include "mqtt" in the subprotocols it offers. +/// 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 @@ -74,8 +75,10 @@ const DEFAULT_SUBPROTOCOL: &str = "mqtt"; /// 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. -const RESERVED_HEADERS: [&str; 9] = [ +/// 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", @@ -85,6 +88,11 @@ const RESERVED_HEADERS: [&str; 9] = [ "sec-websocket-accept", "content-length", "transfer-encoding", + "te", + "trailer", + "keep-alive", + "proxy-connection", + "expect", ]; /// WebSocket transport configuration @@ -94,7 +102,7 @@ pub struct WebSocketConfig { /// Connection timeout pub timeout: Duration, /// Subprotocols to offer, in order of preference (e.g., "mqtt", "mqttv5.0"). - /// "mqtt" is offered when this is empty. + /// "mqtt" is appended when it is not in the list. pub subprotocols: Vec, /// Custom HTTP headers for the WebSocket handshake pub headers: HashMap, @@ -157,8 +165,11 @@ impl WebSocketConfig { /// Sets the WebSocket subprotocols to offer, in order of preference /// - /// The list is sent exactly as given. MQTT-6.0.0-3 requires the client to - /// offer "mqtt", so keep it in the list unless the broker requires otherwise. + /// "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 @@ -168,10 +179,10 @@ impl WebSocketConfig { self } - /// Sets a single WebSocket subprotocol to offer + /// Sets a single WebSocket subprotocol to offer, in preference to "mqtt" /// - /// MQTT-6.0.0-3 requires the client to offer "mqtt"; see - /// [`with_subprotocols`](Self::with_subprotocols) to offer it alongside another. + /// "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()]; @@ -185,8 +196,10 @@ impl WebSocketConfig { /// 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`) or would describe a - /// request body (`Content-Length`, `Transfer-Encoding`) are rejected, as are + /// `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 @@ -386,14 +399,14 @@ impl WebSocketConfig { /// Builds the WebSocket handshake request this configuration produces /// - /// The request carries the configured subprotocols (or "mqtt" when none - /// are configured), the user agent when one is set, and every custom + /// 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 + /// - 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 @@ -403,16 +416,24 @@ impl WebSocketConfig { })?; let headers = request.headers_mut(); - let subprotocols = if self.subprotocols.is_empty() { - DEFAULT_SUBPROTOCOL.to_string() - } else { - if let Some(invalid) = self.subprotocols.iter().find(|p| !is_http_token(p)) { + 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 {invalid:?}: must be a non-empty HTTP token" + "Invalid WebSocket subprotocol {subprotocol:?}: must be a non-empty HTTP token" ))); } - self.subprotocols.join(", ") - }; + 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)?, @@ -1017,6 +1038,11 @@ mod tests { "Sec-WebSocket-Accept", "Content-Length", "transfer-encoding", + "TE", + "Trailer", + "Keep-Alive", + "proxy-connection", + "Expect", ] { let config = WebSocketConfig::new("ws://localhost/mqtt") .unwrap() @@ -1108,6 +1134,40 @@ mod tests { } } + #[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") diff --git a/crates/mqtt5/tests/websocket_config_property_tests.rs b/crates/mqtt5/tests/websocket_config_property_tests.rs index 7f76f892..67556094 100644 --- a/crates/mqtt5/tests/websocket_config_property_tests.rs +++ b/crates/mqtt5/tests/websocket_config_property_tests.rs @@ -82,8 +82,9 @@ fn valid_http_headers() -> impl Strategy> { } // Header names a custom header may not use (lowercase): those reserved for -// the handshake, plus the two that have dedicated setters. -const RESTRICTED_HEADERS: [&str; 11] = [ +// the handshake, hop-by-hop and expectation headers, plus the two that have +// dedicated setters. +const RESTRICTED_HEADERS: [&str; 16] = [ "host", "connection", "upgrade", @@ -93,6 +94,11 @@ const RESTRICTED_HEADERS: [&str; 11] = [ "sec-websocket-accept", "content-length", "transfer-encoding", + "te", + "trailer", + "keep-alive", + "proxy-connection", + "expect", "sec-websocket-protocol", "user-agent", ]; @@ -211,13 +217,17 @@ proptest! { // 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:?} were accepted"); + prop_assert!( + request.is_err(), + "conflicting headers {headers:?} or subprotocols {subprotocols:?} were accepted" + ); return Ok(()); } let request = request.expect("valid configuration builds a request"); @@ -231,11 +241,12 @@ proptest! { for (name, value) in &headers { prop_assert_eq!(sent(name), Some(value.clone())); } - let expected_protocols = if subprotocols.is_empty() { - "mqtt".to_string() - } else { - subprotocols.join(", ") - }; + // "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()); } diff --git a/crates/mqtt5/tests/websocket_handshake_integration.rs b/crates/mqtt5/tests/websocket_handshake_integration.rs index 01d1e1b8..9fca053d 100644 --- a/crates/mqtt5/tests/websocket_handshake_integration.rs +++ b/crates/mqtt5/tests/websocket_handshake_integration.rs @@ -129,3 +129,45 @@ async fn invalid_header_fails_connect_before_dialing() { ); 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(); +}