Kafka-配置SASL/SCRAM认证

1.      创建admin用户

如果选择SCRAM机制，首先需要创建用户凭证。假设要创建一个用户名为admin，密码为admin的用户。

# 进入Kafka的安装目录

cd /usr/local/kafka-3.7.0

bin/kafka-configs.sh --bootstrap-server  hadoop01:9092 --alter --add-config 'SCRAM-SHA-256=[password=admin]' --entity-type users --entity-name admin

 

2.      创建JAAS配置文件

Kafka使用JAAS（Java Authentication and Authorization Service）文件进行安全配置。

创建一个JAAS配置文件（例如  /usr/local/kafka-3.7.0/config/kafka_server_jaas.conf），内容如下。这个文件包含了所有合法的用户及其密码。

KafkaServer {

    org.apache.kafka.common.security.scram.ScramLoginModule required

    username="admin"

    password="admin";

};

注意：

(1). KafkaServer：这部分配置用于Broker进程自身的认证，以及验证来自客户端的连接。

(2). username和password是Broker之间通信的凭证，应与第一步创建的用户凭证一致。

(3). user_用户名="密码"的格式用于定义客户端用户。务必妥善保管此文件，并设置严格的文件权限。

(4). 如果Zookeeper也配置了SASL认证，你还需要在此文件中添加Client部分，用于Broker与Zookeeper之间的认证。 

                                                                                                   

3.      修改Kafka Broker配置

cd /usr/local/kafka-3.7.0/config

编辑Kafka的 server.properties 配置文件，主要修改或添加以下参数:

# 监听器配置为SASL_PLAINTEXT

listeners=SASL_PLAINTEXT://192.168.80.131:9092

advertised.listeners=SASL_PLAINTEXT://192.168.80.131:9092

# 指定Broker间通信使用的安全协议和SASL机制

security.inter.broker.protocol=SASL_PLAINTEXT

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

# 启用SASL机制

sasl.enabled.mechanisms=SCRAM-SHA-256

# （可选但推荐）启用ACL授权，并设置超级用户

authorizer.class.name=kafka.security.authorizer.AclAuthorizer

allow.everyone.if.no.acl.found=false

super.users=User:admin

 

4.      指定JAAS文件并启动Kafka Broker

（1）在启动Kafka时，需要通过环境变量指定JAAS配置文件的路径。

修改/etc/profile文件，将上面内容添加进去：

export KAFKA_OPTS="-Djava.security.auth.login.config=${KAFKA_HOME}/config/kafka_server_jaas.conf"

 

使环境变量生效

source /etc/profile

 

（2）启动Kafka

/usr/local/kafka-3.7.0/bin/kafka-server-start.sh -daemon /usr/local/kafka-3.7.0/config/server.properties

 

5.      配置SASL客户端

客户端（生产者和消费者）也需要进行相应配置才能连接上已启用SASL的Kafka。

在客户端的配置文件中指定以下参数：

vi kafka_client_jaas

security.protocol=SASL_PLAINTEXT

sasl.mechanism=SCRAM-SHA-256

sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin";

 

6.      使用admin用户查询所有topic

/usr/local/kafka-3.7.0/bin/kafka-topics.sh --bootstrap-server 192.168.80.131:9092  --command-config  /usr/local/kafka-3.7.0/config/kafka_client_jaas.conf   --list

 

7.      使用admin用户删除topic 

/usr/local/kafka-3.7.0/bin/kafka-topics.sh --bootstrap-server 192.168.80.131:9092  --command-config  /usr/local/kafka-3.7.0/config/kafka_client_jaas.conf   --delete --topic zgy-tjg-fsy ;

 

8.      使用admin用户创建新用户iot

bin/kafka-configs.sh --bootstrap-server  hadoop01:9092 --alter --add-config 'SCRAM-SHA-256=[password=Iot@123]' --entity-type users --entity-name iot –command-config config/kafka_client_jaas.conf

 

9.      使用iot用户查看topic

在客户端的配置文件中指定以下参数：

vi kafka_client_iot_jaas

security.protocol=SASL_PLAINTEXT

sasl.mechanism=SCRAM-SHA-256

sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="iot" password="Iot@123";

 

使用iot用户查看topic，发现查看不到

/usr/local/kafka-3.7.0/bin/kafka-topics.sh --bootstrap-server 192.168.80.131:9092  --command-config  /usr/local/kafka-3.7.0/config/kafka_client_iot_jaas.conf   --list

 

10.  使用admin用户给iot设置ACL权限

# 首先授予主题读取权限

/usr/local/kafka-3.7.0/bin/kafka-acls.sh --bootstrap-server 192.168.80.131:9092   --add --allow-principal User:iot   --operation Read --operation Describe   --topic zgy-tjg-rcms  --command-config  /usr/local/kafka-3.7.0/config/kafka_client_jaas.conf ;

 

 

# 然后授予消费组权限

/usr/local/kafka-3.7.0/bin/kafka-acls.sh --bootstrap-server 192.168.80.131:9092   --add --allow-principal User:iot   --operation Read --operation Describe   --group iot-group  --command-config  /usr/local/kafka-3.7.0/config/kafka_client_jaas.conf ;

 

 

11.  使用iot用户查询所有topic

/usr/local/kafka-3.7.0/bin/kafka-topics.sh --bootstrap-server 192.168.80.131:9092  --command-config  /usr/local/kafka-3.7.0/config/kafka_client_iot_jaas.conf   --list

 

12.  使用iot用户进行消费

方式一： 使用配置文件（推荐）

在原来的命令中通过 --consumer.config 参数来指定这个配置文件。

/usr/local/kafka-3.7.0/bin/kafka-console-consumer.sh  --bootstrap-server  hadoop01:9092  --topic  zgy-tjg-rcms --consumer.config  /usr/local/kafka-3.7.0/config/kafka_client_iot_jaas.conf  --group  iot-group  --from-beginning

 

方式二：通过命令行参数直接传递。

如果你只是想做一次快速测试，也可以将认证参数直接作为属性在命令中传递。

/usr/local/kafka-3.7.0/bin/kafka-console-consumer.sh  --bootstrap-server hadoop01:9092 --topic zgy-tjg-rcms \

  --consumer-property security.protocol=SASL_PLAINTEXT \

  --consumer-property sasl.mechanism=SCRAM-SHA-256 \

  --consumer-property "sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=\"iot\" password=\"Iot@123\";" \

  --group  iot-group \

  --from-beginning