Kafka 协议模式介绍

        kafka 协议模式有四种：

                PLAINTEXT:                 明文不加密，建议用于开发，测试

                SSL:                              对连接链路进行加密

                SASL_PLAINTEXT:      对身份认证，但是对数据不加密

                SASL_SSL:                  即对身份认证，也对数据加密，生产建议用此方法

        此篇文件专门介绍了SSL和SASL_SSL两种模式的配置 。

        SASL_SSL 介绍时用到kerberos进行身份认证校验，此篇文件默认已经有了kerberos环境，对kerberos进行使用即可，不做其他详细kerberos介绍，我这边后面会单独出一篇文件详细介绍kerberos。

Kafka SSL 配置

        用java工具生成自签证书，这里提供一个ssl.sh自动生产证书的基本

cat ssl.sh

#!/bin/bash
#define
PASSWD=123456
HOSTNAME=wl-kafka-ssl
#precondition

if [ -d "ssl/" ];then
   rm -rf ssl/
   echo "delete already exists ssl/"
fi
mkdir ssl/
echo "create ssl/"


#Generate SSL key and certificate for Kafka broker
keytool -keystore ssl/server.keystore.jks -alias alias-${HOSTNAME} -validity 3650 -genkey -keypass ${PASSWD} -keyalg RSA -dname "CN=${HOSTNAME},OU=aspire,O=aspire,L=beijing,S=beijing,C=cn" -storepass ${PASSWD} -ext SAN=DNS:${HOSTNAME}

#keytool -list -v -keystore ssl/server.keystore.jks

#Creating your own CA

openssl req -new -x509 -keyout ssl/ca-key -out ssl/ca-cert -days 3650 -passout pass:${PASSWD} -subj "/C=cn/ST=beijing/L=beijing/O=aspire/OU=aspire/CN=${HOSTNAME}"

keytool -keystore ssl/client.truststore.jks -alias CARoot -import -file ssl/ca-cert -storepass ${PASSWD}

keytool -keystore ssl/server.truststore.jks -alias CARoot -import -file ssl/ca-cert -storepass ${PASSWD}

#Signing the server certificate

keytool -keystore ssl/server.keystore.jks -alias alias-${HOSTNAME} -certreq -file ssl/server.cert-file -storepass ${PASSWD}

openssl x509 -req -CA ssl/ca-cert -CAkey ssl/ca-key -in ssl/server.cert-file -out ssl/server.cert-signed -days 365 -CAcreateserial -passin pass:${PASSWD}

keytool -keystore ssl/server.keystore.jks -alias CARoot -import -file ssl/ca-cert -storepass ${PASSWD}

keytool -keystore ssl/server.keystore.jks -alias alias-${HOSTNAME} -import -file ssl/server.cert-signed -storepass ${PASSWD}

#Signing the client certificate

keytool -keystore ssl/client.keystore.jks -alias alias-${HOSTNAME} -validity 3650 -genkey -keypass ${PASSWD} -dname "CN=${HOSTNAME},OU=aspire,O=aspire,L=beijing,S=beijing,C=cn" -ext SAN=DNS:${HOSTNAME} -storepass ${PASSWD}

keytool -keystore ssl/client.keystore.jks -alias alias-${HOSTNAME} -certreq -file ssl/client.cert-file -storepass ${PASSWD}

openssl x509 -req -CA ssl/ca-cert -CAkey ssl/ca-key -in ssl/client.cert-file -out ssl/client.cert-signed -days 365 -CAcreateserial -passin pass:${PASSWD}

keytool -keystore ssl/client.keystore.jks -alias CARoot -import -file ssl/ca-cert -storepass ${PASSWD}

keytool -keystore ssl/client.keystore.jks -alias alias-${HOSTNAME} -import -file ssl/client.cert-signed -storepass ${PASSWD}

        kafka 服务端配置

        

#kafka 服务端ssl 配置

#以下这行配置的意思是：Kafka Broker 同时提供明文和SSL加密两种接入方式。明文在 9092 端口，SSL 在 9093 端口
listeners=PLAINTEXT://192.168.1.73:9092,SSL://192.168.1.73:9093

