我需要通过使用 truststore 和 keystore 与客户端建立 TLS 连接。
我的存储文件格式为 P12。我的 Keystore 只有一个私钥和一个公钥。我的 Truststore 只有一个集群证书。
我的问题是,我无法连接到我的 ssl-Kafka。可能是握手失败,但 kafka-go 没有输出任何消息。
以下是我的代码。也许是配置问题吗? 有人知道如何打印错误吗?或者我的配置有什么错误吗?
我的问题是,我无法连接到我的 ssl-Kafka。可能是握手失败,但 kafka-go 没有输出任何消息。
以下是我的代码。也许是配置问题吗? 有人知道如何打印错误吗?或者我的配置有什么错误吗?
getKafkaReader(topic string) *kafka.Reader {
if consumer == nil {
consumer = kafka.NewReader(kafka.ReaderConfig{
Brokers: []string{kafkaConfig.Host},
GroupID: os.Getenv("KAFKA_CONSUMER_GROUP"),
Topic: topic,
Partition: 0,
MinBytes: 10e2, // 1KB
MaxBytes: 10e5, // 1MB
Dialer: getDialer(),
})
}
return consumer
}
func getDialer() *kafka.Dialer {
dialer := &kafka.Dialer{
Timeout: 5 * time.Second,
DualStack: true,
TLS: tlsConfig(),
}
return dialer
}
func tlsConfig() *tls.Config {
// Keystore
keys, _ := ioutil.ReadFile(kafkaConfig.KeyStoreLocation)
blocks, err := p12.ToPEM(keys, kafkaConfig.KeyStorePassword)
if err != nil {
log.Fatal(err.Error())
}
var pemData []byte
for test, b := range blocks {
_ = test
pemData = append(pemData, pem.EncodeToMemory(b)...)
}
cert, err := tls.X509KeyPair(pemData, pemData)
if err != nil {
log.Fatal(err.Error())
}
//Truststore
caCert, err := ioutil.ReadFile("./certificates/ca.pem")
if err != nil {
log.Fatal(err)
}
caCertPool := x509.NewCertPool()
caCertPool.AppendCertsFromPEM(caCert)
config := &tls.Config{
Certificates: []tls.Certificate{cert},
RootCAs: caCertPool,
}
return config
}