【Kafka SASL/SCRAM动态认证集群部署-阿里云开发者社区

目的：配置 SASL/PLAIN 验证，实现了对 Kafka 的权限控制。但 SASL/PLAIN 验证有一个问题：只能在 JAAS 文件 KafkaServer 中配置用户，一旦 Kafka 启动，无法动态新增用户。SASL/SCRAM 验证可以动态新增用户并分配权限。

1. 服务端配置

1、解压安装包

tar -zxvf kafka_2.11-2.4.1.tgz -C /home/xyp9x/

2、改名

mv kafka_2.11-2.4.1 kafka_scram

3、在 kafka_scram 目录下创建 logs、kafka-logs 文件夹

mkdir logs kafka-logs

4、修改配置文件

vi server.properties
将原内容全部清空，输入以下内容：
#broker的全局唯一编号，不能重复
broker.id=0
#是否允许删除topic
delete.topic.enable=true
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘IO的线程数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的最大缓冲区大小
socket.request.max.bytes=104857600
#kafka数据存放的路径
log.dirs=/home/xyp9x/kafka_scram/kafka-logs
#topic在当前broker上的分区个数
num.partitions=1
#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment文件保留的最长时间，超时将被删除
log.retention.hours=168
#配置连接Zookeeper集群地址
zookeeper.connect=bigdata111:2181,bigdata112:2181,bigdata113:2181
#kafka连接zookeeper超时时间90s
zookeeper.connection.timeout.ms=90000

5、配置环境变量

vi /etc/profile
# Kafka_scram
export KAFKA_HOME=/home/xyp9x/kafka_scram
export PATH=$PATH:$KAFKA_HOME/bin

6、分发

rsync -r /etc/profile bigdata112:/etc/
rsync -r /etc/profile bigdata113:/etc/
rsync -r kafka_scram bigdata112:/home/xyp9x/
rsync -r kafka_scram bigdata113:/home/xyp9x/

7）分别在 112 和 113 上修改配置文件

vi server.properties
broker.id=1
vi server.properties
broker.id=2

8）更新 111、112、113 的环境变量

source /etc/profile

9、启动集群（依次在 111、112、113 节点上启动 kafka）

bin/kafka-server-start.sh -daemon config/server.properties

10、创建 SCRAM 证书（SCRAM-SHA-256/SCRAM-SHA-512 是对密码加密的算法，二者有其一即可）

此方法是把凭证(credential)存储在 Zookeeper，可以使用 kafka-configs.sh 在 Zookeeper 中创建凭据。对于每个 SCRAM 机制，必须添加具有机制名称的配置来创建凭证，在启动 Kafka broker 之前创建代理间通信的凭据。所以第一步，在没有设置任何权限的配置下启动 Kafka。

bin/kafka-configs.sh --zookeeper x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --alter --entity-type users --entity-name admin --add-config 'SCRAM-SHA-256=[password=admin],SCRAM-SHA-512=[password=admin]'

11、查看 SCRAM 证书

bin/kafka-configs.sh --zookeeper x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --describe --entity-type users --entity-name admin

12、在 config 目录中创建 kafka_server_jaas.conf 文件

vi kafka_server_jaas.conf
KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="admin";
};

13、修改配置文件

vi server.properties
在原有基础上添加以下内容：
#认证配置
listeners=SASL_PLAINTEXT://:9092
#PLAINTEXT不加密明文传输, 省事, 性能也相对好
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
sasl.enabled.mechanisms=SCRAM-SHA-256
#ACL配置
#设置为true，ACL机制为黑名单机制，只有黑名单中的用户无法访问
#设置为false，ACL机制为白名单机制，只有白名单中的用户可以访问
allow.everyone.if.no.acl.found=false
super.users=User:admin
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

14、在 Kafka 启动脚本中添加配置文件路径

vi kafka-server-start.sh
#!/bin/bash
export KAFKA_OPTS="-Djava.security.auth.login.config=/home/xyp9x/kafka_scram/config/kafka_server_jaas.conf"

15、分发

rsync -r config/kafka_server_jaas.conf bigdata112:/home/xyp9x/kafka_scram/config/
rsync -r config/kafka_server_jaas.conf bigdata113:/home/xyp9x/kafka_scram/config/
rsync -r config/server.properties bigdata112:/home/xyp9x/kafka_scram/config/
rsync -r config/server.properties bigdata113:/home/xyp9x/kafka_scram/config/
rsync -r bin/kafka-server-start.sh bigdata112:/home/xyp9x/kafka_scram/bin/
rsync -r bin/kafka-server-start.sh bigdata113:/home/xyp9x/kafka_scram/bin/

16、分别在 112 和 113 上修改配置文件

vi server.properties
broker.id=1
vi server.properties
broker.id=2

2. 客户端配置

1、为 admin 用户创建一个 jaas 文件

vi kafka_client_jaas_admin.conf
KafkaClient {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="admin";
};

2、在 producer.properties 文件中添加认证协议

vi producer.properties
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256

3、在 consumer.properties 文件中添加认证协议

vi consumer.properties
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256

4、在 kafka-console-producer.sh 脚本中添加 JAAS 文件的路径

vi kafka-console-producer.sh
#!/bin/bash
export KAFKA_OPTS="-Djava.security.auth.login.config=/home/xyp9x/kafka_scram/config/kafka_client_jaas_admin.conf"

5、在 kafka-console-consumer.sh 脚本中添加 JAAS 文件的路径

