>最近有跟视频云对接,用到 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.

Kafka的安全认证机制-SASL/SCRAM验证