目录

 Zookeeper配置SASL

新建zoo_jaas.conf文件

配置zookeeper.properties文件（各节点均需修改）

导入依赖包

独立版zookeeper

Kafka集成版zookeeper

修改启动脚本或环境变量脚本（各节点均需修改）

独立版zookeeper

Kafka集成版zookeeper

Kafka配置SASL_SCRAM

新建kafka_server_jaas.conf文件

配置server.properties文件（各节点均需修改）

新增scram用户

修改启动脚本（各节点均需修改）

重启zookeeper及kafka

测试SASL_SCRAM/SSL认证

为主题、生产、消费脚本创建认证配置文件

开始测试

以下以kafka集成zookeeper为例。

本文仅介绍基于以上已开启SSL认证的前提下，追加SASL_PLAIN安全认证的介绍，即SASL_SSL安全认证的实现，其中部分配置与仅开启SASL/PLAIN存在差异。

 Zookeeper配置SASL

新建zoo_jaas.conf文件

• 进入zookeeper配置目录

cd /opt/tools/kafka_2.13-3.2.1/config

• 新建zoo_jaas.conf文件

vim zoo_jaas.conf

添加如下内容：

Server {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="rbt@123456"
    user_kafka="kafka@123";
};

• 说明
  • Server.username、Server.password为 Zookeeper 内部通信的用户名和密码，因此保证每个 zk 节点该属性一致即可
  • Server.user_xxx 中 xxx 为自定义用户名，用于 zkClient 连接所使用的用户名和密码，即为 kafka 创建的用户名
  • 最后一个账号密码配置的末尾，不可缺失“;”分号。即该配置文件中需有两个“;”分号。
• 复制配置文件到每个节点

scp zoo_jaas.conf root@big2:/opt/tools/kafka_2.13-3.2.1/config

配置zookeeper.properties文件（各节点均需修改）

独立版zookeeper的配置文件为zoo.conf，kafka集成版的配置文件为zookeeper.properties。

• 添加如下配置项

authProvidential.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000
zookeeper.sasl.client=true

导入依赖包

独立版zookeeper

因为使用的权限验证类为：org.apache.kafka.common.security.plain.PlainLoginModule，所以需要导入Kafka相关jar包，kafka-clients相关jar包，在部署kafka服务下的libs目录中可以找到，根据kafka不同版本，相关jar包版本会有所变化。所需要jar包如下，在zookeeper下创建目录zk_sasl_lib将jar包放入（目录名与位置可以随便，后续引用指定即可）

• 拷贝jar文件到zk_sasl_lib目录

• 编辑zookeeper环境变量脚本

vim bin/zkEnv.sh

添加如下内容：

