kafka 加密通信，python-kafka 访问加密服务器

1、证书准备

#!/bin/bash

PASSWORD=pwd123
ADDR=kafka-single
ITEM_NAME=my-kafka
CAROOT_NAME=CARoot

#Step 0: 创建保存目录
mkdir -p /usr/ca/{root,server,client,trust,py-kfk} && cd /usr/ca/
#修改hosts文件
if [ $(awk "BEGIN{flag=0}{if(match($0,/$ADDR/)){flag=1}}END{print flag}" /etc/hosts) -eq 0 ];then
    echo 127.0.0.1 $ADDR >> /etc/hosts
fi

#Step 1 生成服务端秘钥库
KEYSTORE_S=/usr/ca/server/server.keystore.jks
keytool -keystore $KEYSTORE_S -alias $ITEM_NAME -validity 365 -keyalg RSA -genkey -keypass $PASSWORD -dname "CN=$ADDR,OU=sf,O=sf,L=shenzheng,S=GuangDong,C=cn" -storepass $PASSWORD SAN=DNS:$ADDR

#Step 2 生成CA根证书、秘钥
CA_KEY=/usr/ca/root/ca-key
CA_CERT=/usr/ca/root/ca-cert
openssl req -new -x509 -keyout $CA_KEY -out $CA_CERT -days 365 -passout pass:$PASSWORD  -subj "/C=cn/ST=GuangDong/L=shenzheng/O=sf/OU=sf/CN=$ADDR"

#Step 3 通过CA证书创建一个受服务端、客户端信任的证书
TRUSTSTORE_S=/usr/ca/trust/server.truststore.jks
TRUSTSTORE_C=/usr/ca/trust/client.truststore.jks
keytool -keystore $TRUSTSTORE_S -alias $CAROOT_NAME -import -file $CA_CERT -storepass $PASSWORD
keytool -keystore $TRUSTSTORE_C -alias $CAROOT_NAME -import -file $CA_CERT -storepass $PASSWORD

#Step 3 导出服务端证书 server-cert-file
SERVER_CERT_FILE=/usr/ca/server/server-cert-file
keytool -keystore $KEYSTORE_S -alias $ITEM_NAME -certreq -file $SERVER_CERT_FILE -storepass $PASSWORD

#Step 4 使用CA根证书秘钥给服务端证书签名
SERVER_CERT_SIGNED=/usr/ca/server/server-cert-signed
openssl x509 -req -CA $CA_CERT -CAkey $CA_KEY -in $SERVER_CERT_FILE -out $SERVER_CERT_SIGNED -days 365 -CAcreateserial -passin pass:$PASSWORD

#Step 5 将CA根证书导入服务端仓库
keytool -keystore $KEYSTORE_S -alias $CAROOT_NAME -import -file $CA_CERT -storepass $PASSWORD

#Step 6 将签名后的服务端证书导入服务端秘钥库
keytool -keystore $KEYSTORE_S -alias $ITEM_NAME -import -file $SERVER_CERT_SIGNED -storepass $PASSWORD

2. 导出 python-kafka 所需证书

#导出 pem 格式的证书供 kafka-python 使用
#查看服务端秘钥库中的秘钥证书
#keytool -list -rfc -keystore $KEYSTORE_S

#导出CA签名后的服务端证书
SERVER_CERT_SIGNED_PEM_FORMAT=/usr/ca/py-kfk/server_signed.pem
CAROOT_PEM_FORMAT=/usr/ca/py-kfk/CARoot.pem
keytool -exportcert -alias $ITEM_NAME -keystore $KEYSTORE_S -storepass $PASSWORD -rfc -file $SERVER_CERT_SIGNED_PEM_FORMAT 
keytool -exportcert -alias $CAROOT_NAME -keystore $KEYSTORE_S -storepass $PASSWORD -rfc -file $CAROOT_PEM_FORMAT

3. 修改kafka服务器配置

修改 server.properties 配置
listeners=PLAINTEXT://localhost:9092,SASL_SSL://kafka-single:9093
ssl.keystore.location=/usr/ca/server/server.keystore.jks
ssl.keystore.password=pwd123
ssl.key.password=pwd123
ssl.truststore.location=/usr/ca/trust/server.truststore.jks
ssl.truststore.password=pwd123
ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1
ssl.keystore.type=JKS
ssl.truststore.type=JKS
ssl.secure.random.implementation=SHA1PRNG
security.inter.broker.protocol=SSL  #只使用SSL

#使用SASL/PLAIN + TLS 时
#security.inter.broker.protocol=SASL_SSL  
#sasl.mechanism.inter.broker.protocol=PLAIN
#sasl.enabled.mechanisms=PLAIN

4. 启动 zookeeper
~/kafka # bin/zookeeper-server-start.sh config/zookeeper.properties &

5. 启动 broker
~/kafka # bin/kafka-server-start.sh config/server.properties

6. python-kafka 连接服务器，若使用 PLAIN 文本

sasl_mechanism="PLAIN" 即可，SCRAM-SHA-512 需服务器做相应配置

# -*- coding: utf-8 -*-
from kafka import KafkaConsumer, KafkaProducer
import kafka
import ssl
import logging
import time
import json


bootstrap_servers = 'kafka-single:9093'
topic='sqtest'

ssl_certfile = "/usr/ca/py-kfk/certificate.pem"
ssl_cafile = "/usr/ca/py-kfk/CARoot.pem"

context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
context.verify_mode = ssl.CERT_NONE
context.check_hostname = False
context.load_verify_locations(ssl_cafile)

sasl_mechanism="SCRAM-SHA-512"

username="user"
pwd="pwd"

producer = KafkaProducer(
    bootstrap_servers=bootstrap_servers,
    acks='all',
    retries=1,
    api_version=(0, 11, 0, 3),
    ssl_check_hostname=False,
    security_protocol="SASL_SSL",
    ssl_context=context,
    sasl_mechanism=sasl_mechanism,
    sasl_plain_username=username,
    sasl_plain_password=pwd
)
print producer.bootstrap_connected()
print "send message"
for i in range(1):
    producer.send(topic,'localhost test')
    producer.flush()
    producer.close()

#########################################
7.SASL/PLAIN 用户名秘密配置

配置 Kafka Broker
在 Kafka broker 的 config 目录中, 添加一个类似于下面的适当修改过的 JAAS 文件. 在这个例子中, 让我们将它命名为

kafka_server_jaas.conf:

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-secret"
    user_admin="admin-secret"
    user_encrypt="pwd123";
};

#将其路径作为JVM参数，下面行也可直接加入启动脚本中去 bin/kafka-server-start.sh
export KAFKA_OPTS="-Djava.security.auth.login.config=~/kafka/config/kafka_server_jaas.conf"
前台启动broker （启动前记得修改 server.properties 使其支持 SASL/PLAIN + TLS）
~/kafka # bin/kafka-server-start.sh config/server.properties

8.测试数据上报服务器
bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=alice-secret],SCRAM-SHA-512=[password=alice-secret]' --entity-type users --entity-name alice