Window下kafka 单机SASL_SCRAM加密及身份认证

     KAFKA加密认证机制中的SASL主要包括SASL_PLAINTEXT SASL_GSSAPI SASL_SCRAM。这里主要记录一下Windows下搭建配置单机sasl_scram环境。

一、前情提要

    SCRAM是kafka安全机制SASL家族中的一个，通过执行用户名/密码认证（如PLAIN和DIGEST-MD5）的传统机制来解决安全问题，Kafka中的默认SCRAM实现是在Zookeeper中存储SCRAM的证书。包括SCRAM-SHA-256和SCRAM-SHA-512，SHA_256和SHA_512是哈希算法。下面的提到的zookeeper都用的是kafka自带的。

二、配置准备

1：JAVA_HOME 

有JAVA_HOME的环境变量，且java版为1.8及以上，jdk目录无中文和空格；

2：KAFKA项目部署

下载解压KAFKA项目到一个无中文无空格的目录下。配置非加密下的KAFKA环境，参考Windows环境下kafka搭建；

三、JAAS配置

3.1  zookeeper配置

3.1.1 config目录下zookeeper.properties修改为：

dataDir=./data/zookeeper clientPort=2181 maxClientCnxns=0 authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider requireClientAuthScheme=sasl jaasLoginRenew= zookeeper.sasl.client=true 

3.1.2 config目录新增kafka_zoo_jaas.conf，内容为：

Server { org.apache.zookeeper.server.auth.DigestLoginModule required username="admin" password="admin" user_admin="admin"; };

3.1.3 bin/windows/zookeeper-server-start.bat 设置KAFKA_OPTS，修改后内容为：

@echo off IF [%1] EQU [] ( echo USAGE: %0 zookeeper.properties EXIT /B 1 ) SetLocal IF ["%KAFKA_LOG4J_OPTS%"] EQU [""] ( set KAFKA_LOG4J_OPTS=-Dlog4j.configuration=file:%~dp0../../config/log4j.properties ) IF ["%KAFKA_HEAP_OPTS%"] EQU [""] ( set KAFKA_HEAP_OPTS=-Xmx512M -Xms512M ) set KAFKA_OPTS=-Djava.security.auth.login.config=E:/demo/kafka/SASL_SCRAM/broker/2181/kafka_2.12-1.1.1/config/kafka_zoo_jaas.conf "%~dp0kafka-run-class.bat" org.apache.zookeeper.server.quorum.QuorumPeerMain %* EndLocal

3.1.4 启动zookeeper

执行下面的命令，如果出现下面的图片信息则为启动成功。

.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

3.2 设置SCRAM凭证

3.2.1 设置凭证

    在这一步的时候，要先保证zookeeper是启动状态。在kafka_zoo_jaas.conf中，设置了用户admin，密码admin。下面为admin创建scram凭证，执行命令，结果如下图：

.\bin\windows\kafka-configs.bat --zookeeper 192.168.40.150:2181 --alter --add-config "SCRAM-SHA-256=[password=admin],SCRAM-SHA-512=[password=admin]" --entity-type users --entity-name admin

3.2.2 查看凭证

.\bin\windows\kafka-configs.bat --zookeeper 192.168.40.150:2181 --describe --entity-type users --entity-name admin

3.3 配置kafka server

3.3.1 config目录下创建文件kafka_server_jaas.conf，具体的文件内容为：

