gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-12 15:48:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(targets): support Kafka SASL without TLS (#8398)

Browse Source 
* 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 <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* 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 <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
Hauser
2026-10-08 07:57:09 +08:00
committed by GitHub
parent df988cfd86
commit b18c5afc8d
5 changed files with 42 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+4 -4
View File
@@ -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",
Cargo.toml
+1 -1
View File
@@ -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" }
crates/targets/src/check.rs
+12 -9
View File
@@ -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]
crates/targets/src/config/target_args.rs
+8 -4
View File
@@ -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]
crates/targets/src/target/kafka.rs
+17 -12
View File
@@ -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");