一、生成CA证书

1  生成根证书的私钥
openssl genrsa -out ./ca.key
 
2  利用私钥生成一个根证书的申请，一般证书的申请格式都是csr。所以私钥和csr一般需要保存好
openssl req -new -key ca.key -out ca.csr
 
3  自签名的方式签发我们之前的申请的证书，生成的证书为ca.crt
openssl x509 -req -days 3650 -in ca.csr -signkey ca.key -out ca.crt

4 为我们的证书建立第一个序列号，一般都是用4个字符，这个不影响之后的证书颁发等操作

echo FACE（01） > $path/serial。

5 建立ca的证书库，不影响后面的操作，默认配置文件里也有存储的地方。

touch $path/index.txt

6 建立证书回收列表保存失效的证书

openssl ca -gencrl -out $path/ca.crl -crldays 7

$path 为openssl配置文件设置的路径，第4,5,6步可在此不操作，等后续签发证书报错时候根据报错内容添加

二、生成服务端证书

# 建立服务器验证证书的私钥
openssl genrsa -out server.key
 
# 生成证书申请文件，制作签发证书的请求文件时，需要输入Common Name参数，此参数一定为当前主机的IP地址，否则将会显示证书错误。
openssl req -new -key server.key -subj "/CN=本机IP" -out server.csr
# CN=后为本机IP
 
echo subjectAltName = IP:"本机IP" > extfile.cnf
 
# 利用根证书签发服务器身份验证证书
openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key -CAcreateserial -extfile extfile.cnf -out server.crt -days 5000

三、生成客户端证书

# 生成私钥
openssl genrsa  -out client.key 1024
 
# 生成证书请求文件
openssl req -new -key client.key -out  client.csr

# 签发证书
openssl ca -in client.csr -cert ca.crt -keyfile ca.key -out client.crt

四、通过CA证书生成kafka端可信任证书和filebeat端可信任证书

# 生成filebeat端可信任证书
cat ca.crt >  client/filebeat.cert.pem

# 生成kafka端可信任证书
keytool -importcert -file ca.crt -alias ca -noprompt -storepass 123456 -keystore server/kafka_trustchain.jks

五、生成kafka端keystore文件

# create pkcs12 store
openssl pkcs12 -export -out server/localhost.pkcs12 -in server/server.crt -inkey server/server.key -passout "pass:123456"

# import server key into jks keystore
echo "123456" | keytool -importkeystore -srckeystore server/localhost.pkcs12 -destkeystore server/localhost.jks -srcstoretype pkcs12 -storepass "123456"

六、kafka配置

listeners=SSL://192.168.244.145:9092
advertised.listeners=SSL://192.168.244.140:9092

ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1
ssl.keystore.type=JKS
ssl.truststore.type=JKS

security.inter.broker.protocol=SSL
# ssl.endpoint.identification.algorithm=

ssl.keystore.location=/root/kafka/kafka_2.11-2.4.0/ca/server/localhost.jks
ssl.keystore.password=123456
ssl.key.password=123456
ssl.truststore.location=/root/kafka/kafka_2.11-2.4.0/ca/server/kafka_trustchain.jks
ssl.truststore.password=123456
ssl.client.auth=required

七、验证证书是否配置正确

openssl s_client -connect 192.168.244.145:9092 -tls1_2 -CAfile ca/client/filebeat.cert.pem

       # 出现以下提示，说明验证成功

八、filebeat SSL配置

# 将第三步与第四步生成的客户端证书与filebeat认证证书copy到filebeat所在服务器端

# filebeat输出到kafka关于ssl配置的部分

ssl.enabled: true
ssl.key: "/home/python/Desktop/filebeat-7.9.2-linux-x86_64/certs/client.key"
ssl.certificate: "/home/python/Desktop/filebeat-7.9.2-linux-x86_64/certs/client.crt"
ssl.certificate_authorities: ["/home/python/Desktop/filebeat-7.9.2-linux-x86_64/certs/trustchain.cert.pem"]

九、创建kafka消费者配置文件，监听filebeat发来的消息

    1、创建消费者配置文件ssl.properties：

security.protocol=SSL
ssl.endpoint.identification.algorithm=
ssl.keystore.location=/root/kafka/kafka_2.11-2.4.0/ca/server/localhost.jks
ssl.keystore.password=123456
ssl.key.password=123456
ssl.truststore.location=/root/kafka/kafka_2.11-2.4.0/ca/server/kafka_trustchain.jks
ssl.truststore.password=123456
ssl.client.auth=required

     2、启动消费者

./bin/kafka-console-consumer.sh --bootstrap-server 192.168.244.145:9092 --topic test_zml --consumer.config ssl.properties

十、结果呈现

        1、启动filebeat

        2、filebeat发送消息，此时能正常发送到kafka，结果如图所示，说明ssl配置成功：

   

        3、此时如果filebeat关闭ssl，此时发送消息则如图所示，显示消息一直未发送成功，该步骤证明我们配置ssl起了作用

        

十一：参考文档：

       TLS For Filebeat To Kafka：https://discuss.elastic.co/t/tls-for-filebeat-kafka-output/58756/3

       SSL认证之相关证书的生成: https://blog.csdn.net/justry_deng/article/details/88383081

       Kafka配置SSL认证: https://blog.csdn.net/justry_deng/article/details/88383707

       kafka使用SSL加密和认证: https://www.orchome.com/171