Kafka安全认证授权配置

一概述
- 1.1 Kafka的权限分类
- 1.2 实现方式
二安全认证授权配置
- 2.1 zookeeper配置
- - 2.1.1 引入kafka依赖包
  - 2.1.2修改ZK配置文件
  - 2.1.3 创建jaas文件
  - 2.1.4 修改zkEnv.sh
  - 2.1.5 启动zk
- 2.2 Kafka配置
- - 2.2.1 创建超级用户
  - 2.2.2 创建jaas文件
  - 2.2.3 修改kafka配置文件
  - 2.2.4启动kafka
三测试连接
- 3.1 生产者测试
- 3.2 消费者测试
- 3.3 beats工具连接
- 3.4 logstash消费

一概述

1.1 Kafka的权限分类

1）身份认证（Authentication）：对client 与服务器的连接进行身份认证，brokers和zookeeper之间的连接进行Authentication（producer 和 consumer）、其他 brokers、tools与 brokers 之间连接的认证。

2）权限控制（Authorization）：实现对于消息级别的权限控制，clients的读写操作进行Authorization：（生产/消费/group)数据权限。

1.2 实现方式

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

验证方式	Kafka版本	特点
SASL/PLAIN	0.10.0.0	不能动态添加用户
SASL/SCRAM	0.10.2.0	可以动态添加用户
SASL/Kerberos	0.9.0.0	需要独立部署验证服务
SASL/OAUTHBEARER	2.0.0	需自己实现接口实现token的创建和验证，需要额外的oauth服务

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

二安全认证授权配置

本文档主要演示SASL/SCRAM和SASL/PLAIN 搭配使用的方案。版本：kafka_2.12-2.5.1.tgz、apache-zookeeper-3.5.8-bin.tar.gz

2.1 zookeeper配置

2.1.1 引入kafka依赖包

将kafka包下面的相关依赖包复制到zookeeper的目录下，不同的kafka
版本jar包版本会不一样。

cp /mnt/datadisk/kafka/libs/kafka-clients-2.5.1.jar /mnt/datadisk/zookeeper/lib
cp /mnt/datadisk/kafka/libs/lz4-java-1.7.1.jar /mnt/datadisk/zookeeper/lib
cp /mnt/datadisk/kafka/libs/slf4j-api-1.7.30.jar /mnt/datadisk/zookeeper/lib
cp /mnt/datadisk/kafka/libs/slf4j-log4j12-1.7.30.jar /mnt/datadisk/zookeeper/lib
cp /mnt/datadisk/kafka/libs/snappy-java-1.1.7.3.jar /mnt/datadisk/zookeeper/lib

2.1.2修改ZK配置文件

Zookeeper的配置文件zoo.cfg中添加如下内容：

authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000

2.1.3 创建jaas文件

在zookeeper的conf目录下添加安全认证的jaas文件，该文件定义需要链接到Zookeeper服务器的用户名和密码。这里定义文件名为zk-server-jaas.conf，文件内容为：

Server {org.apache.kafka.common.security.plain.PlainLoginModule required
username="zoo_admin"
password="zoo_admin"
user_kafka="kafka";
};

该文件中username和password是定义zookeeper集群节点之间认证的用户名和密码；user_开头配置的用户kafka和密码kafka是提供给外部应用连接zookeeper使用的，后面kafka使用此用户连接zookeeper。

2.1.4 修改zkEnv.sh

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

export SERVER_JVMFLAGS="-Djava.security.auth.login.config=/mnt/datadisk/zookeeper/conf/zk_server_jaas.conf"

2.1.5 启动zk

nohup bin/zkServer.sh start &

2.2 Kafka配置

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

2.2.1 创建超级用户

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

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

正常情况打印信息：Completed updating config for entity: user-principal ‘admin’.

bin/kafka-configs.sh --zookeeper 192.168.13.202:2181 --alter --add-config 'SCRAM-SHA-256=[password=writer],SCRAM-SHA-512=[password=writer]' --entity-type users --entity-name writer

bin/kafka-configs.sh --zookeeper 192.168.13.202:2181 --alter --add-config 'SCRAM-SHA-256=[password=reader],SCRAM-SHA-512=[password=reader]' --entity-type users --entity-name reader

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

bin/zkCli.sh
# 客户端打开后执行下面命令，查看用户节点，可以看到创建成功的用户
ls /config/users

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

bin/kafka-configs.sh --zookeeper 192.168.13.202:2181 --describe --entity-type users --entity-name writer

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

2.2.2 创建jaas文件

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

# 在kafka目录下创建文件
vi config/kafka_server_jaas.conf

kafka_server_jaas.conf文件的内容为：

KafkaServer {org.apache.kafka.common.security.scram.ScramLoginModule required
username="kafka_server_admin"
password="admin";org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin-secret"
user_filebeat="123456";
};Client {org.apache.kafka.common.security.plain.PlainLoginModule required
username="kafka"
password="kafka";
};

