kafka与zookeeper的SSL认证教程

作者 乐维社区(forum.lwops.cn)许远
在构建现代的分布式系统时,确保数据传输的安全性至关重要。Apache Kafka 和 Zookeeper 作为流行的分布式消息队列和协调服务,提供了SSL(Secure Sockets Layer)认证机制,以增强数据传输过程中的安全性。
本文将详细介绍从生成SSL证书到配置服务端和客户端的全过程,确保数据在传输过程中得到充分的保护。

一、配置Kafka账号密码:
1、首先,需要修改kafka配置文件:vim /asop/kafka/kafka_2.11-2.1.0/config/server.properties

broker.id=0
listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://10.176.31.137:9092
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=/asop/kafka/logs
num.partitions=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.connect=localhost:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0

#使用的认证协议
security.inter.broker.protocol=SASL_PLAINTEXT
#SASL机制
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
#完成身份验证的类
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
#如果没有找到ACL(访问控制列表)配置,则允许任何操作。
allow.everyone.if.no.acl.found=false
#需要开启设置超级管理员,设置visitor用户为超级管理员
super.users=User:visitor

2、其次,为server创建登录验证文件,可以根据自己爱好命名文件,如vim /asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf ,文件内容如下

KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username=“visitor”
password=“qaz@123”
user_visitor=“qaz@123”;
};

3、然后修改kafka安装目录vim /asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-start.sh,在文件最上面添加变量

export KAFKA_OPTS=" -Djava.security.auth.login.config=/asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf"
在这里插入图片描述
4、接下来为consumer和producer创建登录验证文件,可以根据爱好命名文件,如kafka_client_jaas.conf,文件内容如下(如果是程序访问,如springboot访问,可以不配置)
vim /asop/kafka/kafka_2.11-2.1.0/config/kafka_client_jaas.conf

KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username=“visitor”
password=“qaz@123”;
};

5、在consumer.properties和producer.properties里分别加上如下配置:
vim /asop/kafka/kafka_2.11-2.1.0/config/consumer.properties
vim /asop/kafka/kafka_2.11-2.1.0/config/producer.properties

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

6、修改kafka安装目录bin/kafka-console-producer.sh和bin/kafka-console-consumer.sh,在文件最上面添加变量
vim /asop/kafka/kafka_2.11-2.1.0/bin/kafka-console-producer.sh
vim /asop/kafka/kafka_2.11-2.1.0/bin/kafka-console-consumer.sh

export KAFKA_OPTS=" -Djava.security.auth.login.config=/asop/kafka/kafka_2.11-2.1.0/config/kafka_client_jaas.conf"在这里插入图片描述
7、分别启动zookeeper和kafka,至此服务端kafka用户登录验证配置完成(先关闭kafka后关闭zookeeper)

关闭服务kafka
/asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-stop.sh -daemon /asop/kafka/kafka_2.11-2.1.0/config/server.properties

启动服务kafka
#/asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-start.sh -daemon /asop/kafka/kafka_2.11-2.1.0/config/server.properties

关闭服务zookeeper-3.4.13
/asop/zk/zookeeper-3.4.13/bin/zkServer.sh stop /asop/zk/zookeeper-3.4.13/conf/zoo.cfg
启动服务zookeeper-3.4.13
/asop/zk/zookeeper-3.4.13/bin/zkServer.sh start /asop/zk/zookeeper-3.4.13/conf/zoo.cfg

8、创建及查看主题

/asop/kafka/kafka_2.11-2.1.0/bin/kafka-console-producer.sh --broker-list 10.176.31.137:9092 --topic cmdb --producer-property security.protocol=SASL_PLAINTEXT --producer-property sasl.mechanism=PLAIN
在这里插入图片描述
接收消息

/asop/kafka/kafka_2.11-2.1.0/bin/kafka-console-consumer.sh --bootstrap-server 10.176.31.137:9092 --topic cmdb --from-beginning --consumer-property security.protocol=SASL_PLAINTEXT --consumer-property sasl.mechanism=PLAIN在这里插入图片描述

