Apache zookeeper kafka 开启SASL安全认证

背景:我之前安装的kafka没有开启安全鉴权,在没有任何凭证的情况下都可以访问kafka。搜了一圈资料,发现有关于sasl、acl相关的,准备试试。

简介

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

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

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

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服务

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

SCRAM认证配置的优点:

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

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

配置zookeeper集群启用SASL

1. 配置zookeeper,启用sasl认证,cat zoo.cfg查看到如下内容:

tickTime=2000
initLimit=1
syncLimit=5
dataDir=/tmp/zookeeper/data
dataLogDir=/tmp/zookeeper/log
clientPort=2181
admin.serverPort=8888
maxClientCnxns=3000
autopurge.snapRetainCount=3
autopurge.purgeInterval=24
server.1=localhost:2888:3888
4lw.commands.whitelist=conf,stat,srvr,mntr.envi
#zk SASL
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
jaasLoginRenew=3600000
requireClientAuthScheme=sasl
zookeeper.sasl.client=true

2. 配置zookeeper JAAS

cat zk_jaas.conf文件内容如下,如果没有改文件则使用vi命令编辑


Server {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    username="admin"
    password="admin123"
    user_kafka="kafka123";
};
  1. 注意:admin用户 是zk 集群之间使用的。kafka用户 是 broker 与 zk 之间使用的。

3. 修改zkEnv.sh

将上一步添加的 jaas 配置文件添加到zookeeper的环境变量中,zkEnv.sh文件最后添加一行:
 
vim zkEnv.sh
 
ZOOBINDIR="${ZOOBINDIR:-/usr/bin}"
ZOOKEEPER_PREFIX="${ZOOBINDIR}/.."
 
# 添加如下 新增变量SERVER_JVMFLAGS:
 
export SERVER_JVMFLAGS="-Djava.security.auth.login.config=../conf/zk_jaas.conf"

配置kafka sasl动态认证

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

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

1. 创建用户

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

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

2. 查看创建的用户信息

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

./kafka-configs.sh --zookeeper localhost:2181 --describe --entity-type users 
 
#(可以单独指定某个用户 --entity-name producer,如下)
./kafka-configs.sh --zookeeper localhost:2181 --describe --entity-type users --entity-name producer

ZK客户端命令行查看:
 
./zkCli.sh  -server localhost:2181
 
ls /config/users

3. 配置kafka jaas文件

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

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

vim kafka_server_jaas.conf

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. kafka 配置文件启用SASL认证

Kafka 服务配置文件 server.propertis,配置认证协议及认证实现类
cat server.properties其它内容都注释掉,然后追加如下内容:

broker.id=0
listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://localhost: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=/tmp/kafka/logs
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=localhost:2181
zookeeper.connection.timeout.ms=60000
group.initial.rebalance.delay.ms=0

Host.name=43.138.0.199

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

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

vi kafka-server-start.sh

exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/home/lighthouse/kafka_2.12-2.2.1/config/kafka_server_jaas.conf  kafka.Kafka "$@"

 6.  kafka客户端配置

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

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

2) 生产者配置

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

config目录下创建一个producer.conf的文件,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的文件,cat 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文件

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=../config/kafka_server_jaas.conf"
 
exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleProducer "$@"
 
 
# 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=../config/kafka_server_jaas.conf"
 
exec $(dirname $0)/kafka-run-class.sh kafka.tools.ConsoleConsumer "$@"

启动kafka

nohup kafka-server-start.sh /path-to-kafka/config/server.properties &

我自己写了一个脚本,同时启动zookeeper和kafka

文件名称叫startup.sh内容如下:

cd /home/lighthouse/zk-3.4.14/zookeeper-3.4.14/bin
./zkServer.sh start
cd /home/lighthouse/kafka_2.12-2.2.1/
nohup bin/kafka-server-start.sh config/server.properties > output.txt &

zookeeper没有启动成功,我找下原因。把整个过程重新整理了一遍,发现zookeeper、kafka启动成功了。

检查验证

./zkCli.sh  -server localhost:2181

ls /brokers/ids

1. 生产者测试

1) 创建一个测试主题test_topic

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

2)查看创建的Topic

./kafka-topics.sh  --list --zookeeper localhost:2181 test_topic

./kafka-topics.sh --describe --zookeeper localhost:2181 --topic test_topic

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

这里使用producer用户认证授权,通过ACL为producer 用户分配操作test_topic权限:

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

注意allow-后面不要换行(会报错),注意整段命令要放在一行上。

  • 启动生产者发送消息

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

./kafka-console-producer.sh --broker-list localhost:9092 --topic test_topic --producer.config ../config/producer.conf

./kafka-console-producer.sh --broker-list localhost:9092 --topic test_topic --producer.config /home/lighthouse/kafka_2.12-2.2.1/config/producer.conf 

