Kafka配置SASL认证

Kafka加密

Kafka认证方式

在本博客中我们使用SASL/PLAIN的方式来进行Kafka加密

环境准备

• Kafka集群环境 请参考之前的Kafka集群搭建 - kafka-broker1- kafka-broker2

集群配置SASL/PLAIN认证

用户校验

修改server.properties 让其支持Kafka的认证(每一个broker节点都需要修改这些配置)

修改server.properties配置

# 备份server.propertiescp /opt/module/kafka/config/server.properties /opt/module/kafka/config/server.properties.bak
vim /opt/module/kafka/config/server.properties

#添加如下内容########## Kafka SASL认证###注意：如果外网也要支持SASL advertised也要同步修改listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://kafka-broker1:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

添加SASL认证配置

vim /opt/module/kafka/config/kafka_server_jaas.conf
# 增加如下配置
KafkaServer {
     org.apache.kafka.common.security.plain.PlainLoginModule required
     username="admin"password="123456"user_admin="123456"user_user2="123456";};#分发文件到不同的机器节点
 xsync kafka_server_jaas.conf

解释：

此配置定义了两个用户（admin和user2）。KafkaServer中的username、password配置的用户和密码，是用来broker和broker连接认证。在本例中，admin是代理broker间通信的用户。user_userName配置为连接到broker的所有用户定义密码，broker使用这些验证所有客户端连接，包括来自其他broker的连接。

请注意，username是服务端连接的账号，password为密码，其中下面的user_admin就是为了集群间通信时候使用的

将JAAS配置文件位置作为JVM参数传递给每个Kafka broker

修改kafka-run-class.sh

#修改kafka-run-class.shKAFKA_OPTS="-Djava.security.auth.login.config=/opt/module/kafka/config/kafka_server_jaas.conf"#分发到其他机器节点
cd/opt/module/kafka/bin
xsync kafka-run-class.sh

启动集群服务

./cluster.sh start

验证权限

集群权限已经启动 所以这个命令是没有任何效果的

为了满足我们的权限列表 我们需要把用户user2作为我们的配置添加进去

• 执行以下脚本

vim /root/user2.conf
# 输入如下内容security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="user2"password="123456";

其中的用户名必须是

admin 123456

或者是

user2 123456

这个时候，我们已经可以看到 权限是认证通过的

测试消费端和生产端(添加配置 验证成功)

权限校验

admin是超级用户
user2是普通用户
我们现在编写权限校验的目的是
1:user2可以自己新建/修改/删除topic
2:只用于user2操作某一个topic 其他topic不允许操作

修改server.properties配置

# 修改server.properties 增加如下配置authorizer.class.name=kafka.security.authorizer.AclAuthorizer
super.users=User:admin

使用权限配置并且超级用户是admin，这个用户和我们在上面配置的列表中的admin用户相互映射

每个机器都进行配置

重启所有的节点,这个时候已经开启了权限

# 配置用户认证文件信息vim /root/admin.conf
# 输入如下内容security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin"password="123456";

修改kafka-acls.sh

此脚本主要用户给指定的用户配置权限

参数/选项描述

--add

增加权限

--remove

删除权限

--allow-host

允许操作的主机

--deny-host

拒绝的主机

--allow-principal

允许操作的用户（通常为User:ClientID）

--deny-principal

拒绝的用户（通常为User:ClientID）

--bootstrap-server

Kafka集群地址，用于连接到Kafka集群

--command-config

认证文件信息，如JAAS配置文件的路径

--group

可以消费特定topic的消费者组

--list

查询所有权限

--operation

指定操作权限，可能的值包括：Describe, DescribeConfigs, Alter, Read, Delete, Create, All, Write, AlterConfigs

--topic

指定要操作的topic

基本命令

• 查询topic的所有权限kafka-acls.sh --bootstrap-server kafka-broker1:9092 --list --command-config /root/admin.conf

• 删除所有权限kafka-acls.sh --bootstrap-server kafka-broker1:9092 --topic user2 --remove --command-config /root/admin.conf
• 当前用户没有权限，现在给user用户赋值权限kafka-acls.sh --bootstrap-server kafka-broker1:9092 --topic user2 --add --allow-principal User:user2 --operation all --command-config /root/admin.conf
• 回收权限kafka-acls.sh --bootstrap-server kafka-broker1:9092 --topic user2 --remove --allow-principal User:user2 --operation all --command-config /root/admin.conf
• 给生产者添加Write权限kafka-acls.sh --bootstrap-server kafka-broker1:9092 --topic user2 --add --allow-principal User:user2 --operation Write --command-config /root/admin.conf
• 给消费者添加Read权限kafka-acls.sh --bootstrap-server kafka-broker1:9092 --topic user2 --add --allow-principal User:user2 --operation Read --group user2 --command-config /root/admin.conf
• 测试生产者kafka-console-producer.sh --bootstrap-server kafka-broker1:9092 --topic user2 --producer.config /root/user2.conf
• 测试消费者kafka-console-consumer.sh --bootstrap-server kafka-broker1:9092 --topic user2 --from-beginning --consumer.config /root/user2.conf --group user2