导语： 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能成功显示出来，即表示配置正确。