1

我的团队正在使用 confluent_kafka 生成实时流，并使用自定义加密被推送到队列中。背后的技术是Java。

我正在尝试使用主题中的 python 来使用消息，并且能够这样做但无法使用 python 对其进行解密。

from confluent_kafka import *
import os
from confluent_kafka.serialization import StringDeserializer

os.environ['HTTPS_PROXY'] = "http://username:password@proxy.com:8080"


c = DeserializingConsumer({
    'bootstrap.servers': 'kafka.abcd.com:1010',
    'group.id':'SharedKafka2',
    'auto.offset.reset': 'latest',
    'security.protocol':'SSL',
    'ssl.ca.location':'CARoot.pem',
    'ssl.certificate.location':'certificate.pem',
    'ssl.key.location':'key.pem',
    'value.deserializer': StringDeserializer()
})

c.subscribe(['SharedKafka2_Topic'])


print("ConsumerStarted")
while True:
    msg = c.poll()
    if msg is None:
        continue
    if msg.error():
        if msg.error().code() == KafkaError._PARTITION_EOF:
            continue
        else:
            print(msg.error())
            break

    print(msg.value())

输出=b'\x85s\n\x0e{!G\xc7\xdfaV\x8e\x03a\xc4\x8e\xe1\xbc\xf4\xf4F\xb5<{\xecD\xc2\xf8jNN

我无法在值反序列化器中提供自定义加密密钥。以下是它在java中的实现方式：

props.put("key.deserializer","com.aexp.sec.kafka.serializers.CryptoDeSerializer");
props.put("value.deserializer","com.aexp.sec.kafka.serializers.CryptoDeSerializer");
props.put("CRYPTO_KEY_DESERIALIZER", StringDeserializer.class.getName());
props.put("CRYPTO_VALUE_DESERIALIZER", StringDeserializer.class.getName());
**props.put(String.format("%s_APIGEE_ENDPOINT", topic), apigeeUrl);
props.put("ENVIRONMENT", environment);
props.put(String.format("%s_KEY_IDENTIFIER", topic), keyIdentifier);
props.put(String.format("%s_APIGEE_CLIENT_ID", topic), clientID);
props.put(String.format("%s_APIGEE_CLIENT_SECRET", topic), clientSecret);**

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

while (true) {
            ConsumerRecords<String, String> records = consumer.poll(1000);
            for (ConsumerRecord<String, String> record : records) {
                buff.write(record.value()+"\n");
                buff.flush();
            }
        }

如何使用 python 自定义解密来解密字符串格式？

4

1 回答 1

0

这不是 Confluent Python 的限制。Java 程序依赖于这个调用的自定义反序列化器com.aexp.sec.kafka.serializers.CryptoDeSerializer，因此这里没有任何人可以提供帮助。您需要做的是在 Python 中提出一个相同的实现，它执行在自定义反序列化器中实现的相同逻辑，然后在创建使用者时注册它：

c = DeserializingConsumer({
'bootstrap.servers': 'kafka.abcd.com:1010',
'group.id':'SharedKafka2',
'auto.offset.reset': 'latest',
'security.protocol':'SSL',
'ssl.ca.location':'CARoot.pem',
'ssl.certificate.location':'certificate.pem',
'ssl.key.location':'key.pem',
'value.deserializer': YourCustomDeserializerInPython()

})

如果可能，请尝试联系在 Java 中实现CryptoDeSerializer该类的人，以了解您的 Python 代码的外观。

于 2020-07-24T14:47:53.723 回答