Kafka集群配置SASL安全认证及使用

IT那活儿 2023-08-21

2901

点击上方“IT那活儿”公众号，关注后了解更多内容，不管IT什么活儿，干就完了！！！

Kafka简介

Kafka是一个高吞吐量、分布式的发布-订阅消息系统。Kafka核心模块使用Scala语言开发，支持多语言（如Java、Python、Go等）客户端，它可以水平扩展和具有高吞吐量特性而被广泛使用，并与多类开源分布式处理系统进行集成使用。

Kafka作为一款开源的、轻量级的、分布式、可分区和具备复制备份的、基于ZooKeeper协调管理的分布式流平台的功能强大的消息系统。与传统消息系统相比，Kafka能够更好的处理活跃的流数据，让数据在各个子系统中高性能、低延迟地不停流转。

实现方式

自0.9.0.0版本开始Kafka社区添加了许多功能用于提高Kafka集群的安全性，Kafka提供SSL或者SASL两种安全策略。SSL方式主要是通过CA令牌实现，此方案主要介绍SASL方式。

2.1 SASL验证

2.2 SSL加密

使用SSL加密在代理和客户端之间，代理之间或代理和工具之间传输的数据。

SCRAM认证配置的优点:

如果使用PLAIN认证有个问题，就是不能动态新增用户，每次添加用户后，需要重启正在运行的Kafka集群才能生效。

因此，在生产环境中，这种认证方式不符合实际业务场景，不利于后期扩展。然而使用SCRAM认证，可以动态新增用户，添加用户后，可以不用重启正在运行的Kafka集群即可进行鉴权。所以生产环境推荐使用SCRAM+PLAIN搭配的认证方案。

配置zookeeper集群启用SASL

3.1 修改 zk 配置启用sasl

[root@master2 config]# vim  /usr/local/zookeeper-3.4.14/conf/zoo.cfg
tickTime=2000
initLimit=1
syncLimit=5
dataDir=/data/zookeeper/data
dataLogDir=/data/zookeeper/datalog
clientPort=2181
admin.serverPort=8888
maxClientCnxns=3000
autopurge.snapRetainCount=3
autopurge.purgeInterval=24
server.1=192.xxx.xxx.112:2888:3888
server.2=192.xxx.xxx.114:2888:3888
server.3=192.xxx.xxx.115:2888:3888
4lw.commands.whitelist=conf,stat,srvr,mntr
#zk SASL
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
jaasLoginRenew=3600000
requireClientAuthScheme=sasl
zookeeper.sasl.client=true

3.2 配置 zk JAAS

[root@master2 config]# vim usr/local/zookeeper-3.4.14/conf/zk_jaas.conf
Server {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    username="admin"
    password="admin123"
    user_kafka="kafka123";
};

注意：

admin用户是zk 集群之间使用的。
kafka用户是 broker 与 zk 之间使用的。

3.3 修改zkEnv.sh

将上一步添加的 jaas 配置文件添加到zookeeper的环境变量中，zkEnv.sh文件最后添加一行：

vim /usr/local/zookeeper-3.4.14/bin/zkEnv.sh
ZOOBINDIR="${ZOOBINDIR:-/usr/bin}"
ZOOKEEPER_PREFIX="${ZOOBINDIR}/.."

# 新增变量SERVER_JVMFLAGS：

export SERVER_JVMFLAGS="-
Djava.security.auth.login.config=/usr/local/zookeeper-3.4.14/conf/zk_jaas.conf"

3.4 引入kafka依赖包

将kafka安装包下面的相关依赖包拷贝到zookeeper的目录下。

注：不同的kafka版本jar包版本会不一样。

3.5 启动zk集群节点并查看集群状态

zkServer.sh restart|status

配置kafka sasl动态认证

SASL/SCRAM认证是把凭证(credential)存储在Zookeeper，使用kafka-configs.sh在Zookeeper中创建凭据。对于每个SCRAM机制，必须添加具有机制名称的配置来创建凭证，所以在启动Kafka broker之前需要创建代理间通信的凭据。

