在 kafka 的 broker 和 client 之间加入 SSL 双向认证

单向认证

首先实现单向的认证，即，client 对 broker 的认证。就像浏览器对服务器的认证一样。
注：密钥生成过程中的密码统一为123456

为broker生成keystore

第一步是为集群的每台机器生成密钥和证书，可以使用java的keytool来生产。我们将生成密钥到一个临时的密钥库，之后我们可以导出并用CA签名它。
keytool 使用可以参考java中Keytool的使用总结

keytool -genkey -keystore server.keystore.jks -alias localhost -validity 365 -keyalg RSA 

生成CA

通过第一步，集群中的每台机器都生成了一个证书来识别机器。但是，证书是未签名的，这意味着攻击者可以创建一个这样的证书来假装成任何机器。因此需要对这些证书进行签名来防止伪造。
在浏览网站的时候，使用HTTPS的网站的证书是由受信任的第三方颁发的，那么这里可以模拟一个假的CA中心，只要支持签名即可。
下面使用 OpenSSL 来生成一个私钥和一个证书，有效期为365天

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

然后会生成 ca-key 和 ca-cert 两个文件。

ca-key：模拟CA的私钥
ca-cert：模拟CA的证书
签名

用上一步生成的 CA 来签名第一步生成的证书，证书哪里来？从第一步生成的 server.keystore.jks 里面导出来。
导出证书，命名为 server.crt。

keytool -keystore server.keystore.jks -alias localhost -certreq -file server.crt

然后用刚才在第二步生成的CA的私钥对刚才导出的证书签名，签名后的证书命名为 server-signed.crt。

openssl x509 -req -CA ca-cert -CAkey ca-key -in server.crt -out server-signed.crt -days 365 -CAcreateserial -passin pass:test123

命令成功执行后，会生成两个文件server-signed.crt和 ca-cert.srl，ca-cert.srl 是 ca-cert的序列号文件，由-CAcreateserial参数生成的
最后，将CA的证书和已经签名的证书导入秘钥仓库：

keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert
keytool -keystore server.keystore.jks -alias localhost -import -file server-signed.crt

使 client 信任broker

为了使 client 信任broker，就需要将broker端的证书加入到客户端信任库里面，那么，按照道理来说，想让client信任所有的broker，那么就需要将所有的broker的证书添加到client的信任列表里面，那么就增加了不好操作性，比如说我新增了一个broker，难道还要挨个通知client增加信任吗？当然不可能。
证书签名体系中，有一个信任链的概念，就是说，如果使用证书A签名了证书B、C、D，那么只要信任了 A，就会信任B、C、D。这样来说就好办了，只要信任CA的证书，就可以信任所有它签名的证书

keytool -keystore client.truststore.jks -alias CARoot -import -file ca-cert

启动broker

在 config/server.properties 文件里面修改如下项：
listeners如果只支持SSL的话，就需要把 security.inter.broker.protocol 也配置为 SSL，即内部交流方式也为SSL。

listeners=PLAINTEXT://192.168.3.146:6667,SSL://192.168.3.146:6668
ssl.endpoint.identification.algorithm= #解决dns匹配问题 忽略对证书主机ip匹配
ssl.key.password=123456
ssl.keystore.location=/usr/hdp/3.1.4.0-315/kafka/conf/cert/server.keystore.jks
ssl.keystore.password=123456

查看日志，无错为启动成功。

启动客户端

这一步，需要把上面生成的 client.truststore.jks 放到客户端侧。
我使用的是java客户端，以消费者为例，连接broker的配置需要额外添加一下几行：

props.put("ssl.endpoint.identification.algorithm", "");
props.put("security.protocol", "SSL");
props.put("ssl.truststore.location", "E:\\cert\\client.truststore.jks");
props.put("ssl.truststore.password", "123456");

验证结果

经过测试，结果正确。
到这里，就可以实现client对broker的认证，如果是任意的一个broker，client是会拒绝连接的。但是，在kafka应用里面，好像并不是很实用，除了数据的加密。更多情况下，broker要防止任意的client进行连接，不能让client随意连接，那么就需要增加broker对client的认证。
其实原理和上面的是一样的，就是为client生成证书，然后让broker信任。增加broker对client的认证

增加broker对client的认证

首先生成各种证书