#kafka  ssl  config
ssl.keystore.location=/data/app/pats_opts/kafka/config/kerberos/server.keystore.jks
ssl.keystore.password=Welcome1234%
ssl.key.password=123456
ssl.truststore.location=/data/app/pats_opts/kafka/config/kerberos /server.truststore.jks
ssl.truststore.password=123456
ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1
ssl.client.auth=required

        kafka 客户端配置（这里以java程序为例）

#java 程序连接kafka ssl 配置信息
security.protocol=SSL
ssl.truststore.location=/opt/bigdata/kafka_2.12-2.7.0/ssl/client.truststore.jks
ssl.truststore.password= 123456
ssl.keystore.location=/opt/bigdata/kafka_2.12-2.7.0/ssl/client.keystore.jks
ssl.keystore.password= 123456
ssl.key.password= 123456

Kafka 之SASL_SSL 配置

        SASL我这里用到的是kerberos服务，申请步骤如下：

        Kerberos 票据申请

                登录到 kerberos 服务器输入命令：

                        kadmin.local

                        addprinc  kafka/192.168.1.73 创建用户名，需要输入密码，此处密码为：123456

                        ​​​​​​​        

                生成keytab文件

                     xst -k /root/kafka.73.keytab kafka/192.168.1.73@BOS.COM   

                初始化：kinit -kt /root/kafka.73.keytab    kafka/192.168.1.73@BOS.COM

                          kafka 服务端配置                        

                        在kafka的配置目录下创建kerberos 文件夹

                        cd /data/app/pats_opts/kafka/config && mkdir kerberos

                        将kerberos 服务器上kakfa.73.keytab、krb5.conf 拷贝到kerberos 目录下

       

        并创建kafka-jaas.conf 文件

        

#kafka-jass.conf 文件内容如下
more kafka-jaas.conf 
KafkaServer {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   serviceName="kafka"
   keyTab="/data/app/pats_opts/kafka/config/kerberos/kafka.73.keytab"
   storeKey=true
   useTicketCache=false
   principal="kafka/192.168.1.73@BOS.COM";
};
KafkaClient {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   serviceName="kafka"
   keyTab="/data/app/pats_opts/kafka/config/kerberos/kafka.73.keytab"
   storeKey=true
   useTicketCache=false
   principal="kafka/192.168.1.73@BOS.COM";
};

        kafka bin目录下的kafka-server-start.sh 启动脚本文件增加一行配置，配置如下：

export	KAFKA_OPTS="-Dzookeeper.sasl.client=false -Djava.security.krb5.conf=/data/app/pats_opts/kafka/config/kerberos/krb5.conf -Djava.security.auth.login.config=/data/app/pats_opts/kafka/config/kerberos/kafka-jaas.conf -Dsun.security.
krb5.debug=true"

                kafka 配置文件修改

                

listeners=PLAINTEXT://:9092,SASL_SSL://:9093
advertised.listeners=PLAINTEXT://192.168.1.73:9092,SASL_SSL://192.168.1.73:9093
group.initial.rebalance.delay.ms=0
advertised.host.name=192.168.1.73

#ssl config
listeners=PLAINTEXT://:9092,SASL_SSL://:9093
advertised.listeners=PLAINTEXT://192.168.1.73:9092,SASL_SSL://192.168.1.73:9093
inter.broker.listener.name=PLAINTEXT
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SASL_SSL:SASL_SSL
#ssl.client.auth=required
ssl.keystore.location=/data/app/pats_opts/kafka/config/kerberos/server.keystore.jks
ssl.keystore.password=123456
ssl.key.password=123456
ssl.truststore.location=/data/app/pats_opts/kafka/config/kerberos/server.truststore.jks
ssl.truststore.type=JKS
ssl.truststore.password=123456
ssl.keystore.type=JKS
ssl.endpoint.identification.algorithm=
#security.protocol=SASL_SSL
#ssl.protocol=TLSv1.2
#ssl.enabled.protocols=TLSv1.2
#ssl.endpoint.identification1.98rithm=
#serviceName=kafka
#sasl_ssl config
sasl.mechanism.inter.broker.protocol=GSSAPI
sasl.enabled.mechanisms=GSSAPI
sasl.kerberos.service.name=kafka