二、zk和kafka配置ssal账号密码:

  1. Zookeeper 配置 SASL
    1.1 新建 zoo_jaas.conf 文件
    zoo_jaas.conf 文件名、文件所在路径没有特殊要求,一般放置在${ZOOKEEPER_HOME}/conf目录下vim /asop/zk/zookeeper-3.4.13/conf/zoo_jaas.conf

Server {
org.apache.kafka.common.security.plain.PlainLoginModule required
username=“admin”
password=“admin@12”
user_kafka=“kafka@123”;
};

    Server.username、Server.password为 Zookeeper 内部通信的用户名和密码,因此保证每个 zk 节点该属性一致即可Server.user_xxx 中 xxx 为自定义用户名,用于 zkClient 连接所使用的用户名和密码,即为 kafka 创建的用户名

1.2 配置 /asop/zk/zookeeper-3.4.13/conf/zoo.cfg 文件
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000
zookeeper.sasl.client=true

zookeeper.sasl.client 设置为 true,开启客户端身份验证,否则zoo_jaas.conf中配置的用户名将不起作用,客户端仍然可以无 jaas 文件连接,只是带有 WARNNING 而已

1.3 导入依赖包
因为使用的权限验证类为:org.apache.kafka.common.security.plain.PlainLoginModule,所以需要 kafka 相关 jar 包,新建文件夹 zk_sasl_lib,如下:从kafka/lib目录下复制以下几个jar包到zookeeper的lib和新建的zk_sasl_lib目录下:

kafka-clients-2.4.1.jar
lz4-java-1.6.0.jar
slf4j-api-1.7.28.jar
slf4j-log4j12-1.7.28.jar
snappy-java-1.1.7.3.jar

mkdir /asop/zk/zookeeper-3.4.13/zk_sasl_lib
cp /asop/kafka/kafka_2.11-2.1.0/libs/kafka-clients-2.1.0.jar /asop/zk/zookeeper-3.4.13/lib/
cp /asop/kafka/kafka_2.11-2.1.0/libs/lz4-java-1.5.0.jar /asop/zk/zookeeper-3.4.13/lib/
cp /asop/kafka/kafka_2.11-2.1.0/libs/slf4j-api-1.7.25.jar /asop/zk/zookeeper-3.4.13/lib/
cp /asop/kafka/kafka_2.11-2.1.0/libs/slf4j-log4j12-1.7.25.jar /asop/zk/zookeeper-3.4.13/lib/
cp /asop/kafka/kafka_2.11-2.1.0/libs/snappy-java-1.1.7.2.jar /asop/zk/zookeeper-3.4.13/lib/
cp /asop/kafka/kafka_2.11-2.1.0/libs/kafka-clients-2.1.0.jar /asop/zk/zookeeper-3.4.13/zk_sasl_lib
cp /asop/kafka/kafka_2.11-2.1.0/libs/lz4-java-1.5.0.jar /asop/zk/zookeeper-3.4.13/zk_sasl_lib
cp /asop/kafka/kafka_2.11-2.1.0/libs/slf4j-api-1.7.25.jar /asop/zk/zookeeper-3.4.13/zk_sasl_lib
cp /asop/kafka/kafka_2.11-2.1.0/libs/slf4j-log4j12-1.7.25.jar /asop/zk/zookeeper-3.4.13/zk_sasl_lib
cp /asop/kafka/kafka_2.11-2.1.0/libs/snappy-java-1.1.7.2.jar /asop/zk/zookeeper-3.4.13/zk_sasl_lib

chmod 755 -R /asop/zk/zookeeper-3.4.13/zk_sasl_lib/
chmod 755 -R /asop/zk/zookeeper-3.4.13/zk_sasl_lib/

1.4 修改 zkEnv.sh 文件/asop/zk/zookeeper-3.4.13/bin/zkEnv.sh
修改前:如果没有就直接加

export SERVER_JVMFLAGS=“-Xmx${ZK_SERVER_HEAP}m $SERVER_JVMFLAGS”

修改后:

