煮酒与君饮

这个屌丝很懒，什么也没留下！

热门标签

Kerberos安全认证-连载12-Kafka Kerberos安全配置及访问

作者：煮酒与君饮 | 2024-08-09 02:25:18

踩

kafka kerberos

1. Kafka配置Kerberos

2. 客户端操作Kafka

3. Java API操作Kafka

4. StructuredStreaming操作Kafka

5. Flink 操作Kafka

技术连载系列，前面内容请参考前面连载11内容:Kerberos安全认证-连载11-HBase Kerberos安全配置及访问_IT贫道的博客-CSDN博客

1. Kafka配置Kerberos

Kafka也支持通过Kerberos进行认证，避免非法用户操作读取Kafka中的数据，对Kafka进行Kerberos认证可以按照如下步骤实现。

1) 创建Kafka服务Princial主体并写入到keytab文件

在kerberos服务端node1节点执行如下命令创建Kafka服务主体:


[root@node1 ~]# kadmin.local -q "addprinc -pw 123456 kafka/node1"
[root@node1 ~]# kadmin.local -q "addprinc -pw 123456 kafka/node2"
[root@node1 ~]# kadmin.local -q "addprinc -pw 123456 kafka/node3"

在kerberos服务端node1节点执行如下命令将Kafka服务主体写入到keytab文件。


#node1节点执行命令，将主体写入到keytab
[root@node1 ~]# kadmin.local -q "ktadd -norandkey -kt /home/keytabs/kafka.service.keytab kafka/node1@EXAMPLE.COM"
[root@node1 ~]# kadmin.local -q "ktadd -norandkey -kt /home/keytabs/kafka.service.keytab kafka/node2@EXAMPLE.COM"
[root@node1 ~]# kadmin.local -q "ktadd -norandkey -kt /home/keytabs/kafka.service.keytab kafka/node3@EXAMPLE.COM"

以上命令执行完成后，在node1节点/home/keytabs目录下生成kafka.service.keytab文件，将该文件分发到各个节点并赋权，这里可以只发送到node1~node3 Kafka所在节点，为了保证各个大数据集群节点的keytabs一致，这里分发到所有节点。


[root@node1 ~]# scp /home/keytabs/kafka.service.keytab node2:/home/keytabs/
[root@node1 ~]# scp /home/keytabs/kafka.service.keytab node3:/home/keytabs/
[root@node1 ~]# scp /home/keytabs/kafka.service.keytab node4:/home/keytabs/
[root@node1 ~]# scp /home/keytabs/kafka.service.keytab node5:/home/keytabs/

分发完成后，在集群各个节点上执行如下命令，修改kafka.service.keytab密钥文件访问权限：

chmod 770 /home/keytabs/kafka.service.keytab

2) 修改配置server.properties文件

在Kafka各个节点KAFKA_HOME/config/server.properties文件中加入如下配置以支持Kerberos安全认证。


#在node1~node3所有节点单独配置
listeners=SASL_PLAINTEXT://:9092
inter.broker.listener.name=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=GSSAPI
sasl.enabled.mechanisms=GSSAPI
sasl.kerberos.service.name=kafka
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
zookeeper.set.acl=false
allow.everyone.if.no.acl.found=true

该配置需要在node1~node3所有节点配置，以上参数解释如下：

listeners=SASL_PLAINTEXT://:9092

指定kafka监听的协议和端口，SASL_PLAINTEXT表示使用SASL（Simple Authentication and Security Layer）机制进行认证和加密通信。

inter.broker.listener.name=SASL_PLAINTEXT

指定Kafka Broker 之间通信使用SASL_PLAINTEXT机制进行认证和加密通信。

sasl.mechanism.inter.broker.protocol=GSSAPI

指定broker之间通信使用SASL机制，GSSAPI是一种基于Kerberos的SASL机制，它使用GSS-API（Generic Security Services Application Programming Interface）进行认证。

sasl.enabled.mechanisms=GSSAPI

只启用了GSSAPI机制，表示Kafka只接受使用Kerberos进行认证的连接。

sasl.kerberos.service.name=kafka

指定Kafka在Kerberos中注册的服务名称，以便用于进行身份验证和授权。

authorizer.class.name=kafka.security.authorizer.AclAuthorizer