for i in /apps/apache-zookeeper-3.8.1-bin/zk_sasl_lib/*.jar;
do
    CLASSPATH="$i:$CLASSPATH"
done

Kafka集成版zookeeper

原生具备，无需配置。

修改启动脚本或环境变量脚本（各节点均需修改）

修改启动脚本或环境变量脚本，使之加载zoo_jaas.conf文件

独立版zookeeper

• 修改环境变量脚本

/opt/tools/apache-zookeeper-3.8.0-bin/bin/zkEnv.sh

在脚本非注释内容的第一行，添加如下内容：

export SERVER_JVMFLAGS="-Djava.security.auth.login.config=/apps/kafka_3.4.1/config/zoo_jaas.conf"

• 修改前：

• 修改后：

Kafka集成版zookeeper

• 修改启动脚本

/opt/tools/kafka_2.13-3.2.1/bin/zookeeper-server-start.sh

在脚本非注释内容的第一行，添加如下内容：

export KAFKA_HEAP_OPTS="-Djava.security.auth.login.config=/apps/kafka_3.4.1/config/zoo_jaas.conf"

• 修改前：

• 修改后：

Kafka配置SASL_SCRAM

新建kafka_server_jaas.conf文件

• 进入kafka配置目录

cd /opt/tools/kafka_2.13-3.2.1/config

• 新建kafka_server_jaas.conf文件

vim kafka_server_jaas.conf

添加如下内容：

KafkaServer {
  org.apache.kafka.common.security.scram.ScramLoginModule required
    username="rbt"
    password="rbt@123456";
};
Client {
  org.apache.kafka.common.security.plain.PlainLoginModule required
    username="kafka"
    password="kafka@123";
};

• 说明
  • KafkaServer.username、KafkaServer.password 为 broker 内部通信的用户名密码，同上
  • KafkaServer.username 配置的用户名作为超级用户使用，此处不再需要以“user_xxx”的格式来声明用户；其他角色的用户不再配置到此处，而是之后使用命令进行添加。
  • Client.username、Client.password 填写 Zookeeper 中注册的账号密码，用于 broker 与 zk 的通信（若 zk 没有配置 SASL 可以忽略、若 zookeeper.sasl.client 为 false 也可以忽略只是带有警告日志）
  • Client中，主要是broker链接到zookeeper，需要和之前配置的zk_server_jaas.conf里的 user_kafka=“kafka@123”;对应，如果zk文件中配置了多个，填写其中一个即可。
• 复制配置文件到每个节点

scp kafka_server_jaas.conf root@big2:/opt/tools/kafka_2.13-3.2.1/config

配置server.properties文件（各节点均需修改）

根据实际情况，修改启动配置文件，如普通版修改server.properties，若为ssl版可能需要修改server-ssl.properties。

• 添加如下配置项

listeners=SASL_SSL://big1:19092
advertised.listeners = SASL_SSL://big1:19092
security.inter.broker.protocol=SASL_SSL
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
sasl.enabled.mechanisms=SCRAM-SHA-512
allow.everyone.if.no.acl.found=false
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
super.users=User:rbt

security.protocol=SASL_SSL
ssl.keystore.location=/opt/tools/kafka_2.13-3.2.1/config/certs/kafka.keystore
ssl.keystore.password=rbt@123456
ssl.key.password=rbt@123456
ssl.truststore.location=/opt/tools/kafka_2.13-3.2.1/config/certs/kafka.truststore
ssl.truststore.password=rbt@123456
ssl.client.auth=required
ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1
ssl.keystore.type=JKS
ssl.truststore.type=JKS
ssl.secure.random.implementation=SHA1PRNG

• 说明
  • super.users 配置超级用户，该用户不受之后的 ACL 配置影响
  • authorizer.class.name配置项，若kafka为3.0.0版本以前，则参数为kafka.security.auth.SimpleAclAuthorizer。若为3.0.0以后的版本，则应为kafka.security.authorizer.AclAuthorizer
  • listeners和advertised.listeners都需配置，不然会报错。这里的地址可以使用IP配置，如：listeners=SASL_PLAINTEXT://192.168.1.1:19092；也可以使用host配置，如：listeners=SASL_PLAINTEXT://big1:19092；亦可以如上图不加地址限制，如：listeners=SASL_PLAINTEXT://:19092。
  • allow.everyone.if.no.acl.found=true，那么所有用户都可以访问没有设置任何 ACL 的资源，类似于黑名单，即只有黑名单下的用户不能访问资源，非黑名单中的用户都可以正常访问kafka的资源；allow.everyone.if.no.acl.found=false时, 也就是默认为false,ACL的机制是白名单机制，只有白名单中的用户才能访问kafka的资源，其他用户为未授权用户。
  • Kafka支持无认证、SASL和SSL访问端口的同时开启（如PLAINTEXT://big1:9092,SSL://big1:9093,SASL_SSL://big1:9094），但为了安全考虑，将其他监听全部去掉，仅保留SASL_PLAINTEXT端口

新增scram用户

bin/kafka-configs.sh --bootstrap-server doris1:9092,doris2:9092,doris3:9092 --alter --add-config 'SCRAM-SHA-256=[password=rbt@123456],SCRAM-SHA-512=[password=rbt@123456]' --entity-type users --entity-name rbt --command-config config/jaas.properties

• 注：
  1. 密码与KafkaServer中的配置要保持一致，否则会报sasl验证失败;
  2. --bootstrap-server因启用了证书认证，因此必须使用host进行访问，使用IP地址将报错。
  3. 因启用了证书认证，因此创建用户时需配置部署SSL时所创建的客户端认证配置文件，如《Kafka SSL认证实现》中的client-ssl.properties文件，此处为jaas.properties。

修改启动脚本（各节点均需修改）

• 修改kafka启动脚本，使之加载kafka_server_jaas.conf文件

/opt/tools/kafka_2.13-3.2.1/bin/kafka-server-start.sh

脚本非注释内容的第一行，添加如下内容：

export KAFKA_HEAP_OPTS="-Djava.security.auth.login.config=/apps/kafka_3.4.1/config/kafka_server_jaas.conf"

• 修改前：

• 修改后：

重启zookeeper及kafka

• 重启zookeeper

/opt/tools/kafka_2.13-3.2.1/bin/zookeeper-server-stop.sh
/opt/tools/kafka_2.13-3.2.1/bin/zookeeper-server-start.sh -daemon /opt/tools/kafka_2.13-3.2.1/config/zookeeper.properties

• 重启kafka

/opt/tools/kafka_2.13-3.2.1/bin/kafka-server-stop.sh
/opt/tools/kafka_2.13-3.2.1/bin/kafka-server-start.sh -daemon /opt/tools/kafka_2.13-3.2.1/config/server.properties

测试SASL_SCRAM/SSL认证

为主题、生产、消费脚本创建认证配置文件

• 进入kafka配置目录

cd /opt/tools/kafka_2.13-3.2.1/config

• 新建jaas.properties文件

vim jaas.properties

添加如下内容：

security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="rbt" password="rbt@123456";
ssl.keystore.location=/opt/tools/kafka_2.13-3.2.1/config/certs/kafka.keystore
ssl.keystore.password=rbt@123456
ssl.key.password=rbt@123456
ssl.truststore.location=/opt/tools/kafka_2.13-3.2.1/config/certs/kafka.truststore
ssl.truststore.password=rbt@123456

• 注：
  1. 还有一种配置方法是将认证账号密码信息创建一个jaas.conf文件，以环境变量的形式配置到各个脚本中，同时脚本启动时再指定此配置文件来指定认证方式（配置文件中不再需要sasl.jaas.config项）；这种方式要多创建文件、修改脚本运行环境、同时还是需要指定此配置文件，较为麻烦，推荐使用以上方法。
  2. sasl.jaas.config中的用户名与密码要与KafkaServer中的配置要保持一致。同时注意最后的分号不可丢失，即使将该行配置至于任何位置。

开始测试

注：--bootstrap-server因启用了证书认证，因此必须使用host进行访问，使用IP地址将报错。

• 进入kafka家目录

cd /opt/tools/kafka_2.13-3.2.1/

• 查看topic 列表

./bin/kafka-topics.sh --list --bootstrap-server doris1:19092,doris2:19092,doris3:19092 --command-config ./config/jaas.properties

 创建topic

./bin/kafka-topics.sh --create --topic ssl-test --bootstrap-server doris1:19092,doris2:19092,doris3:19092  --partitions 3 --replication-factor 3 --config retention.ms=259200000 --command-config ./config/jaas.properties

• 生产

./bin/kafka-console-producer.sh --bootstrap-server doris1:19092,doris2:19092,doris3:19092 --topic test-ssl --producer.config ./config/jaas.properties

• 消费（在集群中其他节点上创建消费）

./bin/kafka-console-consumer.sh --bootstrap-server doris1:19092,doris2:19092,doris3:19092 --topic test-ssl --from-beginning --consumer.config ./config/jaas.properties

• 此时在生产端输入内容，到消费端查看内容，消费端可查看到生产端输入的内容，则表示kafka scram/ssl认证成功实现。