diff --git a/CHANGELOG.md b/CHANGELOG.md index 4f62ca0d..0e2e7bc8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,8 @@ 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]). ### Changed @@ -21,9 +23,6 @@ All notable changes to this project will be documented in this file. that adds the new controller to the voter list. On termination, a new `preStop` hook on the controller container (`kafka`) removes the pod from the voter list before shutdown. - The property `controller.quorum.bootstrap.servers` now contains the headless service names - of all controller role groups instead of individual peer host names. This prevents the - restart controller from restarting all pods in the quorum when a new one is added/deleted. The controller `StatefulSet` is now scaled using `OrderedReady` instead of the `Parallel` strategy to ensure only one voter is added/removed at a time and thus keep the quorum healthy ([#1010]). - Internal operator refactoring: introduce a build() step in the reconciler that @@ -76,6 +75,7 @@ All notable changes to this project will be documented in this file. [#1014]: https://github.com/stackabletech/kafka-operator/pull/1014 [#1017]: https://github.com/stackabletech/kafka-operator/pull/1017 [#1021]: https://github.com/stackabletech/kafka-operator/pull/1021 +[#1024]: https://github.com/stackabletech/kafka-operator/pull/1024 ## [26.7.0] - 2026-07-21 diff --git a/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc b/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc index 76668635..ce9f0f16 100644 --- a/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc +++ b/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc @@ -92,13 +92,34 @@ KRaft mode requires major configuration changes compared to ZooKeeper: with `kafka-storage.sh format --standalone`, bootstrapping a single-node quorum by itself. 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. +* `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 -- but it means the property changes whenever a + controller role group is scaled, so the controller pods are rolled by the restart controller on scale. + +== Kerberos + +Apache Kafka KRaft controllers support Kerberos (GSSAPI) authentication. +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. == Known Issues * Automatic migration from Apache ZooKeeper to KRaft is not supported. -* Kerberos is currently not supported for KRaft in all versions. * The quorum is created once by the controller with the lowest `node.id` using `--standalone`. If this controller loses it's PVC, a new conflicting quorum is created on restart. * A Controller that loses its persistent volume is not re-admitted to the voter set automatically, because it diff --git a/docs/modules/kafka/pages/usage-guide/security.adoc b/docs/modules/kafka/pages/usage-guide/security.adoc index 598bc96c..5e5942c7 100644 --- a/docs/modules/kafka/pages/usage-guide/security.adoc +++ b/docs/modules/kafka/pages/usage-guide/security.adoc @@ -162,13 +162,55 @@ NOTE: When Kerberos is enabled it is also required to enable TLS for maximum sec ==== Clients -In order to keep client configuration as uncluttered as possible, each kerberized Kafka broker has two principals: one for the broker itself and one for the bootstrap service. -The client can connect to the bootstrap service, which returns the broker quorum for use in subsequent operations. -This is transparent as each connection dynamically uses the relevant principal (broker or bootstrap). -In order for this to work, it is necessary for kerberized clusters to define an extra Kafka listener for the bootstrap with a corresponding service (and port). -The bootstrap address is written to the discovery ConfigMap, using the Stackable bootstrap listener with the port being 9095 (secure) for kerberized clusters, and 9092 (non-secure) or 9093 (secure) for non-kerberized ones. +===== Why each broker has two principals -NOTE: Port 9094 is reserved for non-secure kerberized connections which is not currently implemented. +A GSSAPI client derives the service principal it asks the KDC for from the *hostname it connects to*. +Connecting to a Kafka cluster takes two hops over two different addresses: a client first contacts a bootstrap address, receives the cluster metadata, and then connects to individual brokers directly. +Each of those addresses therefore needs its own principal. + +Every broker consequently gets two principals, `kafka/` and `kafka/`, both provisioned into its keytab by the Secret Operator. +The broker's JAAS configuration declares both, as the `bootstrap.KafkaServer` and `client.KafkaServer` login contexts. +No client-side configuration is needed to switch between them: whichever address a client dials, the broker already holds the matching principal. + +To make this work, a kerberized cluster exposes an additional Kafka listener, container port and `Listener` port for the bootstrap address. +These exist only when Kerberos is enabled; without it, clients reach brokers over the client listener alone. + +===== Bootstrap address and ports + +The bootstrap address is published in the xref:reference/discovery.adoc[discovery ConfigMap] under the `KAFKA` key, read from the ingress addresses of the Stackable bootstrap `Listener`. +The port depends on whether Kerberos and TLS are enabled: + +[cols="1,1,1"] +|=== +| Cluster | Port name | Port + +| Kerberos (always TLS) +| `bootstrap` +| 9095 + +| TLS, no Kerberos +| `kafka-tls` +| 9093 + +| No TLS, no Kerberos +| `kafka` +| 9092 +|=== + +NOTE: Kerberos requires TLS, so a kerberized cluster always uses port 9095 for bootstrapping. + +===== Client configuration + +The discovery ConfigMap's `client.properties` carries the properties needed to *reach* the cluster: + +- `security.protocol` +- `sasl.mechanism` +- `sasl.kerberos.service.name` +- and the truststore settings + +But the operator adds *no login configuration* to 'client.properties` because supplying credentials is the client's responsibility. +The operator cannot do it: these clients run outside the Kafka Pods, so they have neither the Pods' keytabs nor their principals. +Add your own `sasl.jaas.config`, or point the JVM at a JAAS file with `java.security.auth.login.config`, naming the principal and keytab the client should authenticate with. == [[authorization]]Authorization diff --git a/docs/modules/kafka/partials/supported-versions.adoc b/docs/modules/kafka/partials/supported-versions.adoc index 5a962cb4..0540998c 100644 --- a/docs/modules/kafka/partials/supported-versions.adoc +++ b/docs/modules/kafka/partials/supported-versions.adoc @@ -2,12 +2,11 @@ // This is a separate file, since it is used by both the direct Kafka documentation, and the overarching // Stackable Platform documentation. -* 4.2.1 (experimental, deprecated) - Requires KRaft, please read on the xref:kafka:usage-guide/kraft-controller.adoc[Kraft migration guide]. -* 4.1.1 (experimental, deprecated) - Requires KRaft, please read on the xref:kafka:usage-guide/kraft-controller.adoc[Kraft migration guide]. +* 4.2.1 +* 4.1.1 * 3.9.2 (LTS) * 3.9.1 (deprecated) -Support for clusters running in Kraft mode (which includes Apache Kafka >= 4.x) is experimental due to the following known issues: +Starting with Apache Kafka >=4.x, the quorum manager has been changed from Apache Zookeeper to the built in KRaft. -* Kerberos authentication is not tested yet. -* Service exposition is not definitive. +If you are running an older Kafka version and plan to upgrade, head on to the xref:kafka:usage-guide/kraft-controller.adoc[Kraft migration guide] for details on how to do so. diff --git a/rust/operator-binary/src/controller/build/command.rs b/rust/operator-binary/src/controller/build/command.rs index dbf609b9..6b8c0d73 100644 --- a/rust/operator-binary/src/controller/build/command.rs +++ b/rust/operator-binary/src/controller/build/command.rs @@ -142,6 +142,7 @@ fn controller_quorum_format_flag(controller_descriptors: &[KafkaPodDescriptor]) } pub fn controller_kafka_container_command( + kafka_security: &ValidatedKafkaSecurity, controller_descriptors: Vec, ) -> String { formatdoc! {" @@ -149,6 +150,7 @@ 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} {derive_pod_index} {export_replica_id} @@ -157,6 +159,12 @@ 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} + {quorum_format_flag} bin/kafka-storage.sh format --cluster-id \"$KAFKA_CLUSTER_ID\" --config /tmp/{properties_file} --ignore-formatted \"$FORMAT_QUORUM_FLAG\" bin/kafka-server-start.sh /tmp/{properties_file} & @@ -165,10 +173,18 @@ pub fn controller_kafka_container_command( {create_vector_shutdown_file_command} ", remove_vector_shutdown_file_command = remove_vector_shutdown_file_command(STACKABLE_LOG_DIR), + // Mirrors `broker_kafka_container_commands`: empty when Kerberos is disabled. + set_realm_env = match kafka_security.has_kerberos_enabled() { + true => format!("export KERBEROS_REALM=$(grep -oP 'default_realm = \\K.*' {STACKABLE_KERBEROS_KRB5_PATH})"), + false => "".to_string(), + }, 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), create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR) } @@ -176,7 +192,11 @@ pub fn controller_kafka_container_command( const KAFKA_METADATA_QUORUM_BINARY: &str = "/stackable/kafka/bin/kafka-metadata-quorum.sh"; -const ADMIN_CLIENT_PROPERTIES_PATH: &str = "/stackable/config/admin-client.properties"; +/// 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`). +const ADMIN_CLIENT_PROPERTIES_PATH: &str = "/tmp/admin-client.properties"; +const ADMIN_CLIENT_PROPERTIES_SOURCE_PATH: &str = "/stackable/config/admin-client.properties"; /// The merged config used only for `add-controller` (self-registration). /// @@ -200,11 +220,6 @@ const CLI_CALL_KILL_AFTER_SECONDS: u32 = 5; /// Shell snippet setting `$BOOTSTRAP_SERVERS` by extracting /// `controller.quorum.bootstrap.servers` from the static, un-rendered `controller.properties` /// ConfigMap file. -/// -/// Reading this at runtime, rather than baking the peer list into this script as a Rust -/// literal, keeps both sidecar scripts' content — and therefore the controller pod -/// template — identical across changes to an existing controller role group's *replica -/// count*. fn extract_bootstrap_servers_command() -> String { format!( r#"BOOTSTRAP_SERVERS=$(grep '^controller.quorum.bootstrap.servers=' {config_dir}/{controller_properties_file} | cut -d= -f2- | sed 's/\\:/:/g')"#, @@ -230,6 +245,12 @@ 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() -> String { + // The sidecar is a separate container and inherits nothing from the kafka container's + // startup, so it derives the realm itself. Harmless when krb5.conf is absent: only the + // Kerberos case has a `${env:KERBEROS_REALM}` placeholder for `config-utils` to resolve. + let set_realm_env = format!( + "KERBEROS_REALM=$(grep -oP 'default_realm = \\K.*' {STACKABLE_KERBEROS_KRB5_PATH} 2>/dev/null) && export KERBEROS_REALM || true" + ); format!( r#" set -uo pipefail @@ -238,9 +259,12 @@ pub fn quorum_manager_container_command() -> String { [ -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} \ + && cp {admin_client_source} {admin_client_config} \ + && config-utils template {admin_client_config} \ && cat /tmp/{controller_properties_file} {admin_client_config} > {add_controller_config}; then QUORUM_CLI={binary} ADMIN_CLIENT_CONFIG={admin_client_config} @@ -266,8 +290,10 @@ pub fn quorum_manager_container_command() -> String { derive_pod_index = DERIVE_POD_INDEX, export_replica_id = EXPORT_REPLICA_ID, extract_bootstrap_servers = extract_bootstrap_servers_command(), + set_realm_env = set_realm_env, config_dir = STACKABLE_CONFIG_DIR, controller_properties_file = ConfigFileName::ControllerProperties, + admin_client_source = ADMIN_CLIENT_PROPERTIES_SOURCE_PATH, admin_client_config = ADMIN_CLIENT_PROPERTIES_PATH, add_controller_config = ADD_CONTROLLER_PROPERTIES_PATH, cli_timeout = CLI_CALL_TIMEOUT_SECONDS, @@ -378,6 +404,65 @@ mod tests { use indoc::indoc; use super::*; + use crate::controller::build::security::tests::{kerberos, plaintext}; + + #[test] + fn controller_command_exports_the_kerberos_realm_when_enabled() { + let command = controller_kafka_container_command(&kerberos(), vec![]); + assert!(command.contains("export KERBEROS_REALM=$(grep -oP 'default_realm = \\K.*'")); + } + + #[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`. + 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")); + } + } + + #[test] + fn quorum_manager_templates_the_admin_client_config() { + 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. + 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(); + 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![]); + assert!( + command.contains( + "cp /stackable/config/admin-client.properties /tmp/admin-client.properties" + ) + ); + assert!(command.contains("config-utils template /tmp/admin-client.properties")); + } #[test] fn quorum_manager_container_command_targets_the_bootstrap_servers_not_localhost() { @@ -446,11 +531,11 @@ mod tests { // `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 plain admin-client config (carries `security.protocol`/`ssl.*`), - // controller.properties first so the client TLS config in admin-client.properties - // wins on any key collision (see `ADD_CONTROLLER_PROPERTIES_PATH`'s doc comment). + // Merges it with the *rendered* admin-client config (carries `security.protocol`, + // `ssl.*` and, under Kerberos, `sasl.jaas.config`), controller.properties first so + // the client config wins on any key collision (see `ADD_CONTROLLER_PROPERTIES_PATH`). assert!(command.contains( - "cat /tmp/controller.properties /stackable/config/admin-client.properties > /tmp/add-controller.properties" + "cat /tmp/controller.properties /tmp/admin-client.properties > /tmp/add-controller.properties" )); // The merged file is what `add-controller` — and only `add-controller` — connects // with; read-only `describe` calls keep using the plain admin-client config. @@ -1119,7 +1204,7 @@ mod tests { pod_descriptor(KafkaRole::Controller, 1, 6), pod_descriptor(KafkaRole::Controller, 2, 7), ]; - let command = controller_kafka_container_command(descriptors); + let command = controller_kafka_container_command(&plaintext(), descriptors); assert!(command.contains(r#"if [ "$REPLICA_ID" = "5" ]; then"#)); assert!(command.contains("FORMAT_QUORUM_FLAG=--standalone")); diff --git a/rust/operator-binary/src/controller/build/kerberos.rs b/rust/operator-binary/src/controller/build/kerberos.rs index cc2f1d91..0335a4fd 100644 --- a/rust/operator-binary/src/controller/build/kerberos.rs +++ b/rust/operator-binary/src/controller/build/kerberos.rs @@ -29,7 +29,7 @@ use crate::{ }, }; -constant!(KERBEROS_VOLUME_NAME: VolumeName = "kerberos"); +constant!(pub KERBEROS_VOLUME_NAME: VolumeName = "kerberos"); #[derive(Snafu, Debug)] pub enum Error { @@ -57,16 +57,31 @@ pub fn add_kerberos_pod_config( ) -> 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 broker + // and bootstrap listeners), so 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: 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 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_VOLUME_NAME) .ephemeral(kerberos_secret_operator_volume) @@ -82,8 +97,8 @@ pub fn add_kerberos_pod_config( Ok(()) } -constant!(KRB5_CONFIG: EnvVarName = "KRB5_CONFIG"); -constant!(KAFKA_OPTS: EnvVarName = "KAFKA_OPTS"); +constant!(pub KRB5_CONFIG: EnvVarName = "KRB5_CONFIG"); +constant!(pub KAFKA_OPTS: EnvVarName = "KAFKA_OPTS"); /// The environment variables the Kerberos configuration requires on the Kafka container, or an /// empty set when Kerberos is disabled. 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 4044e4f6..3849d253 100644 --- a/rust/operator-binary/src/controller/build/properties/controller_properties.rs +++ b/rust/operator-binary/src/controller/build/properties/controller_properties.rs @@ -10,7 +10,7 @@ use crate::{ }, }, crd::{ - KafkaPodDescriptor, + CONTROLLER_POD_FQDN_TEMPLATE, KafkaPodDescriptor, listener::{KafkaListenerConfig, KafkaListenerName}, role::{ KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS, KAFKA_LISTENER_SECURITY_PROTOCOL_MAP, @@ -32,27 +32,27 @@ pub fn build( KAFKA_LOG_DIRS.to_string(), "/stackable/data/kraft".to_string(), ), - (KAFKA_PROCESS_ROLES.to_string(), KafkaRole::Controller.to_string()), ( - "controller.listener.names".to_string(), - KafkaListenerName::Controller.to_string(), + KAFKA_PROCESS_ROLES.to_string(), + KafkaRole::Controller.to_string(), ), ( - KAFKA_NODE_ID.to_string(), - "${env:REPLICA_ID}".to_string(), + "controller.listener.names".to_string(), + KafkaListenerName::Controller.to_string(), ), + (KAFKA_NODE_ID.to_string(), "${env:REPLICA_ID}".to_string()), ( KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS.to_string(), kraft_controllers.clone(), ), ( KAFKA_LISTENERS.to_string(), - "CONTROLLER://${env:POD_NAME}.${env:ROLEGROUP_HEADLESS_SERVICE_NAME}.${env:NAMESPACE}.svc.${env:CLUSTER_DOMAIN}:${env:KAFKA_CLIENT_PORT}".to_string(), + format!("CONTROLLER://{CONTROLLER_POD_FQDN_TEMPLATE}:${{env:KAFKA_CLIENT_PORT}}"), ), ( KAFKA_LISTENER_SECURITY_PROTOCOL_MAP.to_string(), - listener_config - .listener_security_protocol_map_for_controller()), + listener_config.listener_security_protocol_map_for_controller(), + ), ]); result.insert( diff --git a/rust/operator-binary/src/controller/build/properties/listener.rs b/rust/operator-binary/src/controller/build/properties/listener.rs index 3ba733bf..acc63255 100644 --- a/rust/operator-binary/src/controller/build/properties/listener.rs +++ b/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, ) ); } diff --git a/rust/operator-binary/src/controller/build/properties/mod.rs b/rust/operator-binary/src/controller/build/properties/mod.rs index d607d5f3..0eed9d45 100644 --- a/rust/operator-binary/src/controller/build/properties/mod.rs +++ b/rust/operator-binary/src/controller/build/properties/mod.rs @@ -63,18 +63,28 @@ pub fn uses_legacy_log4j(product_version: &str) -> bool { product_version.starts_with("3.") } -/// `controller.quorum.bootstrap.servers` addresses, one per distinct controller role group, -/// pointing at each role group's own headless Service DNS name rather than individual pod -/// FQDNs. +/// `controller.quorum.bootstrap.servers` addresses: one individual pod FQDN per controller, +/// across all controller role groups. /// -/// Only adding or removing a whole role group changes this list. +/// # Why pod FQDNs rather than the role group's headless Service +/// +/// Kerberos forces this. A GSSAPI client derives the service principal from the hostname it +/// dials, so dialling the headless Service asks for `kafka/`, while the CONTROLLER +/// listener's acceptor can offer only a single principal -- the pod's own +/// `kafka/`, which is also what the Raft voter endpoints advertise. Bootstrapping +/// through the Service therefore fails authentication for every peer. +/// +/// The side-effect of using pod FQDNs instead of service names is that this list changes on +/// on every scaling operation (replica count change), so scaling one rolls *all* controller pods. pub(crate) fn kraft_controllers(pod_descriptors: &[KafkaPodDescriptor]) -> Vec { pod_descriptors .iter() .filter(|pd| pd.role == KafkaRole::Controller) .map(|desc| { format!( - "{service}.{namespace}.svc.{cluster_domain}:{client_port}", + "{sts}-{replica}.{service}.{namespace}.svc.{cluster_domain}:{client_port}", + sts = desc.role_group_statefulset_name, + replica = desc.replica, service = desc.role_group_service_name, namespace = desc.namespace, cluster_domain = desc.cluster_domain, @@ -132,7 +142,7 @@ mod tests { } #[test] - fn kraft_controllers_points_at_the_role_group_headless_service_not_individual_pods() { + fn kraft_controllers_lists_individual_pod_fqdns() { let pod_descriptors = vec![ pod_descriptor(KafkaRole::Controller, 0, 9093), pod_descriptor(KafkaRole::Controller, 1, 9093), @@ -143,14 +153,20 @@ mod tests { let quorum_bootstrap_servers = kraft_controllers(&pod_descriptors).join(","); + // Individual pod FQDNs, *not* the role group's headless Service. Under Kerberos the + // GSSAPI service principal is derived from the hostname the peer dials, and the + // CONTROLLER listener's acceptor can only offer one SPN -- the pod's own. Dialling + // the headless Service asks for `kafka/` instead and is rejected. assert_eq!( quorum_bootstrap_servers, - "kafka-controller-default-headless.default.svc.cluster.local:9093" + "kafka-controller-default-0.kafka-controller-default-headless.default.svc.cluster.local:9093,\ + kafka-controller-default-1.kafka-controller-default-headless.default.svc.cluster.local:9093,\ + kafka-controller-default-2.kafka-controller-default-headless.default.svc.cluster.local:9093" ); } #[test] - fn kraft_controllers_is_stable_across_replica_count_changes() { + fn kraft_controllers_grows_with_the_replica_count() { let three_replicas = vec![ pod_descriptor(KafkaRole::Controller, 0, 9093), pod_descriptor(KafkaRole::Controller, 1, 9093), @@ -164,15 +180,19 @@ mod tests { pod_descriptor(KafkaRole::Controller, 4, 9093), ]; - assert_eq!( + // Deliberate consequence of per-pod addressing: unlike the previous headless-Service + // form, this list changes with the replica count, so scaling a controller role group + // rolls the controller pods. + assert_eq!(kraft_controllers(&three_replicas).len(), 3); + assert_eq!(kraft_controllers(&five_replicas).len(), 5); + assert_ne!( kraft_controllers(&three_replicas), kraft_controllers(&five_replicas) ); } #[test] - fn kraft_controllers_lists_every_distinct_role_groups_service_once() { - let mut default_group_pod = pod_descriptor(KafkaRole::Controller, 0, 9093); + fn kraft_controllers_lists_pods_from_every_role_group() { let mut other_group_pod = pod_descriptor(KafkaRole::Controller, 0, 9093); other_group_pod.role_group_statefulset_name = "kafka-controller-other" .parse() @@ -180,18 +200,10 @@ mod tests { other_group_pod.role_group_service_name = "kafka-controller-other-headless" .parse() .expect("valid service name"); - // Second replica of the *same* role group as `default_group_pod` - must not produce - // a second entry for that Service. - let default_group_pod_replica_1 = { - let mut pod = pod_descriptor(KafkaRole::Controller, 1, 9093); - pod.node_id = 1; - pod - }; - default_group_pod.node_id = 0; let pod_descriptors = vec![ - default_group_pod, - default_group_pod_replica_1, + pod_descriptor(KafkaRole::Controller, 0, 9093), + pod_descriptor(KafkaRole::Controller, 1, 9093), other_group_pod, ]; @@ -200,8 +212,9 @@ mod tests { assert_eq!( quorum_bootstrap_servers, vec![ - "kafka-controller-default-headless.default.svc.cluster.local:9093".to_string(), - "kafka-controller-other-headless.default.svc.cluster.local:9093".to_string(), + "kafka-controller-default-0.kafka-controller-default-headless.default.svc.cluster.local:9093".to_string(), + "kafka-controller-default-1.kafka-controller-default-headless.default.svc.cluster.local:9093".to_string(), + "kafka-controller-other-0.kafka-controller-other-headless.default.svc.cluster.local:9093".to_string(), ] ); } diff --git a/rust/operator-binary/src/controller/build/resource/config_map.rs b/rust/operator-binary/src/controller/build/resource/config_map.rs index 49c4e261..9d21f7e3 100644 --- a/rust/operator-binary/src/controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/controller/build/resource/config_map.rs @@ -22,9 +22,10 @@ use crate::{ }, }, crd::{ - STACKABLE_LISTENER_BOOTSTRAP_DIR, STACKABLE_LISTENER_BROKER_DIR, + CONTROLLER_POD_FQDN_TEMPLATE, STACKABLE_LISTENER_BOOTSTRAP_DIR, + STACKABLE_LISTENER_BROKER_DIR, listener::{KafkaListenerConfig, node_address_cmd}, - role::AnyConfig, + role::{AnyConfig, KafkaRole}, }, }; @@ -177,7 +178,7 @@ pub fn build_rolegroup_config_map( // and this tool currently doesn't support the JAAS login configuration format. .add_data( ConfigFileName::Jaas.to_string(), - jaas_config_file(kafka_security.has_kerberos_enabled()), + jaas_config_file(kafka_security.has_kerberos_enabled(), &role), ); // `admin-client.properties` is only needed by the controller-side sidecar running @@ -223,10 +224,40 @@ pub fn build_rolegroup_config_map( // 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 { - match is_kerberos_enabled { - false => String::new(), - true => formatdoc! {" +fn jaas_config_file(is_kerberos_enabled: bool, role: &KafkaRole) -> String { + if !is_kerberos_enabled { + return String::new(); + } + + // Broker pods reach the CONTROLLER listener as SASL clients; the only principals in + // their keytab (see `add_kerberos_pod_config`) are for the broker and bootstrap listener + // addresses, so their CONTROLLER section must reuse the broker address. + // 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`. + let controller_principal_address = match role { + KafkaRole::Broker => node_address_cmd(STACKABLE_LISTENER_BROKER_DIR), + KafkaRole::Controller => CONTROLLER_POD_FQDN_TEMPLATE.to_string(), + }; + + // Unlike the bootstrap and client sections below, 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`). + let controller_section = formatdoc! {" + controller.KafkaServer {{ + com.sun.security.auth.module.Krb5LoginModule required + useKeyTab=true + storeKey=true + keyTab=\"/stackable/kerberos/keytab\" + principal=\"kafka/{controller_principal_address}@${{env:KERBEROS_REALM}}\"; + }}; + "}; + + match role { + KafkaRole::Controller => controller_section, + KafkaRole::Broker => formatdoc! {" bootstrap.KafkaServer {{ com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true @@ -245,6 +276,7 @@ fn jaas_config_file(is_kerberos_enabled: bool) -> String { 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), @@ -255,15 +287,17 @@ fn jaas_config_file(is_kerberos_enabled: bool) -> String { #[cfg(test)] mod tests { use super::jaas_config_file; + use crate::crd::role::KafkaRole; #[test] fn jaas_config_file_empty_without_kerberos() { - assert_eq!(jaas_config_file(false), ""); + assert_eq!(jaas_config_file(false, &KafkaRole::Broker), ""); + assert_eq!(jaas_config_file(false, &KafkaRole::Controller), ""); } #[test] fn jaas_config_file_renders_bootstrap_and_client_sections_with_kerberos() { - let jaas = jaas_config_file(true); + let jaas = jaas_config_file(true, &KafkaRole::Broker); assert!(jaas.contains("bootstrap.KafkaServer")); assert!(jaas.contains("client.KafkaServer")); assert!(jaas.contains("Krb5LoginModule")); diff --git a/rust/operator-binary/src/controller/build/resource/probes.rs b/rust/operator-binary/src/controller/build/resource/probes.rs index 66426b60..468336ed 100644 --- a/rust/operator-binary/src/controller/build/resource/probes.rs +++ b/rust/operator-binary/src/controller/build/resource/probes.rs @@ -11,8 +11,11 @@ use stackable_operator::{ v2::types::common::Port, }; -use crate::controller::{ - build::security::kcat_prober_container_commands, security::ValidatedKafkaSecurity, +use crate::{ + controller::{ + build::security::kcat_prober_container_commands, security::ValidatedKafkaSecurity, + }, + crd::CONTROLLER_POD_FQDN_SHELL, }; #[derive(Snafu, Debug)] @@ -142,7 +145,7 @@ pub fn controller_stuck_unattached_liveness_probe( "bash".to_string(), "-c".to_string(), format!( - "timeout 2 bash -c 'cat < /dev/null > /dev/tcp/localhost/{client_port}' || exit 1\n\ + "timeout 2 bash -c 'cat < /dev/null > /dev/tcp/{CONTROLLER_POD_FQDN_SHELL}/{client_port}' || exit 1\n\ state=$(curl -s --max-time 2 localhost:{metrics_port}/metrics | grep -oE 'kafka_server_raft_metrics_current_state\\{{state=\"[a-z]+\",?\\}}' | grep -oE '\"[a-z]+\"' | tr -d '\"')\n\ [ \"$state\" != \"unattached\" ]" ), diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 401cffdd..d8c3ba42 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -50,7 +50,10 @@ use crate::{ kafka_log_opts, quorum_manager_container_command, }, graceful_shutdown::add_graceful_shutdown_config, - kerberos::{add_kerberos_pod_config, kerberos_env_vars}, + kerberos::{ + KAFKA_OPTS, KERBEROS_VOLUME_NAME, KRB5_CONFIG, add_kerberos_pod_config, + kerberos_env_vars, + }, properties::product_logging::MAX_KAFKA_LOG_FILES_SIZE, recommended_labels_for_role_group_resources, recommended_labels_for_unversioned_role_group_resources, role_group_selector, @@ -67,8 +70,9 @@ use crate::{ BROKER_ID_POD_MAP_DIR, BROKER_ID_POD_MAP_DIR_NAME, KAFKA_HEAP_OPTS, LISTENER_BOOTSTRAP_VOLUME_NAME, LISTENER_BROKER_VOLUME_NAME, LOG_DIRS_VOLUME_NAME, METRICS_PORT, METRICS_PORT_NAME, STACKABLE_CONFIG_DIR, STACKABLE_CONFIG_DIR_NAME, - STACKABLE_DATA_DIR, STACKABLE_LISTENER_BOOTSTRAP_DIR, STACKABLE_LISTENER_BROKER_DIR, - STACKABLE_LOG_CONFIG_DIR, STACKABLE_LOG_CONFIG_DIR_NAME, STACKABLE_LOG_DIR_NAME, + STACKABLE_DATA_DIR, STACKABLE_KERBEROS_DIR, STACKABLE_KERBEROS_KRB5_PATH, + STACKABLE_LISTENER_BOOTSTRAP_DIR, STACKABLE_LISTENER_BROKER_DIR, STACKABLE_LOG_CONFIG_DIR, + STACKABLE_LOG_CONFIG_DIR_NAME, STACKABLE_LOG_DIR_NAME, role::{ AnyConfig, KAFKA_NODE_ID_OFFSET, KafkaRole, broker::BrokerContainer, controller::ControllerContainer, @@ -427,6 +431,11 @@ pub fn build_controller_rolegroup_statefulset( let mut pod_builder = PodBuilder::new(); + if kafka_security.has_kerberos_enabled() { + add_kerberos_pod_config(kafka_security, kafka_role, &mut cb_kafka, &mut pod_builder) + .context(AddKerberosConfigSnafu)?; + } + let node_id_offset = node_id_hash32_offset(kafka_role, role_group_name.as_ref()).to_string(); // Operator-set env vars first (common + controller-specific); the user's `envOverrides` @@ -447,6 +456,10 @@ pub fn build_controller_rolegroup_statefulset( kafka_role, role_group_name, )?) + // 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(validated_rg.env_overrides.clone()) .into(); @@ -484,6 +497,7 @@ pub fn build_controller_rolegroup_statefulset( "-c".to_string(), ]) .args(vec![controller_kafka_container_command( + kafka_security, controller_pod_descriptors, )]); @@ -502,23 +516,16 @@ pub fn build_controller_rolegroup_statefulset( .startup_probe(controller_startup_probe) .liveness_probe(controller_liveness_probe) .readiness_probe(controller_readiness_probe); - // Skipped when Kerberos is enabled, matching `build_quorum_manager_container`'s own - // gating — `admin-client.properties` (the file this removal call relies on) only covers - // the TLS/SSL case. - if !kafka_security.has_kerberos_enabled() { - cb_kafka.lifecycle_pre_stop(LifecycleHandler { - exec: Some(ExecAction { - command: Some(vec![ - "/bin/bash".to_string(), - "-c".to_string(), - controller_remove_self_pre_stop_command( - merged_config.graceful_shutdown_timeout, - ), - ]), - }), - ..LifecycleHandler::default() - }); - } + cb_kafka.lifecycle_pre_stop(LifecycleHandler { + exec: Some(ExecAction { + command: Some(vec![ + "/bin/bash".to_string(), + "-c".to_string(), + controller_remove_self_pre_stop_command(merged_config.graceful_shutdown_timeout), + ]), + }), + ..LifecycleHandler::default() + }); add_log_config_volume( &mut pod_builder, @@ -551,11 +558,11 @@ pub fn build_controller_rolegroup_statefulset( .add_container(kafka_container) .affinity(&merged_config.affinity); - if let Some(quorum_manager_container) = - build_quorum_manager_container(resolved_product_image, kafka_security, quorum_manager_env) - { - pod_builder.add_container(quorum_manager_container); - } + pod_builder.add_container(build_quorum_manager_container( + resolved_product_image, + kafka_security, + quorum_manager_env, + )); add_common_pod_config( &mut pod_builder, @@ -745,17 +752,12 @@ fn add_common_pod_config( // Name of the controller's `quorum-manager` sidecar container. stackable_operator::constant!(QUORUM_MANAGER_CONTAINER_NAME: ContainerName = "quorum-manager"); -/// Builds the `quorum-manager` sidecar for a controller pod. Returns `None` when Kerberos is -/// enabled (the sidecar's admin-client properties file only covers the TLS/SSL case). +/// Builds the `quorum-manager` sidecar for a controller pod. fn build_quorum_manager_container( resolved_product_image: &ResolvedProductImage, kafka_security: &ValidatedKafkaSecurity, env: Vec, -) -> Option { - if kafka_security.has_kerberos_enabled() { - return None; - } - +) -> stackable_operator::k8s_openapi::api::core::v1::Container { let mut cb = new_container_builder(&QUORUM_MANAGER_CONTAINER_NAME); cb.image_from_product_image(resolved_product_image) @@ -794,7 +796,28 @@ 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."); - Some(cb.build()) + if kafka_security.has_kerberos_enabled() { + // `controller_admin_client_properties` authenticates with the pod-scoped keytab + // mounted by `add_kerberos_pod_config`; 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."); + cb.add_env_var(KRB5_CONFIG.to_string(), 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`. + cb.add_env_var( + KAFKA_OPTS.to_string(), + format!("-Djava.security.krb5.conf={STACKABLE_KERBEROS_KRB5_PATH}"), + ); + } + + cb.build() } /// Adds the Vector log-aggregation sidecar container, when the Vector agent is enabled. @@ -1217,8 +1240,10 @@ mod tests { let script = command.last().expect("the exec command has a script arg"); assert!( - script.contains(&format!("/dev/tcp/localhost/{client_port}")), - "expected a TCP reachability check against the controller's own port, script was: {script}" + script.contains(&format!( + "/dev/tcp/$POD_NAME.$ROLEGROUP_HEADLESS_SERVICE_NAME.$NAMESPACE.svc.$CLUSTER_DOMAIN/{client_port}" + )), + "expected the TCP check to dial the address the controller actually binds, script was: {script}" ); assert!( script.contains(r#"[ "$state" != "unattached" ]"#), diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs index b9eb650a..6587faf4 100644 --- a/rust/operator-binary/src/controller/build/security.rs +++ b/rust/operator-binary/src/controller/build/security.rs @@ -26,8 +26,8 @@ use stackable_operator::{ use crate::{ controller::security::ValidatedKafkaSecurity, crd::{ - LISTENER_BOOTSTRAP_VOLUME_NAME, LISTENER_BROKER_VOLUME_NAME, STACKABLE_KERBEROS_KRB5_PATH, - STACKABLE_LISTENER_BROKER_DIR, + CONTROLLER_POD_FQDN_TEMPLATE, LISTENER_BOOTSTRAP_VOLUME_NAME, LISTENER_BROKER_VOLUME_NAME, + STACKABLE_KERBEROS_KRB5_PATH, STACKABLE_LISTENER_BROKER_DIR, listener::{ self, KafkaListenerName, KafkaListenerProtocol, node_address_cmd_env, node_port_cmd_env, }, @@ -51,6 +51,11 @@ 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"; +const STACKABLE_KERBEROS_KEYTAB_PATH: &str = "/stackable/kerberos/keytab"; + 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"); const STACKABLE_TLS_KAFKA_SERVER_DIR: &str = "/stackable/tls-kafka-server"; @@ -173,35 +178,21 @@ pub fn client_properties(security: &ValidatedKafkaSecurity) -> Vec<(String, Opti )); push_client_ssl_stores(&mut props, STACKABLE_TLS_KAFKA_SERVER_DIR); } else if security.has_kerberos_enabled() { - // TODO: to make this configuration file usable out of the box the operator needs to be - // refactored to write out Java jaas files instead of passing command line parameters - // to the Kafka daemon scripts. - // This will simplify the code and the command lines lot. - // It will also make the jaas files reusable by the Kafka shell scripts. props.push(( PROPERTY_SECURITY_PROTOCOL.to_string(), Some(KafkaListenerProtocol::SaslSsl.to_string()), )); push_client_ssl_stores(&mut props, STACKABLE_TLS_KAFKA_SERVER_DIR); + // `sasl.mechanism` is the client-side selector. `sasl.enabled.mechanisms` is the + // broker-side list of accepted mechanisms and has no effect in a client config. props.push(( - PROPERTY_SASL_ENABLED_MECHANISMS.to_string(), + PROPERTY_SASL_MECHANISM.to_string(), Some(SASL_MECHANISM_GSSAPI.to_string()), )); props.push(( PROPERTY_SASL_KERBEROS_SERVICE_NAME.to_string(), Some(KafkaRole::Broker.kerberos_service_name().to_string()), )); - props.push(( - PROPERTY_SASL_INTER_BROKER_MECHANISM.to_string(), - Some(SASL_MECHANISM_GSSAPI.to_string()), - )); - props.push(( - "sasl.jaas.config".to_string(), - Some(format!("com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab=\"{keytab}\" principal=\"{service}/{pod}@{realm}\"", - keytab="/stackable/kerberos/keytab", - service=KafkaRole::Broker.kerberos_service_name(), - pod="todo", - realm="$KERBEROS_REALM")))); } else if security.tls_server_secret_class().is_some() { props.push(( PROPERTY_SECURITY_PROTOCOL.to_string(), @@ -222,15 +213,55 @@ 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 /// `add_controller_volume_and_volume_mounts`. +/// 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 +/// [`add_kerberos_pod_config`][akpc] — 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]. +/// +/// [glc]: crate::controller::build::properties::listener::get_kafka_listener_config +/// [akpc]: crate::controller::build::kerberos::add_kerberos_pod_config +/// [qmcc]: crate::controller::build::command::quorum_manager_container_command pub fn controller_admin_client_properties( - _security: &ValidatedKafkaSecurity, + security: &ValidatedKafkaSecurity, ) -> Vec<(String, Option)> { let mut properties = vec![]; - properties.push(( - PROPERTY_SECURITY_PROTOCOL.to_string(), - Some(KafkaListenerProtocol::Ssl.to_string()), - )); + 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()), + )); + } + push_client_ssl_stores(&mut properties, STACKABLE_TLS_KAFKA_INTERNAL_DIR); properties @@ -473,6 +504,10 @@ pub fn broker_config_settings(security: &ValidatedKafkaSecurity) -> BTreeMap BTreeMap 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); } @@ -719,7 +758,7 @@ pub(crate) mod tests { } /// Plaintext: no TLS, no authentication, no OPA. - fn plaintext() -> ValidatedKafkaSecurity { + pub(crate) fn plaintext() -> ValidatedKafkaSecurity { ValidatedKafkaSecurity::new( no_auth(), SecretClassName::from_str("tls").expect("tls secret class name is valid"), @@ -911,19 +950,96 @@ pub(crate) mod tests { props.get("security.protocol"), Some(&Some("SASL_SSL".to_string())) ); + // `sasl.mechanism`, not the broker-side `sasl.enabled.mechanisms`; and no + // `sasl.jaas.config`, which this out-of-pod consumer cannot use. See + // `discovery_client_properties_carry_no_server_side_or_pod_local_settings`. assert_eq!( - props.get("sasl.enabled.mechanisms"), + props.get("sasl.mechanism"), Some(&Some("GSSAPI".to_string())) ); + assert!(!props.contains_key("sasl.enabled.mechanisms")); assert_eq!( props.get("sasl.kerberos.service.name"), Some(&Some("kafka".to_string())) ); - assert!(props.contains_key("sasl.jaas.config")); + assert!(!props.contains_key("sasl.jaas.config")); + } + + #[test] + fn discovery_client_properties_carry_no_server_side_or_pod_local_settings() { + let props = as_map(client_properties(&kerberos())); + + // The consumer runs outside Kafka pods: it has no keytab and no pod principal, so a + // `sasl.jaas.config` here could only ever be wrong. Clients supply their own. + assert!(!props.contains_key("sasl.jaas.config")); + // Broker-side properties with no meaning in a client config. + assert!(!props.contains_key("sasl.mechanism.inter.broker.protocol")); + assert!(!props.contains_key("sasl.enabled.mechanisms")); + + // What a client actually needs. + 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())) + ); + assert_eq!( + props.get("ssl.truststore.location"), + Some(&Some( + "/stackable/tls-kafka-server/truststore.p12".to_string() + )) + ); } // ---- 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() + )) + ); + } + + #[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())) + ); + assert!(!props.contains_key("sasl.mechanism")); + assert!(!props.contains_key("sasl.jaas.config")); + assert_eq!( + props.get("ssl.keystore.location"), + Some(&Some( + "/stackable/tls-kafka-internal/keystore.p12".to_string() + )) + ); + } + #[test] fn controller_admin_client_properties_uses_the_internal_tls_directory() { let security = server_tls(); @@ -1067,6 +1183,36 @@ pub(crate) mod tests { 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()); + 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()) + .contains_key("sasl.mechanism.controller.protocol") + ); + } + #[test] fn controller_config_kerberos_adds_sasl() { let config = controller_config_settings(&kerberos()); diff --git a/rust/operator-binary/src/crd/listener.rs b/rust/operator-binary/src/crd/listener.rs index 7aabadad..8e014d14 100644 --- a/rust/operator-binary/src/crd/listener.rs +++ b/rust/operator-binary/src/crd/listener.rs @@ -58,13 +58,10 @@ 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 only protocol used is SSL. + /// The protocol used is SSL, or SASL_SSL when Kerberos is enabled. /// The advertised host names are FQDN pod names of the controllers. /// - /// Notes: - /// - /// - there is no listener for client/controller communication - /// - this listener does not support SSL_SASL. + /// Note: there is no listener for client/controller communication. #[strum(serialize = "CONTROLLER")] Controller, } diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index 3402151f..81a30d72 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -76,6 +76,21 @@ constant!(pub STACKABLE_LOG_DIR_NAME: VolumeName = "log"); pub const BROKER_ID_POD_MAP_DIR: &str = "/stackable/broker-id-pod-map"; constant!(pub BROKER_ID_POD_MAP_DIR_NAME: VolumeName = "broker-id-pod-map-dir"); +/// A KRaft controller pod's own fully-qualified domain name, as `config-utils` placeholders +/// resolved at container start. +/// +/// 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}"; + +/// Same as above ([`CONTROLLER_POD_FQDN_TEMPLATE`]) but for use in scripts +/// the operator generates (probes, startup commands) where the shell expands the value +/// rather than `config-utils`. +pub const CONTROLLER_POD_FQDN_SHELL: &str = + "$POD_NAME.$ROLEGROUP_HEADLESS_SERVICE_NAME.$NAMESPACE.svc.$CLUSTER_DOMAIN"; + #[derive(Snafu, Debug)] pub enum Error { #[snafu(display( diff --git a/tests/templates/kuttl/kraft-kerberos/00-assert.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/00-assert.yaml.j2 new file mode 100644 index 00000000..50b1d4c3 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/00-assert.yaml.j2 @@ -0,0 +1,10 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +{% if lookup('env', 'VECTOR_AGGREGATOR') %} +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: vector-aggregator-discovery +{% endif %} diff --git a/tests/templates/kuttl/kraft-kerberos/00-install-vector-aggregator-discovery-configmap.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/00-install-vector-aggregator-discovery-configmap.yaml.j2 new file mode 100644 index 00000000..2d6a0df5 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/00-install-vector-aggregator-discovery-configmap.yaml.j2 @@ -0,0 +1,9 @@ +{% if lookup('env', 'VECTOR_AGGREGATOR') %} +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: vector-aggregator-discovery +data: + ADDRESS: {{ lookup('env', 'VECTOR_AGGREGATOR') }} +{% endif %} diff --git a/tests/templates/kuttl/kraft-kerberos/00-patch-ns.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/00-patch-ns.yaml.j2 new file mode 100644 index 00000000..67185acf --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/00-patch-ns.yaml.j2 @@ -0,0 +1,9 @@ +{% if test_scenario['values']['openshift'] == 'true' %} +# see https://github.com/stackabletech/issues/issues/566 +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + - script: kubectl patch namespace $NAMESPACE -p '{"metadata":{"labels":{"pod-security.kubernetes.io/enforce":"privileged"}}}' + timeout: 120 +{% endif %} diff --git a/tests/templates/kuttl/kraft-kerberos/00-rbac.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/00-rbac.yaml.j2 new file mode 100644 index 00000000..7ee61d23 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/00-rbac.yaml.j2 @@ -0,0 +1,29 @@ +--- +kind: Role +apiVersion: rbac.authorization.k8s.io/v1 +metadata: + name: test-role +rules: +{% if test_scenario['values']['openshift'] == "true" %} + - apiGroups: ["security.openshift.io"] + resources: ["securitycontextconstraints"] + resourceNames: ["privileged"] + verbs: ["use"] +{% endif %} +--- +apiVersion: v1 +kind: ServiceAccount +metadata: + name: test-sa +--- +kind: RoleBinding +apiVersion: rbac.authorization.k8s.io/v1 +metadata: + name: test-rb +subjects: + - kind: ServiceAccount + name: test-sa +roleRef: + kind: Role + name: test-role + apiGroup: rbac.authorization.k8s.io diff --git a/tests/templates/kuttl/kraft-kerberos/01-assert.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/01-assert.yaml.j2 new file mode 100644 index 00000000..d34c1c63 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/01-assert.yaml.j2 @@ -0,0 +1,14 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 300 +{% if test_scenario['values']['kerberos-backend'] == 'mit' %} +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: krb5-kdc +status: + readyReplicas: 1 + replicas: 1 +{% endif %} diff --git a/tests/templates/kuttl/kraft-kerberos/01-install-krb5-kdc.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/01-install-krb5-kdc.yaml.j2 new file mode 100644 index 00000000..69ceec81 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/01-install-krb5-kdc.yaml.j2 @@ -0,0 +1,146 @@ +{% if test_scenario['values']['kerberos-backend'] == 'mit' %} +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: krb5-kdc +spec: + selector: + matchLabels: + app: krb5-kdc + template: + metadata: + labels: + app: krb5-kdc + spec: + serviceAccountName: test-sa + initContainers: + - name: init + image: oci.stackable.tech/sdp/krb5:{{ test_scenario['values']['krb5'] }}-stackable0.0.0-dev + args: + - sh + - -euo + - pipefail + - -c + - | + test -e /var/kerberos/krb5kdc/principal || kdb5_util create -s -P asdf + kadmin.local get_principal -terse root/admin || kadmin.local add_principal -pw asdf root/admin + # stackable-secret-operator principal must match the keytab specified in the SecretClass + kadmin.local get_principal -terse stackable-secret-operator || kadmin.local add_principal -e aes256-cts-hmac-sha384-192:normal -pw asdf stackable-secret-operator + env: + - name: KRB5_CONFIG + value: /stackable/config/krb5.conf + volumeMounts: + - mountPath: /stackable/config + name: config + - mountPath: /var/kerberos/krb5kdc + name: data + containers: + - name: kdc + image: oci.stackable.tech/sdp/krb5:{{ test_scenario['values']['krb5'] }}-stackable0.0.0-dev + args: + - krb5kdc + - -n + env: + - name: KRB5_CONFIG + value: /stackable/config/krb5.conf + volumeMounts: + - mountPath: /stackable/config + name: config + - mountPath: /var/kerberos/krb5kdc + name: data +# Root permissions required on Openshift to bind to privileged port numbers +{% if test_scenario['values']['openshift'] == "true" %} + securityContext: + runAsUser: 0 +{% endif %} + - name: kadmind + image: oci.stackable.tech/sdp/krb5:{{ test_scenario['values']['krb5'] }}-stackable0.0.0-dev + args: + - kadmind + - -nofork + env: + - name: KRB5_CONFIG + value: /stackable/config/krb5.conf + volumeMounts: + - mountPath: /stackable/config + name: config + - mountPath: /var/kerberos/krb5kdc + name: data +# Root permissions required on Openshift to bind to privileged port numbers +{% if test_scenario['values']['openshift'] == "true" %} + securityContext: + runAsUser: 0 +{% endif %} + - name: client + image: oci.stackable.tech/sdp/krb5:{{ test_scenario['values']['krb5'] }}-stackable0.0.0-dev + tty: true + stdin: true + env: + - name: KRB5_CONFIG + value: /stackable/config/krb5.conf + volumeMounts: + - mountPath: /stackable/config + name: config + volumes: + - name: config + configMap: + name: krb5-kdc + volumeClaimTemplates: + - metadata: + name: data + spec: + accessModes: + - ReadWriteOnce + resources: + requests: + storage: 1Gi +--- +apiVersion: v1 +kind: Service +metadata: + name: krb5-kdc +spec: + selector: + app: krb5-kdc + ports: + - name: kadmin + port: 749 + - name: kdc + port: 88 + - name: kdc-udp + port: 88 + protocol: UDP +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: krb5-kdc +data: + krb5.conf: | + [logging] + default = STDERR + kdc = STDERR + admin_server = STDERR + # default = FILE:/var/log/krb5libs.log + # kdc = FILE:/var/log/krb5kdc.log + # admin_server = FILE:/vaggr/log/kadmind.log + [libdefaults] + dns_lookup_realm = false + ticket_lifetime = 24h + renew_lifetime = 7d + forwardable = true + rdns = false + default_realm = {{ test_scenario['values']['kerberos-realm'] }} + spake_preauth_groups = edwards25519 + [realms] + {{ test_scenario['values']['kerberos-realm'] }} = { + acl_file = /stackable/config/kadm5.acl + disable_encrypted_timestamp = false + } + [domain_realm] + .cluster.local = {{ test_scenario['values']['kerberos-realm'] }} + cluster.local = {{ test_scenario['values']['kerberos-realm'] }} + kadm5.acl: | + root/admin *e + stackable-secret-operator *e +{% endif %} diff --git a/tests/templates/kuttl/kraft-kerberos/02-create-kerberos-secretclass.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/02-create-kerberos-secretclass.yaml.j2 new file mode 100644 index 00000000..04ae9a63 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/02-create-kerberos-secretclass.yaml.j2 @@ -0,0 +1,72 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + - script: | + kubectl apply -n $NAMESPACE -f - < 0 %} + custom: "{{ test_scenario['values']['kafka-kraft'].split(',')[1] }}" + productVersion: "{{ test_scenario['values']['kafka-kraft'].split(',')[0] }}" +{% else %} + productVersion: "{{ test_scenario['values']['kafka-kraft'] }}" +{% endif %} + pullPolicy: IfNotPresent + clusterConfig: + # KRaft: metadata is managed by the controllers role, no ZooKeeper involved. + metadataManager: kraft + authentication: + - authenticationClass: kerberos-auth-$NAMESPACE + tls: + # Kerberos requires the use of server and internal TLS! + serverSecretClass: tls +{% if lookup('env', 'VECTOR_AGGREGATOR') %} + vectorAggregatorConfigMapName: vector-aggregator-discovery +{% endif %} + controllers: + config: + logging: + enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} + gracefulShutdownTimeout: 30s # speed up tests + roleGroups: + default: + # 3 controller replicas so that this test actually exercises inter-controller + # (Raft) Kerberos-authenticated traffic on the CONTROLLER listener, not just + # broker-to-controller traffic. + replicas: 3 + brokers: + config: + 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. + bootstrapListenerClass: cluster-internal + gracefulShutdownTimeout: 30s # speed up tests + roleGroups: + default: + replicas: 3 + EOF diff --git a/tests/templates/kuttl/kraft-kerberos/30-access-kafka.txt.j2 b/tests/templates/kuttl/kraft-kerberos/30-access-kafka.txt.j2 new file mode 100644 index 00000000..50a31864 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/30-access-kafka.txt.j2 @@ -0,0 +1,131 @@ +--- +apiVersion: batch/v1 +kind: Job +metadata: + name: access-kafka +spec: + template: + spec: + serviceAccountName: test-sa + containers: + - name: access-kafka +{% if test_scenario['values']['kafka-kraft'].find(",") > 0 %} + image: {{ test_scenario['values']['kafka-kraft'].split(',')[1] }} +{% else %} + image: oci.stackable.tech/sdp/kafka:{{ test_scenario['values']['kafka-kraft'] }}-stackable0.0.0-dev +{% endif %} + command: + - /bin/bash + - /tmp/script/script.sh + env: + - name: KRB5_CONFIG + value: /stackable/kerberos/krb5.conf + - name: KAFKA_OPTS + value: -Djava.security.krb5.conf=/stackable/kerberos/krb5.conf + - name: KAFKA + valueFrom: + configMapKeyRef: + name: test-kafka + key: KAFKA + volumeMounts: + - name: script + mountPath: /tmp/script + - mountPath: /stackable/tls-ca-cert-mount + name: tls-ca-cert-mount + - name: kerberos + mountPath: /stackable/kerberos + volumes: + - name: script + configMap: + name: access-kafka-script + - name: kerberos + ephemeral: + volumeClaimTemplate: + metadata: + annotations: + secrets.stackable.tech/class: kerberos-$NAMESPACE + secrets.stackable.tech/scope: service=access-kafka + secrets.stackable.tech/kerberos.service.names: developer + spec: + storageClassName: secrets.stackable.tech + accessModes: + - ReadWriteOnce + resources: + requests: + storage: "1" + - name: tls-ca-cert-mount + ephemeral: + volumeClaimTemplate: + metadata: + annotations: + secrets.stackable.tech/class: tls + secrets.stackable.tech/scope: pod + spec: + accessModes: + - ReadWriteOnce + resources: + requests: + storage: "1" + storageClassName: secrets.stackable.tech + volumeMode: Filesystem + securityContext: + fsGroup: 1000 + restartPolicy: OnFailure +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: access-kafka-script +data: + script.sh: | + set -euxo pipefail + + export KCAT_CONFIG=/stackable/kcat.conf + TOPIC=test-topic + CONSUMER_GROUP=test-consumer-group + + echo -e -n "\ + metadata.broker.list=$KAFKA\n\ + auto.offset.reset=beginning\n\ + security.protocol=SASL_SSL\n\ + ssl.ca.location=/stackable/tls-ca-cert-mount/ca.crt\n\ + sasl.kerberos.keytab=/stackable/kerberos/keytab\n\ + sasl.kerberos.service.name=kafka\n\ + sasl.kerberos.principal=developer/access-kafka.$NAMESPACE.svc.cluster.local@{{ test_scenario['values']['kerberos-realm'] }}\n\ + sasl.mechanism=GSSAPI\n\ + " > $KCAT_CONFIG + + cat $KCAT_CONFIG + + sent_message="Hello Stackable!" + + echo $sent_message | kcat \ + -t $TOPIC \ + -P + + echo Sent message: \"$sent_message\" + + # Explicit numeric offset (not "-o stored"/"auto.offset.reset=beginning"): the bundled kcat's + # librdkafka (1.7.0) mis-detects broker feature support against Kafka >=4.0 brokers, which + # dropped old low-numbered API versions (KIP-896). It matches ApiVersions by exact version + # instead of range, so it wrongly reports the ListOffsets logical-offset query as unsupported + # ("Failed to query logical offset BEGINNING: Local: Required feature not supported by + # broker") even though the broker supports it fine -- see + # https://github.com/confluentinc/librdkafka/issues/4948. This is unrelated to SASL/Kerberos: + # authentication succeeds either way. The topic is freshly created and this is the only + # message ever produced to it, so offset 0 is always the message we just sent. + received_message=$(kcat \ + -G $CONSUMER_GROUP \ + -o 0 \ + -e \ + $TOPIC) + + echo Received message: \"$received_message\" + + if [ "$received_message" = "$sent_message" ]; then + echo "Test passed" + exit 0 + else + echo "Test failed" + exit 1 + fi diff --git a/tests/templates/kuttl/kraft-kerberos/30-access-kafka.yaml b/tests/templates/kuttl/kraft-kerberos/30-access-kafka.yaml new file mode 100644 index 00000000..eecc0f08 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/30-access-kafka.yaml @@ -0,0 +1,6 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + # We need to replace $NAMESPACE (by KUTTL) + - script: envsubst '$NAMESPACE' < 30-access-kafka.txt | kubectl apply -n $NAMESPACE -f - diff --git a/tests/templates/kuttl/kraft-kerberos/30-assert.yaml b/tests/templates/kuttl/kraft-kerberos/30-assert.yaml new file mode 100644 index 00000000..edc6c317 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/30-assert.yaml @@ -0,0 +1,11 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 600 +--- +apiVersion: batch/v1 +kind: Job +metadata: + name: access-kafka +status: + succeeded: 1 diff --git a/tests/templates/kuttl/kraft-kerberos/60-assert.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/60-assert.yaml.j2 new file mode 100644 index 00000000..a8327181 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/60-assert.yaml.j2 @@ -0,0 +1,34 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 600 +commands: + - script: kubectl -n $NAMESPACE wait --for=condition=available kafkaclusters.kafka.stackable.tech/test-kafka --timeout 301s + - script: | + # 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. + # + kubectl exec -n $NAMESPACE test-kafka-controller-default-0 -c kafka -- \ + /stackable/kafka/bin/kafka-metadata-quorum.sh \ + --bootstrap-controller test-kafka-controller-default-0.test-kafka-controller-default-headless.$NAMESPACE.svc.cluster.local:9093 \ + --command-config /tmp/admin-client.properties \ + describe --replication | tail -n +2 | awk '$NF == "Leader" || $NF == "Follower"' | wc -l | grep -q '^5$' +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: test-kafka-broker-default +status: + readyReplicas: 3 + replicas: 3 +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: test-kafka-controller-default +status: + readyReplicas: 5 + replicas: 5 diff --git a/tests/templates/kuttl/kraft-kerberos/60-scale-controller-up.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/60-scale-controller-up.yaml.j2 new file mode 100644 index 00000000..f1ce309b --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/60-scale-controller-up.yaml.j2 @@ -0,0 +1,54 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +timeout: 600 +commands: + - script: | + kubectl apply -n $NAMESPACE -f - < 0 %} + custom: "{{ test_scenario['values']['kafka-kraft'].split(',')[1] }}" + productVersion: "{{ test_scenario['values']['kafka-kraft'].split(',')[0] }}" +{% else %} + productVersion: "{{ test_scenario['values']['kafka-kraft'] }}" +{% endif %} + pullPolicy: IfNotPresent + clusterConfig: + # KRaft: metadata is managed by the controllers role, no ZooKeeper involved. + metadataManager: kraft + authentication: + - authenticationClass: kerberos-auth-$NAMESPACE + tls: + # Kerberos requires the use of server and internal TLS! + serverSecretClass: tls +{% if lookup('env', 'VECTOR_AGGREGATOR') %} + vectorAggregatorConfigMapName: vector-aggregator-discovery +{% endif %} + controllers: + config: + logging: + enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} + gracefulShutdownTimeout: 30s # speed up tests + roleGroups: + default: + replicas: 5 + brokers: + config: + 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. + bootstrapListenerClass: cluster-internal + gracefulShutdownTimeout: 30s # speed up tests + roleGroups: + default: + replicas: 3 + EOF diff --git a/tests/templates/kuttl/kraft-kerberos/70-assert.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/70-assert.yaml.j2 new file mode 100644 index 00000000..83a0857b --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/70-assert.yaml.j2 @@ -0,0 +1,34 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 600 +commands: + - script: kubectl -n $NAMESPACE wait --for=condition=available kafkaclusters.kafka.stackable.tech/test-kafka --timeout 301s + - script: | + # 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. + # + kubectl exec -n $NAMESPACE test-kafka-controller-default-0 -c kafka -- \ + /stackable/kafka/bin/kafka-metadata-quorum.sh \ + --bootstrap-controller test-kafka-controller-default-0.test-kafka-controller-default-headless.$NAMESPACE.svc.cluster.local:9093 \ + --command-config /tmp/admin-client.properties \ + describe --replication | tail -n +2 | awk '$NF == "Leader" || $NF == "Follower"' | wc -l | grep -q '^3$' +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: test-kafka-broker-default +status: + readyReplicas: 3 + replicas: 3 +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: test-kafka-controller-default +status: + readyReplicas: 3 + replicas: 3 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 new file mode 100644 index 00000000..29fa460b --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/70-scale-controller-down.yaml.j2 @@ -0,0 +1,54 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +timeout: 600 +commands: + - script: | + kubectl apply -n $NAMESPACE -f - < 0 %} + custom: "{{ test_scenario['values']['kafka-kraft'].split(',')[1] }}" + productVersion: "{{ test_scenario['values']['kafka-kraft'].split(',')[0] }}" +{% else %} + productVersion: "{{ test_scenario['values']['kafka-kraft'] }}" +{% endif %} + pullPolicy: IfNotPresent + clusterConfig: + # KRaft: metadata is managed by the controllers role, no ZooKeeper involved. + metadataManager: kraft + authentication: + - authenticationClass: kerberos-auth-$NAMESPACE + tls: + # Kerberos requires the use of server and internal TLS! + serverSecretClass: tls +{% if lookup('env', 'VECTOR_AGGREGATOR') %} + vectorAggregatorConfigMapName: vector-aggregator-discovery +{% endif %} + controllers: + config: + logging: + enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} + gracefulShutdownTimeout: 30s # speed up tests + roleGroups: + default: + replicas: 3 + brokers: + config: + 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. + bootstrapListenerClass: cluster-internal + gracefulShutdownTimeout: 30s # speed up tests + roleGroups: + default: + replicas: 3 + EOF diff --git a/tests/templates/kuttl/kraft-kerberos/README.md b/tests/templates/kuttl/kraft-kerberos/README.md new file mode 100644 index 00000000..a85e47b8 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/README.md @@ -0,0 +1,9 @@ +# Kraft + Kerberos test + +Proves that a KRaft-mode Kafka cluster (`spec.controllers` present, no ZooKeeper) can be +secured with Kerberos authentication (`spec.clusterConfig.authentication` referencing a +Kerberos `AuthenticationClass`) end to end: controllers form a quorum, brokers join, and a +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`. diff --git a/tests/test-definition.yaml b/tests/test-definition.yaml index 9b47243a..5fba37ae 100644 --- a/tests/test-definition.yaml +++ b/tests/test-definition.yaml @@ -131,6 +131,14 @@ tests: - zookeeper-latest - kafka-latest - openshift + - name: kraft-kerberos + dimensions: + - kafka-kraft + - krb5 + - kerberos-realm + - kerberos-backend + - openshift + - broker-listener-class - name: kerberos dimensions: - kafka