使用相对路径和绝对路径都报错了,错误相同

org.apache.kafka.common.KafkaException: Failed to construct kafka producer
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:431)
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:299)
        at kafka.tools.ConsoleProducer$.main(ConsoleProducer.scala:44)
        at kafka.tools.ConsoleProducer.main(ConsoleProducer.scala)
Caused by: java.lang.IllegalArgumentException: Login module control flag not specified in JAAS config
        at org.apache.kafka.common.security.JaasConfig.parseAppConfigurationEntry(JaasConfig.java:110)
        at org.apache.kafka.common.security.JaasConfig.<init>(JaasConfig.java:63)
        at org.apache.kafka.common.security.JaasContext.load(JaasContext.java:90)
        at org.apache.kafka.common.security.JaasContext.loadClientContext(JaasContext.java:84)
        at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:124)
        at org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:67)
        at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:99)
        at org.apache.kafka.clients.producer.KafkaProducer.newSender(KafkaProducer.java:439)
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:420)
        ... 3 more

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/463528.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

拼多多商品详情接口数据采集

拼多多商品详情接口数据采集是一个相对专业的任务&#xff0c;通常涉及到使用API接口或第三方采集工具等技术手段。以下是一些基本步骤和注意事项&#xff0c;供您参考&#xff1a; 请求示例&#xff0c;API接口接入Anzexi58 申请开发者账号&#xff1a;如果您打算使用API接口…

基于java+springboot+vue实现的停车场车位预约系统(文末源码+Lw+ppt)23-442

摘 要 本系统为用户而设计制作合庆镇停车场车位预约系统&#xff0c;旨在实现合庆镇停车场车位预约智能化、现代化管理。本合庆镇停车场车位预约管理自动化系统的开发和研制的最终目的是将合庆镇停车场车位预约的运作模式从手工记录数据转变为网络信息查询管理&#xff0c;…

高光谱数据应用于植被监测与分析与数据获

1. 常用高光谱数据 (1) 航空成像光谱仪系统 国内系统&#xff1a;MAIS、OMIS-1、OMIS-2、PHI、WHI、LASIS 国外系统&#xff1a;AIS、AVIRIS、TRWIS、GERIS、HYDICEAISA、DAIS、CASI、HYMAP (2) 航天成像光谱仪 Hyperion/EO-1 环境与减灾小卫星星座&#xff08;HJ-1B&…

ChatGPT GPT4科研应用、数据分析与机器学习、论文高效写作、AI绘图技术

原文链接&#xff1a;ChatGPT GPT4科研应用、数据分析与机器学习、论文高效写作、AI绘图技术https://mp.weixin.qq.com/s?__bizMzUzNTczMDMxMg&mid2247596849&idx3&sn111d68286f9752008bca95a5ec575bb3&chksmfa823ad6cdf5b3c0c446eceb5cf29cccc3161d746bdd9f2…

IPy,一个超级有用的Python库

更多内容在 Python 的 IPy 库是一个专门用于处理 IP 地址和网络相关操作的第三方库。它提供了丰富的功能&#xff0c;可以帮助用户轻松地进行 IP 地址的解析、格式化、比较和范围计算等操作。以下是 IPy 库的一些主要特性和功能&#xff1a; IP 地址解析与表示&#xff1a;IPy …

展开说说:Android之SharedPreferences

SharedPreferences 是一种轻量级的数据持久化存储机制。以key/value键值对形式存储在xml文件&#xff0c;用于保存一些应用程序数据。保存在 /data/data/PACKAGE_NAME/shared_prefs/xxx.Xml文件。 SharedPreferences 只能存储string&#xff0c;int&#xff0c;float&#xff…

实战!wsl 与主机网络通信,在 wsl 中搭建服务器。学了计算机网络,但只能刷刷面试题?那也太无聊了!这篇文章可以让你检测你的计网知识!

前言&#xff08;碎碎念&#xff09;&#xff1a;每次发布文章时&#xff0c;我都是一个纠结的过程。因为我给自己写笔记时&#xff0c;只需要记录自己不清晰或者易忘的知识点就可以了&#xff0c;但一旦想要作为文章发布&#xff0c;那么我就得考虑到很多人是纯新手&#xff0…

【Git】本地仓库关联远程仓库

Git 本地项目关联远程仓库 本地 本地已有项目 ● 项目 07.GitLocalTest 包含有一个js ○ test.js 远程仓库 ● 远程仓库地址 ○ https://github.com/Sonnenlicht77/gitTest.git ○ 仓库只有一个 readme.md 关联 1.本地 1.1 本地仓库 ● git init ● git add . ● gi…

[Node.js] Node.js 版本控制工具 nvm/n/fnm

