Deploying KafkaConnect with TLS

[GrSof]

Hello,

I have enabled TLS communication in my Kafka cluster to connect with an external client.

>>>>>>>>:~/kafka/strimzi-0.27.1$ cat examples/kafka/kafka-ephemeral.yaml
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    version: 2.8.0
    replicas: 3
    authorization:
      type: simple
      superUsers:
        - CN = my-user
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
      - name: external
        port: 9094
        type: nodeport
        tls: true
        authentication:
          type: tls
        configuration:
          bootstrap:
             nodePort: 32100
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      default.replication.factor: 1
      min.insync.replicas: 1
      inter.broker.protocol.version: "2.8"
    storage:
      type: ephemeral
  zookeeper:
    replicas: 3
    storage:
      type: ephemeral
  entityOperator:
    topicOperator: {}
    userOperator: {}

I have also create a KafkaUser

>>>>>>>>:~/kafka/strimzi-0.27.1$ cat examples/user/kafka-user-original.yaml
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: tls
  authorization:
    type: simple
    acls:
      # Example consumer Acls for topic my-topic using consumer group my-group
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Read
        host: "*"
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Describe
        host: "*"
#      - resource:
#          type: group
#          name: my-group
#          patternType: literal
#        operation: Read
#        host: "*"
      # Example Producer Acls for topic my-topic
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Write
        host: "*"
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Create
        host: "*"
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Describe
        host: "*"

My KafkaConnect resource looks like this:

>>>>>>>>:~/kafka/strimzi-0.27.1$ cat examples/connect/kafka-connect.yaml
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
#  annotations:
#  # use-connector-resources configures this KafkaConnect
#  # to use KafkaConnector resources to avoid
#  # needing to call the Connect REST API directly
#    strimzi.io/use-connector-resources: "true"
spec:
  version: 2.8.0
  replicas: 1
  authentication:
    type: tls
    certificateAndKey:
      certificate: user.crt
      key: user.key
      secretName: my-user
  bootstrapServers: my-cluster-kafka-bootstrap:9093
  tls:
    trustedCertificates:
      - secretName: my-cluster-cluster-ca-cert
        certificate: ca.crt
      - secretName: my-user
        certificate: ca.crt
  config:
    # group.id: connect-cluster
    offset.storage.topic: connect-cluster-offsets
    config.storage.topic: connect-cluster-configs
    status.storage.topic: connect-cluster-status
    # -1 means it will use the default replication factor configured in the broker
    config.storage.replication.factor: -1
    offset.storage.replication.factor: -1
    status.storage.replication.factor: -1

When I deploy the KafkaConnect I get the following error:

2022-10-21 09:09:14,415 ERROR [Worker clientId=connect-1, groupId=connect-cluster] Uncaught exception in herder work thread, exiting:  (org.apache.kafka.connect.runtime.distributed.DistributedHerder) [DistributedHerder-connect-1-1]
org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [connect-cluster-offsets]

How does TLS influence the connection to the connect topics? Am I missing something in my yaml files? Are the certificates well configured in tha KafkaConnect yaml file?

Thank you very much

[kinihun]

Kafka Connect requires more permission in order to connect to Kafka, it needs to be able to produce and consume to offset, config and status topis. The permissions on the Kafka User is insufficient.

[scholzj]

I think you are approaching it from the wrong end. You should first sort out your listener configuration. Decide who connects where and how they should be configured, what security you actually want to use etc.. Once you have that solved, it should be more clear to you what do you want to do with your Connect. The docs and the examples should be able to help you with the details about the listeners.

[GrSof]

Hello, again,

After a long time, I think now I know better Kafka to have a clear picture of what I wanted. I have 3 listeners:

• 2 internal: 1 plain and 1 with TLS. For communication between components of the cluster.
• 1 external: with authentication and authorization. For clients outside the cluster to connect to.

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    version: 2.8.0
    replicas: 3
    authorization:
      type: simple
      superUsers:
        - CN=my-user
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
      - name: external
        port: 9094
        type: nodeport
        tls: true
        authentication:
          type: tls
        configuration:
          bootstrap:
             nodePort: 32100
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      default.replication.factor: 1
      min.insync.replicas: 1
      inter.broker.protocol.version: "2.8"
    storage:
      type: ephemeral
  zookeeper:
    replicas: 3
    storage:
      type: ephemeral
  entityOperator:
    topicOperator: {}
    userOperator: {}

Once, that is in place and works correctly. I create a KafkaUser called my-user.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  authentication:
    type: tls
  authorization:
    type: simple
    acls:
      # Example consumer Acls for topic my-topic using consumer group my-group
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Read
        host: "*"
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Describe
        host: "*"
      - resource:
          type: group
          name: connect-cluster
          patternType: literal
        operation: Read
        host: "*"
      # Example Producer Acls for topic my-topic
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Write
        host: "*"
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Create
        host: "*"
      - resource:
          type: topic
          name: my-topic
          patternType: literal
        operation: Describe
        host: "*"

That works fine and I can communicate with the external Listener (on NodePort 32100) using the certificates generated.

Now, I want to deplot the KafkaConnect resource. This component is within the cluster so I want to use the internal listener (my-cluster-kafka-bootstrap.my-kafka-project.svc:9092), without authentication or authorization.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  version: 2.8.0
  image: kafka-connect:0.3
  replicas: 1
  bootstrapServers: my-cluster-kafka-bootstrap.my-kafka-project.svc:9092
  config:
    group.id: connect-cluster
    offset.storage.topic: connect-cluster-offsets
    config.storage.topic: connect-cluster-configs
    status.storage.topic: connect-cluster-status
    config.storage.replication.factor: 3
    offset.storage.replication.factor: 1
    status.storage.replication.factor: 1
  template:
    pod:
      imagePullSecrets:
        - name: dockersecret1