vi kafka-console-consumer.sh
#!/bin/bash
export KAFKA_OPTS="-Djava.security.auth.login.config=/home/xyp9x/kafka_scram/config/kafka_client_jaas_admin.conf"

6、分发

rsync -r config/kafka_client_jaas_admin.conf bigdata112:/home/xyp9x/kafka_scram/config/
rsync -r config/kafka_client_jaas_admin.conf bigdata113:/home/xyp9x/kafka_scram/config/
rsync -r config/producer.properties bigdata112:/home/xyp9x/kafka_scram/config/
rsync -r config/producer.properties bigdata113:/home/xyp9x/kafka_scram/config/
rsync -r config/consumer.properties bigdata112:/home/xyp9x/kafka_scram/config/
rsync -r config/consumer.properties bigdata113:/home/xyp9x/kafka_scram/config/
rsync -r bin/kafka-console-producer.sh bigdata112:/home/xyp9x/kafka_scram/bin/
rsync -r bin/kafka-console-producer.sh bigdata113:/home/xyp9x/kafka_scram/bin/
rsync -r bin/kafka-console-consumer.sh bigdata112:/home/xyp9x/kafka_scram/bin/
rsync -r bin/kafka-console-consumer.sh bigdata113:/home/xyp9x/kafka_scram/bin/

7、启动 kafka-broker

bin/kafka-server-start.sh -daemon config/server.properties

8、创建 topic

bin/kafka-topics.sh --zookeeper x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --create --partitions 3 --replication-factor 3 --topic test

9、超级用户启动生产者

bin/kafka-console-producer.sh --broker-list x.x.1.111:9092,x.x.1.112:9092,x.x.1.113:9092 --topic test --producer.config config/producer.properties

10、超级用户启动消费者

bin/kafka-console-consumer.sh --bootstrap-server x.x.1.111:9092,x.x.1.112:9092,x.x.1.113:9092 --topic test --consumer.config config/consumer.properties

3. ACL 操作（在配置好 SASL 后，启动 Zookeeper 集群和 Kafka 集群之后，就可以使用 kafka-acls.sh 脚本来操作 ACL 机制）

1、创建普通用户 SCRAM 证书

bin/kafka-configs.sh --zookeeper x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --alter --entity-type users --entity-name xyp9x --add-config 'SCRAM-SHA-256=[iterations=8192,password=xyp9x],SCRAM-SHA-512=[password=xyp9x]'

2、查看 SCRAM 证书

bin/kafka-configs.sh --zookeeper x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --describe --entity-type users --entity-name xyp9x

3、删除 SCRAM 证书

bin/kafka-configs.sh --zookeeper x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --alter --delete-config 'SCRAM-SHA-512' --delete-config 'SCRAM-SHA-256' --entity-type users --entity-name xyp9x

4、查看 ACL 授权

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --list

5、对普通用户生产者授写权限

//允许xyp9x用户从所有IP地址写
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --add --allow-principal User:xyp9x --operation Write --topic test
//禁止xyp9x用户从所有IP地址写
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --remove --allow-principal User:xyp9x --operation Write --topic test
//允许xyp9x用户从x.x.1.111和x.x.1.112写
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --add --allow-principal User:xyp9x --allow-host x.x.1.111 --allow-host x.x.1.112 --operation Write --topic test
//禁止xyp9x用户从x.x.1.111和x.x.1.112写
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --remove --allow-principal User:xyp9x --allow-host x.x.1.111 --allow-host x.x.1.112 --operation Write --topic test

6、对普通用户消费者组授权

//允许xyp9x用户使用test-consumer-group消费组消费消息
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --add --allow-principal User:xyp9x --operation Read --group test-consumer-group
//禁止xyp9x用户使用test-consumer-group消费组消费消息
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --remove --allow-principal User:xyp9x --operation Read --group test-consumer-group

7、对普通用户消费者授读权限

//允许xyp9x用户从所有IP地址读
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --add --allow-principal User:xyp9x --operation Read --topic test
//禁止xyp9x用户从所有IP地址读
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --remove --allow-principal User:xyp9x --operation Read --topic test
//允许xyp9x用户从x.x.1.111和x.x.1.112读
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --add --allow-principal User:xyp9x --allow-host x.x.1.111 --allow-host x.x.1.112 --operation Read --topic test
//禁止xyp9x用户从x.x.1.111和x.x.1.112读
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=x.x.1.111:2181,x.x.1.112:2181,x.x.1.113:2181 --remove --allow-principal User:xyp9x --allow-host x.x.1.111 --allow-host x.x.1.112 --operation Read --topic test

8、总结

SASL/SCRAM验证方法可以在Kafka服务启动之后，动态的新增用户分并配权限，在业务变动频繁，开发人员多的

【Kafka SASL/SCRAM动态认证集群部署

1. 服务端配置

2. 客户端配置

3. ACL 操作（在配置好 SASL 后，启动 Zookeeper 集群和 Kafka 集群之后，就可以使用 kafka-acls.sh 脚本来操作 ACL 机制）

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

【Kafka SASL/SCRAM动态认证集群部署

1. 服务端配置

2. 客户端配置

3. ACL 操作（在配置好 SASL 后，启动 Zookeeper 集群和 Kafka 集群之后，就可以使用 kafka-acls.sh 脚本来操作 ACL 机制）

热门文章

最新文章

相关课程

相关电子书

相关实验场景