参考：https://kafka.apache.org/documentation/#security

单向认证

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

为broke生成keystore

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

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

可以使用

keytool -list -v -keystore server.keystore.jks

来查看证书的信息。
可以看到这个仓库里面有一个刚才生成的证书，不过，它包含了哪些含义呢？待查？？？？？？？？？？？？？？

生成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 信任broke

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

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

启动broke

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

listeners=PLAINTEXT://10.40.20.61:9092,SSL://10.40.20.61:9093
ssl.keystore.location=/usr/local/kafka_2.11-0.10.1.0/serverssl/server.keystore.jks
ssl.keystore.password=test123
ssl.key.password=test123

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

启动客户端

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

        props.put("security.protocol", "SSL");
        props.put("ssl.truststore.location", "D:\\client.truststore.jks");
        props.put("ssl.truststore.password", "test123");

不管是什么客户端，也就是增加这几个参数，这也是最小配置

        security.protocol=SSL
        ssl.truststore.location=/var/private/ssl/kafka.client.truststore.jks
        ssl.truststore.password=test1234

验证结果

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

增加broke对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:test123
#导入
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

配置broke

ssl.truststore.location=/usr/local/kafka_2.11-0.10.1.0/clientssl/server.truststore.jks
ssl.truststore.password=test123

ssl.client.auth=required

启动。

配置client

在client端，增加连接参数：

ssl.keystore.location=/var/private/ssl/kafka.client.keystore.jks
ssl.keystore.password=test123
ssl.key.password=test123

那么，我这个java客户端，就可以这么添加：

        props.put("ssl.keystore.location", "D:\\client.keystore.jks");
        props.put("ssl.keystore.password", "test123");
        props.put("ssl.key.password", "test123");

启动client

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

总结

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

broke 配置 server.properties：

listeners=PLAINTEXT://10.40.20.61:9092,SSL://10.40.20.61:9093

ssl.client.auth=required

ssl.keystore.location=/usr/local/kafka_2.11-0.10.1.0/serverssl/server.keystore.jks
ssl.keystore.password=test123
ssl.key.password=test123
ssl.truststore.location=/usr/local/kafka_2.11-0.10.1.0/clientssl/server.truststore.jks
ssl.truststore.password=test123

client 连接参数：

        props.put("bootstrap.servers", "10.40.20.61:9093");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        props.put("security.protocol", "SSL");
        props.put("ssl.truststore.location", "D:\\client.truststore.jks");
        props.put("ssl.truststore.password", "test123");

        props.put("ssl.keystore.location", "D:\\client.keystore.jks");
        props.put("ssl.keystore.password", "test123");
        props.put("ssl.key.password", "test123");