Use Custom TLS when Configuring Kafka (w/ Strimzi crd)

[themagiccog]

#4406 (comment)

Hello @scholzj the links you have above dont seem to be working. Can you please provided updated link?
I am having the same issues, I want my Kafka CR definition with listener.type: ingress, to use my own tls cert.

Currently, when i build, I dont see the secretName in the ingress resource that is built from kafka CR. Below is a snippet of my Kafka CR definition:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: kafka-cluster-nginx
spec:
  kafka:
    version: 3.6.0
    replicas: 3
    authorization:
      type: simple
    listeners:
    - name: plain
      port: 9092
      type: internal
      tls: false
      authentication:
        type: scram-sha-512
    - name: external
      port: 9094
      type: ingress
      tls: true 
      authentication:
        type: tls
      configuration:
        class: nginx
        brokerCertChainAndKey:
          secretName: kafka-wildcard-cert 
          certificate: tls.crt
          key: tls.key
        bootstrap:
          host: dev.kafka.me.com 
          annotations:
            nginx.ingress.kubernetes.io/ssl-passthrough: "true"
        brokers:
        - broker: 0
          host: devb0.kafka.me.com
          annotations:
            nginx.ingress.kubernetes.io/ssl-passthrough: "true"
        - broker: 1
          host: devb1.kafka.me.com
          annotations:
            nginx.ingress.kubernetes.io/ssl-passthrough: "true"
        - broker: 2
          host: devb2.kafka.me.com
          annotations:
            nginx.ingress.kubernetes.io/ssl-passthrough: "true"
    config:
      auto.create.topics.enable: "false"
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      log.message.format.version: "3.6"
      inter.broker.protocol.version: "3.6"
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 16Gi
        class: managed-premium
        deleteClaim: false
  zookeeper:
    replicas: 3
    storage:
      type: persistent-claim
      size: 4Gi
      class: managed-premium
      deleteClaim: false

[scholzj]

As explained many times in the past. You have to configure your Ingress to TLS passthrough => and as such, it does not care about the certificate. So your custom certificate is never configured in the Ingress, it is configured in the Kafka cluster.

[themagiccog]

Apologies if you have mentioned this. I realize that I have to do this but don't know how to do it.
I believe the links to the articles in the previous discussion points to the instructions but the links don't work.

I would appreciate if you point me in the right direction.
@scholzj

[scholzj]

Sorry, the right direction for what? You seem to have the custom certificate configured in brokerCertChainAndKey. So assuming you have the right content of the secret, that should be all you need. (Plus enabling the TLS Passthrough in the Kubernetes Nginx Ingress controller - IIRC there is some command line option for it in the controller).

[themagiccog]

Thanks @scholzj for your reply.

Here is what I have done:
I have created a nginx-ingress-controller and specified the passthru.
See helm command below:

helm upgrade --install  nginx-ingress-kafka ingress-nginx/ingress-nginx \
  --create-namespace \
  --namespace kafka-strimzi \
  --set controller.ingressClassResource.name=nginx-kafka \
  --set controller.extraArgs.default-ssl-certificate="kafka-strimzi/kafka-wildcard-cert" \
  --set controller.extraArgs.enable-ssl-passthrough=true \
  --set controller.service.annotations."service\.beta\.kubernetes\.io/azure-load-balancer-internal-subnet"=subnet-for-aks-apps \
  --set controller.service.annotations."service\.beta\.kubernetes\.io/azure-load-balancer-internal"=true \
  --set controller.service.annotations."service\.beta\.kubernetes\.io/azure-load-balancer-health-probe-request-path"=/healthz

I have also created my Kafka definition as follows:

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: kafka-cluster-nginx
spec:
  kafka:
    version: 3.6.0
    replicas: 3
    authorization:
      type: simple
    listeners:
    - name: plain
      port: 9092
      type: internal
      tls: false
      authentication:
        type: scram-sha-512
    - name: external
      port: 9094
      type: ingress
      tls: true 
      authentication:
        type: tls
      configuration:
        class: nginx-kafka
        brokerCertChainAndKey:
          secretName: kafka-wildcard-cert 
          certificate: tls.crt
          key: tls.key
        bootstrap:
          host: dev.kafka.cog.com
        brokers:
        - broker: 0
          host: devb0.kafka.cog.com
        - broker: 1
          host: devb1.kafka.cog.com
        - broker: 2
          host: devb2.kafka.cog.com
    config:
      auto.create.topics.enable: "false"
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      log.message.format.version: "3.6"
      inter.broker.protocol.version: "3.6"
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 16Gi
        class: managed-premium
        deleteClaim: false
  zookeeper:
    replicas: 3
    storage:
      type: persistent-claim
      size: 4Gi
      class: managed-premium
      deleteClaim: false

