Kafka SASL/PLAIN静态认证集群部署_kafka plain

Kafka SASL/PLAIN静态认证集群部署

在Kafka中，SASL机制包含三种，它们分别是Kerberos、PLAIN、SCRAM。以PLAIN认证为示例。

1. 服务端配置

1、解压安装包

tar -zxvf kafka_2.11-2.4.1.tgz -C /home/xyp9x/
1

2、改名

mv kafka_2.11-2.4.1 kafka_sasl
1

3、在kafka_sasl目录下创建logs、kafka-logs文件夹

mkdir logs kafka-logs
1

4、在config目录中创建kafka_server_jaas.conf文件，前三行是配置管理员账户（该账户与上面server.properties中配置的super.users一样）后面的即 user_用户名=“该用户的密码”。之后配置ACL的时候需要用到这里配置的用户

vi kafka_server_jaas.conf

KafkaServer {
	org.apache.kafka.common.security.plain.PlainLoginModule required
	username="admin"
	password="admin"
	user_admin="admin"
	user_xyp9x="xyp9x";
};
1
2
3
4
5
6
7
8
9

5、修改配置文件

vi server.properties

将原内容全部清空，输入以下内容：

#broker的全局唯一编号，不能重复
broker.id=0
#是否允许删除topic
delete.topic.enable=true
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘IO的线程数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的最大缓冲区大小
socket.request.max.bytes=104857600
#kafka数据存放的路径
log.dirs=/home/xyp9x/kafka_sasl/kafka-logs
#topic在当前broker上的分区个数
num.partitions=1
#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment文件保留的最长时间，超时将被删除
log.retention.hours=168
#配置连接Zookeeper集群地址
zookeeper.connect=bigdata111:2181,bigdata112:2181,bigdata113:2181
#kafka连接zookeeper超时时间90s
zookeeper.connection.timeout.ms=90000

#认证配置
listeners=SASL_PLAINTEXT://192.168.1.111:9092
#PLAINTEXT不加密明文传输, 省事, 性能也相对好
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
#ACL配置
#设置为true，ACL机制为黑名单机制，只有黑名单中的用户无法访问
#设置为false，ACL机制为白名单机制，只有白名单中的用户可以访问
allow.everyone.if.no.acl.found=false
super.users=User:admin
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

6、在Kafka启动脚本中添加配置文件路径

vi kafka-server-start.sh

#!/bin/bash
export KAFKA_OPTS="-Djava.security.auth.login.config=/home/xyp9x/kafka_sasl/config/kafka_server_jaas.conf"
1
2
3
4

7、配置环境变量

vi /etc/profile

# Kafka_sasl
export KAFKA_HOME=/home/xyp9x/kafka_sasl
export PATH=$PATH:$KAFKA_HOME/bin
1
2
3
4
5

8、分发

rsync -r /etc/profile bigdata112:/etc/
rsync -r /etc/profile bigdata113:/etc/
rsync -r kafka_sasl bigdata112:/home/xyp9x/
rsync -r kafka_sasl bigdata113:/home/xyp9x/
1
2
3
4

9、分别在112和113上修改配置文件

vi server.properties
broker.id=1
listeners=SASL_PLAINTEXT://192.168.1.112:9092

vi server.properties
broker.id=2
listeners=SASL_PLAINTEXT://192.168.1.113:9092
1
2
3
4
5
6
7

10、更新111、112、113的环境变量

source /etc/profile
1

2.客户端配置（当服务端配置启用了SASL/PLAIN，那么Client连接的时候需要配置认证信息）

1、在config目录创建kafka_client_jaas.conf

vi kafka_client_jaas.conf

KafkaClient {
	org.apache.kafka.common.security.plain.PlainLoginModule required
	username="admin"
	password="admin";
};
1
2
3
4
5
6
7

2、在producer.properties文件中添加认证协议

vi producer.properties

security.protocol=SASL_PLAINTEXT 
sasl.mechanism=PLAIN
1
2
3
4

3、在consumer.properties文件中添加认证协议

vi consumer.properties

security.protocol=SASL_PLAINTEXT 
sasl.mechanism=PLAIN
1
2
3
4

4、在kafka-console-producer.sh脚本中添加JAAS文件的路径

vi kafka-console-producer.sh

#!/bin/bash
export KAFKA_OPTS="-Djava.security.auth.login.config=/home/xyp9x/kafka_sasl/config/kafka_client_jaas.conf"
1
2
3
4

