基本概念
监听器（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 readerkafka-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)