指定Kafka使用的授权器类。

zookeeper.set.acl=false

是否在ZooKeeper中设置ACL（Access Control List），这里设置为false，表示不对ZooKeeper节点设置ACL。

allow.everyone.if.no.acl.found=true

指定当没有匹配的ACL规则时，是否允许所有用户访问。

3) 准备kafka_jaas.conf文件

在node1~node3各个节点中准备kafka_jaas.conf文件，该文件配置kafka服务端和zookeeper客户端的身份验证和授权配置，由于zookeeper开启了Kerberos认证，所以这里需要进行zookeeper客户端的身份验证配置。

这里在各个kafka节点KAFKA_HOME/config/目录中创建kafka_jaas.conf文件，内容如下:


KafkaServer {
 com.sun.security.auth.module.Krb5LoginModule required
 useKeyTab=true
 storeKey=true
 keyTab="/home/keytabs/kafka.service.keytab"
 serviceName="kafka"
 principal="kafka/node1@EXAMPLE.COM";
};
Client {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
serviceName="zookeeper"
keyTab="/home/keytabs/zookeeper.service.keytab" principal="zookeeper/node3@EXAMPLE.COM";
};

以上在可以在node1 kafka客户端配置完成后，分发到node2~node3节点中，如下：


#在node1节点分发，分发后在node2、node3节点配置对应Server principal
[root@node1 config]# scp /software/kafka_2.12-3.3.1/config/kafka_jaas.conf node2:/software/kafka_2.12-3.3.1/config/
[root@node1 config]# scp /software/kafka_2.12-3.3.1/config/kafka_jaas.conf node3:/software/kafka_2.12-3.3.1/config/

注意：在node1~node3各个kafka节点中该文件中的KafkaServer对应的principal不同，为对应各个节点hostname。

4) 准备kafka_client_jaas.conf

在各个kafka节点配置kafka_client_jaas.conf配置文件，该文件作用主要是对kafka 客户端进行身份认证。这里在node1~node3节点KAFKA_HOME/config/中创建kafka_client_jaas.conf文件，内容如下：


KafkaClient {
 com.sun.security.auth.module.Krb5LoginModule required
 useKeyTab=true
 storeKey=true
 keyTab="/home/keytabs/kafka.service.keytab"
 serviceName="kafka"
 principal="kafka/node1@EXAMPLE.COM";
};

可以在node1节点配置该文件后分发到node2~node3节点中：


#分发到node2、node3节点，需要在对应节点配置对应的principal
[root@node1 config]# scp /software/kafka_2.12-3.3.1/config/kafka_client_jaas.conf node2:/software/kafka_2.12-3.3.1/config/
[root@node1 config]# scp /software/kafka_2.12-3.3.1/config/kafka_client_jaas.conf node3:/software/kafka_2.12-3.3.1/config/

以上文件分发完成后，需要在对应节点修改配置对应的Principal信息为对应的hostname。

5) 修改启动脚本kafka-server-start.sh

在kafka各个节点中配置KAFKA_HOME/bin/kafka-server-start.sh启动脚本，在该脚本中加入kafka_jaas.conf配置，加入的内容如下：


#在node1~node3各个节点都要配置kafka-server-start.sh
export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/software/kafka_2.12-3.3.1/config/kafka_jaas.conf"

6) 修改kafka操作脚本

在当前集群中node1~node3节点是kafka服务端，同时如果在这3个节点上进行kafka 命令操作，这三个节点也是kafka客户端。在操作kafka时我们通常会操作KAFKA_HOME中的kafka-topic.sh、kafka-console-producer.sh、kafka-console-consumer.sh脚本，这些脚本需要进行kerberos认证，可以通过前面配置的kafka_client_jaas.conf文件进行Kerberos认证，所以这里在各个脚本中加入如下配置，避免在操作对应脚本时没有进行认证从而没有操作权限。

在node1~node3各个kafka 客户端配置以上脚本，可以先在node1节点进行配置各文件然后分发到其他kafka客户端，对应操作文件增加如下配置:


#vim /software/kafka_2.12-3.3.1/bin/kafka-topics.sh
export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/software/kafka_2.12-3.3.1/config/kafka_client_jaas.conf"
 