In that case, the yaml above, that deploys the KafkaConnect instance, should work because it uses the internal plain listener. Is that rigth?
Is this a good listeners configuration? Please help me untangle this.

Best

Sofia

[scholzj]

I'm afraid this is a wrong expectation. The authorization is in Kafka configured on a cluster-wide level. So it applies to all listeners. For the external one where you have enabled authentication, the user will be authenticated and authorized against its ACLs. For the internal listeners without the authentication, all users will have the username ANONYMOUS and will be authorized against it. That means that by default, they will not be allowed to do anything.

Normally the best (and recommended solution) would be to use authentication on all listeners if you want to use authorization. But if it fits you security requirements, you can also for example:

• Disable authorization completely
• Make the ANONYMOUS used a super user in .spec.kafka.authorization.superUsers and essentially bypass the the authorization for the listeners without authentication

These are of course not completely secure, but if it works for you, you can of course do that 🤷 .

[GrSof]

sHello,

Your latest message got me to think and I've tried to go with disable authorization:

Option 1: Disable authorization completely. My favorite. Here I get to deploy the KafkaConnect instance and a Connector but I don't seem to be able to find the certificate necessary to connect using a kafkacat client.
Files:
------------kafka-ephemeral.yaml--------------

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    version: 2.8.0
    replicas: 3
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
      - name: external
        port: 9094
        type: nodeport
        tls: true
        authentication:
          type: tls
        configuration:
          bootstrap:
             nodePort: 32100
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      default.replication.factor: 1
      min.insync.replicas: 1
      inter.broker.protocol.version: "2.8"
    storage:
      type: ephemeral
  zookeeper:
    replicas: 3
    storage:
      type: ephemeral
  entityOperator:
    topicOperator: {}
    userOperator: {}

------------kafka-connect.yaml------------

kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  version: 2.8.0
  image: kafka-connect:0.3
  replicas: 1
  bootstrapServers: my-cluster-kafka-bootstrap.my-kafka-project.svc:9092
  config:
    group.id: connect-cluster
    offset.storage.topic: connect-cluster-offsets
    config.storage.topic: connect-cluster-configs
    status.storage.topic: connect-cluster-status
    config.storage.replication.factor: 3
    offset.storage.replication.factor: 1
    status.storage.replication.factor: 1
  template:
    pod:
      imagePullSecrets:
        - name: dockersecret1

Above I don't need any authentication or authorization parameters because I'm using the plain listener (9092) and there is no authorization.

The secrets that are created after that deployment are:

NAME                                     TYPE                                  DATA   AGE
default-token-zmg6c                      kubernetes.io/service-account-token   3      18m
my-cluster-clients-ca                    Opaque                                1      18m
my-cluster-clients-ca-cert               Opaque                                3      18m
my-cluster-cluster-ca                    Opaque                                1      18m
my-cluster-cluster-ca-cert               Opaque                                3      18m
my-cluster-cluster-operator-certs        Opaque                                4      18m
my-cluster-entity-operator-certs         Opaque                                4      16m
my-cluster-entity-operator-token-x6qg4   kubernetes.io/service-account-token   3      16m
my-cluster-kafka-brokers                 Opaque                                12     17m
my-cluster-kafka-token-mdmvf             kubernetes.io/service-account-token   3      17m
my-cluster-zookeeper-nodes               Opaque                                12     18m
my-cluster-zookeeper-token-4mxgx         kubernetes.io/service-account-token   3      18m
my-connect-cluster-connect-token-vkntc   kubernetes.io/service-account-token   3      15m

The kafkacat command for the consumer, when there is authentication and authorization is:

kafkacat -C -b 10.0.32.11:32100 -t my-topic -X security.protocol=SSL -X ssl.ca.location=ca.crt -X ssl.certificate.location=my-user.crt -X ssl.key.location=my-user.key

Now, if I don't have authorization the command should work like this:?

kafkacat -C -b 10.0.32.11:32100 -t my-topic -X security.protocol=SSL -X ssl.ca.location=ca.crt

But I get this

~/kafka-certificates$ kafkacat -C -b 10.0.32.11:32100 -t my-topic -X security.protocol=SSL -X ssl.ca.location=ca.crt
% ERROR: Failed to query metadata for topic my-topic: Local: Broker transport failure

Logs kafka

2023-03-17 13:41:56,371 INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Failed authentication with /10.42.1.0 (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SSL-10]
2023-03-17 13:41:57,586 INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Failed authentication with /10.42.1.0 (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SSL-11]
2023-03-17 13:42:05,716 INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Failed authentication with /10.42.1.0 (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SSL-12]
2023-03-17 13:42:06,924 INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Failed authentication with /10.42.1.0 (SSL handshake failed) (org.apache.kafka.common.network.Selector) [data-plane-kafka-network-thread-0-ListenerName(EXTERNAL-9094)-SSL-10]

Am I missing a certificate? Which one? I know you are not the developers of kafkacat but maybe you have an idea of what happens. Sorry to bother if this question is out of place.

[scholzj]

The ca.crt should come from the secret named <cluster-name>-cluster-ca-cert -> in your case form my-cluster-cluster-ca-cert