diff --git a/CHANGELOG.md b/CHANGELOG.md index a2a1d42c..bae39d67 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,11 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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`. +- **`WebSocketConfig::with_tls_auto`, and the `with_ca_cert_*` and `with_client_auth_*` methods that call it, now accept a host name URL** (#190). They parsed the URL's host and port as a socket address, so they failed with "Invalid host/port combination" for anything but an IP address. A wss:// connection takes its server and certificate name from the URL and does not use `TlsConfig::addr`, so for a host name that is now the unspecified address `0.0.0.0` with the URL's port; set a real address before using such a config with `TlsTransport`. +- **`MqttClient` now uses its stored TLS configuration for `wss://` connections** (#190), as it already did for `mqtts://`. A CA, client certificate or key set with `set_tls_config` was ignored over WebSocket, and `set_insecure_tls(true)` replaced the stored configuration with a bare one that disables verification. - **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 +- **ALPN protocols on a `WebSocketConfig`'s TLS configuration are now offered on `wss://`.** They were ignored before. A WebSocket server negotiates HTTP, so one that advertises only `http/1.1`, as this crate's `wss://` listener does, now rejects a handshake that offers only an MQTT protocol such as `mqtt` with `NoApplicationProtocol`. `MqttClient` does not offer its stored configuration's ALPN protocols on `wss://` (for example those set with `connect_with_tls`, such as AWS IoT's `x-amzn-mqtt-ca`), since they select MQTT directly over TLS. - **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. @@ -28,6 +32,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- `TlsConfig::client_config`, which returns the rustls client configuration a `TlsConfig` describes. - `WebSocketConfig::build_handshake_request`, which returns the upgrade request the configuration produces. ## [mqtt5 0.47.3] - 2026-10-09 diff --git a/crates/mqtt5/Cargo.toml b/crates/mqtt5/Cargo.toml index 9d7ebbc1..e208a4e1 100644 --- a/crates/mqtt5/Cargo.toml +++ b/crates/mqtt5/Cargo.toml @@ -45,7 +45,7 @@ opentelemetry_sdk = ["dep:opentelemetry_sdk"] opentelemetry-otlp = ["dep:opentelemetry-otlp"] tracing-opentelemetry = ["dep:tracing-opentelemetry"] tokio = ["dep:tokio"] -transport-websocket = ["dep:tokio-tungstenite", "dep:futures-util"] +transport-websocket = ["dep:tokio-tungstenite", "dep:futures-util", "dep:rustls-native-certs"] transport-quic = ["dep:quinn"] codec-gzip = ["dep:flate2"] codec-deflate = ["dep:flate2"] @@ -177,6 +177,7 @@ rustls-pki-types = "1.15.0" tokio = { version = "1.47", features = ["full"] } tokio-rustls = { version = "0.26", default-features = false, features = ["ring", "tls12"] } tokio-tungstenite = { version = "0.29", default-features = false, features = ["connect", "rustls-tls-native-roots"], optional = true } +rustls-native-certs = { version = "0.8", optional = true } quinn = { version = "0.11.11", default-features = false, features = ["rustls-ring", "runtime-tokio", "platform-verifier"], optional = true } webpki-roots = "1.0.8" diff --git a/crates/mqtt5/src/client/inner.rs b/crates/mqtt5/src/client/inner.rs index 1013ecfc..f560c784 100644 --- a/crates/mqtt5/src/client/inner.rs +++ b/crates/mqtt5/src/client/inner.rs @@ -282,8 +282,25 @@ impl MqttClient { let mut config = WebSocketConfig::new(url) .map_err(|e| MqttError::ConnectionError(format!("Invalid WebSocket URL: {e}")))?; - if insecure { - let tls_config = TlsConfig::new(addr, host).with_verify_server_cert(false); + // As in connect_tls: a stored TLS config applies to wss:// too. With + // none, an insecure connection still needs one to disable + // verification; a secure one keeps the WebSocket transport's default. + // The stored ALPN protocols are left out: they name the protocol for + // MQTT directly over TLS (mqtts://, e.g. AWS IoT's x-amzn-mqtt-ca), + // while a WebSocket server negotiates HTTP. + let stored = self.tls_config.read().await.clone(); + let tls_config = match stored { + Some(mut cfg) => { + cfg.addr = addr; + cfg.hostname = host.to_string(); + cfg.verify_server_cert = !insecure; + cfg.alpn_protocols = None; + Some(cfg) + } + None if insecure => Some(TlsConfig::new(addr, host).with_verify_server_cert(false)), + None => None, + }; + if let Some(tls_config) = tls_config { config = config.with_tls_config(tls_config); } diff --git a/crates/mqtt5/src/transport/tls.rs b/crates/mqtt5/src/transport/tls.rs index d818df58..5579d658 100644 --- a/crates/mqtt5/src/transport/tls.rs +++ b/crates/mqtt5/src/transport/tls.rs @@ -21,6 +21,13 @@ pub type TlsWriteHalf = WriteHalf>; #[derive(Debug)] pub struct TlsConfig { /// Server address + /// + /// [`TlsTransport`] dials it. A wss:// connection does not use it: the + /// WebSocket URL selects the server, so a config made by + /// [`WebSocketConfig::with_tls_auto`](crate::transport::websocket::WebSocketConfig::with_tls_auto) + /// for a host name URL holds the unspecified address `0.0.0.0` with the + /// URL's port, and must be given a real address before use with + /// [`TlsTransport`]. pub addr: SocketAddr, /// Server hostname for SNI pub hostname: String, @@ -47,10 +54,7 @@ impl Clone for TlsConfig { hostname: self.hostname.clone(), connect_timeout: self.connect_timeout, client_cert: self.client_cert.clone(), - client_key: self.client_key.as_ref().map(|key| { - PrivateKeyDer::try_from(key.secret_der().to_vec()) - .expect("Failed to clone private key") - }), + client_key: self.client_key.as_ref().map(PrivateKeyDer::clone_key), root_certs: self.root_certs.clone(), use_system_roots: self.use_system_roots, verify_server_cert: self.verify_server_cert, @@ -307,6 +311,120 @@ impl TlsConfig { self.root_certs = Some(vec![ca_cert]); Ok(()) } + + /// Builds the rustls client configuration this TLS configuration describes + /// + /// The `addr` and `hostname` fields are not part of it: they select the + /// server to dial and the name to verify, which the caller supplies when + /// connecting. + /// + /// With `use_system_roots`, the system roots are the bundled + /// `webpki-roots`, as [`TlsTransport`] uses them. + /// + /// # Errors + /// + /// Returns an error if a root certificate cannot be added or the client + /// certificate and key cannot be used for client authentication + pub fn client_config(&self) -> Result { + self.client_config_with(SystemRoots::Bundled) + } + + /// Builds the rustls client configuration, taking the system roots that + /// `use_system_roots` adds from `system_roots` + /// + /// Roots are only loaded when the server certificate is verified. + pub(crate) fn client_config_with(&self, system_roots: SystemRoots) -> Result { + let config_builder = if self.verify_server_cert { + let mut root_store = RootCertStore::empty(); + + // Add custom root certificates + if let Some(ref root_certs) = self.root_certs { + for cert in root_certs { + root_store.add(cert.clone()).map_err(|e| { + MqttError::ProtocolError(format!("Failed to add root cert: {e}")) + })?; + } + } + + // Add system roots if requested + if self.use_system_roots { + system_roots.add_to(&mut root_store)?; + } + + ClientConfig::builder().with_root_certificates(root_store) + } else { + // Disable certificate verification for testing with self-signed certs + ClientConfig::builder() + .dangerous() + .with_custom_certificate_verifier(Arc::new(NoVerification)) + }; + + let mut config = if let (Some(cert), Some(key)) = ( + self.client_cert.clone(), + self.client_key.as_ref().map(PrivateKeyDer::clone_key), + ) { + config_builder + .with_client_auth_cert(cert, key) + .map_err(|e| { + MqttError::ProtocolError(format!("Failed to configure client auth: {e}")) + })? + } else { + config_builder.with_no_client_auth() + }; + + // Configure ALPN protocols if provided + if let Some(ref protocols) = self.alpn_protocols { + config.alpn_protocols.clone_from(protocols); + } + + Ok(config) + } +} + +/// Where the roots added by [`TlsConfig::use_system_roots`] come from +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum SystemRoots { + /// The bundled `webpki-roots` + Bundled, + /// The platform's certificate store, loaded with `rustls-native-certs` + #[cfg(feature = "transport-websocket")] + Native, +} + +impl SystemRoots { + // Only loading native roots can fail. + #[cfg_attr(not(feature = "transport-websocket"), allow(clippy::unnecessary_wraps))] + fn add_to(self, root_store: &mut RootCertStore) -> Result<()> { + match self { + Self::Bundled => { + root_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned()); + Ok(()) + } + #[cfg(feature = "transport-websocket")] + Self::Native => add_native_roots(root_store), + } + } +} + +/// Adds the platform's root certificates, as tokio-tungstenite does for a +/// wss:// connection without a TLS configuration: load errors are logged, and +/// only finding no usable certificate at all is an error, unless custom roots +/// were already added. +#[cfg(feature = "transport-websocket")] +fn add_native_roots(root_store: &mut RootCertStore) -> Result<()> { + let loaded = rustls_native_certs::load_native_certs(); + if !loaded.errors.is_empty() { + tracing::warn!(errors = ?loaded.errors, "errors loading native root certificates"); + } + let (added, ignored) = root_store.add_parsable_certificates(loaded.certs); + debug!(added, ignored, "loaded native root certificates"); + if root_store.is_empty() { + return Err(MqttError::ConnectionError(format!( + "no native root certificates found (errors: {:?})", + loaded.errors + ))); + } + Ok(()) } /// TLS transport implementation @@ -349,54 +467,8 @@ impl TlsTransport { /// # Errors /// /// Returns an error if the operation fails - fn build_tls_config(&mut self) -> Result { - let mut root_store = RootCertStore::empty(); - - // Add system roots if requested - if self.config.use_system_roots { - root_store.extend(webpki_roots::TLS_SERVER_ROOTS.to_vec()); - } - - // Add custom root certificates - if let Some(ref root_certs) = self.config.root_certs { - for cert in root_certs { - root_store.add(cert.clone()).map_err(|e| { - MqttError::ProtocolError(format!("Failed to add root cert: {e}")) - })?; - } - } - - let config_builder = if self.config.verify_server_cert { - ClientConfig::builder().with_root_certificates(root_store) - } else { - // Disable certificate verification for testing with self-signed certs - ClientConfig::builder() - .dangerous() - .with_custom_certificate_verifier(Arc::new(NoVerification)) - }; - - let mut config = if let (Some(cert), Some(key)) = ( - self.config.client_cert.clone(), - self.config.client_key.as_ref().map(|k| { - PrivateKeyDer::try_from(k.secret_der().to_vec()) - .expect("Failed to clone private key") - }), - ) { - config_builder - .with_client_auth_cert(cert, key) - .map_err(|e| { - MqttError::ProtocolError(format!("Failed to configure client auth: {e}")) - })? - } else { - config_builder.with_no_client_auth() - }; - - // Configure ALPN protocols if provided - if let Some(ref protocols) = self.config.alpn_protocols { - config.alpn_protocols.clone_from(protocols); - } - - Ok(config) + fn build_tls_config(&self) -> Result { + self.config.client_config() } } diff --git a/crates/mqtt5/src/transport/websocket.rs b/crates/mqtt5/src/transport/websocket.rs index 7bb7fa78..3c53fb89 100644 --- a/crates/mqtt5/src/transport/websocket.rs +++ b/crates/mqtt5/src/transport/websocket.rs @@ -43,12 +43,13 @@ use crate::error::{MqttError, Result}; use crate::packet::Packet; use crate::time::Duration; use crate::transport::packet_io::{decode_buffered_packet, PacketReader, PacketWriter}; -use crate::transport::tls::TlsConfig; +use crate::transport::tls::{SystemRoots, TlsConfig}; use crate::Transport; use bytes::{Buf, Bytes, BytesMut}; use futures_util::{stream::SplitSink, stream::SplitStream, StreamExt}; use std::collections::HashMap; use std::net::SocketAddr; +use std::sync::Arc; use tokio::net::TcpStream; use tokio_tungstenite::{ tungstenite::{ @@ -59,7 +60,7 @@ use tokio_tungstenite::{ }, protocol::Message, }, - MaybeTlsStream, WebSocketStream, + Connector, MaybeTlsStream, WebSocketStream, }; use tracing::{debug, error, info, instrument}; use url::Url; @@ -218,6 +219,24 @@ impl WebSocketConfig { } /// Sets a custom TLS configuration for wss:// connections + /// + /// Its root certificates, system-roots setting, server-certificate + /// verification, client certificate and ALPN protocols are used for the + /// TLS handshake. With `use_system_roots`, the system roots are the + /// platform's native root certificates, as for a wss:// connection with no + /// TLS configuration, rather than the bundled `webpki-roots` that + /// [`TlsTransport`](crate::transport::tls::TlsTransport) uses. + /// Its `addr`, `hostname` and `connect_timeout` are not used: the server + /// and the name verified against its certificate come from the WebSocket + /// URL, and the timeout from [`with_timeout`](Self::with_timeout). + /// + /// ALPN protocols are offered as configured. A WebSocket server negotiates + /// HTTP, so one that advertises only `http/1.1`, as this crate's broker + /// does, rejects a handshake offering only an MQTT protocol such as + /// `mqtt` with `NoApplicationProtocol`. + /// + /// Without a TLS configuration, a wss:// connection verifies the server + /// against the platform's native root certificates. #[must_use] pub fn with_tls_config(mut self, tls_config: TlsConfig) -> Self { self.tls_config = Some(tls_config); @@ -227,14 +246,18 @@ impl WebSocketConfig { /// Creates a TLS configuration automatically from the WebSocket URL /// /// This is a convenience method that creates a TLS config with the same - /// host and port as the WebSocket URL. + /// host and port as the WebSocket URL. The host may be a name or an IP + /// address. A wss:// connection does not use the config's `addr`, so for a + /// host name it is the unspecified address `0.0.0.0` with the URL's port, + /// rather than a resolved one. Set a real address before using the config + /// with [`TlsTransport`](crate::transport::tls::TlsTransport), which would + /// otherwise dial the local host. /// /// # Errors /// /// Returns an error if: /// - The URL is not a secure WebSocket (wss://) /// - The URL does not have a valid host - /// - The host/port combination cannot be parsed as a socket address pub fn with_tls_auto(mut self) -> Result { if !self.is_secure() { return Err(MqttError::ProtocolError( @@ -246,9 +269,10 @@ impl WebSocketConfig { MqttError::ProtocolError("WebSocket URL must have a host".to_string()) })?; - let addr: SocketAddr = format!("{host}:{}", self.port()) - .parse() - .map_err(|e| MqttError::ProtocolError(format!("Invalid host/port combination: {e}")))?; + let port = self.port(); + let addr = format!("{host}:{port}").parse().unwrap_or_else(|_| { + SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED), port) + }); let tls_config = TlsConfig::new(addr, host); self.tls_config = Some(tls_config); @@ -679,40 +703,21 @@ impl Transport for WebSocketTransport { let request = self.config.build_handshake_request()?; - let ws_result = if self.config.is_secure() - && self - .config - .tls_config - .as_ref() - .is_some_and(|cfg| !cfg.verify_server_cert) - { - use tokio_tungstenite::Connector; - - let tls = rustls::ClientConfig::builder() - .dangerous() - .with_custom_certificate_verifier(std::sync::Arc::new(NoVerifier)) - .with_no_client_auth(); - - let connector = Connector::Rustls(std::sync::Arc::new(tls)); - - tokio::time::timeout( - self.config.timeout, - tokio_tungstenite::connect_async_tls_with_config( - request, - None, - false, - Some(connector), - ), - ) - .await - } else { - tokio::time::timeout( - self.config.timeout, - tokio_tungstenite::connect_async(request), - ) - .await + // With no TLS configuration, tokio-tungstenite builds its own client + // config from the platform's native root certificates. + let connector = match self.config.tls_config.as_ref() { + Some(tls_config) if self.config.is_secure() => Some(Connector::Rustls(Arc::new( + tls_config.client_config_with(SystemRoots::Native)?, + ))), + _ => None, }; + let ws_result = tokio::time::timeout( + self.config.timeout, + tokio_tungstenite::connect_async_tls_with_config(request, None, false, connector), + ) + .await; + match ws_result { Ok(Ok((ws_stream, response))) => { if let Some(protocol) = response.headers().get("Sec-WebSocket-Protocol") { @@ -825,54 +830,6 @@ impl Transport for WebSocketTransport { } } -#[derive(Debug)] -struct NoVerifier; - -impl rustls::client::danger::ServerCertVerifier for NoVerifier { - fn verify_server_cert( - &self, - _end_entity: &rustls::pki_types::CertificateDer<'_>, - _intermediates: &[rustls::pki_types::CertificateDer<'_>], - _server_name: &rustls::pki_types::ServerName<'_>, - _ocsp_response: &[u8], - _now: rustls::pki_types::UnixTime, - ) -> std::result::Result { - Ok(rustls::client::danger::ServerCertVerified::assertion()) - } - - fn verify_tls12_signature( - &self, - _message: &[u8], - _cert: &rustls::pki_types::CertificateDer<'_>, - _dss: &rustls::DigitallySignedStruct, - ) -> std::result::Result { - Ok(rustls::client::danger::HandshakeSignatureValid::assertion()) - } - - fn verify_tls13_signature( - &self, - _message: &[u8], - _cert: &rustls::pki_types::CertificateDer<'_>, - _dss: &rustls::DigitallySignedStruct, - ) -> std::result::Result { - Ok(rustls::client::danger::HandshakeSignatureValid::assertion()) - } - - fn supported_verify_schemes(&self) -> Vec { - vec![ - rustls::SignatureScheme::RSA_PKCS1_SHA256, - rustls::SignatureScheme::RSA_PKCS1_SHA384, - rustls::SignatureScheme::RSA_PKCS1_SHA512, - rustls::SignatureScheme::ECDSA_NISTP256_SHA256, - rustls::SignatureScheme::ECDSA_NISTP384_SHA384, - rustls::SignatureScheme::RSA_PSS_SHA256, - rustls::SignatureScheme::RSA_PSS_SHA384, - rustls::SignatureScheme::RSA_PSS_SHA512, - rustls::SignatureScheme::ED25519, - ] - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/crates/mqtt5/tests/websocket_tls_integration.rs b/crates/mqtt5/tests/websocket_tls_integration.rs new file mode 100644 index 00000000..297cdbf9 --- /dev/null +++ b/crates/mqtt5/tests/websocket_tls_integration.rs @@ -0,0 +1,244 @@ +#![cfg(feature = "transport-websocket")] + +//! Checks that a `wss://` connection uses `WebSocketConfig::tls_config`: its +//! root certificates, client certificate and ALPN protocols, as seen by a +//! TLS server, rather than only being stored on the configuration. +//! +//! Requires the certificates from `scripts/generate_test_certs.sh`. + +mod wss_test_server; + +use mqtt5::transport::tls::TlsConfig; +use mqtt5::transport::websocket::WebSocketConfig; +use mqtt5::{ConnectOptions, MqttClient}; +use rustls::pki_types::PrivateKeyDer; +use tokio::time::timeout; +use wss_test_server::{ + certs, connect, key, server_config, wss_config, wss_server, CA, CLIENT_CERT, CLIENT_KEY, + SERVER_CERT, SERVER_KEY, WAIT, +}; + +#[tokio::test] +async fn custom_ca_from_tls_config_is_trusted() { + let (addr, server) = wss_server(server_config( + certs(SERVER_CERT), + key(SERVER_KEY), + false, + &[], + )) + .await; + let config = wss_config(addr) + .with_ca_cert_from_file(CA) + .expect("load CA"); + + connect(config, server) + .await + .expect("server certificate issued by the configured CA was rejected"); +} + +#[tokio::test] +async fn certificate_from_another_ca_is_rejected() { + let other = rcgen::generate_simple_self_signed(vec!["127.0.0.1".to_string()]) + .expect("generate certificate"); + let (addr, server) = wss_server(server_config( + vec![other.cert.der().clone()], + PrivateKeyDer::try_from(other.signing_key.serialize_der()).expect("key"), + false, + &[], + )) + .await; + let config = wss_config(addr) + .with_ca_cert_from_file(CA) + .expect("load CA"); + + let err = connect(config, server) + .await + .expect_err("certificate not issued by the configured CA was accepted"); + assert!(err.contains("UnknownIssuer"), "unexpected error: {err}"); +} + +#[tokio::test] +async fn client_certificate_from_tls_config_is_presented() { + let (addr, server) = wss_server(server_config( + certs(SERVER_CERT), + key(SERVER_KEY), + true, + &[], + )) + .await; + let config = wss_config(addr) + .with_client_auth_from_files(CLIENT_CERT, CLIENT_KEY) + .expect("load client auth") + .with_ca_cert_from_file(CA) + .expect("load CA"); + + let handshake = connect(config, server) + .await + .expect("server requiring a client certificate rejected the connection"); + assert!(handshake.client_presented_cert); +} + +#[tokio::test] +async fn client_certificate_is_presented_with_verification_disabled() { + let (addr, server) = wss_server(server_config( + certs(SERVER_CERT), + key(SERVER_KEY), + true, + &[], + )) + .await; + let mut tls_config = TlsConfig::new(addr, "127.0.0.1").with_verify_server_cert(false); + tls_config + .load_client_cert_pem(CLIENT_CERT) + .expect("load client cert"); + tls_config + .load_client_key_pem(CLIENT_KEY) + .expect("load client key"); + let config = wss_config(addr).with_tls_config(tls_config); + + let handshake = connect(config, server) + .await + .expect("server requiring a client certificate rejected the connection"); + assert!(handshake.client_presented_cert); +} + +#[tokio::test] +async fn alpn_protocols_from_tls_config_are_offered() { + let (addr, server) = wss_server(server_config( + certs(SERVER_CERT), + key(SERVER_KEY), + false, + &[b"mqtt-test", b"http/1.1"], + )) + .await; + let mut tls_config = TlsConfig::new(addr, "127.0.0.1").with_alpn_protocols(&["mqtt-test"]); + tls_config.load_ca_cert_pem(CA).expect("load CA"); + let config = wss_config(addr).with_tls_config(tls_config); + + let handshake = connect(config, server).await.expect("connect"); + assert_eq!(handshake.alpn.as_deref(), Some(&b"mqtt-test"[..])); +} + +#[tokio::test] +async fn without_tls_config_private_ca_is_not_trusted() { + let (addr, server) = wss_server(server_config( + certs(SERVER_CERT), + key(SERVER_KEY), + false, + &[], + )) + .await; + + let err = connect(wss_config(addr), server) + .await + .expect_err("certificate from a private CA was accepted with no TLS configuration"); + assert!(err.contains("UnknownIssuer"), "unexpected error: {err}"); +} + +#[tokio::test] +async fn custom_ca_works_with_a_host_name_url() { + let (addr, server) = wss_server(server_config( + certs(SERVER_CERT), + key(SERVER_KEY), + false, + &[], + )) + .await; + let config = WebSocketConfig::new(&format!("wss://localhost:{}/mqtt", addr.port())) + .expect("config") + .with_ca_cert_from_file(CA) + .expect("a host name URL should accept a CA"); + + connect(config, server) + .await + .expect("server certificate for localhost, issued by the configured CA, was rejected"); +} + +#[tokio::test] +async fn certificate_is_verified_against_the_url_host_not_tls_config_hostname() { + let (addr, server) = wss_server(server_config( + certs(SERVER_CERT), + key(SERVER_KEY), + false, + &[], + )) + .await; + let mut tls_config = TlsConfig::new(addr, "wrong.example"); + tls_config.load_ca_cert_pem(CA).expect("load CA"); + let config = wss_config(addr).with_tls_config(tls_config); + + connect(config, server) + .await + .expect("TlsConfig::hostname was used instead of the URL host"); +} + +#[tokio::test] +async fn mqtt_client_presents_its_stored_tls_config_over_wss() { + let (addr, server) = wss_server(server_config( + certs(SERVER_CERT), + key(SERVER_KEY), + true, + &[], + )) + .await; + let client = MqttClient::with_options( + ConnectOptions::new("wss-stored-tls").with_automatic_reconnect(false), + ); + client + .set_tls_config( + Some(std::fs::read(CLIENT_CERT).expect("read client cert")), + Some(std::fs::read(CLIENT_KEY).expect("read client key")), + Some(std::fs::read(CA).expect("read CA")), + ) + .await; + + // The test server stops after the WebSocket handshake and never answers + // CONNECT, so only the TLS handshake it observed is checked. + let connecting = tokio::spawn(async move { + let _ = client.connect(&format!("wss://{addr}/mqtt")).await; + }); + let handshake = timeout(WAIT, server) + .await + .expect("server timed out") + .expect("server saw no completed handshake: the stored TLS config was not used"); + assert!(handshake.client_presented_cert); + connecting.abort(); +} + +#[tokio::test] +async fn mqtt_client_does_not_offer_its_stored_alpn_over_wss() { + // Like this crate's wss:// listener, the server advertises only http/1.1, + // so a client offering an MQTT protocol would get NoApplicationProtocol. + let (addr, server) = wss_server(server_config( + certs(SERVER_CERT), + key(SERVER_KEY), + false, + &[b"http/1.1"], + )) + .await; + let client = MqttClient::with_options( + ConnectOptions::new("wss-stored-alpn").with_automatic_reconnect(false), + ); + + // connect_with_tls stores its configuration before dialing; a closed port + // makes it fail at once, leaving that configuration stored. + let closed = std::net::TcpListener::bind("127.0.0.1:0") + .and_then(|listener| listener.local_addr()) + .expect("pick a port"); + let mut stored = TlsConfig::new(closed, "127.0.0.1").with_alpn_protocols(&["mqtt"]); + stored.load_ca_cert_pem(CA).expect("load CA"); + client + .connect_with_tls(stored) + .await + .expect_err("nothing listens on the closed port"); + + let connecting = tokio::spawn(async move { + let _ = client.connect(&format!("wss://{addr}/mqtt")).await; + }); + let handshake = timeout(WAIT, server) + .await + .expect("server timed out") + .expect("server saw no completed handshake: the stored ALPN was offered"); + assert_eq!(handshake.alpn, None); + connecting.abort(); +} diff --git a/crates/mqtt5/tests/websocket_tls_native_roots.rs b/crates/mqtt5/tests/websocket_tls_native_roots.rs new file mode 100644 index 00000000..d16c0126 --- /dev/null +++ b/crates/mqtt5/tests/websocket_tls_native_roots.rs @@ -0,0 +1,99 @@ +#![cfg(feature = "transport-websocket")] + +//! Checks that `use_system_roots` on a wss:// connection's TLS configuration +//! means the platform's root certificates, as it does for a wss:// connection +//! with no TLS configuration. +//! +//! `SSL_CERT_FILE` stands in for a CA installed on the platform: +//! rustls-native-certs loads that file instead of the platform store. The +//! variable is process-wide, so these tests have their own binary. +//! +//! Requires the certificates from `scripts/generate_test_certs.sh`. + +mod wss_test_server; + +use mqtt5::transport::tls::TlsConfig; +use mqtt5::{ConnectOptions, MqttClient}; +use std::sync::Once; +use tokio::time::timeout; +use wss_test_server::{ + certs, connect, key, server_config, wss_config, wss_server, CA, CLIENT_CERT, CLIENT_KEY, + SERVER_CERT, SERVER_KEY, WAIT, +}; + +/// Makes the test CA a platform root for this process. Every test calls it +/// before connecting, and it writes the variable only once. +fn install_test_ca_as_platform_root() { + static INSTALL: Once = Once::new(); + INSTALL.call_once(|| std::env::set_var("SSL_CERT_FILE", CA)); +} + +#[tokio::test] +async fn platform_roots_are_used_without_a_configured_ca() { + install_test_ca_as_platform_root(); + let (addr, server) = wss_server(server_config( + certs(SERVER_CERT), + key(SERVER_KEY), + false, + &[b"mqtt-test"], + )) + .await; + let tls_config = TlsConfig::new(addr, "127.0.0.1").with_alpn_protocols(&["mqtt-test"]); + + let handshake = connect(wss_config(addr).with_tls_config(tls_config), server) + .await + .expect("server certificate from a platform-installed CA was rejected"); + assert_eq!(handshake.alpn.as_deref(), Some(&b"mqtt-test"[..])); +} + +#[tokio::test] +async fn platform_roots_are_not_used_when_system_roots_are_off() { + install_test_ca_as_platform_root(); + let (addr, server) = wss_server(server_config( + certs(SERVER_CERT), + key(SERVER_KEY), + false, + &[], + )) + .await; + let tls_config = TlsConfig::new(addr, "127.0.0.1").with_system_roots(false); + + let err = connect(wss_config(addr).with_tls_config(tls_config), server) + .await + .expect_err("server certificate was trusted with no roots configured"); + assert!(err.contains("UnknownIssuer"), "unexpected error: {err}"); +} + +#[tokio::test] +async fn mqtt_client_with_cert_and_key_but_no_ca_trusts_platform_roots() { + install_test_ca_as_platform_root(); + let (addr, server) = wss_server(server_config( + certs(SERVER_CERT), + key(SERVER_KEY), + true, + &[], + )) + .await; + let client = MqttClient::with_options( + ConnectOptions::new("wss-platform-roots").with_automatic_reconnect(false), + ); + client + .set_tls_config( + Some(std::fs::read(CLIENT_CERT).expect("read client cert")), + Some(std::fs::read(CLIENT_KEY).expect("read client key")), + None, + ) + .await; + + // The test server stops after the WebSocket handshake and never answers + // CONNECT, so only the TLS handshake it observed is checked. + let connecting = tokio::spawn(async move { + let _ = client.connect(&format!("wss://{addr}/mqtt")).await; + }); + let handshake = timeout(WAIT, server) + .await + .expect("server timed out") + .expect("server saw no completed handshake: the platform roots were not used"); + assert!(handshake.client_presented_cert); + connecting.abort(); +} diff --git a/crates/mqtt5/tests/wss_test_server/mod.rs b/crates/mqtt5/tests/wss_test_server/mod.rs new file mode 100644 index 00000000..a55c7c0d --- /dev/null +++ b/crates/mqtt5/tests/wss_test_server/mod.rs @@ -0,0 +1,143 @@ +//! A loopback WSS server for the wss:// TLS tests, reporting what it observed +//! about each TLS handshake. +//! +//! Requires the certificates from `scripts/generate_test_certs.sh`. +#![allow(dead_code)] + +use mqtt5::transport::websocket::{WebSocketConfig, WebSocketTransport}; +use mqtt5::Transport; +use rustls::pki_types::pem::PemObject; +use rustls::pki_types::{CertificateDer, PrivateKeyDer}; +use rustls::server::WebPkiClientVerifier; +use rustls::{RootCertStore, ServerConfig}; +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; +use tokio::net::TcpListener; +use tokio::sync::oneshot; +use tokio::time::timeout; +use tokio_rustls::TlsAcceptor; +use tokio_tungstenite::tungstenite::handshake::server::{ + Callback, ErrorResponse, Request, Response, +}; +use tokio_tungstenite::tungstenite::http::HeaderValue; + +pub const WAIT: Duration = Duration::from_secs(5); +pub const CA: &str = "../../test_certs/ca.pem"; +pub const SERVER_CERT: &str = "../../test_certs/server.pem"; +pub const SERVER_KEY: &str = "../../test_certs/server.key"; +pub const CLIENT_CERT: &str = "../../test_certs/client.pem"; +pub const CLIENT_KEY: &str = "../../test_certs/client.key"; + +/// What the server observed about a TLS handshake that completed. +#[derive(Debug)] +pub struct Handshake { + pub alpn: Option>, + pub client_presented_cert: bool, +} + +pub fn certs(path: &str) -> Vec> { + CertificateDer::pem_file_iter(path) + .unwrap_or_else(|e| panic!("read {path}: {e}")) + .collect::>() + .unwrap_or_else(|e| panic!("parse {path}: {e}")) +} + +pub fn key(path: &str) -> PrivateKeyDer<'static> { + PrivateKeyDer::from_pem_file(path).unwrap_or_else(|e| panic!("read {path}: {e}")) +} + +pub fn server_config( + cert: Vec>, + key: PrivateKeyDer<'static>, + require_client_cert: bool, + alpn: &[&[u8]], +) -> ServerConfig { + let _ = rustls::crypto::ring::default_provider().install_default(); + let builder = ServerConfig::builder(); + let mut config = if require_client_cert { + let mut roots = RootCertStore::empty(); + for ca in certs(CA) { + roots.add(ca).expect("add CA"); + } + let verifier = WebPkiClientVerifier::builder(Arc::new(roots)) + .build() + .expect("client verifier"); + builder.with_client_cert_verifier(verifier) + } else { + builder.with_no_client_auth() + } + .with_single_cert(cert, key) + .expect("server cert"); + config.alpn_protocols = alpn.iter().map(|p| p.to_vec()).collect(); + config +} + +/// Starts a one-connection WSS server. The receiver yields the handshake +/// details if the TLS and WebSocket handshakes both complete. +pub async fn wss_server(config: ServerConfig) -> (SocketAddr, oneshot::Receiver) { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("addr"); + let acceptor = TlsAcceptor::from(Arc::new(config)); + let (tx, rx) = oneshot::channel(); + tokio::spawn(async move { + let Ok((stream, _)) = listener.accept().await else { + return; + }; + let Ok(tls) = acceptor.accept(stream).await else { + return; + }; + let (_, session) = tls.get_ref(); + let handshake = Handshake { + alpn: session.alpn_protocol().map(<[u8]>::to_vec), + client_presented_cert: session.peer_certificates().is_some(), + }; + let Ok(ws) = tokio_tungstenite::accept_hdr_async(tls, SelectMqtt).await else { + return; + }; + let _ = tx.send(handshake); + // Hold the connection open until the client closes it. + let _ws = ws; + tokio::time::sleep(WAIT).await; + }); + (addr, rx) +} + +/// Selects the "mqtt" subprotocol the client offers by default, as a broker +/// would; the client fails the handshake if the server selects none. +pub struct SelectMqtt; + +impl Callback for SelectMqtt { + fn on_request(self, _req: &Request, mut resp: Response) -> Result { + resp.headers_mut() + .insert("Sec-WebSocket-Protocol", HeaderValue::from_static("mqtt")); + Ok(resp) + } +} + +pub fn wss_config(addr: SocketAddr) -> WebSocketConfig { + WebSocketConfig::new(&format!("wss://{addr}/mqtt")).expect("config") +} + +/// Connects and returns the server's view of the handshake, or the client's +/// error if the connection failed. +pub async fn connect( + config: WebSocketConfig, + server: oneshot::Receiver, +) -> Result { + let mut transport = WebSocketTransport::new(config); + let result = timeout(WAIT, transport.connect()) + .await + .expect("connect timed out"); + match result { + Ok(()) => { + let handshake = timeout(WAIT, server) + .await + .expect("server timed out") + .expect("server saw no completed handshake"); + transport.close().await.expect("close"); + Ok(handshake) + } + Err(e) => Err(e.to_string()), + } +}