前言

通常，我们会给Kafka增加SASL以加强对kafka的安全访问，以满足漏扫对于kafka的访问安全漏扫要求。但漏扫的时候也会同样会对Kafka集群中的zk集群进行扫描。所以zk集群同样涉及到要配置kafka的安全认证。

Zookeeper（Kafka内置）单独添加SASL认证及ACL

创建zk.jaas文件

以此路径/opt/soft/kafka/kafka_2.12-2.5.1/config/zk.jaas为例：

Server {
  org.apache.kafka.common.security.plain.PlainLoginModule required 
  user_super="adminsecret"
  user_zkclient="12345678";
};
Client {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="zkclient"
  password="12345678";
};

修改zookeeper.properties

增加以下配置

authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl

修改kafka-run-class.sh

添加以下内容（# Launch mode上一行添加）：

export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/soft/kafka/kafka_2.12-2.5.1/config/zk.jaas"

修改server.properties

增加以下配置

zookeeper.client.sasl=true

重启zk

登录zk客户端，设置ACL

进入bin目录，执行：

./zookeeper-shell.sh  <本机IP>:2181

输入：

setAcl / ip:<希望有权限访问的IP>:cdrwa

重启Kafka

验证zk的sasl是否配置成功

使用一台非集群内的zk节点上的zk软件，来模拟入侵。也就说这个节点的ip并没有被设置getAcl

执行zkCli.sh工具连接Zookeeper:

zkCli.sh -server <想要入侵的zk节点ip>:2181
ls /
Insufficient permission :   

出现“Insufficient permission :”表明没有权限访问。

也可以使用内置的zk命令尝试：

zookeeper-shell.sh <想要入侵的zk节点ip>:2181
ls /
Insufficient permission :   
```bash