我这里配置的 Kafka和生产者/消费者之间采用SASL/PLAIN和SASL/SCRAM两种方式共同完成认证，授权使用ACL方式。PLAIN方式的用户是在jaas文件中写死的，不能动态的添加；SCRAM支持动态的添加用户。

4.1 创建超级用户

配置SASL/SCRAM认证的第一步，是配置可以连接到kafka集群的用户。本案例创建了3个用户：admin，producer，consumer。kafka_server_admin用户用于broker之间的认证通信，producer用户用于生产者连接kafka，consumer用户用于消费者连接kafka。

创建admin用户：

kafka-configs.sh --zookeeper 192.xxx.xxx.112:2181 --alter --
add-config 'SCRAM-SHA-256=
[iterations=8192,password=admin123],SCRAM-SHA-512=
[password=admin123]' --entity-type users --entity-name admin

Completed Updating config for entity: user-principal 'admin'. #正常情况打印信息这串信息。

PS：用户 admin 这里配置admin用户用于实现broker间的通讯。

4.2 查看创建用户信息

kafka-configs 脚本是用来设置主题级别参数的。其实，它的功能还有很多。比如在这个例子中，我们使用它来创建 SASL/SCRAM 认证中的用户信息。可以使用下列命令来查看刚才创建的用户数据。

kafka-configs.sh --zookeeper 192.xxx.xxx.112:2181 --describe --entity-type users 
（可以单独指定某个用户 --entity-name producer，如下）
kafka-configs.sh --zookeeper 192.xxx.xxx.112:2181 --describe --entity-type users --entity-name producer

Configs for user-principal 'producer' are SCRAM-SHA-512=salt=czVldW1mNDRlcTgzdnJydWxrOXB0YThxbw==,stored_key=Hjrex55CEcLSA2b1n8bvL5CfjziLPHD8EoduoCkNrT0xcVDthoQSi4hvt7szU55pJP/LTbpQkZdV66nueVzKmg==,server_key=OcXwU1KyxBU1tUdlu9ikpfwl+y+ne121iD4amN7zKl8I84KeBZlrwz1IKB1ICFFiAP9XoRRj0QMgHbCfhsL/wA==,iterations=4096,SCRAM-SHA-256=salt=NDJzZmZ0MmhxMDI0N2ZrbXh6bTQwYnZzcA==,stored_key=EIzpjfD4JzItzjeh16UvpQEGyYoZbesR0GRsQB58Als=,server_key=5LAvxe/KjCas7w4GM+KlkT/il99peOozXY/HPZeSF14=,iterations=8192

这段命令包含了 writer 用户加密算法 SCRAM-SHA-256 以及 SCRAM-SHA-512 对应的盐值 (Salt)、ServerKey 和 StoreKey，这些都是 SCRAM 机制的术语。

还可以在zk客户端中可以查看创建成功的用户：

zkCli.sh  -server 192.xxx.xxx.112:2181
[zk: 192.xxx.xxx.112:2181(CONNECTED) 0] ls /config/users
[admin, producer, consumer]
[zk: 192.xxx.xxx.112:2181(CONNECTED) 1]

备注:

删除用户[producer]的SCRAM证书：

kafka-configs.sh --zookeeper 192.xxx.xxx.112:2181 --alter --delete-config 'SCRAM-SHA-256' --entity-type users --entity-name producer

4.3 配置kafka jaas文件

配置了用户之后，我们需要为 Broker 创建一个对应的 JAAS 文件。在实际场景中，需要为每台单独的物理 Broker 机器都创建一份 JAAS 文件。

Kafka 的 jaas认证配置文件，配置的是登录类，超管密码和管理的帐号密码列表。

