Kafka 认证机制
认证：认证要解决的是你要证明你是谁的问题
授权：要解决的则是你能做什么的问题
A 主要机制
1） 基于SSL 和 SASL的安全认证机制
2） 基于SSL认证主要是指Broker 和客户端的双路认证（2-way authentication ）,SSL 加密已经启用了客户端向Broker的证书，双路认证就是Broker 也要认证客户端的证书。
3）SASL做客户端认证，SASL是提供认证和数据安全服务的框架。 OAUTHBEARER：是基于 OAuth 2 认证框架的新机制
B 机制的建议
可以使用 SSL 来做通信加密，使用 SASL 来做 Kafka 的认证实现
C： 配置实例
流程：
用户创建
参数配置
1） 创建用户
admin 用户用于实现 Broker 间通信，
writer 用户用于生产消息，
reader 用户用于消费消息
a> 创建命令
$ cd kafka_2.12-2.3.0/
$ bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[password=admin],SCRAM-SHA-512=[password=admin]' --entity-type users --entity-name admin
Completed Updating config for entity: user-principal 'admin'.
$ bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[password=writer],SCRAM-SHA-512=[password=writer]' --entity-type users --entity-name writer
Completed Updating config for entity: user-principal 'writer'.
$ bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[password=reader],SCRAM-SHA-512=[password=reader]' --entity-type users --entity-name reader
Completed Updating config for entity: user-principal 'reader'.
b> 查看创建命令
$ bin/kafka-configs.sh --zookeeper localhost:2181 --describe --entity-type users --entity-name writer
2) 创建JAAS文件
文件名：kafka-broker.jaas
KafkaServer {
org.apache.kafka.common.security.scram.ScramLoginModule required
username="admin"
password="admin";
};
tips: 两个分号，文件中不能有空格
3）配置Broker的server.properties
开启 SCRAM 认证机制，并启用 SHA-256 算法
sasl.enabled.mechanisms=SCRAM-SHA-256
Broker 间通信也开启 SCRAM 认证，同样使用 SHA-256 算法
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
Broker 间通信不配置 SSL
security.inter.broker.protocol=SASL_PLAINTEXT
isteners 使用 SASL_PLAINTEXT
listeners=SASL_PLAINTEXT://localhost:9092
4） 启动Broker
启动第1台
$KAFKA_OPTS=-Djava.security.auth.login.config=<your_path>/kafka-broker.jaas bin/kafka-server-start.sh config/server1.properties
......
[2019-07-02 13:30:34,822] INFO Kafka commitId: fc1aaa116b661c8a (org.apache.kafka.common.utils.AppInfoParser)
[2019-07-02 13:30:34,822] INFO Kafka startTimeMs: 1562045434820 (org.apache.kafka.common.utils.AppInfoParser)
[2019-07-02 13:30:34,823] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)
启动第2台
$KAFKA_OPTS=-Djava.security.auth.login.config=<your_path>/kafka-broker.jaas bin/kafka-server-start.sh config/server2.properties
......
[2019-07-02 13:32:31,976] INFO Kafka commitId: fc1aaa116b661c8a (org.apache.kafka.common.utils.AppInfoParser)
[2019-07-02 13:32:31,976] INFO Kafka startTimeMs: 1562045551973 (org.apache.kafka.common.utils.AppInfoParser)
[2019-07-02 13:32:31,978] INFO [KafkaServer id=1] started (kafka.server.KafkaServer)
5） 发送消息
用kafka-console-producer 脚本来发送消息，由于启用了认证，客户端需要做一些相应的配置。新建producer.conf 内容如下：
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="writer" password="writer";
运行脚本：
$ bin/kafka-console-producer.sh --broker-list localhost:9092,localhost:9093 --topic test --producer.config <your_path>/producer.conf
>hello, world //注释输入消息
>
6 ） 消费消息
用 kafka-console-consumer 脚本创建一个名 consumer.conf
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="reader" password="reader";
运行console Consumer 程序
$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092,localhost:9093 --topic test --from-beginning --consumer.config <your_path>/consumer.conf
hello, world
消息消费成功
7） 动态增减用户
a> 删除writer 用户
$ bin/kafka-configs.sh --zookeeper localhost:2181 --alter --delete-config 'SCRAM-SHA-256' --entity-type users --entity-name writer
Completed Updating config for entity: user-principal 'writer'.
$ bin/kafka-configs.sh --zookeeper localhost:2181 --alter --delete-config 'SCRAM-SHA-512' --entity-type users --entity-name writer
Completed Updating config for entity: user-principal 'writer'.
b> 添加new_writer用户
$ bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=new_writer]' --entity-type users --entity-name new_writer
Completed Updating config for entity: user-principal 'new_writer'.
如果再使用writer用户发消息就会报错，修改为new_writer 恢复正常
$ bin/kafka-console-producer.sh --broker-list localhost:9092,localhost:9093 --topic test --producer.config /Users/huxi/testenv/producer.conf
>[2019-07-02 13:54:29,695] ERROR [Producer clientId=console-producer] Connection to node -1 (localhost/127.0.0.1:9092) failed authentication due to: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-256 (org.apache.kafka.clients.NetworkClient)