In this post, we are going to use 2 different clients to connect the Kafka broker with 2-way SSL. We will use Advantco Kafka adapter as well as Kafka console to produce and consume messages from the broker.

With 2-way SSL, here is how a client requests a resource in an encrypted channel:

1.       A client requests a protected topic from the broker
2.       The broker presents its identity by a certificate
3.       The client verifies the broker certificate
4.       In the other way, the client also sends its certificate to the broker for the verification
5.       The broker verifies the client certificate
6.       If success, the broker grants access to the topic to the client

In this post we just focus on how to setup on client side, and we assume the broker is set up with SSL/TLS and ready to use. We will need this information from the broker for the connectivity:

a.       The port that the broker listening for SSL (SSL://{port} or SASL_SSL://{port}).
b.       The certificate for the server authentication. It could be exported from the truststore (server.truststore.jks) or the keystore (server.keystore.jks). Let say a cluster with multiple brokers and we have a different keystore for each broker, but the CA Root certificate to sign the CSR for each keystore/broker is the same. Let name it as caroot_godaddy.crt for reference later.
c.       The client authentication flag by setting ssl.client.auth has to be “requested” or “required”.

We are using terms “keystore” and “truststore” frequently throughout the article. Let explain it first, “keystore” is a repository/file that stores private-public keys and certificates, and “truststore” is a repository/file that stores only certificates.

Now we have the certificate from the broker (caroot_godaddy.crt) for the server authentication (aka 1-way SSL). For the client authentication (second way SSL), we need a keystore which includes public-private keys and a signed certificate (self-signed or from a trusted CA) and a CA Root certificate that signed the certificate.

Create a keytstore and a CA Root certificate

The CA Root certificate is to sign the certificate signing request (CSR), and we can generate it to sign the certificate (self-signed) or we can export it from a trusted CA.

Let create a keystore for the client.

Create a client keystore

keytool -genkey -keystore client.keystore.jks -validity 3650 -storepass "password1" -keypass "password1" -dname "CN=client1" -alias client1 -storetype pkcs12

Create a client cert sign request (CSR)

keytool -keystore client.keystore.jks -certreq -file client-cert-sign-request -alias client1 -storepass "password1" -keypass "password1"

The CSR file could be signed by a trusted CA or by self-signed.

Here are steps if we want to sign the certificate by ourselves

openssl req -new -x509 -keyout ca-key -out ca-cert -days 3650

openssl x509 -req -CA ca-cert -CAkey ca-key -in client-cert-sign-request -out client-cert-signed -days 3650 -CAcreateserial -passin pass:password1

cat ca-cert client-cert-signed > completeChain.crt

Now we will import the certificate chain to the keystore.

keytool -keystore client.keystore.jks -import -file completeChain.crt -alias client1 -storepass "password1" -keypass "password1" -noprompt

The keystore is ready to use. The last step is to export the CA Root certificate.

If we signed the certificate by ourselves then the ca-cert is the CA Root certificate.

We are using KeyStore Explorer to export the CA Root certificate.

Import the CA Root certificate to Kafka broker

This step should be performed by Kafka team. We send the CA Root certificate file from the previous step and ask them to import it to the broker truststore.

Assume the broker truststore file name is server.truststore.jks

keytool -keystore server.truststore.jks -alias carootsapcpip0520 -import -file carootsapcpip0520.crt -storepass "{password of truststore}" -noprompt

This step is not always required. Like in a corporation we are using the same trusted CA to sign certificates for multiple system including Kafka brokers, then the CA Root certificate is already in the truststore.

Make a connection with Advantco Kafka Adapter

There are different versions for 2 platforms SAP CPI/HCI and PI/PO, but the adapter configuration is identical.

The 2-way SSL can be used for none authentication mode as long as other authentication modes such as Kerberos, Plain, SCRAM..

We have a keystore (client.keystore.jks)and a certificate (caroot_godaddy.crt)from the previous steps, let import it to the Key Storage for referencing in the channel configuration later.

Java KeyStore (JKS) format of the keystore is not supported by PI/PO Key Storage, we have to convert it to P12 format.

Convert JKS to P12

keytool -importkeystore -srckeystore client.keystore.jks -destkeystore client.keystore.p12 -deststoretype pkcs12

Key Storage View SAP_KAFKA_CLIENT

Similar with SAP CPI/HCI, create 2 entries one for Keystore and one for Certificate. The cloud platform supports JKS, so we don’t have to convert the keystore to P12 format.

Here is an example of 2-way SSL with Kerberos

Test the connectivity with Kafka console

The best way to test 2-way SSL is using Kafka console, we don’t have to write any line of code to test it.

Simply download Kafka from Apache Kafka website to the client, it includes kafka-console-producer and kafka-console-consumer in bin directory. We will use one of it to test the connectivity.

To use the console we have to create 2 things:

Wrapper the server certificate (caroot_godaddy.crt) to a client truststore

keytool -keystore client.truststore.jks -alias CARoot -import -file caroot_godaddy.crt -storepass "password1" -noprompt

Create a client property file client-ssl.properties and copy it to kafka bin directory, should have following lines:

security.protocol=SSL
ssl.truststore.location=client.truststore.jks
ssl.truststore.password=password1
ssl.endpoint.identification.algorithm=
ssl.keystore.location=client.keystore.jks
ssl.keystore.password=password1
ssl.key.password=password1

It’s ready to use the kafka console producer to produce a test message to a topic

./bin/kafka-console-producer --broker-list {broker name}:{port} --producer.config client-ssl.properties –topic {topic name}

We are done. Hope you find it useful.