From e53ba71568c30c18d31691d54d90d749d78fb4c0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fabr=C3=ADcio=20Bracht?= Date: Sun, 11 Oct 2026 13:27:19 -0700 Subject: [PATCH] apply tls flags to wss urls in cli pub, sub and bench --- crates/mqtt5/tests/cli_wss_tls.rs | 159 ++++++++++++++++++ crates/mqttv5-cli/src/commands/bench_cmd.rs | 8 +- crates/mqttv5-cli/src/commands/client_args.rs | 28 ++- crates/mqttv5-cli/src/commands/pub_cmd.rs | 6 +- crates/mqttv5-cli/src/commands/sub_cmd.rs | 6 +- 5 files changed, 193 insertions(+), 14 deletions(-) create mode 100644 crates/mqtt5/tests/cli_wss_tls.rs diff --git a/crates/mqtt5/tests/cli_wss_tls.rs b/crates/mqtt5/tests/cli_wss_tls.rs new file mode 100644 index 00000000..05501d6a --- /dev/null +++ b/crates/mqtt5/tests/cli_wss_tls.rs @@ -0,0 +1,159 @@ +#![cfg(feature = "broker")] +mod common; +use common::cli_helpers::{run_cli_command, run_cli_pub, CliResult}; +use common::{find_workspace_root, TestBroker}; +use mqtt5::broker::config::{ + BrokerConfig, StorageBackend, StorageConfig, TlsConfig as BrokerTlsConfig, WebSocketConfig, +}; +use std::net::SocketAddr; +use std::sync::atomic::{AtomicU16, Ordering}; + +static WSS_PORT_COUNTER: AtomicU16 = AtomicU16::new(21200); + +async fn start_wss_broker() -> (TestBroker, String) { + let _ = rustls::crypto::ring::default_provider().install_default(); + + let workspace_root = find_workspace_root(); + let wss_port = WSS_PORT_COUNTER.fetch_add(1, Ordering::SeqCst); + + let tls_config = BrokerTlsConfig::new( + workspace_root.join("test_certs/server.pem"), + workspace_root.join("test_certs/server.key"), + ) + .with_require_client_cert(false) + .with_bind_address("127.0.0.1:0".parse::().unwrap()); + + let ws_tls_config = WebSocketConfig::new() + .with_bind_addresses(vec![format!("127.0.0.1:{wss_port}").parse().unwrap()]) + .with_path("/mqtt".to_string()); + + let storage_config = StorageConfig { + backend: StorageBackend::Memory, + enable_persistence: true, + ..Default::default() + }; + + let config = BrokerConfig::default() + .with_bind_address("127.0.0.1:0".parse::().unwrap()) + .with_storage(storage_config) + .with_tls(tls_config) + .with_websocket_tls(ws_tls_config); + + let broker = TestBroker::start_with_config(config).await; + (broker, format!("wss://localhost:{wss_port}/mqtt")) +} + +fn ca_cert() -> String { + find_workspace_root() + .join("test_certs/ca.pem") + .display() + .to_string() +} + +fn assert_succeeded(command: &str, result: &CliResult) { + assert!( + result.success, + "{command} over wss:// with --ca-cert failed\nstdout: {}\nstderr: {}", + result.stdout, result.stderr + ); +} + +#[tokio::test] +async fn pub_uses_ca_cert_for_wss() { + let (_broker, url) = start_wss_broker().await; + let ca = ca_cert(); + + let result = run_cli_pub(&url, "cli/wss/pub", "hello", &["--ca-cert", &ca]).await; + + assert_succeeded("pub", &result); +} + +#[tokio::test] +async fn sub_uses_ca_cert_for_wss() { + let (_broker, url) = start_wss_broker().await; + let ca = ca_cert(); + + let published = run_cli_pub( + &url, + "cli/wss/sub", + "retained", + &["--ca-cert", &ca, "--retain"], + ) + .await; + assert_succeeded("pub", &published); + + let result = run_cli_command(&[ + "sub", + "--url", + &url, + "--topic", + "cli/wss/sub", + "--count", + "1", + "--non-interactive", + "--ca-cert", + &ca, + ]) + .await; + + assert_succeeded("sub", &result); + assert!( + result.stdout_contains("retained"), + "stdout: {}", + result.stdout + ); +} + +#[tokio::test] +async fn bench_throughput_uses_ca_cert_for_wss() { + let (_broker, url) = start_wss_broker().await; + let ca = ca_cert(); + + let result = run_cli_command(&[ + "bench", + "--url", + &url, + "--duration", + "1", + "--warmup", + "0", + "--ca-cert", + &ca, + ]) + .await; + + assert_succeeded("bench --mode throughput", &result); +} + +#[tokio::test] +async fn bench_connections_uses_ca_cert_for_wss() { + let (_broker, url) = start_wss_broker().await; + let ca = ca_cert(); + + let result = run_cli_command(&[ + "bench", + "--mode", + "connections", + "--url", + &url, + "--duration", + "1", + "--warmup", + "0", + "--concurrency", + "1", + "--ca-cert", + &ca, + ]) + .await; + + assert_succeeded("bench --mode connections", &result); + let output: serde_json::Value = serde_json::from_str(&result.stdout).unwrap(); + let results = &output["results"]; + assert_eq!(results["failed"], 0, "stderr: {}", result.stderr); + assert!( + results["successful"].as_u64().unwrap() > 0, + "stderr: {}", + result.stderr + ); +} diff --git a/crates/mqttv5-cli/src/commands/bench_cmd.rs b/crates/mqttv5-cli/src/commands/bench_cmd.rs index afa87b4b..eb662c87 100644 --- a/crates/mqttv5-cli/src/commands/bench_cmd.rs +++ b/crates/mqttv5-cli/src/commands/bench_cmd.rs @@ -16,6 +16,7 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use tokio::time::Instant; +use super::client_args::uses_tls; use super::parsers::{parse_duration_secs, parse_stream_strategy}; #[derive(Debug, Clone, Copy, ValueEnum, Default)] @@ -563,8 +564,7 @@ async fn configure_transport(client: &MqttClient, cmd: &BenchCommand, url: &str) if cmd.quic_early_data { client.set_quic_early_data(true).await; } - let is_secure = - url.starts_with("ssl://") || url.starts_with("mqtts://") || url.starts_with("quics://"); + let is_secure = uses_tls(url); let has_certs = cmd.cert.is_some() || cmd.key.is_some() || cmd.ca_cert.is_some(); if is_secure && has_certs { let cert_pem = if let Some(p) = &cmd.cert { @@ -1349,9 +1349,7 @@ fn spawn_connection_workers( concurrency: usize, state: &ConnectionBenchState, ) -> Vec> { - let is_secure = state.broker_url.starts_with("ssl://") - || state.broker_url.starts_with("mqtts://") - || state.broker_url.starts_with("quics://"); + let is_secure = uses_tls(&state.broker_url); let has_certs = state.cert_pem.is_some() || state.key_pem.is_some() || state.ca_pem.is_some(); let mut handles = Vec::with_capacity(concurrency); diff --git a/crates/mqttv5-cli/src/commands/client_args.rs b/crates/mqttv5-cli/src/commands/client_args.rs index 4ef2a1d5..fe40937c 100644 --- a/crates/mqttv5-cli/src/commands/client_args.rs +++ b/crates/mqttv5-cli/src/commands/client_args.rs @@ -13,6 +13,12 @@ pub fn parse_qos(s: &str) -> Result { } } +pub fn uses_tls(url: &str) -> bool { + ["mqtts://", "ssl://", "wss://", "quics://"] + .iter() + .any(|scheme| url.starts_with(scheme)) +} + pub fn parse_protocol_version(s: &str) -> Result { match s { "3.1.1" | "311" | "4" => Ok(ProtocolVersion::V311), @@ -150,7 +156,7 @@ pub struct QuicArgs { #[cfg(test)] mod tests { - use super::{SessionArgs, RESUMED_SESSION_EXPIRY_SECS}; + use super::{uses_tls, SessionArgs, RESUMED_SESSION_EXPIRY_SECS}; fn session(no_clean_start: bool, session_expiry: Option) -> SessionArgs { SessionArgs { @@ -179,4 +185,24 @@ mod tests { fn clean_start_without_expiry_sends_none() { assert_eq!(session(false, None).session_expiry_secs(), None); } + + #[test] + fn secure_schemes_use_tls() { + for url in [ + "mqtts://h:8883", + "ssl://h:8883", + "wss://h/mqtt", + "quics://h:14567", + ] { + assert!(uses_tls(url), "{url}"); + } + for url in [ + "mqtt://h:1883", + "tcp://h:1883", + "ws://h/mqtt", + "quic://h:14567", + ] { + assert!(!uses_tls(url), "{url}"); + } + } } diff --git a/crates/mqttv5-cli/src/commands/pub_cmd.rs b/crates/mqttv5-cli/src/commands/pub_cmd.rs index 4f199217..912f0f72 100644 --- a/crates/mqttv5-cli/src/commands/pub_cmd.rs +++ b/crates/mqttv5-cli/src/commands/pub_cmd.rs @@ -16,7 +16,7 @@ use tokio::signal; use tokio::sync::Notify; use tracing::{debug, info, warn}; -use super::client_args::{parse_qos, QuicArgs, SessionArgs, TlsArgs, WillArgs}; +use super::client_args::{parse_qos, uses_tls, QuicArgs, SessionArgs, TlsArgs, WillArgs}; use super::parsers::{ calculate_wait_until, duration_secs_to_u32, parse_duration_millis, parse_duration_secs, }; @@ -395,9 +395,7 @@ async fn configure_tls_certs( broker_url: &str, cmd: &PubCommand, ) -> Result<()> { - let is_secure = broker_url.starts_with("ssl://") - || broker_url.starts_with("mqtts://") - || broker_url.starts_with("quics://"); + let is_secure = uses_tls(broker_url); let has_certs = cmd.tls.cert.is_some() || cmd.tls.key.is_some() || cmd.tls.ca_cert.is_some(); if is_secure && has_certs { diff --git a/crates/mqttv5-cli/src/commands/sub_cmd.rs b/crates/mqttv5-cli/src/commands/sub_cmd.rs index 8ab966e1..0bf215fe 100644 --- a/crates/mqttv5-cli/src/commands/sub_cmd.rs +++ b/crates/mqttv5-cli/src/commands/sub_cmd.rs @@ -12,7 +12,7 @@ use tokio::signal; use tokio::sync::Notify; use tracing::{debug, info}; -use super::client_args::{parse_qos, QuicArgs, SessionArgs, TlsArgs, WillArgs}; +use super::client_args::{parse_qos, uses_tls, QuicArgs, SessionArgs, TlsArgs, WillArgs}; use super::parsers::{duration_secs_to_u32, parse_duration_secs}; #[derive(Args)] @@ -304,9 +304,7 @@ async fn configure_tls_certs( broker_url: &str, cmd: &SubCommand, ) -> Result<()> { - let is_secure = broker_url.starts_with("ssl://") - || broker_url.starts_with("mqtts://") - || broker_url.starts_with("quics://"); + let is_secure = uses_tls(broker_url); if !is_secure || (cmd.tls.cert.is_none() && cmd.tls.key.is_none() && cmd.tls.ca_cert.is_none()) {