But when I connect to broker (using command):

 kafka-console-producer.sh \
  --bootstrap-server dev.kafka.cog.com:443 \
  --topic test-topic-nginx  \
  --producer-property security.protocol=SASL_SSL \
  --producer-property sasl.mechanism=SCRAM-SHA-512 \
  --producer-property sasl.jaas.config="org.apache.kafka.common.security.scram.ScramLoginModule required username=\"test-user-nginx\" password=\"xxxxxxxxxxxxxxxxxxxxxxxxxx\";"

I end up with this error:

>[2024-01-14 18:10:24,278] ERROR [Producer clientId=console-producer] Connection to node -1 (dev.kafka.cog.com/10.55.109.14:443) failed authentication due to: SSL handshake failed (org.apache.kafka.clients.NetworkClient)
[2024-01-14 18:10:24,279] WARN [Producer clientId=console-producer] Bootstrap broker dev.kafka.cog.com:443 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)
[2024-01-14 18:10:24,404] ERROR [Producer clientId=console-producer] Connection to node -1 (dev.kafka.cog.com/10.55.109.14:443) failed authentication due to: SSL handshake failed (org.apache.kafka.clients.NetworkClient)
[2024-01-14 18:10:24,404] WARN [Producer clientId=console-producer] Bootstrap broker dev.kafka.cog.com:443 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)

Is there something I am doing wrong?
I verified the cert is correct by running:
openssl s_client -connect dev.kafka.cog.com:443

Thanks for your anticipated feedback.

[scholzj]

Now here is the issue I found. My wild card is *.cog.com and I found that Kafka only recognizes one parameter when wildcard is used. In my case, my wildcard had 2 parameters, i.e. dev and kafka [dev.kafka].cog.com. This was the reason why I was getting the error above.

As far as I know, this is not a Kafka thing. This is how it works everywhere. The * wildcard can represent only one DNS segment.