#vim /software/kafka_2.12-3.3.1/bin/kafka-console-producer.sh
export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/software/kafka_2.12-3.3.1/config/kafka_client_jaas.conf"
 
#vim /software/kafka_2.12-3.3.1/bin/kafka-console-consumer.sh
export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/software/kafka_2.12-3.3.1/config/kafka_client_jaas.conf"

在node1 kafka 客户端配置完成后，将配置后的各个脚本文件分发到其他kafka客户端节点上：


[root@node1 ~]# cd /software/kafka_2.12-3.3.1/bin/
[root@node1 bin]# scp ./kafka-topics.sh ./kafka-console-producer.sh ./kafka-console-consumer.sh node2:`pwd`
[root@node1 bin]# scp ./kafka-topics.sh ./kafka-console-producer.sh ./kafka-console-consumer.sh node3:`pwd`

7) 准备client.properites

在node1~node3各个kafka 客户端中准备client.properties配置文件，该配置文件内容如下，这里将该文件创建在各个节点的/root目录下。


security.protocol=SASL_PLAINTEXT
sasl.mechanism=GSSAPI
sasl.kerberos.service.name=kafka

当Kafka通过Kerberos认证后，在执行KAFKA_HOME/bin目录下的脚本时，需要使用正确的协议、SASL机制及kerberos服务，所以这里将以上信息配置到client.properties文件中，在执行各个脚本时需要通过参数指定该文件，这样客户端可以和服务端正常通信。

在node1配置完成/root/client.properties文件后，分发到node2~node3节点中：


[root@node1 ~]# scp ./client.properties node2:`pwd`
[root@node1 ~]# scp ./client.properties node3:`pwd`

8) 启动kafka集群

启动kafka集群前需要先启动Zookeeper，然后在各个kafka服务节点启动kafka，完成kafka集群启动。操作如下：


#node3~node5各节点启动zookeeper
zkServer.sh start
 
#node1~node3各节点启动kafka
startKafka.sh

2. 客户端操作Kafka

启动Kafka集群后，通过如下命令在Kafka集群查询、创建topic以及向topic中写入数据，以下命令可以执行在node1~node3各个kafka 客户端中。


#创建kafka topic 
kafka-topics.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --create --topic test --partitions 3 --replication-factor 1 --command-config  /root/client.properties
 
#查看集群topic
kafka-topics.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --list --command-config /root/client.properties
 
#向kafka topic中写入数据
[root@node1 ~]# kafka-console-producer.sh --broker-list node1:9092,node2:9092,node3:9092 --topic test --producer.config /root/client.properties 
>1
>2
>3
>4
>5
 
#读取kafka topic中的数据
[root@node1 ~]# kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --topic test --from-beginning  --consumer.config /root/client.properties 
1
2
3
4
5

3. Java API操作Kafka

可以按照如下步骤实现Java API操作Kerberos认证的Kafka数据。

1) 准备krb5.conf文件

将node1 kerberos服务端/etc/krb5.conf文件存放在IDEA项目中的resources资源目录中或者本地Window固定的某个目录中，用于编写代码时指定访问Kerberos的Realm。

2) 准备用户keytab文件

在kerberos服务端node1节点上将生成的kafka.server.keytab文件存入到window路径中，这里放在项目resource资源目录下，后续需要该文件进行客户端认证。

3) 准备kafka_client_jaas.conf文件

将Kafka 中kafka_client_jaas.conf文件放在window中某个路径中，并修改该文件中keytab路径为window路径：


KafkaClient {
 com.sun.security.auth.module.Krb5LoginModule required
 useKeyTab=true
 storeKey=true
 keyTab="D:/idea_space/KerberosAuth/KerberosAuthKafka/src/main/resources/kafka.service.keytab"
 serviceName="kafka"
 principal="kafka/node1@EXAMPLE.COM";
};

特别需要注意的是该文件中指定window路径时使用“/”或“\\”隔开各目录，否则客户端认证时读取不到keytab文件。这里将修改后的kafka_client_jaas.conf文件存入到项目resource资源目录下。

4) 编写Java代码向Kafka topic中读写数据

编写代码前，需要在项目pom.xml中引入如下依赖：


<!-- kafka client依赖包 -->
<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>3.3.1</version>
</dependency>

