fix: kraft kerberos support

CHANGELOG.md

@@ -4,6 +4,11 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Added
+
+- Kerberos authentication now works with KRaft controllers (`spec.controllers`), securing the
+  `CONTROLLER` listener used for broker/controller and controller/controller Raft RPC traffic ([#TBD]).
+
 ### Changed
 
 - Internal operator refactoring: introduce a build() step in the reconciler that

docs/modules/kafka/pages/usage-guide/kraft-controller.adoc

@@ -78,6 +78,17 @@ controllers:
 
 The configuration of overrides, JVM arguments etc. is similar to the Broker and documented on the xref:concepts:overrides.adoc[concepts page].
 
+== Kerberos
+
+Kerberos authentication is supported for KRaft clusters: enabling a Kerberos `AuthenticationClass`
+secures the `CLIENT`, `BOOTSTRAP` and `CONTROLLER` listeners with GSSAPI, including
+controller-to-controller and broker-to-controller Raft RPC traffic on the `CONTROLLER` listener.
+The `INTERNAL` inter-broker listener continues to use mutual TLS and is unaffected by Kerberos.
+
+NOTE: SASL/SCRAM is not supported for the controller listener by Apache Kafka itself
+(https://issues.apache.org/jira/browse/KAFKA-15513[KAFKA-15513]); this does not affect Kerberos
+(GSSAPI), which authenticates against the external KDC rather than Kafka-internal credential storage.
+
 == Internal operator details
 
 KRaft mode requires major configuration changes compared to ZooKeeper:
@@ -91,7 +102,6 @@ KRaft mode requires major configuration changes compared to ZooKeeper:
 
 * Automatic migration from Apache ZooKeeper to KRaft is not supported.
 * Scaling controller replicas might lead to unstable clusters.
-* Kerberos is currently not supported for KRaft in all versions.
 
 == Troubleshooting
 

docs/modules/kafka/partials/supported-versions.adoc

@@ -12,5 +12,5 @@ Support for clusters running in Kraft mode (which includes Apache Kafka 4.x.x) i
 Also there are some known issues such as:
 
 * Controller scaling is not reliable.
-* Kerberos authentication is not tested yet.
 * Service exposition is not definitive.
+* Kerberos authentication for KRaft is implemented, unit-tested, and has been verified end-to-end against a live cluster (Kafka 4.2.1, 3 controller replicas); it has not yet been exercised across all supported versions in CI.

rust/operator-binary/src/controller/build/command.rs

@@ -157,6 +157,7 @@ wait_for_termination()
 "#;
 
 pub fn controller_kafka_container_command(
+    kafka_security: &ValidatedKafkaSecurity,
     controller_descriptors: Vec<KafkaPodDescriptor>,
     product_version: &str,
 ) -> String {
@@ -165,23 +166,40 @@ pub fn controller_kafka_container_command(
         {remove_vector_shutdown_file_command}
         prepare_signal_handlers
         containerdebug --output={STACKABLE_LOG_DIR}/containerdebug-state.json --loop &
-
+        {set_realm_env}
         POD_INDEX=$(echo \"$POD_NAME\" | grep -oE '[0-9]+$')
         export REPLICA_ID=$((POD_INDEX+NODE_ID_OFFSET))
 
         cp {config_dir}/{properties_file} /tmp/{properties_file}
 
         config-utils template /tmp/{properties_file}
-
+        {jaas_setup}
         bin/kafka-storage.sh format --cluster-id \"$KAFKA_CLUSTER_ID\" --config /tmp/{properties_file} --ignore-formatted {initial_controller_command}
         bin/kafka-server-start.sh /tmp/{properties_file} &
 
         wait_for_termination $!
         {create_vector_shutdown_file_command}
         ",
         remove_vector_shutdown_file_command = remove_vector_shutdown_file_command(STACKABLE_LOG_DIR),
+        // When Kerberos is disabled this resolves to an empty string, so the surrounding
+        // template lines collapse to the same single blank line that was present before
+        // Kerberos support was added (byte-identical output for non-Kerberos setups).
+        set_realm_env = match kafka_security.has_kerberos_enabled() {
+            true => format!("export KERBEROS_REALM=$(grep -oP 'default_realm = \\K.*' {STACKABLE_KERBEROS_KRB5_PATH})\n"),
+            false => "".to_string(),
+        },
         config_dir = STACKABLE_CONFIG_DIR,
         properties_file = ConfigFileName::ControllerProperties,
+        // Same as `set_realm_env`: empty when Kerberos is disabled, preserving the
+        // pre-Kerberos-support blank-line layout.
+        jaas_setup = match kafka_security.has_kerberos_enabled() {
+            true => format!(
+                "\ncp {config_dir}/{jaas_file} /tmp/{jaas_file}\nconfig-utils template /tmp/{jaas_file}\n",
+                config_dir = STACKABLE_CONFIG_DIR,
+                jaas_file = ConfigFileName::Jaas,
+            ),
+            false => "".to_string(),
+        },
         initial_controller_command = initial_controllers_command(&controller_descriptors, product_version),
         create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR)
     }
@@ -207,3 +225,115 @@ fn initial_controllers_command(
         ),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::str::FromStr;
+
+    use stackable_operator::{
+        builder::meta::ObjectMetaBuilder,
+        crd::authentication::{core, kerberos},
+        v2::types::kubernetes::SecretClassName,
+    };
+
+    use super::*;
+    use crate::crd::authentication::ResolvedAuthenticationClasses;
+
+    fn kerberos_auth_class() -> core::v1alpha1::AuthenticationClass {
+        core::v1alpha1::AuthenticationClass {
+            metadata: ObjectMetaBuilder::new().name("kerberos-auth").build(),
+            spec: core::v1alpha1::AuthenticationClassSpec {
+                provider: core::v1alpha1::AuthenticationClassProvider::Kerberos(
+                    kerberos::v1alpha1::AuthenticationProvider {
+                        kerberos_secret_class: "kerberos-secret-class".to_string(),
+                    },
+                ),
+            },
+        }
+    }
+
+    /// Kerberos, which also requires server and internal TLS.
+    fn kerberos_security() -> ValidatedKafkaSecurity {
+        ValidatedKafkaSecurity::new(
+            ResolvedAuthenticationClasses::new(vec![kerberos_auth_class()]),
+            SecretClassName::from_str("tls").expect("tls secret class name is valid"),
+            Some("tls".parse().unwrap()),
+            None,
+        )
+    }
+
+    /// Plaintext: no TLS, no authentication, no OPA.
+    fn plaintext_security() -> ValidatedKafkaSecurity {
+        ValidatedKafkaSecurity::new(
+            ResolvedAuthenticationClasses::new(vec![]),
+            SecretClassName::from_str("tls").expect("tls secret class name is valid"),
+            None,
+            None,
+        )
+    }
+
+    #[test]
+    fn controller_command_exports_kerberos_realm_and_templates_jaas_when_enabled() {
+        let command = controller_kafka_container_command(&kerberos_security(), vec![], "4.1.1");
+        assert!(command.contains("export KERBEROS_REALM="));
+        assert!(command.contains(&format!(
+            "cp {}/jaas.properties /tmp/jaas.properties",
+            STACKABLE_CONFIG_DIR
+        )));
+        assert!(command.contains("config-utils template /tmp/jaas.properties"));
+    }
+
+    #[test]
+    fn controller_command_skips_kerberos_setup_when_disabled() {
+        let command = controller_kafka_container_command(&plaintext_security(), vec![], "4.1.1");
+        assert!(!command.contains("KERBEROS_REALM"));
+        assert!(!command.contains("jaas.properties"));
+    }
+
+    /// Mirrors `controller_kafka_container_command` as it existed at commit `d9942ad`
+    /// (immediately before Kerberos support was added), before it took a `kafka_security`
+    /// parameter. Used to pin down that Kerberos-disabled output is byte-identical to the
+    /// pre-Kerberos-support output, per the plan's Global Constraint.
+    fn pre_kerberos_controller_kafka_container_command(
+        controller_descriptors: Vec<KafkaPodDescriptor>,
+        product_version: &str,
+    ) -> String {
+        formatdoc! {"
+            {BASH_TRAP_FUNCTIONS}
+            {remove_vector_shutdown_file_command}
+            prepare_signal_handlers
+            containerdebug --output={STACKABLE_LOG_DIR}/containerdebug-state.json --loop &
+
+            POD_INDEX=$(echo \"$POD_NAME\" | grep -oE '[0-9]+$')
+            export REPLICA_ID=$((POD_INDEX+NODE_ID_OFFSET))
+
+            cp {config_dir}/{properties_file} /tmp/{properties_file}
+
+            config-utils template /tmp/{properties_file}
+
+            bin/kafka-storage.sh format --cluster-id \"$KAFKA_CLUSTER_ID\" --config /tmp/{properties_file} --ignore-formatted {initial_controller_command}
+            bin/kafka-server-start.sh /tmp/{properties_file} &
+
+            wait_for_termination $!
+            {create_vector_shutdown_file_command}
+            ",
+            remove_vector_shutdown_file_command = remove_vector_shutdown_file_command(STACKABLE_LOG_DIR),
+            config_dir = STACKABLE_CONFIG_DIR,
+            properties_file = ConfigFileName::ControllerProperties,
+            initial_controller_command = initial_controllers_command(&controller_descriptors, product_version),
+            create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR)
+        }
+    }
+
+    #[test]
+    fn controller_command_is_byte_identical_to_pre_kerberos_output_when_disabled() {
+        let actual = controller_kafka_container_command(&plaintext_security(), vec![], "4.1.1");
+        let expected = pre_kerberos_controller_kafka_container_command(vec![], "4.1.1");
+
+        assert_eq!(
+            actual, expected,
+            "controller_kafka_container_command must produce byte-identical output to the \
+             pre-Kerberos-support implementation when Kerberos is disabled"
+        );
+    }
+}

rust/operator-binary/src/controller/build/kerberos.rs

@@ -41,30 +41,49 @@ pub enum Error {
 pub fn add_kerberos_pod_config(
     kafka_security: &ValidatedKafkaSecurity,
     role: &KafkaRole,
-    cb_kcat_prober: &mut ContainerBuilder,
+    cb_kcat_prober: Option<&mut ContainerBuilder>,
     cb_kafka: &mut ContainerBuilder,
     pb: &mut PodBuilder,
 ) -> Result<(), Error> {
     if let Some(kerberos_secret_class) = kafka_security.kerberos_secret_class() {
-        // Mount keytab
-        let kerberos_secret_operator_volume = SecretOperatorVolumeSourceBuilder::new(
+        let mut volume_builder = SecretOperatorVolumeSourceBuilder::new(
             kerberos_secret_class,
             // We need both public (krb5.conf) and private (keytab) parts.
             SecretClassVolumeProvisionParts::PublicPrivate,
-        )
-        .with_listener_volume_scope(LISTENER_BROKER_VOLUME_NAME)
-        .with_listener_volume_scope(LISTENER_BOOTSTRAP_VOLUME_NAME)
-        .with_kerberos_service_name(role.kerberos_service_name())
-        .build()
-        .context(KerberosSecretVolumeSnafu)?;
+        );
+        match role {
+            // Brokers are exposed through listener-operator `Listener` volumes (the client
+            // and bootstrap listeners); the keytab principal must cover both.
+            KafkaRole::Broker => {
+                volume_builder
+                    .with_listener_volume_scope(LISTENER_BROKER_VOLUME_NAME)
+                    .with_listener_volume_scope(LISTENER_BOOTSTRAP_VOLUME_NAME);
+            }
+            // KRaft controllers have no listener-operator `Listener` volume (see
+            // `controller/build/mod.rs`, "Only broker role groups get a bootstrap Listener"):
+            // they're only reachable through their own StatefulSet pod DNS name, so the keytab
+            // must be pod-scoped, matching how the controller's internal TLS cert is provisioned
+            // in `add_controller_volume_and_volume_mounts`.
+            KafkaRole::Controller => {
+                volume_builder.with_pod_scope();
+            }
+        };
+        let kerberos_secret_operator_volume = volume_builder
+            .with_kerberos_service_name(role.kerberos_service_name())
+            .build()
+            .context(KerberosSecretVolumeSnafu)?;
         pb.add_volume(
             VolumeBuilder::new("kerberos")
                 .ephemeral(kerberos_secret_operator_volume)
                 .build(),
         )
         .context(AddVolumeSnafu)?;
 
-        for cb in [cb_kafka, cb_kcat_prober] {
+        let mut containers: Vec<&mut ContainerBuilder> = vec![cb_kafka];
+        if let Some(cb_kcat_prober) = cb_kcat_prober {
+            containers.push(cb_kcat_prober);
+        }
+        for cb in containers {
             cb.add_volume_mount("kerberos", STACKABLE_KERBEROS_DIR)
                 .context(AddVolumeMountSnafu)?;
             cb.add_env_var("KRB5_CONFIG", STACKABLE_KERBEROS_KRB5_PATH);
@@ -77,3 +96,85 @@ pub fn add_kerberos_pod_config(
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use stackable_operator::{
+        builder::{meta::ObjectMetaBuilder, pod::container::ContainerBuilder},
+        crd::authentication::{core, kerberos},
+    };
+
+    use super::*;
+    use crate::crd::authentication::ResolvedAuthenticationClasses;
+
+    fn kerberos_security() -> ValidatedKafkaSecurity {
+        ValidatedKafkaSecurity::new(
+            ResolvedAuthenticationClasses::new(vec![core::v1alpha1::AuthenticationClass {
+                metadata: ObjectMetaBuilder::new().name("kerberos-auth").build(),
+                spec: core::v1alpha1::AuthenticationClassSpec {
+                    provider: core::v1alpha1::AuthenticationClassProvider::Kerberos(
+                        kerberos::v1alpha1::AuthenticationProvider {
+                            kerberos_secret_class: "kerberos-secret-class".to_string(),
+                        },
+                    ),
+                },
+            }]),
+            "tls".parse().unwrap(),
+            Some("tls".parse().unwrap()),
+            None,
+        )
+    }
+
+    #[test]
+    fn controller_role_mounts_pod_scoped_keytab_without_kcat_container() {
+        let mut pb = PodBuilder::new();
+        let mut cb_kafka = ContainerBuilder::new("kafka").expect("valid container name");
+
+        add_kerberos_pod_config(
+            &kerberos_security(),
+            &KafkaRole::Controller,
+            None,
+            &mut cb_kafka,
+            &mut pb,
+        )
+        .expect("kerberos pod config for controller role");
+
+        let pod = pb.build_template();
+        let kerberos_volume = pod
+            .spec
+            .as_ref()
+            .and_then(|spec| spec.volumes.as_ref())
+            .and_then(|volumes| volumes.iter().find(|v| v.name == "kerberos"))
+            .expect("kerberos volume must be present");
+        let ephemeral = kerberos_volume
+            .ephemeral
+            .as_ref()
+            .expect("kerberos volume must be an ephemeral (secret-operator) volume");
+        let annotations = ephemeral
+            .volume_claim_template
+            .as_ref()
+            .and_then(|t| t.metadata.as_ref())
+            .and_then(|m| m.annotations.as_ref())
+            .expect("volume claim template must carry secrets.stackable.tech annotations");
+        // Pod-scoping (`with_pod_scope()`) is expressed as a `secrets.stackable.tech/scope: pod`
+        // annotation (same as the controller's internal TLS cert, see
+        // `add_controller_volume_and_volume_mounts`) -- it must not mention a listener volume.
+        assert_eq!(
+            annotations
+                .get("secrets.stackable.tech/scope")
+                .map(String::as_str),
+            Some("pod"),
+            "controller keytab must be pod-scoped only, not listener-volume-scoped: {annotations:?}"
+        );
+
+        let kafka_container = cb_kafka.build();
+        let env_names: Vec<_> = kafka_container
+            .env
+            .unwrap_or_default()
+            .into_iter()
+            .map(|e| e.name)
+            .collect();
+        assert!(env_names.contains(&"KRB5_CONFIG".to_string()));
+        assert!(env_names.contains(&"KAFKA_OPTS".to_string()));
+    }
+}

rust/operator-binary/src/controller/build/properties/listener.rs

@@ -108,8 +108,14 @@ 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, KafkaListenerProtocol::Ssl);
+    listener_security_protocol_map.insert(
+        KafkaListenerName::Controller,
+        if kafka_security.has_kerberos_enabled() {
+            KafkaListenerProtocol::SaslSsl
+        } else {
+            KafkaListenerProtocol::Ssl
+        },
+    );
 
     // BOOTSTRAP
     if kafka_security.has_kerberos_enabled() {
@@ -492,7 +498,7 @@ mod tests {
                 bootstrap_name = KafkaListenerName::Bootstrap,
                 bootstrap_protocol = KafkaListenerProtocol::SaslSsl,
                 controller_name = KafkaListenerName::Controller,
-                controller_protocol = KafkaListenerProtocol::Ssl,
+                controller_protocol = KafkaListenerProtocol::SaslSsl,
             )
         );
     }