KafkaServer配置的是kafka的账号和密码。Client配置了broker到Zookeeper的连接用户名密码，这里要和前面2.1.3节zookeeper配置中的zk_server_jaas.conf中user_kafka的账号和密码相同。中间部分配置的是PLAIN认证方式的账户和密码，其中filebeat是账户名，123456是密码。
关于这个文件内容，需要注意以下两点：

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

2.2.3 修改kafka配置文件

server.properties文件中添加如下内容:

# 启用ACL
allow.everyone.if.no.acl.found=false
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
# 设置本例中admin为超级用户；在Zookeeper的“/kafka/config/users”下存在用户
super.users=User:kafka_server_admin
# 同时启用SCRAM和PLAIN机制
sasl.enabled.mechanisms=SCRAM-SHA-256,PLAIN
# 为broker间通讯开启SCRAM机制，采用SCRAM-SHA-512算法
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
# broker间通讯使用PLAINTEXT，本例中不演示SSL配置
security.inter.broker.protocol=SASL_PLAINTEXT
# 配置listeners使用SASL_PLAINTEXT
listeners=SASL_PLAINTEXT://:9092
# 配置advertised.listeners
advertised.listeners=SASL_PLAINTEXT://192.168.13.202:9092

启动脚本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=/mnt/datadisk/kafka/config/kafka_server_jaas.conf kafka.Kafka "$@"

kafka配置文件server.properties完整配置如下：

# kafka不同节点的唯一标识
broker.id=0#认证配置
listeners=SASL_PLAINTEXT://:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
sasl.enabled.mechanisms=SCRAM-SHA-256,PLAIN
#ACL配置
allow.everyone.if.no.acl.found=false
super.users=User:kafka_server_admin
authorizer.class.name=kafka.security.authorizer.AclAuthorizer# ip为本机ip
advertised.listeners=SASL_PLAINTEXT://192.168.13.202:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
# 必改项 数据和日志分离, kafka中数据的存放目录
log.dirs=/mnt/datadisk/kafka/data
# 默认的分区数量
num.partitions=3
# 默认的副本数量
default.replication.factor=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
# zookeeper集群地址
zookeeper.connect=192.168.13.202:2181
zookeeper.connection.timeout.ms=90000
# 可以删除topic
delete.topic.enable=true
group.initial.rebalance.delay.ms=0
# 消息体大小
message.max.bytes=10485760
socket.request.max.bytes=524288000
replica.lag.time.max.ms=120000

2.2.4启动kafka

nohup bin/kafka-server-start.sh config/server.properties &

三测试连接

3.1 生产者测试

使用kafka-console-producer.sh脚本测试生产者，由于开启安全认证和授权，因此客户端需要做相应的配置。config目录下创建一个producer.conf的文件，文件内容如下：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="writer" password="writer";

创建一个测试主题test_topic：

bin/kafka-topics.sh --zookeeper 192.168.13.202:2181 --create --replication-factor 1 --partitions 1 --topic test_topic

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

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:writer --operation Write --topic 'test_topic'

2.分配producer权限

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:writer --producer --topic 'test_topic'

启动生产者发送消息：

bin/kafka-console-producer.sh --broker-list 192.168.13.202:9092 --topic test_topic --producer.config /mnt/datadisk/kafka/config/producer.conf

3.2 消费者测试

使用kafka-console-consumer.sh脚本测试生产者，由于开启安全认证和授权，因此客户端需要做相应的配置。config目录下创建一个consumer.conf的文件，文件内容如下：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="reader" password="reader";

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

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:reader --operation Read --topic 'test_topic'

2.分配consumer的权限

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:reader --consumer --topic test_topic --group test_group

启动消费者消费消息：

bin/kafka-console-consumer.sh --bootstrap-server 192.168.13.202:9092 --topic test_topic --group 'test_group' --from-beginning --consumer.config /mnt/datadisk/kafka/config/consumer.conf

3.3 beats工具连接

kafka官网提供了许多beats采集工具，用于不同场景下采集数据，感兴趣的可以去官网了解下，这里就不介绍了。笔者这里使用的是beats工具是7.9版本的，采集的数据如果要发送到kafka，只支持SASL/PLAIN认证方式，据说高版本的beats工具支持SCRAM方式的。这里介绍下filebeat-7.9.1的配置方式。PLAIN类型的用户在2.2.2章节已经创建，在filebeat中用此用户名和密码即可完成认证，授权参考3.1章节。
filebeat的配置文件filebeat.yml中输出kafka配置中加入用户名和密码，如下：

可以启动一个消费者监控filebeat采集的数据，启动filebeat，并在filebeat采集的文件中输入内容，可以看到消费者可以获取对应的消息：