However, because I have set up domains in my local PC only (i.e. host-file), the interconnections with the brokers do not recognize the domain (as the host-file config doesn't propagate into the k8s cluster).

Sorry, I do not folow this. The interconnections between brokers have their own listeners and advertised hosts. They do not care about your domain or your loadbalancer / ingress listener.

---

I'm also not sure what do you expect from the service.beta.kubernetes.io/azure-load-balancer-health-probe-request-path annotation. Kafka is not HTTP, so there is no /healthz endpoint. So not sure what that does and how it impacts your services. But I guess you need to clarify that with Azure if needed.

Also, the way you have it configured, the advertised name for the brokers will be the address of the loadbalancer. So not sure if that will really use your DNS names. Normally, the LBs get some IP or Azure DNS name that will be used (which in general is fine, but you might need to reflect it in your certificate).

[themagiccog]

This is the error I am getting:

[2024-01-17 10:32:24,251] DEBUG [Producer clientId=console-producer] Initialize connection to node 10.55.109.8:9094 (id: 2 rack: null) for sending metadata request (org.apache.kafka.clients.NetworkClient)
[2024-01-17 10:32:24,251] DEBUG Resolved host 10.55.109.8 as 10.55.109.8 (org.apache.kafka.clients.ClientUtils)
[2024-01-17 10:32:24,251] DEBUG [Producer clientId=console-producer] Initiating connection to node 10.55.109.8:9094 (id: 2 rack: null) using address /10.55.109.8 (org.apache.kafka.clients.NetworkClient)
[2024-01-17 10:32:25,268] DEBUG [Producer clientId=console-producer] Set SASL client state to SEND_APIVERSIONS_REQUEST (org.apache.kafka.common.security.authenticator.SaslClientAuthenticator)
[2024-01-17 10:32:25,269] DEBUG [Producer clientId=console-producer] Creating SaslClient: client=null;service=kafka;serviceHostname=10.55.109.8;mechs=[SCRAM-SHA-512] (org.apache.kafka.common.security.authenticator.SaslClientAuthenticator)
[2024-01-17 10:32:25,269] DEBUG Setting SASL/SCRAM_SHA_512 client state to SEND_CLIENT_FIRST_MESSAGE (org.apache.kafka.common.security.scram.internals.ScramSaslClient)
[2024-01-17 10:32:25,270] DEBUG [Producer clientId=console-producer] Created socket with SO_RCVBUF = 32768, SO_SNDBUF = 102400, SO_TIMEOUT = 0 to node 2 (org.apache.kafka.common.network.Selector)
[2024-01-17 10:32:25,271] DEBUG [Producer clientId=console-producer] Completed connection to node 2. Fetching API versions. (org.apache.kafka.clients.NetworkClient)
[2024-01-17 10:32:25,283] DEBUG [SslTransportLayer channelId=2 key=channel=java.nio.channels.SocketChannel[connection-pending remote=/10.55.109.8:9094], selector=sun.nio.ch.EPollSelectorImpl@27c1ac82, interestOps=8, readyOps=0] SSL Handshake failed (org.apache.kafka.common.network.SslTransportLayer)
javax.net.ssl.SSLHandshakeException: No subject alternative names matching IP address 10.55.109.8 found
        at java.base/sun.security.ssl.Alert.createSSLException(Alert.java:131)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:360)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:303)
        at java.base/sun.security.ssl.TransportContext.fatal(TransportContext.java:298)
        at java.base/sun.security.ssl.CertificateMessage$T13CertificateConsumer.checkServerCerts(CertificateMessage.java:1357)
        at java.base/sun.security.ssl.CertificateMessage$T13CertificateConsumer.onConsumeCertificate(CertificateMessage.java:1232)
        at java.base/sun.security.ssl.CertificateMessage$T13CertificateConsumer.consume(CertificateMessage.java:1175)
        at java.base/sun.security.ssl.SSLHandshake.consume(SSLHandshake.java:392)
        at java.base/sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:443)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1076)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1063)
        at java.base/java.security.AccessController.doPrivileged(Native Method)
        at java.base/sun.security.ssl.SSLEngineImpl$DelegatedTask.run(SSLEngineImpl.java:1010)
        at org.apache.kafka.common.network.SslTransportLayer.runDelegatedTasks(SslTransportLayer.java:435)
        at org.apache.kafka.common.network.SslTransportLayer.handshakeUnwrap(SslTransportLayer.java:523)
        at org.apache.kafka.common.network.SslTransportLayer.doHandshake(SslTransportLayer.java:373)
        at org.apache.kafka.common.network.SslTransportLayer.handshake(SslTransportLayer.java:293)
        at org.apache.kafka.common.network.KafkaChannel.prepare(KafkaChannel.java:178)
        at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:543)
        at org.apache.kafka.common.network.Selector.poll(Selector.java:481)
        at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:571)
        at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:344)
        at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:247)
        at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.security.cert.CertificateException: No subject alternative names matching IP address 10.55.109.8 found
        at java.base/sun.security.util.HostnameChecker.matchIP(HostnameChecker.java:165)
        at java.base/sun.security.util.HostnameChecker.match(HostnameChecker.java:101)
        at java.base/sun.security.ssl.X509TrustManagerImpl.checkIdentity(X509TrustManagerImpl.java:461)
        at java.base/sun.security.ssl.X509TrustManagerImpl.checkIdentity(X509TrustManagerImpl.java:435)
        at java.base/sun.security.ssl.X509TrustManagerImpl.checkTrusted(X509TrustManagerImpl.java:283)
        at java.base/sun.security.ssl.X509TrustManagerImpl.checkServerTrusted(X509TrustManagerImpl.java:141)
        at java.base/sun.security.ssl.CertificateMessage$T13CertificateConsumer.checkServerCerts(CertificateMessage.java:1335)
        ... 19 more
[2024-01-17 10:32:25,283] INFO [Producer clientId=console-producer] Failed authentication with 10.55.109.8/10.55.109.8 (channelId=2) (SSL handshake failed) (org.apache.kafka.common.network.Selector)
[2024-01-17 10:32:25,284] INFO [Producer clientId=console-producer] Node 2 disconnected. (org.apache.kafka.clients.NetworkClient)
[2024-01-17 10:32:25,284] ERROR [Producer clientId=console-producer] Connection to node 2 (10.55.109.8/10.55.109.8:9094) failed authentication due to: SSL handshake failed (org.apache.kafka.clients.NetworkClient)

Like you said, it seems like the interconnections btwn brokers is just IP Address Based. Can I assign hostnames to the broker IPs so that it can be recognized by the certs?

[scholzj]

You can use the advertisedHost field int he Kafka CR to redefine it. Check the API reference for the details and examples.

[themagiccog]

Hey @scholzj
I just tried the advertisedHost and I am happy to report that it works!!