for jar in /asop/zk/zookeeper-3.4.13/zk_sasl_lib/*.jar;
do
CLASSPATH=“ j a r : jar: jar:CLASSPATH”
done

export SERVER_JVMFLAGS=" -Djava.security.auth.login.config=/asop/zk/zookeeper-3.4.13/conf/zoo_jaas.conf "

重启 Zookeeper 服务即可

关闭服务zookeeper-3.4.13
/asop/zk/zookeeper-3.4.13/bin/zkServer.sh stop /asop/zk/zookeeper-3.4.13/conf/zoo.cfg
启动服务zookeeper-3.4.13
/asop/zk/zookeeper-3.4.13/bin/zkServer.sh start /asop/zk/zookeeper-3.4.13/conf/zoo.cfg

  1. Kakfa 配置 SASL
    2.1 新建 kafka_server_jaas.conf 文件
    kafka_server_jaas.conf 文件名和存放路径没有要求,一般放置在${KAFKA_HOME}/config目录下/asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf

KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username=“visitor”
password=“qaz@123”
user_visitor=“qaz@123”;
};
Client{
org.apache.kafka.common.security.plain.PlainLoginModule required
username=“kafka”
password=“kafka@123”;
};

    KafkaServer.username、KafkaServer.password 为 broker 内部通信的用户名密码,同上

KafkaServer.user_xxx 其中 xxx 必须和 KafkaServer.username 配置的用户名一致,密码也一致
KafkaServer.user_producer、KafkaServer.user_consumer 为了之后的 ACL 做准备,达到消费者生产者使用不同账号且消费者账号只能消费数据,生产者账号只能生产数据
Client.username、Client.password 填写 Zookeeper 中注册的账号密码,用于 broker 与 zk 的通信(若 zk 没有配置 SASL 可以忽略、若 zookeeper.sasl.client 为 false 也可以忽略只是带有,日志如下)

[2021-06-29 17:14:30,204] WARN SASL configuration failed: javax.security.auth.login.LoginException: No JAAS configuration section named ‘Client’ was found in specified JAAS configuration file: ‘/Users/wjun/env/kafka/config/kafka_server_jaas.conf’. Will continue connection to Zookeeper server without SASL authentication, if Zookeeper server allows it. (org.apache.zookeeper.ClientCnxn)

2.2 修改 server.properties 文件
broker.id=0
listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://192.168.157.198:9092
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=/asop/kafka/logs
num.partitions=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.connect=127.0.0.1:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0

#使用的认证协议
security.inter.broker.protocol=SASL_PLAINTEXT
#SASL机制
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
#完成身份验证的类
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
#如果没有找到ACL(访问控制列表)配置,则允许任何操作。
allow.everyone.if.no.acl.found=false
#需要开启设置超级管理员,设置visitor用户为超级管理员
super.users=User:visitor

其中 localhost 需要修改成 IP地址

super.users 配置超级用户,该用户不受之后的 ACL 配置影响

2.3 修改启动脚本
修改 kafka-server-start.sh 文件,使之加载到 kafka_server_jaas.conf 文件/asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-start.sh

修改前:

if [ “x$KAFKA_HEAP_OPTS” = “x” ]; then
export KAFKA_HEAP_OPTS=“-Xmx1G -Xms1G”
fi

修改后:
(先在首行加这一行,如果有了就不用加了)export KAFKA_OPTS=" -Djava.security.auth.login.config=/asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf"
if [ “x$KAFKA_HEAP_OPTS” = “x” ]; then
export KAFKA_HEAP_OPTS=“-Xmx1G -Xms1G -Djava.security.auth.login.config=/asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf”
fi

设置zookeeper的ACL规则
/asop/zk/zookeeper-3.4.13/bin/zkCli.sh #进入zk的命令行模式

addauth digest admin:admin@12 #切换登陆用户(超级管理员是在zk的配置文件/asop/zk/zookeeper-3.4.13/conf/zoo_jaas.conf里面)

setAcl / ip:127.0.0.1:cdrwa,auth:kafka:kafka@123:cdrwa #(设置可以登陆的IP和用户账号密码,admin是上面的zk的配置文件里面定义的管理员,Kafka用户是/asop/kafka/kafka_2.11-2.1.0/config/kafka_server_jaas.conf文件里面的定义的kafka连接zk的用户 (Client下面的))

addauth digest kafka:kafka@123 #再切换为kafka用户再设置一次acl
setAcl / ip:127.0.0.1:cdrwa,auth:kafka:kafka@123:cdrwa

注意:如果要加白名单IP或者用户要在原来的基础上加,不然会覆盖
setAcl / ip:127.0.0.1:cdrwa,auth:kafka:kafka@123:cdrwa,auth:admin:admin@12:cdrwa,ip:1.1.1.1

需要恢复权限,不设置acl的话就运行
setAcl / world:anyone:cdrwa

重启 kafka 服务即可

关闭服务kafka
#/asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-stop.sh -daemon /asop/kafka/kafka_2.11-2.1.0/config/server.properties

启动服务kafka
#/asop/kafka/kafka_2.11-2.1.0/bin/kafka-server-start.sh -daemon /asop/kafka/kafka_2.11-2.1.0/config/server.properties

至此,完成kafka与zookeeper配置ssl认证。更多运维技巧欢迎关注乐维社区,更多运维问题也欢迎留言提问。

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

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

汇凯金业:如何判断黄金的买入时机

黄金,作为全球公认的避险资产,其价格波动受到多种因素的影响,包括经济数据、货币政策、地缘政治风险等。对于投资者而言,把握黄金的最佳买入点是实现投资收益最大化的关键。本文将探讨影响黄金价格的主要因素,并提供一…

八大排序之——计数排序全方位剖析!(小白也能轻松看懂!)

目录 1. 计数排序的思想动图 2. 从思想到代码的实现 >1.创建临时数组 >2.统计次数 >3.排序 >4.简单版本 3. 是否可以优化呢~ 4. 计数排序的时空复杂度 5.总结 计数排序的优点 计数排序的局限性 6、完结散花 个人主页:秋风起,再归来…

【腾讯云业务运营暑期实习面试题】

题目: (全程大概50来分钟左右,面试官挺好的,不清楚的问题也在一直引导我,总体来说非常好,挺喜欢这个面试官的) 1、自我介绍 2、讲讲文件的权限以及把目录下所有文件都修改 文件的权限 rwx --&…

动态规划(DFS -> 记忆化搜索 ->动态规划)

问题一: 首先看一个最经典的问题:上台阶问题。P1255 数楼梯 - 洛谷 | 计算机科学教育新生态 (luogu.com.cn) 我们首先看一下,如何用DFS的方法进行解题。 假设我们要上到第5级台阶: 可以看出上到第五级台阶时,可能是…

oak相机使用oak官网方式标定

目录 一、depthai ROS驱动 一、depthai ROS驱动 (1)驱动下载地址:2. C 开发快速上手 — DepthAI Docs 0.3.0.0 documentation sudo apt install ./depthai_2.17.1_arm64.deb //运行 Python3 utilities/cam_test.py -mres 400 -cams rgb,m …

探索大模型:袋鼠云在 Text To SQL 上的实践与优化

Text To SQL 指的是将自然语言转化为能够在关系型数据库中执行的结构化查询语言(简称 SQL)。近年来,伴随人工智能大模型技术的不断进步,Text To SQL 任务的成功率显著提升,这得益于大模型的推理、理解以及指令遵循等能…

自闭症学校排名前十:为星儿点亮未来

在自闭症教育领域,有许多优秀的学校和机构为自闭症儿童提供着专业的帮助和支持。 星贝育园:以其独特的教育理念和个性化的教学方法脱颖而出。学校拥有一支经验丰富、富有爱心的教师团队,为孩子们提供全方位的关爱和教育。注重培养孩子的综合能…

手机通讯录大营救,恢复sim卡联系人的3个重要方法

在数字化世界的浩瀚海洋中,手机通讯录就像一艘承载着人际关系的生命之船。然而,当这艘船遭遇风浪,即sim卡上的联系人信息意外丢失时,我们该如何进行一场惊心动魄的大营救,找回那些珍贵的联系人呢?别担心&am…

Backend - C# 的日志Lognet4

目录 一、安装 log4net 插件 (一)作用 (二)操作 (三)注意 二、配置 (一)配置AssemblyInfo.cs (二)配置log4net.config 1. 创建log4net.config文件&#xff08…

空间自回归模型及 Stata 具体操作步骤

目录 一、理论原理 二、数据准备 三、程序代码及解释 四、代码运行结果 一、理论原理 空间自回归模型(Spatial Autoregressive Model,SAR)是一种用于分析具有空间相关性的数据的统计模型。它假设观测值之间的相关性不仅取决于传统的时间或…

xxl-job从2.3.0升级到2.4.1版本遇到的问题及解决方法

一、maven升级版本 <!-- xxl-job包 --> <dependency><groupId>com.xuxueli</groupId><artifactId>xxl-job-core</artifactId><version>2.4.1</version> </dependency> 二、在nacos对应服务的配置文件增加accessToken配…

精益创业方法论在创业实践中的应用:以乔布斯视角探索创新与变革的艺术——张驰咨询

史蒂夫乔布斯以其非凡的愿景、不懈的迭代精神与对产品极致的追求&#xff0c;成为无数创业者心中的灯塔。本文将借鉴乔布斯的创新思维与精益创业方法论相结合&#xff0c;构建一套融合理论与实践的深度框架&#xff0c;旨在指导创业者在不确定的市场环境中高效探索、快速验证并…

用C# 代码调整16位整数大小端的4种方法

四种方法: short BLC(short s) {byte high (byte)((s - s % 256) / 256); //数字减去 低8位, 得到的数字再除以256得到高8位byte low (byte)(s % 256); //数字对256取余数, 得到低8位byte[] change1 { high, low };return BitConverter.ToInt16(change1); }short BLC2(sh…

使用ffmpeg将一个目录下的mkv格式的视频文件转换成mp4格式

最近学剪辑&#xff0c;从BT种子下载的素材资源都是mkv格式的&#xff0c;不能直接导入到视频剪辑软件中。这种情况下需要用一些格式转换工具进行转换&#xff0c;也可以使用ffmpeg进行编辑。 ffmpeg是一个命令行工具&#xff0c;用来对本地的音频视频软件进行编辑。ffmpeg我也…

【动态规划】回文串问题

一、经验总结 对于回文串问题&#xff0c;传统的以i位置为结尾的状态表示已经不能满足要求&#xff0c;无法推导状态转移方程。应该创建一个二维dp表&#xff0c;将所有子串[i, j]的状态表示出来二维dp表的初始化和填表顺序略微复杂&#xff0c;有时需要借助网格图像分析 二、…

Web安全:SQL注入

一、SQL注入三要素 1、用户可以对输入的参数值进行修改。 2、后端不对用户输入的参数值进行严格过滤。 3、用户修改后的参数值可以被带入后端中成功执行&#xff0c;并返回一定结果。 二、SQL注入原理 简单来说&#xff0c;用户输入的值会被插入到SQL语句中&#xff0c;然后…

秋招突击——7/10——复习{}——新作{在排序数组中查找元素的第一个最后一个位置、搜索旋转排序数组}

文章目录 引言复习新作在排序数组中查找元素的第一个和最后一个位置个人实现参考实现 搜索旋转排序数组个人实现参考实现 总结 引言 复习 新作 在排序数组中查找元素的第一个和最后一个位置 题目链接 注意 非递减序列》元素是递增或者相等&#xff0c;并不是严格递增的找到…

什么软件可以AI生成PPT?交给这5款AI PPT工具就完事

话说在当下快节奏的工作中&#xff0c;PPT制作几乎已经成为不可或缺的一部分~每天不是在做PPT的路上&#xff0c;就是在改PPT的途中。 好在幸运的是&#xff0c;现在可有不少好用的AI PPT制作工具能够来帮助我们轻松应对这一难题&#xff01;今天就来给大家分享5款实在又百搭的…

Autosar网络管理:发出第一帧网络管理报文的方法

Autosar网络管理:发出第一帧网络管理报文的方法 1. 为什么要第一帧发出网络管理报文 很多OEM要求CAN网络管理第一帧发出的是网络管理报文,目的是为了快速唤醒CAN网络 2. 节点外发第一帧报文不是网络管理报文的原因 首先根据AUTOSAR CANNM规范要求,节点要能够发出网络管理…

TCP协议双向网络通讯---Python实现

本篇文章是博主在人工智能、网络通讯等领域学习时&#xff0c;用于个人学习、研究或者欣赏使用&#xff0c;并基于博主对人工智能等领域的一些理解而记录的学习摘录和笔记&#xff0c;若有不当和侵权之处&#xff0c;指出后将会立即改正&#xff0c;还望谅解。文章分类在Python…