Java API读写Kafka Topic代码如下：


/**
 *  Java API 操作Kerbros认证的Kafka
 *  使用 JAAS 来进行 Kerberos 认证
 *  注意：kafka_client_jaas.conf文件中的keytab文件路径需要使用双斜杠或者反单斜杠
 */
public class OperateAuthKafka {
    public static void main(String[] args) {
        //准备JAAS配置文件路径
        String kafkaClientJaasFile = "D:\\idea_space\\KerberosAuth\\KerberosAuthKafka\\src\\main\\resources\\kafka_client_jaas.conf";
        // Kerberos配置文件路径
        String krb5FilePath = "D:\\idea_space\\KerberosAuth\\KerberosAuthKafka\\src\\main\\resources\\krb5.conf";
 
        System.setProperty("java.security.auth.login.config", kafkaClientJaasFile);
        System.setProperty("java.security.krb5.conf", krb5FilePath);
 
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
        props.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //kerberos安全认证
        props.setProperty("security.protocol", "SASL_PLAINTEXT");
        props.setProperty("sasl.mechanism", "GSSAPI");
        props.setProperty("sasl.kerberos.service.name", "kafka");
 
        //向Kafka topic中发送消息
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);
        kafkaProducer.send(new ProducerRecord<>("test", "100"));
        kafkaProducer.send(new ProducerRecord<>("test", "200"));
        kafkaProducer.send(new ProducerRecord<>("test", "300"));
        kafkaProducer.send(new ProducerRecord<>("test", "400"));
        kafkaProducer.send(new ProducerRecord<>("test", "500"));
 
        kafkaProducer.close();
        System.out.println("消息发送成功");
 
        /**
         * 从Kafka topic中消费消息
         */
        props.setProperty("group.id", "test"+ UUID.randomUUID());
        //设置消费的位置，earliest表示从头开始消费，latest表示从最新的位置开始消费
        props.setProperty("auto.offset.reset", "earliest");
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
        kafkaConsumer.subscribe(Arrays.asList("test"));
        while (true) {
            // 拉取数据
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : consumerRecords) {
                // 获取数据对应的分区号
                int partition = record.partition();
                // 对应数据值
                String value = record.value();
                //对应数据的偏移量
                long lastoffset = record.offset();
                //对应数据发送的key
                String key = record.key();
 
                System.out.println("数据的key为:"+ key +
                        ",数据的value为:" + value +
                        ",数据的offset为:"+ lastoffset +
                        ",数据的分区为:"+ partition);
            }
        }
 
    }
}

4. StructuredStreaming操作Kafka

StructuredStreaming操作Kafka时同样需要准备krb5.conf、kafka.service.keytab、kafka_client_jaas.conf配置文件，步骤参考Java API操作Kafka部分。

编写代码前，需要在项目pom.xml中引入如下依赖：


<!-- SparkSQL -->
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-sql_2.12</artifactId>
  <version>3.4.0</version>
</dependency>
<!-- Kafka 0.10+ Source For Structured Streaming-->
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-sql-kafka-0-10_2.12</artifactId>
  <version>3.4.0</version>
</dependency>

StructuredStreaming操作Kafka代码如下：


/**
 * StructuredStreaming 读取Kerberos 认证的Kafka数据
 */
public class StructuredStreamingReadAuthKafka {
    public static void main(String[] args) throws TimeoutException, StreamingQueryException {
        //准备JAAS配置文件路径
        String kafkaClientJaasFile = "D:\\idea_space\\KerberosAuth\\KerberosAuthKafka\\src\\main\\resources\\kafka_client_jaas.conf";
        // Kerberos配置文件路径
        String krb5FilePath = "D:\\idea_space\\KerberosAuth\\KerberosAuthKafka\\src\\main\\resources\\krb5.conf";
 
        System.setProperty("java.security.auth.login.config", kafkaClientJaasFile);
        System.setProperty("java.security.krb5.conf", krb5FilePath);
 
        //1.创建对象
        SparkSession spark = SparkSession.builder()
                .master("local")
                .appName("kafka source")
                .config("spark.sql.shuffle.partitions", 1)
                .getOrCreate();
 
        spark.sparkContext().setLogLevel("Error");
 
        //2.读取kafka 数据
        Dataset<Row> df = spark.readStream()
                .format("kafka")
                .option("kafka.bootstrap.servers", "node1:9092,node2:9092,node3:9092")
                .option("subscribe", "test")
                .option("startingOffsets", "earliest")
                //kerberos安全认证
                .option("kafka.security.protocol", "SASL_PLAINTEXT")
                .option("kafka.sasl.mechanism", "GSSAPI")
                .option("kafka.sasl.kerberos.service.name", "kafka")
                .load();
 
        Dataset<Row> result = df.selectExpr("cast (key as string)", "cast (value as string)");
 
        StreamingQuery query = result.writeStream()
                .format("console")
                .start();
 
        query.awaitTermination();
    }
}

