Kafka配置Kerberos安全认证及与Java程序集成

Background

  • 本文主要介绍在 Kafka 中如何配置 Kerberos 认证，以及 java 使用 JAAS 来进行 Kerberos 认证连接。
  • 本文演示为单机版。

所用软件版本

查看 Kerberos 版本命令：klist -V

软件名称版本
jdk1.8.0_202
kafka2.12-2.2.1
kerberos1.15.1

1、Kerberos Server 安装

  • Kerberos 是一种由 MIT(麻省理工大学)提出的网络身份验证协议，它旨在通过使用密钥加密技术为客户端和服务器应用程序提供强身份验证。
  • Kerberos 是一种基于加密 Ticket 的身份认证协议，主要由三个部分组成：Key Distribution Center (即KDC)、Client 和 Service：
    客户端会先访问两次 KDC，然后再访问目标服务，如：HTTP 服务、Zookeeper 服务、Kafka 服务等。

12c395d0ab9fd49425fcfe07f585060.png

  • 在线安装
yum install krb5-server krb5-workstation krb5-libs -y
  • 配置主机名映射

在 /etc/hosts文件中新增本机映射（我这里的主机名是monkey）。

127.0.0.1 monkey
  • 配置 krb5.conf

根据需要修改 /etc/krb5.conf文件，其中WLF.COM你可以改成你需要的，还有monkey是你的主机映射。

# Configuration snippets may be placed in this directory as well
includedir /etc/krb5.conf.d/[logging]default = FILE:/var/log/krb5libs.logkdc = FILE:/var/log/krb5kdc.logadmin_server = FILE:/var/log/kadmind.log[libdefaults]dns_lookup_realm = falseticket_lifetime = 24hrenew_lifetime = 7dforwardable = truerdns = falsepkinit_anchors = FILE:/etc/pki/tls/certs/ca-bundle.crtdefault_realm = WLF.COM#default_ccache_name = KEYRING:persistent:%{uid}[realms]
WLF.COM = {kdc = monkeyadmin_server = monkey
}[domain_realm]
.monkey = WLF.COM
monkey = WLF.COM
  • 配置 kdc.conf
  • 修改/var/kerberos/krb5kdc/kdc.conf，kdc的专属配置文件。
  • Java 使用 aes256-cts 验证方式需要安装额外的 jar 包，所以为了方便不用哈。
