0
0

Zookeeper & Kafka 开启安全认证的配置

导语： zookeeper 和 kafka 在默认情况下，是没有开启安全认证的，那么任意客户端可以在不需要任何身份认证的情况下访问zookeeper和kafka下的各节点，甚至可以进行节点的增加，修改以及删除的动作。注意，前面的动作是基于客户端能访问服务端所在的网络，如果进行了物理隔绝或者做了防火墙限制，那前述内容就不一定成立。但是，在某些对安全加固要求比较严格的客户或者生产环境中，那就必须开启安全认证才行。除了最基本的身份认证以外，还有针对每个节点的权限访问，但本文不涉及该话题。

进入正题，先从zookeeper开始配置，zookeeper官网提供了认证配置的参考，点击下方官网地址，即可查看详情。配置分两种情况：

  1. 客户端和服务端的双向认证
  2. 服务端与服务端的双向认证

如果是非集群模式下，仅配置客户端和服务端的双向认证即可。集群模式下，则需要客户端和服务端的认证以及zookeeper服务器之间的双向认证。

以下是各配置的详细内容：

zookeeper

一. 配置 Client-Server 双向认证(官网地址)

该功能于 3.4.0 开始引入，低于 3.4.0 的版本的zookeeper则应先升级。

1. 配置zookeeper服务端

Zookeeper 使用的是Java自带的认证和授权服务(简称：JAAS)，详细内容请看官网，该链接是 Java 8 的 JAAS 的介绍。

  • 准备jaas 配置文件，包含客户端和服务端，为了简单，使用 DIGEST-MD5 认证方式。其他认证方式，请参考上面提供的官网链接。

比如新建文件server.jaas.conf，内容如下：

  1. Server {
  2.        # 使用摘要认证模块
  3.        org.apache.zookeeper.server.auth.DigestLoginModule required
  4.        # 用户名：super，密码：adminsecret
  5.        user_super="adminsecret"
  6.        # 用户名：bob，密码：bobsecret
  7.        user_bob="bobsecret";
  8. };
  1. 注意：上面的*.jaas.conf文件中的注释需要删掉，否则会导致程序启动失败；大括号里面，最后一行的分号(;)必须得存在，否则会出现找不到认证配置的错误
  • Server 端修改配置 zoo.cfg(kafka自带的配置文件名为zookeeper.properties)
  1. # 强制进行SASL认证
  2. sessionRequireClientSASLAuth=true
  4. authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
  • 修改 zookeeper 的启动脚本，增加jvm参数，比如使用的是kafka自带的zookeeper，则可修改 kafka/bin/kafka-run-class.sh 文件的最后一段内容：
  1. #  新增变量SERVER_JVMFLAGS
  2. SERVER_JVMFLAGS="-Djava.security.auth.login.config=/{path}/server.jaas.conf"
  4. # Launch mode
  5. if [ "x$DAEMON_MODE" = "xtrue" ]; then
  6.   nohup "$JAVA" $SERVER_JVMFLAGS $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp "$CLASSPATH" $KAFKA_OPTS "$@" > "$CONSOLE_OUTPUT_FILE" 2>&1 < /dev/null &
  7. else
  8.   exec "$JAVA" $SERVER_JVMFLAGS $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp "$CLASSPATH" $KAFKA_OPTS "$@"
  9. fi

/{path}/server.jaas.conf， {path} 是 server.jaas.conf 文件存放的绝对路径

2. 配置客户端

新建client.jass.conf，内容如下：

  1. Client {
  2.        # 使用摘要认证模块，与 server.jaas.conf 保持一致
  3.        org.apache.zookeeper.server.auth.DigestLoginModule required
  4.        # 用户：bob，与 server.jaas.conf 保持一致
  5.        username="bob"
  6.        # 密码：bobsecret，与 server.jaas.conf 保持一致
  7.        password="bobsecret";
  8. };
  1. 注意：上面的*.jaas.conf文件中的注释需要删掉，否则会导致程序启动失败；大括号里面，最后一行的分号(;)必须得存在，否则会出现找不到认证配置的错误
  • 修改 客户端 的启动脚本，增加jvm参数，参考上面的方法进行修改即可，下面列出变量的格式。
  1. CLIENT_JVMFLAGS="-Djava.security.auth.login.config=/{path}/client.jaas.conf"

/{path}/client.jaas.conf， {path} 是 server.jaas.conf 文件存放的绝对路径

二. 配置 Server-Server 双向认证(官网地址)

该配置仅适用于集群的情况

该功能于 3.4.10 开始引入，低于 3.4.10 的版本则应先升级。

1. 修改zoo.conf(kafka自带的配置文件名为zookeeper.properties)

增加如下配置：

  1. quorum.auth.enableSasl=true
  2. quorum.auth.learnerRequireSasl=true
  3. quorum.auth.serverRequireSasl=true
  4. quorum.auth.learner.saslLoginContext=QuorumLearner
  5. quorum.auth.server.saslLoginContext=QuorumServer
  6. quorum.cnxn.threads.size=20

2. 修改server.jaas.conf文件