vim /usr/local/kafka_2.12-2.3.0/config/kafka_server_jaas.conf
KafkaServer {
    #org.apache.kafka.common.security.plain.PlainLoginModule required
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="admin123"
    user_admin="admin123"
    user_producer="producer123"
    user_consumer="consumer123";
};
KafkaClient {
    #org.apache.kafka.common.security.plain.PlainLoginModule required
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="admin123"
    user_producer="producer123"
    user_consumer="consumer123";
};
Client {
    #org.apache.kafka.common.security.plain.PlainLoginModule required
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="kafka"
    password="kafka123";
};
KafkaServer { 
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username ="admin"
    password="admin123"
    user_admin="admin123"
    user_producer="producer123"
    user_consumer="consumer123";
};
KafkaClient { 
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="admin123"
    user_producer="producer123"
    user_consumer="consumer123";
};
Client { 
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="kafka"
    password="kafka123";
};

KafkaServer中usename配置的是kafka服务端使用的账号和密码，后面的user_xxx事预设的普通帐号认证信息。
中间部分配置的是PLAIN认证方式的账户和密码，其中producer1是账户名，producer123是密码。
Client配置了broker到Zookeeper的连接用户名密码，这里要和前面zookeeper配置中的 zk_jaas.conf.conf 中 user_kafka 的账号和密码相同。

关于这个文件内容，需要注意以下两点：

1）不要忘记最后一行和倒数第二行结尾处的分号；
2）JAAS 文件中不需要任何空格键。

4.4 kafka 配置文件启用SASL认证

Kafka 服务配置文件 server.propertis，配置认证协议及认证实现类。

************代理节点1 ************

[root@master2 config]# cat server.properties
broker.id=0
listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://192.xxx.xxx.112:9092
sasl.enabled.mechanisms=SCRAM-SHA-256,PLAIN
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
security.inter.broker.protocol=SASL_PLAINTEXT
allow.everyone.if.no.acl.found=false
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:admin
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/datalogs
num.partitions=3
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.hours=168
log.retention.bytes=1073741824
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
delete.topic.enable=true
auto.create.topics.enable=false
zookeeper.connect=192.xxx.xxx.112:2181,192.xxx.xxx.114:2181,192.xxx.xxx.115:2181
zookeeper.connection.timeout.ms=60000
group.initial.rebalance.delay.ms=0

************代理节点2 ************

[root@node1 config]# cat server.properties
broker.id=1
listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://192.xxx.xxx.114:9092
sasl.enabled.mechanisms=SCRAM-SHA-256,PLAIN
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
security.inter.broker.protocol=SASL_PLAINTEXT
allow.everyone.if.no.acl.found=false
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:admin
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/datalogs
num.partitions=3
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.hours=168
log.retention.bytes=1073741824
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
delete.topic.enable=true
auto.create.topics.enable=false
zookeeper.connect=192.xxx.xxx.112:2181,192.xxx.xxx.114:2181,192.xxx.xxx.115:2181
zookeeper.connection.timeout.ms=60000
group.initial.rebalance.delay.ms=0

************代理节点3 ************

[root@node2 config]# cat server.properties
broker.id=2
listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://192.xxx.xxx.115:9092
sasl.enabled.mechanisms=SCRAM-SHA-256,PLAIN
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
security.inter.broker.protocol=SASL_PLAINTEXT
allow.everyone.if.no.acl.found=false
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:admin
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/datalogs
num.partitions=3
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.hours=168
log.retention.bytes=1073741824
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
delete.topic.enable=true
auto.create.topics.enable=false
zookeeper.connect=192.xxx.xxx.112:2181,192.xxx.xxx.114:2181,192.xxx.xxx.115:2181
zookeeper.connection.timeout.ms=60000
group.initial.rebalance.delay.ms=0

4.5 kafka 启动脚本添加认证文件路径的环境变量

Kafka 安全认证可以直接通过环境变量 -Djava.security.auth.login.config 设置，修改 Kafka 启动脚本 kafka-start-server.sh 文件最后一行，增加一个参数指向 jaas 配置文件的绝对路径：

vim /usr/local/kafka_2.12-2.3.0/bin/kafka-server-start.sh
#exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/usr/local/kafka_2.12-2.3.0/config/kafka_server_jaas.conf kafka.Kafka "$@"

