基于 SASL/SCRAM 让 Kafka 实现动态授权认证的方法_Java

一、说明

在大数据处理和分析中 apache kafka 已经成为了一个核心组件。然而在生产环境中部署 kafka 时，安全性是一个必须要考虑的重要因素。sasl（简单认证与安全层）和 scram（基于密码的认证机制的盐化挑战响应认证机制）提供了一种方法来增强 kafka 集群的安全性。

本文将从零开始部署 zookeeper 和 kafka 并通过配置 sasl/scram 和 acl（访问控制列表）来增强 kafka 的安全性。

二、kafka 的安全机制

kafka 社区在 0.9.0.0 版本正式添加了安全特性，可以满足各种安全性的要求，包括：

kafka 与 zookeeper 之间的安全通信；
kafka 集群 zookeeper 之间的安全通信；
客户端与服务端之间的安全通信；
消息级别的权限控制，可以控制客户端（生产者或消费者）的读写操作权限。

认证方式	引入版本	适用场景
ssl	0.9.0	ssl做信道加密比较多，ssl认证不如sasl所以一般都会使用ssl来做通信加密。
sasl/gssapi	0.9.9	主要是给 kerberos 使用的。如果你的公司已经做了 kerberos 认证（比如使用 active directory），那么使用 gssapi 是最方便的了。因为你不需要额外地搭建 kerberos，只要让你们的 kerberos 管理员给每个 broker 和要访问 kafka 集群的操作系统用户申请 principal 就好了。
sasl/plain	0.10.2	简单的用户名密码认证，通常与ssl结合使用，对于小公司来说，没必要搭建公司级别的kerberos，使用它就比较合适。
sasl/scram	0.10.2	plain的加强版本，支持动态的用户增减。
deleation token	1.1	delegation token 是在 1.1 版本引入的，它是一种轻量级的认证机制，主要目的是补充现有的 sasl 或 ssl 认证。如果要使用 delegation token，你需要先配置好 sasl 认证，然后再利用 kafka 提供的 api 去获取对应的 delegation token。这样 broker 和客户端在做认证的时候，可以直接使用这个 token，不用每次都去 kdc 获取对应的 ticket（kerberos 认证）或传输 keystore 文件（ssl 认证）。
sasl/oauthbearer	2.0	oauth 2框架的集成。

三、环境和软件准备

从 apache kafka 官网下载对应版本的 kafka 并解压到你选择的目录。

确保已经安装 java 才能运行 kafka，可以通过运行 java -version 来检查 java 环境。

四、部署 zookeeper

使用 kafka 内置的 zookeeper

4.1. 启用 sasl 认证

进入 config 目录，修改 zookeeper.properties 配置文件增加以下内容：

authprovider.1=org.apache.zookeeper.server.auth.saslauthenticationprovider
jaasloginrenew=3600000
requireclientauthscheme=sasl
zookeeper.sasl.client=true

4.2. 配置 jaas

在 config 目录下创建 zk_jaas.conf 文件，内容如下：

server {
    org.apache.zookeeper.server.auth.digestloginmodule required
    username="admin"
    password="admin"
    user_admin="admin"
    user_zkclient="zkclient";
};

其作用是创建了一个 server 节点，其中

org.apache.zookeeper.server.auth.digestloginmodule required 是认证逻辑的处理类；
username、password 是zookeeper之间通讯的用户名和密码；
user_admin="admin" 的结构是 user_[username]=[password] 定义 kafka-broker（zookeeper客户端）连接到 zookeeper 时用的用户名和密码。

注意：server 内部最后一行的 ; 和 } 后的 ; 不能缺少！

4.3. 修改启动文件

进入 bin 目录，修改 zookeeper-server-start.sh 文件；

在 export kafka_heap_opts= 配置项的参数后添加 jaas 的配置：

export kafka_heap_opts="-xmx512m -xms512m -djava.security.auth.login.config=../config/zk_jaas.conf"

4.4. 启动 zookeeper

执行命令：./zookeeper-server-start.sh -daemon ../config/zookeeper.properties

-daemon 参数配置后台运行

4.5. 测试

可以从官网 apache zookeeper 下载对应版本的 zookeeper 并解压；

添加 jaas 配置，在 confi 目录下创建 zk_client_jaas.conf 文件：

client{
    org.apache.zookeeper.server.auth.digestloginmodule required
    username="zkclient"
    password="zkclient";
};

修改 bin 目录下的 zkcli.sh 文件，在启动命令中增加 jaas 的配置：

