当前位置：首页 > news >正文

Python客户端使用SASL_SSL连接Kafka需要将jks密钥转换为pem密钥，需要转化成p12格式再转换pem才能适配confluent_kafka包

news 2025/6/27 21:20:37

证书生成

生成证书以及jks参考以下文章
https://blog.csdn.net/qq_41527073/article/details/121148600

证书转换jks -> pem

需要转化成p12以下转换才能适配confluent_kafka包，直接jks转pem会报错不能使用，具体参考以下文章
https://www.ngui.cc/zz/1104321.html?action=onClick
在这里插入图片描述
keytool -importkeystore -srckeystore server.truststore.jks -destkeystore server.p12 -deststoretype PKCS12
openssl pkcs12 -in server.p12 -nokeys -out server.cer.pem
keytool -importkeystore -srckeystore server.keystore.jks -destkeystore client.p12 -deststoretype PKCS12
openssl pkcs12 -in client.p12 -nokeys -out client.cer.pem
openssl pkcs12 -in client.p12 -nodes -nocerts -out client.key.pem
(转换过程)
生成证书目录结构：
在这里插入图片描述

python客户端示例代码

import rsa, json
import time, sys
from kafka import KafkaProducer
import confluent_kafka
import ssl
def encrypt(msg):with open('public.pem', 'rb') as publickfile:p = publickfile.read()pubkey = rsa.PublicKey.load_pkcs1(p)original_text = msg.encode('utf8')crypt_text = rsa.encrypt(original_text, pubkey)return crypt_textdef decrypt(data):with open('private.pem', 'rb') as privatefile:p = privatefile.read()privkey = rsa.PrivateKey.load_pkcs1(p)crypt_text = dataoriginal_text = rsa.decrypt(crypt_text, privkey)return original_text.decode('utf8')def produce_message():producer: KafkaProducer = Nonesuccess = 0conn_error = 0msg = {"type": "webclone","api": "delete","state": True,"nodename": "node-1","uuid": "asjdkjrh"}context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)context.set_ciphers('TLSv1:TLSv1.2')context.check_hostname = Falsecontext.verify_mode = ssl.CERT_NONEwhile success < 5:try:producer.send('message_push', value=json.dumps(msg))producer.flush()success += 1except KeyboardInterrupt:breakexcept:while True:try:producer = KafkaProducer(bootstrap_servers=['172.XX.X.XX:9093'],acks = 1,security_protocol="SASL_SSL",ssl_cafile="/Kafka/config/ssl/server.cer.pem",ssl_certfile="/Kafka/config/ssl/client.cer.pem",ssl_keyfile="/Kafka/config/ssl/client.key.pem",ssl_context=context,sasl_mechanism="PLAIN",sasl_plain_username="kafka",sasl_plain_password="XXXXX",api_version = (2, 0),)breakexcept Exception as e:producer = Noneprint(e)conn_error += 1time.sleep(1)print(f"connect error: {conn_error}")if __name__ == '__main__':if len(sys.argv) < 2:print("Usage: python3 test.py produce|consume")sys.exit(1)start_time = time.time()if sys.argv[1] == 'produce':produce_message()elif sys.argv[1] == 'consume':consume_message()end_time = time.time()print(f"start at: {start_time}, end at: {end_time}, cost: {end_time - start_time} seconds")