1. Toy模板网首页
  2. > Toy博客

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

10月前 作者：WaiSaa 分类：Toy博客 阅读(61) 违法举报

这篇具有很好参考价值的文章主要介绍了Kafka配置Kerberos安全认证及与Java程序集成。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

Background

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

所用软件版本

查看 Kerberos 版本命令：klist -V

软件名称 版本
jdk 1.8.0_202
kafka 2.12-2.2.1
kerberos 1.15.1

1、Kerberos Server 安装

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

Kafka配置Kerberos安全认证及与Java程序集成,Java,kafka,安全,java,kerberos

  • 在线安装
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.log
 kdc = FILE:/var/log/krb5kdc.log
 admin_server = FILE:/var/log/kadmind.log

[libdefaults]
 dns_lookup_realm = false
 ticket_lifetime = 24h
 renew_lifetime = 7d
 forwardable = true
 rdns = false
 pkinit_anchors = FILE:/etc/pki/tls/certs/ca-bundle.crt
 default_realm = WLF.COM
 #default_ccache_name = KEYRING:persistent:%{uid}

[realms]
WLF.COM = {
 kdc = monkey
 admin_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 = 88
 kdc_tcp_ports = 88

[realms]
 WLF.COM = {
  #master_key_type = aes256-cts
  acl_file = /var/kerberos/krb5kdc/kadm5.acl
  dict_file = /usr/share/dict/words
  admin_keytab = /var/kerberos/krb5kdc/kadm5.keytab
  max_file = 24h
  max_renewable_life = 7d
  supported_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/

Kafka配置Kerberos安全认证及与Java程序集成,Java,kafka,安全,java,kerberos

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

Kafka配置Kerberos安全认证及与Java程序集成,Java,kafka,安全,java,kerberos

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 required
   useKeyTab=true
   keyTab="/opt/kafka_2.12-2.2.1/config/kerberos/kafka-server.keytab"
   storeKey=true
   useTicketCache=false
   principal="kafka-server/monkey@WLF.COM";
};
  • KafkaClient 配置
KafkaClient {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   keyTab="/opt/kafka_2.12-2.2.1/config/kerberos/kafka-client.keytab"
   storeKey=true
   useTicketCache=false
   principal="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
    # 先启动zk
    echo "start zookeeper ..."
    nohup $KAFKA_HOME/bin/zookeeper-server-start.sh $KAFKA_HOME/config/zookeeper.properties > $LOG_ZK 2>&1 &
    sleep 5
    # 再启动kafka
    nohup $KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/server.properties > $LOG_KAFKA 2>&1 &
}

function stop(){
    if [[ $pids ]]; then
        kill -9 $pids
        msg='Stopped success'
        custom_print $msg
    else
        msg='The service is already down'
        custom_print $msg
    fi
}

function restart(){
    if [[ $pids ]]; then
        kill -9 $pids
    fi
    run
    msg='Restart success'
    custom_print $msg
}

function status(){
    jps | egrep "Kafka|QuorumPeerMain"
    if [[ $pids ]]; then
        # 黄底蓝字
        msg='RUNNING'
        custom_print $msg
    else
        # 蓝底黑字
        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 ]]; then
        tail -f $1
    else
        msg='No log has been generated yet'
        custom_print $msg
    fi
}

# 判断输入参数
if [[ $operate = "run" || $operate = "start" ]]; then
    if [[ $pids ]]; then
        msg='The service is already running'
        custom_print $msg
    else
        run
        msg='Start success'
        custom_print $msg
    fi
elif [[ $operate = "kil" || $operate = "stop" ]]; then
    stop
elif [[ $operate = "res" || $operate = "restart" ]]; then
    restart
elif [[ $operate = "sta" || $operate = "status" ]]; then
    status
elif [[ $operate = "log" ]]; then
    if [[ $param = "zk" ]]; then
        log $LOG_ZK
    else
        log $LOG_KAFKA
    fi
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 $param
    msg="$param create success"
    custom_print $msg
elif [[ $operate = "del" ]]; then
    if [[ -z $param ]]; then
        topics=`$KAFKA_HOME/bin/kafka-topics.sh --zookeeper $ZK_SERVER:2181 --list`
        for topic in $topics; do
            if [[ $topic != "__consumer_offsets" ]]; then
                $KAFKA_HOME/bin/kafka-topics.sh --zookeeper $ZK_SERVER:2181 --delete --topic $topic> /dev/null
                msg="$topic delete success"
                custom_print $msg
            fi
        done
    else
        $KAFKA_HOME/bin/kafka-topics.sh --zookeeper $ZK_SERVER:2181 --delete --topic $param > /dev/null
        msg="$param delete success"
        custom_print $msg
    fi
elif [[ $operate = "consume" && ! -z $param ]]; then
    if [[ -z $CONF_SASL ]]; then
        $KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server $KAFKA_SERVER:9092 --from-beginning --topic $param
    else
        $KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server $KAFKA_SERVER:9092 --from-beginning --topic $param --consumer.config $CONF_SASL
    fi
elif [[ $operate = "produce" && ! -z $param ]]; then
    if [[ -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_SASL
    fi
else
    custom_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路径！文章来源地址https://www.toymoban.com/news/detail-814610.html

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");
        // sasl
        props.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();
    }
}

到了这里，关于Kafka配置Kerberos安全认证及与Java程序集成的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处： 如若内容造成侵权/违法违规/事实不符，请点击违法举报进行投诉反馈，一经查实，立即删除！

分享到：
领支付宝红包 赞助服务器费用

