feat: add Kerberos authentication for Kafka (#762)

* added enum for either vector of autentication classes or kerberos secret name

* initial kerberos impl

* implement kerberos specifics

* wip: integration test

* revert complex enum and use parallel struct (CRD decision pending)

* call shell explicitly for kerberos probe to allow variable substitution

* working test

* revert class name change and formatting

* added validation

* linting

* more linting

* refactor: add kerberos to list of authentication classes instead of dedicated struct

* changelog

* reverted operator-rs ref and corrected test

* fixed changes due to operator-rs to 0.78.0

* added docs/example

* improved comments

* Update rust/crd/src/authentication.rs

Co-authored-by: Sebastian Bernauer <sebastian.bernauer@stackable.de>

* Update rust/crd/src/lib.rs

Co-authored-by: Sebastian Bernauer <sebastian.bernauer@stackable.de>

* Update rust/operator-binary/src/kafka_controller.rs

Co-authored-by: Sebastian Bernauer <sebastian.bernauer@stackable.de>

* Update rust/operator-binary/src/kafka_controller.rs

Co-authored-by: Sebastian Bernauer <sebastian.bernauer@stackable.de>

* fixed review suggestions

* formatting: new lines between enum elements

* review suggestions

* add use-client-tls dimension and cleanup test

* add constants for kerberos paths

* test: Update kerberos tests to always use TLS

* added check that TLS is enabled for Kerberos

* regenerate charts

* formatting

* corrected validation check

* Update rust/operator-binary/src/kerberos.rs

Co-authored-by: Sebastian Bernauer <sebastian.bernauer@stackable.de>

* use listener volume scope for kerberos volume and replace FQDN with listener in advertised listeners

* added custom iamge usage from previous merge from main

* Update rust/crd/src/authentication.rs

Co-authored-by: Siegfried Weber <mail@siegfriedweber.net>

* remove unecessary test

* removed unused Error

* regenerate charts

* combine cases where internal tls is required

* working test with broker listeners instead of listener bootstrap

* add listener for bootstrapper

* removed duplicate check

* add bootstrap configs for client_auth as well

* use correct port in discovery for kerberos. Removed bootstrap changes for non-kerberos.

* use discovery in kerberos test

* added unit test for kerberos config

* added note about client connections and ports

* Update docs/modules/kafka/pages/usage-guide/security.adoc

Co-authored-by: Siegfried Weber <mail@siegfriedweber.net>

* clarified comment

---------

Co-authored-by: Sebastian Bernauer <sebastian.bernauer@stackable.de>
Co-authored-by: Sebastian Bernauer <sebastian.bernauer@stackable.tech>
Co-authored-by: Siegfried Weber <mail@siegfriedweber.net>

Author: 3 people

CHANGELOG.md

@@ -7,6 +7,7 @@ All notable changes to this project will be documented in this file.
 ### Added
 
 - Support version `3.8.0` ([#753]).
+- Add support for Kerberos authentication ([#762]).
 - The operator can now run on Kubernetes clusters using a non-default cluster domain.
   Use the env var `KUBERNETES_CLUSTER_DOMAIN` or the operator Helm chart property `kubernetesClusterDomain` to set a non-default cluster domain ([#771]).
 
@@ -35,6 +36,7 @@ All notable changes to this project will be documented in this file.
 [#741]: https://github.com/stackabletech/kafka-operator/pull/741
 [#750]: https://github.com/stackabletech/kafka-operator/pull/750
 [#753]: https://github.com/stackabletech/kafka-operator/pull/753
+[#762]: https://github.com/stackabletech/kafka-operator/pull/762
 [#771]: https://github.com/stackabletech/kafka-operator/pull/771
 [#773]: https://github.com/stackabletech/kafka-operator/pull/773
 

deploy/helm/kafka-operator/crds/crds.yaml

@@ -565,6 +565,10 @@ spec:
                               Only affects client connections. This setting controls: - If clients need to authenticate themselves against the broker via TLS - Which ca.crt to use when validating the provided client certs
 
                               This will override the server TLS settings (if set) in `spec.clusterConfig.tls.serverSecretClass`.
+
+                              ## Kerberos provider
+
+                              This affects client connections and also requires TLS for encryption. This setting is used to reference an `AuthenticationClass` and in turn, a `SecretClass` that is used to create keytabs.
                             type: string
                         required:
                           - authenticationClass

docs/modules/kafka/pages/usage-guide/security.adoc

@@ -53,6 +53,10 @@ You can create your own secrets and reference them e.g. in the `spec.clusterConf
 == Authentication
 
 The internal or broker-to-broker communication is authenticated via TLS.
+For client-to-server communication, authentication can be achieved with either TLS or Kerberos.
+
+=== TLS
+
 In order to enforce TLS authentication for client-to-server communication, you can set an `AuthenticationClass` reference in the custom resource provided by the xref:commons-operator:index.adoc[Commons Operator].
 
 [source,yaml]
@@ -101,6 +105,65 @@ spec:
 <3> The reference to a `SecretClass`.
 <4> The `SecretClass` that is referenced by the `AuthenticationClass` in order to provide certificates.
 
+=== Kerberos
+
+Similarly, you can set an `AuthenticationClass` reference for a Kerberos authentication provider:
+
+[source,yaml]
+----
+apiVersion: authentication.stackable.tech/v1alpha1
+kind: AuthenticationClass
+metadata:
+  name: kafka-client-kerberos # <2>
+spec:
+  provider:
+    kerberos:
+      kerberosSecretClass: kafka-client-auth-secret # <3>
+---
+apiVersion: secrets.stackable.tech/v1alpha1
+kind: SecretClass
+metadata:
+  name: kafka-client-auth-secret # <4>
+spec:
+  backend:
+    kerberosKeytab:
+      ...
+---
+apiVersion: kafka.stackable.tech/v1alpha1
+kind: KafkaCluster
+metadata:
+  name: simple-kafka
+spec:
+  image:
+    productVersion: 3.7.1
+  clusterConfig:
+    authentication:
+      - authenticationClass: kafka-client-kerberos # <1>
+    tls:
+      serverSecretClass: tls # <5>
+    zookeeperConfigMapName: simple-kafka-znode
+  brokers:
+    roleGroups:
+      default:
+        replicas: 3
+----
+<1> The `clusterConfig.authentication.authenticationClass` can be set to use Kerberos for authentication. This is optional.
+<2> The referenced `AuthenticationClass` that references a `SecretClass` to provide Kerberos keytabs.
+<3> The reference to a `SecretClass`.
+<4> The `SecretClass` that is referenced by the `AuthenticationClass` in order to provide keytabs.
+<5> The SecretClass that will be used for encryption.
+
+NOTE: When Kerberos is enabled it is also required to enable TLS for maximum security.
+
+==== 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.
+
+NOTE: Port 9094 is reserved for non-secure kerberized connections which is not currently implemented.
 
 == [[authorization]]Authorization
 

rust/crd/src/authentication.rs

@@ -9,7 +9,7 @@ use stackable_operator::{
     schemars::{self, JsonSchema},
 };
 
-const SUPPORTED_AUTHENTICATION_CLASS_PROVIDERS: [&str; 1] = ["TLS"];
+pub const SUPPORTED_AUTHENTICATION_CLASS_PROVIDERS: [&str; 2] = ["TLS", "Kerberos"];
 
 #[derive(Snafu, Debug)]
 pub enum Error {
@@ -18,9 +18,10 @@ pub enum Error {
         source: stackable_operator::client::Error,
         authentication_class: ObjectRef<AuthenticationClass>,
     },
-    // TODO: Adapt message if multiple authentication classes are supported
-    #[snafu(display("only one authentication class is currently supported. Possible Authentication class providers are {SUPPORTED_AUTHENTICATION_CLASS_PROVIDERS:?}"))]
+
+    #[snafu(display("only one authentication class at a time is currently supported. Possible Authentication class providers are {SUPPORTED_AUTHENTICATION_CLASS_PROVIDERS:?}"))]
     MultipleAuthenticationClassesProvided,
+
     #[snafu(display(
         "failed to use authentication provider [{provider}] for authentication class [{authentication_class}] - supported providers: {SUPPORTED_AUTHENTICATION_CLASS_PROVIDERS:?}",
     ))]
@@ -42,6 +43,12 @@ pub struct KafkaAuthentication {
     /// - Which ca.crt to use when validating the provided client certs
     ///
     /// This will override the server TLS settings (if set) in `spec.clusterConfig.tls.serverSecretClass`.
+    ///
+    /// ## Kerberos provider
+    ///
+    /// This affects client connections and also requires TLS for encryption.
+    /// This setting is used to reference an `AuthenticationClass` and in turn, a `SecretClass` that is
+    /// used to create keytabs.
     pub authentication_class: String,
 }
 
@@ -90,6 +97,13 @@ impl ResolvedAuthenticationClasses {
             .find(|auth| matches!(auth.spec.provider, AuthenticationClassProvider::Tls(_)))
     }
 
+    /// Return the (first) Kerberos `AuthenticationClass` if available
+    pub fn get_kerberos_authentication_class(&self) -> Option<&AuthenticationClass> {
+        self.resolved_authentication_classes
+            .iter()
+            .find(|auth| matches!(auth.spec.provider, AuthenticationClassProvider::Kerberos(_)))
+    }
+
     /// Validates the resolved AuthenticationClasses.
     /// Currently errors out if:
     /// - More than one AuthenticationClass was provided
@@ -101,8 +115,11 @@ impl ResolvedAuthenticationClasses {
 
         for auth_class in &self.resolved_authentication_classes {
             match &auth_class.spec.provider {
-                AuthenticationClassProvider::Tls(_) => {}
-                _ => {
+                // explicitly list each branch so new elements do not get overlooked
+                AuthenticationClassProvider::Tls(_) | AuthenticationClassProvider::Kerberos(_) => {}
+                AuthenticationClassProvider::Static(_)
+                | AuthenticationClassProvider::Ldap(_)
+                | AuthenticationClassProvider::Oidc(_) => {
                     return Err(Error::AuthenticationProviderNotSupported {
                         authentication_class: ObjectRef::from_obj(auth_class),
                         provider: auth_class.spec.provider.to_string(),

rust/crd/src/lib.rs

@@ -5,11 +5,11 @@ pub mod listener;
 pub mod security;
 pub mod tls;
 
-use crate::authentication::KafkaAuthentication;
 use crate::authorization::KafkaAuthorization;
 use crate::tls::KafkaTls;
 
 use affinity::get_affinity;
+use authentication::KafkaAuthentication;
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
@@ -63,6 +63,9 @@ pub const STACKABLE_DATA_DIR: &str = "/stackable/data";
 pub const STACKABLE_CONFIG_DIR: &str = "/stackable/config";
 pub const STACKABLE_LOG_CONFIG_DIR: &str = "/stackable/log_config";
 pub const STACKABLE_LOG_DIR: &str = "/stackable/log";
+// kerberos
+pub const STACKABLE_KERBEROS_DIR: &str = "/stackable/kerberos";
+pub const STACKABLE_KERBEROS_KRB5_PATH: &str = "/stackable/kerberos/krb5.conf";
 
 const DEFAULT_BROKER_GRACEFUL_SHUTDOWN_TIMEOUT: Duration = Duration::from_minutes_unchecked(30);
 
@@ -335,6 +338,13 @@ impl KafkaRole {
         }
         roles
     }
+
+    /// A Kerberos principal has three parts, with the form username/fully.qualified.domain.name@YOUR-REALM.COM.
+    /// We only have one role and will use "kafka" everywhere (which e.g. differs from the current hdfs implementation,
+    /// but is similar to HBase).
+    pub fn kerberos_service_name(&self) -> &'static str {
+        "kafka"
+    }
 }
 
 #[derive(Clone, Debug, Default, PartialEq, Fragment, JsonSchema)]