"$java" "-dzookeeper.log.dir=${zoo_log_dir}" "-dzookeeper.root.logger=${zoo_log4j_prop}" "-dzookeeper.log.file=${zoo_log_file}" \
     -cp "$classpath" $client_jvmflags $jvmflags \
     "-djava.security.auth.login.config=../conf/zk_client_jaas.conf" \
     org.apache.zookeeper.zookeepermain "$@"

执行 zkcli.sh 连接本机已经启动好的 zookeeper

进入 kafka 的 log 目录，查看内置 zk 的日志 zookeeper.out 显示以下内容：

info adding sasl authorization for authorizationid: zkclient (org.apache.zookeeper.server.zookeeperserver)

代表 zookeeper 的 sasl 认证已经配置成功。

五、部署 kafka

5.1. 配置 kafka broker

进入 config 目录，修改 server.properties 配置文件增加以下内容：

listeners=sasl_plaintext://:9092
advertised.listeners=sasl_plaintext://localhost:9092
security.inter.broker.protocol=sasl_plaintext
sasl.mechanism.inter.broker.protocol=scram-sha-256
sasl.enabled.mechanisms=scram-sha-256
authorizer.class.name=kafka.security.authorizer.aclauthorizer
allow.everyone.if.no.acl.found=false
super.users=user:admin

authorizer.class.name 开启 acl 授权机制并指定实现类；
allow.everyone.if.no.acl.found 如果没有找到acl（访问控制列表）配置，是否允许任何操作；这里设置为 false 指除了超级管理员，其他用户必须配置 acl 才能访问资源；
super.users 超级管理员，无需配置 acl 拥有所有权限的用户。

5.2. 配置 jaas

在 config 目录下创建 kafka_server_jaas.conf 文件，内容如下：

kafkaserver {
    org.apache.kafka.common.security.scram.scramloginmodule required
    username="admin"
    password="admin";
};
client{
    org.apache.kafka.common.security.plain.plainloginmodule required
    username="zkclient"
    password="zkclient";
};

kafkaserver 中的 username，password 用于 kafka 集群 broker 节点之间通信用的账号密码；
kafkaserver 中的 user_test="test" 用于 kafka 客户端（producer，consumer）连接broker时，用该配置下user_[username]=[password]结构配置的账号密码登录；
client 用于 broker 和 zookeeper 之间的认证，对应 zk_jaas.conf 中的【user_zkclient="zkclient"】配置；
user_admin="admin" 的结构是 user_[username]=[password] 定义 kafka-broker（zookeeper客户端）连接到 zookeeper 时用的用户名和密码。

5.3. 修改启动文件

进入 bin 目录，修改 kafka-server-start.sh 文件；

在 export kafka_heap_opts= 配置项的参数后添加 jaas 的配置：

export kafka_heap_opts="-xmx1g -xms1g -djava.security.auth.login.config=../config/kafka_server_jaas.conf"

5.4. 创建 scram 用户

在启动 kafka 之前需要先创建好用户，在 bin 目录下执行以下内容：

分别创建 admin(超级管理员) 和 test(客户端用户)

./kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'scram-sha-256=[password=admin]' --entity-type users --entity-name admin
./kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'scram-sha-256=[password=test]' --entity-type users --entity-name test

sasl/scram 认证的用户信息是动态创建存储在 zookeeper 中，由于上面的配置 kafka_server_jaas.conf 中 broker 之间的通信是通过 admin 用户的，如果该用户不存在会 启动报错。

5.5. 启动 kafka

执行命令：./kafka-server-start.sh -daemon ../config/server.properties

-daemon 参数配置后台运行

六、验证 sasl/scram 鉴权

6.1. 客户端认证配置

6.1.1. 管理员配置

进入 config 目录创建 cmd.properties 内容如下：

security.protocol=sasl_plaintext
sasl.mechanism=scram-sha-256
sasl.jaas.config=org.apache.kafka.common.security.scram.scramloginmodule required username="admin" password="admin";

配置认证的类型以及登录逻辑的处理类和用户，使用超级管理员 admin

注意最后的 ; 是必须加上的。

6.1.2. 生产者配置

修改 config 目录下的 producer.properties 增加以下内容：

security.protocol=sasl_plaintext
sasl.mechanism=scram-sha-256
sasl.jaas.config=org.apache.kafka.common.security.scram.scramloginmodule required username="admin" password="admin";

生产者也使用超级管理员 admin 来发送消息。

6.1.3. 消费者配置

修改 config 目录下的 consumer.properties 增加以下内容：

security.protocol=sasl_plaintext
sasl.mechanism=scram-sha-256
sasl.jaas.config=org.apache.kafka.common.security.scram.scramloginmodule required username="test" password="test";