Client { org.apache.zookeeper.server.auth.DigestLoginModule required username="admin" password="admin"; }; KafkaServer { org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin" user_admin="admin"; };

其中client用来broker认证zookeeper，kafkaServer用于broker和broker的通讯，及client的认证

3.3.2 server.properties文件修改，修改后的文件内容为：

 Server Basics broker.id=2 port=9091 host.name=192.168.40.150 advertised.port=9091 advertised.host.name=192.168.40.150 Socket Server Settings listeners=SASL_PLAINTEXT://192.168.40.150:9091 advertised.listeners=SASL_PLAINTEXT://192.168.40.150:9091 #SASL机制 security.inter.broker.protocol=SASL_PLAINTEXT sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256 sasl.enabled.mechanisms=SCRAM-SHA-256,SCRAM-SHA-512 num.network.threads=3 num.io.threads=8 socket.send.buffer.bytes= socket.receive.buffer.bytes= socket.request.max.bytes= Log Basics log.dirs=./tmp/kafka-logs num.partitions=1 num.recovery.threads.per.data.dir=1 Internal Topic Settings offsets.topic.replication.factor=1 transaction.state.log.replication.factor=1 transaction.state.log.min.isr=1 Log Flush Policy #log.flush.interval.messages=10000 #log.flush.interval.ms=1000 Log Retention Policy log.retention.hours=168 #log.retention.bytes= log.segment.bytes= log.retention.check.interval.ms= Zookeeper zookeeper.connect=192.168.40.150:2181 zookeeper.set.acl=true zookeeper.connection.timeout.ms=60000 Group Coordinator Settings group.initial.rebalance.delay.ms=0

这里选用了SCRAM(PLAIN、DIGEST-MD5) 中的PLAIN，授权协议选择了SCRAM-SHA-256

3.3.3 修改bin/windows/kafka-server-start.bat

将上面创建kafka_server_jaas.conf设置为JVM参数，修改后内容为：

@echo off IF [%1] EQU [] ( echo USAGE: %0 server.properties EXIT /B 1 ) SetLocal set KAFKA_OPTS=-Djava.security.auth.login.config=E:/demo/kafka/SASL_SCRAM/broker/2181/kafka_2.12-1.1.1/config/kafka_server_jaas.conf IF ["%KAFKA_LOG4J_OPTS%"] EQU [""] ( set KAFKA_LOG4J_OPTS=-Dlog4j.configuration=file:%~dp0../../config/log4j.properties ) IF ["%KAFKA_HEAP_OPTS%"] EQU [""] ( rem detect OS architecture wmic os get osarchitecture | find /i "32-bit" >nul 2>&1 IF NOT ERRORLEVEL 1 ( rem 32-bit OS set KAFKA_HEAP_OPTS=-Xmx512M -Xms512M ) ELSE ( rem 64-bit OS set KAFKA_HEAP_OPTS=-Xmx1G -Xms1G ) ) "%~dp0kafka-run-class.bat" kafka.Kafka %* EndLocal

3.3.4 启动kafka server

用下面的命令启动，且出现下面的截图则为启动成功

.\bin\windows\kafka-server-start.bat .\config\server.properties

 3.4 创建TOPIC

3.4.1修改bin/windows/kafka-topics.bat，增加jvm参数，修改后的内容如下：

@echo off SetLocal set KAFKA_OPTS=-Djava.security.auth.login.config=E:/demo/kafka/SASL_SCRAM/broker/2181/kafka_2.12-1.1.1/config/kafka_server_jaas.conf "%~dp0kafka-run-class.bat" kafka.admin.TopicCommand %* EndLocal

3.4.2 创建topic

使用下面的命令创建topic NOTICE：

.\bin\windows\kafka-topics.bat --create --zookeeper 192.168.40.150:2181 --replication-factor 1 --partitions 1 --topic NOTICE

3.5 创建生产者

3.5.1 修改config下producer.properties

和上面server.properties和kafka_server_jaas.conf修改的信息保持一致，修改后的内容为：

 Producer Basics bootstrap.servers=192.168.40.150:9091 compression.type=none sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin"; security.protocol=SASL_PLAINTEXT sasl.mechanism=SCRAM-SHA-256 

3.5.2 打开生产者并生产消息

.\bin\windows\kafka-console-producer.bat --broker-list 192.168.40.150:9091 --topic NOTICE --producer.config .\config\producer.properties

 3.6 创建消费者

3.6.1 修改config目录下consumer.properties

具体修改和producer.properties一样，修改后的信息为：

bootstrap.servers=192.168.40.150:2181 group.id=test-consumer-group sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin"; security.protocol=SASL_PLAINTEXT sasl.mechanism=SCRAM-SHA-256

 3.6.2 创建消费者并消费数据

.\bin\windows\kafka-console-consumer.bat --bootstrap-server 192.168.40.150:9091 --topic NOTICE --consumer.config .\config\consumer.properties --from-beginning