Secure Kafka Cluster

_secure_kafka_cluster_commands

# env
export KAFKA_HOST="my.kafka.hostname"
export KAFKA_OPTS="-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf"

# create topics
kafka-topics --create --topic securing-kafka --replication-factor 1 --partitions 3 --zookeeper $KAFKA_HOST:2181

# producer acl
kafka-acls --authorizer-properties zookeeper.connect=$KAFKA_HOST:2181 --add --allow-principal User:kafkaclient --producer --topic securing-kafka

# consumer acl
kafka-acls --authorizer-properties zookeeper.connect=$KAFKA_HOST:2181 --add --allow-principal User:kafkaclient --consumer --topic securing-kafka --group securing-kafka-group

# start the producer
kafka-console-producer --broker-list $KAFKA_HOST:9093 --topic securing-kafka --producer.config /etc/kafka/producer_ssl.properties

# start the consumer
kafka-console-consumer --bootstrap-server $KAFKA_HOST:9093 --topic securing-kafka --new-consumer --from-beginning --consumer.config /etc/kafka/consumer_ssl.properties

consumer_ssl.properties

bootstrap.servers=my.kafka.hostname:9093
group.id=securing-kafka-group
security.protocol=SSL
ssl.truststore.location=/etc/security/tls/kafka.client.truststore.jks
ssl.truststore.password=test1234
ssl.keystore.location=/etc/security/tls/kafka.client.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234

create_keystores_and_trustores

#!/bin/bash
PASSWORD=test1234
VALIDITY=365
keytool -keystore kafka.server.keystore.jks -alias localhost -validity $VALIDITY -genkey
openssl req -new -x509 -keyout ca-key -out ca-cert -days $VALIDITY
keytool -keystore kafka.server.truststore.jks -alias CARoot -import -file ca-cert
keytool -keystore kafka.client.truststore.jks -alias CARoot -import -file ca-cert
keytool -keystore kafka.server.keystore.jks -alias localhost -certreq -file cert-file
openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days $VALIDITY -CAcreateserial -passin pass:$PASSWORD
keytool -keystore kafka.server.keystore.jks -alias CARoot -import -file ca-cert
keytool -keystore kafka.server.keystore.jks -alias localhost -import -file cert-signed
keytool -keystore kafka.client.keystore.jks -alias localhost -validity $VALIDITY -genkey
keytool -keystore kafka.client.keystore.jks -alias localhost -certreq -file cert-file
openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days $VALIDITY -CAcreateserial -passin pass:$PASSWORD
keytool -keystore kafka.client.keystore.jks -alias CARoot -import -file ca-cert
keytool -keystore kafka.client.keystore.jks -alias localhost -import -file cert-signed

kafka_server_jaas.conf

KafkaServer {
    com.sun.security.auth.module.Krb5LoginModule required
    useKeyTab=true
    storeKey=true
    keyTab="/etc/security/keytabs/kafka.keytab"
    principal="kafka/my.kafka.hostname@kerberos.realm";
};

Client {
    com.sun.security.auth.module.Krb5LoginModule required
    useKeyTab=true
    storeKey=true
    keyTab="/etc/security/keytabs/kafka.keytab"
    principal="kafka/my.kafka.hostname@kerberos.realm";
};

producer_ssl.properties

bootstrap.servers=my.kafka.hostname:9093
security.protocol=SSL
ssl.truststore.location=/etc/security/tls/kafka.client.truststore.jks
ssl.truststore.password=test1234
ssl.keystore.location=/etc/security/tls/kafka.client.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234

server.properties

broker.id=0
listeners=SSL://:9093,SASL_SSL://:9095
security.inter.broker.protocol=SSL
zookeeper.connect=my.kafka.hostname:2181
log.dirs=/var/lib/kafka
zookeeper.set.acl=true
ssl.client.auth=required
ssl.keystore.location=/etc/security/tls/kafka.server.keystore.jks
ssl.keystore.password=test1234
ssl.key.password=test1234
ssl.truststore.location=/etc/security/tls/kafka.server.truststore.jks
ssl.truststore.password=test1234
sasl.kerberos.service.name=kafka
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:CN=my.kafka.hostname,OU=,O=Confluent,L=London,ST=London,C=GB

start-zk-and-kafka

export KAFKA_HEAP_OPTS='-Xmx256M'
export KAFKA_OPTS='-Djava.security.auth.login.config=/etc/kafka/zookeeper_jaas.conf'
/usr/bin/zookeeper-server-start /etc/kafka/zookeeper.properties &
sleep 5
export KAFKA_OPTS='-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf'
/usr/bin/kafka-server-start /etc/kafka/server.properties &

zookeeper.properties

dataDir=/var/lib/zookeeper
clientPort=2181
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000

zookeeper_jaas.conf

Server {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  keyTab="/etc/security/keytabs/zookeeper.keytab"
  storeKey=true
  useTicketCache=false
  principal="zookeeper/my.kafka.hostname@kerberos.realm";
};

Please review all occurrences of my.kafka.hostname and kerberos.realm and replace it with the correct value.