diff --git a/CHANGELOG.md b/CHANGELOG.md index bae39d67..9c0dfdcb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Breaking + +- **`ConnectOptions` has a new public field, `websocket_config`.** Code that builds `ConnectOptions` with a struct literal must add it. The field exists only with the `transport-websocket` feature, which is on by default. + ### Fixed - **`wss://` connections now use the TLS configuration on `WebSocketConfig`** (#190). With server-certificate verification on, `WebSocketTransport::connect` used tokio-tungstenite's own client configuration built from the platform's root certificates, so a CA set with `with_ca_cert_from_file` or `with_ca_cert_from_bytes` was never trusted (a server it issued failed with `UnknownIssuer`), a client certificate set with `with_client_auth_from_files` or `with_client_auth_from_bytes` was never presented, and `use_system_roots` and ALPN protocols had no effect. With verification off, the client certificate and ALPN protocols were dropped. The handshake now uses the rustls configuration `TlsTransport` builds from a `TlsConfig`, in both cases, except that `use_system_roots` adds the platform's root certificates, as a `wss://` connection without a TLS configuration already did, rather than the bundled `webpki-roots`. @@ -32,6 +36,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **`ConnectOptions::with_websocket_config`, so `MqttClient` can send custom WebSocket handshake headers** (#165). `MqttClient` built each connection's `WebSocketConfig` from the URL alone, so its ws:// and wss:// connections could not carry custom headers, subprotocols or a user agent. The configuration set here is applied on every connection attempt, including automatic reconnects, with the URL taken from the address passed to `connect`; for wss://, its TLS configuration, when it has one, is used in place of the one stored with `set_tls_config` or `connect_with_tls`. A connection that follows a server redirect (`UseAnotherServer` or `ServerMoved`) does not use it, so its headers are never sent to a host the broker names. +- `WebSocketConfig` implements `Clone`. - `TlsConfig::client_config`, which returns the rustls client configuration a `TlsConfig` describes. - `WebSocketConfig::build_handshake_request`, which returns the upgrade request the configuration produces. diff --git a/crates/mqtt5/src/client/inner.rs b/crates/mqtt5/src/client/inner.rs index f560c784..b037933f 100644 --- a/crates/mqtt5/src/client/inner.rs +++ b/crates/mqtt5/src/client/inner.rs @@ -20,6 +20,8 @@ use crate::transport::{TcpTransport, TlsTransport, TransportType}; use super::connection::ConnectionEvent; use super::direct::{AutomaticReconnectLifecycle, StoredSubscription}; use super::state::ClientTransportType; +#[cfg(feature = "transport-websocket")] +use super::state::WebSocketSettings; use super::MqttClient; impl MqttClient { @@ -71,7 +73,10 @@ impl MqttClient { { let (host, port) = Self::split_host_port(rest, 80)?; Ok(( - ClientTransportType::WebSocket(address.to_string()), + ClientTransportType::WebSocket( + address.to_string(), + WebSocketSettings::Configured, + ), host, port, )) @@ -88,7 +93,10 @@ impl MqttClient { { let (host, port) = Self::split_host_port(rest, 443)?; Ok(( - ClientTransportType::WebSocketSecure(address.to_string()), + ClientTransportType::WebSocketSecure( + address.to_string(), + WebSocketSettings::Configured, + ), host, port, )) @@ -208,10 +216,13 @@ impl MqttClient { ClientTransportType::Tcp => Self::connect_tcp(addr).await, ClientTransportType::Tls => self.connect_tls(addr, host).await, #[cfg(feature = "transport-websocket")] - ClientTransportType::WebSocket(url) => Self::connect_websocket(&url).await, + ClientTransportType::WebSocket(url, settings) => { + self.connect_websocket(&url, settings).await + } #[cfg(feature = "transport-websocket")] - ClientTransportType::WebSocketSecure(url) => { - self.connect_websocket_secure(addr, host, &url).await + ClientTransportType::WebSocketSecure(url, settings) => { + self.connect_websocket_secure(addr, host, &url, settings) + .await } #[cfg(feature = "transport-quic")] ClientTransportType::Quic => self.connect_quic(addr, host).await, @@ -259,16 +270,37 @@ impl MqttClient { Ok(TransportType::Tls(Box::new(tls_transport))) } + /// The WebSocket configuration for a connection to `url`: the one set with + /// `ConnectOptions::with_websocket_config`, read on every attempt so + /// reconnects use it too, or a default one, which a server redirect + /// always gets. #[cfg(feature = "transport-websocket")] - async fn connect_websocket(url: &str) -> Result { - let config = WebSocketConfig::new(url) - .map_err(|e| MqttError::ConnectionError(format!("Invalid WebSocket URL: {e}")))?; - let mut ws_transport = WebSocketTransport::new(config); - ws_transport - .connect() - .await - .map_err(|e| MqttError::ConnectionError(format!("WebSocket connect failed: {e}")))?; - Ok(TransportType::WebSocket(Box::new(ws_transport))) + async fn websocket_config_for( + &self, + url: &str, + settings: WebSocketSettings, + ) -> Result { + let configured = match settings { + WebSocketSettings::Configured => { + self.inner.read().await.options.websocket_config.clone() + } + WebSocketSettings::Default => None, + }; + match configured { + Some(config) => WebSocketConfig::clone(&config).with_url(url), + None => WebSocketConfig::new(url), + } + .map_err(|e| MqttError::ConnectionError(format!("Invalid WebSocket URL: {e}"))) + } + + #[cfg(feature = "transport-websocket")] + async fn connect_websocket( + &self, + url: &str, + settings: WebSocketSettings, + ) -> Result { + let config = self.websocket_config_for(url, settings).await?; + Self::open_websocket(config).await } #[cfg(feature = "transport-websocket")] @@ -277,10 +309,17 @@ impl MqttClient { addr: std::net::SocketAddr, host: &str, url: &str, + settings: WebSocketSettings, ) -> Result { let insecure = self.transport_config.read().await.insecure_tls; - let mut config = WebSocketConfig::new(url) - .map_err(|e| MqttError::ConnectionError(format!("Invalid WebSocket URL: {e}")))?; + let mut config = self.websocket_config_for(url, settings).await?; + + if let Some(tls_config) = config.tls_config.as_mut() { + if insecure { + tls_config.verify_server_cert = false; + } + return Self::open_websocket(config).await; + } // As in connect_tls: a stored TLS config applies to wss:// too. With // none, an insecure connection still needs one to disable @@ -304,6 +343,11 @@ impl MqttClient { config = config.with_tls_config(tls_config); } + Self::open_websocket(config).await + } + + #[cfg(feature = "transport-websocket")] + async fn open_websocket(config: WebSocketConfig) -> Result { let mut ws_transport = WebSocketTransport::new(config); ws_transport .connect() @@ -471,10 +515,18 @@ impl MqttClient { } pub(crate) async fn connect_internal(&self, address: &str) -> Result { + Box::pin(self.connect_following_redirects(address, 0)).await + } + + async fn connect_following_redirects( + &self, + address: &str, + redirects_followed: u8, + ) -> Result { const MAX_REDIRECTS: u8 = 3; self.inner.read().await.options.validate_deferred_ack()?; let mut current_address = address.to_string(); - let mut redirect_count: u8 = 0; + let mut redirect_count = redirects_followed; loop { let client_id = self.inner.read().await.options.client_id.clone(); @@ -485,7 +537,12 @@ impl MqttClient { "Connection attempt" ); - let (client_transport_type, host, port) = Self::parse_address(¤t_address)?; + let (requested_transport_type, host, port) = Self::parse_address(¤t_address)?; + let client_transport_type = if redirect_count == 0 { + requested_transport_type + } else { + requested_transport_type.redirected() + }; let addrs = Self::resolve_addresses(host, port).await?; let addresses_to_try = Self::select_addresses_for_connection(&addrs, host); @@ -569,7 +626,7 @@ impl MqttClient { reason = ?reason, "Following server redirect from TLS connection" ); - return Box::pin(self.connect_internal(&url)).await; + return Box::pin(self.connect_following_redirects(&url, 1)).await; } Err(MqttError::ConnectionError( "Server redirect missing server reference".to_string(), diff --git a/crates/mqtt5/src/client/mod.rs b/crates/mqtt5/src/client/mod.rs index 72a2fa56..5e346650 100644 --- a/crates/mqtt5/src/client/mod.rs +++ b/crates/mqtt5/src/client/mod.rs @@ -1344,7 +1344,7 @@ mod tests { let (transport, host, port) = MqttClient::parse_address("ws://localhost").unwrap(); assert!(matches!( transport, - state::ClientTransportType::WebSocket(_) + state::ClientTransportType::WebSocket(..) )); assert_eq!(host, "localhost"); assert_eq!(port, 80); @@ -1352,7 +1352,7 @@ mod tests { let (transport, host, port) = MqttClient::parse_address("ws://localhost:8080").unwrap(); assert!(matches!( transport, - state::ClientTransportType::WebSocket(_) + state::ClientTransportType::WebSocket(..) )); assert_eq!(host, "localhost"); assert_eq!(port, 8080); @@ -1360,7 +1360,7 @@ mod tests { let (transport, host, port) = MqttClient::parse_address("wss://secure.broker").unwrap(); assert!(matches!( transport, - state::ClientTransportType::WebSocketSecure(_) + state::ClientTransportType::WebSocketSecure(..) )); assert_eq!(host, "secure.broker"); assert_eq!(port, 443); @@ -1369,14 +1369,14 @@ mod tests { MqttClient::parse_address("wss://secure.broker:8443").unwrap(); assert!(matches!( transport, - state::ClientTransportType::WebSocketSecure(_) + state::ClientTransportType::WebSocketSecure(..) )); assert_eq!(host, "secure.broker"); assert_eq!(port, 8443); let (transport, host, port) = MqttClient::parse_address("ws://broker.emqx.io:8083/mqtt").unwrap(); - if let state::ClientTransportType::WebSocket(url) = transport { + if let state::ClientTransportType::WebSocket(url, _) = transport { assert_eq!(url, "ws://broker.emqx.io:8083/mqtt"); } else { panic!("Expected WebSocket transport type"); @@ -1386,7 +1386,7 @@ mod tests { let (transport, host, port) = MqttClient::parse_address("wss://broker.hivemq.com:8884/mqtt").unwrap(); - if let state::ClientTransportType::WebSocketSecure(url) = transport { + if let state::ClientTransportType::WebSocketSecure(url, _) = transport { assert_eq!(url, "wss://broker.hivemq.com:8884/mqtt"); } else { panic!("Expected WebSocketSecure transport type"); diff --git a/crates/mqtt5/src/client/state.rs b/crates/mqtt5/src/client/state.rs index c8353d1a..5f2c2728 100644 --- a/crates/mqtt5/src/client/state.rs +++ b/crates/mqtt5/src/client/state.rs @@ -12,20 +12,39 @@ use super::direct::AutomaticReconnectLifecycle; use super::direct::{StoredSubscription, SubscriptionPersistence}; use super::MqttClient; +#[cfg(feature = "transport-websocket")] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum WebSocketSettings { + Configured, + Default, +} + #[derive(Debug, Clone)] pub(crate) enum ClientTransportType { Tcp, Tls, #[cfg(feature = "transport-websocket")] - WebSocket(String), + WebSocket(String, WebSocketSettings), #[cfg(feature = "transport-websocket")] - WebSocketSecure(String), + WebSocketSecure(String, WebSocketSettings), #[cfg(feature = "transport-quic")] Quic, #[cfg(feature = "transport-quic")] QuicSecure, } +impl ClientTransportType { + pub(crate) fn redirected(self) -> Self { + match self { + #[cfg(feature = "transport-websocket")] + Self::WebSocket(url, _) => Self::WebSocket(url, WebSocketSettings::Default), + #[cfg(feature = "transport-websocket")] + Self::WebSocketSecure(url, _) => Self::WebSocketSecure(url, WebSocketSettings::Default), + other => other, + } + } +} + impl MqttClient { pub(crate) async fn trigger_connection_event(&self, event: ConnectionEvent) { super::fire_connection_event(&self.connection_event_callbacks, event).await; diff --git a/crates/mqtt5/src/transport/websocket.rs b/crates/mqtt5/src/transport/websocket.rs index 3c53fb89..a35b51bf 100644 --- a/crates/mqtt5/src/transport/websocket.rs +++ b/crates/mqtt5/src/transport/websocket.rs @@ -97,6 +97,7 @@ const RESERVED_HEADERS: [&str; 14] = [ ]; /// WebSocket transport configuration +#[derive(Clone)] pub struct WebSocketConfig { /// WebSocket URL (ws:// or wss://) pub url: Url, @@ -135,20 +136,8 @@ impl WebSocketConfig { /// /// Returns an error if the URL is invalid or uses an unsupported scheme pub fn new(url: &str) -> Result { - let parsed_url = Url::parse(url) - .map_err(|e| MqttError::ProtocolError(format!("Invalid WebSocket URL: {e}")))?; - - match parsed_url.scheme() { - "ws" | "wss" => {} - scheme => { - return Err(MqttError::ProtocolError(format!( - "Unsupported WebSocket scheme: {scheme}. Use 'ws' or 'wss'" - ))); - } - } - Ok(Self { - url: parsed_url, + url: parse_websocket_url(url)?, timeout: Duration::from_secs(30), subprotocols: vec![DEFAULT_SUBPROTOCOL.to_string()], headers: HashMap::new(), @@ -157,6 +146,16 @@ impl WebSocketConfig { }) } + /// Returns this configuration with its URL replaced + /// + /// Used by `MqttClient`, which connects to the address it is given with + /// the rest of the configuration from + /// [`ConnectOptions::with_websocket_config`](crate::ConnectOptions::with_websocket_config). + pub(crate) fn with_url(mut self, url: &str) -> Result { + self.url = parse_websocket_url(url)?; + Ok(self) + } + /// Sets the connection timeout #[must_use] pub fn with_timeout(mut self, timeout: Duration) -> Self { @@ -510,6 +509,19 @@ fn header_value(name: &str, value: &str) -> Result { }) } +/// Parses a WebSocket URL, accepting only the ws and wss schemes +fn parse_websocket_url(url: &str) -> Result { + let parsed_url = Url::parse(url) + .map_err(|e| MqttError::ProtocolError(format!("Invalid WebSocket URL: {e}")))?; + + match parsed_url.scheme() { + "ws" | "wss" => Ok(parsed_url), + scheme => Err(MqttError::ProtocolError(format!( + "Unsupported WebSocket scheme: {scheme}. Use 'ws' or 'wss'" + ))), + } +} + /// 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 { diff --git a/crates/mqtt5/src/types.rs b/crates/mqtt5/src/types.rs index f613bf0e..16e05866 100644 --- a/crates/mqtt5/src/types.rs +++ b/crates/mqtt5/src/types.rs @@ -47,6 +47,10 @@ pub struct ConnectOptions { /// Session Expiry Interval (`with_session_expiry_interval`): under MQTT v5 an absent /// Session Expiry Interval means 0, so the session ends when the connection closes. pub resume_existing_session: bool, + /// WebSocket configuration for ws:// and wss:// connections; see + /// [`ConnectOptions::with_websocket_config`]. + #[cfg(feature = "transport-websocket")] + pub websocket_config: Option>, } impl ConnectOptions { @@ -60,9 +64,47 @@ impl ConnectOptions { codec_registry: None, deferred_ack: false, resume_existing_session: false, + #[cfg(feature = "transport-websocket")] + websocket_config: None, } } + /// Sets the WebSocket configuration for ws:// and wss:// connections + /// + /// It is applied on every connection attempt, including automatic + /// reconnects: its custom headers, subprotocols, user agent and timeout go + /// into each upgrade request, and for wss:// its TLS configuration, when it + /// has one, is used in place of the one stored with `set_tls_config` or + /// `connect_with_tls` (`set_insecure_tls(true)` still disables + /// verification). The client connects to the address passed to `connect`, + /// so the configuration's own `url` is not used; build it from that + /// address or any URL with the same scheme. A connection that follows a + /// server redirect (`UseAnotherServer` or `ServerMoved`) does not use it, + /// so its headers only go to the address passed to `connect`. + /// + /// ```rust,no_run + /// # use mqtt5::{ConnectOptions, MqttClient}; + /// # use mqtt5::transport::websocket::WebSocketConfig; + /// # async fn example() -> mqtt5::Result<()> { + /// let address = "wss://broker.example.com/mqtt"; + /// let websocket = WebSocketConfig::new(address)? + /// .with_header("x-amz-customauthorizer-name", "my-authorizer"); + /// let options = ConnectOptions::new("client-1").with_websocket_config(websocket); + /// let client = MqttClient::with_options(options); + /// client.connect(address).await?; + /// # Ok(()) + /// # } + /// ``` + #[cfg(feature = "transport-websocket")] + #[must_use] + pub fn with_websocket_config( + mut self, + config: crate::transport::websocket::WebSocketConfig, + ) -> Self { + self.websocket_config = Some(Arc::new(config)); + self + } + /// Opts in to resuming a broker-held session without local session state. /// /// See [`ConnectOptions::resume_existing_session`]. @@ -248,7 +290,8 @@ impl ConnectOptions { impl std::fmt::Debug for ConnectOptions { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("ConnectOptions") + let mut debug = f.debug_struct("ConnectOptions"); + debug .field("protocol_options", &self.protocol_options) .field("session_config", &self.session_config) .field("reconnect_config", &self.reconnect_config) @@ -258,8 +301,10 @@ impl std::fmt::Debug for ConnectOptions { &self.codec_registry.as_ref().map(|_| "CodecRegistry"), ) .field("deferred_ack", &self.deferred_ack) - .field("resume_existing_session", &self.resume_existing_session) - .finish() + .field("resume_existing_session", &self.resume_existing_session); + #[cfg(feature = "transport-websocket")] + debug.field("websocket_config", &self.websocket_config); + debug.finish() } } diff --git a/crates/mqtt5/tests/broker_quic_integration.rs b/crates/mqtt5/tests/broker_quic_integration.rs index 7b8d00b2..1591fd45 100644 --- a/crates/mqtt5/tests/broker_quic_integration.rs +++ b/crates/mqtt5/tests/broker_quic_integration.rs @@ -164,22 +164,22 @@ async fn test_broker_default_quic_port() { #[tokio::test] async fn test_control_only_reconnect_publish_does_not_reenter_disconnect_loop() { - assert_control_only_reconnect_publish_stable(false).await; + Box::pin(assert_control_only_reconnect_publish_stable(false)).await; } #[tokio::test] async fn test_control_only_clean_session_reconnect_publish_does_not_reenter_disconnect_loop() { - assert_control_only_reconnect_publish_stable(true).await; + Box::pin(assert_control_only_reconnect_publish_stable(true)).await; } #[tokio::test] async fn test_control_only_clean_session_qos0_burst_after_reconnect_does_not_disconnect() { - assert_control_only_reconnect_qos0_burst_stable(false).await; + Box::pin(assert_control_only_reconnect_qos0_burst_stable(false)).await; } #[tokio::test] async fn test_control_only_secure_clean_session_qos0_burst_after_reconnect_does_not_disconnect() { - assert_control_only_reconnect_qos0_burst_stable(true).await; + Box::pin(assert_control_only_reconnect_qos0_burst_stable(true)).await; } async fn assert_control_only_reconnect_publish_stable(clean_start: bool) { diff --git a/crates/mqtt5/tests/clean_start_prebind_subscribe.rs b/crates/mqtt5/tests/clean_start_prebind_subscribe.rs index 97cde09b..7b554184 100644 --- a/crates/mqtt5/tests/clean_start_prebind_subscribe.rs +++ b/crates/mqtt5/tests/clean_start_prebind_subscribe.rs @@ -124,7 +124,7 @@ async fn retained_received_after_takeover_during_hand_off(clean_start: bool) -> #[tokio::test] async fn clean_start_subscribe_during_hand_off_still_gets_retained_message() { assert_eq!( - retained_received_after_takeover_during_hand_off(true).await, + Box::pin(retained_received_after_takeover_during_hand_off(true)).await, 1 ); } @@ -132,7 +132,7 @@ async fn clean_start_subscribe_during_hand_off_still_gets_retained_message() { #[tokio::test] async fn resumed_subscribe_during_hand_off_gets_retained_message() { assert_eq!( - retained_received_after_takeover_during_hand_off(false).await, + Box::pin(retained_received_after_takeover_during_hand_off(false)).await, 1 ); } diff --git a/crates/mqtt5/tests/client_websocket_config.rs b/crates/mqtt5/tests/client_websocket_config.rs new file mode 100644 index 00000000..22f5b1b5 --- /dev/null +++ b/crates/mqtt5/tests/client_websocket_config.rs @@ -0,0 +1,199 @@ +#![cfg(feature = "transport-websocket")] + +//! Checks that `ConnectOptions::with_websocket_config` reaches the upgrade +//! request of every connection `MqttClient` makes, including automatic +//! reconnects, as seen by a server. +//! +//! The wss:// test requires the certificates from +//! `scripts/generate_test_certs.sh`. + +mod wss_test_server; + +use futures::{SinkExt, StreamExt}; +use mqtt5::time::Duration; +use mqtt5::transport::websocket::WebSocketConfig; +use mqtt5::{ConnectOptions, MqttClient}; +use tokio::net::TcpListener; +use tokio::sync::mpsc; +use tokio::time::timeout; +use tokio_tungstenite::tungstenite::handshake::server::{ + Callback, ErrorResponse, Request, Response, +}; +use tokio_tungstenite::tungstenite::http::HeaderMap; +use tokio_tungstenite::tungstenite::Message; +use wss_test_server::{certs, key, server_config, wss_server, CA, SERVER_CERT, SERVER_KEY, WAIT}; + +/// An MQTT v5 CONNACK accepting the connection: no session, success, no +/// properties. +const CONNACK: [u8; 5] = [0x20, 0x03, 0x00, 0x00, 0x00]; + +/// Sends each upgrade request's headers to the test and selects the first +/// offered subprotocol, as a broker would. +struct Capture(mpsc::UnboundedSender); + +impl Callback for Capture { + fn on_request(self, req: &Request, mut resp: Response) -> Result { + let _ = self.0.send(req.headers().clone()); + if let Some(first) = req + .headers() + .get("Sec-WebSocket-Protocol") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.split(',').next()) + { + let first = first.trim().parse().expect("subprotocol header value"); + resp.headers_mut().insert("Sec-WebSocket-Protocol", first); + } + Ok(resp) + } +} + +/// Starts a ws:// server that answers each connection's CONNECT with +/// `connack` and then closes it, so the client reconnects. The receiver yields +/// the headers of every upgrade request. +async fn closing_mqtt_server(connack: Vec) -> (String, mpsc::UnboundedReceiver) { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("addr"); + let (tx, rx) = mpsc::unbounded_channel(); + tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + let capture = Capture(tx.clone()); + let connack = connack.clone(); + tokio::spawn(async move { + let Ok(mut ws) = tokio_tungstenite::accept_hdr_async(stream, capture).await else { + return; + }; + if !matches!(ws.next().await, Some(Ok(Message::Binary(_)))) { + return; + } + let _ = ws.send(Message::Binary(connack.into())).await; + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + }); + } + }); + (format!("ws://{addr}/mqtt"), rx) +} + +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 websocket_config_is_sent_on_connect_and_on_reconnect() { + let (url, mut requests) = closing_mqtt_server(CONNACK.to_vec()).await; + let websocket = WebSocketConfig::new("ws://placeholder.invalid/other") + .expect("config") + .with_header("Authorization", "Bearer token123") + .with_subprotocol("mqttv5.0") + .with_user_agent("configured/1.0"); + let options = ConnectOptions::new("ws-config-client") + .with_websocket_config(websocket) + .with_reconnect_delay(Duration::from_millis(100), Duration::from_millis(500)); + let client = MqttClient::with_options(options); + + timeout(WAIT, client.connect(&url)) + .await + .expect("connect timed out") + .expect("connect"); + + for attempt in ["connect", "reconnect"] { + let headers = timeout(WAIT, requests.recv()) + .await + .unwrap_or_else(|_| panic!("no upgrade request for the {attempt}")) + .expect("server stopped"); + assert_eq!( + header(&headers, "Authorization"), + Some("Bearer token123"), + "{attempt}" + ); + assert_eq!( + header(&headers, "Sec-WebSocket-Protocol"), + Some("mqttv5.0, mqtt"), + "{attempt}" + ); + assert_eq!( + header(&headers, "User-Agent"), + Some("configured/1.0"), + "{attempt}" + ); + assert_eq!( + header(&headers, "Host"), + url.strip_prefix("ws://") + .and_then(|r| r.strip_suffix("/mqtt")), + "{attempt}" + ); + } + + let _ = client.disconnect().await; +} + +fn redirect_connack(server_reference: &str) -> Vec { + let reference = server_reference.as_bytes(); + let reference_len = u16::try_from(reference.len()).expect("reference length"); + let properties_len = u8::try_from(3 + reference.len()).expect("properties length"); + let mut connack = vec![0x20, properties_len + 3, 0x00, 0x9C, properties_len, 0x1C]; + connack.extend_from_slice(&reference_len.to_be_bytes()); + connack.extend_from_slice(reference); + connack +} + +#[tokio::test] +async fn websocket_config_is_not_sent_to_a_redirect_target() { + let (target_url, mut target_requests) = closing_mqtt_server(CONNACK.to_vec()).await; + let (origin_url, mut origin_requests) = + closing_mqtt_server(redirect_connack(&target_url)).await; + let websocket = WebSocketConfig::new("ws://placeholder.invalid/other") + .expect("config") + .with_header("Authorization", "Bearer token123"); + let client = MqttClient::with_options( + ConnectOptions::new("ws-redirect-client") + .with_websocket_config(websocket) + .with_automatic_reconnect(false), + ); + + timeout(WAIT, client.connect(&origin_url)) + .await + .expect("connect timed out") + .expect("connect"); + + let origin = timeout(WAIT, origin_requests.recv()) + .await + .expect("no upgrade request for the origin") + .expect("origin server stopped"); + assert_eq!(header(&origin, "Authorization"), Some("Bearer token123")); + let target = timeout(WAIT, target_requests.recv()) + .await + .expect("no upgrade request for the redirect target") + .expect("target server stopped"); + assert_eq!(header(&target, "Authorization"), None); + + let _ = client.disconnect().await; +} + +#[tokio::test] +async fn websocket_config_tls_is_used_for_wss() { + let (addr, server) = wss_server(server_config( + certs(SERVER_CERT), + key(SERVER_KEY), + false, + &[], + )) + .await; + let websocket = WebSocketConfig::new("wss://localhost/mqtt") + .expect("config") + .with_ca_cert_from_file(CA) + .expect("load CA"); + let client = MqttClient::with_options( + ConnectOptions::new("wss-config-client") + .with_websocket_config(websocket) + .with_automatic_reconnect(false), + ); + + let connecting = tokio::spawn(async move { + let _ = client.connect(&format!("wss://{addr}/mqtt")).await; + }); + timeout(WAIT, server) + .await + .expect("server timed out") + .expect("server saw no completed handshake: the configured CA was not used"); + connecting.abort(); +} diff --git a/crates/mqtt5/tests/quic_multistream_integration.rs b/crates/mqtt5/tests/quic_multistream_integration.rs index ec36b511..301c2e1b 100644 --- a/crates/mqtt5/tests/quic_multistream_integration.rs +++ b/crates/mqtt5/tests/quic_multistream_integration.rs @@ -822,12 +822,12 @@ async fn test_discard_flow_without_flow_headers_returns_error() { #[tokio::test] async fn test_quic_per_publish_under_tight_stream_limit() { - let received = publish_and_count( + let received = Box::pin(publish_and_count( test_quic_config().with_max_concurrent_streams(2), StreamStrategy::DataPerPublish, 20, 16, - ) + )) .await; assert_eq!( received, 20, @@ -837,12 +837,12 @@ async fn test_quic_per_publish_under_tight_stream_limit() { #[tokio::test] async fn test_quic_control_only_under_small_stream_window() { - let received = publish_and_count( + let received = Box::pin(publish_and_count( test_quic_config().with_stream_receive_window(2048), StreamStrategy::ControlOnly, 50, 512, - ) + )) .await; assert_eq!( received, 50, @@ -852,12 +852,12 @@ async fn test_quic_control_only_under_small_stream_window() { #[tokio::test] async fn test_quic_delivery_with_segmentation_offload_disabled() { - let received = publish_and_count( + let received = Box::pin(publish_and_count( test_quic_config().with_disable_segmentation_offload(true), StreamStrategy::ControlOnly, 50, 512, - ) + )) .await; assert_eq!( received, 50,