背景

承接上篇，继续 Kafka 的简单帐号密码认证的 SASL/SCRAM 认证方式这里会多增加一步密钥注册到 Zookeeper 的操作。

本文的实践过程中，Kafka 版本为 kafka_2.11-2.3.1 ，只对 Kafka 服务做安全认证；Zookeeper 使用自带的，不做安全认证。

关键配置列表：

1. Kafka 服务配置文件 server.propertis，配置认证协议及认证实现类；
2. Kafka 的 jaas.config 认证配置文件，登录类，超管密码和管理的帐号密码列表；
3. Kafka 服务启动脚本 kafka-start-server.sh，设置安全认证环境变量；
4. Kafka 认证文件中各个密钥注册。
5. Kafka 客户端连接配置认证属性。

1、Kafka 服务配置文件

进入 kafka 应用的 config 目录，修改 server.properties 文件，在尾部添加认证协议配置：

# 修改listeners
listeners=SASL_PLAINTEXT://:9092

# 权限配置
allow.everyone.if.no.acl.found=true
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
sasl.enabled.mechanisms=SCRAM-SHA-256
super.users=User:admin

配置说明：

与 Plain 的区别在于：

sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
sasl.enabled.mechanisms=SCRAM-SHA-256

即指定 SASL 的安全认证方式为 SCRAM，这是一种加密算法。有没有其他的值呢？

特别注意：这里认证监听配置 listeners 中，默认的 hostname 是 localhost ，如果需要被远程主机连接，必须改为特定 IP

listeners=SASL_PLAINTEXT://192.xx.xx.xxx:9092

这里的 IP 决定了认证流程中的

public SaslClientAuthenticator(Map<String, ?> configs, String node, Subject subject, String servicePrincipal, String host, String mechanism, boolean handshakeRequestEnable, TransportLayer transportLayer) throws IOException {
        this.node = node;
        this.subject = subject;
        this.host = host;
        }

这里的 host 属性，如果远程连接一个 添加了认证的 Kafka 服务器，而目标服务器默认是 localhost ，会无法成功连接。

2、Kafka jaas 认证配置文件

在 config 目录下创建一个 jaas 配置文件 kafka-server-jaas-scram.conf ，使用 SCRAM 帐号密码方式，配置内容如下：

KafkaServer {
  org.apache.kafka.common.security.scram.ScramLoginModule required
  username="admin"
  password="admin"
  user_admin="admin"
  user_test1="test1"
  user_test2="test2"
  user_test3="test3"
  user_test4="test4";
};

基本语法注意事项：

1. 第一行：固定名称指定 KafkaServer 端的配置。
2. 第二行：安全认证类名为 ScramLoginModule，与 Kafka 属性文件中的协议类型一致。
3. 第三、四行：服务端使用的帐号和密码。
4. 第五行，超管帐号的密码。
5. 后面都是预设普通帐号认证信息，语法为 user_真正的用户名=''密码"
6. 最后一行用户名密码定义后，必须以分号 ; 结尾。

3、Kafka 启动脚本添加认证文件路径的环境变量

Kafka 安全认证可以直接通过环境变量 -Djava.security.auth.login.config 设置，修改 Kafka 启动脚本 kafka-start-server.sh 文件最后一行，增加一个参数指向第二步的 jaas 配置文件的绝对路径：

exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/Downloads/kafka_2.11-2.3.1/config/kafka-server-jaas-scram.conf kafka.Kafka "$@"

4、启动 Kafka

第一步，启动 Zookeeper

bin/zookeeper-server-start.sh config/zookeeper.properties

第二步，向 Zookeeper 注册 jaas.config 文件中定义的各个用户的密码，这里只注册两个，超管和 test1 ：

bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[password=admin]' --entity-type users --entity-name admin

bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[password=test1]' --entity-type users --entity-name test1

这一步必须配置，否则启动 Kafka 的时候会报错：

Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-256 (org.apache.kafka.clients.NetworkClient)

第三步，启动 Kafka

bin/kafka-server-start.sh config/server.properties

5、Kafka 连接客户端认证配置

用户 Java 编写一个 Kafka 消费者，用帐号 test1/test1 进行安全认证，主要代码如下：

//默认是30000ms
kafkaProps.put(DataShareConstant.REQUEST_TIMEOUT, "5000");
kafkaProps.put("transaction.timeout.ms", "5000");
kafkaProps.put("max.block.ms", "5000"); // 该属性决定连接超时的
kafkaProps.setProperty("security.protocol", SecurityProtocol.SASL_PLAINTEXT.name);
kafkaProps.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
kafkaProps.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"test1\" password=\"test1\";");

注意，这里设置 JAAS 属性需要与 Kafka Server 中的属性一致：

1. 安全认证协议类型：security.protocol，SASL
2. SASL 的协议类型，PLAIN
3. SASL 认证配置信息，PlainLoginModule，帐号密码来自第二步配置预设的帐号。

值得注意的是，在做 Kafka 连通性测试时，因为 Kafka 存在失败重试机制，建议用同步请求，配置 max.block.ms 设置连接超时时间，迅速获取连接结构。

它默认是一分钟，如果需要通过页面配置 Kafka 信息，这个默认时间是不友好的。

启示录

Plain 模式的 SASL 安全认证实践起来比较容易，主要就是注意客户端和服务端协议类型的一致。

有几个常见问题：

1、如果服务器配置是 PLAIN，而客户端使用 SCRAM，则会报异常：

org.apache.kafka.common.errors.UnsupportedSaslMechanismException: Client SASL mechanism 'PLAIN' not enabled in the server, enabled mechanisms are [SCRAM-SHA-256]

2、客户端使用普通帐号，没有配置 ACL 权限时，默认写入 Kafka 时会报认证失败错误：Invalid username or password，这时帐号密码没问题的话，需要检查 ACL 权限。

3、Kafka 服务端的安全认证配置文件必须注册到 Zookeeper 中才能生效。