15.4. Example: Setting up secure client access

Configure the Kafka cluster with a Kafka listener.

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.

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.

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

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.

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

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

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

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

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

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

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

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

Specify the public cluster CA certificate.

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.

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

Specify the public certificate and private key.

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.

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-----");