4.6 kafka客户端配置

1）配置consumer.properties和producer.properties,都要加入以下配置

ecurity.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512

2）生产者配置

使用kafka-console-producer.sh脚本测试生产者，由于开启安全认证和授权，此时使用console-producer脚本来尝试发送消息，那么消息会发送失败，原因是没有指定合法的认证用户，因此客户端需要做相应的配置，需要创建一个名为producer.conf的配置文件给producer程序使用。

config目录下创建一个producer.conf的文件，文件内容如下：

[root@master2 config]# cat producer.conf
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.Scra
mLoginModule required username="producer" password="producer123";

注意：Topic设置写权限。

3）消费者配置

使用kafka-console-consumer.sh脚本测试生产者，由于开启安全认证和授权，因此客户端需要做相应的配置。需要为 consumer 用户创建consumer.conf给消费者程序，同时设置对topic的读权限。

config目录下创建一个consumer.conf的文件，文件内容如下：

[root@master2 config]# vim consumer.conf
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.Scra
mLoginModule required username="consumer" password="consumer123";

注意：Topic设置读权限。

4）在生产者和消费者启动脚本中引入JAAS文件

[root@master2 kafka_2.12-2.3.0]# vim bin/kafka-console-producer.sh

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx512M"
fi
export KAFKA_OPTS="-Djava.security.auth.login.config=
/usr/local/kafka_2.12-2.3.0/config/kafka_server_jaas.conf"exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleProducer "$@"

[root@master2 kafka_2.12-2.3.0]# vim bin/kafka-console-consumer.sh
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx512M"
fi
export KAFKA_OPTS="-Djava.security.auth.login.config=/usr/local/kafka_2.12-2.3.0/config/kafka_server_jaas.conf"
exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleConsumer "$@"

4.7 启动kafka集群

nohup kafka-server-start.sh /usr/local/kafka_2.12-2.3.0/config/server.properties &

4.8 查看kafka集群是否注册到zk

zkCli.sh  -server localhost:2181
[zk: localhost:2181(CONNECTED) 13] ls /brokers/ids
[0, 1, 2]

测试及使用

5.1 生产者测试

1）创建一个测试主题test_topic

kafka-topics.sh --zookeeper 192.xxx.xxx.112:2181 --create --
replication-factor 3 --partitions 3 --topic test_topic
Created topic test_topic.

2）查看创建的Topic

[root@node2 ~]# kafka-topics.sh  --list --zookeeper 192.xxx.xxx.112:2181
test_topic
[root@node2 ~]# kafka-topics.sh --describe --zookeeper 192.xxx.xxx.112:2181 --topic test_topic
Topic:test_topic        PartitionCount:3        ReplicationFactor:3     Configs:
Topic: test_topic       Partition: 0 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2
Topic: test_topic       Partition: 1 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0
Topic: test_topic       Partition: 2 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1

继续........

注：kafka开启认证后，当生产者往Topic写数据需要为主题配置权限（write），即对生产者赋予写的权限。

这里使用producer用户认证授权，通过ACL为producer 用户分配操作test_topic权限（下面两个命令二选一即可）：

分配写的权限

kafka-acls.sh --authorizer-properties 
zookeeper.connect=192.xxx.xxx.114:2181 --add --allow-
principal User:producer --operation Write --topic 'test_topic'

启动生产者发送消息

特别注意：在生产者生产消息之前如果不设置生产者用户的ACL权限会报错:

[2022-12-18 16:48:18,155] WARN [Producer clientId=console-
producer] Error while fetching metadata with correlation id 
3 : {test_topic=TOPIC_AUTHORIZATION_FAILED} 
(org.apache.kafka.clients.NetworkClient)

[2022-12-18 16:48:18,156] ERROR [Producer clientId=console-
producer] Topic authorization failed for topics [test_topic] 
(org.apache.kafka.clients.Metadata)

