From 7040fc087532b1dffa2b610f96d6bc0e5faa129a Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 13 Nov 2026 16:02:43 +0100 Subject: [PATCH 1/2] fix: Kraft controllers should always require client auth --- CHANGELOG.md | 4 ++ .../build/properties/controller_properties.rs | 5 +- .../src/controller/build/security.rs | 63 ++++++++++--------- 3 files changed, 40 insertions(+), 32 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b17eabd9..4beb7e70 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -86,6 +86,10 @@ All notable changes to this project will be documented in this file. `CONTAINERDEBUG_LOG_DIRECTORY`, which containerdebug stopped reading in its version 0.2.0 when it moved to `stackable-telemetry`; the directory is now set via `FILE_LOG_DIRECTORY`. The file is rotated hourly and at most two files are kept, because rotation is opt-in ([#1055]). +- The CONTROLLER listener of KRaft controllers now always requires a client certificate, like the + INTERNAL listener of brokers. Previously, it accepted TLS connections without a client certificate + unless a TLS AuthenticationClass or OPA was configured. With Kerberos, clients are still + authenticated via SASL instead ([#XXXX]). ### Removed diff --git a/rust/operator-binary/src/controller/build/properties/controller_properties.rs b/rust/operator-binary/src/controller/build/properties/controller_properties.rs index c5de74c6..9c754d88 100644 --- a/rust/operator-binary/src/controller/build/properties/controller_properties.rs +++ b/rust/operator-binary/src/controller/build/properties/controller_properties.rs @@ -74,10 +74,7 @@ pub fn build( } result.extend(opa_authorizer_properties(cluster_config, pod_descriptors)); - result.extend(controller_config_settings( - &cluster_config.kafka_security, - cluster_config.opa_connect().is_some(), - )); + result.extend(controller_config_settings(&cluster_config.kafka_security)); result.extend(graceful_shutdown_config_properties()); result.extend(overrides); diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs index 141f5a23..6f679a19 100644 --- a/rust/operator-binary/src/controller/build/security.rs +++ b/rust/operator-binary/src/controller/build/security.rs @@ -563,10 +563,7 @@ pub fn broker_config_settings(security: &ValidatedKafkaSecurity) -> BTreeMap BTreeMap { +pub fn controller_config_settings(security: &ValidatedKafkaSecurity) -> BTreeMap { let mut config = BTreeMap::new(); insert_listener_ssl_stores( @@ -582,17 +579,15 @@ pub fn controller_config_settings( &KafkaListenerName::Internal, STACKABLE_TLS_KAFKA_INTERNAL_DIR, ); - // We set either client tls with authentication or client tls without authentication - // If authentication is explicitly required we do not want to have any other CAs to - // be trusted. + // The CONTROLLER listener only serves node-to-node traffic (brokers, other controllers and + // the quorum-manager tooling), all of which present the internal certificate. Like the + // INTERNAL listener, it therefore always requires client authentication, so that nothing + // without an internal certificate can talk to the KRaft quorum. + // This also gives the nodes a non-ANONYMOUS principal (the subject of their internal + // certificate), which OPA uses to identify them as super users. // - // With OPA, the nodes are identified as super users by the subject of their internal - // certificate. Without client authentication, the principal on the CONTROLLER listener is - // ANONYMOUS. With Kerberos, the principal is derived from the SASL authentication instead. - if security.tls_client_authentication_class().is_some() - || (opa_authorization && !security.has_kerberos_enabled()) - { - // client auth required + // With Kerberos, the listener is SASL_SSL and clients are authenticated via SASL instead. + if !security.has_kerberos_enabled() { config.insert( KafkaListenerName::Controller.listener_ssl_client_auth(), SSL_CLIENT_AUTH_REQUIRED.to_string(), @@ -1072,7 +1067,7 @@ mod tests { #[test] fn controller_config_opa_adds_authorizer_truststore() { - let config = controller_config_settings(&opa(), false); + let config = controller_config_settings(&opa()); assert!(config.contains_key("opa.authorizer.truststore.path")); assert!(config.contains_key("opa.authorizer.truststore.password")); assert!(config.contains_key("opa.authorizer.truststore.type")); @@ -1081,30 +1076,42 @@ mod tests { #[test] fn controller_config_plaintext_has_internal_tls() { // Internal TLS is mandatory, so the controller config is non-empty even for a plaintext - // cluster: it carries the controller + internal SSL stores, but no client-cert auth - // (there is no client authentication class). - let config = controller_config_settings(&plaintext(), false); + // cluster: it carries the controller + internal SSL stores and requires client certs on + // the CONTROLLER listener. + let config = controller_config_settings(&plaintext()); assert!( config.contains_key(&KafkaListenerName::Controller.listener_ssl_keystore_location()) ); assert!(config.contains_key(&KafkaListenerName::Internal.listener_ssl_keystore_location())); - assert!(!config.contains_key(&KafkaListenerName::Controller.listener_ssl_client_auth())); + assert_eq!( + config + .get(&KafkaListenerName::Controller.listener_ssl_client_auth()) + .map(String::as_str), + Some("required") + ); } #[test] - fn controller_config_opa_requires_client_auth_unless_kerberos() { + fn controller_config_requires_client_auth_unless_kerberos() { let key = KafkaListenerName::Controller.listener_ssl_client_auth(); - let config = controller_config_settings(&internal_tls(), true); - assert_eq!(config.get(&key).map(String::as_str), Some("required")); + for security in [ + plaintext(), + server_tls(), + client_auth_tls(), + internal_tls(), + opa(), + ] { + let config = controller_config_settings(&security); + assert_eq!(config.get(&key).map(String::as_str), Some("required")); + } - assert!(!controller_config_settings(&internal_tls(), false).contains_key(&key)); - assert!(!controller_config_settings(&kerberos(), true).contains_key(&key)); + assert!(!controller_config_settings(&kerberos()).contains_key(&key)); } #[test] fn controller_config_internal_tls_covers_controller_and_internal() { - let config = controller_config_settings(&internal_tls(), false); + let config = controller_config_settings(&internal_tls()); assert!(config.contains_key("listener.name.controller.ssl.keystore.location")); assert!(config.contains_key("listener.name.internal.ssl.keystore.location")); } @@ -1120,7 +1127,7 @@ mod tests { #[test] fn controller_config_sets_the_controller_sasl_mechanism_with_kerberos() { - let config = controller_config_settings(&kerberos(), false); + let config = controller_config_settings(&kerberos()); assert_eq!( config.get("sasl.mechanism.controller.protocol"), Some(&"GSSAPI".to_string()) @@ -1134,14 +1141,14 @@ mod tests { .contains_key("sasl.mechanism.controller.protocol") ); assert!( - !controller_config_settings(&internal_tls(), false) + !controller_config_settings(&internal_tls()) .contains_key("sasl.mechanism.controller.protocol") ); } #[test] fn controller_config_kerberos_adds_sasl() { - let config = controller_config_settings(&kerberos(), false); + let config = controller_config_settings(&kerberos()); assert_eq!( config.get("sasl.enabled.mechanisms"), Some(&"GSSAPI".to_string()) From 52d81469fffb4dc6a0de5d36013eecf0900c5186 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 13 Nov 2026 16:06:41 +0100 Subject: [PATCH 2/2] update changelog --- CHANGELOG.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4beb7e70..6fb0cb1b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -89,7 +89,7 @@ All notable changes to this project will be documented in this file. - The CONTROLLER listener of KRaft controllers now always requires a client certificate, like the INTERNAL listener of brokers. Previously, it accepted TLS connections without a client certificate unless a TLS AuthenticationClass or OPA was configured. With Kerberos, clients are still - authenticated via SASL instead ([#XXXX]). + authenticated via SASL instead ([#1065]). ### Removed @@ -128,6 +128,7 @@ All notable changes to this project will be documented in this file. [#1051]: https://github.com/stackabletech/kafka-operator/pull/1051 [#1052]: https://github.com/stackabletech/kafka-operator/pull/1052 [#1055]: https://github.com/stackabletech/kafka-operator/pull/1055 +[#1065]: https://github.com/stackabletech/kafka-operator/pull/1065 ## [26.7.0] - 2026-07-21