gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-04 11:15:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(mqtt): migrate client and harden TLS config (#2413)

Browse Source 
Signed-off-by: houseme <housemecn@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: houseme <4829346+houseme@users.noreply.github.com>
Co-authored-by: 安正超 <anzhengchao@gmail.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
houseme
2026-04-07 15:28:29 +08:00
committed by GitHub
parent 9ed6487e2f
commit d6158c0481
16 changed files with 815 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+98 -37
View File
@@ -600,6 +600,26 @@ dependencies = [
"syn 2.0.117", "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]] [[package]]
name = "atoi" name = "atoi"
version = "2.0.0" version = "2.0.0"
@@ -3542,9 +3562,9 @@ dependencies = [
[[package]] [[package]]
name = "flume" name = "flume"
version = "0.11.1" version = "0.12.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be"
dependencies = [ dependencies = [
"futures-core", "futures-core",
"futures-sink", "futures-sink",
@@ -5511,6 +5531,16 @@ dependencies = [
"uuid", "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]] [[package]]
name = "multimap" name = "multimap"
version = "0.10.1" version = "0.10.1"
@@ -7508,23 +7538,54 @@ dependencies = [
] ]
[[package]] [[package]]
name = "rumqttc" name = "rumqttc-core-next"
version = "0.25.1" version = "0.29.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0feff8d882bff0b2fddaf99355a10336d43dd3ed44204f85ece28cf9626ab519" checksum = "bad77a37d1cf72bf95973879c3b6542530889a27ced6861d98de80145ea0e871"
dependencies = [ dependencies = [
"bytes", "async-tungstenite",
"fixedbitset", "futures-io",
"flume",
"futures-util", "futures-util",
"log", "http 1.4.0",
"rustls-native-certs", "rustls-native-certs",
"rustls-pemfile", "rustls-pki-types",
"rustls-webpki 0.102.8", "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", "thiserror 2.0.18",
"tokio", "tokio",
"tokio-rustls", "tokio-rustls",
"tokio-stream",
"tokio-util", "tokio-util",
] ]
@@ -7722,7 +7783,7 @@ dependencies = [
"futures", "futures",
"hashbrown 0.16.1", "hashbrown 0.16.1",
"metrics", "metrics",
"rumqttc", "rumqttc-next",
"rustfs-config", "rustfs-config",
"rustfs-ecstore", "rustfs-ecstore",
"rustfs-s3-common", "rustfs-s3-common",
@@ -8150,7 +8211,7 @@ dependencies = [
"hashbrown 0.16.1", "hashbrown 0.16.1",
"quick-xml 0.39.2", "quick-xml 0.39.2",
"rayon", "rayon",
"rumqttc", "rumqttc-next",
"rustc-hash", "rustc-hash",
"rustfs-config", "rustfs-config",
"rustfs-ecstore", "rustfs-ecstore",
@@ -8454,11 +8515,13 @@ version = "0.0.5"
dependencies = [ dependencies = [
"async-trait", "async-trait",
"criterion", "criterion",
"hyper-rustls",
"reqwest 0.13.2", "reqwest 0.13.2",
"rumqttc", "rumqttc-next",
"rustfs-config", "rustfs-config",
"rustfs-s3-common", "rustfs-s3-common",
"rustfs-utils", "rustfs-utils",
"rustls",
"serde", "serde",
"serde_json", "serde_json",
"snap", "snap",
@@ -8639,7 +8702,7 @@ dependencies = [
"once_cell", "once_cell",
"ring", "ring",
"rustls-pki-types", "rustls-pki-types",
"rustls-webpki 0.103.10", "rustls-webpki",
"subtle", "subtle",
"zeroize", "zeroize",
] ]
@@ -8656,15 +8719,6 @@ dependencies = [
"security-framework", "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]] [[package]]
name = "rustls-pki-types" name = "rustls-pki-types"
version = "1.14.0" version = "1.14.0"
@@ -8689,7 +8743,7 @@ dependencies = [
"rustls", "rustls",
"rustls-native-certs", "rustls-native-certs",
"rustls-platform-verifier-android", "rustls-platform-verifier-android",
"rustls-webpki 0.103.10", "rustls-webpki",
"security-framework", "security-framework",
"security-framework-sys", "security-framework-sys",
"webpki-root-certs", "webpki-root-certs",
@@ -8702,17 +8756,6 @@ version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" 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]] [[package]]
name = "rustls-webpki" name = "rustls-webpki"
version = "0.103.10" version = "0.103.10"
@@ -10234,6 +10277,24 @@ version = "0.2.5"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" 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]] [[package]]
name = "twox-hash" name = "twox-hash"
version = "2.1.2" version = "2.1.2"
Cargo.toml
+1 -1
View File
@@ -248,7 +248,7 @@ rayon = "1.11.0"
reed-solomon-erasure = { version = "6.0", default-features = false, features = ["std", "simd-accel"] } reed-solomon-erasure = { version = "6.0", default-features = false, features = ["std", "simd-accel"] }
reed-solomon-simd = "3.1.0" reed-solomon-simd = "3.1.0"
regex = { version = "1.12.3" } 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"] } rustix = { version = "1.1.4", features = ["fs"] }
rust-embed = { version = "8.11.0" } rust-embed = { version = "8.11.0" }
rustc-hash = { version = "2.1.2" } rustc-hash = { version = "2.1.2" }
crates/audit/src/factory.rs
+23 -8
View File
@@ -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::{AUDIT_MQTT_KEYS, AUDIT_WEBHOOK_KEYS, ENV_AUDIT_MQTT_KEYS, ENV_AUDIT_WEBHOOK_KEYS};
use rustfs_config::{ use rustfs_config::{
AUDIT_DEFAULT_DIR, DEFAULT_LIMIT, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR, 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_AUTH_TOKEN, WEBHOOK_CLIENT_CA, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR,
WEBHOOK_QUEUE_LIMIT, WEBHOOK_SKIP_TLS_VERIFY, WEBHOOK_QUEUE_LIMIT, WEBHOOK_SKIP_TLS_VERIFY,
}; };
@@ -27,7 +28,10 @@ use rustfs_ecstore::config::KVS;
use rustfs_targets::{ use rustfs_targets::{
Target, Target,
error::TargetError, error::TargetError,
target::{mqtt::MQTTArgs, webhook::WebhookArgs}, target::{
mqtt::{MQTTArgs, MQTTTlsConfig, validate_mqtt_broker_url},
webhook::WebhookArgs,
},
}; };
use std::time::Duration; use std::time::Duration;
use tracing::{debug, warn}; use tracing::{debug, warn};
@@ -166,6 +170,14 @@ impl TargetFactory for MQTTTargetFactory {
.and_then(|v| v.parse::<u64>().ok()) .and_then(|v| v.parse::<u64>().ok())
.map(Duration::from_secs) .map(Duration::from_secs)
.unwrap_or_else(|| Duration::from_secs(30)), .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_dir: config.lookup(MQTT_QUEUE_DIR).unwrap_or(AUDIT_DEFAULT_DIR.to_string()),
queue_limit: config queue_limit: config
.lookup(MQTT_QUEUE_LIMIT) .lookup(MQTT_QUEUE_LIMIT)
@@ -185,12 +197,15 @@ impl TargetFactory for MQTTTargetFactory {
let url = Url::parse(&broker) let url = Url::parse(&broker)
.map_err(|e| TargetError::Configuration(format!("Invalid broker URL: {e} (value: '{broker}')")))?; .map_err(|e| TargetError::Configuration(format!("Invalid broker URL: {e} (value: '{broker}')")))?;
match url.scheme() { let tls = MQTTTlsConfig::from_values(
"tcp" | "ssl" | "ws" | "wss" | "mqtt" | "mqtts" => {} config.lookup(MQTT_TLS_POLICY).as_deref(),
_ => { config.lookup(MQTT_TLS_CA).as_deref(),
return Err(TargetError::Configuration("Unsupported broker URL scheme".to_string())); 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() { if config.lookup(MQTT_TOPIC).is_none() {
return Err(TargetError::Configuration("Missing MQTT topic".to_string())); return Err(TargetError::Configuration("Missing MQTT topic".to_string()));
crates/audit/tests/config_parsing_test.rs
+6
View File
@@ -53,6 +53,12 @@ fn test_mqtt_valid_fields() {
"reconnect_interval", "reconnect_interval",
"queue_dir", "queue_dir",
"queue_limit", "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 // This tests the MQTT configuration fields we support
crates/config/src/audit/mqtt.rs
+19 -1
View File
@@ -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_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_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_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. /// 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_ENABLE,
ENV_AUDIT_MQTT_BROKER, ENV_AUDIT_MQTT_BROKER,
ENV_AUDIT_MQTT_TOPIC, 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_KEEP_ALIVE_INTERVAL,
ENV_AUDIT_MQTT_QUEUE_DIR, ENV_AUDIT_MQTT_QUEUE_DIR,
ENV_AUDIT_MQTT_QUEUE_LIMIT, 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. /// 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_KEEP_ALIVE_INTERVAL,
crate::MQTT_QUEUE_DIR, crate::MQTT_QUEUE_DIR,
crate::MQTT_QUEUE_LIMIT, 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, crate::COMMENT_KEY,
]; ];
crates/config/src/constants/targets.rs
+6
View File
@@ -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_KEEP_ALIVE_INTERVAL: &str = "keep_alive_interval";
pub const MQTT_QUEUE_DIR: &str = "queue_dir"; pub const MQTT_QUEUE_DIR: &str = "queue_dir";
pub const MQTT_QUEUE_LIMIT: &str = "queue_limit"; 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. /// Environment variable controlling whether target queue files are Snappy-compressed.
/// Applies to both notify and audit target queue stores. /// Applies to both notify and audit target queue stores.
crates/config/src/notify/mqtt.rs
+19 -1
View File
@@ -24,6 +24,12 @@ pub const NOTIFY_MQTT_KEYS: &[&str] = &[
crate::MQTT_KEEP_ALIVE_INTERVAL, crate::MQTT_KEEP_ALIVE_INTERVAL,
crate::MQTT_QUEUE_DIR, crate::MQTT_QUEUE_DIR,
crate::MQTT_QUEUE_LIMIT, 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, 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_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_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_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_ENABLE,
ENV_NOTIFY_MQTT_BROKER, ENV_NOTIFY_MQTT_BROKER,
ENV_NOTIFY_MQTT_TOPIC, 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_KEEP_ALIVE_INTERVAL,
ENV_NOTIFY_MQTT_QUEUE_DIR, ENV_NOTIFY_MQTT_QUEUE_DIR,
ENV_NOTIFY_MQTT_QUEUE_LIMIT, 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,
]; ];
crates/ecstore/src/config/audit.rs
+32 -1
View File
@@ -15,7 +15,8 @@
use crate::config::{KV, KVS}; use crate::config::{KV, KVS};
use rustfs_config::{ use rustfs_config::{
COMMENT_KEY, DEFAULT_LIMIT, ENABLE_KEY, EVENT_DEFAULT_DIR, EnableState, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, 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_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, 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<KVS> = LazyLock::new(|| {
value: DEFAULT_LIMIT.to_string(), value: DEFAULT_LIMIT.to_string(),
hidden_if_empty: false, 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 { KV {
key: COMMENT_KEY.to_owned(), key: COMMENT_KEY.to_owned(),
value: "".to_owned(), value: "".to_owned(),
crates/ecstore/src/config/notify.rs
+32 -1
View File
@@ -15,7 +15,8 @@
use crate::config::{KV, KVS}; use crate::config::{KV, KVS};
use rustfs_config::{ use rustfs_config::{
COMMENT_KEY, DEFAULT_LIMIT, ENABLE_KEY, EVENT_DEFAULT_DIR, EnableState, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, 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_CLIENT_CA, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT,
WEBHOOK_SKIP_TLS_VERIFY, WEBHOOK_SKIP_TLS_VERIFY,
}; };
@@ -133,6 +134,36 @@ pub static DEFAULT_NOTIFY_MQTT_KVS: LazyLock<KVS> = LazyLock::new(|| {
value: DEFAULT_LIMIT.to_string(), value: DEFAULT_LIMIT.to_string(),
hidden_if_empty: false, 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 { KV {
key: COMMENT_KEY.to_owned(), key: COMMENT_KEY.to_owned(),
value: "".to_owned(), value: "".to_owned(),
crates/notify/src/factory.rs
+23 -8
View File
@@ -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::notify::{ENV_NOTIFY_MQTT_KEYS, ENV_NOTIFY_WEBHOOK_KEYS, NOTIFY_MQTT_KEYS, NOTIFY_WEBHOOK_KEYS};
use rustfs_config::{ use rustfs_config::{
DEFAULT_LIMIT, EVENT_DEFAULT_DIR, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR, 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_AUTH_TOKEN, WEBHOOK_CLIENT_CA, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR,
WEBHOOK_QUEUE_LIMIT, WEBHOOK_SKIP_TLS_VERIFY, WEBHOOK_QUEUE_LIMIT, WEBHOOK_SKIP_TLS_VERIFY,
}; };
@@ -27,7 +28,10 @@ use rustfs_ecstore::config::KVS;
use rustfs_targets::{ use rustfs_targets::{
Target, Target,
error::TargetError, error::TargetError,
target::{mqtt::MQTTArgs, webhook::WebhookArgs}, target::{
mqtt::{MQTTArgs, MQTTTlsConfig, validate_mqtt_broker_url},
webhook::WebhookArgs,
},
}; };
use std::time::Duration; use std::time::Duration;
use tracing::{debug, warn}; use tracing::{debug, warn};
@@ -166,6 +170,14 @@ impl TargetFactory for MQTTTargetFactory {
.and_then(|v| v.parse::<u64>().ok()) .and_then(|v| v.parse::<u64>().ok())
.map(Duration::from_secs) .map(Duration::from_secs)
.unwrap_or_else(|| Duration::from_secs(30)), .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_dir: config.lookup(MQTT_QUEUE_DIR).unwrap_or(EVENT_DEFAULT_DIR.to_string()),
queue_limit: config queue_limit: config
.lookup(MQTT_QUEUE_LIMIT) .lookup(MQTT_QUEUE_LIMIT)
@@ -185,12 +197,15 @@ impl TargetFactory for MQTTTargetFactory {
let url = Url::parse(&broker) let url = Url::parse(&broker)
.map_err(|e| TargetError::Configuration(format!("Invalid broker URL: {e} (value: '{broker}')")))?; .map_err(|e| TargetError::Configuration(format!("Invalid broker URL: {e} (value: '{broker}')")))?;
match url.scheme() { let tls = MQTTTlsConfig::from_values(
"tcp" | "ssl" | "ws" | "wss" | "mqtt" | "mqtts" => {} config.lookup(MQTT_TLS_POLICY).as_deref(),
_ => { config.lookup(MQTT_TLS_CA).as_deref(),
return Err(TargetError::Configuration("Unsupported broker URL scheme".to_string())); 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() { if config.lookup(MQTT_TOPIC).is_none() {
return Err(TargetError::Configuration("Missing MQTT topic".to_string())); return Err(TargetError::Configuration("Missing MQTT topic".to_string()));
crates/targets/Cargo.toml
+3 -1
View File
@@ -13,11 +13,13 @@ documentation = "https://docs.rs/rustfs-target/latest/rustfs_target/"
[dependencies] [dependencies]
rustfs-config = { workspace = true, features = ["notify", "constants", "audit"] } 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 } rustfs-s3-common = { workspace = true }
async-trait = { workspace = true } async-trait = { workspace = true }
hyper-rustls = { workspace = true }
reqwest = { workspace = true } reqwest = { workspace = true }
rumqttc = { workspace = true } rumqttc = { workspace = true }
rustls = { workspace = true }
serde = { workspace = true } serde = { workspace = true }
serde_json = { workspace = true } serde_json = { workspace = true }
snap = { workspace = true } snap = { workspace = true }
crates/targets/src/check.rs
+34 -33
View File
@@ -21,7 +21,9 @@
/// * `password` - Optional password for authentication /// * `password` - Optional password for authentication
/// # Returns /// # Returns
/// * `Ok(())` - If the connection is successful /// * `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 /// # Example
/// ```rust,no_run /// ```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( pub async fn check_mqtt_broker_available(
broker_url: &str, broker_url: &str,
topic: &str, topic: &str,
username: Option<&str>, username: Option<&str>,
password: Option<&str>, password: Option<&str>,
) -> Result<(), String> { ) -> Result<(), crate::TargetError> {
use rumqttc::{AsyncClient, MqttOptions, QoS}; use crate::target::mqtt::MQTTTlsConfig;
let url = rustfs_utils::parse_url(broker_url).map_err(|e| format!("Broker URL parsing failed:{e}"))?;
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(); let url = url.url();
match url.scheme() { // build_mqtt_options returns TargetError directly; Configuration variants propagate as-is.
"tcp" | "ssl" | "ws" | "wss" | "mqtt" | "mqtts" | "tls" | "tcps" => {} let mqtt_options = build_mqtt_options(
_ => return Err("unsupported broker url scheme".to_string()), "rustfs_check".to_string(),
} url,
username,
let host = url.host_str().ok_or("Broker is missing host")?; password,
let port = url.port().unwrap_or(1883); tls,
let mut mqtt_options = MqttOptions::new("rustfs_check", host, port); std::time::Duration::from_secs(5),
None,
// 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));
let (client, mut eventloop) = AsyncClient::new(mqtt_options, 1); let (client, mut eventloop) = AsyncClient::new(mqtt_options, 1);
// Try to connect and subscribe // Try to connect and subscribe
client client
.subscribe(topic, QoS::AtLeastOnce) .subscribe(topic, QoS::AtLeastOnce)
.await .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 // Wait for eventloop to receive at least one event
match tokio::time::timeout(std::time::Duration::from_secs(3), eventloop.poll()).await { match tokio::time::timeout(std::time::Duration::from_secs(3), eventloop.poll()).await {
Ok(Ok(_)) => Ok(()), Ok(Ok(_)) => Ok(()),
Ok(Err(e)) => Err(format!("MQTT connection failed:{e}")), Ok(Err(e)) => Err(crate::TargetError::Network(format!("MQTT connection failed: {e}"))),
Err(_) => Err("MQTT connection timeout".to_string()), Err(_) => Err(crate::TargetError::Timeout("MQTT connection timed out".to_string())),
} }
} }
crates/targets/src/lib.rs
+1 -1
View File
@@ -18,7 +18,7 @@ pub mod error;
pub mod store; pub mod store;
pub mod target; 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 error::{StoreError, TargetError};
pub use rustfs_s3_common::EventName; pub use rustfs_s3_common::EventName;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
crates/targets/src/target/mqtt.rs
+482 -39
View File
@@ -20,13 +20,21 @@ use crate::{
target::{ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetType}, target::{ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetType},
}; };
use async_trait::async_trait; 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::Serialize;
use serde::de::DeserializeOwned; use serde::de::DeserializeOwned;
use std::sync::Arc; use std::sync::Arc;
use std::{ use std::{
marker::PhantomData, marker::PhantomData,
path::PathBuf, path::{Path, PathBuf},
sync::atomic::{AtomicBool, Ordering}, sync::atomic::{AtomicBool, Ordering},
time::Duration, time::Duration,
}; };
@@ -36,6 +44,381 @@ use url::Url;
const DEFAULT_CONNECTION_TIMEOUT: Duration = Duration::from_secs(15); 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 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<Self, TargetError> {
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<MQTTTlsPolicy>,
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<String>,
}
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<Self, TargetError> {
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::<EnableState>()
.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<Vec<String>, 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<String, TargetError> {
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<rustls::RootCertStore, TargetError> {
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<Transport, TargetError> {
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<u32>,
) -> Result<MqttOptions, TargetError> {
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 /// Arguments for configuring an MQTT target
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
@@ -52,6 +435,8 @@ pub struct MQTTArgs {
pub username: String, pub username: String,
/// The password for the broker /// The password for the broker
pub password: String, pub password: String,
/// Explicit TLS configuration for secure MQTT transports
pub tls: MQTTTlsConfig,
/// The maximum interval for reconnection attempts (Note: rumqttc has internal strategy) /// The maximum interval for reconnection attempts (Note: rumqttc has internal strategy)
pub max_reconnect_interval: Duration, pub max_reconnect_interval: Duration,
/// The keep alive interval /// The keep alive interval
@@ -70,12 +455,7 @@ impl MQTTArgs {
return Ok(()); return Ok(());
} }
match self.broker.scheme() { validate_mqtt_broker_url(&self.broker, &self.tls)?;
"ws" | "wss" | "tcp" | "ssl" | "tls" | "tcps" | "mqtt" | "mqtts" => {}
_ => {
return Err(TargetError::Configuration("unknown protocol in broker address".to_string()));
}
}
if self.topic.is_empty() { if self.topic.is_empty() {
return Err(TargetError::Configuration("MQTT topic cannot be empty".to_string())); return Err(TargetError::Configuration("MQTT topic cannot be empty".to_string()));
@@ -187,16 +567,15 @@ where
.init_cell .init_cell
.get_or_try_init(|| async { .get_or_try_init(|| async {
debug!(target_id = %target_id_clone, "Initializing MQTT background task."); debug!(target_id = %target_id_clone, "Initializing MQTT background task.");
let host = args_clone.broker.host_str().unwrap_or("localhost"); let mqtt_options = build_mqtt_options(
let port = args_clone.broker.port().unwrap_or(1883); format!("rustfs_notify_{}", uuid::Uuid::new_v4()),
let mut mqtt_options = MqttOptions::new(format!("rustfs_notify_{}", uuid::Uuid::new_v4()), host, port); &args_clone.broker,
mqtt_options Some(args_clone.username.as_str()),
.set_keep_alive(args_clone.keep_alive) Some(args_clone.password.as_str()),
.set_max_packet_size(100 * 1024 * 1024, 100 * 1024 * 1024); // 100MB &args_clone.tls,
args_clone.keep_alive,
if !args_clone.username.is_empty() { Some(MAX_MQTT_PACKET_SIZE_BYTES),
mqtt_options.set_credentials(args_clone.username.clone(), args_clone.password.clone()); )?;
}
let (new_client, eventloop) = AsyncClient::new(mqtt_options, 10); let (new_client, eventloop) = AsyncClient::new(mqtt_options, 10);
@@ -341,40 +720,40 @@ async fn run_mqtt_event_loop(
polled_event_result = async { polled_event_result = async {
if !initial_connection_established || !connected_status.load(Ordering::SeqCst) { if !initial_connection_established || !connected_status.load(Ordering::SeqCst) {
match tokio::time::timeout(EVENT_LOOP_POLL_TIMEOUT, eventloop.poll()).await { match tokio::time::timeout(EVENT_LOOP_POLL_TIMEOUT, eventloop.poll()).await {
Ok(Ok(event)) => Ok(event), Ok(result) => Some(result),
Ok(Err(e)) => Err(e),
Err(_) => { Err(_) => {
debug!(target_id = %target_id, "MQTT poll timed out (EVENT_LOOP_POLL_TIMEOUT) while not connected or status pending."); 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 { } else {
eventloop.poll().await Some(eventloop.poll().await)
} }
} => { } => {
match polled_event_result { match polled_event_result {
Ok(notification) => { Some(Ok(notification)) => {
trace!(target_id = %target_id, event = ?notification, "Received MQTT event"); trace!(target_id = %target_id, event = ?notification, "Received MQTT event");
match notification { match notification {
rumqttc::Event::Incoming(Packet::ConnAck(_conn_ack)) => { rumqttc::Event::Incoming(Incoming::ConnAck(_conn_ack)) => {
info!(target_id = %target_id, "MQTT connected (ConnAck)."); info!(target_id = %target_id, "MQTT connected (ConnAck).");
connected_status.store(true, Ordering::SeqCst); connected_status.store(true, Ordering::SeqCst);
initial_connection_established = true; initial_connection_established = true;
} }
rumqttc::Event::Incoming(Packet::Publish(publish)) => { rumqttc::Event::Incoming(Incoming::Publish(publish)) => {
debug!(target_id = %target_id, topic = %publish.topic, payload_len = publish.payload.len(), "Received message on subscribed topic."); 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."); info!(target_id = %target_id, "Received Disconnect packet from broker. MQTT connection lost.");
connected_status.store(false, Ordering::SeqCst); 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."); 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); 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); trace!(target_id = %target_id, "Received PubAck for pkid: {}", puback.pkid);
} }
// Process other incoming packet types as needed (PubRec, PubRel, PubComp, UnsubAck) // 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); connected_status.store(false, Ordering::SeqCst);
error!(target_id = %target_id, error = %e, "Error from MQTT event loop poll"); 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, if matches!(e,
ConnectionError::Io(_) | ConnectionError::Io(_) |
ConnectionError::NetworkTimeout | ConnectionError::Timeout(_) |
ConnectionError::ConnectionRefused(_) | ConnectionError::ConnectionRefused(_) |
ConnectionError::Tls(_) ConnectionError::Tls(_)
) { ) {
@@ -422,6 +796,10 @@ async fn run_mqtt_event_loop(
// Sleep briefly to avoid busy cycles in case of rapid failure. // Sleep briefly to avoid busy cycles in case of rapid failure.
tokio::time::sleep(Duration::from_secs(1)).await; 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 { match state_err {
// If StateError is caused by deserialization issues, check the underlying MqttBytesError // 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 rumqttc::StateError::Deserialization(mqtt_bytes_err) => { // The type of mqtt_bytes_err is &rumqttc::mqttbytes::Error
matches!( matches!(
mqtt_bytes_err, mqtt_bytes_err,
MqttBytesError::InvalidProtocol // Invalid agreement MqttBytesError::InvalidProtocol // Invalid agreement
| MqttBytesError::InvalidProtocolLevel(_) // Invalid protocol level | MqttBytesError::InvalidProtocolLevel(_) // Invalid protocol level
@@ -451,7 +829,7 @@ fn is_fatal_mqtt_error(err: &ConnectionError) -> bool {
| MqttBytesError::InvalidPacketType(_) // Invalid package type | MqttBytesError::InvalidPacketType(_) // Invalid package type
| MqttBytesError::MalformedPacket // Package format error | MqttBytesError::MalformedPacket // Package format error
| MqttBytesError::PayloadTooLong // Too long load | 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) | MqttBytesError::TopicNotUtf8 // Topic Non-UTF-8 (Serious Agreement Violation)
) )
} }
@@ -640,3 +1018,68 @@ where
self.args.enable 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"));
}
}
rustfs/src/admin/handlers/audit.rs
+18 -3
View File
@@ -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::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_config::{DEFAULT_DELIMITER, ENABLE_KEY, ENV_PREFIX, EnableState, MAX_ADMIN_REQUEST_BODY_SIZE};
use rustfs_ecstore::config::Config; 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 s3s::{Body, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet}; use std::collections::{HashMap, HashSet};
@@ -521,9 +521,24 @@ impl Operation for AuditTargetConfig {
.ok_or_else(|| s3_error!(InvalidArgument, "topic is required"))?; .ok_or_else(|| s3_error!(InvalidArgument, "topic is required"))?;
let username = kv_map.get(rustfs_config::MQTT_USERNAME).map(String::as_str); 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); 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 .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") { if let Some(queue_dir) = kv_map.get("queue_dir") {
validate_queue_dir(queue_dir.as_str()).await?; validate_queue_dir(queue_dir.as_str()).await?;
rustfs/src/admin/handlers/event.rs
+18 -3
View File
@@ -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_config::{DEFAULT_DELIMITER, ENABLE_KEY, ENV_PREFIX, EnableState, MAX_ADMIN_REQUEST_BODY_SIZE};
use rustfs_ecstore::config::Config; 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 s3s::{Body, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet}; use std::collections::{HashMap, HashSet};
@@ -450,9 +450,24 @@ impl Operation for NotificationTarget {
.ok_or_else(|| s3_error!(InvalidArgument, "topic is required"))?; .ok_or_else(|| s3_error!(InvalidArgument, "topic is required"))?;
let username = kv_map.get(rustfs_config::MQTT_USERNAME).map(String::as_str); 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); 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 .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") { if let Some(queue_dir) = kv_map.get("queue_dir") {
validate_queue_dir(queue_dir.as_str()).await?; validate_queue_dir(queue_dir.as_str()).await?;