From cc7346939cdce49ab359f88e432e47a69a8a58e3 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 13 Nov 2026 17:52:44 +0100 Subject: [PATCH 1/2] refactor: KRaft controllers only use mTLS --- CHANGELOG.md | 12 +- .../pages/usage-guide/kraft-controller.adoc | 27 +- .../kafka/pages/usage-guide/security.adoc | 2 +- .../src/controller/build/command.rs | 57 ++-- .../src/controller/build/kerberos.rs | 78 ++---- .../src/controller/build/mod.rs | 1 - .../build/properties/controller_properties.rs | 5 +- .../controller/build/properties/listener.rs | 15 +- .../src/controller/build/properties/mod.rs | 59 ++-- .../src/controller/build/role_group/broker.rs | 13 +- .../controller/build/role_group/controller.rs | 143 ++-------- .../src/controller/build/role_group/mod.rs | 23 +- .../src/controller/build/security.rs | 262 ++++-------------- rust/operator-binary/src/crd/listener.rs | 3 +- rust/operator-binary/src/crd/mod.rs | 1 - .../kuttl/kraft-kerberos/60-assert.yaml.j2 | 4 +- .../kuttl/kraft-kerberos/70-assert.yaml.j2 | 4 +- .../70-scale-controller-down.yaml.j2 | 6 +- .../templates/kuttl/kraft-kerberos/README.md | 4 + tests/templates/kuttl/opa/20_opa.yaml.j2 | 5 - 20 files changed, 191 insertions(+), 533 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b17eabd9e..4074756f5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,8 +8,9 @@ All notable changes to this project will be documented in this file. - Support floating tags for product images via the new `spec.image.stackableVersionPolicy` field ([#1021]). -- Support Kerberos (GSSAPI) authentication on KRaft controllers, covering both - broker-to-controller and controller-to-controller (Raft) traffic ([#1024]). +- Support Kerberos (GSSAPI) client authentication for KRaft clusters, including dynamic controller + quorum scaling. Brokers and controllers authenticate each other with their internal TLS + certificates (mTLS on the CONTROLLER listener), not with Kerberos ([#1024], [#XXXX]). - The logs of the `containerdebug` process are now collected by the Vector agent and forwarded to the Vector aggregator ([#1055]). - Support for Kafka 4.3.1 ([#1022]). @@ -23,7 +24,8 @@ All notable changes to this project will be documented in this file. them. The operator removes the finalizers of KafkaTopics whose KafkaCluster is deleted ([#1050]). - The OPA authorizer is now also configured on KRaft controllers. Previously, requests that brokers forward to the controllers (e.g. creating or deleting topics) were not authorized at all. - The Kafka nodes are now configured as super users, so that traffic between them does not hit OPA. + The Kafka nodes are now configured as super users, identified by the subjects of their internal + TLS certificates, so that traffic between them does not hit OPA. Controllers now get the same affinity to OPA as brokers ([#1051]). ### Changed @@ -32,6 +34,10 @@ All notable changes to this project will be documented in this file. Abbreviated values such as `26.7` are no longer accepted ([#1021]). - BREAKING: `spec.image.pullPolicy` now defaults to `IfNotPresent` for non-floating tags instead of always defaulting to `Always` ([#1021]). +- BREAKING: The CONTROLLER listener of KRaft controllers now always requires TLS client + authentication, also when no authentication or authorization is configured. Tools connecting + directly to a controller (e.g. `kafka-metadata-quorum.sh`) must present a certificate of the + internal `SecretClass` ([#XXXX]). - The dynamic KRaft quorum created by the operator is now scaled automatically. Previously, manual intervention was needed after every scale operation. This change introduces a new side-car container (`quorum-manager`) to all controller pods diff --git a/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc b/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc index 100498d44..b9afd81fc 100644 --- a/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc +++ b/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc @@ -114,32 +114,19 @@ KRaft mode requires major configuration changes compared to ZooKeeper mode: Every other controller formats with `--no-initial-controllers` and joins purely through the sidecar's `add-controller` call. Brokers always format with `--no-initial-controllers` too; they are never voters. * `controller.quorum.bootstrap.servers` points at each controller role group's own headless Service DNS name, not - individual pod addresses, unless Kerberos is enabled. See below for details related to Kerberos. + individual pod addresses, unless Kerberos is enabled. + With Kerberos, it lists the individual pod FQDNs of all controllers, so whenever the number of controller replicas + changes, **both** the controller **and** broker pods are rolled by the restart controller. == Kerberos -Apache Kafka KRaft controllers support Kerberos (GSSAPI) authentication. +KRaft clusters support Kerberos (GSSAPI) authentication for clients. Reference an `AuthenticationClass` with the Kerberos provider as described in xref:usage-guide/security.adoc[Security]. -When Kerberos is enabled: - -* The `CONTROLLER` listener uses `SASL_SSL` instead of `SSL`, with `GSSAPI` as the mechanism - (`sasl.mechanism.controller.protocol`). This covers both broker-to-controller traffic and - the controller-to-controller (Raft) traffic between quorum members. -* Controller keytabs are *pod-scoped*, unlike broker keytabs, which are scoped to their - listener volumes. Controllers are reachable only under their own StatefulSet pod DNS name, - which is therefore the principal in their keytab. -* Each pod's `jaas.properties` gains a `controller.KafkaServer` login context. Unlike the - broker-side contexts it does not set `isInitiator=false`, because a controller must be able - to initiate GSSAPI connections to its peers, not only accept them. -* The `quorum-manager` sidecar and the `preStop` hook authenticate to the CONTROLLER listener - as the controller's own pod principal. -* `controller.quorum.bootstrap.servers` lists the individual pod FQDNs of all controllers. - This is required by Kerberos: a GSSAPI client derives the service principal from the hostname - it dials, and the CONTROLLER listener can only offer the pod's own principal. - It means that, with Kerberos enabled, whenever the controller replicas is changed, - **both** the controller **and** broker pods are rolled by the restart controller. +Kerberos is only used on the client-facing listeners of the brokers. +The `CONTROLLER` listener always uses `SSL` with mutual authentication: brokers and controllers authenticate each other with the certificates of the `internalSecretClass`, also when Kerberos is enabled. +Controllers therefore have no keytab. == Known Issues diff --git a/docs/modules/kafka/pages/usage-guide/security.adoc b/docs/modules/kafka/pages/usage-guide/security.adoc index b0957bd7d..039a62e96 100644 --- a/docs/modules/kafka/pages/usage-guide/security.adoc +++ b/docs/modules/kafka/pages/usage-guide/security.adoc @@ -244,7 +244,7 @@ The operator sets `super.users` to the principals of all nodes of the cluster: * The nodes authenticate each other with the TLS certificates that the secret operator provides for the `internalSecretClass`. The principal is the subject DN of the certificate, which contains the FQDN of the pod and a generic common name, e.g. `DC=local,DC=cluster,DC=svc,DC=default,DC=simple-kafka-broker-default-headless,DC=simple-kafka-broker-default-0,CN=generated certificate for pod`. -* If Kerberos is enabled, the short names of the Kerberos services of the brokers and the controllers are added as well. +* Kerberos principals are never added, also when Kerberos is enabled: the nodes do not use Kerberos to authenticate each other. Kafka compares the principal of a request with each `super.users` entry for exact equality; wildcards are not supported. The common name `CN=generated certificate for pod` is therefore part of every certificate-based entry. diff --git a/rust/operator-binary/src/controller/build/command.rs b/rust/operator-binary/src/controller/build/command.rs index 63e6d489e..f9db0e00d 100644 --- a/rust/operator-binary/src/controller/build/command.rs +++ b/rust/operator-binary/src/controller/build/command.rs @@ -158,7 +158,6 @@ pub fn controller_kafka_container_command( prepare_signal_handlers mkdir --parents \"${file_log_directory}\" || true containerdebug --output={STACKABLE_LOG_DIR}/containerdebug-state.json --loop & - {set_realm_env} {import_opa_tls_cert} @@ -169,9 +168,6 @@ pub fn controller_kafka_container_command( config-utils template /tmp/{properties_file} - cp {config_dir}/{jaas_file} /tmp/{jaas_file} - config-utils template /tmp/{jaas_file} - cp {admin_client_source} {admin_client_config} config-utils template {admin_client_config} @@ -184,14 +180,11 @@ pub fn controller_kafka_container_command( ", remove_vector_shutdown_file_command = remove_vector_shutdown_file_command(STACKABLE_LOG_DIR), file_log_directory = FILE_LOG_DIRECTORY.as_ref(), - // Mirrors `broker_kafka_container_commands`: empty when Kerberos is disabled. - set_realm_env = export_kerberos_realm_command(kafka_security).unwrap_or_default(), import_opa_tls_cert = copy_opa_tls_cert_command(kafka_security), derive_pod_index = DERIVE_POD_INDEX, export_replica_id = EXPORT_REPLICA_ID, config_dir = STACKABLE_CONFIG_DIR, properties_file = ConfigFileName::ControllerProperties, - jaas_file = ConfigFileName::Jaas, admin_client_source = ADMIN_CLIENT_PROPERTIES_SOURCE_PATH, admin_client_config = ADMIN_CLIENT_PROPERTIES_PATH, quorum_format_flag = controller_quorum_format_flag(&controller_descriptors), @@ -201,9 +194,8 @@ pub fn controller_kafka_container_command( const KAFKA_METADATA_QUORUM_BINARY: &str = "/stackable/kafka/bin/kafka-metadata-quorum.sh"; -/// The rendered admin-client config. The raw ConfigMap file is copied here and passed through -/// `config-utils template` first, because under Kerberos its `sasl.jaas.config` carries -/// `${env:...}` placeholders (see `controller_admin_client_properties`). +/// The rendered admin-client config. Like every other config file, the raw ConfigMap file is +/// copied here and passed through `config-utils template` before use. const ADMIN_CLIENT_PROPERTIES_PATH: &str = "/tmp/admin-client.properties"; const ADMIN_CLIENT_PROPERTIES_SOURCE_PATH: &str = "/stackable/config/admin-client.properties"; @@ -253,7 +245,7 @@ const CONTROLLER_QUORUM_MANAGER_LOOP_SCRIPT: &str = /// The sidecar's main-loop command: while this controller's local Raft state is `observer`, /// admit it into the quorum's voter set once that is safe. -pub fn quorum_manager_container_command(security: &ValidatedKafkaSecurity) -> String { +pub fn quorum_manager_container_command() -> String { format!( r#" set -uo pipefail @@ -262,7 +254,6 @@ pub fn quorum_manager_container_command(security: &ValidatedKafkaSecurity) -> St [ -n "$POD_INDEX" ] || exit 0 {export_replica_id} {extract_bootstrap_servers} - {set_realm_env} if cp {config_dir}/{controller_properties_file} /tmp/{controller_properties_file} \ && config-utils template /tmp/{controller_properties_file} \ @@ -293,7 +284,6 @@ pub fn quorum_manager_container_command(security: &ValidatedKafkaSecurity) -> St derive_pod_index = DERIVE_POD_INDEX, export_replica_id = EXPORT_REPLICA_ID, extract_bootstrap_servers = extract_bootstrap_servers_command(), - set_realm_env = export_kerberos_realm_command(security).unwrap_or_default(), config_dir = STACKABLE_CONFIG_DIR, controller_properties_file = ConfigFileName::ControllerProperties, admin_client_source = ADMIN_CLIENT_PROPERTIES_SOURCE_PATH, @@ -404,46 +394,31 @@ mod tests { use crate::controller::test_support::{kerberos, plaintext}; #[test] - fn controller_command_does_not_export_a_realm_without_kerberos() { - let command = controller_kafka_container_command(&plaintext(), vec![]); - assert!(!command.contains("KERBEROS_REALM")); - } - - #[test] - fn controller_command_always_templates_the_jaas_file() { - // `jaas.properties` is always present in the ConfigMap (empty when Kerberos is off), - // so the copy is unconditional, matching `broker_start_command`. + fn controller_command_has_no_kerberos_setup() { + // The controllers never use Kerberos: the nodes authenticate each other with their + // internal pod certificates, also when Kerberos is enabled for clients. for security in [kerberos(), plaintext()] { let command = controller_kafka_container_command(&security, vec![]); - assert!(command.contains("cp /stackable/config/jaas.properties /tmp/jaas.properties")); - assert!(command.contains("config-utils template /tmp/jaas.properties")); + assert!(!command.contains("KERBEROS_REALM")); + assert!(!command.contains("jaas.properties")); } + assert!(!quorum_manager_container_command().contains("KERBEROS_REALM")); } #[test] fn quorum_manager_templates_the_admin_client_config() { - let command = quorum_manager_container_command(&kerberos()); + let command = quorum_manager_container_command(); assert!( command.contains( "cp /stackable/config/admin-client.properties /tmp/admin-client.properties" ) ); assert!(command.contains("config-utils template /tmp/admin-client.properties")); - // It must connect with the *rendered* copy, not the raw ConfigMap file, or the - // `${env:...}` placeholders in `sasl.jaas.config` reach the JAAS parser verbatim. + // It must connect with the *rendered* copy, not the raw ConfigMap file. assert!(command.contains("ADMIN_CLIENT_CONFIG=/tmp/admin-client.properties")); assert!(!command.contains("ADMIN_CLIENT_CONFIG=/stackable/config/admin-client.properties")); } - #[test] - fn quorum_manager_exports_the_kerberos_realm() { - // The sidecar is a separate container: it inherits nothing from the kafka container's - // startup, so it must derive $KERBEROS_REALM itself for `config-utils template` to - // resolve the principal. - let command = quorum_manager_container_command(&kerberos()); - assert!(command.contains("KERBEROS_REALM")); - } - #[test] fn controller_command_templates_the_admin_client_config_for_pre_stop() { let command = controller_kafka_container_command(&kerberos(), vec![]); @@ -457,7 +432,7 @@ mod tests { #[test] fn quorum_manager_container_command_targets_the_bootstrap_servers_not_localhost() { - let command = quorum_manager_container_command(&kerberos()); + let command = quorum_manager_container_command(); assert!(command.contains( "grep '^controller.quorum.bootstrap.servers=' /stackable/config/controller.properties" )); @@ -475,7 +450,7 @@ mod tests { /// removed. Backgrounding it and having the trap actively `kill` it closes that window. #[test] fn quorum_manager_container_command_kills_an_in_flight_add_controller_attempt_on_term() { - let command = quorum_manager_container_command(&kerberos()); + let command = quorum_manager_container_command(); assert!(command.contains("trap 'handle_term_signal' TERM")); // The poll interval is slept in the background and `wait`ed on, so bash can run the // trap immediately instead of only after a foreground `sleep` returns. @@ -517,13 +492,13 @@ mod tests { #[test] fn quorum_manager_container_command_string_merges_controller_and_admin_client_properties_for_add_controller() { - let command = quorum_manager_container_command(&kerberos()); + let command = quorum_manager_container_command(); // Renders this controller's own `controller.properties` (carries `node.id` and // `listeners`) via the same REPLICA_ID derivation used by the `kafka` container. assert!(command.contains("export REPLICA_ID=$((POD_INDEX + NODE_ID_OFFSET))")); assert!(command.contains("config-utils template /tmp/controller.properties")); // Merges it with the *rendered* admin-client config (carries `security.protocol`, - // `ssl.*` and, under Kerberos, `sasl.jaas.config`), controller.properties first so + // `ssl.*`), controller.properties first so // the client config wins on any key collision (see `ADD_CONTROLLER_PROPERTIES_PATH`). assert!(command.contains( "cat /tmp/controller.properties /tmp/admin-client.properties > /tmp/add-controller.properties" @@ -846,7 +821,7 @@ mod tests { /// its own `timeout` wrapper, delaying pod termination by minutes. #[test] fn every_cli_call_has_a_kill_after_so_timeout_is_actually_enforced() { - let container_command = quorum_manager_container_command(&kerberos()); + let container_command = quorum_manager_container_command(); // Both scripts are handed the CLI path by their operator-generated preamble and // then refer to it by variable, so that assignment is the only place the literal diff --git a/rust/operator-binary/src/controller/build/kerberos.rs b/rust/operator-binary/src/controller/build/kerberos.rs index 0f7af858f..527b6ffe3 100644 --- a/rust/operator-binary/src/controller/build/kerberos.rs +++ b/rust/operator-binary/src/controller/build/kerberos.rs @@ -42,86 +42,40 @@ pub enum Error { /// Brokers are exposed through listener-operator `Listener` volumes (the broker and bootstrap /// listeners), so the keytab's principals must cover both. /// -/// # Panics -/// -/// Panics if the volume mount cannot be added to the container builder. Only call this on a -/// container builder whose mount paths are still distinct from the ones added here. -pub fn mount_broker_kerberos_volume( - kafka_security: &ValidatedKafkaSecurity, - cb_kafka: &mut ContainerBuilder, -) -> Result, Error> { - let Some(kerberos_secret_class) = kafka_security.kerberos_secret_class() else { - return Ok(None); - }; - let mut keytab = keytab_volume_source(&kerberos_secret_class); - keytab - .with_listener_volume_scope(&*LISTENER_BROKER_VOLUME_NAME) - .with_listener_volume_scope(&*LISTENER_BOOTSTRAP_VOLUME_NAME); - mount_keytab_volume( - &mut keytab, - KafkaRole::Broker.kerberos_service_name(), - cb_kafka, - ) - .map(Some) -} - -/// Mounts the Kerberos keytab and `krb5.conf` volume into a controller's `kafka` container and -/// returns the pod volume, when Kerberos is enabled; `None` otherwise. -/// -/// KRaft controllers have no listener-operator `Listener` volume: they are only reachable through -/// their own StatefulSet pod DNS name, so the keytab must be pod-scoped, matching how the -/// controller's internal TLS certificate is provisioned in -/// [`mount_controller_tls_volume`](super::security::mount_controller_tls_volume). +/// Only brokers get a keytab: Kerberos is used on the client-facing listeners only. The Kafka +/// nodes authenticate each other with their internal pod certificates, so KRaft controllers +/// have no Kerberos configuration at all. /// /// # Panics /// /// Panics if the volume mount cannot be added to the container builder. Only call this on a /// container builder whose mount paths are still distinct from the ones added here. -pub fn mount_controller_kerberos_volume( +pub fn mount_broker_kerberos_volume( kafka_security: &ValidatedKafkaSecurity, cb_kafka: &mut ContainerBuilder, ) -> Result, Error> { let Some(kerberos_secret_class) = kafka_security.kerberos_secret_class() else { return Ok(None); }; - let mut keytab = keytab_volume_source(&kerberos_secret_class); - keytab.with_pod_scope(); - mount_keytab_volume( - &mut keytab, - KafkaRole::Controller.kerberos_service_name(), - cb_kafka, - ) - .map(Some) -} - -/// The secret-operator volume source for the keytab and `krb5.conf`, before it is scoped to the -/// addresses the role's principals must cover. -fn keytab_volume_source(kerberos_secret_class: &str) -> SecretOperatorVolumeSourceBuilder { - SecretOperatorVolumeSourceBuilder::new( - kerberos_secret_class, + let kerberos_secret_operator_volume = SecretOperatorVolumeSourceBuilder::new( + &kerberos_secret_class, // We need both public (krb5.conf) and private (keytab) parts. SecretClassVolumeProvisionParts::PublicPrivate, ) -} - -/// Builds the scoped `keytab` volume source into the pod volume and mounts it into the `kafka` -/// container. -fn mount_keytab_volume( - keytab: &mut SecretOperatorVolumeSourceBuilder, - kerberos_service_name: &str, - cb_kafka: &mut ContainerBuilder, -) -> Result { - let kerberos_secret_operator_volume = keytab - .with_kerberos_service_name(kerberos_service_name) - .build() - .context(KerberosSecretVolumeSnafu)?; + .with_listener_volume_scope(&*LISTENER_BROKER_VOLUME_NAME) + .with_listener_volume_scope(&*LISTENER_BOOTSTRAP_VOLUME_NAME) + .with_kerberos_service_name(KafkaRole::Broker.kerberos_service_name()) + .build() + .context(KerberosSecretVolumeSnafu)?; cb_kafka .add_volume_mount(&*KERBEROS_VOLUME_NAME, STACKABLE_KERBEROS_DIR) .expect("The mount paths are statically defined and there should be no duplicates."); - Ok(VolumeBuilder::new(&*KERBEROS_VOLUME_NAME) - .ephemeral(kerberos_secret_operator_volume) - .build()) + Ok(Some( + VolumeBuilder::new(&*KERBEROS_VOLUME_NAME) + .ephemeral(kerberos_secret_operator_volume) + .build(), + )) } constant!(pub KRB5_CONFIG: EnvVarName = "KRB5_CONFIG"); diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index 30664d8ab..f3943a33a 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -741,7 +741,6 @@ mod tests { &*STACKABLE_LOG_DIR_NAME, ], KafkaRole::Controller => vec![ - &*KERBEROS_VOLUME_NAME, &*STACKABLE_LOG_CONFIG_DIR_NAME, &*STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME, &*OPA_TLS_VOLUME_NAME, 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 c5de74c62..9c754d882 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/properties/listener.rs b/rust/operator-binary/src/controller/build/properties/listener.rs index 2cb533769..f21f075c6 100644 --- a/rust/operator-binary/src/controller/build/properties/listener.rs +++ b/rust/operator-binary/src/controller/build/properties/listener.rs @@ -114,14 +114,11 @@ pub fn get_kafka_listener_config( port: kafka_security.internal_port().to_string(), }); listener_security_protocol_map.insert(KafkaListenerName::Internal, KafkaListenerProtocol::Ssl); - listener_security_protocol_map.insert( - KafkaListenerName::Controller, - if kafka_security.has_kerberos_enabled() { - KafkaListenerProtocol::SaslSsl - } else { - KafkaListenerProtocol::Ssl - }, - ); + // The Kafka nodes authenticate each other with the pod certificates of the internal + // `SecretClass`, also when Kerberos is enabled. Kerberos principals do not identify a + // specific node, so they must not be used for inter-node traffic. + listener_security_protocol_map + .insert(KafkaListenerName::Controller, KafkaListenerProtocol::Ssl); // BOOTSTRAP if kafka_security.has_kerberos_enabled() { @@ -515,7 +512,7 @@ mod tests { bootstrap_name = KafkaListenerName::Bootstrap, bootstrap_protocol = KafkaListenerProtocol::SaslSsl, controller_name = KafkaListenerName::Controller, - controller_protocol = KafkaListenerProtocol::SaslSsl, + controller_protocol = KafkaListenerProtocol::Ssl, ) ); } diff --git a/rust/operator-binary/src/controller/build/properties/mod.rs b/rust/operator-binary/src/controller/build/properties/mod.rs index 29dca78a3..565776a9d 100644 --- a/rust/operator-binary/src/controller/build/properties/mod.rs +++ b/rust/operator-binary/src/controller/build/properties/mod.rs @@ -124,13 +124,7 @@ pub(crate) fn opa_authorizer_properties( "opa.authorizer.url".to_string(), opa_connect_string.to_string(), ), - ( - "super.users".to_string(), - super_users( - pod_descriptors, - cluster_config.kafka_security.has_kerberos_enabled(), - ), - ), + ("super.users".to_string(), super_users(pod_descriptors)), ]) } @@ -139,7 +133,14 @@ pub(crate) fn opa_authorizer_properties( /// Internal traffic is authenticated with the pod certificates of the internal `SecretClass` /// (mTLS), whose subject DN is `CN=generated certificate for pod,DC=,DC=, /// DC=,DC=svc,DC=`, see the secret-operator. -/// With Kerberos, the node principals are mapped to the short name of the service. +/// +/// # Why there are no Kerberos principals +/// +/// The nodes never authenticate each other with Kerberos, also when Kerberos is enabled for +/// clients: the INTERNAL and CONTROLLER listeners are SSL. Kerberos principals must not be added +/// here. Kafka compares `super.users` with the principal after +/// `sasl.kerberos.principal.to.local.rules`, which by default map `kafka/@` to +/// `kafka` for any host, so an entry like `User:kafka` would not be tied to the Kafka nodes. /// /// # Why the `CN=generated certificate for pod` entry is required /// @@ -159,8 +160,8 @@ pub(crate) fn opa_authorizer_properties( /// See the Kafka documentation: /// - [Authorization and ACLs](https://kafka.apache.org/44/security/authorization-and-acls/) /// - [`ssl.principal.mapping.rules`](https://kafka.apache.org/44/security/authorization-and-acls/#customizing-ssl-user-name) -fn super_users(pod_descriptors: &[KafkaPodDescriptor], kerberos_enabled: bool) -> String { - let mut principals: BTreeSet = pod_descriptors +fn super_users(pod_descriptors: &[KafkaPodDescriptor]) -> String { + let principals: BTreeSet = pod_descriptors .iter() .map(|desc| { // The certificate encodes `CN`, then the `DC`s from the pod up to the cluster domain @@ -173,14 +174,6 @@ fn super_users(pod_descriptors: &[KafkaPodDescriptor], kerberos_enabled: bool) - }) .collect(); - if kerberos_enabled { - principals.extend( - [KafkaRole::Broker, KafkaRole::Controller] - .iter() - .map(|role| format!("User:{}", role.kerberos_service_name())), - ); - } - principals.into_iter().collect::>().join(";") } @@ -237,18 +230,38 @@ mod tests { ]; assert_eq!( - super_users(&pod_descriptors, false), + super_users(&pod_descriptors), "User:DC=local,DC=cluster,DC=svc,DC=default,DC=kafka-controller-default-headless,DC=kafka-controller-default-0,CN=generated certificate for pod;\ User:DC=local,DC=cluster,DC=svc,DC=default,DC=kafka-controller-default-headless,DC=kafka-controller-default-1,CN=generated certificate for pod" ); } #[test] - fn super_users_contains_the_service_name_with_kerberos() { - let pod_descriptors = vec![pod_descriptor(KafkaRole::Controller, 0, 9093)]; + fn opa_super_users_contain_only_pod_certificate_subjects_with_kerberos() { + let pod_descriptors = vec![ + pod_descriptor(KafkaRole::Controller, 0, 9093), + pod_descriptor(KafkaRole::Broker, 1, 9092), + ]; + let cluster_config = ValidatedClusterConfig { + kafka_security: crate::controller::test_support::kerberos(), + authorization_config: Some(crate::crd::authorization::KafkaAuthorizationConfig { + opa_connect: "http://opa.default.svc.cluster.local:8081/v1/data/kafka".to_owned(), + secret_class: None, + }), + metadata_manager: crate::crd::MetadataManager::KRaft, + zookeeper_config_map_name: None, + broker_id_pod_config_map_name: None, + }; + + let properties = opa_authorizer_properties(&cluster_config, &pod_descriptors); + let super_users = properties.get("super.users").expect("super.users is set"); - assert!(super_users(&pod_descriptors, true).ends_with(";User:kafka")); - assert!(!super_users(&pod_descriptors, false).contains("User:kafka")); + for entry in super_users.split(';') { + assert!( + entry.starts_with("User:") && entry.ends_with(",CN=generated certificate for pod"), + "every super user must be a pod certificate subject, got [{entry}]" + ); + } } #[test] diff --git a/rust/operator-binary/src/controller/build/role_group/broker.rs b/rust/operator-binary/src/controller/build/role_group/broker.rs index d496124dd..99b019b8b 100644 --- a/rust/operator-binary/src/controller/build/role_group/broker.rs +++ b/rust/operator-binary/src/controller/build/role_group/broker.rs @@ -18,7 +18,7 @@ use stackable_operator::{ }, }; -use super::{RoleGroupBuilder, controller_jaas_section}; +use super::RoleGroupBuilder; use crate::{ controller::{ BrokerRoleGroupConfig, PodDescriptorsError, RoleGroupName, ValidatedBrokerConfig, @@ -367,12 +367,6 @@ fn jaas_config_file(is_kerberos_enabled: bool) -> String { return String::new(); } - // Broker pods reach the CONTROLLER listener as SASL clients; the only principals in - // their keytab (see `mount_broker_kerberos_volume`) are for the broker and bootstrap listener - // addresses, so their CONTROLLER section must reuse the broker address. - let controller_section = - controller_jaas_section(&node_address_cmd(STACKABLE_LISTENER_BROKER_DIR)); - formatdoc! {" bootstrap.KafkaServer {{ com.sun.security.auth.module.Krb5LoginModule required @@ -391,8 +385,6 @@ fn jaas_config_file(is_kerberos_enabled: bool) -> String { keyTab=\"{STACKABLE_KERBEROS_KEYTAB_PATH}\" principal=\"kafka/{broker_address}@${{env:KERBEROS_REALM}}\"; }}; - - {controller_section} ", bootstrap_address = node_address_cmd(STACKABLE_LISTENER_BOOTSTRAP_DIR), broker_address = node_address_cmd(STACKABLE_LISTENER_BROKER_DIR), @@ -483,5 +475,8 @@ mod tests { // The bootstrap and client principals embed distinct listener addresses. assert!(jaas.contains("/stackable/listener-bootstrap")); assert!(jaas.contains("/stackable/listener-broker")); + // The nodes authenticate each other with their internal pod certificates, never with + // Kerberos, so there is no login context for the CONTROLLER listener. + assert!(!jaas.contains("controller.KafkaServer")); } } diff --git a/rust/operator-binary/src/controller/build/role_group/controller.rs b/rust/operator-binary/src/controller/build/role_group/controller.rs index 3368dd11f..e1f201b9f 100644 --- a/rust/operator-binary/src/controller/build/role_group/controller.rs +++ b/rust/operator-binary/src/controller/build/role_group/controller.rs @@ -18,7 +18,7 @@ use stackable_operator::{ }, }; -use super::{RoleGroupBuilder, controller_jaas_section}; +use super::RoleGroupBuilder; use crate::{ controller::{ ControllerRoleGroupConfig, PodDescriptorsError, RoleGroupName, ValidatedCluster, @@ -31,10 +31,6 @@ use crate::{ common_kafka_env, common_operator_env_vars, container_ports, log_config_volume, }, jvm, - kerberos::{ - self, KAFKA_OPTS, KERBEROS_VOLUME_NAME, KRB5_CONFIG, kerberos_env_vars, - mount_controller_kerberos_volume, - }, properties::{ ConfigFileName, controller_properties, listener::get_kafka_listener_config, }, @@ -52,9 +48,8 @@ use crate::{ security::ValidatedKafkaSecurity, }, crd::{ - CONTROLLER_POD_FQDN_TEMPLATE, KAFKA_HEAP_OPTS, LOG_DIRS_VOLUME_NAME, METRICS_PORT, - STACKABLE_CONFIG_DIR, STACKABLE_CONFIG_DIR_NAME, STACKABLE_DATA_DIR, - STACKABLE_KERBEROS_DIR, STACKABLE_KERBEROS_KRB5_PATH, STACKABLE_LOG_CONFIG_DIR, + KAFKA_HEAP_OPTS, LOG_DIRS_VOLUME_NAME, METRICS_PORT, STACKABLE_CONFIG_DIR, + STACKABLE_CONFIG_DIR_NAME, STACKABLE_DATA_DIR, STACKABLE_LOG_CONFIG_DIR, STACKABLE_LOG_CONFIG_DIR_NAME, STACKABLE_LOG_DIR_NAME, role::{KAFKA_NODE_ID_OFFSET, KafkaRole, controller::ControllerContainer}, }, @@ -83,9 +78,6 @@ pub enum Error { #[snafu(display("failed to build the TLS volume"))] BuildInternalTlsVolume { source: security::Error }, - #[snafu(display("failed to build the Kerberos volume"))] - BuildKerberosVolume { source: kerberos::Error }, - #[snafu(display("failed to build container probe"))] BuildProbe { source: probes::Error }, @@ -154,8 +146,11 @@ pub(crate) fn build<'a>( }) } -/// `controller.properties` rendered with the role group's overrides, the controller's JAAS -/// config, and `admin-client.properties`. +/// `controller.properties` rendered with the role group's overrides, and +/// `admin-client.properties`. +/// +/// Controllers have no JAAS config: the CONTROLLER listener is SSL also when Kerberos is +/// enabled, and the nodes authenticate each other with their internal pod certificates. fn role_config_map_data( cluster: &ValidatedCluster, role_group_name: &RoleGroupName, @@ -190,20 +185,12 @@ fn role_config_map_data( file: ConfigFileName::ControllerProperties, })?, ), - // This file contains the JAAS configuration for Kerberos authentication - // It has the ".properties" extension but is not a Java properties file. - // It is processed by `config-utils` to substitute "env:" and "file:" variables - // and this tool currently doesn't support the JAAS login configuration format. - ( - ConfigFileName::Jaas.to_string(), - jaas_config_file(kafka_security.has_kerberos_enabled()), - ), // `admin-client.properties` is only needed by the controller-side sidecar running // `kafka-metadata-quorum.sh` against the CONTROLLER listener; brokers don't need it. ( ConfigFileName::AdminClient.to_string(), to_java_properties_string( - controller_admin_client_properties(kafka_security) + controller_admin_client_properties() .iter() .filter_map(|(k, v)| v.as_ref().map(|v| (k, v))), ) @@ -229,19 +216,11 @@ fn role_pod( let mut cb_kafka = new_container_builder(ControllerContainer::Kafka.name()); - // The push order below (Kerberos, `log-config`, then the internal TLS volume, which is not - // next to Kerberos as it is for brokers) is part of the pod template: a pod's `volumes` is - // an ordered list, so changing the order changes the template of every deployed StatefulSet - // and restarts its pods. The order is pinned by `pod_volumes_keep_their_order` in - // `build/mod.rs`. + // The push order below is part of the pod template: a pod's `volumes` is an ordered list, + // so changing the order changes the template of every deployed StatefulSet and restarts its pods. + // The order is pinned by `pod_volumes_keep_their_order` in `build/mod.rs`. let mut pod_volumes: Vec = Vec::new(); - if let Some(kerberos_volume) = mount_controller_kerberos_volume(kafka_security, &mut cb_kafka) - .context(BuildKerberosVolumeSnafu)? - { - pod_volumes.push(kerberos_volume); - } - let node_id_offset = node_id_hash32_offset(&ROLE, role_group_name.as_ref()).to_string(); // Operator-set env vars first (common + controller-specific); the user's `envOverrides` @@ -265,10 +244,6 @@ fn role_pod( ) .context(ConstructJvmArgumentsSnafu)?, ) - // Kerberos env goes on the `kafka` container only. `controller_shared_env` is also - // the `quorum-manager` sidecar's base, and `KAFKA_OPTS` points the JVM at - // `/tmp/jaas.properties`, which only the `kafka` container renders. - .merge(kerberos_env_vars(kafka_security)) .merge(rg_config.env_overrides.clone()) .into(); @@ -354,7 +329,7 @@ fn role_pod( let containers = vec![ cb_kafka.build(), - build_quorum_manager_container(image, kafka_security, quorum_manager_env), + build_quorum_manager_container(image, quorum_manager_env), ]; Ok((containers, pod_volumes)) @@ -399,12 +374,11 @@ fn controller_pod_shared_env_vars( /// Builds the `quorum-manager` sidecar for a controller pod. fn build_quorum_manager_container( resolved_product_image: &ResolvedProductImage, - kafka_security: &ValidatedKafkaSecurity, env: EnvVarSet, ) -> Container { let mut cb = new_container_builder(&QUORUM_MANAGER_CONTAINER_NAME); - let mut local_env = EnvVarSet::new() + let local_env = EnvVarSet::new() // `kafka-metadata-quorum.sh` goes through `kafka-run-class.sh`, which defaults // `KAFKA_HEAP_OPTS` to `-Xmx256M` when unset. Set an explicit, modest heap so the // JVM's max heap plus its base/metaspace/SSL-buffer overhead stays comfortably @@ -415,7 +389,7 @@ fn build_quorum_manager_container( .command(vec![ "/bin/bash".to_string(), "-c".to_string(), - quorum_manager_container_command(kafka_security), + quorum_manager_container_command(), ]) .resources( ResourceRequirementsBuilder::new() @@ -441,49 +415,12 @@ fn build_quorum_manager_container( .add_volume_mount(&*LOG_DIRS_VOLUME_NAME, STACKABLE_DATA_DIR) .expect("The mount paths are statically defined and there should be no duplicates."); - if kafka_security.has_kerberos_enabled() { - // `controller_admin_client_properties` authenticates with the pod-scoped keytab - // mounted by `mount_controller_kerberos_volume`; the volume is already on the pod, this - // container just needs its own mount and `KRB5_CONFIG`. It deliberately does *not* - // get `KAFKA_OPTS`: that points the JVM at `/tmp/jaas.properties`, which only the - // `kafka` container renders. - cb.add_volume_mount(&*KERBEROS_VOLUME_NAME, STACKABLE_KERBEROS_DIR) - .expect("The mount paths are statically defined and there should be no duplicates."); - - local_env = local_env - .with_value(&KRB5_CONFIG, STACKABLE_KERBEROS_KRB5_PATH) - // `KRB5_CONFIG` only reaches native MIT tools; the JVM reads the - // `java.security.krb5.conf` system property, without which the admin client fails - // with "Unable to locate KDC for realm". Unlike the `kafka` container's `KAFKA_OPTS` - // this deliberately omits `java.security.auth.login.config`: that points at - // `/tmp/jaas.properties`, which only the `kafka` container renders. This container - // authenticates with the inline `sasl.jaas.config` in `admin-client.properties`. - .with_value( - &KAFKA_OPTS, - format!("-Djava.security.krb5.conf={STACKABLE_KERBEROS_KRB5_PATH}"), - ); - } - // This allows to override explicit env vars set in this function. cb.add_env_vars(local_env.merge(env)); cb.build() } -// Generate JAAS configuration file for Kerberos authentication -// or an empty string if Kerberos is not enabled. -// See https://docs.oracle.com/javase/8/docs/technotes/guides/security/jgss/tutorials/LoginConfigFile.html -fn jaas_config_file(is_kerberos_enabled: bool) -> String { - if !is_kerberos_enabled { - return String::new(); - } - - // Controller pods have no listener-operator `Listener` volume; their keytab is - // pod-scoped, so their CONTROLLER section uses their own pod FQDN — the same expression - // already used for `KAFKA_LISTENERS` in `controller_properties.rs`. - controller_jaas_section(CONTROLLER_POD_FQDN_TEMPLATE) -} - #[cfg(test)] mod tests { use super::*; @@ -502,43 +439,26 @@ mod tests { } } - /// The `quorum-manager` sidecar sets `KAFKA_HEAP_OPTS` unconditionally, and - /// `KRB5_CONFIG`/`KAFKA_OPTS` when Kerberos is enabled. All three must still be - /// overridable by the caller's `env` (in practice, the rolegroup's `envOverrides`), the - /// same way [`env_overrides_override_operator_set_env_vars`] proves it for the `kafka` - /// container's env. + /// The `quorum-manager` sidecar sets `KAFKA_HEAP_OPTS`, which must still be overridable by + /// the rolegroup's `envOverrides`. #[test] fn quorum_manager_container_env_can_be_overridden() { let resolved_product_image = test_resolved_product_image(); - let kafka_security = crate::controller::test_support::kerberos(); - assert!( - kafka_security.has_kerberos_enabled(), - "sanity check failed: the fixture must have Kerberos enabled so that KRB5_CONFIG \ - and KAFKA_OPTS are set in the first place" - ); - - let overrides = EnvVarSet::new() - .with_value(&KAFKA_HEAP_OPTS, "-Xmx999M") - .with_value(&KRB5_CONFIG, "/custom/krb5.conf") - .with_value(&KAFKA_OPTS, "-Doverridden=true"); + let overrides = EnvVarSet::new().with_value(&KAFKA_HEAP_OPTS, "-Xmx999M"); - let container = - build_quorum_manager_container(&resolved_product_image, &kafka_security, overrides); + let container = build_quorum_manager_container(&resolved_product_image, overrides); let env = container.env.expect("the sidecar has env vars"); - for (name, expected_value) in [ - ("KAFKA_HEAP_OPTS", "-Xmx999M"), - ("KRB5_CONFIG", "/custom/krb5.conf"), - ("KAFKA_OPTS", "-Doverridden=true"), - ] { - let matching: Vec<_> = env.iter().filter(|env_var| env_var.name == name).collect(); - assert_eq!( - matching.len(), - 1, - "the override must replace the operator-set {name}, not duplicate it" - ); - assert_eq!(matching[0].value.as_deref(), Some(expected_value)); - } + let matching: Vec<_> = env + .iter() + .filter(|env_var| env_var.name == "KAFKA_HEAP_OPTS") + .collect(); + assert_eq!( + matching.len(), + 1, + "the override must replace the operator-set KAFKA_HEAP_OPTS, not duplicate it" + ); + assert_eq!(matching[0].value.as_deref(), Some("-Xmx999M")); } #[test] @@ -550,9 +470,4 @@ mod tests { let _ = *CLUSTER_DOMAIN; let _ = *QUORUM_MANAGER_CONTAINER_NAME; } - - #[test] - fn jaas_config_file_empty_without_kerberos() { - assert_eq!(jaas_config_file(false), ""); - } } diff --git a/rust/operator-binary/src/controller/build/role_group/mod.rs b/rust/operator-binary/src/controller/build/role_group/mod.rs index 5c07d0a48..08f912ea9 100644 --- a/rust/operator-binary/src/controller/build/role_group/mod.rs +++ b/rust/operator-binary/src/controller/build/role_group/mod.rs @@ -14,7 +14,6 @@ pub(crate) mod broker; pub(crate) mod controller; -use indoc::formatdoc; use snafu::{ResultExt, Snafu}; use stackable_operator::{ commons::affinity::StackableAffinity, @@ -29,7 +28,7 @@ use stackable_operator::{ use super::resource; use crate::{ controller::{RoleGroupName, ValidatedCluster, validate::ValidatedLogging}, - crd::{STACKABLE_KERBEROS_KEYTAB_PATH, role::KafkaRole}, + crd::role::KafkaRole, }; /// Why emitting one role group's Kubernetes objects failed. The caller adds which role and role @@ -109,23 +108,3 @@ impl RoleGroupBuilder<'_> { resource::statefulset::build_statefulset(self).context(StatefulSetSnafu) } } - -/// The JAAS section for the CONTROLLER listener, which both roles render, authenticating as -/// `kafka/`. -/// -/// Unlike the broker's bootstrap and client sections, this context is used for BOTH sides of -/// every CONTROLLER-listener connection: brokers connect out to controllers, and controllers -/// connect to each other for Raft. This is the only listener in this operator where the process -/// must act as a GSSAPI initiator as well as an acceptor, so `isInitiator` is intentionally left -/// at its default (`true`). -fn controller_jaas_section(controller_principal_address: &str) -> String { - formatdoc! {" - controller.KafkaServer {{ - com.sun.security.auth.module.Krb5LoginModule required - useKeyTab=true - storeKey=true - keyTab=\"{STACKABLE_KERBEROS_KEYTAB_PATH}\" - principal=\"kafka/{controller_principal_address}@${{env:KERBEROS_REALM}}\"; - }}; - "} -} diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs index 141f5a23e..b869c06bd 100644 --- a/rust/operator-binary/src/controller/build/security.rs +++ b/rust/operator-binary/src/controller/build/security.rs @@ -21,7 +21,7 @@ use stackable_operator::{ use crate::{ controller::{build::command::export_kerberos_realm_command, security::ValidatedKafkaSecurity}, crd::{ - CONTROLLER_POD_FQDN_TEMPLATE, LISTENER_BOOTSTRAP_VOLUME_NAME, LISTENER_BROKER_VOLUME_NAME, + LISTENER_BOOTSTRAP_VOLUME_NAME, LISTENER_BROKER_VOLUME_NAME, STACKABLE_KERBEROS_KEYTAB_PATH, STACKABLE_LISTENER_BROKER_DIR, listener::{ self, KafkaListenerName, KafkaListenerProtocol, node_address_cmd_env, node_port_cmd_env, @@ -46,9 +46,7 @@ const PROPERTY_SECURITY_PROTOCOL: &str = "security.protocol"; const PROPERTY_SASL_ENABLED_MECHANISMS: &str = "sasl.enabled.mechanisms"; const PROPERTY_SASL_KERBEROS_SERVICE_NAME: &str = "sasl.kerberos.service.name"; const PROPERTY_SASL_INTER_BROKER_MECHANISM: &str = "sasl.mechanism.inter.broker.protocol"; -const PROPERTY_SASL_CONTROLLER_MECHANISM: &str = "sasl.mechanism.controller.protocol"; const PROPERTY_SASL_MECHANISM: &str = "sasl.mechanism"; -const PROPERTY_SASL_JAAS_CONFIG: &str = "sasl.jaas.config"; pub(crate) const STACKABLE_TLS_KAFKA_INTERNAL_DIR: &str = "/stackable/tls-kafka-internal"; constant!(pub(crate) STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME: VolumeName = "tls-kafka-internal"); @@ -205,54 +203,17 @@ pub fn client_properties(security: &ValidatedKafkaSecurity) -> Vec<(String, Opti /// (e.g. `kafka-metadata-quorum.sh`) talking to the CONTROLLER listener from *inside* a /// controller pod, over the `tls-kafka-internal` volume mounted by /// `mount_controller_tls_volume`. -/// When Kerberos is enabled the CONTROLLER listener is `SASL_SSL` (see -/// [`get_kafka_listener_config`][glc]), so these calls must authenticate with GSSAPI. They do -/// so as the controller's *own* pod principal, from the pod-scoped keytab mounted by -/// [`mount_controller_kerberos_volume`] — the correct identity for a voter registering itself. /// -/// The principal contains `${env:…}` placeholders, so the rendered file must be passed -/// through `config-utils template` before use; see [`quorum_manager_container_command`][qmcc]. +/// The CONTROLLER listener is always `SSL`, also when Kerberos is enabled (see +/// [`get_kafka_listener_config`][glc]), so the tool authenticates with the controller's internal +/// pod certificate. /// /// [glc]: crate::controller::build::properties::listener::get_kafka_listener_config -/// [`mount_controller_kerberos_volume`]: crate::controller::build::kerberos::mount_controller_kerberos_volume -/// [qmcc]: crate::controller::build::command::quorum_manager_container_command -pub fn controller_admin_client_properties( - security: &ValidatedKafkaSecurity, -) -> Vec<(String, Option)> { - let mut properties = vec![]; - - if security.has_kerberos_enabled() { - properties.push(( - PROPERTY_SECURITY_PROTOCOL.to_string(), - Some(KafkaListenerProtocol::SaslSsl.to_string()), - )); - // Client-side mechanism selection. `sasl.enabled.mechanisms` is the *broker-side* - // list of accepted mechanisms and has no effect here. - properties.push(( - PROPERTY_SASL_MECHANISM.to_string(), - Some(SASL_MECHANISM_GSSAPI.to_string()), - )); - properties.push(( - PROPERTY_SASL_KERBEROS_SERVICE_NAME.to_string(), - Some(KafkaRole::Controller.kerberos_service_name().to_string()), - )); - properties.push(( - PROPERTY_SASL_JAAS_CONFIG.to_string(), - Some(format!( - "com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true \ - storeKey=true keyTab=\"{keytab}\" \ - principal=\"{service}/{pod_fqdn}@${{env:KERBEROS_REALM}}\";", - keytab = STACKABLE_KERBEROS_KEYTAB_PATH, - service = KafkaRole::Controller.kerberos_service_name(), - pod_fqdn = CONTROLLER_POD_FQDN_TEMPLATE, - )), - )); - } else { - properties.push(( - PROPERTY_SECURITY_PROTOCOL.to_string(), - Some(KafkaListenerProtocol::Ssl.to_string()), - )); - } +pub fn controller_admin_client_properties() -> Vec<(String, Option)> { + let mut properties = vec![( + PROPERTY_SECURITY_PROTOCOL.to_string(), + Some(KafkaListenerProtocol::Ssl.to_string()), + )]; push_client_ssl_stores(&mut properties, STACKABLE_TLS_KAFKA_INTERNAL_DIR); @@ -524,10 +485,6 @@ pub fn broker_config_settings(security: &ValidatedKafkaSecurity) -> BTreeMap BTreeMap BTreeMap { +pub fn controller_config_settings(security: &ValidatedKafkaSecurity) -> BTreeMap { let mut config = BTreeMap::new(); insert_listener_ssl_stores( @@ -582,46 +536,19 @@ 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. - // - // 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 - config.insert( - KafkaListenerName::Controller.listener_ssl_client_auth(), - SSL_CLIENT_AUTH_REQUIRED.to_string(), - ); - } + // The CONTROLLER listener is SSL in every configuration, Kerberos included, and always + // requires client authentication, like the INTERNAL listener: the Kafka nodes authenticate + // each other with the pod certificates of the internal `SecretClass`, whose subjects are + // pinned to the pod, and never with Kerberos. All clients of this listener (brokers, the + // other controllers and the quorum management admin client) present that certificate. With + // OPA, the nodes are identified as super users by its subject. + config.insert( + KafkaListenerName::Controller.listener_ssl_client_auth(), + SSL_CLIENT_AUTH_REQUIRED.to_string(), + ); insert_opa_truststore_settings(&mut config, security); - // Kerberos - if security.has_kerberos_enabled() { - config.insert( - PROPERTY_SASL_ENABLED_MECHANISMS.to_string(), - SASL_MECHANISM_GSSAPI.to_string(), - ); - config.insert( - PROPERTY_SASL_KERBEROS_SERVICE_NAME.to_string(), - KafkaRole::Controller.kerberos_service_name().to_string(), - ); - config.insert( - PROPERTY_SASL_INTER_BROKER_MECHANISM.to_string(), - SASL_MECHANISM_GSSAPI.to_string(), - ); - config.insert( - PROPERTY_SASL_CONTROLLER_MECHANISM.to_string(), - SASL_MECHANISM_GSSAPI.to_string(), - ); - tracing::debug!("Kerberos configs added: [{:#?}]", config); - } - config } @@ -907,32 +834,9 @@ mod tests { // ---- controller_admin_client_properties ---- #[test] - fn admin_client_uses_gssapi_over_sasl_ssl_with_kerberos() { - let props = as_map(controller_admin_client_properties(&kerberos())); - assert_eq!( - props.get("security.protocol"), - Some(&Some("SASL_SSL".to_string())) - ); - assert_eq!( - props.get("sasl.mechanism"), - Some(&Some("GSSAPI".to_string())) - ); - assert_eq!( - props.get("sasl.kerberos.service.name"), - Some(&Some("kafka".to_string())) - ); - // The internal TLS stores stay: SASL_SSL is still SSL underneath. - assert_eq!( - props.get("ssl.truststore.location"), - Some(&Some( - "/stackable/tls-kafka-internal/truststore.p12".to_string() - )) - ); - } + fn controller_admin_client_properties_use_the_internal_pod_certificate() { + let props = as_map(controller_admin_client_properties()); - #[test] - fn admin_client_is_unchanged_without_kerberos() { - let props = as_map(controller_admin_client_properties(&internal_tls())); assert_eq!( props.get("security.protocol"), Some(&Some("SSL".to_string())) @@ -945,13 +849,6 @@ mod tests { "/stackable/tls-kafka-internal/keystore.p12".to_string() )) ); - } - - #[test] - fn controller_admin_client_properties_uses_the_internal_tls_directory() { - let security = server_tls(); - let props = as_map(controller_admin_client_properties(&security)); - assert_eq!( props.get("security.protocol"), Some(&Some("SSL".to_string())) @@ -968,39 +865,6 @@ mod tests { ); } - #[test] - fn controller_admin_client_properties_includes_keystore_when_client_auth_is_required() { - let security = client_auth_tls(); - let props = as_map(controller_admin_client_properties(&security)); - - assert_eq!( - props.get("ssl.keystore.location"), - Some(&Some( - "/stackable/tls-kafka-internal/keystore.p12".to_string() - )) - ); - } - - #[test] - fn controller_admin_client_properties_always_uses_tls_even_without_external_client_tls() { - // Internal (broker/controller) TLS is mandatory (`tls_internal_secret_class()` always - // returns a SecretClass, defaulting to "tls"), and `mount_controller_tls_volume` - // unconditionally mounts both the keystore and truststore on controller pods, independent - // of the external client TLS/authentication settings. So even the "plaintext" fixture - // (no external client TLS, no client-cert auth) still needs SSL to reach the CONTROLLER - // listener - mirroring `controller_config_settings`'s unconditional treatment of the same - // listener (see `controller_config_plaintext_has_internal_tls`). - let security = plaintext(); - let props = as_map(controller_admin_client_properties(&security)); - - assert_eq!( - props.get("security.protocol"), - Some(&Some("SSL".to_string())) - ); - assert!(props.contains_key("ssl.truststore.location")); - assert!(props.contains_key("ssl.keystore.location")); - } - // ---- broker_config_settings ---- #[test] @@ -1072,7 +936,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,74 +945,52 @@ 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. + 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())); } #[test] - fn controller_config_opa_requires_client_auth_unless_kerberos() { + fn controller_config_always_requires_client_auth() { 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")); - - assert!(!controller_config_settings(&internal_tls(), false).contains_key(&key)); - assert!(!controller_config_settings(&kerberos(), true).contains_key(&key)); + for security in [ + plaintext(), + server_tls(), + client_auth_tls(), + internal_tls(), + kerberos(), + opa(), + ] { + assert_eq!( + controller_config_settings(&security) + .get(&key) + .map(String::as_str), + Some("required") + ); + } } #[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")); } #[test] - fn broker_config_sets_the_controller_sasl_mechanism_with_kerberos() { - let config = broker_config_settings(&kerberos()); - assert_eq!( - config.get("sasl.mechanism.controller.protocol"), - Some(&"GSSAPI".to_string()) - ); - } - - #[test] - fn controller_config_sets_the_controller_sasl_mechanism_with_kerberos() { - let config = controller_config_settings(&kerberos(), false); - assert_eq!( - config.get("sasl.mechanism.controller.protocol"), - Some(&"GSSAPI".to_string()) - ); - } - - #[test] - fn controller_sasl_mechanism_is_absent_without_kerberos() { - assert!( - !broker_config_settings(&internal_tls()) - .contains_key("sasl.mechanism.controller.protocol") - ); - assert!( - !controller_config_settings(&internal_tls(), false) - .contains_key("sasl.mechanism.controller.protocol") - ); - } - - #[test] - fn controller_config_kerberos_adds_sasl() { - let config = controller_config_settings(&kerberos(), false); - assert_eq!( - config.get("sasl.enabled.mechanisms"), - Some(&"GSSAPI".to_string()) - ); - assert_eq!( - config.get("sasl.kerberos.service.name"), - Some(&"kafka".to_string()) - ); + fn controller_listener_never_uses_sasl() { + for security in [kerberos(), internal_tls()] { + let broker_config = broker_config_settings(&security); + let controller_config = controller_config_settings(&security); + + assert!(!broker_config.contains_key("sasl.mechanism.controller.protocol")); + assert!(!controller_config.contains_key("sasl.mechanism.controller.protocol")); + assert!(!controller_config.contains_key("sasl.enabled.mechanisms")); + assert!(!controller_config.contains_key("sasl.kerberos.service.name")); + } } } diff --git a/rust/operator-binary/src/crd/listener.rs b/rust/operator-binary/src/crd/listener.rs index 8e014d14d..4bd6e3f5a 100644 --- a/rust/operator-binary/src/crd/listener.rs +++ b/rust/operator-binary/src/crd/listener.rs @@ -58,7 +58,8 @@ pub enum KafkaListenerName { /// This listener is defined when Kraft mode is enabled. /// It is responsible for broker/controller as well as controller/controller communications /// and therefore it is present on *both* brokers and controller properties files. - /// The protocol used is SSL, or SASL_SSL when Kerberos is enabled. + /// The only protocol used is SSL with the internal pod certificates, also when Kerberos is + /// enabled. /// The advertised host names are FQDN pod names of the controllers. /// /// Note: there is no listener for client/controller communication. diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index 5035c1213..8ff31fd2f 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -85,7 +85,6 @@ constant!(pub BROKER_ID_POD_MAP_DIR_NAME: VolumeName = "broker-id-pod-map-dir"); /// Used for: /// - the address where the CONTROLLER listener is *bound*. /// - the endpoint registered with the quorum's voter. -/// - when Kerberos is enabled, it is also the host in the controller's Kerberos service principal. pub const CONTROLLER_POD_FQDN_TEMPLATE: &str = "${env:POD_NAME}.${env:ROLEGROUP_HEADLESS_SERVICE_NAME}.${env:NAMESPACE}.svc.${env:CLUSTER_DOMAIN}"; #[derive(Snafu, Debug)] diff --git a/tests/templates/kuttl/kraft-kerberos/60-assert.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/60-assert.yaml.j2 index a8327181c..e93e0aed9 100644 --- a/tests/templates/kuttl/kraft-kerberos/60-assert.yaml.j2 +++ b/tests/templates/kuttl/kraft-kerberos/60-assert.yaml.j2 @@ -8,8 +8,8 @@ commands: # The voter set itself must have changed, not just the StatefulSet: a controller that # starts but never joins the quorum is exactly what this test guards against. # - # Kerberos-specific: the admin client must use the *rendered* /tmp copy. The raw - # ConfigMap file still has unresolved ${env:...} placeholders in sasl.jaas.config. + # The admin client connects to the CONTROLLER listener with the controller's internal + # pod certificate (mTLS), also with Kerberos enabled. # kubectl exec -n $NAMESPACE test-kafka-controller-default-0 -c kafka -- \ /stackable/kafka/bin/kafka-metadata-quorum.sh \ diff --git a/tests/templates/kuttl/kraft-kerberos/70-assert.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/70-assert.yaml.j2 index 83a0857b0..a8594734c 100644 --- a/tests/templates/kuttl/kraft-kerberos/70-assert.yaml.j2 +++ b/tests/templates/kuttl/kraft-kerberos/70-assert.yaml.j2 @@ -8,8 +8,8 @@ commands: # The voter set itself must have changed, not just the StatefulSet: a controller that # starts but never joins the quorum is exactly what this test guards against. # - # Kerberos-specific: the admin client must use the *rendered* /tmp copy. The raw - # ConfigMap file still has unresolved ${env:...} placeholders in sasl.jaas.config. + # The admin client connects to the CONTROLLER listener with the controller's internal + # pod certificate (mTLS), also with Kerberos enabled. # kubectl exec -n $NAMESPACE test-kafka-controller-default-0 -c kafka -- \ /stackable/kafka/bin/kafka-metadata-quorum.sh \ diff --git a/tests/templates/kuttl/kraft-kerberos/70-scale-controller-down.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/70-scale-controller-down.yaml.j2 index 69ade0780..4cb03758e 100644 --- a/tests/templates/kuttl/kraft-kerberos/70-scale-controller-down.yaml.j2 +++ b/tests/templates/kuttl/kraft-kerberos/70-scale-controller-down.yaml.j2 @@ -43,9 +43,9 @@ commands: logging: enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} brokerListenerClass: {{ test_scenario['values']['broker-listener-class'] }} - # bootstrap-listener-class is orthogonal to this test's focus on Kerberos over the - # CONTROLLER listener (that axis is already covered by the plain `kerberos` test - # case), so it is pinned here rather than parameterized as a test dimension. + # bootstrap-listener-class is orthogonal to this test's focus on controller quorum + # scaling with Kerberos enabled (that axis is already covered by the plain `kerberos` + # test case), so it is pinned here rather than parameterized as a test dimension. bootstrapListenerClass: cluster-internal gracefulShutdownTimeout: 30s # speed up tests roleGroups: diff --git a/tests/templates/kuttl/kraft-kerberos/README.md b/tests/templates/kuttl/kraft-kerberos/README.md index a85e47b8d..54f849451 100644 --- a/tests/templates/kuttl/kraft-kerberos/README.md +++ b/tests/templates/kuttl/kraft-kerberos/README.md @@ -7,3 +7,7 @@ client can authenticate via GSSAPI to produce/consume a message. This bundles the KRaft cluster setup from `smoke-kraft` with the KDC deployment, `SecretClass`/`AuthenticationClass` and produce/consume job from `kerberos`. + +The brokers and controllers authenticate each other with their internal pod certificates (mTLS +on the CONTROLLER listener), not with Kerberos. The scale-up and scale-down steps check that the +quorum manager still admits and removes controllers in this setup. diff --git a/tests/templates/kuttl/opa/20_opa.yaml.j2 b/tests/templates/kuttl/opa/20_opa.yaml.j2 index 51c6f411a..75affce4a 100644 --- a/tests/templates/kuttl/opa/20_opa.yaml.j2 +++ b/tests/templates/kuttl/opa/20_opa.yaml.j2 @@ -35,11 +35,6 @@ data: ) } - allow if { - is_external_request - input.requestContext.principal.name == "kafka" - } - allow if { is_external_request input.requestContext.principal.name == "developer" From f20bf00f0be400b109bb50aa669e0e2fa4c06aae Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Sat, 14 Nov 2026 17:48:36 +0100 Subject: [PATCH 2/2] docs: more accurate descriptions for KafkaListenerNames --- rust/operator-binary/src/crd/listener.rs | 59 ++++++++++++++---------- 1 file changed, 35 insertions(+), 24 deletions(-) diff --git a/rust/operator-binary/src/crd/listener.rs b/rust/operator-binary/src/crd/listener.rs index 4bd6e3f5a..ef2892c91 100644 --- a/rust/operator-binary/src/crd/listener.rs +++ b/rust/operator-binary/src/crd/listener.rs @@ -16,53 +16,64 @@ const LISTENER_PORTS_DIR: &str = "ports"; #[derive(strum::Display, Debug, EnumString)] pub enum KafkaListenerProtocol { - /// Unencrypted and unauthenticated HTTP connections + /// Unencrypted and unauthenticated Kafka protocol connections #[strum(serialize = "PLAINTEXT")] Plaintext, - /// Encrypted and server-authenticated HTTPS connections + /// TLS-encrypted Kafka protocol connections, with optional client certificate authentication + /// (`ssl.client.auth`) #[strum(serialize = "SSL")] Ssl, - /// Kerberos authentication + /// SASL authentication over TLS. The operator only configures the GSSAPI (Kerberos) mechanism. #[strum(serialize = "SASL_SSL")] SaslSsl, } #[derive(strum::Display, Debug, EnumString, Ord, Eq, PartialEq, PartialOrd)] pub enum KafkaListenerName { - /// The purpose of this listener is to handle client/broker communications. - /// It can be configured to use the SSL or SASL_SSL (Kerberos) protocols - /// if the brokers use TLS for communication (and possible authentication). - /// The PLAINTEXT protocol is used when `spec.clusterConfig.tls.serverSecretClass: null` - /// The advertised listener hosts are derived from the pod (broker) listener volume. + /// Handles client/broker communication. The protocol is: + /// + /// - SASL_SSL if Kerberos is enabled + /// - SSL if `spec.clusterConfig.tls.serverSecretClass` or a TLS AuthenticationClass is set + /// (client certificates are required with the AuthenticationClass) + /// - PLAINTEXT otherwise + /// + /// The advertised host and port are read from the broker listener volume. #[strum(serialize = "CLIENT")] Client, - /// The purpose if this listener is to handle broker internal communications. - /// Unlike the client listener, the only protocol used here is SSL even when - /// `spec.clusterConfig.tls.internalSecretClass: null`. - /// The advertised listener hosts are the same as the client (broker listener host) - /// but with a different port. + /// Handles broker/broker communication (`inter.broker.listener.name`). + /// The protocol is always SSL with client certificates required, using the pod certificates + /// of `spec.clusterConfig.tls.internalSecretClass` (default `tls`), also when Kerberos is + /// enabled. + /// The advertised host is the FQDN of the broker pod in the headless service. #[strum(serialize = "INTERNAL")] Internal, - /// This is almost identical with the `Client` listener with the following exceptions: + /// Like the `Client` listener with SASL_SSL, with the following differences: /// /// - it is only defined if Kerberos is enabled /// - it uses a different port - /// - the keystore associated with the listener volume uses a different CA + /// - the Kerberos principal is derived from the bootstrap listener address instead of the + /// broker listener address /// - /// Note: the corresponding K8S service is *always* defined, not just if Kerberos is enabled - /// and it is published in the discovery ConfigMap for clients to consume. + /// It uses the same TLS keystore as the `Client` listener, whose certificate covers both the + /// broker and the bootstrap listener addresses. + /// + /// Note: the bootstrap `Listener` is created for every broker role group, not only if + /// Kerberos is enabled. Its addresses are published in the discovery ConfigMap, with the + /// bootstrap port if Kerberos is enabled and the client port otherwise. #[strum(serialize = "BOOTSTRAP")] Bootstrap, - /// This listener is defined when Kraft mode is enabled. - /// It is responsible for broker/controller as well as controller/controller communications - /// and therefore it is present on *both* brokers and controller properties files. - /// The only protocol used is SSL with the internal pod certificates, also when Kerberos is - /// enabled. - /// The advertised host names are FQDN pod names of the controllers. + /// Handles broker/controller and controller/controller communication in KRaft mode. + /// Only the controllers bind this listener; brokers connect to it as clients, so it appears + /// in the `controller.listener.names` and the security protocol map of both properties files. + /// The protocol is always SSL with client certificates required, using the pod certificates + /// of the internal `SecretClass`, also when Kerberos is enabled. + /// The controllers are addressed by their pod FQDNs (`controller.quorum.bootstrap.servers`). /// - /// Note: there is no listener for client/controller communication. + /// Note: there is no listener for client/controller communication. The only other clients + /// are the quorum management admin tools running inside the controller pods, which use the + /// same pod certificates. #[strum(serialize = "CONTROLLER")] Controller, }