一、目的

配置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验证相关推荐

Kafka动态认证SASL/SCRAM配置+整合springboot配置
记录: zookeeper启动命令: [root@master-yzjgxh2571705819-1651919082731-99-0727183 bin]# ./zkServer.sh start ...
Kafka安全认证SASL/PLAINTEXT，账号密码认证
环境centos7 kafka集群和zookeeper集群默认都是不带用户密码的. 1. 配置zookeeper集群SASL zookeeper所有节点都是对等的,只是各个节点角色可能不相同.以下步骤 ...
Kafka安装配置（SASL/SCRAM动态认证）
SASL/SCRAM验证方法可以在Kafka服务启动之后,动态的新增用户分并配权限,在业务变动频繁,开发人员多的情况下比SASL/PLAIN方法更加灵活. Zookeeper:3.4.13,kafka ...
Kafka SASL/SCRAM动态认证集群部署
Kafka SASL/SCRAM动态认证集群部署目的:配置SASL/PLAIN验证,实现了对Kafka的权限控制.但SASL/PLAIN验证有一个问题:只能在JAAS文件KafkaServer中配置 ...
Kafka安全认证-KerberosSCRAM
SASL/SCRAM动态认证配置SASL/PLAIN验证,实现了对Kafka的权限控制.但SASL/PLAIN验证有一个问题:只能在JAAS文件KafkaServer中配置用户,一但Kafka启动, ...
大数据Hadoop之——Kafka安全机制（Kafka SSL认证实现）
文章目录一.概述 1)SASL认证概述 2)Delegation Token认证概述 3)SSL认证概述(本章实现) 二.各种安全认证机制对比和使用场景三.Kafka SSL认证实现 1)创建ss ...
kafka身份认证 maxwell_Kafka 使用SASL / SCRAM进行身份验证
使用SASL / SCRAM进行身份验证请先在不配置任何身份验证的情况下启动Kafka 1. 创建SCRAM Credentials 1.1 创建broker通信用户(或称超级用户) bash Em ...
kafka动态权限认证（SASL SCRAM + ACL）
kafka动态权限认证(SASL SCRAM + ACL) 创建三个测试用户 bin/kafka-configs.sh --zookeeper 192.168.x.x:2181 --alter --a ...
Kafka SASL/SCRAM+ACL实现动态创建用户及权限控制
文章目录 SASL_SCRAM+ACL实现动态创建用户及权限控制使用SASL / SCRAM进行身份验证 1. 创建SCRAM Credentials 创建broker建通信用户(或称超级用户) 创 ...

Kafka动态认证SASL/SCRAM验证