diff --git a/Cargo.lock b/Cargo.lock index 456df56fb..ffcd09ac1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -600,6 +600,26 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "async-tungstenite" +version = "0.34.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eb26fbd2a93308b1c1b74ac4e494a11ac10db57c476882240573bdf961463520" +dependencies = [ + "atomic-waker", + "futures-core", + "futures-io", + "futures-task", + "futures-util", + "log", + "pin-project-lite", + "rustls-native-certs", + "rustls-pki-types", + "tokio", + "tokio-rustls", + "tungstenite", +] + [[package]] name = "atoi" version = "2.0.0" @@ -3542,9 +3562,9 @@ dependencies = [ [[package]] name = "flume" -version = "0.11.1" +version = "0.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" +checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" dependencies = [ "futures-core", "futures-sink", @@ -5511,6 +5531,16 @@ dependencies = [ "uuid", ] +[[package]] +name = "mqttbytes-core-next" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4f2068d9ea771ce85a7b8caa23d9e542b10b118e99b2b16dbcac626211c9903c" +dependencies = [ + "bytes", + "thiserror 2.0.18", +] + [[package]] name = "multimap" version = "0.10.1" @@ -7508,23 +7538,54 @@ dependencies = [ ] [[package]] -name = "rumqttc" -version = "0.25.1" +name = "rumqttc-core-next" +version = "0.29.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0feff8d882bff0b2fddaf99355a10336d43dd3ed44204f85ece28cf9626ab519" +checksum = "bad77a37d1cf72bf95973879c3b6542530889a27ced6861d98de80145ea0e871" dependencies = [ - "bytes", - "fixedbitset", - "flume", + "async-tungstenite", + "futures-io", "futures-util", - "log", + "http 1.4.0", "rustls-native-certs", - "rustls-pemfile", - "rustls-webpki 0.102.8", + "rustls-pki-types", + "rustls-webpki", + "thiserror 2.0.18", + "tokio", + "tokio-rustls", +] + +[[package]] +name = "rumqttc-next" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "70844faf7766027294c088fdc456096cb0877eb4a346cc8119b7fc444eeb6afa" +dependencies = [ + "rumqttc-v5-next", +] + +[[package]] +name = "rumqttc-v5-next" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8418a883899469a505b30076738ddf344f8c625690fb10602ea965786a094b0" +dependencies = [ + "async-tungstenite", + "bytes", + "fixedbitset", + "flume", + "futures-io", + "futures-util", + "http 1.4.0", + "log", + "mqttbytes-core-next", + "rumqttc-core-next", + "rustls-native-certs", + "rustls-pki-types", + "rustls-webpki", "thiserror 2.0.18", "tokio", "tokio-rustls", - "tokio-stream", "tokio-util", ] @@ -7722,7 +7783,7 @@ dependencies = [ "futures", "hashbrown 0.16.1", "metrics", - "rumqttc", + "rumqttc-next", "rustfs-config", "rustfs-ecstore", "rustfs-s3-common", @@ -8150,7 +8211,7 @@ dependencies = [ "hashbrown 0.16.1", "quick-xml 0.39.2", "rayon", - "rumqttc", + "rumqttc-next", "rustc-hash", "rustfs-config", "rustfs-ecstore", @@ -8454,11 +8515,13 @@ version = "0.0.5" dependencies = [ "async-trait", "criterion", + "hyper-rustls", "reqwest 0.13.2", - "rumqttc", + "rumqttc-next", "rustfs-config", "rustfs-s3-common", "rustfs-utils", + "rustls", "serde", "serde_json", "snap", @@ -8639,7 +8702,7 @@ dependencies = [ "once_cell", "ring", "rustls-pki-types", - "rustls-webpki 0.103.10", + "rustls-webpki", "subtle", "zeroize", ] @@ -8656,15 +8719,6 @@ dependencies = [ "security-framework", ] -[[package]] -name = "rustls-pemfile" -version = "2.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dce314e5fee3f39953d46bb63bb8a46d40c2f8fb7cc5a3b6cab2bde9721d6e50" -dependencies = [ - "rustls-pki-types", -] - [[package]] name = "rustls-pki-types" version = "1.14.0" @@ -8689,7 +8743,7 @@ dependencies = [ "rustls", "rustls-native-certs", "rustls-platform-verifier-android", - "rustls-webpki 0.103.10", + "rustls-webpki", "security-framework", "security-framework-sys", "webpki-root-certs", @@ -8702,17 +8756,6 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" -[[package]] -name = "rustls-webpki" -version = "0.102.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "64ca1bc8749bd4cf37b5ce386cc146580777b4e8572c7b97baf22c83f444bee9" -dependencies = [ - "ring", - "rustls-pki-types", - "untrusted 0.9.0", -] - [[package]] name = "rustls-webpki" version = "0.103.10" @@ -10234,6 +10277,24 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8" +dependencies = [ + "bytes", + "data-encoding", + "http 1.4.0", + "httparse", + "log", + "rand 0.9.2", + "rustls", + "rustls-pki-types", + "sha1 0.10.6", + "thiserror 2.0.18", +] + [[package]] name = "twox-hash" version = "2.1.2" diff --git a/Cargo.toml b/Cargo.toml index 3cd70f6c6..bd7f37777 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -248,7 +248,7 @@ rayon = "1.11.0" reed-solomon-erasure = { version = "6.0", default-features = false, features = ["std", "simd-accel"] } reed-solomon-simd = "3.1.0" regex = { version = "1.12.3" } -rumqttc = { version = "0.25.1" } +rumqttc = { package = "rumqttc-next", version = "0.29.0", features = ["websocket"] } rustix = { version = "1.1.4", features = ["fs"] } rust-embed = { version = "8.11.0" } rustc-hash = { version = "2.1.2" } diff --git a/crates/audit/src/factory.rs b/crates/audit/src/factory.rs index 6674db26c..bdad7415c 100644 --- a/crates/audit/src/factory.rs +++ b/crates/audit/src/factory.rs @@ -19,7 +19,8 @@ use rumqttc::QoS; use rustfs_config::audit::{AUDIT_MQTT_KEYS, AUDIT_WEBHOOK_KEYS, ENV_AUDIT_MQTT_KEYS, ENV_AUDIT_WEBHOOK_KEYS}; use rustfs_config::{ AUDIT_DEFAULT_DIR, DEFAULT_LIMIT, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR, - MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, RUSTFS_WEBHOOK_SKIP_TLS_VERIFY_DEFAULT, + MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MQTT_TLS_POLICY, + MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_TOPIC, MQTT_USERNAME, MQTT_WS_PATH_ALLOWLIST, RUSTFS_WEBHOOK_SKIP_TLS_VERIFY_DEFAULT, WEBHOOK_AUTH_TOKEN, WEBHOOK_CLIENT_CA, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT, WEBHOOK_SKIP_TLS_VERIFY, }; @@ -27,7 +28,10 @@ use rustfs_ecstore::config::KVS; use rustfs_targets::{ Target, error::TargetError, - target::{mqtt::MQTTArgs, webhook::WebhookArgs}, + target::{ + mqtt::{MQTTArgs, MQTTTlsConfig, validate_mqtt_broker_url}, + webhook::WebhookArgs, + }, }; use std::time::Duration; use tracing::{debug, warn}; @@ -166,6 +170,14 @@ impl TargetFactory for MQTTTargetFactory { .and_then(|v| v.parse::().ok()) .map(Duration::from_secs) .unwrap_or_else(|| Duration::from_secs(30)), + tls: MQTTTlsConfig::from_values( + config.lookup(MQTT_TLS_POLICY).as_deref(), + config.lookup(MQTT_TLS_CA).as_deref(), + config.lookup(MQTT_TLS_CLIENT_CERT).as_deref(), + config.lookup(MQTT_TLS_CLIENT_KEY).as_deref(), + config.lookup(MQTT_TLS_TRUST_LEAF_AS_CA).as_deref(), + config.lookup(MQTT_WS_PATH_ALLOWLIST).as_deref(), + )?, queue_dir: config.lookup(MQTT_QUEUE_DIR).unwrap_or(AUDIT_DEFAULT_DIR.to_string()), queue_limit: config .lookup(MQTT_QUEUE_LIMIT) @@ -185,12 +197,15 @@ impl TargetFactory for MQTTTargetFactory { let url = Url::parse(&broker) .map_err(|e| TargetError::Configuration(format!("Invalid broker URL: {e} (value: '{broker}')")))?; - match url.scheme() { - "tcp" | "ssl" | "ws" | "wss" | "mqtt" | "mqtts" => {} - _ => { - return Err(TargetError::Configuration("Unsupported broker URL scheme".to_string())); - } - } + let tls = MQTTTlsConfig::from_values( + config.lookup(MQTT_TLS_POLICY).as_deref(), + config.lookup(MQTT_TLS_CA).as_deref(), + config.lookup(MQTT_TLS_CLIENT_CERT).as_deref(), + config.lookup(MQTT_TLS_CLIENT_KEY).as_deref(), + config.lookup(MQTT_TLS_TRUST_LEAF_AS_CA).as_deref(), + config.lookup(MQTT_WS_PATH_ALLOWLIST).as_deref(), + )?; + validate_mqtt_broker_url(&url, &tls)?; if config.lookup(MQTT_TOPIC).is_none() { return Err(TargetError::Configuration("Missing MQTT topic".to_string())); diff --git a/crates/audit/tests/config_parsing_test.rs b/crates/audit/tests/config_parsing_test.rs index 6b8edceb2..74869efa2 100644 --- a/crates/audit/tests/config_parsing_test.rs +++ b/crates/audit/tests/config_parsing_test.rs @@ -53,6 +53,12 @@ fn test_mqtt_valid_fields() { "reconnect_interval", "queue_dir", "queue_limit", + "tls_policy", + "tls_ca", + "tls_client_cert", + "tls_client_key", + "tls_trust_leaf_as_ca", + "ws_path_allowlist", ]; // This tests the MQTT configuration fields we support diff --git a/crates/config/src/audit/mqtt.rs b/crates/config/src/audit/mqtt.rs index 0c30e64c2..541a8cd8f 100644 --- a/crates/config/src/audit/mqtt.rs +++ b/crates/config/src/audit/mqtt.rs @@ -23,9 +23,15 @@ pub const ENV_AUDIT_MQTT_RECONNECT_INTERVAL: &str = "RUSTFS_AUDIT_MQTT_RECONNECT pub const ENV_AUDIT_MQTT_KEEP_ALIVE_INTERVAL: &str = "RUSTFS_AUDIT_MQTT_KEEP_ALIVE_INTERVAL"; pub const ENV_AUDIT_MQTT_QUEUE_DIR: &str = "RUSTFS_AUDIT_MQTT_QUEUE_DIR"; pub const ENV_AUDIT_MQTT_QUEUE_LIMIT: &str = "RUSTFS_AUDIT_MQTT_QUEUE_LIMIT"; +pub const ENV_AUDIT_MQTT_TLS_POLICY: &str = "RUSTFS_AUDIT_MQTT_TLS_POLICY"; +pub const ENV_AUDIT_MQTT_TLS_CA: &str = "RUSTFS_AUDIT_MQTT_TLS_CA"; +pub const ENV_AUDIT_MQTT_TLS_CLIENT_CERT: &str = "RUSTFS_AUDIT_MQTT_TLS_CLIENT_CERT"; +pub const ENV_AUDIT_MQTT_TLS_CLIENT_KEY: &str = "RUSTFS_AUDIT_MQTT_TLS_CLIENT_KEY"; +pub const ENV_AUDIT_MQTT_TLS_TRUST_LEAF_AS_CA: &str = "RUSTFS_AUDIT_MQTT_TLS_TRUST_LEAF_AS_CA"; +pub const ENV_AUDIT_MQTT_WS_PATH_ALLOWLIST: &str = "RUSTFS_AUDIT_MQTT_WS_PATH_ALLOWLIST"; /// A list of all valid configuration keys for an MQTT target. -pub const ENV_AUDIT_MQTT_KEYS: &[&str; 10] = &[ +pub const ENV_AUDIT_MQTT_KEYS: &[&str; 16] = &[ ENV_AUDIT_MQTT_ENABLE, ENV_AUDIT_MQTT_BROKER, ENV_AUDIT_MQTT_TOPIC, @@ -36,6 +42,12 @@ pub const ENV_AUDIT_MQTT_KEYS: &[&str; 10] = &[ ENV_AUDIT_MQTT_KEEP_ALIVE_INTERVAL, ENV_AUDIT_MQTT_QUEUE_DIR, ENV_AUDIT_MQTT_QUEUE_LIMIT, + ENV_AUDIT_MQTT_TLS_POLICY, + ENV_AUDIT_MQTT_TLS_CA, + ENV_AUDIT_MQTT_TLS_CLIENT_CERT, + ENV_AUDIT_MQTT_TLS_CLIENT_KEY, + ENV_AUDIT_MQTT_TLS_TRUST_LEAF_AS_CA, + ENV_AUDIT_MQTT_WS_PATH_ALLOWLIST, ]; /// A list of all valid configuration keys for an MQTT target. @@ -50,5 +62,11 @@ pub const AUDIT_MQTT_KEYS: &[&str] = &[ crate::MQTT_KEEP_ALIVE_INTERVAL, crate::MQTT_QUEUE_DIR, crate::MQTT_QUEUE_LIMIT, + crate::MQTT_TLS_POLICY, + crate::MQTT_TLS_CA, + crate::MQTT_TLS_CLIENT_CERT, + crate::MQTT_TLS_CLIENT_KEY, + crate::MQTT_TLS_TRUST_LEAF_AS_CA, + crate::MQTT_WS_PATH_ALLOWLIST, crate::COMMENT_KEY, ]; diff --git a/crates/config/src/constants/targets.rs b/crates/config/src/constants/targets.rs index ce341ae44..d6b07faf8 100644 --- a/crates/config/src/constants/targets.rs +++ b/crates/config/src/constants/targets.rs @@ -34,6 +34,12 @@ pub const MQTT_RECONNECT_INTERVAL: &str = "reconnect_interval"; pub const MQTT_KEEP_ALIVE_INTERVAL: &str = "keep_alive_interval"; pub const MQTT_QUEUE_DIR: &str = "queue_dir"; pub const MQTT_QUEUE_LIMIT: &str = "queue_limit"; +pub const MQTT_TLS_POLICY: &str = "tls_policy"; +pub const MQTT_TLS_CA: &str = "tls_ca"; +pub const MQTT_TLS_CLIENT_CERT: &str = "tls_client_cert"; +pub const MQTT_TLS_CLIENT_KEY: &str = "tls_client_key"; +pub const MQTT_TLS_TRUST_LEAF_AS_CA: &str = "tls_trust_leaf_as_ca"; +pub const MQTT_WS_PATH_ALLOWLIST: &str = "ws_path_allowlist"; /// Environment variable controlling whether target queue files are Snappy-compressed. /// Applies to both notify and audit target queue stores. diff --git a/crates/config/src/notify/mqtt.rs b/crates/config/src/notify/mqtt.rs index ca6585ffc..aba56a82e 100644 --- a/crates/config/src/notify/mqtt.rs +++ b/crates/config/src/notify/mqtt.rs @@ -24,6 +24,12 @@ pub const NOTIFY_MQTT_KEYS: &[&str] = &[ crate::MQTT_KEEP_ALIVE_INTERVAL, crate::MQTT_QUEUE_DIR, crate::MQTT_QUEUE_LIMIT, + crate::MQTT_TLS_POLICY, + crate::MQTT_TLS_CA, + crate::MQTT_TLS_CLIENT_CERT, + crate::MQTT_TLS_CLIENT_KEY, + crate::MQTT_TLS_TRUST_LEAF_AS_CA, + crate::MQTT_WS_PATH_ALLOWLIST, crate::COMMENT_KEY, ]; @@ -38,8 +44,14 @@ pub const ENV_NOTIFY_MQTT_RECONNECT_INTERVAL: &str = "RUSTFS_NOTIFY_MQTT_RECONNE pub const ENV_NOTIFY_MQTT_KEEP_ALIVE_INTERVAL: &str = "RUSTFS_NOTIFY_MQTT_KEEP_ALIVE_INTERVAL"; pub const ENV_NOTIFY_MQTT_QUEUE_DIR: &str = "RUSTFS_NOTIFY_MQTT_QUEUE_DIR"; pub const ENV_NOTIFY_MQTT_QUEUE_LIMIT: &str = "RUSTFS_NOTIFY_MQTT_QUEUE_LIMIT"; +pub const ENV_NOTIFY_MQTT_TLS_POLICY: &str = "RUSTFS_NOTIFY_MQTT_TLS_POLICY"; +pub const ENV_NOTIFY_MQTT_TLS_CA: &str = "RUSTFS_NOTIFY_MQTT_TLS_CA"; +pub const ENV_NOTIFY_MQTT_TLS_CLIENT_CERT: &str = "RUSTFS_NOTIFY_MQTT_TLS_CLIENT_CERT"; +pub const ENV_NOTIFY_MQTT_TLS_CLIENT_KEY: &str = "RUSTFS_NOTIFY_MQTT_TLS_CLIENT_KEY"; +pub const ENV_NOTIFY_MQTT_TLS_TRUST_LEAF_AS_CA: &str = "RUSTFS_NOTIFY_MQTT_TLS_TRUST_LEAF_AS_CA"; +pub const ENV_NOTIFY_MQTT_WS_PATH_ALLOWLIST: &str = "RUSTFS_NOTIFY_MQTT_WS_PATH_ALLOWLIST"; -pub const ENV_NOTIFY_MQTT_KEYS: &[&str; 10] = &[ +pub const ENV_NOTIFY_MQTT_KEYS: &[&str; 16] = &[ ENV_NOTIFY_MQTT_ENABLE, ENV_NOTIFY_MQTT_BROKER, ENV_NOTIFY_MQTT_TOPIC, @@ -50,4 +62,10 @@ pub const ENV_NOTIFY_MQTT_KEYS: &[&str; 10] = &[ ENV_NOTIFY_MQTT_KEEP_ALIVE_INTERVAL, ENV_NOTIFY_MQTT_QUEUE_DIR, ENV_NOTIFY_MQTT_QUEUE_LIMIT, + ENV_NOTIFY_MQTT_TLS_POLICY, + ENV_NOTIFY_MQTT_TLS_CA, + ENV_NOTIFY_MQTT_TLS_CLIENT_CERT, + ENV_NOTIFY_MQTT_TLS_CLIENT_KEY, + ENV_NOTIFY_MQTT_TLS_TRUST_LEAF_AS_CA, + ENV_NOTIFY_MQTT_WS_PATH_ALLOWLIST, ]; diff --git a/crates/ecstore/src/config/audit.rs b/crates/ecstore/src/config/audit.rs index 5574daeb4..79c3ec173 100644 --- a/crates/ecstore/src/config/audit.rs +++ b/crates/ecstore/src/config/audit.rs @@ -15,7 +15,8 @@ use crate::config::{KV, KVS}; use rustfs_config::{ COMMENT_KEY, DEFAULT_LIMIT, ENABLE_KEY, EVENT_DEFAULT_DIR, EnableState, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, - MQTT_QOS, MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, WEBHOOK_AUTH_TOKEN, + MQTT_QOS, MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, + MQTT_TLS_POLICY, MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_TOPIC, MQTT_USERNAME, MQTT_WS_PATH_ALLOWLIST, WEBHOOK_AUTH_TOKEN, WEBHOOK_BATCH_SIZE, WEBHOOK_CLIENT_CA, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_HTTP_TIMEOUT, WEBHOOK_MAX_RETRY, WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT, WEBHOOK_RETRY_INTERVAL, WEBHOOK_SKIP_TLS_VERIFY, }; @@ -154,6 +155,36 @@ pub static DEFAULT_AUDIT_MQTT_KVS: LazyLock = LazyLock::new(|| { value: DEFAULT_LIMIT.to_string(), hidden_if_empty: false, }, + KV { + key: MQTT_TLS_POLICY.to_owned(), + value: "".to_owned(), + hidden_if_empty: true, + }, + KV { + key: MQTT_TLS_CA.to_owned(), + value: "".to_owned(), + hidden_if_empty: true, + }, + KV { + key: MQTT_TLS_CLIENT_CERT.to_owned(), + value: "".to_owned(), + hidden_if_empty: true, + }, + KV { + key: MQTT_TLS_CLIENT_KEY.to_owned(), + value: "".to_owned(), + hidden_if_empty: true, + }, + KV { + key: MQTT_TLS_TRUST_LEAF_AS_CA.to_owned(), + value: "".to_owned(), + hidden_if_empty: true, + }, + KV { + key: MQTT_WS_PATH_ALLOWLIST.to_owned(), + value: "".to_owned(), + hidden_if_empty: true, + }, KV { key: COMMENT_KEY.to_owned(), value: "".to_owned(), diff --git a/crates/ecstore/src/config/notify.rs b/crates/ecstore/src/config/notify.rs index 5f3bb9442..05ce628ec 100644 --- a/crates/ecstore/src/config/notify.rs +++ b/crates/ecstore/src/config/notify.rs @@ -15,7 +15,8 @@ use crate::config::{KV, KVS}; use rustfs_config::{ COMMENT_KEY, DEFAULT_LIMIT, ENABLE_KEY, EVENT_DEFAULT_DIR, EnableState, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, - MQTT_QOS, MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, WEBHOOK_AUTH_TOKEN, + MQTT_QOS, MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, + MQTT_TLS_POLICY, MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_TOPIC, MQTT_USERNAME, MQTT_WS_PATH_ALLOWLIST, WEBHOOK_AUTH_TOKEN, WEBHOOK_CLIENT_CA, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT, WEBHOOK_SKIP_TLS_VERIFY, }; @@ -133,6 +134,36 @@ pub static DEFAULT_NOTIFY_MQTT_KVS: LazyLock = LazyLock::new(|| { value: DEFAULT_LIMIT.to_string(), hidden_if_empty: false, }, + KV { + key: MQTT_TLS_POLICY.to_owned(), + value: "".to_owned(), + hidden_if_empty: true, + }, + KV { + key: MQTT_TLS_CA.to_owned(), + value: "".to_owned(), + hidden_if_empty: true, + }, + KV { + key: MQTT_TLS_CLIENT_CERT.to_owned(), + value: "".to_owned(), + hidden_if_empty: true, + }, + KV { + key: MQTT_TLS_CLIENT_KEY.to_owned(), + value: "".to_owned(), + hidden_if_empty: true, + }, + KV { + key: MQTT_TLS_TRUST_LEAF_AS_CA.to_owned(), + value: "".to_owned(), + hidden_if_empty: true, + }, + KV { + key: MQTT_WS_PATH_ALLOWLIST.to_owned(), + value: "".to_owned(), + hidden_if_empty: true, + }, KV { key: COMMENT_KEY.to_owned(), value: "".to_owned(), diff --git a/crates/notify/src/factory.rs b/crates/notify/src/factory.rs index 0d91d8d0a..9c7d79b43 100644 --- a/crates/notify/src/factory.rs +++ b/crates/notify/src/factory.rs @@ -19,7 +19,8 @@ use rumqttc::QoS; use rustfs_config::notify::{ENV_NOTIFY_MQTT_KEYS, ENV_NOTIFY_WEBHOOK_KEYS, NOTIFY_MQTT_KEYS, NOTIFY_WEBHOOK_KEYS}; use rustfs_config::{ DEFAULT_LIMIT, EVENT_DEFAULT_DIR, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR, - MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, RUSTFS_WEBHOOK_SKIP_TLS_VERIFY_DEFAULT, + MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MQTT_TLS_POLICY, + MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_TOPIC, MQTT_USERNAME, MQTT_WS_PATH_ALLOWLIST, RUSTFS_WEBHOOK_SKIP_TLS_VERIFY_DEFAULT, WEBHOOK_AUTH_TOKEN, WEBHOOK_CLIENT_CA, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT, WEBHOOK_SKIP_TLS_VERIFY, }; @@ -27,7 +28,10 @@ use rustfs_ecstore::config::KVS; use rustfs_targets::{ Target, error::TargetError, - target::{mqtt::MQTTArgs, webhook::WebhookArgs}, + target::{ + mqtt::{MQTTArgs, MQTTTlsConfig, validate_mqtt_broker_url}, + webhook::WebhookArgs, + }, }; use std::time::Duration; use tracing::{debug, warn}; @@ -166,6 +170,14 @@ impl TargetFactory for MQTTTargetFactory { .and_then(|v| v.parse::().ok()) .map(Duration::from_secs) .unwrap_or_else(|| Duration::from_secs(30)), + tls: MQTTTlsConfig::from_values( + config.lookup(MQTT_TLS_POLICY).as_deref(), + config.lookup(MQTT_TLS_CA).as_deref(), + config.lookup(MQTT_TLS_CLIENT_CERT).as_deref(), + config.lookup(MQTT_TLS_CLIENT_KEY).as_deref(), + config.lookup(MQTT_TLS_TRUST_LEAF_AS_CA).as_deref(), + config.lookup(MQTT_WS_PATH_ALLOWLIST).as_deref(), + )?, queue_dir: config.lookup(MQTT_QUEUE_DIR).unwrap_or(EVENT_DEFAULT_DIR.to_string()), queue_limit: config .lookup(MQTT_QUEUE_LIMIT) @@ -185,12 +197,15 @@ impl TargetFactory for MQTTTargetFactory { let url = Url::parse(&broker) .map_err(|e| TargetError::Configuration(format!("Invalid broker URL: {e} (value: '{broker}')")))?; - match url.scheme() { - "tcp" | "ssl" | "ws" | "wss" | "mqtt" | "mqtts" => {} - _ => { - return Err(TargetError::Configuration("Unsupported broker URL scheme".to_string())); - } - } + let tls = MQTTTlsConfig::from_values( + config.lookup(MQTT_TLS_POLICY).as_deref(), + config.lookup(MQTT_TLS_CA).as_deref(), + config.lookup(MQTT_TLS_CLIENT_CERT).as_deref(), + config.lookup(MQTT_TLS_CLIENT_KEY).as_deref(), + config.lookup(MQTT_TLS_TRUST_LEAF_AS_CA).as_deref(), + config.lookup(MQTT_WS_PATH_ALLOWLIST).as_deref(), + )?; + validate_mqtt_broker_url(&url, &tls)?; if config.lookup(MQTT_TOPIC).is_none() { return Err(TargetError::Configuration("Missing MQTT topic".to_string())); diff --git a/crates/targets/Cargo.toml b/crates/targets/Cargo.toml index 65fe2079c..5c423af31 100644 --- a/crates/targets/Cargo.toml +++ b/crates/targets/Cargo.toml @@ -13,11 +13,13 @@ documentation = "https://docs.rs/rustfs-target/latest/rustfs_target/" [dependencies] rustfs-config = { workspace = true, features = ["notify", "constants", "audit"] } -rustfs-utils = { workspace = true, features = ["sys", "notify"] } +rustfs-utils = { workspace = true, features = ["sys", "notify", "tls"] } rustfs-s3-common = { workspace = true } async-trait = { workspace = true } +hyper-rustls = { workspace = true } reqwest = { workspace = true } rumqttc = { workspace = true } +rustls = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } snap = { workspace = true } diff --git a/crates/targets/src/check.rs b/crates/targets/src/check.rs index fcc85b507..09d9c8bf8 100644 --- a/crates/targets/src/check.rs +++ b/crates/targets/src/check.rs @@ -21,7 +21,9 @@ /// * `password` - Optional password for authentication /// # Returns /// * `Ok(())` - If the connection is successful -/// * `Err(String)` - If the connection fails, contains an error message +/// * `Err(TargetError)` - If the check fails. +/// `TargetError::Configuration` indicates a bad configuration (invalid URL, TLS settings, etc.). +/// Other variants indicate a connectivity or runtime failure. /// /// # Example /// ```rust,no_run @@ -40,54 +42,53 @@ /// } /// } /// ``` -/// # Note -/// Need to add `rumqttc` and `url` dependencies in `Cargo.toml` -/// ```toml -/// [dependencies] -/// rumqttc = "0.25.0" -/// url = "2.5.7" -/// tokio = { version = "1", features = ["full"] } -/// ``` /// pub async fn check_mqtt_broker_available( broker_url: &str, topic: &str, username: Option<&str>, password: Option<&str>, -) -> Result<(), String> { - use rumqttc::{AsyncClient, MqttOptions, QoS}; - let url = rustfs_utils::parse_url(broker_url).map_err(|e| format!("Broker URL parsing failed:{e}"))?; +) -> Result<(), crate::TargetError> { + use crate::target::mqtt::MQTTTlsConfig; + + check_mqtt_broker_available_with_tls(broker_url, topic, username, password, &MQTTTlsConfig::default()).await +} + +pub async fn check_mqtt_broker_available_with_tls( + broker_url: &str, + topic: &str, + username: Option<&str>, + password: Option<&str>, + tls: &crate::target::mqtt::MQTTTlsConfig, +) -> Result<(), crate::TargetError> { + use crate::target::mqtt::build_mqtt_options; + use rumqttc::{AsyncClient, QoS}; + + let url = rustfs_utils::parse_url(broker_url) + .map_err(|e| crate::TargetError::Configuration(format!("Broker URL parsing failed: {e}")))?; let url = url.url(); - match url.scheme() { - "tcp" | "ssl" | "ws" | "wss" | "mqtt" | "mqtts" | "tls" | "tcps" => {} - _ => return Err("unsupported broker url scheme".to_string()), - } - - let host = url.host_str().ok_or("Broker is missing host")?; - let port = url.port().unwrap_or(1883); - let mut mqtt_options = MqttOptions::new("rustfs_check", host, port); - - // Set credentials if provided - if let Some(user) = username - && !user.is_empty() - { - let pass = password.unwrap_or(""); - mqtt_options.set_credentials(user, pass); - } - - mqtt_options.set_keep_alive(std::time::Duration::from_secs(5)); + // build_mqtt_options returns TargetError directly; Configuration variants propagate as-is. + let mqtt_options = build_mqtt_options( + "rustfs_check".to_string(), + url, + username, + password, + tls, + std::time::Duration::from_secs(5), + None, + )?; let (client, mut eventloop) = AsyncClient::new(mqtt_options, 1); // Try to connect and subscribe client .subscribe(topic, QoS::AtLeastOnce) .await - .map_err(|e| format!("MQTT subscription failed:{e}"))?; + .map_err(|e| crate::TargetError::Network(format!("MQTT subscription failed: {e}")))?; // Wait for eventloop to receive at least one event match tokio::time::timeout(std::time::Duration::from_secs(3), eventloop.poll()).await { Ok(Ok(_)) => Ok(()), - Ok(Err(e)) => Err(format!("MQTT connection failed:{e}")), - Err(_) => Err("MQTT connection timeout".to_string()), + Ok(Err(e)) => Err(crate::TargetError::Network(format!("MQTT connection failed: {e}"))), + Err(_) => Err(crate::TargetError::Timeout("MQTT connection timed out".to_string())), } } diff --git a/crates/targets/src/lib.rs b/crates/targets/src/lib.rs index b46287fc6..cd2fadfdb 100644 --- a/crates/targets/src/lib.rs +++ b/crates/targets/src/lib.rs @@ -18,7 +18,7 @@ pub mod error; pub mod store; pub mod target; -pub use check::check_mqtt_broker_available; +pub use check::{check_mqtt_broker_available, check_mqtt_broker_available_with_tls}; pub use error::{StoreError, TargetError}; pub use rustfs_s3_common::EventName; use serde::{Deserialize, Serialize}; diff --git a/crates/targets/src/target/mqtt.rs b/crates/targets/src/target/mqtt.rs index df32b2fff..d6522fac0 100644 --- a/crates/targets/src/target/mqtt.rs +++ b/crates/targets/src/target/mqtt.rs @@ -20,13 +20,21 @@ use crate::{ target::{ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetType}, }; use async_trait::async_trait; -use rumqttc::{AsyncClient, ConnectionError, EventLoop, MqttOptions, Outgoing, Packet, QoS, mqttbytes::Error as MqttBytesError}; +use hyper_rustls::ConfigBuilderExt; +use rumqttc::{ + AsyncClient, Broker, ConnectionError, EventLoop, Incoming, MqttOptions, Outgoing, QoS, Transport, + mqttbytes::Error as MqttBytesError, +}; +use rustfs_config::{ + EnableState, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_WS_PATH_ALLOWLIST, +}; +use rustls::ClientConfig; use serde::Serialize; use serde::de::DeserializeOwned; use std::sync::Arc; use std::{ marker::PhantomData, - path::PathBuf, + path::{Path, PathBuf}, sync::atomic::{AtomicBool, Ordering}, time::Duration, }; @@ -36,6 +44,381 @@ use url::Url; const DEFAULT_CONNECTION_TIMEOUT: Duration = Duration::from_secs(15); const EVENT_LOOP_POLL_TIMEOUT: Duration = Duration::from_secs(10); // For initial connection check in task +const DEFAULT_MQTT_TCP_PORT: u16 = 1883; +const DEFAULT_MQTT_TLS_PORT: u16 = 8883; +const DEFAULT_MQTT_WSS_PORT: u16 = 443; +const MAX_MQTT_PACKET_SIZE_BYTES: u32 = 100 * 1024 * 1024; +const DEFAULT_MQTT_WS_PATH_ALLOWLIST: &[&str] = &["/", "/mqtt"]; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MQTTTlsPolicy { + SystemCa, + CustomCa, +} + +impl MQTTTlsPolicy { + fn parse(value: &str) -> Result { + match value.trim() { + value if value.eq_ignore_ascii_case("system_ca") => Ok(Self::SystemCa), + value if value.eq_ignore_ascii_case("custom_ca") => Ok(Self::CustomCa), + _ => Err(TargetError::Configuration( + "MQTT tls_policy must be one of: system_ca, custom_ca".to_string(), + )), + } + } +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct MQTTTlsConfig { + pub policy: Option, + pub ca_path: String, + pub client_cert_path: String, + pub client_key_path: String, + pub trust_leaf_as_ca: bool, + pub ws_path_allowlist: Vec, +} + +impl MQTTTlsConfig { + pub fn from_values( + policy: Option<&str>, + ca_path: Option<&str>, + client_cert_path: Option<&str>, + client_key_path: Option<&str>, + trust_leaf_as_ca: Option<&str>, + ws_path_allowlist: Option<&str>, + ) -> Result { + let policy = match policy.map(str::trim).filter(|value| !value.is_empty()) { + Some(value) => Some(MQTTTlsPolicy::parse(value)?), + None => None, + }; + + let trust_leaf_as_ca = match trust_leaf_as_ca.map(str::trim).filter(|value| !value.is_empty()) { + Some(value) => value + .parse::() + .map(EnableState::is_enabled) + .map_err(|_| TargetError::Configuration(format!("Invalid value for {MQTT_TLS_TRUST_LEAF_AS_CA}")))?, + None => false, + }; + + let ws_path_allowlist = match ws_path_allowlist.map(str::trim).filter(|value| !value.is_empty()) { + Some(value) => parse_ws_path_allowlist(value)?, + None => Vec::new(), + }; + + Ok(Self { + policy, + ca_path: ca_path.unwrap_or_default().trim().to_string(), + client_cert_path: client_cert_path.unwrap_or_default().trim().to_string(), + client_key_path: client_key_path.unwrap_or_default().trim().to_string(), + trust_leaf_as_ca, + ws_path_allowlist, + }) + } + + fn effective_ws_path_allowlist(&self) -> Vec<&str> { + if self.ws_path_allowlist.is_empty() { + DEFAULT_MQTT_WS_PATH_ALLOWLIST.to_vec() + } else { + self.ws_path_allowlist.iter().map(String::as_str).collect() + } + } +} + +fn parse_ws_path_allowlist(value: &str) -> Result, TargetError> { + let mut allowlist = Vec::new(); + for raw in value.split(',') { + let path = raw.trim(); + if path.is_empty() { + continue; + } + if !path.starts_with('/') || path.contains('?') || path.contains('#') { + return Err(TargetError::Configuration(format!( + "{MQTT_WS_PATH_ALLOWLIST} entries must be absolute paths without query or fragment" + ))); + } + allowlist.push(path.to_string()); + } + + if allowlist.is_empty() { + return Err(TargetError::Configuration(format!( + "{MQTT_WS_PATH_ALLOWLIST} must contain at least one websocket path" + ))); + } + + Ok(allowlist) +} + +fn keep_alive_seconds(duration: Duration) -> u16 { + duration.as_secs().min(u64::from(u16::MAX)) as u16 +} + +fn default_broker_port(scheme: &str) -> u16 { + match scheme { + "ssl" | "tls" | "tcps" | "mqtts" => DEFAULT_MQTT_TLS_PORT, + "wss" => DEFAULT_MQTT_WSS_PORT, + _ => DEFAULT_MQTT_TCP_PORT, + } +} + +fn websocket_broker_url(broker: &Url, secure: bool) -> Result { + let mut url = broker.clone(); + url.set_scheme("ws") + .map_err(|_| TargetError::Configuration("Failed to normalize websocket broker URL scheme".to_string()))?; + + if secure && url.port().is_none() { + url.set_port(Some(DEFAULT_MQTT_WSS_PORT)) + .map_err(|_| TargetError::Configuration("Failed to set default secure websocket broker port".to_string()))?; + } + + Ok(url.to_string()) +} + +fn ensure_rustls_provider_installed() { + if rustls::crypto::CryptoProvider::get_default().is_none() + && rustls::crypto::aws_lc_rs::default_provider().install_default().is_err() + { + debug!("rustls crypto provider was installed concurrently, skipping aws-lc-rs install"); + } +} + +fn validate_path_is_absolute(path: &str, field: &str) -> Result<(), TargetError> { + if !Path::new(path).is_absolute() { + return Err(TargetError::Configuration(format!("{field} must be an absolute path"))); + } + Ok(()) +} + +fn build_root_store(ca_path: &str, trust_leaf_as_ca: bool) -> Result { + let certs = + rustfs_utils::load_certs(ca_path).map_err(|e| TargetError::Configuration(format!("Failed to load MQTT tls_ca: {e}")))?; + let mut store = rustls::RootCertStore::empty(); + + if trust_leaf_as_ca { + let (valid, invalid) = store.add_parsable_certificates(certs); + if valid == 0 { + return Err(TargetError::Configuration(format!( + "MQTT tls_ca did not contain any parsable trust anchors (ignored {invalid} entries)" + ))); + } + } else { + for cert in certs { + store + .add(cert) + .map_err(|e| TargetError::Configuration(format!("Failed to add MQTT tls_ca to root store: {e}")))?; + } + } + + Ok(store) +} + +fn build_mqtt_tls_transport(broker: &Url, tls: &MQTTTlsConfig) -> Result { + ensure_rustls_provider_installed(); + + let client_config = match tls + .policy + .ok_or_else(|| TargetError::Configuration("Secure MQTT schemes require an explicit tls_policy".to_string()))? + { + MQTTTlsPolicy::SystemCa => { + let builder = ClientConfig::builder() + .with_native_roots() + .map_err(|e| TargetError::Configuration(format!("Failed to load native root certificates: {e}")))?; + + if tls.client_cert_path.is_empty() { + builder.with_no_client_auth() + } else { + let certs = rustfs_utils::load_certs(&tls.client_cert_path) + .map_err(|e| TargetError::Configuration(format!("Failed to load MQTT tls_client_cert: {e}")))?; + let key = rustfs_utils::load_private_key(&tls.client_key_path) + .map_err(|e| TargetError::Configuration(format!("Failed to load MQTT tls_client_key: {e}")))?; + builder + .with_client_auth_cert(certs, key) + .map_err(|e| TargetError::Configuration(format!("Failed to build MQTT client mTLS identity: {e}")))? + } + } + MQTTTlsPolicy::CustomCa => { + let builder = ClientConfig::builder().with_root_certificates(build_root_store(&tls.ca_path, tls.trust_leaf_as_ca)?); + + if tls.client_cert_path.is_empty() { + builder.with_no_client_auth() + } else { + let certs = rustfs_utils::load_certs(&tls.client_cert_path) + .map_err(|e| TargetError::Configuration(format!("Failed to load MQTT tls_client_cert: {e}")))?; + let key = rustfs_utils::load_private_key(&tls.client_key_path) + .map_err(|e| TargetError::Configuration(format!("Failed to load MQTT tls_client_key: {e}")))?; + builder + .with_client_auth_cert(certs, key) + .map_err(|e| TargetError::Configuration(format!("Failed to build MQTT client mTLS identity: {e}")))? + } + } + }; + + if matches!(broker.scheme(), "wss") { + Ok(Transport::wss_with_config(client_config.into())) + } else { + Ok(Transport::tls_with_config(client_config.into())) + } +} + +pub fn validate_mqtt_broker_url(broker: &Url, tls: &MQTTTlsConfig) -> Result<(), TargetError> { + match broker.scheme() { + "ws" | "wss" | "tcp" | "ssl" | "tls" | "tcps" | "mqtt" | "mqtts" => {} + _ => { + return Err(TargetError::Configuration("unknown protocol in broker address".to_string())); + } + } + + if !broker.username().is_empty() || broker.password().is_some() { + return Err(TargetError::Configuration("Broker URL must not embed username or password".to_string())); + } + + broker + .host_str() + .ok_or_else(|| TargetError::Configuration("Broker is missing host".to_string()))?; + + let secure_scheme = matches!(broker.scheme(), "wss" | "ssl" | "tls" | "tcps" | "mqtts"); + let websocket_scheme = matches!(broker.scheme(), "ws" | "wss"); + + if !websocket_scheme { + if !matches!(broker.path(), "" | "/") { + return Err(TargetError::Configuration( + "Broker URL path is only supported for ws/wss schemes".to_string(), + )); + } + + if broker.query().is_some() { + return Err(TargetError::Configuration( + "Broker URL query is only supported for ws/wss schemes".to_string(), + )); + } + + if broker.fragment().is_some() { + return Err(TargetError::Configuration( + "Broker URL fragment is only supported for ws/wss schemes".to_string(), + )); + } + + if !tls.ws_path_allowlist.is_empty() { + return Err(TargetError::Configuration(format!( + "{MQTT_WS_PATH_ALLOWLIST} is only supported for ws/wss schemes" + ))); + } + } else if !tls + .effective_ws_path_allowlist() + .iter() + .any(|allowed_path| *allowed_path == broker.path()) + { + return Err(TargetError::Configuration(format!( + "Websocket broker path '{}' is not in the {MQTT_WS_PATH_ALLOWLIST} allowlist", + broker.path() + ))); + } + + if secure_scheme { + let policy = tls + .policy + .ok_or_else(|| TargetError::Configuration("Secure MQTT schemes require an explicit tls_policy".to_string()))?; + + if !tls.client_cert_path.is_empty() { + validate_path_is_absolute(&tls.client_cert_path, MQTT_TLS_CLIENT_CERT)?; + } + + if !tls.client_key_path.is_empty() { + validate_path_is_absolute(&tls.client_key_path, MQTT_TLS_CLIENT_KEY)?; + } + + if tls.client_cert_path.is_empty() != tls.client_key_path.is_empty() { + return Err(TargetError::Configuration( + "MQTT tls_client_cert and tls_client_key must be specified together".to_string(), + )); + } + + match policy { + MQTTTlsPolicy::SystemCa => { + if !tls.ca_path.is_empty() { + return Err(TargetError::Configuration(format!( + "{MQTT_TLS_CA} is not allowed when tls_policy=system_ca" + ))); + } + if tls.trust_leaf_as_ca { + return Err(TargetError::Configuration(format!( + "{MQTT_TLS_TRUST_LEAF_AS_CA} requires tls_policy=custom_ca" + ))); + } + } + MQTTTlsPolicy::CustomCa => { + if tls.ca_path.is_empty() { + return Err(TargetError::Configuration(format!("{MQTT_TLS_CA} is required when tls_policy=custom_ca"))); + } + validate_path_is_absolute(&tls.ca_path, MQTT_TLS_CA)?; + } + } + } else if tls.policy.is_some() + || !tls.ca_path.is_empty() + || !tls.client_cert_path.is_empty() + || !tls.client_key_path.is_empty() + || tls.trust_leaf_as_ca + { + return Err(TargetError::Configuration( + "TLS settings are only allowed for mqtts/ssl/tls/tcps/wss schemes".to_string(), + )); + } + + Ok(()) +} + +pub(crate) fn build_mqtt_options( + client_id: String, + broker: &Url, + username: Option<&str>, + password: Option<&str>, + tls: &MQTTTlsConfig, + keep_alive: Duration, + max_packet_size: Option, +) -> Result { + validate_mqtt_broker_url(broker, tls)?; + + let host = broker + .host_str() + .ok_or_else(|| TargetError::Configuration("Broker is missing host".to_string()))?; + let port = broker.port().unwrap_or_else(|| default_broker_port(broker.scheme())); + let mut mqtt_options = match broker.scheme() { + "tcp" | "mqtt" => MqttOptions::new(client_id, (host, port)), + "ssl" | "tls" | "tcps" | "mqtts" => { + let mut options = MqttOptions::new(client_id, (host, port)); + options.set_transport(build_mqtt_tls_transport(broker, tls)?); + options + } + "ws" => { + let websocket_broker = Broker::websocket(broker.as_str().to_string()) + .map_err(|e| TargetError::Configuration(format!("Invalid websocket broker URL: {e}")))?; + MqttOptions::new(client_id, websocket_broker) + } + "wss" => { + let websocket_broker = Broker::websocket(websocket_broker_url(broker, true)?) + .map_err(|e| TargetError::Configuration(format!("Invalid secure websocket broker URL: {e}")))?; + let mut options = MqttOptions::new(client_id, websocket_broker); + options.set_transport(build_mqtt_tls_transport(broker, tls)?); + options + } + _ => { + return Err(TargetError::Configuration("unknown protocol in broker address".to_string())); + } + }; + + mqtt_options.set_keep_alive(keep_alive_seconds(keep_alive)); + + if let Some(max_packet_size) = max_packet_size { + mqtt_options.set_max_packet_size(Some(max_packet_size)); + } + + if let Some(user) = username + && !user.is_empty() + { + mqtt_options.set_credentials(user.to_string(), password.unwrap_or("").to_string()); + } + + Ok(mqtt_options) +} /// Arguments for configuring an MQTT target #[derive(Debug, Clone)] @@ -52,6 +435,8 @@ pub struct MQTTArgs { pub username: String, /// The password for the broker pub password: String, + /// Explicit TLS configuration for secure MQTT transports + pub tls: MQTTTlsConfig, /// The maximum interval for reconnection attempts (Note: rumqttc has internal strategy) pub max_reconnect_interval: Duration, /// The keep alive interval @@ -70,12 +455,7 @@ impl MQTTArgs { return Ok(()); } - match self.broker.scheme() { - "ws" | "wss" | "tcp" | "ssl" | "tls" | "tcps" | "mqtt" | "mqtts" => {} - _ => { - return Err(TargetError::Configuration("unknown protocol in broker address".to_string())); - } - } + validate_mqtt_broker_url(&self.broker, &self.tls)?; if self.topic.is_empty() { return Err(TargetError::Configuration("MQTT topic cannot be empty".to_string())); @@ -187,16 +567,15 @@ where .init_cell .get_or_try_init(|| async { debug!(target_id = %target_id_clone, "Initializing MQTT background task."); - let host = args_clone.broker.host_str().unwrap_or("localhost"); - let port = args_clone.broker.port().unwrap_or(1883); - let mut mqtt_options = MqttOptions::new(format!("rustfs_notify_{}", uuid::Uuid::new_v4()), host, port); - mqtt_options - .set_keep_alive(args_clone.keep_alive) - .set_max_packet_size(100 * 1024 * 1024, 100 * 1024 * 1024); // 100MB - - if !args_clone.username.is_empty() { - mqtt_options.set_credentials(args_clone.username.clone(), args_clone.password.clone()); - } + let mqtt_options = build_mqtt_options( + format!("rustfs_notify_{}", uuid::Uuid::new_v4()), + &args_clone.broker, + Some(args_clone.username.as_str()), + Some(args_clone.password.as_str()), + &args_clone.tls, + args_clone.keep_alive, + Some(MAX_MQTT_PACKET_SIZE_BYTES), + )?; let (new_client, eventloop) = AsyncClient::new(mqtt_options, 10); @@ -341,40 +720,40 @@ async fn run_mqtt_event_loop( polled_event_result = async { if !initial_connection_established || !connected_status.load(Ordering::SeqCst) { match tokio::time::timeout(EVENT_LOOP_POLL_TIMEOUT, eventloop.poll()).await { - Ok(Ok(event)) => Ok(event), - Ok(Err(e)) => Err(e), + Ok(result) => Some(result), Err(_) => { debug!(target_id = %target_id, "MQTT poll timed out (EVENT_LOOP_POLL_TIMEOUT) while not connected or status pending."); - Err(ConnectionError::NetworkTimeout) + connected_status.store(false, Ordering::SeqCst); + None } } } else { - eventloop.poll().await + Some(eventloop.poll().await) } } => { match polled_event_result { - Ok(notification) => { + Some(Ok(notification)) => { trace!(target_id = %target_id, event = ?notification, "Received MQTT event"); match notification { - rumqttc::Event::Incoming(Packet::ConnAck(_conn_ack)) => { + rumqttc::Event::Incoming(Incoming::ConnAck(_conn_ack)) => { info!(target_id = %target_id, "MQTT connected (ConnAck)."); connected_status.store(true, Ordering::SeqCst); initial_connection_established = true; } - rumqttc::Event::Incoming(Packet::Publish(publish)) => { - debug!(target_id = %target_id, topic = %publish.topic, payload_len = publish.payload.len(), "Received message on subscribed topic."); + rumqttc::Event::Incoming(Incoming::Publish(publish)) => { + debug!(target_id = %target_id, topic = ?publish.topic, payload_len = publish.payload.len(), "Received message on subscribed topic."); } - rumqttc::Event::Incoming(Packet::Disconnect) => { + rumqttc::Event::Incoming(Incoming::Disconnect(_)) => { info!(target_id = %target_id, "Received Disconnect packet from broker. MQTT connection lost."); connected_status.store(false, Ordering::SeqCst); } - rumqttc::Event::Incoming(Packet::PingResp) => { + rumqttc::Event::Incoming(Incoming::PingResp(_)) => { trace!(target_id = %target_id, "Received PingResp from broker. Connection is alive."); } - rumqttc::Event::Incoming(Packet::SubAck(suback)) => { + rumqttc::Event::Incoming(Incoming::SubAck(suback)) => { trace!(target_id = %target_id, "Received SubAck for pkid: {}", suback.pkid); } - rumqttc::Event::Incoming(Packet::PubAck(puback)) => { + rumqttc::Event::Incoming(Incoming::PubAck(puback)) => { trace!(target_id = %target_id, "Received PubAck for pkid: {}", puback.pkid); } // Process other incoming packet types as needed (PubRec, PubRel, PubComp, UnsubAck) @@ -393,18 +772,13 @@ async fn run_mqtt_event_loop( } } } - Err(e) => { + Some(Err(e)) => { connected_status.store(false, Ordering::SeqCst); error!(target_id = %target_id, error = %e, "Error from MQTT event loop poll"); - if matches!(e, ConnectionError::NetworkTimeout) && (!initial_connection_established || !connected_status.load(Ordering::SeqCst)) { - warn!(target_id = %target_id, "Timeout during initial poll or pending state, will retry."); - continue; - } - if matches!(e, ConnectionError::Io(_) | - ConnectionError::NetworkTimeout | + ConnectionError::Timeout(_) | ConnectionError::ConnectionRefused(_) | ConnectionError::Tls(_) ) { @@ -422,6 +796,10 @@ async fn run_mqtt_event_loop( // Sleep briefly to avoid busy cycles in case of rapid failure. tokio::time::sleep(Duration::from_secs(1)).await; } + None => { + warn!(target_id = %target_id, "Timeout during initial poll or pending state, will retry."); + continue; + } } } } @@ -443,7 +821,7 @@ fn is_fatal_mqtt_error(err: &ConnectionError) -> bool { match state_err { // If StateError is caused by deserialization issues, check the underlying MqttBytesError rumqttc::StateError::Deserialization(mqtt_bytes_err) => { // The type of mqtt_bytes_err is &rumqttc::mqttbytes::Error - matches!( + matches!( mqtt_bytes_err, MqttBytesError::InvalidProtocol // Invalid agreement | MqttBytesError::InvalidProtocolLevel(_) // Invalid protocol level @@ -451,7 +829,7 @@ fn is_fatal_mqtt_error(err: &ConnectionError) -> bool { | MqttBytesError::InvalidPacketType(_) // Invalid package type | MqttBytesError::MalformedPacket // Package format error | MqttBytesError::PayloadTooLong // Too long load - | MqttBytesError::PayloadSizeLimitExceeded(_) // Load size limit exceeded + | MqttBytesError::PayloadSizeLimitExceeded { .. } // Load size limit exceeded | MqttBytesError::TopicNotUtf8 // Topic Non-UTF-8 (Serious Agreement Violation) ) } @@ -640,3 +1018,68 @@ where self.args.enable } } + +#[cfg(test)] +mod tests { + use super::{MQTTTlsConfig, validate_mqtt_broker_url}; + use url::Url; + + #[test] + fn validate_mqtt_broker_url_rejects_non_websocket_path() { + let url = Url::parse("mqtt://broker.example.com:1883/custom").expect("valid url"); + let err = validate_mqtt_broker_url(&url, &MQTTTlsConfig::default()).expect_err("non-websocket path should be rejected"); + assert!(err.to_string().contains("path is only supported")); + } + + #[test] + fn validate_mqtt_broker_url_rejects_non_websocket_query() { + let url = Url::parse("mqtt://broker.example.com:1883?client_id=test").expect("valid url"); + let err = validate_mqtt_broker_url(&url, &MQTTTlsConfig::default()).expect_err("non-websocket query should be rejected"); + assert!(err.to_string().contains("query is only supported")); + } + + #[test] + fn validate_mqtt_broker_url_rejects_non_websocket_fragment() { + let url = Url::parse("mqtt://broker.example.com:1883/#section").expect("valid url"); + let err = + validate_mqtt_broker_url(&url, &MQTTTlsConfig::default()).expect_err("non-websocket fragment should be rejected"); + assert!(err.to_string().contains("fragment is only supported")); + } + + #[test] + fn validate_mqtt_broker_url_allows_websocket_path_and_query() { + let url = Url::parse("ws://broker.example.com:8080/mqtt?client_id=test").expect("valid url"); + validate_mqtt_broker_url(&url, &MQTTTlsConfig::default()).expect("websocket path and query should be allowed"); + } + + #[test] + fn validate_mqtt_broker_url_rejects_url_embedded_credentials() { + let url = Url::parse("mqtt://user:pass@broker.example.com:1883").expect("valid url"); + let err = validate_mqtt_broker_url(&url, &MQTTTlsConfig::default()).expect_err("url credentials should be rejected"); + assert!(err.to_string().contains("must not embed username or password")); + } + + #[test] + fn validate_mqtt_broker_url_requires_explicit_tls_policy_for_secure_scheme() { + let url = Url::parse("mqtts://broker.example.com:8883").expect("valid url"); + let err = validate_mqtt_broker_url(&url, &MQTTTlsConfig::default()) + .expect_err("secure scheme should require explicit tls policy"); + assert!(err.to_string().contains("explicit tls_policy")); + } + + #[test] + fn validate_mqtt_broker_url_rejects_disallowed_websocket_path() { + let url = Url::parse("wss://broker.example.com/private").expect("valid url"); + let tls = MQTTTlsConfig::from_values(Some("system_ca"), None, None, None, None, Some("/mqtt")).expect("valid tls config"); + let err = validate_mqtt_broker_url(&url, &tls).expect_err("path outside allowlist should be rejected"); + assert!(err.to_string().contains("allowlist")); + } + + #[test] + fn validate_mqtt_broker_url_requires_tls_ca_for_custom_ca_policy() { + let url = Url::parse("mqtts://broker.example.com:8883").expect("valid url"); + let tls = MQTTTlsConfig::from_values(Some("custom_ca"), None, None, None, None, None).expect("valid tls config"); + let err = validate_mqtt_broker_url(&url, &tls).expect_err("custom_ca policy without path should be rejected"); + assert!(err.to_string().contains("tls_ca")); + } +} diff --git a/rustfs/src/admin/handlers/audit.rs b/rustfs/src/admin/handlers/audit.rs index 1325afa2d..b6395071f 100644 --- a/rustfs/src/admin/handlers/audit.rs +++ b/rustfs/src/admin/handlers/audit.rs @@ -24,7 +24,7 @@ use rustfs_audit::{audit_system, start_audit_system as start_global_audit_system use rustfs_config::audit::{AUDIT_MQTT_KEYS, AUDIT_MQTT_SUB_SYS, AUDIT_ROUTE_PREFIX, AUDIT_WEBHOOK_KEYS, AUDIT_WEBHOOK_SUB_SYS}; use rustfs_config::{DEFAULT_DELIMITER, ENABLE_KEY, ENV_PREFIX, EnableState, MAX_ADMIN_REQUEST_BODY_SIZE}; use rustfs_ecstore::config::Config; -use rustfs_targets::check_mqtt_broker_available; +use rustfs_targets::{TargetError, check_mqtt_broker_available_with_tls, target::mqtt::MQTTTlsConfig}; use s3s::{Body, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error}; use serde::{Deserialize, Serialize}; use std::collections::{HashMap, HashSet}; @@ -521,9 +521,24 @@ impl Operation for AuditTargetConfig { .ok_or_else(|| s3_error!(InvalidArgument, "topic is required"))?; let username = kv_map.get(rustfs_config::MQTT_USERNAME).map(String::as_str); let password = kv_map.get(rustfs_config::MQTT_PASSWORD).map(String::as_str); - check_mqtt_broker_available(endpoint, topic, username, password) + let tls = MQTTTlsConfig::from_values( + kv_map.get(rustfs_config::MQTT_TLS_POLICY).map(String::as_str), + kv_map.get(rustfs_config::MQTT_TLS_CA).map(String::as_str), + kv_map.get(rustfs_config::MQTT_TLS_CLIENT_CERT).map(String::as_str), + kv_map.get(rustfs_config::MQTT_TLS_CLIENT_KEY).map(String::as_str), + kv_map.get(rustfs_config::MQTT_TLS_TRUST_LEAF_AS_CA).map(String::as_str), + kv_map.get(rustfs_config::MQTT_WS_PATH_ALLOWLIST).map(String::as_str), + ) + .map_err(|e| s3_error!(InvalidArgument, "invalid MQTT TLS settings: {}", e))?; + let parsed_broker = Url::parse(endpoint).map_err(|e| s3_error!(InvalidArgument, "invalid broker URL: {}", e))?; + rustfs_targets::target::mqtt::validate_mqtt_broker_url(&parsed_broker, &tls) + .map_err(|e| s3_error!(InvalidArgument, "{}", e))?; + check_mqtt_broker_available_with_tls(parsed_broker.as_str(), topic, username, password, &tls) .await - .map_err(|e| s3_error!(InvalidArgument, "MQTT Broker unavailable: {}", e))?; + .map_err(|e| match e { + TargetError::Configuration(_) => s3_error!(InvalidArgument, "{}", e), + _ => s3_error!(InvalidArgument, "MQTT broker check failed: {}", e), + })?; if let Some(queue_dir) = kv_map.get("queue_dir") { validate_queue_dir(queue_dir.as_str()).await?; diff --git a/rustfs/src/admin/handlers/event.rs b/rustfs/src/admin/handlers/event.rs index 07b6f73b0..ffb60f572 100644 --- a/rustfs/src/admin/handlers/event.rs +++ b/rustfs/src/admin/handlers/event.rs @@ -25,7 +25,7 @@ use rustfs_config::notify::{ }; use rustfs_config::{DEFAULT_DELIMITER, ENABLE_KEY, ENV_PREFIX, EnableState, MAX_ADMIN_REQUEST_BODY_SIZE}; use rustfs_ecstore::config::Config; -use rustfs_targets::check_mqtt_broker_available; +use rustfs_targets::{TargetError, check_mqtt_broker_available_with_tls, target::mqtt::MQTTTlsConfig}; use s3s::{Body, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error}; use serde::{Deserialize, Serialize}; use std::collections::{HashMap, HashSet}; @@ -450,9 +450,24 @@ impl Operation for NotificationTarget { .ok_or_else(|| s3_error!(InvalidArgument, "topic is required"))?; let username = kv_map.get(rustfs_config::MQTT_USERNAME).map(String::as_str); let password = kv_map.get(rustfs_config::MQTT_PASSWORD).map(String::as_str); - check_mqtt_broker_available(endpoint, topic, username, password) + let tls = MQTTTlsConfig::from_values( + kv_map.get(rustfs_config::MQTT_TLS_POLICY).map(String::as_str), + kv_map.get(rustfs_config::MQTT_TLS_CA).map(String::as_str), + kv_map.get(rustfs_config::MQTT_TLS_CLIENT_CERT).map(String::as_str), + kv_map.get(rustfs_config::MQTT_TLS_CLIENT_KEY).map(String::as_str), + kv_map.get(rustfs_config::MQTT_TLS_TRUST_LEAF_AS_CA).map(String::as_str), + kv_map.get(rustfs_config::MQTT_WS_PATH_ALLOWLIST).map(String::as_str), + ) + .map_err(|e| s3_error!(InvalidArgument, "invalid MQTT TLS settings: {}", e))?; + let parsed_broker = Url::parse(endpoint).map_err(|e| s3_error!(InvalidArgument, "invalid broker URL: {}", e))?; + rustfs_targets::target::mqtt::validate_mqtt_broker_url(&parsed_broker, &tls) + .map_err(|e| s3_error!(InvalidArgument, "{}", e))?; + check_mqtt_broker_available_with_tls(parsed_broker.as_str(), topic, username, password, &tls) .await - .map_err(|e| s3_error!(InvalidArgument, "MQTT Broker unavailable: {}", e))?; + .map_err(|e| match e { + TargetError::Configuration(_) => s3_error!(InvalidArgument, "{}", e), + _ => s3_error!(InvalidArgument, "MQTT broker check failed: {}", e), + })?; if let Some(queue_dir) = kv_map.get("queue_dir") { validate_queue_dir(queue_dir.as_str()).await?;