I really appreciate your assistance with this so far.
I will post a snippet of my code in the main thread and close this discussion when I am done.
Once again, thank you!

[scholzj]

Glad it now works for you and thanks for sharing the write-up you shared. That could be very useful for others! 👍

[themagiccog]

Intro

If you are having some issues with Custom TLS (when using security.protocol: SASL_SSL) you can follow the steps below.

My Kafka will be using a loadbalancer finalizer, and I am using Azure's Internal Load balancer for this. Feel free to modify this accordingly based on your cloud provider.

Quick note on security protocols:
SSL: means that you are using TLS for both the Broker comms (tls certs in cluster certs secrets) and also TLS for External comms (tls certs in user secrets)
SASL_SSL: means that you are using SASL (SCRAM-SHA-512) for Broker Comms and TLS for External comms (tls certs in user secrets) <--- I AM USING THIS
SASL_PLAINTEXT: means that you are using SASL (SCRAM-SHA-512) for Broker comms and PLAINTEXT for External comms (passwords in user secrets)

---

Prerequisites

Install OLM (Operator Lifestyle Manager) and the Strimzi Operator by running commands below.

# Install Operator Lifecycle Manager
curl -L https://github.com/operator-framework/operator-lifecycle-manager/releases/download/v0.26.0/install.sh -o install.sh
chmod +x install.sh
./install.sh v0.26.0

# Install the Strimzi Operator
kubectl apply -f https://operatorhub.io/install/strimzi-kafka-operator.yaml

# Create namespace and use
kubectl create ns kafka-strimzi
kubectl config set-context --current --namespace=kafka-strimzi

---

Create Secret w/ Custom Certs

You need to have your certificate and private-key ready.

Important Notes:

• Your certificate must contain the Domain Names that will be associated with the Kafka Endpoints.
• Ensure that the Cert also includes Root CA for the cert,
• Ensure that the Certs dont have bag-attributes (for a less noisy cert)
• and ensure private-key is decrypted.

The next thing is to create a secret with the Cert and Key in the same namespace you intend to deploy the cluster.
In my case, my cert is a wildcard cert (with ca chain contained) and I am using a key that has already been decrypted.

# Create Secret w/ Certs
kubectl create secret tls kafka-wildcard-cert \
  --key wildcard-decrypted.key \
  --cert wildcard-chain.crt

Optional:
If you are using a Root CA Cert that is not among the popular ones, you may need to create a truststore so that it can be used by client to communicate with the Kafka Cluster:

keytool -keystore client.truststore.jks -alias CARoot -import -file wildcard-chain.crt -storepass password -noprompt

---

Create Kafka Cluster

Now its time to create the Kafka Cluster.
We have to ensure that we specify the brokerCertChainAndKey configuration and specify the advertisedHost property on the brokers. This is very necessary if you only have Domain Names specified in your certificate.

Our definition file is as follows:
kubectl apply -f kafka-cluster.yaml

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: kafka-cluster
spec:
  maintenanceTimeWindows:
    - "* * 7-9 * * ?"
  clientsCa:
    generateCertificateAuthority: true
    renewalDays: 30
    validityDays: 365
  clusterCa:
    generateCertificateAuthority: true
    renewalDays: 30
    validityDays: 365
  kafka:
    version: 3.6.0
    replicas: 3
    authorization:
      type: simple
    listeners:
    - name: plain
      port: 9092
      type: internal
      tls: false
      authentication:
        type: scram-sha-512
    - name: external
      port: 9094
      type: loadbalancer
      tls: true #Since we are using Custom TLS for External Comms, we add the brokerCertChainAndKey below
      authentication:
        type: scram-sha-512 
      configuration:
        brokerCertChainAndKey:
          secretName: kafka-wildcard-cert 
          certificate: tls.crt
          key: tls.key
        bootstrap:
          annotations:
            service.beta.kubernetes.io/azure-load-balancer-health-probe-request-path: /healthz
            service.beta.kubernetes.io/azure-load-balancer-internal: "true"
            service.beta.kubernetes.io/azure-load-balancer-internal-subnet: subnet-for-aks-apps
        brokers:
        - broker: 0
          annotations:
            service.beta.kubernetes.io/azure-load-balancer-health-probe-request-path: /healthz
            service.beta.kubernetes.io/azure-load-balancer-internal: "true"
            service.beta.kubernetes.io/azure-load-balancer-internal-subnet: subnet-for-aks-apps
          advertisedHost: b0-<domain-name>
        - broker: 1
          annotations:
            service.beta.kubernetes.io/azure-load-balancer-health-probe-request-path: /healthz
            service.beta.kubernetes.io/azure-load-balancer-internal: "true"
            service.beta.kubernetes.io/azure-load-balancer-internal-subnet: subnet-for-aks-apps
          advertisedHost: b1-<domain-name>
        - broker: 2
          annotations:
            service.beta.kubernetes.io/azure-load-balancer-health-probe-request-path: /healthz
            service.beta.kubernetes.io/azure-load-balancer-internal: "true"
            service.beta.kubernetes.io/azure-load-balancer-internal-subnet: subnet-for-aks-apps
          advertisedHost: b2-<domain-name>
    config:
      auto.create.topics.enable: "false"
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      log.message.format.version: "3.6"
      inter.broker.protocol.version: "3.6"
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 16Gi
        class: managed-premium
        deleteClaim: false
  zookeeper:
    replicas: 3
    storage:
      type: persistent-claim
      size: 4Gi
      class: managed-premium
      deleteClaim: false
  entityOperator:
    topicOperator: {}
    userOperator: {}
  cruiseControl: {}

