出发点

公司Kafka一直没做安全验证，由于是诱捕程序故需要面向外网连接，需要增加Kafka连接验证，保证Kafka不被非法连接，故开始研究Kafka安全验证
使用Kafka版本为2.4.0版本，主要参考官方文档

官网

官网对2.4版本安全验证介绍以及使用方式地址：
https://kafka.apache.org/24/documentation.html#security

具体流程

使用 SASL/PLAIN 进行身份验证
SASL/PLAIN 是一种简单的用户名/密码身份验证机制，通常与 TLS 一起使用以进行加密以实现安全身份验证。 Kafka 支持 SASL/PLAIN 的默认实现，可以扩展用于生产用途，如此处所述。

用户名用作 ACL 等配置的身份验证。
配置 Kafka 代理
将一个经过适当修改的 JAAS 文件添加到每个 Kafka 代理的配置目录中，类似于下面的文件，在这个例子中我们称之为 kafka_server_jaas.conf： 此配置定义了两个用户（admin 和 alice）。代理使用 KafkaServer 部分中的属性用户名和密码来启动与其他代理的连接。在此示例中，admin 是代理间通信的用户。属性集user_用户名定义 连接到代理的所有用户的密码，代理验证所有客户端连接，包括 来自使用这些属性的其他经纪人的人。

重要配置kafka_server_jaas.conf

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-secret"
    user_admin="admin-secret"
    user_alice="alice-secret";
};

将 JAAS 配置文件位置作为 JVM 参数传递给每个 Kafka 代理：

注意

以下配置需要添加到Kafka启动脚本中以添加JVM虚拟机运行参数

需要改为自己的kafka_server_jaas.conf配置文件路径

-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf

我自己的Kafka的路径为/opt/kafka/bin/kafka-server-start.sh
具体内容如下
主要配置为KAFKA_HEAP_OPTS="-Xmx1G -Xms1G -Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf"

#!/bin/bash
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

if [ $# -lt 1 ];
then
        echo "USAGE: $0 [-daemon] server.properties [--override property=value]*"
        exit 1
fi
base_dir=$(dirname $0)

if [ "x$KAFKA_LOG4J_OPTS" = "x" ]; then
    export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/../config/log4j.properties"
fi

if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G -Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf"
fi

EXTRA_ARGS=${EXTRA_ARGS-'-name kafkaServer -loggc'}

COMMAND=$1
case $COMMAND in
  -daemon)
    EXTRA_ARGS="-daemon "$EXTRA_ARGS
    shift
    ;;
  *)
    ;;
esac

exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"

在服务器属性中配置 SASL 端口和 SASL 机制，如此处所述。例如：

listeners=SASL_SSL://host.name:port
    security.inter.broker.protocol=SASL_SSL
    sasl.mechanism.inter.broker.protocol=PLAIN
    sasl.enabled.mechanisms=PLAIN

配置 Kafka 客户端

要在客户端上配置 SASL 身份验证，请执行以下操作：
在 producer.properties 或 consumer.properties 中为每个客户机配置 JAAS 配置属性。 登录模块描述了生产者和消费者等客户端如何连接到 Kafka 代理。 以下是 PLAIN 机制的客户端配置示例：

sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required 
    username="alice" 
    password="alice-secret";

客户端使用选项用户名和密码进行配置 客户端连接的用户。在此示例中，客户端以用户 alice 身份连接到代理。 JVM 中的不同客户机可以通过指定不同的用户名作为不同的用户进行连接 和 中的密码。sasl.jaas.config

客户机的 JAAS 配置也可以指定为类似于代理的 JVM 参数 如此处所述。客户端使用名为 KafkaClient 的登录部分。此选项只允许一个用户访问来自 JVM 的所有客户机连接。

在生产者属性或消费者属性中配置以下属性：
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
在生产中使用 SASL/PLAIN
SASL/PLAIN 应仅与 SSL 一起使用作为传输层，以确保明文密码不会在未加密的情况下在线传输。
Kafka 中 SASL/PLAIN 的默认实现指定 JAAS 配置文件中的用户名和密码，如下所示。从 Kafka 2.0 版本开始，您可以避免在磁盘上存储明文密码 通过配置您自己的回调处理程序，这些处理程序使用配置选项和 从外部源获取用户名和密码。sasl.server.callback.handler.classsasl.client.callback.handler.class
在生产系统中，外部身份验证服务器可以实现密码身份验证。从卡夫卡2.0版本开始， 您可以通过配置 来插入自己的回调处理程序，这些处理程序使用外部身份验证服务器进行密码验证。sasl.server.callback.handler.class

使用Java客户端进行连接的配置类：

package com.xxx.xxx.config;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

import java.util.Map;

/**
 * @FileName: KafkaSecurityConfig.java
 * @Description: KafkaSecurityConfig.java类说明
 * @Date: 2023/2/1 17:16
 */
@Configuration
@EnableKafka
public class KafkaSecurityConfig {
    public class KafkaProducerConfig {
        @Autowired
        private KafkaProperties kafkaProperties;

        /**
         * 消费者配置
         */
        @Bean
        public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            Map<String, Object> props = kafkaProperties.buildConsumerProperties();
            props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=alice password=alice-secret;");
            props.put("security.protocol", "SASL_PLAINTEXT");
            props.put("sasl.mechanism", "PLAIN");
            factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
            factory.setConcurrency(2);
            factory.getContainerProperties().setPollTimeout(1500);
            return factory;
        }


        /**
         * 生产者配置
         */
        @Bean
        public KafkaTemplate<String, String> kafkaTemplate() {
            Map<String, Object> props = kafkaProperties.buildProducerProperties();
            props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=alice password=alice-secret;");
            props.put("security.protocol", "SASL_PLAINTEXT");
            props.put("sasl.mechanism", "PLAIN");
            return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
        }
    }
}