生成证书

首先在本地生成一个 server.keystore.jks 的 jks 证书，有效期为365天。会依次让你键入：密码->重输密码->名与姓->组织单位->组织名->城市->省份->国家两位代码（CN即可）->确认信息（yes或no）->密码->重输密码

建议建立一个 ca 目录来存放这些文件，在该目录下去执行一下命令。

keytool -keystore server.keystore.jks -alias localhost -validity 365 -genkey -keyalg RSA
-keyalg 指定算法为 RSA

生成成功后会有一个警告：JKS 密钥库使用专有格式。建议使用keytool -importkeystore -srckeystore server.keystore.jks -destkeystore server.keystore.jks -deststoretype pkcs12迁移到行业标准格式 PKCS12。可迁移也可以忽略，我这里选择忽略

Warning:
The JKS keystore uses a proprietary format. It is recommended to migrate to PKCS12 which is an industry standard format using 
"keytool -importkeystore -srckeystore server.keystore.jks -destkeystore server.keystore.jks -deststoretype pkcs12".

校验生成证书内容

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

生成CA

通过第一步，集群中的每台机器都生成了一个证书来识别机器。但是，证书是未签名的，这意味着攻击者可以创建一个这样的证书来假装成任何机器。因此需要对这些证书进行签名来防止伪造。

在浏览网站的时候，使用HTTPS的网站的证书是由受信任的第三方颁发的，那么这里可以模拟一个假的CA中心，只要支持签名即可。下面使用 OpenSSL 来生成一个私钥和一个证书，有效期为365天

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

键入命令后会提示：输入密码->校验输入密码->国家两位代码（CN即可）->省份->城市->名与姓->组织名—组织单位->名与姓(域名)->邮箱 ，此输入步骤与上面生成证书世输入步骤相反。

完成后会多出两个文件ca-key 模拟CA的私钥 和 ca-cert 模拟CA的证书。

为了使 client 信任 broker，就需要将 broker 端的证书加入到客户端信任库里面，那么，按照道理来说，想让client信任所有的 broker ，那么就需要将所有的 broker 的证书添加到client的信任列表里面，那么就增加了不好操作性，比如说我新增了一个 broker，难道还要挨个通知client增加信任吗？当然不可能。

证书签名体系中，有一个信任链的概念，就是说，如果使用证书A签名了证书B、C、D，那么只要信任了 A，就会信任B、C、D。这样来说就好办了，只要信任CA的证书，就可以信任所有它签名的证书。

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

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

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

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

# pass:后面输入你CA的密码
openssl x509 -req -CA ca-cert -CAkey ca-key -in server.crt -out server-signed.crt -days 365 -CAcreateserial -passin pass:CAPassword

命令成功执行后，会生成两个文件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

配置KAFKA

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

listeners=PLAINTEXT://xxx.xxx.xx.xxx:9092,SSL://xxx.xxx.xx.xxx:9093
#解决dns匹配问题 忽略对证书主机ip匹配,设置为空可使证书的主机名与kafka的主机名不用保持一致
ssl.endpoint.identification.algorithm= 
# 键入CA 密钥密码
ssl.key.password=
# 键入证书密钥密码
ssl.keystore.password=
# 配置 server.keystore.jks 位置
ssl.keystore.location=
# 生成 server.truststore.jks  的密码
ssl.truststore.password=
# 配置 server.truststore.jks 位置
ssl.truststore.location=/opt/software/kafka/ca/
# 设置 clients 也要开启认证。
ssl.client.auth=required
# kafka broker 内部数据传输协议
security.inter.broker.protocol=SSL

客户端配置

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

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

到这里，就可以实现 client 对 broker 的认证，如果是任意的一个 broker，client 是会拒绝连接的。但是，在kafka应用里面，好像并不是很实用，除了数据的加密。更多情况下，broker 要防止任意的 client 进行连接，不能让 client 随意连接，那么可以增加 broker 对 client 的认证，也可以开启 SASL(账号密码) 配置。

broker 对 client

其实原理和上面的是一样的，就是为 client 生成证书，然后让 broker 信任。首先生成各种证书

#为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
#签名 pass:CA密码
openssl x509 -req -CA ca-cert -CAkey ca-key -in client.crt -out client-signed.crt -days 365 -CAcreateserial -passin pass:CAPassword
#导入
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

配置client，border 不需要改动

# ca密码
props.put("ssl.key.password", "");
# client.keystore.jks 的位置
props.put("ssl.keystore.location", "");
# client.keystore.jks 的密码
props.put("ssl.keystore.password", "");

把以上两个配置合并就组成了 broker 和 client 的双向信任。

SASL/PLAIN进行身份验证

SASL/PLAIN是一种简单的用户名/密码身份验证机制，通常与TLS一起用于加密以实现安全身份验证。Kafka支持SASL/PLAIN的默认实现，可以扩展到生产环境中使用。我们需要向 Kafka 的 config 目录中添加一个JAAS文件。

# 添加文件
vim /opt/software/kafka/config/kafka_server_jaas.conf
# 添加如下配置
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
        username="admin-kafka"
        password="admin-password"
        # 必须有一个用户对应 username 账户 和 password 密码
        user_admin-kafka="admin-password"
        # 其余用户随便设置
        user_test="test123";
};

JAAS文件定义了链接Kafka broker 时所需要的用户名密码及 broker 各个节点之间相互通信的用户名密码，其中：

• username 定义一个公共的用户名，用于节点之间进行通信
• password 做为公共用户名的密码
• user_xxxx 为自定义的用户，主要是客户端用来连接 kafka 的，所有可以使用的用户必须在此定义，不能再之后新增。user_xxxx 中 xxxx 是用户名，等号后面是密码，这里大小写一个字都不能差。

修改kafka启动脚本 kafka-server-start.sh，将 JAAS 配置文件位置作为 JVM 参数传递给 Kafka

# 修改前
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"
# 修改后
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/opt/software/kafka/config/kafka_server_jaas.conf kafka.Kafka "$@"

修改 kafka 配置文件下的 server.properties

# 修改之前
listeners=PLAINTEXT://xxx.xxx.xx.xxx:9092,SSL://xxx.xxxx.xx.xxx:9093
# 修改之后
listeners=SASL_SSL://xxx.xxx.xx.xxx:9092

security.inter.broker.protocol=SASL_SSL
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
allow.everyone.if.no.acl.found=true
# 超级用户设置 对应必须设置的用户
super.users=User:admin-kafka

• allow.everyone.if.no.acl.found=true，整个ACL机制为黑名单机制，即只有黑名单中的用户不能访问资源，非黑名中的用户都可以正常访问 kafka 的资源
• allow.everyone.if.no.acl.found=false，默认为 false，ACL的机制是白名单机制，只有白名单中的用户才能访问 kafka 的资源，其他用户为未授权用户。

配置完毕后 kafka 就可以重新启动了。启动后看到报错，这里其实建议把 Zookeeper 的安全认证也加上，具体和 kafka 差不多，不加不会影响，就会输出个报错而已。

ERROR [ZooKeeperClient ACL authorizer] Auth failed. (kafka.zookeeper.ZooKeeperClient)

客户端代码配置

props.put("security.protocol", "SASL_PLAINTEXT");
props.put("sasl.mechanism", "PLAIN");
props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"test\" password=\"test123\";");