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

免责申明：

本文系转载，版权归原作者所有，如若侵权请联系我们进行删除！

《数据治理行业实践白皮书》下载地址：https://fs80.cn/4w2atu

《数栈V6.0产品白皮书》下载地址：https://fs80.cn/cw0iw1

想了解或咨询更多有关袋鼠云大数据产品、行业解决方案、客户案例的朋友，浏览袋鼠云官网：https://www.dtstack.com/?src=bbs

同时，欢迎对大数据开源项目有兴趣的同学加入「袋鼠云开源框架钉钉技术群」，交流最新开源技术信息，群号码：30537511，项目地址：https://github.com/DTStack