Security

Encryption

The internal and client communication can be encrypted with TLS. This requires the Secret Operator to be present in order to provide certificates. The utilized certificates can be changed in a top-level config.

---
apiVersion: kafka.stackable.tech/v1alpha1
kind: KafkaCluster
metadata:
  name: simple-kafka
spec:
  image:
    productVersion: 3.9.2
  clusterConfig:
    zookeeperConfigMapName: simple-kafka-znode
    tls:
      serverSecretClass: tls (1)
      internalSecretClass: kafka-internal-tls (2)
  brokers:
    config:
      requestedSecretLifetime: 7d (3)
    roleGroups:
      default:
        replicas: 3

1	The `spec.clusterConfig.tls.serverSecretClass` refers to the client-to-server encryption. Defaults to the `tls` secret. Can be deactivated by setting `serverSecretClass` to `null`.
2	The `spec.clusterConfig.tls.internalSecretClass` refers to the broker-to-broker internal encryption. This must be explicitly set or defaults to `tls`. May be disabled by setting `internalSecretClass` to `null`.
3	The lifetime for autoTls certificates generated by the secret operator. Only a lifetime up to the `maxCertificateLifetime` setting in the SecretClass is applied.

The tls secret is deployed from the Secret Operator and looks like this:

---
apiVersion: secrets.stackable.tech/v1alpha1
kind: SecretClass
metadata:
  name: tls
spec:
  backend:
    autoTls:
      ca:
        secret:
          name: secret-provisioner-tls-ca
          namespace: default
        autoGenerate: true
      maxCertificateLifetime: 15d

You can create your own secrets and reference them e.g. in the spec.clusterConfig.tls.serverSecretClass or spec.clusterConfig.tls.internalSecretClass to use different certificates.

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 Commons Operator.

---
apiVersion: authentication.stackable.tech/v1alpha1
kind: AuthenticationClass
metadata:
  name: kafka-client-tls (2)
spec:
  provider:
    tls:
      clientCertSecretClass: kafka-client-auth-secret (3)
---
apiVersion: secrets.stackable.tech/v1alpha1
kind: SecretClass
metadata:
  name: kafka-client-auth-secret (4)
spec:
  backend:
    autoTls:
      ca:
        secret:
          name: secret-provisioner-tls-kafka-client-ca
          namespace: default
        autoGenerate: true
---
apiVersion: kafka.stackable.tech/v1alpha1
kind: KafkaCluster
metadata:
  name: simple-kafka
spec:
  image:
    productVersion: 3.9.2
  clusterConfig:
    authentication:
      - authenticationClass: kafka-client-tls (1)
    zookeeperConfigMapName: simple-kafka-znode
  brokers:
    roleGroups:
      default:
        replicas: 3

1	The `clusterConfig.authentication.authenticationClass` can be set to use TLS for authentication. This is optional.
2	The referenced `AuthenticationClass` that references a `SecretClass` to provide certificates.
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:

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.9.2
  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.

When Kerberos is enabled it is also required to enable TLS for maximum security.

Clients

Why each broker has two principals

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/<bootstrap-address> and kafka/<broker-address>, 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 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:

Cluster Port name Port

Cluster	Port name	Port
Kerberos (always TLS)	`bootstrap`	9095
TLS, no Kerberos	`kafka-tls`	9093
No TLS, no Kerberos	`kafka`	9092

Kerberos (always TLS)

bootstrap

9095

TLS, no Kerberos

kafka-tls

9093

No TLS, no Kerberos

kafka

9092

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

If you wish to include integration with Open Policy Agent and already have an OPA cluster, then you can include an opa field pointing to the OPA cluster discovery ConfigMap and the required package. The package is optional and defaults to the metadata.name field:

---
apiVersion: kafka.stackable.tech/v1alpha1
kind: KafkaCluster
metadata:
  name: simple-kafka
spec:
  image:
    productVersion: 3.9.2
  clusterConfig:
    authorization:
      opa:
        configMapName: simple-opa
        package: kafka
    zookeeperConfigMapName: simple-kafka-znode
  brokers:
    roleGroups:
      default:
        replicas: 1

You can change some opa cache properties by overriding:

---
apiVersion: kafka.stackable.tech/v1alpha1
kind: KafkaCluster
metadata:
  name: simple-kafka
spec:
  image:
    productVersion: 3.9.2
  clusterConfig:
    authorization:
      opa:
        configMapName: simple-opa
        package: kafka
    zookeeperConfigMapName: simple-kafka-znode
  brokers:
    configOverrides:
      broker.properties:
        opa.authorizer.cache.initial.capacity: "100"
        opa.authorizer.cache.maximum.size: "100"
        opa.authorizer.cache.expire.after.seconds: "10"
    roleGroups:
      default:
        replicas: 1

A full list of settings and their respective defaults can be found here.

The OPA Rego rules authorize requests from external clients. Communication between the Kafka nodes is not authorized by OPA, see [authorization-internal-traffic].

Internal traffic

The Kafka nodes (brokers and controllers) are configured as super.users, so that the traffic between them is allowed without asking OPA. You do not need to write Rego rules for it. Requests that a broker forwards to a controller on behalf of a client, such as creating a topic, are still authorized against the principal of the original client.

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.

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. The secret operator always sets it in the subject of the pod certificates.

Everything that presents a certificate with the subject DN of a node, issued by the CA of the internalSecretClass, is treated as a Kafka node. Restrict who can obtain such certificates accordingly.

If you previously set super.users for the Kafka nodes yourself, for example with configOverrides, you can remove that override. The operator now sets the principals of all nodes automatically. Keep an override only if you need additional super users, such as specific client principals, and then include the entries for the nodes as well, because the override replaces the value set by the operator.

Scaling a role changes the nodes of the cluster and with them the node properties. This restarts all Pods of the cluster, see Scaling operations restart the whole cluster.