Zookeeper和Kafka安全配置
买房靠丈母娘
编辑于 2021年06月17日 22:09
收录于文集
共1篇

一共涉及这么几个安全项：

  1. zookeeper的节点内容安全。对zookeeper的节点进行ACL限制。

  2. zookeeper的4字指令安全。禁止4字指令查询zookeeper状态。

  3. kafka的消息安全。对kafka的消息内容添加ACL。

官方文档：http://kafka.apache.org/documentation/#security

使用的Kafka版本2.13-2.8.0，解压后主目录：/data/kafka_2.13-2.8.0/

下面的文件及命令都是相对主目录位置，配置文件中列出的内容只涉及安全相关配置，文件中已有的要做修改，没有的增加。

1、可以直接用kafka自带的zookeeper，如果用单独的zookeeper，需要导入相应的kafka包：

kafka-clients-2.8.0.jar

lz4-java-1.7.1.jar

2、config/zookeeper.properties文件配置

代码块
HTML
自动换行
复制代码
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000
zookeeper.sasl.client=true
4lw.commands.whitelist=
复制成功

其中"4lw.commands.whitelist=&#​34;是禁止zookeeper的4字指令。

3、增加config/jaas.conf文件，不能有注释

代码块
HTML
自动换行
复制代码
Server {
   org.apache.kafka.common.security.plain.PlainLoginModule required
   username="admin"
   password="123456"
   user_kafka="123456"
   user_admin="123456";
};
Client {
   org.apache.kafka.common.security.plain.PlainLoginModule required
   username="kafka"
   password="123456";
};
KafkaServer {
   org.apache.kafka.common.security.plain.PlainLoginModule required
   username="kafka"
   password="123456"
   user_admin="123456"
   user_kafka="123456";
};
KafkaClient {
   org.apache.kafka.common.security.plain.PlainLoginModule required
   username="kafka"
   password="123456";
};
复制成功

其中Server是zookeeper服务端配置，配置中添加了2个账号admin和kafka，username和password指明zookeeper间通信使用的账号密码。

Client是kafka与zookeeper通信时的账号及密码。可以用-Dzookeeper.sasl.clientconfig=ZkClient指定不同的配置。

KafkaServer和KafkaClient分别是kafka的服务器端及客户端配置。

4、给zookeeper生成一个超级用户super的密码

代码块
Shell
自动换行
复制代码
$ cd /data/kafka_2.13-2.8.0/libs/
$ java -cp "./zookeeper-3.7.0.jar:./slf4j-api-1.7.30.jar" org.apache.zookeeper.server.auth.DigestAuthenticationProvider super:123456
复制成功

只取最后一行的输出：

super:123456->super:BBO7K8dPkoek/JxIHqXxM75QRpI=

5、增加启动配置

如果用的是单独的zookeeper：

1.在zkStart.sh中

代码块
Shell
自动换行
复制代码
export JVMFLAGS="$JVMFLAGS -Dzookeeper.DigestAuthenticationProvider.superDigest=super:BBO7K8dPkoek/JxIHqXxM75QRpI="
复制成功

2.在kafka-run-class.sh中

代码块
Shell
自动换行
复制代码
export KAFKA_OPTS="$KAFKA_OPTS -Dzookeeper.sasl.client=true -Dzookeeper.sasl.client.username=kafka -Djava.security.auth.login.config=/data/kafka_2.13-2.8.0/config/jaas.conf"
复制成功

如果用的是kafka自带的zookeeper，只改kafka-run-class.sh即可：

代码块
Shell
自动换行
复制代码
export KAFKA_OPTS="$KAFKA_OPTS -Dzookeeper.sasl.client=true -Dzookeeper.DigestAuthenticationProvider.superDigest=super:BBO7K8dPkoek/JxIHqXxM75QRpI= -Dzookeeper.sasl.client.username=kafka -Djava.security.auth.login.config=/data/kafka_2.13-2.8.0/config/jaas.conf"
复制成功

6、启动zookeeper

代码块
Shell
自动换行
复制代码
$ cd /data/kafka_2.13-2.8.0/bin/
$ ./zookeeper-server-start.sh -daemon ../config/zookeeper.properties
复制成功

7、修改config/server.properties

代码块
HTML
自动换行
复制代码
broker.id=1
listeners=SASL_PLAINTEXT://192.168.181.132:9092
advertised.listeners=SASL_PLAINTEXT://192.168.181.132:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=PLAIN  
sasl.mechanism.inter.broker.protocol=PLAIN
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
allow.everyone.if.no.acl.found=false
super.users=User:admin
zookeeper.set.acl=true
复制成功

8、启动kafka

