kafka 加密通信，python-kafka 訪問加密服務器

1、證書准備

#!/bin/bash

PASSWORD=pwd123
ADDR=kafka-single
ITEM_NAME=localhost 
 6 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 -storepass $PASSWORD -dname "CN=SX, OU=SX, O=sx, L=SZ, ST=GD, C=CN"
19 
#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/O=SX/OU=sx/CN=KAFKA-SERVER/emailAddress=[e-mail]"

#Step 3 通過CA證書創建一個受服務端、客戶端信任的證書
TRUSTSTORE_S=/usr/ca/trust/server.truststore.jks
TRUSTSTORE_C=/usr/ca/trust/client.truststore.jks
echo y | keytool -keystore $TRUSTSTORE_S -alias $CAROOT_NAME -import -file $CA_CERT -storepass $PASSWORD
echo y | 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根證書導入服務端倉庫
echo y | 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.導出 pem 格式的證書供 kafka-python 使用

#查看服務端秘鑰庫中的秘鑰證書
keytool -list -rfc -keystore $KEYSTORE_S -storepass $PASSWORD

#導出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

 

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='test'

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="custom"
pwd="pwd123"

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_custom="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.kafka 的SCRAM機制依賴zookeeper存儲憑證，在zk中創建：如配置使用 SASL_SSL/SCRAM 時
bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[password=pwd123],SCRAM-SHA-512=[password=pwd123]' --entity-type users --entity-name custom

# jaas.conf 如下：KafkaServer {

        org.apache.kafka.common.security.scram.ScramLoginModule required
        username="cumstom"
        password="pwd123";
    };

執行報錯

rror while executing config command org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /config/changes/config_change_

org.I0Itec.zkclient.exception.ZkNoNodeException: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /config/changes/config_change_

原因ZK缺少節點，登錄zk 后創建節點

bash bin/zkCli.sh

[zk: localhost:2181(CONNECTED) 0] create /config changes
Node already exists: /config
[zk: localhost:2181(CONNECTED) 1] create /config/changes config_change_

 

然后再執行