Internal SSL listener just won't work!

[ETisREAL]

Bug Description

I am trying to use an SSL internal lister, however I can't get it to work for the life of me. I've look through ALL the docs, issues etc and still I can't get it to work and I really don't know why

Steps to reproduce

Here is the error I am getting:
%3|1701600701.934|FAIL|sample#producer-1| [thrd:ssl://kafka-kafka-bootstrap.default.svc:9093/bootstrap]: ssl://kafka-kafka-bootstrap.default.svc:9093/bootstrap: SSL handshake failed: error:1416F086:SSL routines:tls_process_server_certificate:certificate verify failed: broker certificate could not be verified, verify that ssl.ca.location is correctly configured or root CA certificates are installed (install ca-certificates package) (after 13ms in state SSL_HANDSHAKE, 1 identical error(s) suppressed)

This is my client code:

package main

import (
	"fmt"
	"io"
	"net/http"
	"github.com/confluentinc/confluent-kafka-go/kafka"
)

func getRoot(w http.ResponseWriter, r http.Request) {
	io.WriteString(w, "This is my website!\n")
	w.Write([]byte{})
}

func startHTTPServer() {
	http.HandleFunc("/", getRoot)

	err := http.ListenAndServe(":8080", nil)
	if err != nil {
		fmt.Printf("error instantiating server: %v", err)
	}
}

func main() {

	go startHTTPServer()

	bootstrapServers := "kafka-kafka-bootstrap.default.svc:9093"
    caLocation := "/certs/ca.crt"
    topic := "test-topic"

    userCertLocation := "/certs/user.crt"
    userKeyLocation := "/certs/user.key"
    userKeyPassword := "pzy1riN1r5vzSTpfoH0mT7LSM27WrM9j"

    producerConfig := &kafka.ConfigMap{
		"bootstrap.servers": bootstrapServers,
		"security.protocol": "SSL",
		"ssl.ca.location": caLocation,
		"ssl.certificate.location": userCertLocation,
		"ssl.key.location": userKeyLocation,
		"ssl.key.password": userKeyPassword,
		"client.id": "sample",
		"acks": "all",
	}

	
	p, err := kafka.NewProducer(producerConfig)
	if err != nil {
		fmt.Printf("Failed to create producer: %s\n", err)
		return
	}

	delivery_chan := make(chan kafka.Event, 100)
	err = p.Produce(&kafka.Message{
		TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
		Value: []byte("send some value")},
		delivery_chan,
	)

	e := <-delivery_chan
	m := e.(*kafka.Message)

	if m.TopicPartition.Error != nil {
		fmt.Printf("Delivery failed: %v\n", m.TopicPartition.Error)
	} else {
		fmt.Printf(
			"Delivered message to topic %s [%d] at offset %v\n",
			*m.TopicPartition.Topic,
			m.TopicPartition.Partition,
			m.TopicPartition.Offset,
		)
	}
	close(delivery_chan)
}

This is my kafka cluster configuration:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: kafka
spec:
  kafka:
    version: 3.6.0
    replicas: 3
    authorization:
      type: simple
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      default.replication.factor: 3
      min.insync.replicas: 2
      inter.broker.protocol.version: "3.6"
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 4Gi
        deleteClaim: true
  zookeeper:
    replicas: 3
    storage:
      type: persistent-claim
      size: 4Gi
      deleteClaim: true
  entityOperator:
    tlsSidecar:
      resources:
        requests:
          cpu: 100m
          memory: 64Mi
        limits:
          cpu: 150m
          memory: 128Mi
    topicOperator: {}
    userOperator: {}

as far as I can tell, this is what I am supposed to do cluster's side

Then I am creating a user using the relative operator. Here is the manifest:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: go-client
  labels:
    strimzi.io/cluster: kafka
spec:
  authentication:
    type: tls

and to ensure the mTLS works properly, I create a svc for the deployment to get a DNS record that corresponds t the CN of the created user (In my case go-client)

apiVersion: apps/v1
kind: Deployment
metadata:
  name: go-client
spec:
  selector:
    matchLabels:
      app: go-client
  template:
    metadata:
      labels:
        app: go-client
    spec:
      containers:
      - name: go-client
        image: golang
        tty: true
        resources:
          requests:
            memory: "128Mi"
            cpu: "100m"
          limits:
            memory: "256Mi"
            cpu: "200m"
        ports:
        - containerPort: 8080
        command: ["/bin/bash"]
        args:
          - "-c"
          - apt update && apt install -y nano gcc ca-certificates && tail -f /dev/null
        volumeMounts:
          - name: secret-volume
            mountPath: /certs/
      volumes:
      - name: secret-volume
        secret:
          secretName: go-client

---
apiVersion: v1
kind: Service
metadata:
  name: go-client
spec:
  selector:
    app: go-client
  ports:
  - port: 8080
    targetPort: 8080

How is this not working??

Expected behavior

I should be able to connect successfully to Kafka using mTLS, since I:

1. Created the user using the users-operator
2. Mounted the certificates, key and password and pointed to them correctly in my code
3. Created a svc for my deployment to create a DNS record

Strimzi version

0.38.0

Kubernetes version

1.27.3-gke.100

Installation method

Helm chart

Infrastructure

GKE

Configuration files and logs

No response

Additional context

No response

[scholzj]

• This does not seem like a Strimzi issue, so I converted it to Discussion
• Can you please fix the formatting so that the code and YAMLs are readable?
• You should also explain where did you get the ca.crt file -> did you get it from the <cluster-name>-cluster-ca-cert secret?

[ETisREAL]

Hi :) Thank you for the quick reply first of all.

I fixed the formatting, sorry it was a backtick issue...

As per the ca.crt, I am pointing to the one that is included in the created user's secret (and mounted together with the other fields in the secret), together with the user.crt, user.key and user.password (doing a cat of the file and inserting the content of course, not just pointing to the file)

For discolure, I tried pointing to the '-cluster-ca-cert' ca.crt , but in that case I still get an error

[scholzj]

You should definitely use the ca.crt from the <cluster-name>-cluster-ca-cert secret and not from the user secret. The User secret has the CA that signed it, but that is not what signed the user certificates. I do not know the Confluent clients -> but I'm not sure why do you set the key password => the PEM files are unencrypted. The password in the user secret applies to the PKCS12 file.

[ETisREAL]

OMG using the cluster CA and not using the password did it. You have no idea what a favor you just did me. I owe you one man, fr.

Thank you so, so much man. Will give it back to you someday :)

[scholzj]

Glad it helped 👍