Argo Kafka Eventsource配置SASL_SSL(SCRAM-SHA-512)遇阻，求安全方案

使用v1.7.6版本的Argo Kafka Eventsource，配置SASL_SSL（SCRAM-SHA-512）认证时无法正常连接Kafka集群，错误提示为kafka: client has run out of available brokers to talk to: unexpected EOF。仅当添加insecureSkipVerify: true跳过TLS证书验证时，连接才能正常工作，但该方式存在安全风险，需寻求安全的解决办法。

原EventSource配置 #

apiVersion: argoproj.io/v1alpha1
kind: EventSource
metadata:
  name: kafka
  namespace: MY_NAMESPACE
spec:
  eventBusName: MY_EVENTBUS
  kafka:
    kafka-event:
      # kafka broker url
      url: MY_BROKER_URL
      # name of the kafka topic
      topic: MY_TOPIC
      # jsonBody specifies that all event body payload coming from this
      # source will be JSON
      jsonBody: true
      # partition id
      # partition: "1"
      # optional backoff time for connection retries.
      # if not provided, default connection backoff time will be used.
      connectionBackoff:
        # duration in nanoseconds, or strings like "3s", "2m". following value is 10 seconds
        duration: 10s
        # how many backoffs
        steps: 5
        # factor to increase on each step.
        # setting factor > 1 makes backoff exponential.
        factor: 2
        jitter: 0.2
      # Use a consumer group, if this is used you do not need to specify a "partition: <id>"
      consumerGroup:
        groupName: argo-test-group
        oldest: false
        rebalanceStrategy: range
        limitEventsPerSecond: 1
        version: "2.5.0"
      #    Enable SASL authentication (not to be used with TLS)
      sasl:
        mechanism: SCRAM-SHA-512
        passwordSecret:
          key: password
          name: kafka-sasl-creds
        userSecret:
          key: username
          name: kafka-sasl-creds

错误日志 #

{"level":"info","ts":1679338599.3659816,"logger":"argo-events.eventsource","caller":"kafka/start.go:81","msg":"start kafka event source...","eventSourceName":"kafka","eventSourceType":"kafka","eventName":"kafka-event"}
{"level":"error","ts":1679338600.1314366,"logger":"argo-events.eventsource","caller":"kafka/start.go:122","msg":"Error creating consumer group client: kafka: client has run out of available brokers to talk to: unexpected EOF","eventSourceName":"kafka","eventSourceType":"kafka","eventName":"kafka-event","stacktrace":"github.com/argoproj/argo-events/eventsources/sources/kafka.(*EventListener).consumerGroupConsumer\n\t/home/runner/work/argo-events/argo-events/eventsources/sources/kafka/start.go:122\ngithub.com/argoproj/argo-events/eventsources/sources/kafka.(*EventListener).StartListening\n\t/home/runner/work/argo-events/argo-events/eventsources/sources/kafka/start.go:87\ngithub.com/argoproj/argo-events/eventsources.(*EventSourceAdaptor).run.func3.1\n\t/home/runner/work/argo-events/argo-events/eventsources/eventing.go:503\ngithub.com/argoproj/argo-events/common.DoWithRetry.func1\n\t/home/runner/work/argo-events/argo-events/common/retry.go:106\nk8s.io/apimachinery/pkg/util/wait.ConditionFunc.WithContext.func1\n\t/home/runner/go/pkg/mod/k8s.io/apimachinery@v0.24.3/pkg/util/wait/wait.go:220\nk8s.io/apimachinery/pkg/util/wait.runConditionWithCrashProtectionWithContext\n\t/home/runner/go/pkg/mod/k8s.io/apimachinery@v0.24.3/pkg/util/wait/wait.go:233\nk8s.io/apimachinery/pkg/util/wait.runConditionWithCrashProtection\n\t/home/runner/go/pkg/mod/k8s.io/apimachinery@v0.24.3/pkg/util/wait/wait.go:226\nk8s.io/apimachinery/pkg/util/wait.ExponentialBackoff\n\t/home/runner/go/pkg/mod/k8s.io/apimachinery@v0.24.3/pkg/util/wait/wait.go:421\ngithub.com/argoproj/argo-events/common.DoWithRetry\n\t/home/runner/work/argo-events/argo-events/common/retry.go:105\ngithub.com/argoproj/argo-events/eventsources.(*EventSourceAdaptor).run.func3\n\t/home/runner/work/argo-events/argo-events/eventsources/eventing.go:502"}

临时可行但不安全的配置 #

添加以下TLS配置后连接正常，但跳过了证书验证：

tls:
  insecureSkipVerify: true

安全解决方案 #

问题根源是Eventsource客户端无法信任Kafka broker的TLS证书，需通过指定CA证书让客户端验证证书有效性，步骤如下：

1. 创建包含Kafka CA证书的Secret
   将Kafka集群的CA证书保存为文件（如kafka-ca.crt），然后在对应Namespace创建Secret：

kubectl create secret generic kafka-ca-cert --from-file=ca.crt=./kafka-ca.crt -n MY_NAMESPACE

2. 修改EventSource配置，添加CA证书引用
   去掉insecureSkipVerify: true，替换为CA证书的Secret配置：

tls:
  caSecret:
    name: kafka-ca-cert
    key: ca.crt

修改后的完整EventSource配置 #

apiVersion: argoproj.io/v1alpha1
kind: EventSource
metadata:
  name: kafka
  namespace: MY_NAMESPACE
spec:
  eventBusName: MY_EVENTBUS
  kafka:
    kafka-event:
      url: MY_BROKER_URL
      topic: MY_TOPIC
      jsonBody: true
      connectionBackoff:
        duration: 10s
        steps: 5
        factor: 2
        jitter: 0.2
      consumerGroup:
        groupName: argo-test-group
        oldest: false
        rebalanceStrategy: range
        limitEventsPerSecond: 1
        version: "2.5.0"
      sasl:
        mechanism: SCRAM-SHA-512
        passwordSecret:
          key: password
          name: kafka-sasl-creds
        userSecret:
          key: username
          name: kafka-sasl-creds
      tls:
        caSecret:
          name: kafka-ca-cert
          key: ca.crt

该配置通过指定可信CA证书，让Eventsource客户端能正常验证Kafka broker的TLS证书，同时保持SASL_SSL认证的安全性。

内容的提问来源于stack exchange，提问作者Aman