[2022-12-18 16:48:18,158] ERROR Error when sending message 
to topic test_topic with key: null, value: 5 bytes with 
error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [test_topic]

kafka-console-producer.sh --broker-list 192.xxx.xxx.114:9092 --topic test_topic --producer.config /usr/local/kafka_2.12-2.3.0/config/producer.conf
>hello
>world
>end

5.2 消费者测试

注：kafka开启认证后，当消费者拉取主题的消息时需要为Topic配置权限（read），即对生产者赋予写的权限。

这里使用consumer用户认证授权，通过ACL为consumer用户分配操作test_topic权限，同时需要对消费者组授权（给groupId配权），执行一下2个步骤:

1）分配读的权限

kafka-acls.sh --authorizer-properties 
zookeeper.connect=192.xxx.xxx.114:2181 --add --allow-
principal User:consumer--operation Read --topic 'test_topic'

2）分配consumer的权限

kafka-acls.sh --authorizer-properties 
zookeeper.connect=192.xxx.xxx.114:2181 --add --allow-
principal User:consumer--consumer --topic test_topic --group test_group

启动消费者消费消息：

读取消息前，如果对开启消费者用户读取topic权限，会报错：

[2022-12-18 17:48:59,996] WARN [Consumer clientId=consumer-1, 
groupId=test_group] Error while fetching metadata with correlation id 2 : {test_topic=TOPIC_AUTHORIZATION_FAILED} 
(org.apache.kafka.clients.NetworkClient)
[2022-12-18 17:49:00,015] ERROR [Consumer clientId=consumer-1, 
groupId=test_group] Topic authorization failed for topics 
[test_topic] (org.apache.kafka.clients.Metadata)
[2022-12-18 17:49:00,018] ERROR Error processing message, 
terminating consumer process:  
(kafka.tools.ConsoleConsumer$)
org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [test_topic]
Processed a total of 0 messages
#kafka-console-consumer.sh --bootstrap-server 192.xxx.xxx.112:9092,192.xxx.xxx.114:9092,192.xxx.xxx.115:9092  --topic test --group test-group --from-beginning --consumer.config /usr/local/kafka_2.12-2.3.0/config/consumer.conf
kafka-console-consumer.sh --bootstrap-server 192.xxx.xxx.112:9092  --topic test_topic --group test-group --from-beginning --consumer.config /usr/local/kafka_2.12-2.3.0/config/consumer.conf
hello
world
end

5.3 测试案例

1）创建Topic

kafka-topics.sh --zookeeper 192.xxx.xxx.115:2181 --create --
partitions 2 --replication-factor 1 --topic test1

2）生产者端配置

#添加写权限
kafka-acls.sh --authorizer 
kafka.security.auth.SimpleAclAuthorizer --authorizer-
properties zookeeper.connect=192.xxx.xxx.114:2181 --add --
allow-principal User:producer --operation Write --topic test1
#写入消息
kafka-console-producer.sh --broker-list 
192.xxx.xxx.112:9092,192.xxx.xxx.114:9092,192.xxx.xxx.115:9092 --topic test1 
--producer.config /usr/local/kafka_2.12-
2.3.0/config/producer.conf
>hello
>world

3）消费者端配置

#添加读权限
kafka-acls.sh --authorizer 
kafka.security.auth.SimpleAclAuthorizer --authorizer-
properties zookeeper.connect=192.xxx.xxx.114:2181 --add --
allow-principal User:consumer --operation Read --topic test1
#添加消费者组权限
kafka-acls.sh --authorizer 
kafka.security.auth.SimpleAclAuthorizer --authorizer-
properties zookeeper.connect=192.xxx.xxx.114:2181 --add --
allow-principal User:consumer --operation Read --group test-group
#拉取消息
kafka-console-consumer.sh --bootstrap-server 
192.xxx.xxx.112:9092,192.xxx.xxx.114:9092,192.xxx.xxx.115:9092  
--topic test1 --from-beginning --consumer.config 
/usr/local/kafka_2.12-2.3.0/config/consumer.conf --group test-group
hello
World