kafka 搭建

1. 先处理好zookeeper 和 kafka，这边就不再赘述了

kafka ssl 证书搭建

1. 依赖环境，安装 openssl 和 keytool 工具

1. SSL 证书原理

1. ssl 加密是基于非对称加密的加密通讯方式，它大致包含两个部分，一部分是验证对方的身份，另一部分是对双方交流的数据进行加密
2. 非对称加密中，最重要的就是公钥和私钥，如果是对称式加密，那么公钥和私钥就是同一个，对称式加密的问题就是，一旦有人知道了密钥，那么加密就会失效。但非对称式加密不同，假设A 和 B 通信，黑客C知道了A的公钥，黑客并不能对数据进行解密，除非它还掌握了私钥，这是因为A用公钥加密，只有使用B的私钥才能对其进行解密。工作模式如下图。
   
3. ssl 还包含了身份验证，这里的身份验证是通过第三方(证书中心)，证书中心(Certificate Authority 简称CA)是一个签发证书的地方，一般在公网的证书都是需要交钱的，所以一般我们都是用自己搭建的证书中心，然后给自己签发证书。
4. 证书是什么呢，证书在我们这里主要是指这个服务的一个身份，且ca证书会将我们上面提到的公钥进行绑定
5. 证书中除了密钥以外，一般还包含了，域名，国家，城市，证书有效期等等
6. 由于证书包含了大量的信息，所以通信时会携带证书进行身份验证，数据加密等操作

openssl 工具介绍

1. 上文提到，我们的最终目的是为了产生一个携带公钥，身份的证书
2. 而证书需要一个证书中心签发
3. 公钥需要先产生一个私钥
4. 且证书为了安全，还会对证书进行二次加密
5. 证书签发过程如下图
   

1. 生成CA

# 生成CA私钥(ca.key)
openssl genrsa -des3 -out ca.key 2048 
# 生成CA证书签名请求(ca.csr)
openssl req -new -key ca.key -out ca.csr
# 生成自签名CA证书(ca.cert)
openssl x509 -req -days 3650 -in ca.csr -signkey ca.key -out ca.crt

2. 生成Server证书

# 生成服务端私钥(server.key)
openssl genrsa -des3 -out server.key 2048 
# 生成服务端证书签名请求(server.csr)
openssl req -new -key server.key -out server.csr
# 使用ca证书签署服务端csr以生成服务端证书(server.cert)
openssl ca -days 3650 -in server.csr -out server.crt -cert ca.crt -keyfile ca.key 

3. 生成Client证书（用于双向认证）

# 生成客户端私钥(client.key)
openssl genrsa -des3 -out client.key 2048
# 生成客户端证书签名请求(client.csr)
openssl req -new -key client.key -out client.csr
# 使用ca证书签署客户端csr以生成客户端证书(client.cert)
openssl ca -days 3650 -in client.csr -out client.crt -cert ca.crt -keyfile ca.key

Keytool 工具介绍

1. keytool 是java自带的证书管理工具，由于kafka 是java编写的，所以kafka环境一般都会有 keytool
2. keystore 是一个证书数据库，他将密钥和证书存放在一起
3. truststore 仅包含一个证书，与keystore 的区别是，keystore 是用于服务器认证客户端，truststore 则相反
4. 证书生成过程

