Kafka 安全认证SASL

基本概念

监听器（Listener）： Kafka Broker 可以启动多个实例，每个实例又可以监听多个端口，一个监听器就是其中一个实例 IP + 端口的组合。

监听器名字（Listener Name）：为监听器起的名字，便于描述监听器的作用。一个 Kafka Broker 可以启动多个监听器，有了名字，也就能够在需要引用监听器的地方，通过名字指明引用的具体是哪个监听器。

SASL（Simple Authentication and Security Layer）：网络协议中使用的认证层，关键词是 Authentication，用于服务访问的身份认证。SASL 只是提供了一种认证框架，具体实现上可以选择不同的机制（Mechanism）。

SCRAM（Salted Challenge Response Authentication Mechanism）：SASL 机制家族的一种，是一套包含服务器和客户端双向确认的用户认证体系，配合信道加密可以比较好的抵御中间人、拖库、伪造等攻击。

在版本0.9.0.0中，Kafka社区添加了许多功能，这些功能单独或一起使用可以提高Kafka集群的安全性。目前支持以下安全措施：

监听器安全协议（Listener Security Protocol）：Kafka 支持的安全协议有如下四种

• PLAINTEXT
• SSL
• SASL_PLAINTEXT
• SASL_SSL

1）使用 SSL 或 SASL 对客户端（生产者和消费者）、其他代理和工具与代理的连接进行身份验证。Kafka支持以下SASL机制：

• SASL/GSSAPI (Kerberos) - 从版本0.9.0.0开始
• SASL/PLAIN - 从版本0.10.0.0开始
• SASL/SCRAM-SHA-256 和SASL/SCRAM-SHA-512-从版本0.10.2.0开始
• SASL/OAUTHBEARER - 从版本2.0开始

2）从代理到 ZooKeeper 的连接身份验证

3）使用 SSL 对代理和客户端之间、代理之间或代理和工具之间传输的数据进行加密（请注意，启用 SSL 时会出现性能下降，其程度取决于 CPU 类型和 JVM 实现。）

4）客户端读/写操作的授权

安全协议由两部分组成，前半部分（SASL）决定是否验证用户身份；后半部分（PLAINTEXT/SSL）决定是否对传输信道加密，类似 HTTP 和 HTTPS。前两种安全协议不带 SASL，所以通信不需要配置用户凭证信息；而如果使用后两种安全协议通信，则必须配置用户凭证信息。

SASL机制常见场景如下

SASL 机制是配合安全协议来使用的，如果监听器选择了 SASL_PLAINTEXT 或 SASL_SSL 安全协议，则需要进一步设置 SASL 机制。由于 PLAINTEXT 或 SSL 安全协议并不会用到 SASL 机制，所以无需设置。

五种 SASL 机制中 GSSAPI 适用于使用 Kerberos 的场景， OAUTHBEARER 适用于使用 OAuth 2 授权框架的场景，SCRAM-SHA-512 可以认为是 SCRAM-SHA-256 的加强版，二者都是使用 SCRAM 机制。SCRAM 机制比 PLAIN 机制的优势有两点：
（1）提供更好的安全性（密码不会在信道上传输）
（2）可以动态增加用户凭证信息，无需重启服务。

JAAS（Java Authentication and Authorization Service）：Java 认证和授权服务，使用可插拔方式将认证授权逻辑和应用程序分离，实现认证授权服务的模块通过 JAAS 配置文件来指定。Kafka 使用 JAAS 来实现 SASL 认证机制，实现模块为 org.apache.kafka.common.security.scram.ScramLoginModule，并且将 JAAS 配置融合到了 Broker 配置文件中。

参考文章:https://xie.infoq.cn/article/0264cc0705d3c2f6cd4914e07

使用SASL进行身份验证

SASL 可以与 PLAINTEXT 或 SSL 一起使用，作为分别使用安全协议 SASL_PLAINTEXT 或 SASL_SSL 的传输层。如果使用SASL_SSL，则还必须配置SSL。

Kafka支持以下SASL机制：

• GSSAPI (Kerberos)
• PLAIN
• SCRAM-SHA-256
• SCRAM-SHA-512
• OAUTHBEARER

环境说明

本次采用3个Broker，单节点zk提供演示环境

开启SASL会影响Kafka性能，增加CPU耗损

[zk: localhost:2181(CONNECTED) 0] ls /brokers/ids
[0, 1, 2]
[zk: localhost:2181(CONNECTED) 1]

目前Kafka为3台节点

接下来为Kafka broker配置SASL

