Chapter 16. Securing access to a Kafka cluster

Procedure

1. Configure the networkPolicyPeers property to define the application pods or namespaces allowed to access the Kafka cluster.

   This example shows configuration for a tls listener to allow connections only from application pods with the label app set to kafka-client:

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     kafka:
       # ...
       listeners:
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication:
             type: tls
           networkPolicyPeers:
             - podSelector:
                 matchLabels:
                   app: kafka-client
       # ...
     zookeeper:
       # ...

2. Apply the changes to the Kafka resource configuration.

Procedure

1. Create a Secret containing your private key and server certificate:

   oc create secret generic <my_secret> --from-file=<my_listener_key.key> --from-file=<my_listener_certificate.crt>

2. Edit the Kafka resource for your cluster.

   Configure the listener to use your Secret, certificate file, and private key file in the configuration.brokerCertChainAndKey property.

   Example configuration for a loadbalancer external listener with TLS encryption enabled

   # ...
   listeners:
     - name: plain
       port: 9092
       type: internal
       tls: false
     - name: external3
       port: 9094
       type: loadbalancer
       tls: true
       configuration:
         brokerCertChainAndKey:
           secretName: my-secret
           certificate: my-listener-certificate.crt
           key: my-listener-key.key
   # ...

   Example configuration for a TLS listener

   # ...
   listeners:
     - name: plain
       port: 9092
       type: internal
       tls: false
     - name: tls
       port: 9093
       type: internal
       tls: true
       configuration:
         brokerCertChainAndKey:
           secretName: my-secret
           certificate: my-listener-certificate.crt
           key: my-listener-key.key
   # ...

3. Apply the changes to the Kafka resource configuration.

   The Cluster Operator starts a rolling update of the Kafka cluster, which updates the configuration of the listeners.

   Note

   A rolling update is also started if you update a Kafka listener certificate in a Secret that is already used by a listener.

1. Configure the Kafka cluster with a Kafka listener.

   • Define the authentication required to access the Kafka broker through the listener.

   • Enable authorization on the Kafka broker.

     Example listener configuration

     apiVersion: kafka.strimzi.io/v1beta2
     kind: Kafka
     metadata:
       name: my-cluster
       namespace: myproject
     spec:
       kafka:
         # ...
         listeners: 1
         - name: external1 2
           port: 9094 3
           type: <listener_type> 4
           tls: true 5
           authentication:
             type: tls 6
           configuration: 7
             #...
         authorization: 8
           type: simple
           superUsers:
             - super-user-name 9
       # ...

     1

     Configuration options for enabling external listeners are described in the Generic Kafka listener schema reference.

     2

     Name to identify the listener. Must be unique within the Kafka cluster.

     3

     Port number used by the listener inside Kafka. The port number has to be unique within a given Kafka cluster. Allowed port numbers are 9092 and higher with the exception of ports 9404 and 9999, which are already used for Prometheus and JMX. Depending on the listener type, the port number might not be the same as the port number that connects Kafka clients.

     4

     External listener type specified as route (OpenShift only), loadbalancer, nodeport or ingress (Kubernetes only). An internal listener is specified as internal or cluster-ip.

     5

     Required. TLS encryption on the listener. For route and ingress type listeners it must be set to true. For mTLS authentication, also use the authentication property.

     6

     Client authentication mechanism on the listener. For server and client authentication using mTLS, you specify tls: true and authentication.type: tls.

     7

     (Optional) Depending on the requirements of the listener type, you can specify additional listener configuration.

     8

     Authorization specified as simple, which uses the AclAuthorizer and StandardAuthorizer Kafka plugins.

     9

     (Optional) Super users can access all brokers regardless of any access restrictions defined in ACLs.

     Warning

     An OpenShift route address comprises the Kafka cluster name, the listener name, the project name, and the domain of the router. For example, my-cluster-kafka-external1-bootstrap-my-project.domain.com (<cluster_name>-kafka-<listener_name>-bootstrap-<namespace>.<domain>). Each DNS label (between periods “.”) must not exceed 63 characters, and the total length of the address must not exceed 255 characters.

2. Apply the changes to the Kafka resource configuration.

   The Kafka cluster is configured with a Kafka broker listener using mTLS authentication.

   A service is created for each Kafka broker pod.

   A service is created to serve as the bootstrap address for connection to the Kafka cluster.

   A service is also created as the external bootstrap address for external connection to the Kafka cluster using nodeport listeners.

   The cluster CA certificate to verify the identity of the kafka brokers is also created in the secret <cluster_name>-cluster-ca-cert.

   Note

   If you scale your Kafka cluster while using external listeners, it might trigger a rolling update of all Kafka brokers. This depends on the configuration.