代码块
Shell
自动换行
复制代码
$ cd /data/kafka_2.13-2.8.0/bin/
$ ./kafka-server-start.sh -daemon ../config/server.properties
复制成功

9、zookeeper添加ACL

一定要用kafka自带的zookeeper客户端，这个客户端将通过kafka用户的sasl方式登录zookeeper，启动类是org.apache.zookeeper.ZooKeeperMainWithTlsSupportForKafka。单独zookeeper客户端启动类是org.apache.zookeeper.ZooKeeperMain，不能以sasl方式登录。

代码块
Shell
自动换行
复制代码
$ cd /data/kafka_2.13-2.8.0/bin/
$ ./zookeeper-shell.sh localhost:2181
复制成功

添加认证用户：

代码块
Shell
自动换行
复制代码
addauth digest admin:123456
addauth digest kafka:123456
复制成功

配置ACL：

代码块
Shell
自动换行
复制代码
setAcl / auth:admin:cdrwa
setAcl / auth:kafka:cdrwa
复制成功

查看ACL：

代码块
Shell
自动换行
复制代码
getAcl /
'sasl,'kafka
: cdrwa
'digest,'admin:0uek/hZ/V9fgiM35b0Z2226acMQ=
: cdrwa
'digest,'kafka:AHviFxZyZ/iL8MsBeboZ4XZzv/M=
: cdrwa
复制成功

不能用这个命令配置ACL：

代码块
Shell
自动换行
复制代码
setAcl / sasl:kafka:cdrwa
复制成功

将导致/的ACL只有：

代码块
Shell
自动换行
复制代码
getAcl /
'sasl,'kafka
: cdrwa
复制成功

如果用单独的zookeeper客户端配置ACL，只能配置成digest或者sasl的方式，不能使2个kafka认证同时生效。

10、测试：

用单独的zookeeper客户端登录：

代码块
Shell
自动换行
复制代码
./zkCli.sh
[zk: localhost:2181(CONNECTED) 0] ls /
Insufficient permission : /
复制成功

授权登录：

代码块
Shell
自动换行
复制代码
addauth digest admin:123456
ls /
[admin, brokers, cluster, config, consumers, controller_epoch, delegation_token, feature, isr_change_notification, kafka-acl, kafka-acl-changes, kafka-acl-extended, kafka-acl-extended-changes, latest_producer_id_block, log_dir_event_notification, zookeeper]
复制成功

11、给kafka服务器端的kafka用户授权

给用户kafka所有topic的producer权限

代码块
Shell
自动换行
复制代码
$ ./kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --topic * --add --allow-principal User:kafka --producer --group=*
复制成功

给用户kafka所有topic的consumer权限

代码块
Shell
自动换行
复制代码
$ ./kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --topic * --add --allow-principal User:kafka --consumer --group=*
复制成功

给用户kafka所有topic的操作权限

代码块
Shell
自动换行
复制代码
$ ./kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 --topic * --add --allow-principal User:kafka --operation all --group "*"
复制成功

12、测试kafka

代码块
Shell
自动换行
复制代码
$ cd /data/kafka_2.13-2.8.0/bin/
复制成功

1) 显示所有topic

代码块
Shell
自动换行
复制代码
$ ./kafka-topics.sh --list --zookeeper localhost:2181
复制成功

2) 创建topic

代码块
Shell
自动换行
复制代码
$ ./kafka-topics.sh --create --zookeeper localhost:2181 --topic test --partitions 1 --replication-factor 1
复制成功

3) 发送消息

修改producer.properties

security.protocol=SASL_PLAINTEXT

sasl.mechanism=PLAIN

代码块
Shell
自动换行
复制代码
$ ./kafka-console-producer.sh --topic test --bootstrap-server 192.168.181.132:9092 --producer.config ../config/producer.properties
复制成功

4) 读取消息

修改consumer.properties

security.protocol=SASL_PLAINTEXT

sasl.mechanism=PLAIN

代码块
Shell
自动换行
复制代码
$ ./kafka-console-consumer.sh --bootstrap-server 192.168.181.132:9092 --topic test --from-beginning --consumer.config ../config/consumer.properties
复制成功

5、删除topic

代码块
Shell
自动换行
复制代码
$ ./kafka-topics.sh --zookeeper localhost:2181 --delete --topic test
复制成功

13、spring配置

代码块
HTML
自动换行
复制代码
spring:
  kafka:    
    properties:
      sasl.mechanism: PLAIN
      security.protocol: SASL_PLAINTEXT
      zookeeper.sasl.client: true
      sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="123456";
复制成功

14、遗留问题

子目录不会继承根目录的ACL限制，其它一些目录，例如/kafka-acl目录仍然是任意可读的，只是不能修改。

kafka zookeeper acl
本文为我原创，未经授权禁止转载
cv11773508