增加如下配置：

  1. QuorumServer {
  2.        org.apache.zookeeper.server.auth.DigestLoginModule required
  3.        user_test="test";
  4. };
  5.  
  6. QuorumLearner {
  7.        org.apache.zookeeper.server.auth.DigestLoginModule required
  8.        username="test"
  9.        password="test";
  10. };

好了，zookeepeer以及客户端的配置都完成之后，我们再来配置kafka，kafka同样包含客户端和服务端的配置，请看下文。

kafka

1. 修改server.properties文件

  1. # kafka所在主机的ip地址
  2. listeners=SASL_PLAINTEXT://ip:9092
  4. security.inter.broker.protocol=SASL_PLAINTEXT
  5. sasl.mechanism.inter.broker.protocol=PLAIN
  6. sasl.enabled.mechanisms=PLAIN

2. 客户端增加以下属性

如果客户端使用了producer.properties和consumer.properties文件，则添加以下属性至各自的文件中。

  1. security.protocol=SASL_PLAINTEXT
  2. sasl.mechanism=PLAIN

如果客户端并未配置producer.properties和consumer.properties文件，因此需要在创建producer和consumer的代码中加上如上两个属性，代码大致如下：

  1.     props.put(SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
  2.     props.put("sasl.mechanism", "PLAIN");

kafka的配置也完成，重启所有服务，检查日志，是否启动正常，以及认证是否已生效。

3. kafka_topics.sh的使用

由于开启SASL后，使用kafka自带的脚本kafka-topics.sh查看topic时，如果按照未开启SASL的方式执行，命令如：

  1. ./kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --list

会提示连接超时的错误。如下：

Error while executing topic command : Timed out waiting for a node assignment. Call: listTopics
[2022-10-18 10:08:32,274] ERROR org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: listTopics
(kafka.admin.TopicCommand$)

因此，需要新增配置参数。新建一个名为topics.properties(或其他名字)的文件，配置参数如下：

  1. security.protocol=SASL_PLAINTEXT
  2. sasl.mechanism=PLAIN

这个配置和前面的producer.properties以及consumer.properties是一样的。在官网上，我没有找到这部分的描述，是通过阅读源码得来的。从kafka-topics.sh脚本中，可以看到源码的入口是：

  1. kafka.admin.TopicCommand

根据 TopicCommand.scala 源码找到 main 方法的入口，因为我们命令中使用的是 --list参数，所以找到 topicService.listTopics(opts) 这一行，跟踪该行代码，可以看到listTopics 方法并没有与参数有关的代码，所以我们可以大胆猜测，参数应该是在 topicService 对象中。接下来就对 topicService 对象的初始化进行阅读，发现和参数有关的代码如下：

  1. Admin.create(commandConfig)...KafkaAdminClient.createInternal()...
  2. channelBuilder =ClientUtils.createChannelBuilder(config, time, logContext);// 与SASL有关的就是下面这段代码publicstaticChannelBuildercreateChannelBuilder(AbstractConfig config,Time time,LogContext logContext){SecurityProtocol securityProtocol =SecurityProtocol.forName(config.getString(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));String clientSaslMechanism = config.getString(SaslConfigs.SASL_MECHANISM);returnChannelBuilders.clientChannelBuilder(securityProtocol,JaasContext.Type.CLIENT, config,null,
  3.                 clientSaslMechanism, time,true, logContext);}

从上面的代码得知，开启SASL后，需要用到两个参数：

  1. CommonClientConfigs.SECURITY_PROTOCOL_CONFIG
  2. SaslConfigs.SASL_MECHANISM

参数定义如下：

  1. publicstaticfinalString SECURITY_PROTOCOL_CONFIG ="security.protocol";publicstaticfinalString SASL_MECHANISM ="sasl.mechanism";

这个就是我们前面定义的topics.properties文件里面的内容了。
配置参数定义好了，如何引用上述配置文件呢，我们继续从头开始扒源码：

  1. // 入参是args
  2. val opts =newTopicCommandOptions(args)...// topicService 对象的第一个参数是opts.commandConfig
  3. val topicService =TopicService(opts.commandConfig, opts.bootstrapServer)...// 接下来看看opts.commandConfig的定义
  4. def commandConfig:Properties=if(has(commandConfigOpt))Utils.loadProps(options.valueOf(commandConfigOpt))elsenewProperties()// 上面的代码对 commandConfigOpt 进行判断，下面我们看下 commandConfigOpt 的定义：private val commandConfigOpt = parser.accepts("command-config","Property file containing configs to be passed to Admin Client. "+"This is used only with --bootstrap-server option for describing and altering broker configs.").withRequiredArg
  5.       .describedAs("command config property file").ofType(classOf[String])...// 由此可知，命令行的参数是 --command-config

通过上面的源码走读，我们可以得出，开启SASL后，查看topic的命令应该如下：

  1. ./kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --list --command-config="xxx/kafka/config/topics.properties"

topic能成功显示出来，即表示配置正确。

标签： kafka java-zookeeper zookeeper
本文转载自: https://blog.csdn.net/lhjlhj123123/article/details/126257810
版权归原作者 IT布道 所有， 如有侵权，请联系我们删除。
登录后发布评论

“Zookeeper & Kafka 开启安全认证的配置”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航
zookeeperkafka