#!/bin/bash
# 生成CA
openssl req -new -x509 -keyout ca-key -out ca-certs -days  7300 -passout pass:ds1994 -subj "/C=cn/ST=shenzhen/L=shenzhen/O=aspire/OU=aspire/CN=kafka-cluster"
# 生成server keystore truststore
keytool -keystore server.keystore.jks -alias ds-kafka-single -validity 7300 -genkey -keypass ds1994 -keyalg RSA -dname "CN=kafka-cluster,OU=aspire,O=aspire,L=shenzhen,S=shenzhen,C=cn" -storepass ds1994
keytool -keystore server.truststore.jks -alias CARoot -import -file ca-certs  -storepass ds1994
keytool -keystore server.keystore.jks -alias ds-kafka-single -certreq -file server.cert-file -storepass ds1994
openssl x509 -req -CA ca-certs -CAkey ca-key -in server.cert-file -out server.cert-singed -days 7300 -CAcreateserial -passin pass:ds1994
keytool -keystore server.keystore.jks -alias CARoot -import -file ca-certs -storepass ds1994
keytool -keystore server.keystore.jks -alias ds-kafka-single -import -file server.cert-singed -storepass ds1994
# 生成client keystore truststore
keytool -keystore client.truststore.jks -alias CARoot -import -file ca-certs  -storepass ds1994
keytool -keystore client.keystore.jks -alias ds-kafka-client -validity 7300 -genkey -keypass ds1994 -dname "CN=kafka-cluster,OU=aspire,O=aspire,L=shenzhen,S=shenzhen,C=cn" -storepass ds1994
keytool -keystore client.keystore.jks -alias ds-kafka-client -certreq -file client.cert-file -storepass ds1994
openssl x509 -req -CA ca-certs -CAkey ca-key -in client.cert-file -out client.cert-signed -days 7300 -CAcreateserial -passin pass:ds1994
keytool -keystore client.keystore.jks -alias CARoot -import -file ca-certs -storepass ds1994
keytool -keystore client.keystore.jks -alias ds-kafka-client -import -file client.cert-signed -storepass ds1994

kafka 配置

ssl.keystore.location=/opt/apps/certs/server.keystore.jks
ssl.keystore.password=ds1994
ssl.key.password=ds1994
ssl.truststore.location=/opt/apps/certs/server.truststore.jks
ssl.truststore.password=ds1994
ssl.ca.location=/opt/apps/certs/ca-certs
ssl.certification.location=/opt/apps/certs/ca-certs
ssl.key.location=/opt/apps/certs/ca-key
# 此项非常重要
ssl.client.auth=none
ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1
ssl.keystore.type=JKS
ssl.truststore.type=JKS
# 此项非常重要
ssl.endpoint.identification.algorithm=
security.inter.broker.protocol=SSL
kafka.security.protocol=SSL
broker.id=0
listeners=PLAINTEXT://1.1.1.1:9093,SSL://1.1.1.1:9092
advertised.listeners=PLAINTEXT://1.1.1.1:9093,SSL://1.1.1.1:9092
num.network.threads=5
num.io.threads=9
socket.send.buffer.bytes=2048000
socket.receive.buffer.bytes=2048000
socket.request.max.bytes=104857600
log.dirs=/data/kafka01,/data/kafka02,/data/kafka03
num.replica.fetchers=2
min.insync.replicas=2
replica.lag.time.max.ms=30000
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=11.1.109.25:2181,11.1.100.100:2181,11.0.59.65:2181
zookeeper.connection.timeout.ms=30000
zookeeper.session.timeout.ms=30000
group.initial.rebalance.delay.ms=3000
log.message.timestamp.type=LogAppendTime

producer 配置

1. 配置详情 test.properties

bootstrap.servers=ip:9092
security.protocol=SSL
ssl.truststore.location=/opt/apps/certs/client.truststore.jks
ssl.truststore.password=ds1994
ssl.keystore.password=ds1994
ssl.keystore.location=/opt/apps/certs/client.keystore.jks
ssl.endpoint.identification.algorithm=

2. bin/kafka-console-producer.sh --broker-list ip:9092 --producer-property test.properties
3. 避坑指南，一定要写 ssl.endpoint.identification.algorithm=, 否则会导致ssl handshake failed

参考文章

1. openssl系列文章
2. 证书中各个字段的详细信息
3. CA证书字段说明
4. openssl 自签发证书
5. openssl 工具介绍
6. kafka 搭建及测试