5、在kafka-console-consumer.sh脚本中添加JAAS文件的路径

vi kafka-console-consumer.sh

#!/bin/bash
export KAFKA_OPTS="-Djava.security.auth.login.config=/home/xyp9x/kafka_sasl/config/kafka_client_jaas.conf"
1
2
3
4

6、分发

rsync -r config/kafka_client_jaas.conf bigdata112:/home/xyp9x/kafka_sasl/config/
rsync -r config/kafka_client_jaas.conf bigdata113:/home/xyp9x/kafka_sasl/config/
rsync -r config/producer.properties bigdata112:/home/xyp9x/kafka_sasl/config/
rsync -r config/producer.properties bigdata113:/home/xyp9x/kafka_sasl/config/
rsync -r config/consumer.properties bigdata112:/home/xyp9x/kafka_sasl/config/
rsync -r config/consumer.properties bigdata113:/home/xyp9x/kafka_sasl/config/
rsync -r bin/kafka-console-producer.sh bigdata112:/home/xyp9x/kafka_sasl/bin/
rsync -r bin/kafka-console-producer.sh bigdata113:/home/xyp9x/kafka_sasl/bin/
rsync -r bin/kafka-console-consumer.sh bigdata112:/home/xyp9x/kafka_sasl/bin/
rsync -r bin/kafka-console-consumer.sh bigdata113:/home/xyp9x/kafka_sasl/bin/
1
2
3
4
5
6
7
8
9
10

7、启动kafka-broker

bin/kafka-server-start.sh -daemon config/server.properties
1

8、创建topic

bin/kafka-topics.sh --zookeeper 192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --create --replication-factor 3 --partitions 3 --topic test
1

9、超级用户启动生产者

bin/kafka-console-producer.sh --broker-list 192.168.1.111:9092,192.168.1.112:9092,192.168.1.113:9092 --topic test --producer.config config/producer.properties
1

10、超级用户启动消费者

bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.111:9092,192.168.1.112:9092,192.168.1.113:9092 --topic test --consumer.config config/consumer.properties
1

3. ACL操作（在配置好SASL后，启动Zookeeper集群和Kafka集群之后，就可以使用kafka-acls.sh脚本来操作ACL机制）

1、查看ACL授权

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --list 
1

2、对普通用户生产者授权

//允许xyp9x用户从所有IP地址写
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --add --allow-principal User:xyp9x --operation Write --topic test

//禁止xyp9x用户从所有IP地址写
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --remove --allow-principal User:xyp9x --operation Write --topic test

//允许xyp9x用户从192.168.1.111和192.168.1.112写
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --add --allow-principal User:xyp9x --allow-host 192.168.1.111 --allow-host 192.168.1.112 --operation Write --topic test

//禁止xyp9x用户从192.168.1.111和192.168.1.112写
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --remove --allow-principal User:xyp9x --allow-host 192.168.1.111 --allow-host 192.168.1.112 --operation Write --topic test
1
2
3
4
5
6
7
8
9
10
11

3、对普通用户消费者组授权

//允许xyp9x用户使用test-consumer-group消费组消费消息
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --add --allow-principal User:xyp9x --operation Read --group test-consumer-group

//禁止xyp9x用户使用test-consumer-group消费组消费消息
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --remove --allow-principal User:xyp9x --operation Read --group test-consumer-group
1
2
3
4
5

4、对普通用户消费者授权

//允许xyp9x用户从所有IP地址读
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --add --allow-principal User:xyp9x --operation Read --topic kafka_acl_topic

//禁止xyp9x用户从所有IP地址读
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --remove --allow-principal User:xyp9x --operation Read --topic kafka_acl_topic

//允许xyp9x用户从192.168.1.111和192.168.1.112读
bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --add --allow-principal User:xyp9x --allow-host 192.168.1.111 --allow-host 192.168.1.112 --operation Read --topic kafka_acl_topic

//禁止xyp9x用户从192.168.1.111和192.168.1.112读
bin/kafka-acls.sh --authorizer-properties zookeeper.connect=192.168.1.111:2181,192.168.1.112:2181,192.168.1.113:2181 --remove --allow-principal User:xyp9x --allow-host 192.168.1.111 --allow-host 192.168.1.112 --operation Read --topic kafka_acl_topic
1
2
3
4
5
6
7
8
9
10
11