5. Flink 操作Kafka

Flink操作Kafka时同样需要准备krb5.conf、kafka.service.keytab、kafka_client_jaas.conf配置文件，步骤参考Java API操作Kafka部分。

编写代码前，需要在项目pom.xml中引入如下依赖：


<!-- Flink批和流开发依赖包 -->
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-clients</artifactId>
  <version>1.16.0</version>
</dependency>
 
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-java</artifactId>
  <version>1.16.0</version>
</dependency>
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-kafka</artifactId>
  <version>1.16.0</version>
</dependency>

Flink操作Kafka代码如下：


/**
 * Flink 读取Kerberos 认证的Kafka数据
 */
public class FlinkReadAuthKafka {
    public static void main(String[] args) throws Exception {
        //准备JAAS配置文件路径
        String kafkaClientJaasFile = "D:\\idea_space\\KerberosAuth\\KerberosAuthKafka\\src\\main\\resources\\kafka_client_jaas.conf";
        // Kerberos配置文件路径
        String krb5FilePath = "D:\\idea_space\\KerberosAuth\\KerberosAuthKafka\\src\\main\\resources\\krb5.conf";
 
        System.setProperty("java.security.auth.login.config", kafkaClientJaasFile);
        System.setProperty("java.security.krb5.conf", krb5FilePath);
 
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
        KafkaSource<Tuple2<String, String>> kafkaSource = KafkaSource.<Tuple2<String, String>>builder()
                .setBootstrapServers("node1:9092,node2:9092,node3:9092") //设置Kafka 集群节点
                .setTopics("test") //设置读取的topic
                .setGroupId("my-test-group") //设置消费者组
                //kerberos安全认证
                .setProperty("security.protocol", "SASL_PLAINTEXT")
                .setProperty("sasl.mechanism", "GSSAPI")
                .setProperty("sasl.kerberos.service.name", "kafka")
                
                .setStartingOffsets(OffsetsInitializer.earliest()) //设置读取数据位置
                .setDeserializer(new KafkaRecordDeserializationSchema<Tuple2<String, String>>() {
                    //设置key ,value 数据获取后如何处理
                    @Override
                    public void deserialize(ConsumerRecord<byte[], byte[]> consumerRecord, Collector<Tuple2<String, String>> collector) throws IOException {
                        String key = null;
                        String value = null;
                        if(consumerRecord.key() != null){
                            key = new String(consumerRecord.key(), "UTF-8");
                        }
                        if(consumerRecord.value() != null){
                            value = new String(consumerRecord.value(), "UTF-8");
                        }
                        collector.collect(Tuple2.of(key, value));
                    }
 
                    //设置置返回的二元组类型
                    @Override
                    public TypeInformation<Tuple2<String, String>> getProducedType() {
                        return TypeInformation.of(new TypeHint<Tuple2<String, String>>() {
                        });
                    }
                })
                .build();
 
        DataStreamSource<Tuple2<String, String>> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka-source");
 
        kafkaDS.print();
 
        env.execute();
    }
}

欢迎点赞、评论、收藏，关注IT贫道，获取IT技术知识！

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/煮酒与君饮/article/detail/950987

Kerberos安全认证-连载12-Kafka Kerberos安全配置及访问

1. Kafka配置Kerberos

​​​​​​​2. 客户端操作Kafka

​​​​​​​3. Java API操作Kafka

4. StructuredStreaming操作Kafka

5. Flink 操作Kafka

2. 客户端操作Kafka

3. Java API操作Kafka