文章目录 Node.jsNode.js 版本控制工具介绍nvm (Node Version Manager)nfnm (Fast Node Manager) Node.js Node.js 是一个开源与跨平台的 JavaScript 运行环境&#xff0c;它允许 JavaScript 运行&#xff0c; Node.js 使用事件驱动、非阻塞的 I/O 模型&#xff0c;使其轻量且…

HSE化工应急安全生产管理平台:衢州某巨大型化工企业的成功应用

在化工行业中&#xff0c;安全生产一直是至关重要的议题。为了提高生产安全性、降低成本并提升企业形象&#xff0c;衢州某巨大型化工企业引入了HSE化工应急安全生产管理平台&#xff0c;取得了显著的改善和获益。 该平台的核心功能包括风险管理和应急预案制定。通过对化工生产…

JVM学习-垃圾回收专题

目录 1.如何判断对象可以回收 1.1引用计数法 1.2可达性分析算法 1.3五种引用 1.4拓展&#xff1a;直接内存 2.垃圾回收算法 2.1标记清除算法 2.2标记整理算法 2.3复制 3.分代垃圾回收 3.垃圾回收器 3.1串行垃圾回收器 3.2吞吐量优先垃圾回收器 3.3响应时间优先垃圾回收器…

51单片机-AT24C02(I2C总线)

目录 一&#xff0c;介绍及元件工作原理 7.时序结构&#xff08;重要&#xff09; 8.i2C总线数据帧&#xff08;重要&#xff09; 二&#xff0c;应用 一&#xff0c;介绍及元件工作原理 1.元件介绍 2.存储器 3.地址总线和数据总线 地址总线只能一次选中一行 4.引脚及应用…

python之自动化(django)

1、安装 我用的是pip install Django 在命令行中安装 然后django-admin startproject autotext&#xff08;在命令行中&#xff09; 这句话是创建一个django 项目 然后切换到你所创建项目的目录下 输入&#xff1a; python manage.py runserver 当你出现以下错误时 You…

[沉淀之华] 自研基于SpringBoot Mybaits 构建低代码数据治理脚手架分享:涵盖数据同步、数据比对、数据归档、数据恢复为一体

文章目录 成果演示背景整体能力功能描述相关细节安装使用 成果演示 Github地址&#xff1a;数据治理脚手架 wiki&#xff1a;kg-ctl-core使用文档 背景 为什么要做这个&#xff1f; 一个老生常谈且不得不谈问题&#xff1a;随着业务日益发展&#xff0c;如果不做数据迁移&…

【大模型系列】问答理解定位(Qwen-VL/Llama2/GPT)

文章目录 1 Qwen-VL(2023, Alibaba)1.1 网络结构1.2 模型训练 2 Llama2(2023, Meta)2.1 网络结构2.1.1 MHA/GQA/MQA2.1.2 RoPE(Rotary Position Embedding, 旋转式位置编码)2.1.3 RMSNorm 2.2 推理2.2.1 集束搜索(beam search)2.2.2 RoPE外推 3 GPT系列(OpenAI) 1 Qwen-VL(2023…

深度强化学习(六)(改进价值学习)

深度强化学习(六)(改进价值学习) 一.经验回放 把智能体与环境交互的记录(即经验)储存到 一个数组里&#xff0c;事后反复利用这些经验训练智能体。这个数组被称为经验回放数组&#xff08;replay buffer&#xff09;。 具体来说, 把智能体的轨迹划分成 ( s t , a t , r t ,…

软件架构设计 C/S与B/S架构的区别

一、什么是C/S架构&#xff1f; C/S是Client/Server的缩写。服务器通常采用高性能的PC、工作站或小型机&#xff0c;并采用大型数据库系统&#xff0c;如Oracle或SQLServer。 C/S架构软件有一个特点&#xff0c;就是如果用户要使用的话&#xff0c;需要下载一个客户端&#x…

【Linux系列】计算机系统中的架构与发行版:理解与区分

&#x1f49d;&#x1f49d;&#x1f49d;欢迎来到我的博客&#xff0c;很高兴能够在这里和您见面&#xff01;希望您在这里可以感受到一份轻松愉快的氛围&#xff0c;不仅可以获得有趣的内容和知识&#xff0c;也可以畅所欲言、分享您的想法和见解。 推荐:kwan 的首页,持续学…

m3u8,一个超酷的 Python 库!

更多资料获取 &#x1f4da; 个人网站&#xff1a;ipengtao.com 大家好&#xff0c;今天为大家分享一个超酷的 Python 库 - m3u8。 Github地址&#xff1a;https://github.com/globocom/m3u8 在网络视频传输中&#xff0c;HLS&#xff08;HTTP Live Streaming&#xff09;是一…

unity内存优化之AB包篇(微信小游戏)

1.搭建资源服务器使用(HFS软件(https://www.pianshen.com/article/54621708008/)) using System.Collections; using System.Collections.Generic; using UnityEngine;using System;public class Singleton<T> where T : class, new() {private static readonly Lazy<…