#为client生成证书
keytool -genkey -keystore client.keystore.jks -alias localhost -validity 365 -keyalg RSA
#为client生成CA
openssl req -new -x509 -keyout ca-key -out ca-cert -days 365
#导出证书
keytool -keystore client.keystore.jks -alias localhost -certreq -file client.crt
#签名
openssl x509 -req -CA ca-cert -CAkey ca-key -in client.crt -out client-signed.crt -days 365 -CAcreateserial -passin pass:123456
#导入
keytool -keystore client.keystore.jks -alias CARoot -import -file ca-cert
keytool -keystore client.keystore.jks -alias localhost -import -file client-signed.crt

keytool -keystore server.truststore.jks -alias CARoot -import -file ca-cert

配置broker

ssl.client.auth=required
ssl.truststore.location=/usr/hdp/3.1.4.0-315/kafka/conf/cert/server.truststore.jks
ssl.truststore.password=123456

启动。

配置client

在client端，增加连接参数：

props.put("ssl.key.password", "123456");
props.put("ssl.keystore.location", "E:\\cert\\client.keystore.jks");
props.put("ssl.keystore.password", "123456");

启动client

我以消费者为例，看是否能够连接成功，如果能够连接成功，那就也就意味着配置成功了。
幸运的，连接成功了，模拟发送数据也能够接收到。

总结

没啥可总结的，把配置合在一起贴一下吧。

broker 配置 server.properties：

listeners=PLAINTEXT://192.168.3.146:6667,SSL://192.168.3.146:6668
ssl.client.auth=required
ssl.endpoint.identification.algorithm=
ssl.key.password=123456
ssl.keystore.location=/usr/hdp/3.1.4.0-315/kafka/conf/cert/server.keystore.jks
ssl.keystore.password=123456
ssl.truststore.location=/usr/hdp/3.1.4.0-315/kafka/conf/cert/server.truststore.jks
ssl.truststore.password=123456

client

props.put("security.protocol", "SSL");
props.put("ssl.truststore.location", "E:\\cert\\client.truststore.jks");
props.put("ssl.truststore.password", "123456");
props.put("ssl.key.password", "123456");
props.put("ssl.keystore.location", "E:\\cert\\client.keystore.jks");
props.put("ssl.keystore.password", "123456");
props.put("ssl.endpoint.identification.algorithm", "");

整体执行的shell命令 重新整理

keytool -genkey -keystore server.keystore.jks -alias localhost -validity 3650 -keyalg RSA   -storepass 123456 -keypass 123456 -dname "CN=MyName, OU=MyOrgUnit, O=MyOrg, L=MyCity, ST=MyState, C=MyCountry"
openssl req -new -x509 -days 3650 -keyout ca-key.pem -out ca-cert.pem -passout pass:123456
keytool -keystore server.keystore.jks -alias localhost -certreq -file server.crt -storepass 123456 -keypass 123456
openssl x509 -req -CA ca-cert -CAkey ca-key -in server.crt -out server-signed.crt -days 3650 -CAcreateserial -passin pass:123456
openssl x509 -req -CA ca-cert.pem -CAkey ca-key.pem -in server.crt -out server-signed.crt -days 3650 -CAcreateserial -passin pass:123456
keytool -keystore server.keystore.jks -alias CARoot -import -file ca-cert.pem -storepass 123456 -keypass 123456
keytool -keystore server.keystore.jks -alias localhost -import -file server-signed.crt -storepass 123456 -keypass 123456
keytool -keystore client.truststore.jks -alias CARoot -import -file ca-cert.pem -storepass 123456 -keypass 123456

keytool -genkey -keystore client.keystore.jks -alias localhost -validity 3650 -keyalg RSA -storepass 123456 -keypass 123456 -dname "CN=MyName, OU=MyOrgUnit, O=MyOrg, L=MyCity, ST=MyState, C=MyCountry"
keytool -keystore client.keystore.jks -alias localhost -certreq -file client.crt -storepass 123456 -keypass 123456
openssl x509 -req -CA ca-cert -CAkey ca-key -in client.crt -out client-signed.crt -days 3650 -CAcreateserial -passin pass:123456
openssl x509 -req -CA ca-cert.pem -CAkey ca-key.pem -in client.crt -out client-signed.crt -days 3650 -CAcreateserial -passin pass:123456
keytool -keystore client.keystore.jks -alias CARoot -import -file ca-cert.pem -storepass 123456 -keypass 123456
keytool -keystore client.keystore.jks -alias localhost -import -file client-signed.crt -storepass 123456 -keypass 123456
keytool -keystore server.truststore.jks -alias CARoot -import -file ca-cert.pem -storepass 123456 -keypass 123456