From b18c5afc8d0bc241a7db53bf51af9dd368f4bc13 Mon Sep 17 00:00:00 2001 From: Hauser Date: Thu, 8 Oct 2026 07:57:09 +0800 Subject: [PATCH] feat(targets): support Kafka SASL without TLS (#8398) * feat(targets): support Kafka SASL without TLS Propagate the explicit Kafka transport mode through notify and audit targets while retaining TLS as the default. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * chore(deps): update RustFS Kafka crates to 1.4.0 Use the published Kafka clients after the SASL plaintext transport support release. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: heihutu Co-authored-by: zhi22915 --- Cargo.lock | 8 +++---- Cargo.toml | 2 +- crates/targets/src/check.rs | 21 +++++++++-------- crates/targets/src/config/target_args.rs | 12 ++++++---- crates/targets/src/target/kafka.rs | 29 ++++++++++++++---------- 5 files changed, 42 insertions(+), 30 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e3eda47a6..249730c39 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10078,9 +10078,9 @@ dependencies = [ [[package]] name = "rustfs-kafka" -version = "1.3.1" +version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "759f8ddf709b497006e0f422a89890f11ac86c84c2c39df282891477c22f1a13" +checksum = "bd2f8b46ec4dd462c2235948d1efa2103957230c171e118c6ff8e5d1a845ab48" dependencies = [ "base64 0.23.1", "bytes", @@ -10103,9 +10103,9 @@ dependencies = [ [[package]] name = "rustfs-kafka-async" -version = "1.3.1" +version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5aa574a13d91ef529e3a08f059a5c3ee8d4c3b16cf62f62c72595a6dc4f79f8f" +checksum = "4b3b98705f6f9f5071b8793d29935de764c31668c6e9877553b229eb67127af6" dependencies = [ "base64 0.23.1", "bytes", diff --git a/Cargo.toml b/Cargo.toml index dc105b562..6a5791efb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -166,7 +166,7 @@ http-body-util = "0.1.5" memchr = "2.8.3" minlz = "1.2.3" reqwest = "0.13.5" -rustfs-kafka-async = { version = "1.3.1" } +rustfs-kafka-async = { version = "1.4.0" } socket2 = { version = "0.6.5" } tokio = { version = "1.53.2" } tokio-rustls = { default-features = false, version = "0.26.6" } diff --git a/crates/targets/src/check.rs b/crates/targets/src/check.rs index c56c14895..af882c3bb 100644 --- a/crates/targets/src/check.rs +++ b/crates/targets/src/check.rs @@ -355,21 +355,24 @@ mod tests { } #[test] - fn check_kafka_broker_available_rejects_sasl_without_tls_before_connecting() { + fn check_kafka_broker_available_accepts_sasl_without_tls_configuration() { let mut args = kafka_args(); args.sasl_enable = true; args.sasl_username = "user".to_string(); args.sasl_password = "secret".to_string(); - let err = tokio::runtime::Runtime::new() - .expect("runtime") - .block_on(check_kafka_broker_available(&args)) - .expect_err("SASL without TLS should fail before opening a network connection"); + assert!(args.validate().is_ok()); + let security = args + .security_config(false) + .expect("valid SASL configuration") + .expect("SASL should configure security"); + assert!(!security.tls_enabled()); - match err { - TargetError::Configuration(msg) => assert!(msg.contains("requires tls_enable")), - other => panic!("expected configuration error, got {other:?}"), - } + let result = tokio::runtime::Runtime::new() + .expect("runtime") + .block_on(check_kafka_broker_available(&args)); + + assert!(!matches!(result, Err(TargetError::Configuration(_)))); } #[test] diff --git a/crates/targets/src/config/target_args.rs b/crates/targets/src/config/target_args.rs index d96c7b719..d8432c7af 100644 --- a/crates/targets/src/config/target_args.rs +++ b/crates/targets/src/config/target_args.rs @@ -888,15 +888,19 @@ mod tests { } #[test] - fn validate_kafka_config_rejects_sasl_without_tls() { + fn validate_kafka_config_accepts_sasl_without_tls() { let mut config = kafka_base_config(); config.insert(KAFKA_SASL_ENABLE.to_string(), "on".to_string()); config.insert(KAFKA_SASL_USERNAME.to_string(), "user".to_string()); config.insert(KAFKA_SASL_PASSWORD.to_string(), "secret".to_string()); - let err = validate_kafka_config(&config, "").expect_err("SASL without TLS should fail"); - - assert!(err.to_string().contains("requires tls_enable")); + for target_type in [TargetType::NotifyEvent, TargetType::AuditLog] { + let args = build_kafka_args(&config, "", target_type).expect("SASL plaintext config should parse"); + assert_eq!(args.target_type, target_type); + assert!(!args.tls_enable); + assert!(args.sasl_enable); + assert!(args.validate().is_ok()); + } } #[test] diff --git a/crates/targets/src/target/kafka.rs b/crates/targets/src/target/kafka.rs index 7888fd13b..7c59123c7 100644 --- a/crates/targets/src/target/kafka.rs +++ b/crates/targets/src/target/kafka.rs @@ -130,7 +130,8 @@ pub struct KafkaArgs { pub tls_client_cert: String, /// Optional path to client private key for mTLS pub tls_client_key: String, - /// Whether to enable SASL authentication over the TLS transport + /// Whether to enable SASL authentication. With TLS disabled, SASL and event + /// traffic use plaintext transport and are visible to network observers. pub sasl_enable: bool, /// SASL mechanism (PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512) pub sasl_mechanism: String, @@ -224,11 +225,6 @@ impl KafkaArgs { } if self.sasl_enable { - if !self.tls_enable { - return Err(TargetError::Configuration( - "kafka sasl_enable requires tls_enable for SASL_SSL".to_string(), - )); - } normalize_kafka_sasl_mechanism(&self.sasl_mechanism)?; if self.sasl_username.is_empty() || self.sasl_password.is_empty() { return Err(TargetError::Configuration( @@ -256,8 +252,8 @@ impl KafkaArgs { return Ok(None); } - let mut security = SecurityConfig::new(); - if !self.tls_ca.is_empty() { + let mut security = SecurityConfig::new().with_tls_enabled(self.tls_enable); + if self.tls_enable && !self.tls_ca.is_empty() { if validate_tls_files { let certs = load_cert_bundle_der_bytes(&self.tls_ca) .map_err(|e| TargetError::Configuration(format!("Failed to parse Kafka tls_ca: {e}")))?; @@ -269,7 +265,7 @@ impl KafkaArgs { } security = security.with_ca_cert(self.tls_ca.clone()); } - if !self.tls_client_cert.is_empty() && !self.tls_client_key.is_empty() { + if self.tls_enable && !self.tls_client_cert.is_empty() && !self.tls_client_key.is_empty() { if validate_tls_files { let certs = load_cert_bundle_der_bytes(&self.tls_client_cert) .map_err(|e| TargetError::Configuration(format!("Failed to parse Kafka tls_client_cert: {e}")))?; @@ -871,7 +867,7 @@ mod tests { } #[test] - fn test_validate_sasl_requires_tls() { + fn test_validate_sasl_allows_plaintext_transport() { let args = KafkaArgs { sasl_enable: true, sasl_mechanism: KAFKA_SASL_SCRAM_SHA_512.to_string(), @@ -879,8 +875,16 @@ mod tests { sasl_password: "secret".to_string(), ..base_args() }; - let err = args.validate().expect_err("SASL without TLS should fail"); - assert!(err.to_string().contains("requires tls_enable")); + assert!(args.validate().is_ok()); + let security = args + .security_config(false) + .expect("valid SASL configuration") + .expect("SASL should configure security"); + assert!(!security.tls_enabled()); + assert_eq!( + security.sasl_config().expect("SASL should be configured").mechanism(), + KAFKA_SASL_SCRAM_SHA_512 + ); } #[test] @@ -928,6 +932,7 @@ mod tests { .expect("security should be configured"); let sasl = security.sasl_config().expect("SASL should be configured"); + assert!(security.tls_enabled()); assert_eq!(sasl.mechanism(), KAFKA_SASL_SCRAM_SHA_512); assert_eq!(sasl.username(), "user"); assert_eq!(sasl.password(), "secret");