>最近有跟视频云对接，用到 Kafka 消息队列，发现公司使用了安全认证机制 SASL/SCRAM，所以研究一下这方面的内容。（以前公司好像没有使用安全认证） kafka 提供了多种[安全认证机制](http://kafka.apache.org/documentation/#security)，主要分为 SSL 和 SASL 两大类。 SASL 主要由以下几种方式： 1. SASL/GSSAPI (Kerberos) - starting at version 0.9.0.0 2. SASL/PLAIN - starting at version 0.10.0.0 3. SASL/SCRAM-SHA-256 and SASL/SCRAM-SHA-512 - starting at version 0.10.2.0 4. SASL/OAUTHBEARER - starting at version 2.0 几种方式具体的实现方法可以看看[中文文档](https://kafka.apachecn.org/documentation.html#security_sasl)，里面有详细的介绍，本次主要介绍 SASL/SCRAM 认证。 **SASL/SCRAM验证可以动态新增用户并分配权限。** ## 前期准备 本次主要是在 windows 进行验证测试。 apache-zookeeper-3.7.0 kafka-2.3.0 首先启动 zookeeper 和 kafka ``` ###### 启动 zookeeper D:\work-soft\tool\apache-zookeeper-3.7.0-bin\bin>zkServer.cmd ###### 启动 kafka D:\work-soft\tool\kafka>bin\windows\kafka-server-start.bat config\server.properties ``` ## 创建SCRAM证书 ``` bin\windows\kafka-configs.bat --zookeeper localhost:2181 --alter --add-config SCRAM-SHA-256=[iterations=8192,password=alice-secret],SCRAM-SHA-512=[password=alice-secret] --entity-type users --entity-name alice bin\windows\kafka-configs.bat --zookeeper 127.0.0.1:2181 --alter --add-config SCRAM-SHA-256=[password=admin-sec],SCRAM-SHA-512=[password=admin-sec] --entity-type users --entity-name admin ``` >创建 SCRAM 证书时如果出现 `requirement failed: Unknown Dynamic Configuration: Set('SCRAM-SHA-256).` 报错，可能是因为 --add-config 后面的参数加了单引号 ''，去掉'' 重新执行就OK了。<br> 详情可以查看：https://github.com/confluentinc/confluent-kafka-dotnet/issues/741 查看证书： ``` bin\windows\kafka-configs.bat --zookeeper localhost:2181 --describe --entity-type users --entity-name alice ``` 删除证书： ``` bin\windows\kafka-configs.bat --zookeeper localhost:2181 --alter --delete-config SCRAM-SHA-512 --delete-config SCRAM-SHA-256 --entity-type users --entity-name alice ```   ## 服务端配置 在 kafka 配置文件目录 config 创建文件 kafka-server-jass.conf，如我的目录是：D:\work-soft\tool\kafka\config 文件内容为： ``` KafkaServer { org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-sec"; }; ``` 然后在zookeeper, kafka和consumer、producer 的启动脚本中添加如下脚本： kafka-server-start.bat ： ``` set KAFKA_OPTS=-Djava.security.auth.login.config=D:\work-soft\tool\kafka\config\kafka-server-jass.conf ```  zookeeper 的添加方式是在 zkServer.cmd 里面添加： ``` "-Djava.security.auth.login.config=D:\work-soft\tool\kafka\config\kafka-server-jass.conf" ```  最后在 kafka 配置文件 server.properties 里添加： ``` listeners=SASL_PLAINTEXT://localhost:9020 security.inter.broker.protocol=SASL_PLAINTEXT sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256 sasl.enabled.mechanisms=SCRAM-SHA-256 ``` 完成以上操作后重启 Zookeeper 和 Kafka。 重启后发现在无认证的情况下无法直接连接上 kafka，需要配置相关内容。   ``` security.protocol=SASL_PLAINTEXT sasl.mechanism=SCRAM-SHA-256 sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-sec"; ``` Java application.yml 的配置方式： ``` spring: kafka: bootstrap-servers: localhost:9020 producer: properties: sasl.mechanism: SCRAM-SHA-256 security.protocol: SASL_PLAINTEXT sasl.jaas.config: org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-sec"; ``` 配置完成重新启动应用，发送消息:  使用 Offset Explorer 查看：  End.