一、概述
上回已经完成kafka+zookeeper的基础功能的实现,但是因为默认不认证存在很大的安全风险,这次完成SASL_PLAINTEXT的认证类型实践。
二、安全配置
2.1 zookeeper SASL配置部分
2.1.1 创建conf/java.env文件,添加以下配置信息
export SERVER_JVMFLAGS="-Djava.security.auth.login.config=/opt/soft/zookeeper/conf/sasl.conf -Dzookeeper.allowSaslFailedClients=false"
export CLIENT_JVMFLAGS="-Djava.security.auth.login.config=/opt/soft/zookeeper/conf/sasl.conf -Dzookeeper.allowSaslFailedClients=false"
2.1.2 创建conf/sasl.conf文件,添加如下配置信息,该配置会用于kafka登陆认证
Server {
org.apache.zookeeper.server.auth.DigestLoginModule required
user_admin="password";
};
2.1.3 在conf/zoo.conf文件添加如下配置信息
authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
2.2 kafka配置部分
2.2.1 创建config/kafka_server_jaas.conf配置文件,内容如下
KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="password"
user_admin="admin123"
user_test="test123";
};
### Client为登陆zookeeper配置
Client {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="password";
};
### kafka客户端登陆配置
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin123";
};
2.2.2 在config/server.properties文件增加一下关于SASL的认证配置信息
####################################SASL SETTING########################################
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
#authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer ##3.0版本已经弃用
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=true
2.2.3 修改bin/kafka-run-class.sh文件,增加认证配置
KAFKA_OPTS="-Djava.security.auth.login.config=/opt/soft/kafka/config/kafka_server_jaas.conf"
2.2.4 在config/producer.properties 和 config/consumer.properties均需要添加以下认证配置
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin123";
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
2.2.5 在bin/kafka-console-producer.sh 和bin/kafka-console-consumer.sh均需要添加以下配置
export KAFKA_HEAP_OPTS="-Xmx512M -Djava.security.auth.login.config=kafka_server_jaas.conf"
三、测试验证
3.1 控制台生产者测试,需要配置producer.properties
bin/kafka-console-producer.sh --bootstrap-server 10.126.38.160:9092 --topic test2023 \
--producer.config config/producer.properties
3.2 控制台消费者测试,需要配置consumer.properties
./bin/kafka-console-consumer.sh --bootstrap-server 10.126.38.160:9092 --topic test2023 \
--consumer.config config/consumer.properties
3.3 使用python作为消费者连接测试
from kafka import KafkaConsumer
import time
import json
BOOTSTRAP_SERVERS = '10.126.38.160:9092'
TOPIC = 'test2023'
consumer = KafkaConsumer(TOPIC,
bootstrap_servers=BOOTSTRAP_SERVERS,
auto_offset_reset='earliest',
security_protocol='SASL_PLAINTEXT',
sasl_mechanism='PLAIN',
sasl_plain_username='admin',
sasl_plain_password='password',
api_version=(0, 10),
receive_buffer_bytes=1024,
enable_auto_commit='False')
for msg in consumer:
print(msg)
四、总结
通过以上配置基本能够实现SASL的配置功能。
标签:zookeeper,kafka,Centos7,conf,sasl,SASL,security,password,config From: https://blog.51cto.com/u_15131458/6085644