feat: Support Customer TLS certificates (#20879)

bbernays · web-flow · commit 208ef9183b03 · 2025-06-09T07:59:39.000Z
diff --git a/plugins/destination/kafka/client/client.go b/plugins/destination/kafka/client/client.go
@@ -3,8 +3,10 @@ package client
 import (
 	"context"
 	"crypto/tls"
+	"crypto/x509"
 	"encoding/json"
 	"fmt"
+	"os"
 	"time"
 
 	"github.com/IBM/sarama"
@@ -27,6 +29,29 @@ type Client struct {
 	*filetypes.Client
 }
 
+func createTLSConfiguration(userSpec *spec.Spec) (*tls.Config, error) {
+	t := &tls.Config{
+		InsecureSkipVerify: !userSpec.EnforceTLSVerification,
+	}
+	if userSpec.TlsDetails != nil && *userSpec.TlsDetails.CertFile != "" && *userSpec.TlsDetails.KeyFile != "" && *userSpec.TlsDetails.CaFile != "" {
+		cert, err := tls.LoadX509KeyPair(*userSpec.TlsDetails.CertFile, *userSpec.TlsDetails.KeyFile)
+		if err != nil {
+			return nil, fmt.Errorf("failed to load X509 key pair: %w", err)
+		}
+
+		caCert, err := os.ReadFile(*userSpec.TlsDetails.CaFile)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read CA file: %w", err)
+		}
+
+		caCertPool := x509.NewCertPool()
+		caCertPool.AppendCertsFromPEM(caCert)
+		t.Certificates = []tls.Certificate{cert}
+		t.RootCAs = caCertPool
+	}
+	return t, nil
+}
+
 func New(_ context.Context, logger zerolog.Logger, s []byte, opts plugin.NewClientOptions) (plugin.Client, error) {
 	c := &Client{
 		logger: logger.With().Str("module", "dest-kafka").Logger(),
@@ -58,11 +83,16 @@ func New(_ context.Context, logger zerolog.Logger, s []byte, opts plugin.NewClie
 	c.conf.ClientID = `cwc|1c04a227-aef8-47a9-9353-e20bbb6a9616|cq-destination-kafka|` + internalPlugin.Version
 
 	if c.spec.SASLUsername != "" {
+		tlsConfig, err := createTLSConfiguration(c.spec)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create TLS configuration: %w", err)
+		}
+
 		c.conf.Net.SASL.Enable = true
 		c.conf.Net.SASL.User = c.spec.SASLUsername
 		c.conf.Net.SASL.Password = c.spec.SASLPassword
 		c.conf.Net.TLS.Enable = true
-		c.conf.Net.TLS.Config = &tls.Config{InsecureSkipVerify: !c.spec.EnforceTLSVerification}
+		c.conf.Net.TLS.Config = tlsConfig
 		c.conf.Net.SASL.Handshake = true
 	}
 
diff --git a/plugins/destination/kafka/client/spec/schema.json b/plugins/destination/kafka/client/spec/schema.json
diff --git a/plugins/destination/kafka/client/spec/spec.go b/plugins/destination/kafka/client/spec/spec.go
@@ -13,6 +13,15 @@ type topicDetails struct {
 	ReplicationFactor int `json:"replication_factor,omitempty" jsonschema:"minimum=1,default=1"`
 }
 
+type TlsDetails struct {
+	// Path to the certificate file for client authentication
+	CertFile *string `json:"cert_file,omitempty"`
+	// Path to the key file for client authentication
+	KeyFile *string `json:"key_file,omitempty"`
+	// Path to the certificate authority file for TLS client authentication
+	CaFile *string `json:"ca_file,omitempty"`
+}
+
 type Spec struct {
 	filetypes.FileSpec
 
@@ -35,6 +44,9 @@ type Spec struct {
 	// Enforce TLS Verification when configuring a username to connect to Kafka.
 	EnforceTLSVerification bool `json:"enforce_tls_verification,omitempty"`
 
+	// TLS details for client authentication.
+	TlsDetails *TlsDetails `json:"tls_details,omitempty"`
+
 	// Number of records to write before starting a new object.
 	BatchSize int64 `json:"batch_size" jsonschema:"minimum=1,default=1000"`
 
diff --git a/plugins/destination/kafka/docs/overview.md b/plugins/destination/kafka/docs/overview.md
@@ -61,6 +61,10 @@ This is the (nested) plugin spec
 
   Optional parameters to set topic details.
 
+- `tls_details` ([tls_details](#tls_details)) (optional)
+
+  Optional parameters to set TLS details for the Kafka connection.
+
 
 ### format_spec
 
@@ -107,3 +111,19 @@ Reserved for future use.
 
   Replication factor for the topic.
 
+
+
+### tls_details
+- `ca_file_path` (`string`) (optional) (default: empty)
+
+   Path to the certificate authority file for TLS client authentication
+
+- `cert_file_path` (`string`) (optional) (default: empty)
+
+  Path to the certificate file for client authentication
+
+- `key_file_path` (`string`) (optional) (default: empty)
+
+  Path to the key file for client authentication.
+
+