相关文章

  • Kerberos安全认证-连载11-HBase Kerberos安全配置及访问_kerberos hbase(1)

    Kerberos安全认证-连载11-HBase Kerberos安全配置及访问_kerberos hbase(1)

    这里需要将该值设置为false，否则Hadoop HA 不能正常启动。 skipACL=yes 跳过Zookeeper 访问控制列表（ACL）验证，允许连接zookeper后进行读取和写入。这里建议跳过，否则配置HBase 启动后不能向Zookeeper中写入数据。 这里在node3节点进行zoo.cfg文件的配置，配置完成后，将zoo.cfg文件分发

    2024年04月13日
    浏览(53)

  • Kafka增加安全验证安全认证，SASL认证，并通过spring boot-Java客户端连接配置

    公司Kafka一直没做安全验证，由于是诱捕程序故需要面向外网连接，需要增加Kafka连接验证，保证Kafka不被非法连接，故开始研究Kafka安全验证 使用Kafka版本为2.4.0版本，主要参考官方文档 官网对2.4版本安全验证介绍以及使用方式地址： https://kafka.apache.org/24/documentation.html#secu

    2024年02月01日
    浏览(66)
  • 企业级大数据安全架构（十）DBeaver连接Hive的Kerberos认证配置

    企业级大数据安全架构（十）DBeaver连接Hive的Kerberos认证配置

    1.配置本地hosts 因为Kerberos认证过程及集群服务中，很多是以主机名的形式进行访问的，所以工作机要设置hosts. 域名映射，我们通过部署CDH的集群的每一台机器都已经配置了host(文件为/etc/hosts)，工作机也需要配置window的host文件，如果提示无法修改，一般是需要管理员权限的原

    2024年02月21日
    浏览(46)
  • Kafka三种认证模式，Kafka 安全认证及权限控制详细配置与搭建

    Kafka三种认证模式，Kafka 安全认证及权限控制详细配置与搭建

    Kafka三种认证模式，Kafka 安全认证及权限控制详细配置与搭建。 Kafka三种认证模式 使用kerberos认证

    2024年02月04日
    浏览(52)
  • 【网络安全】图解 Kerberos：身份认证

    【网络安全】图解 Kerberos：身份认证

    Kerberos 是一种身份认证协议，被广泛运用在大数据生态中，甚至可以说是大数据身份认证的事实标准。本文将详细说明 Kerberos 原理。 Kerberos 一词来源于古希腊神话中的 Cerberus —— 守护地狱之门的三头犬。下图是 Kerberos 的 LOGO。 一句话来说，Kerberos 是一种基于加密 Ticket 的身

    2024年02月06日
    浏览(56)

  • kafka配置SASL/PLAIN 安全认证

    为zookeeper添加SASL支持，在配置文件zoo.cfg添加 新建zk_server_jaas.conf文件，为Zookeeper添加账号认证信息.这个文件你放在哪里随意，只要后面zkEnv配置正确的路径就好了。我是放在 /opt/zookeeper/conf/home 路径下。zk_server_jaas.conf文件的内容如下 Server { org.apache.kafka.common.security.plain.Plai

    2024年02月10日
    浏览(39)
  • kafka 安装 以及 Kraft 模式、安全认证配置

    kafka 安装 以及 Kraft 模式、安全认证配置

    常见安装：zookeeper + kafka zookeeper 单节点安装 ： apach官网下载对应包：apache-zookeeper-3.7.1-bin.tar.gz 修改对应配置文件 /conf/zoo_sample.cfg ，配置端口以及数据目录 sh zkServer.sh start 启动 、 sh zkServer.sh stop 停止、 sh zkServer.sh status 状态 sh zkCli.sh -server 客户端 zookeeper 集群安装： 在每个

    2024年02月10日
    浏览(50)
  • CDH数仓项目(三) —— Kerberos安全认证和Sentry权限管理

    CDH数仓项目(三) —— Kerberos安全认证和Sentry权限管理

    本文基于《CDH数仓项目(一) —— CDH安装部署搭建详细流程》和《CDH数仓项目(二) —— 用户行为数仓和业务数仓搭建》和搭建CDH数仓。本章节主要介绍基于CDH数仓的Kerberos认证和Sentry权限管理 Kerberos是一种计算机网络授权协议，用来在非安全网络中，对个人通信以安全的手段进

    2023年04月22日
    浏览(54)
  • 第一章 Kafka 配置部署及SASL_PLAINTEXT安全认证

    第一章 Kafka 配置部署及SASL_PLAINTEXT安全认证

    第一章 Kafka 配置部署及SASL_PLAINTEXT安全认证 第二章  Spring Boot 整合 Kafka消息队列 生产者 第三章  Spring Boot 整合 Kafka消息队列 消息者 Kafka下载地址：Apache Kafka 2.1、修改 Zookeeper 配置文件 config/zookeeper.properties 2.2、Zookeeper 配置文件修改内容 2.2、Zookeeper 配置文件增加配置说明

    2024年01月16日
    浏览(42)
  • Apache zookeeper kafka 开启SASL安全认证_kafka开启认证(1)

    Apache zookeeper kafka 开启SASL安全认证_kafka开启认证(1)

    先自我介绍一下，小编浙江大学毕业，去过华为、字节跳动等大厂，目前阿里P7 深知大多数程序员，想要提升技能，往往是自己摸索成长，但自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！ 因此收集整理了一份《2024年最新网络安全全套学习资料》

    2024年04月23日
    浏览(35)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包，优惠每天领

二维码1

领取红包

二维码2

领红包