apply tls flags to wss urls in cli pub, sub and bench

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::<SocketAddr>().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::<SocketAddr>().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
+    );
+}

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<tokio::task::JoinHandle<()>> {
-    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);

crates/mqttv5-cli/src/commands/client_args.rs

@@ -13,6 +13,12 @@ pub fn parse_qos(s: &str) -> Result<QoS, String> {
     }
 }
 
+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<ProtocolVersion, String> {
     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<u64>) -> 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}");
+        }
+    }
 }

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 {

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())
     {