一、目的

配置SASL/PLAIN驗證，實現了對Kafka的權限控制。但SASL/PLAIN驗證有一個問題：只能在JAAS文件KafkaServer中配置用戶，一但Kafka啟動，無法動態新增用戶。SASL/SCRAM驗證可以動態新增用戶并分配權限

1.1啟動Zookeeper和Kafka

版本：kafka_2.12-2.7.0.tgz
此方法是把憑證(credential)存儲在Zookeeper，可以使用kafka-configs.sh在Zookeeper中創建憑據。對于每個SCRAM機制，必須添加具有機制名稱的配置來創建憑證，在啟動Kafka broker之前創建代理間通信的憑據。
所以第一步，在沒有設置任何權限的配置下啟動Kafka和Zookeeper。

1.2 創建SCRAM證書

1）創建broker建通信用戶：admin(在使用sasl之前必須先創建,否則啟動報錯)

bin/kafka-configs.sh --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

2）創建生產用戶：producer

bin/kafka-configs.sh --zookeeper 127.0.0.1:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=prod-sec], SCRAM-SHA-512=[password=prod-sec]' --entity-type users --entity-name producer

3）創建消費用戶：consumer

bin/kafka-configs.sh --zookeeper 127.0.0.1:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=cons-sec], SCRAM-SHA-512=[password=cons-sec]' --entity-type users --entity-name consumer

SCRAM-SHA-256/SCRAM-SHA-512是對密碼加密的算法，二者有其一即可

1.3查看SCRAM證書

bin/kafka-configs.sh --zookeeper localhost:2181 --describe --entity-type users --entity-name consumer bin/kafka-configs.sh --zookeeper localhost:2181 --describe --entity-type users --entity-name producer

1.4刪除SCRAM證書

bin/kafka-configs.sh --zookeeper localhost:2181 --alter --delete-config 'SCRAM-SHA-512' --delete-config 'SCRAM-SHA-256' --entity-type users --entity-name producer

1.5服務端配置

在用戶證書創建完畢之后開始Kafka服務端的配置：
1）創建JAAS文件：

KafkaServer { org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-sec"; };

2）將JAAS配置文件位置作為JVM參數傳遞給每個Kafka Broker【bin/kafka-server-start.sh】添加
-Djava.security.auth.login.config
=/home/test/kiki/kafka/ka/config/kafka_server_jaas.conf

exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config =/home/test/kiki/kafka/ka/config/kafka_server_jaas.conf kafka.Kafka "$@"

3）配置server.properties【config/server.properties】

#認證配置 listeners=SASL_PLAINTEXT://:9092 security.inter.broker.protocol=SASL_PLAINTEXT sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256 sasl.enabled.mechanisms=SCRAM-SHA-256 #ACL配置 allow.everyone.if.no.acl.found=false super.users=User:admin authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

可以根據自己的需求選擇SASL_SSL或SASL_PLAINTEXT, PLAINTEXT為不加密明文傳輸，性能好與SSL
4）重啟Kafka和Zookeeper

1.6客戶端配置

1）為我們創建的三個用戶分別創建三個JAAS文件:分別命名為
kafka_client_scram_admin_jaas.conf
kafka_client_scram_producer_jaas.conf
kafka_client_scram_consumer_jaas.conf

KafkaClient { org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-sec"; };KafkaClient { org.apache.kafka.common.security.scram.ScramLoginModule required username="consumer" password="cons-sec"; };KafkaClient { org.apache.kafka.common.security.scram.ScramLoginModule required username="producer" password="prod-sec"; };

2）修改啟動腳本引入JAAS文件：
生產者配置：
配置bin/kafka-console-producer.sh

exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config =/home/test/kiki/kafka/ka/config/kafka_client_scram_producer_jaas.conf

消費者配置：
配置bin/kafka-console-consumer.sh

exec $(dirname $0)/kafka-run-class.sh -Djava.security.auth.login.config =/home/test/kiki/kafka/ka/config/kafka_client_scram_consumer_jaas.conf

3)配置consumer.properties和producer.properties,都要加入以下配置

security.protocol=SASL_PLAINTEXT sasl.mechanism=SCRAM-SHA-256

4）創建主題

[test@police ka]$ bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic test --partitions 2 --replication-factor 1

5)啟動生產

[test@police ka]$ bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic test --producer.config config/producer.properties

發現會報權限相關的錯
6）對生產者賦予寫的權限

bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add--allow-principal User:producer --operation Write

查看權限：

bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --list

7)對消費者賦予讀的權限

bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add--allow-principal User:consumer --operation Read

此時啟動消費者

[test@police ka]$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning --consumer.config config/consumer.properties

此時依舊會報錯，報未對消費者組授權

bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:consumer --operation Read --group test-group

此時再啟動消費者，可以發現能正常消費生產者的消息
8）查看權限

bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --list

1.7總結

SASL/SCRAM驗證方法可以在Kafka服務啟動之后，動態的新增用戶分并配權限，在業務變動頻繁，開發人員多的情況下比SASL/PLAIN方法更加靈活。
參考：
https://cloud.tencent.com/developer/article/1491674
https://cloud.tencent.com/developer/article/1588581

總結

以上是生活随笔為你收集整理的Kafka动态认证SASL/SCRAM验证的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。