[kdcdefaults]kdc_ports = 88kdc_tcp_ports = 88[realms]WLF.COM = {#master_key_type = aes256-ctsacl_file = /var/kerberos/krb5kdc/kadm5.acldict_file = /usr/share/dict/wordsadmin_keytab = /var/kerberos/krb5kdc/kadm5.keytabmax_file = 24hmax_renewable_life = 7dsupported_enctypes = aes128-cts:normal des3-hmac-sha1:normal arcfour-hmac:normal camellia256-cts:normal camellia128-cts:normal des-hmac-sha1:normal des-cbc-md5:normal des-cbc-crc:normal}
  • 配置 kadm5.acl
  • 修改权限相关配置文件 /var/kerberos/krb5kdc/kadm5.acl
  • 其中前一个*号是通配符，表示像名为“abc/admin”或“xxx/admin”的人都可以使用此工具（远程或本地）管理kerberos数据库，后一个跟权限有关，*表示所有权限。WLF.COM是上面配置的realm。
*/admin@WLF.COM *
  • 初始化KDC数据库
kdb5_util create -r WLF.COM -s
ll -a /var/kerberos/krb5kdc/

image.png

  • 启动 Kerberos 服务
systemctl start krb5kdc kadmin
systemctl status krb5kdc kadmin

image.png

Kerberos 服务机器上可以使用 kadmin.local 来执行各种管理的操作！

2、Kafka 开启 Kerberos 认证

所有Kerberos 相关配置文件（java连接所需），我们都放在 Kafka 的 config/kerberos/目录下的(kerberos 目录需新建)，把krb5.conf也拷贝过去。

  • 创建 keytab
cd /opt/kafka_2.12-2.2.1/
cp /etc/krb5.conf config/kerberos/cd
kadmin.local -q "add_principal -randkey kafka-server/monkey@WLF.COM"
kadmin.local -q "add_principal -randkey kafka-client@WLF.COM"
kadmin.local -q "xst -k config/kerberos/kafka-server.keytab kafka-server/monkey@WLF.COM"
kadmin.local -q "xst -k config/kerberos/kafka-client.keytab kafka-client@WLF.COM"
  • Kafka 服务端配置

修改config/server.properties配置文件，新增如下内容：

listeners=SASL_PLAINTEXT://monkey:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=GSSAPI
sasl.enabled.mechanisms=GSSAPI
sasl.kerberos.service.name=kafka-server
  • KafkaServer 配置

新建config/kerberos/kafka-server-jaas.conf文件，内容如下：

KafkaServer {com.sun.security.auth.module.Krb5LoginModule requireduseKeyTab=truekeyTab="/opt/kafka_2.12-2.2.1/config/kerberos/kafka-server.keytab"storeKey=trueuseTicketCache=falseprincipal="kafka-server/monkey@WLF.COM";
};
  • KafkaClient 配置
KafkaClient {com.sun.security.auth.module.Krb5LoginModule requireduseKeyTab=truekeyTab="/opt/kafka_2.12-2.2.1/config/kerberos/kafka-client.keytab"storeKey=trueuseTicketCache=falseprincipal="kafka-client@WLF.COM";
};
  • 修改bin/kafka-server-start.sh，倒数第二行增加如下配置：
export KAFKA_OPTS="-Dzookeeper.sasl.client=false -Dzookeeper.sasl.client.username=zk-server -Djava.security.krb5.conf=/opt/kafka_2.12-2.2.1/config/kerberos/krb5.conf -Djava.security.auth.login.config=/opt/kafka_2.12-2.2.1/config/kerberos/kafka-server-jaas.conf"
  • 修改bin/kafka-topics.sh、kafka-console-producer.sh、bin/kafka-console-consumer.sh ，倒数第二行增加如下配置：
export KAFKA_OPTS="-Djava.security.krb5.conf=/opt/kafka_2.12-2.2.1/config/kerberos/krb5.conf -Djava.security.auth.login.config=/opt/kafka_2.12-2.2.1/config/kerberos/kafka-client-jaas.conf"
  • 新建 config/kerberos/client.properties 文件，内容如下：
security.protocol=SASL_PLAINTEXT
sasl.mechanism=GSSAPI
sasl.kerberos.service.name=kafka-server
  • 附赠kafka操作脚本operate.sh
#/bin/bash:<<!
【脚本说明】
1、此脚本用于操作kafka：启动、停止、查看运行状态、重启、查看日志、查询所有主题、创建主题、删除主题、订阅或消费主题数据；
2、建议把脚本放在kafka安装目录下；
3、适用单机版。
注意安装kafka修改配置文件：
# IP替换为Kafka所在主机的IP
sed -i '31 a listeners=PLAINTEXT://localhost:9092' config/server.properties
!# kafka安装目录
KAFKA_HOME=/opt/kafka_2.12-2.2.1
# zookeeper地址
ZK_SERVER=monkey
# kafka地址
KAFKA_SERVER=monkey
# zk启动日志
LOG_ZK=$KAFKA_HOME/logs/zookeeper-run.log
# kafka启动日志
LOG_KAFKA=$KAFKA_HOME/logs/kafka-run.log
# sasl
CONF_SASL=config/kerberos/client.properties# 操作
operate=$1
# 参数
param=$2# 进程
pids=`ps -ef | egrep "Kafka|QuorumPeerMain" | egrep -v grep | awk '{print $2}'`# 提示信息
msg='Please input params [<run>|<kil>|<res>|<sta>|<log> [zk]|<list>|<add> <{topic}>|<del> [{topic}]|<consume> <{topic}>|<produce> <{topic}>]'# 定制化shell输出
function custom_print(){echo -e "\033[5;34m ***** \033[0m"echo -e "\033[32m $@ ! \033[0m"echo -e "\033[5;34m ***** \033[0m"
}function run(){rm -rf $LOG_ZK $LOG_KAFKA# 先启动zkecho "start zookeeper ..."nohup $KAFKA_HOME/bin/zookeeper-server-start.sh $KAFKA_HOME/config/zookeeper.properties > $LOG_ZK 2>&1 &sleep 5# 再启动kafkanohup $KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties > $LOG_KAFKA 2>&1 &
}function stop(){if [[ $pids ]]; thenkill -9 $pidsmsg='Stopped success'custom_print $msgelsemsg='The service is already down'custom_print $msgfi
}function restart(){if [[ $pids ]]; thenkill -9 $pidsfirunmsg='Restart success'custom_print $msg
}function status(){jps | egrep "Kafka|QuorumPeerMain"if [[ $pids ]]; then# 黄底蓝字msg='RUNNING'custom_print $msgelse# 蓝底黑字echo -e "\033[5;34m ***** \033[0m"echo -e "\033[31m STOPPED ! \033[0m"echo -e "\033[5;34m ***** \033[0m"fi
}function log(){if [[ -e $1 ]]; thentail -f $1elsemsg='No log has been generated yet'custom_print $msgfi
}# 判断输入参数
if [[ $operate = "run" || $operate = "start" ]]; thenif [[ $pids ]]; thenmsg='The service is already running'custom_print $msgelserunmsg='Start success'custom_print $msgfi
elif [[ $operate = "kil" || $operate = "stop" ]]; thenstop
elif [[ $operate = "res" || $operate = "restart" ]]; thenrestart
elif [[ $operate = "sta" || $operate = "status" ]]; thenstatus
elif [[ $operate = "log" ]]; thenif [[ $param = "zk" ]]; thenlog $LOG_ZKelselog $LOG_KAFKAfi
elif [[ $operate = "list" ]]; then$KAFKA_HOME/bin/kafka-topics.sh --zookeeper $ZK_SERVER:2181 --list
elif [[ $operate = "add" && ! -z $param ]]; then$KAFKA_HOME/bin/kafka-topics.sh --create --bootstrap-server $KAFKA_SERVER:9092 --replication-factor 1 --partitions 1 --topic $parammsg="$param create success"custom_print $msg
elif [[ $operate = "del" ]]; thenif [[ -z $param ]]; thentopics=`$KAFKA_HOME/bin/kafka-topics.sh --zookeeper $ZK_SERVER:2181 --list`for topic in $topics; doif [[ $topic != "__consumer_offsets" ]]; then$KAFKA_HOME/bin/kafka-topics.sh --zookeeper $ZK_SERVER:2181 --delete --topic $topic> /dev/nullmsg="$topic delete success"custom_print $msgfidoneelse$KAFKA_HOME/bin/kafka-topics.sh --zookeeper $ZK_SERVER:2181 --delete --topic $param > /dev/nullmsg="$param delete success"custom_print $msgfi
elif [[ $operate = "consume" && ! -z $param ]]; thenif [[ -z $CONF_SASL ]]; then$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server $KAFKA_SERVER:9092 --from-beginning --topic $paramelse$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server $KAFKA_SERVER:9092 --from-beginning --topic $param --consumer.config $CONF_SASLfi
elif [[ $operate = "produce" && ! -z $param ]]; thenif [[ -z $CONF_SASL ]]; then$KAFKA_HOME/bin/kafka-console-producer.sh --broker-list $KAFKA_SERVER:9092 --topic $param else$KAFKA_HOME/bin/kafka-console-producer.sh --broker-list $KAFKA_SERVER:9092 --topic $param --producer.config $CONF_SASLfi
elsecustom_print $msg
fi

3、java 程序连接 Kafka

  • 防火墙放行88/udp端口
  • kdc服务默认端口是88。
firewall-cmd --zone=public --add-port=88/udp --permanent
firewall-cmd --reload
  • 引入maven依赖
<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.2.1</version>
</dependency>
  • 程序示例

注意：需要修改 kafka-client-jaas.conf配置文件中配置的kafka-client.keytab路径！

package com.cloudansys;import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.Future;public class TestKafkaKerberos {public static void main(String[] args) {// 消费者testConsumer();// 生产者testProducer();}private static void testConsumer() {System.setProperty("java.security.auth.login.config", "F:\\test\\kerberos\\kafka-client-jaas.conf");System.setProperty("java.security.krb5.conf", "F:\\test\\kerberos\\krb5.conf");Properties props = new Properties();props.put("bootstrap.servers", "monkey:9092");props.put("group.id", "test_group");props.put("enable.auto.commit", "true");props.put("auto.commit.interval.ms", "1000");props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");// saslprops.put("sasl.mechanism", "GSSAPI");props.put("security.protocol", "SASL_PLAINTEXT");props.put("sasl.kerberos.service.name", "kafka-server");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);String topic = "test";consumer.subscribe(Collections.singletonList(topic));while (true) {try {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> record : records) {System.out.printf("offset = %d, partition = %d, key = %s, value = %s%n",record.offset(), record.partition(), record.key(), record.value());}} catch (Exception e) {e.printStackTrace();}}}private static void testProducer() {// JAAS配置文件路径和Kerberos配置文件路径System.setProperty("java.security.auth.login.config", "F:\\test\\kerberos\\kafka-client-jaas.conf");System.setProperty("java.security.krb5.conf", "F:\\test\\kerberos\\krb5.conf");// kafka属性配置Properties props = new Properties();props.put("bootstrap.servers", "monkey:9092");props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");// kerberos安全认证props.put("security.protocol", "SASL_PLAINTEXT");props.put("sasl.mechanism", "GSSAPI");props.put("sasl.kerberos.service.name", "kafka-server");String topic = "test";String msg = "this is a test msg";KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);ProducerRecord<String, String> record = new ProducerRecord<>(topic, msg);// 发送消息记录Future<RecordMetadata> future = kafkaProducer.send(record);try {RecordMetadata metadata = future.get();System.out.printf("Message sent to Kafka topic=%s, partition=%d, offset=%d\n", metadata.topic(), metadata.partition(), metadata.offset());} catch (Exception e) {e.printStackTrace();}kafkaProducer.close();}
}

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

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

相关文章

【STM32】STM32学习笔记-串口发送和接收(27)

【STM32】STM32学习笔记-串口发送和接收(27)

00. 目录 文章目录 00. 目录01. 串口简介02. 串口相关API2.1 USART_Init2.2 USART_InitTypeDef2.3 USART_Cmd2.4 USART_SendData2.5 USART_ReceiveData 03. 串口发送接线图04. USB转串口模块05. 串口发送程序示例06. 串口发送支持printf07. 串口发送支持printf_v208.09.10. 01.…
阅读更多...
Linux网络基础及bonding实际操作

Linux网络基础及bonding实际操作

1.查看linux基础的网络配置 网关&#xff1a;route -n ip地址&#xff1a;ifconfig 或 ip a DNS服务器&#xff1a;cat /etc/resolv.conf 主机名&#xff1a;hostname 路由&#xff1a;route -n 网络连接状态&#xff1a;ss 或 netstat 2.临时修改网卡名称 3.永久修…
阅读更多...
在全志T113-i平台上实现H.265视频解码步骤详解

在全志T113-i平台上实现H.265视频解码步骤详解

H.265&#xff0c;也被称为HEVC(HighEfficiency Video Coding)&#xff0c;作为H.264的继任者&#xff0c;提供了更好的视频压缩和更高的视频质。H.265通过引入更多先进的编码技术&#xff0c;如更强大的运动估计和更高效的变换编码&#xff0c;对比H.264进行了改进。这些改进使…
阅读更多...
宏景eHR 多处 SQL注入漏洞复现

宏景eHR 多处 SQL注入漏洞复现

0x01 产品简介 宏景eHR人力资源管理软件是一款人力资源管理与数字化应用相融合，满足动态化、协同化、流程化、战略化需求的软件。 0x02 漏洞概述 宏景eHR view、trainplan_tree.jsp等接口处存在SQL注入漏洞，未经过身份认证的远程攻击者可利用此漏洞执行任意SQL指令，从而窃…
阅读更多...
Python web自动化测试框架搭建(功能接口)——通用模块

Python web自动化测试框架搭建(功能接口)——通用模块

1、通用模块&#xff1a; config.conf: 公共配置文件&#xff0c;配置报告、日志、截图路径&#xff0c;以及邮件相关配置 [report] reportpath E:\workspace\WebAutomation\src\functiontest\Report\2017-07-18 screen_path E:\workspace\WebAutomation\src\functiontest\R…
阅读更多...
vue上传文件加进度条，fake-progress一起使用

vue上传文件加进度条，fake-progress一起使用

el-upload上传过程中加进度条&#xff0c;进度条el-progress配合fake-progress一起使用&#xff0c;效果如下&#xff1a; 安装 npm install fake-progress 在用到的文件里面引用 import Fakeprogress from "fake-progress"; 这个进度条主要是假的进度条&#xff…
阅读更多...
数字信号处理 唐向宏著 pdf ＋课后答案 免费下载

数字信号处理 唐向宏著 pdf ＋课后答案 免费下载

数字信号处理——原理、实现与仿真 pdf 唐向宏著 &#xff0b;课后答案 杭州电子科技大学 费劲心思在网上花钱买的&#xff0c;共享给大家 永久链接&#xff1a;https://wwi.lanzoup.com/b0140pf4f 密码&#xff1a;aflj 里面除了有原书PDF&#xff0c;还有课后题答案
阅读更多...
OpenHarmony——基于HDF驱动框架构建的Display驱动模型

OpenHarmony——基于HDF驱动框架构建的Display驱动模型

概述 功能简介 LCD&#xff08;Liquid Crystal Display&#xff09;驱动编程&#xff0c;通过对显示器上电、初始化显示器驱动IC&#xff08;Integrated Circuit&#xff09;内部寄存器等操作&#xff0c;使其可以正常工作。 基于HDF&#xff08;Hardware Driver Foundation…
阅读更多...
数字化新时代，智慧社区再升级：低代码平台助您轻松打造

数字化新时代，智慧社区再升级：低代码平台助您轻松打造

随着科技的飞速发展&#xff0c;社区的数字化转型已成为大势所趋。党的二十大报告提出&#xff0c;要“完善网格化管理、精细化服务、信息化支撑的基层治理平台&#xff0c;健全城乡社区治理体系”&#xff0c;更是为社区数字化建设&#xff0c;打造智慧社区指明了方向和路径。…
阅读更多...
HDFS WebHDFS 读写文件分析及HTTP Chunk Transfer Coding相关问题探究

HDFS WebHDFS 读写文件分析及HTTP Chunk Transfer Coding相关问题探究

文章目录 前言需要回答的问题DataNode端基于Netty的WebHDFS Service的实现 基于重定向的文件写入流程写入一个大文件时WebHDFS和Hadoop Native的块分布差异 基于重定向的数据读取流程尝试读取一个小文件尝试读取一个大文件 读写过程中的Chunk Transfer-Encoding支持写文件使用C…
阅读更多...
性能测试?

性能测试?

一、什么是性能测试 先看下百度百科对它的定义 性能测试是通过自动化的测试工具模拟多种正常、峰值以及异常负载条件来对系统的各项性能指标进行测试。我们可以认为性能测试是&#xff1a;通过在测试环境下对系统或构件的性能进行探测&#xff0c;用以验证在生产环境下系统性…
阅读更多...
[C#]winform部署yolov5-onnx模型

[C#]winform部署yolov5-onnx模型

【官方框架地址】 https://github.com/ultralytics/yolov5 【算法介绍】 Yolov5&#xff0c;全称为You Only Look Once version 5&#xff0c;是计算机视觉领域目标检测算法的一个里程碑式模型。该模型由ultralytics团队开发&#xff0c;并因其简洁高效的特点而备受关注。Yol…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023