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

出发点

公司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

1. KafkaServer {
2.     org.apache.kafka.common.security.plain.PlainLoginModule required
3.     username="admin"
4.     password="admin-secret"
5.     user_admin="admin-secret"
6.     user_alice="alice-secret";
7. };

复制代码

将 JAAS 配置文件位置作为 JVM 参数转达给每个 Kafka 代理：
注意

以下配置需要添加到Kafka启动脚本中以添加JVM虚拟机运行参数
需要改为自己的kafka_server_jaas.conf配置文件路径

1. -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"

1. #!/bin/bash
2. # Licensed to the Apache Software Foundation (ASF) under one or more
3. # contributor license agreements.  See the NOTICE file distributed with
4. # this work for additional information regarding copyright ownership.
5. # The ASF licenses this file to You under the Apache License, Version 2.0
6. # (the "License"); you may not use this file except in compliance with
7. # the License.  You may obtain a copy of the License at
8. #
9. #    http://www.apache.org/licenses/LICENSE-2.0
10. #
11. # Unless required by applicable law or agreed to in writing, software
12. # distributed under the License is distributed on an "AS IS" BASIS,
13. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14. # See the License for the specific language governing permissions and
15. # limitations under the License.
17. if [ $# -lt 1 ];
18. then
19.         echo "USAGE: $0 [-daemon] server.properties [--override property=value]*"
20.         exit 1
21. fi
22. base_dir=$(dirname $0)
24. if [ "x$KAFKA_LOG4J_OPTS" = "x" ]; then
25.     export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/../config/log4j.properties"
26. fi
28. if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
29.     export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G -Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf"
30. fi
32. EXTRA_ARGS=${EXTRA_ARGS-'-name kafkaServer -loggc'}
34. COMMAND=$1
35. case $COMMAND in
36.   -daemon)
37.     EXTRA_ARGS="-daemon "$EXTRA_ARGS
38.     shift
39.     ;;
40.   *)
41.     ;;
42. esac
44. exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"

复制代码

在服务器属性中配置 SASL 端口和 SASL 机制，如此地方述。比方：

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

复制代码

配置 Kafka 客户端

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

1. sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
2.     username="alice" \
3.     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客户端举行连接的配置类：

1. package com.xxx.xxx.config;
3. import org.springframework.beans.factory.annotation.Autowired;
4. import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
5. import org.springframework.context.annotation.Bean;
6. import org.springframework.context.annotation.Configuration;
7. import org.springframework.kafka.annotation.EnableKafka;
8. import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
9. import org.springframework.kafka.config.KafkaListenerContainerFactory;
10. import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
11. import org.springframework.kafka.core.DefaultKafkaProducerFactory;
12. import org.springframework.kafka.core.KafkaTemplate;
13. import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
15. import java.util.Map;
17. /**
18. * @FileName: KafkaSecurityConfig.java
19. * @Description: KafkaSecurityConfig.java类说明
20. * @Date: 2023/2/1 17:16
21. */
22. @Configuration
23. @EnableKafka
24. public class KafkaSecurityConfig {
25.     public class KafkaProducerConfig {
26.         @Autowired
27.         private KafkaProperties kafkaProperties;
29.         /**
30.          * 消费者配置
31.          */
32.         @Bean
33.         public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
34.             ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
35.             Map<String, Object> props = kafkaProperties.buildConsumerProperties();
36.             props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=alice password=alice-secret;");
37.             props.put("security.protocol", "SASL_PLAINTEXT");
38.             props.put("sasl.mechanism", "PLAIN");
39.             factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
40.             factory.setConcurrency(2);
41.             factory.getContainerProperties().setPollTimeout(1500);
42.             return factory;
43.         }
46.         /**
47.          * 生产者配置
48.          */
49.         @Bean
50.         public KafkaTemplate<String, String> kafkaTemplate() {
51.             Map<String, Object> props = kafkaProperties.buildProducerProperties();
52.             props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=alice password=alice-secret;");
53.             props.put("security.protocol", "SASL_PLAINTEXT");
54.             props.put("sasl.mechanism", "PLAIN");
55.             return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
56.         }
57.     }
58. }

复制代码

学习网络安全技术的方法无非三种:
第一种是报网络安全专业，现在叫网络空间安全专业，主要专业课程:步伐计划、盘算机组成原理原理、数据布局、操作系统原理、数据库系统、 盘算机网络、人工智能、自然语言处理、社会盘算、网络安全法律法规、网络安全、内容安全、数字取证、呆板学习，多媒体技术，信息检索、舆情分析等。
第二种是自学，就是在网上找资源、找教程，大概是想办法认识一-些大佬，抱紧大腿，不过这种方法很耗时间，而且学习没有规划，可能很长一段时间感觉自己没有进步，容易劝退。
如果你对网络安全入门感爱好，那么你需要的话可以点击这里