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