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 can authorize both external and internal communication. For internal communication, the principal name is the subject DN of the broker’s certificate. The subject DN 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. An OpaCluster and the Rego rules that verify the subject DNs could look as follows:

---
apiVersion: opa.stackable.tech/v1alpha1
kind: OpaCluster
metadata:
  name: simple-opa
spec:
  image:
    productVersion: 1.16.2
  servers:
    config:
      logging:
        containers:
          opa:
            loggers:
              decision:
                level: INFO
    roleGroups:
      default: {}
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: kafka-rego-rules
  labels:
    opa.stackable.tech/bundle: "true"
data:
  kafka.rego: |
    package kafka

    is_internal_request if input.requestContext.listenerName == "INTERNAL"
    is_external_request if not is_internal_request

    # FQDNs of the brokers as regular expressions
    brokers_fqdn := [
      `local`,                                 # cluster domain
      `cluster`,                               # cluster domain
      `svc`,
      `default`,                               # namespace
      `simple-kafka-broker-default-headless`,  # Pod subdomain
      `simple-kafka-broker-default-[0-9]+`,    # Pod name
    ]

    domain_components = [concat("", [`DC=`, dc]) | some dc in brokers_fqdn]
    common_name = `CN=generated certificate for pod`
    subject_dn_fields = array.concat(domain_components, [common_name])
    brokers_subject_dn_pattern := concat("", ["^", concat(",", subject_dn_fields), "$"])

    default allow := false

    allow if {
      is_internal_request

      regex.match(
        brokers_subject_dn_pattern,
        input.requestContext.principal.name,
      )
    }

    allow if {
      is_external_request

      # TODO Add your own rules
    }