You can also create topics and user as follows:

kubectl apply -f test-topic.yaml

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: test-topic
  labels:
    strimzi.io/cluster: kafka-cluster #kafka-cluster-tls-nginx #kafka-cluster-tls-lb
spec:
  partitions: 3
  replicas: 3

test-user.yaml

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: test-user
  labels:
    strimzi.io/cluster: kafka-cluster #kafka-cluster-tls-nginx #kafka-cluster-tls-lb
spec:
  authentication:
    type: scram-sha-512
  authorization:
    type: simple
    acls:
      - resource:
          type: topic
          name: '*'
          patternType: literal
        operation: Read
      - resource:
          type: topic
          name: '*'
          patternType: literal
        operation: Write
      - resource:
          type: group
          name: '*'
          patternType: literal
        operation: Read
      - resource:
          type: group
          name: '*'
          patternType: literal
        operation: Write

In other to test the Kafka communication, we can run the producer and consumer commands as shown:
Ensure you have Kafka installed locally or you can leverage 3rd party tools.

You can get the test-user password with the following command:

kubectl get secret test-user -o jsonpath='{.data.user\.password}' | base64 -d

producer:

#producer
bin/kafka-console-producer.sh \
  --bootstrap-server <domain-name>:9094 \
  --topic test-topic \
  --producer-property security.protocol=SASL_SSL \
  --producer-property sasl.mechanism=SCRAM-SHA-512 \
  --producer-property sasl.jaas.config="org.apache.kafka.common.security.scram.ScramLoginModule required username=\"test-user\" password=\"<user-password>\";"

consumer

bin/kafka-console-consumer.sh \
  --bootstrap-server <domain-name>:9094 \
  --topic test-topic \
  --from-beginning \
  --consumer-property security.protocol=SASL_SSL \
  --consumer-property sasl.mechanism=SCRAM-SHA-512 \
  --consumer-property sasl.jaas.config="org.apache.kafka.common.security.scram.ScramLoginModule required username=\"test-user\" password=\"<user-password>\";" \
  --group my-consumer-group

Optional:
If needed, you can create a trustkey with the following command below.
Save in location so app has access to it (if necessary). Then you can specify truststore and location as one of the properties, i.e.:
producer:

bin/kafka-console-producer.sh \
  --bootstrap-server <domain-name>:9094 \
  --topic test-topic \
  --producer-property security.protocol=SASL_SSL \
  --producer-property sasl.mechanism=SCRAM-SHA-512 \
  --producer-property ssl.truststore.location=client.truststore.jks \
  --producer-property ssl.truststore.password=password \
  --producer-property sasl.jaas.config="org.apache.kafka.common.security.scram.ScramLoginModule required username=\"test-user\" password=\"<user-password>\";"

consumer:

bin/kafka-console-consumer.sh \
  --bootstrap-server <domain-name>:9094 \
  --topic test-topic \
  --from-beginning \
  --consumer-property security.protocol=SASL_SSL \
  --consumer-property sasl.mechanism=SCRAM-SHA-512 \
  --consumer-property ssl.truststore.location=client.truststore.jks \
  --consumer-property ssl.truststore.password=password \
  --consumer-property sasl.jaas.config="org.apache.kafka.common.security.scram.ScramLoginModule required username=\"test-user\" password=\"<user-password>\";" \
  --group my-consumer-group

You should now be able to communicate with your cluster using Custom TLS.
Thank you.