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

一、说明

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

本文将从零开始部署
ZooKeeper
和
Kafka
并通过配置
SASL/SCRAM
和
ACL
（访问控制列表）来增强
Kafka
的安全性。

二、Kafka 的安全机制

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

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

三、环境和软件准备

从
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

为 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 权限。

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