消费者使用 test 用户来接收消息。

6.2. 创建topic

在 bin 目录下执行以下命令：

./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic test --partitions 1 --replication-factor 1 --command-config ../config/cmd.properties

bootstrap-server 配置 kafka 服务端的地址
topic 指定topic名称
command-config 指定命令的认证配置，这里使用上面创建的 管理员配置

创建成功后可以通过以下命令查看存在的 topic 清单：

./kafka-topics.sh --bootstrap-server localhost:9092 --list --command-config ../config/cmd.properties

6.3. 创建消费者

6.3.1. 执行 kafka-console-consumer

在 bin 目录下执行以下命令：

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --consumer.config ../config/consumer.properties

执行命令后会发现以下报错信息：

error error processing message, terminating consumer process:  (kafka.tools.consoleconsumer$)
org.apache.kafka.common.errors.saslauthenticationexception: authentication failed during authentication due to invalid credentials with sasl mechanism scram-sha-256
processed a total of 0 messages

authentication failed 认证失败，由于消费者的认证使用的是 test 用户，而该用户还未配置任何 acl 权限。

6.3.2. 配置用户 acl 权限

kafka 的 acl (access control lists) 允许你定义哪些用户可以访问哪些主题，并且可以执行哪些操作（如读、写、创建、删除等）。

执行以下命令：

./kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal user:test --operation read --topic test --group test-consumer-group

bash 复制全屏

为 test 用户在资源 topic[test] 下分配只读权限

执行成功，可以通过以下命令查看资源所分配的所有 acl 清单：

./kafka-acls.sh --bootstrap-server localhost:9092 --topic test --list --command-config ../config/cmd.properties

重新创建消费者：

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --consumer.config ../config/consumer.properties

执行成功后该 shell 窗口会一直阻塞等待消息。

6.4. 创建生产者

新开一个 shell 窗口 在 bin 目录下执行以下命令：

./kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test --producer.config ../config/producer.properties

由于生产者的认证使用的是 admin 为 超级管理员 所以无需配置 acl 权限。

执行成功后会出现 > 符号，输入内容之后，切换到 消费者 窗口就可以看到了。

到此这篇关于基于 sasl/scram 让 kafka 实现动态授权认证的文章就介绍到这了,更多相关kafka动态授权认证内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！

基于 SASL/SCRAM 让 Kafka 实现动态授权认证的方法

2024年07月29日 • Java •我要评论

一、说明

二、kafka 的安全机制

三、环境和软件准备

四、部署 zookeeper

4.1. 启用 sasl 认证

4.2. 配置 jaas

4.3. 修改启动文件

4.4. 启动 zookeeper

4.5. 测试

五、部署 kafka

5.1. 配置 kafka broker

5.2. 配置 jaas

5.3. 修改启动文件

5.4. 创建 scram 用户

5.5. 启动 kafka

六、验证 sasl/scram 鉴权

6.1. 客户端认证配置

6.1.1. 管理员配置

6.1.2. 生产者配置

6.1.3. 消费者配置

6.2. 创建topic

6.3. 创建消费者

6.3.1. 执行 kafka-console-consumer

6.3.2. 配置用户 acl 权限

6.4. 创建生产者

相关文章:

Spring Boot中进行文件上传和文件下载功能实现

SpringBoot参数验证的几种方式小结

k8s的安装部署，详细过程展示（保姆级安装教程）

发表评论


验证码：

基于 SASL/SCRAM 让 Kafka 实现动态授权认证的方法

2024年07月29日 • Java •我要评论

一、说明

二、kafka 的安全机制

三、环境和软件准备

四、部署 zookeeper

4.1. 启用 sasl 认证

4.2. 配置 jaas

4.3. 修改启动文件

4.4. 启动 zookeeper

4.5. 测试

五、部署 kafka

5.1. 配置 kafka broker

5.2. 配置 jaas

5.3. 修改启动文件

5.4. 创建 scram 用户

5.5. 启动 kafka

六、验证 sasl/scram 鉴权

6.1. 客户端认证配置

6.1.1. 管理员配置

6.1.2. 生产者配置

6.1.3. 消费者配置

6.2. 创建topic

6.3. 创建消费者

6.3.1. 执行 kafka-console-consumer

6.3.2. 配置用户 acl 权限

6.4. 创建生产者

相关文章:

Spring Boot中进行 文件上传和 文件下载功能实现

SpringBoot参数验证的几种方式小结

k8s的安装部署，详细过程展示（保姆级安装教程）

发表评论

Spring Boot中进行文件上传和文件下载功能实现