refactor: KRaft controllers only use mTLS

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

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
 

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.

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

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<Option<Volume>, 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<Option<Volume>, 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<Volume, Error> {
-    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");

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,