feat(kafka): add support for custom certificates (#3321)

Hello there!

In our company we use custom certificates to authenticate to our Kafka
instances. Unfortunately PeerDB doesn't support uploading custom
certificates for Kafka peers.

This PR adds the possibility. The PR reuses the same logic as used for
Clickhouse peers.

Thank you for your work, review, feedback & eventual merge ❤️🙏

<img width="1320" height="1338" alt="image"
src="https://github.com/user-attachments/assets/9c422c5f-8f75-4352-917e-74c4d0dc2390"
/>

Author: JanPetr

flow/connectors/kafka/kafka.go

@@ -3,6 +3,7 @@ package connkafka
 import (
 	"context"
 	"crypto/tls"
+	"crypto/x509"
 	"errors"
 	"fmt"
 	"log/slog"
@@ -75,7 +76,26 @@ func NewKafkaConnector(
 		kgo.WithLogger(kgoLogger(logger)),
 	)
 	if !config.DisableTls {
-		optionalOpts = append(optionalOpts, kgo.DialTLSConfig(&tls.Config{MinVersion: tls.VersionTLS13}))
+		tlsSetting := &tls.Config{MinVersion: tls.VersionTLS12}
+		if config.Certificate != nil || config.PrivateKey != nil {
+			if config.Certificate == nil || config.PrivateKey == nil {
+				return nil, errors.New("both certificate and private key must be provided if using certificate-based authentication")
+			}
+			cert, err := tls.X509KeyPair([]byte(*config.Certificate), []byte(*config.PrivateKey))
+			if err != nil {
+				return nil, fmt.Errorf("failed to parse provided certificate: %w", err)
+			}
+			tlsSetting.Certificates = []tls.Certificate{cert}
+		}
+		if config.RootCa != nil {
+			caPool := x509.NewCertPool()
+			if !caPool.AppendCertsFromPEM([]byte(*config.RootCa)) {
+				return nil, errors.New("failed to parse provided root CA")
+			}
+			tlsSetting.RootCAs = caPool
+		}
+
+		optionalOpts = append(optionalOpts, kgo.DialTLSConfig(tlsSetting))
 	}
 	switch config.Partitioner {
 	case "LeastBackup":

nexus/analyzer/src/lib.rs

@@ -859,6 +859,9 @@ fn parse_db_options(db_type: DbType, with_options: &[SqlOption]) -> anyhow::Resu
                     .get("disable_tls")
                     .and_then(|s| s.parse::<bool>().ok())
                     .unwrap_or_default(),
+                certificate: opts.get("certificate").map(|s| s.to_string()),
+                private_key: opts.get("private_key").map(|s| s.to_string()),
+                root_ca: opts.get("root_ca").map(|s| s.to_string()),
             };
             Config::KafkaConfig(kafka_config)
         }

protos/peers.proto

@@ -237,6 +237,9 @@ message KafkaConfig {
   string sasl = 4;
   bool disable_tls = 5;
   string partitioner = 6;
+  optional string certificate = 7 [(peerdb_redacted) = true];
+  optional string private_key = 8 [(peerdb_redacted) = true];
+  optional string root_ca = 9 [(peerdb_redacted) = true];
 }
 
 enum ElasticsearchAuthType {

ui/app/peers/create/[peerType]/helpers/ka.ts

@@ -61,6 +61,54 @@ export const kaSetting: PeerSetting[] = [
     tips: 'If you are using a non-TLS connection for Kafka server, check this box.',
     optional: true,
   },
+  {
+    label: 'Certificate',
+    stateHandler: (value, setter) => {
+      if (!value) {
+        // remove key from state if empty
+        setter((curr) => {
+          const newCurr = { ...curr } as KafkaConfig;
+          delete newCurr.certificate;
+          return newCurr;
+        });
+      } else setter((curr) => ({ ...curr, certificate: value as string }));
+    },
+    type: 'file',
+    optional: true,
+    tips: 'This is only needed if the user is authenticated via certificate.',
+  },
+  {
+    label: 'Private Key',
+    stateHandler: (value, setter) => {
+      if (!value) {
+        // remove key from state if empty
+        setter((curr) => {
+          const newCurr = { ...curr } as KafkaConfig;
+          delete newCurr.privateKey;
+          return newCurr;
+        });
+      } else setter((curr) => ({ ...curr, privateKey: value as string }));
+    },
+    type: 'file',
+    optional: true,
+    tips: 'This is only needed if the user is authenticated via certificate.',
+  },
+  {
+    label: 'Root Certificate',
+    stateHandler: (value, setter) => {
+      if (!value) {
+        // remove key from state if empty
+        setter((curr) => {
+          const newCurr = { ...curr } as KafkaConfig;
+          delete newCurr.rootCa;
+          return newCurr;
+        });
+      } else setter((curr) => ({ ...curr, rootCa: value as string }));
+    },
+    type: 'file',
+    optional: true,
+    tips: 'If not provided, host CA roots will be used.',
+  },
 ];
 
 export const blankKafkaSetting: KafkaConfig = {

ui/components/PeerForms/KafkaConfig.tsx

@@ -9,6 +9,7 @@ import { Switch } from '@/lib/Switch/Switch';
 import { TextField } from '@/lib/TextField';
 import { Tooltip } from '@/lib/Tooltip';
 import ReactSelect from 'react-select';
+import { handleFieldChange } from "@/components/PeerForms/common";
 
 interface KafkaProps {
   setter: PeerSetter;
@@ -99,7 +100,7 @@ export default function KafkaForm({ setter }: KafkaProps) {
                   type={setting.type}
                   defaultValue={setting.default}
                   onChange={(e: React.ChangeEvent<HTMLInputElement>) =>
-                    setting.stateHandler(e.target.value, setter)
+                    handleFieldChange(e, setting, setter)
                   }
                 />
                 {setting.tips && (