3.4 logstash消费

logstash可以使用SCRAM方式认证，logstash的config目录下新增kafka-client-jaas.conf文件，文件内容如下：

KafkaClient {org.apache.kafka.common.security.scram.ScramLoginModule required
username="reader"
password="reader";
};

修改congfig目录下对应的conf处理文件，在input模块中引入SCRAM认证：

security_protocol => "SASL_PLAINTEXT"
sasl_mechanism => "SCRAM-SHA-256"
jaas_path => "/mnt/datadisk/logstash/config/kafka-client-jaas.conf"

启动生产者往logstash监控的topic中发送数据，测试如下：

下一篇：【Java版 Kafka ACL使用实战】

Kafka安全认证授权配置相关推荐

配置方法_经济权配置账户与六类经济项——经济权配置方法认识
(本文由上海复斯管理咨询公司研究并发布.) 经济权配置账户,是表达配置体自身经济权配置方案并有效呈现其经济管理模式特征的框架性工具.可拆分为收益权配置账户和经济管理权配置账户,并分别表达收益权配置和经 ...
ELK5.3+Kafka集群配置
[一]资源准备 # 3台4C*8G, 安装Zookeeper.Kafka.Logstash--Broker(input: filebeat; output: Kafka) 10.101.2.23 10 ...
汽车位置服务之kafka集群配置注意事项
1) 重复消费问题. 问题描述采用kafka读取消息进行处理时,consumer会重复读取afka队列中的数据. 问题原因 kafka的consumer消费数据时首先会从broker里读取一批 ...
Kafka教程(安装/配置/开发/面试题)
[rabbitmq教程]https://bigbird.blog.csdn.net/article/details/81436980 [Flink教程]https://blog.csdn.net/he ...
kafka集群配置(三台机器)
前提:kafka基于zookeeper配置解压安装包 ###将压缩包压缩到指定位置tar -zxvf kafka_2.11-1.01.tgz -C /usr/local/src/ 改名 mv kaf ...
kafka安装及配置过程
来来先扫一下,一码不扫何以扫天下简介 Kafka是LinkedIn开源的分布式发布-订阅消息系统,目前归属于Apache顶级项目.Kafka主要特点是基于Pull的模式来处理消息消费,追求高吞吐量, ...
linux环境kafka安装及配置
linux环境kafka下载安装下载资源安装zookeeper kafka安装及配置 kafka安装(单体) kafka集群配置方式 kafka开启kerberos认证 kafka自带zookee ...
小木大数据-kafka安装及配置
kafka安装及配置大家好,今天我要给大家介绍一下kafka的安装及配置的方式. 首先介绍一下什么叫kafka.Kafka是一个发布订阅消息系统,它的用途小木我理解的是,我们有一个温度传感器,然后k ...
kafka外网映射公司有公网ip kafka外网访问 kafka外网配置
前提条件: 由于公司要和第三方对接,采用kafka对接数据,传输用SSL加密. 环境配置: 内网三台机器:192.168.1.55,192.168.1.56,192.168.1.57 全部绑定内网的静 ...
Kafka监控工具KafkaOffsetMonitor配置及使用
一.KafkaOffsetMonitor简述 KafkaOffsetMonitor是Kafka的一款客户端消费监控工具,用来实时监控Kafka服务的Consumer以及它们所在的Partition中的 ...

Kafka安全认证授权配置

Kafka安全认证授权配置

一概述

1.1 Kafka的权限分类

1.2 实现方式

二安全认证授权配置

2.1 zookeeper配置

2.1.1 引入kafka依赖包

2.1.2修改ZK配置文件

2.1.3 创建jaas文件

2.1.4 修改zkEnv.sh

2.1.5 启动zk

2.2 Kafka配置

2.2.1 创建超级用户

2.2.2 创建jaas文件

2.2.3 修改kafka配置文件

2.2.4启动kafka

三测试连接

3.1 生产者测试

3.2 消费者测试

3.3 beats工具连接

3.4 logstash消费

Kafka安全认证授权配置相关推荐

最新文章

热门文章

Kafka安全认证授权配置

Kafka安全认证授权配置

一 概述

1.1 Kafka的权限分类

1.2 实现方式

二 安全认证授权配置

2.1 zookeeper配置

2.1.1 引入kafka依赖包

2.1.2修改ZK配置文件

2.1.3 创建jaas文件

2.1.4 修改zkEnv.sh

2.1.5 启动zk

2.2 Kafka配置

2.2.1 创建超级用户

2.2.2 创建jaas文件

2.2.3 修改kafka配置文件

2.2.4启动kafka

三 测试连接

3.1 生产者测试

3.2 消费者测试

3.3 beats工具连接

3.4 logstash消费

Kafka安全认证授权配置相关推荐

最新文章

热门文章

一概述

二安全认证授权配置

三测试连接