3. Retrieve the bootstrap address you can use to access the Kafka cluster from the status of the Kafka resource.

   oc get kafka <kafka_cluster_name> -o=jsonpath='{.status.listeners[?(@.name=="<listener_name>")].bootstrapServers}{"\n"}'

   For example:

   oc get kafka my-cluster -o=jsonpath='{.status.listeners[?(@.name=="external")].bootstrapServers}{"\n"}'

   Use the bootstrap address in your Kafka client to connect to the Kafka cluster.

1. Create or modify a user representing the client that requires access to the Kafka cluster.

   • Specify the same authentication type as the Kafka listener.

   • Specify the authorization ACLs for simple authorization.

     Example user configuration

     apiVersion: kafka.strimzi.io/v1beta2
     kind: KafkaUser
     metadata:
       name: my-user
       labels:
         strimzi.io/cluster: my-cluster 1
     spec:
       authentication:
         type: tls 2
       authorization:
         type: simple
         acls: 3
           - resource:
               type: topic
               name: my-topic
               patternType: literal
             operations:
               - Describe
               - Read
           - resource:
               type: group
               name: my-group
               patternType: literal
             operations:
               - Read

     1

     The label must match the label of the Kafka cluster.

     2

     Authentication specified as mutual tls.

     3

     Simple authorization requires an accompanying list of ACL rules to apply to the user. The rules define the operations allowed on Kafka resources based on the username (my-user).

2. Apply the changes to the KafkaUser resource configuration.

   The user is created, as well as a secret with the same name as the KafkaUser resource. The secret contains a public and private key for mTLS authentication.

   Example secret with user credentials

   apiVersion: v1
   kind: Secret
   metadata:
     name: my-user
     labels:
       strimzi.io/kind: KafkaUser
       strimzi.io/cluster: my-cluster
   type: Opaque
   data:
     ca.crt: <public_key> # Public key of the clients CA used to sign this user certificate
     user.crt: <user_certificate> # Public key of the user
     user.key: <user_private_key> # Private key of the user
     user.p12: <store> # PKCS #12 store for user certificates and keys
     user.password: <password_for_store> # Protects the PKCS #12 store

3. Extract the cluster CA certificate from the <cluster_name>-cluster-ca-cert secret of the Kafka cluster.

   oc get secret <cluster_name>-cluster-ca-cert -o jsonpath='{.data.ca\.crt}' | base64 -d > ca.crt

4. Extract the user CA certificate from the <user_name> secret.

   oc get secret <user_name> -o jsonpath='{.data.user\.crt}' | base64 -d > user.crt

5. Extract the private key of the user from the <user_name> secret.

   oc get secret <user_name> -o jsonpath='{.data.user\.key}' | base64 -d > user.key

6. Configure your client with the bootstrap address hostname and port for connecting to the Kafka cluster:

   props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "<hostname>:<port>");

7. Configure your client with the truststore credentials to verify the identity of the Kafka cluster.

   Specify the public cluster CA certificate.

   Example truststore configuration

   props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
   props.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, "PEM");
   props.put(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG, "<ca.crt_file_content>");

   SSL is the specified security protocol for mTLS authentication. Specify SASL_SSL for SCRAM-SHA-512 authentication over TLS. PEM is the file format of the truststore.

8. Configure your client with the keystore credentials to verify the user when connecting to the Kafka cluster.

   Specify the public certificate and private key.

   Example keystore configuration

   props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
   props.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, "PEM");
   props.put(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG, "<user.crt_file_content>");
   props.put(SslConfigs.SSL_KEYSTORE_KEY_CONFIG, "<user.key_file_content>");

   Add the keystore certificate and the private key directly to the configuration. Add as a single-line format. Between the BEGIN CERTIFICATE and END CERTIFICATE delimiters, start with a newline character (\n). End each line from the original certificate with \n too.

   Example keystore configuration

   props.put(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG, "-----BEGIN CERTIFICATE----- \n<user_certificate_content_line_1>\n<user_certificate_content_line_n>\n-----END CERTIFICATE---");
   props.put(SslConfigs.SSL_KEYSTORE_KEY_CONFIG, "----BEGIN PRIVATE KEY-----\n<user_key_content_line_1>\n<user_key_content_line_n>\n-----END PRIVATE KEY-----");