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

Posted IT布道

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了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,内容如下:

Server 
       # 使用摘要认证模块
       org.apache.zookeeper.server.auth.DigestLoginModule required
       # 用户名:super,密码:adminsecret
       user_super="adminsecret"
       # 用户名:bob,密码:bobsecret
       user_bob="bobsecret";
;

注意:上面的*.jaas.conf文件中的注释需要删掉,否则会导致程序启动失败;大括号里面,最后一行的分号(;)必须得存在,否则会出现找不到认证配置的错误

  • Server 端修改配置 zoo.cfg(kafka自带的配置文件名为zookeeper.properties)
# 强制进行SASL认证
sessionRequireClientSASLAuth=true

authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
  • 修改 zookeeper 的启动脚本,增加jvm参数,比如使用的是kafka自带的zookeeper,则可修改 kafka/bin/kafka-run-class.sh 文件的最后一段内容:
#  新增变量SERVER_JVMFLAGS
SERVER_JVMFLAGS="-Djava.security.auth.login.config=/path/server.jaas.conf"

# Launch mode
if [ "x$DAEMON_MODE" = "xtrue" ]; then
  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 &
else
  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 "$@"
fi

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

2. 配置客户端

新建client.jass.conf,内容如下:

Client 
       # 使用摘要认证模块,与 server.jaas.conf 保持一致
       org.apache.zookeeper.server.auth.DigestLoginModule required
       # 用户:bob,与 server.jaas.conf 保持一致
       username="bob"
       # 密码:bobsecret,与 server.jaas.conf 保持一致
       password="bobsecret";
;

注意:上面的*.jaas.conf文件中的注释需要删掉,否则会导致程序启动失败;大括号里面,最后一行的分号(;)必须得存在,否则会出现找不到认证配置的错误

  • 修改 客户端 的启动脚本,增加jvm参数,参考上面的方法进行修改即可,下面列出变量的格式。
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)

增加如下配置:

quorum.auth.enableSasl=true
quorum.auth.learnerRequireSasl=true
quorum.auth.serverRequireSasl=true
quorum.auth.learner.saslLoginContext=QuorumLearner
quorum.auth.server.saslLoginContext=QuorumServer
quorum.cnxn.threads.size=20

2. 修改server.jaas.conf文件

增加如下配置:

QuorumServer 
       org.apache.zookeeper.server.auth.DigestLoginModule required
       user_test="test";
;
 
QuorumLearner 
       org.apache.zookeeper.server.auth.DigestLoginModule required
       username="test"
       password="test";
;

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

kafka

1. 修改server.properties文件

# kafka所在主机的ip地址
listeners=SASL_PLAINTEXT://ip:9092

security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

2. 客户端增加以下属性

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

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

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

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

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

3. kafka_topics.sh的使用

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

./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(或其他名字)的文件,配置参数如下:

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

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

kafka.admin.TopicCommand 

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

Admin.create(commandConfig)
...
KafkaAdminClient.createInternal()
...
channelBuilder = ClientUtils.createChannelBuilder(config, time, logContext);

// 与SASL有关的就是下面这段代码
public static ChannelBuilder createChannelBuilder(AbstractConfig config, Time time, LogContext logContext) 
        SecurityProtocol securityProtocol = SecurityProtocol.forName(config.getString(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
        String clientSaslMechanism = config.getString(SaslConfigs.SASL_MECHANISM);
        return ChannelBuilders.clientChannelBuilder(securityProtocol, JaasContext.Type.CLIENT, config, null,
                clientSaslMechanism, time, true, logContext);
    

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

CommonClientConfigs.SECURITY_PROTOCOL_CONFIG
SaslConfigs.SASL_MECHANISM

参数定义如下:

public static final String SECURITY_PROTOCOL_CONFIG = "security.protocol";
public static final String SASL_MECHANISM = "sasl.mechanism";

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

// 入参是args
val opts = new TopicCommandOptions(args)
...
// topicService 对象的第一个参数是opts.commandConfig
val topicService = TopicService(opts.commandConfig, opts.bootstrapServer)
...
// 接下来看看opts.commandConfig的定义
def commandConfig: Properties = if (has(commandConfigOpt)) Utils.loadProps(options.valueOf(commandConfigOpt)) else new Properties()
// 上面的代码对 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
      .describedAs("command config property file")
      .ofType(classOf[String])
...
// 由此可知,命令行的参数是 --command-config

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

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

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

以上是关于Zookeeper & Kafka 开启安全认证的配置的主要内容,如果未能解决你的问题,请参考以下文章

kafka源码环境搭建

Zookeeper & Kafka 错误 KeeperErrorCode=NodeExists

zookeeper and kafka

Linux版kafka&zookeeper启动方式

使用脚本来管理Kafka的服务

学习笔记Kafka—— Kafka安装配置—— JDK&Zookeeper安装Zookeeper 常用操作