创建admin密码admin用户

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

如果想创建普通用户，可以执行下面的命令

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

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

kafka-configs 脚本是用来设置topic级别参数的。其实，它的功能还有很多。比如在这个例子中，我们使用它来创建 SASL/SCRAM 认证中的用户信息。我们可以使用下列命令来查看刚才创建的用户数据。

[root@kafka1 kafka]# sh bin/kafka-configs.sh --zookeeper localhost:2181 --describe --entity-type users  --entity-name writer

Configs for user-principal 'writer' are SCRAM-SHA-512=salt=dzFqNzd2MG5sM3NkYXFmeTZ2M2xkY3hjNA==,stored_key=5koNTaS3B9vEmlDhOf0drBJ9iDR3ZXbNeJ9M8DiWT5LfctpIDv1Ckqapu6HGbGhgSBaR/moYxi1y773IYlRRmA==,server_key=mLAo7yKfAXlraliQ4baDFITtvq6NJu5X1P2jyG7///2ffpg1WSSR9AaOPUaOZV2xqfwkCAiaz5wV1m9nPnM2VQ==,iterations=4096,SCRAM-SHA-256=salt=aDVvcjU5ZmZ5ODFmcHZtbzl6OHVnd2xxdQ==,stored_key=UpSygvAqGJNjdEjksuytrCpvYfX+CbpeDjMm+vC5PoE=,server_key=qiQIu8gdcKJGmzFtzvwYYG393FgbKR8cskrZhJPAWWI=,iterations=4096

这段命令包含了 writer 用户加密算法 SCRAM-SHA-256 以及 SCRAM-SHA-512 对应的盐值 (Salt)、ServerKey 和 StoreKey。这些都是 SCRAM 机制的术语，我们不需要了解它们的含义，因为它们并不影响我们接下来的配置。

创建 JAAS 文件

配置了用户之后，我们需要为每个 Broker 创建一个对应的 JAAS 文件。需要为每台单独的物理 Broker 机器都创建一份 JAAS 文件。

vim /usr/local/kafka/kafka-broker.jaas

KafkaServer {
org.apache.kafka.common.security.scram.ScramLoginModule required
username="admin"
password="admin";
};

这里，我们使用 admin 用户实现 Broker 之间的通信。接下来，我们来配置 Broker 的 server.properties 文件，下面这些内容，是需要单独配置的：

[root@kafka1 kafka]# cat /usr/local/kafka/config/server.properties
broker.id=0

##SASL CONFIG###
sasl.enabled.mechanisms=SCRAM-SHA-256
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256
security.inter.broker.protocol=SASL_PLAINTEXT
listeners=SASL_PLAINTEXT://:9092
advertised.listeners=SASL_PLAINTEXT://kafka3:9092
#####
...

• 开启 SCRAM 认证机制，并启用 SHA-256 算法
• 为 Broker 间通信也开启 SCRAM 认证，同样使用 SHA-256 算法；
• Broker 间通信不配置 SSL
• 设置 listeners 使用 SASL_PLAINTEXT，依然是不使用 SSL。
• SASL_PLAINTEXT://kafka3:9092 中kafka3需要添加host解析

修改启动脚本

[root@kafka1 kafka]# vim bin/kafka-server-start.sh

#添加下面的参数
export KAFKA_OPTS="-Djava.security.auth.login.config=/usr/local/kafka/kafka-broker.jaas"

创建认证文件

root@kafka3:~# cat /usr/local/kafka/config/pro.conf
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin";

生产消费者配置

生产者生产数据

root@kafka3:/usr/local/kafka/config# /usr/local/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic one  --producer.config /usr/local/kafka/config/pro.conf

>abcdocker test   #输入生产者数据

消费数据

需要指定认证文件

root@kafka3:/usr/local/kafka/config# /usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic one --from-beginning --consumer.config /usr/local/kafka/config/pro.conf

#以下为输出内容...
abcdocker test

不添加认证文件，报错如下

root@kafka3:/usr/local/kafka/config# /usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic one --from-beginning
[2023-09-01 08:49:49,354] WARN [Consumer clientId=consumer-1, groupId=console-consumer-64456] Bootstrap broker localhost:9092 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)
[2023-09-01 08:49:49,709] WARN [Consumer clientId=consumer-1, groupId=console-consumer-64456] Bootstrap broker localhost:9092 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)
[2023-09-01 08:49:50,114] WARN [Consumer clientId=consumer-1, groupId=console-consumer-64456] Bootstrap broker localhost:9092 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)