auth with client cert in kafka plugin (#607)

DmitryRomanov · web-flow · commit 6de60c84614d · 2024-03-26T20:26:38.000+07:00
* auth with client cert in kafka plugin

* because you can set content of cert or key

* e2e test for kafka auth by client cert
diff --git a/e2e/kafka_auth/.gitignore b/e2e/kafka_auth/.gitignore
@@ -0,0 +1 @@
+/certs/
diff --git a/e2e/kafka_auth/docker-compose.yml b/e2e/kafka_auth/docker-compose.yml
@@ -1,4 +1,4 @@
-version: "2"
+version: "2.1"
 
 services:
   zookeeper:
@@ -9,31 +9,51 @@ services:
       - "zookeeper_data:/bitnami"
     environment:
       - ALLOW_ANONYMOUS_LOGIN=yes
+  init-certs:
+      image: docker.io/bitnami/kafka:3.6
+      command: /tmp/generate.sh
+      working_dir: /tmp/
+      user: 0:0
+      volumes:
+        - ./certs/:/tmp/certs/
+        - "./generate.sh:/tmp/generate.sh"
   kafka:
     image: docker.io/bitnami/kafka:3.6
+    container_name: kafka
     depends_on:
-      - zookeeper
+      zookeeper:
+        condition: service_started
+      init-certs:
+        condition: service_completed_successfully
     ports:
-      - "9093:9092"
+      - "9093:9093"
     volumes:
       - "kafka_data:/bitnami"
+      - ./certs/kafka.truststore.jks:/bitnami/kafka/config/certs/kafka.truststore.jks
+      - ./certs/kafka.keystore.jks:/bitnami/kafka/config/certs/kafka.keystore.jks
     environment:
       # Zookeeper
       - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
       # Listeners
-      - KAFKA_CFG_LISTENERS=SASL_PLAINTEXT://:9092,PLAINTEXT://:9094
-      - KAFKA_CFG_ADVERTISED_LISTENERS=SASL_PLAINTEXT://localhost:9093,PLAINTEXT://:9094
-      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT
+      - KAFKA_CFG_LISTENERS=SASL_SSL://:9093,SASL_PLAINTEXT://:9095,PLAINTEXT://:9094
+      - KAFKA_CFG_ADVERTISED_LISTENERS=SASL_SSL://localhost:9093,SASL_PLAINTEXT://localhost:9095,PLAINTEXT://:9094
+      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL,SSL:SSL
       # Inter broker
       - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=PLAINTEXT
       - KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL=PLAIN
       # Security
       - KAFKA_CFG_SASL_ENABLED_MECHANISMS=PLAIN,SCRAM-SHA-256,SCRAM-SHA-512
       - KAFKA_CLIENT_USERS=user,admin
       - KAFKA_CLIENT_PASSWORDS=pass,admin
+      - KAFKA_CFG_SSL_KEYSTORE_LOCATION=/opt/bitnami/kafka/config/certs/kafka.keystore.jks
+      - KAFKA_CFG_SSL_KEYSTORE_PASSWORD=supersecret
+      - KAFKA_CFG_SSL_KEY_PASSWORD=supersecret
+      - KAFKA_CFG_SSL_TRUSTSTORE_LOCATION=/opt/bitnami/kafka/config/certs/kafka.truststore.jks
+      - KAFKA_CFG_SSL_TRUSTSTORE_PASSWORD=supersecret
+      - KAFKA_CFG_SSL_CLIENT_AUTH=required
 
 volumes:
   zookeeper_data:
     driver: local
   kafka_data:
-    driver: local
+    driver: local
diff --git a/e2e/kafka_auth/generate.sh b/e2e/kafka_auth/generate.sh
@@ -0,0 +1,53 @@
+#!/usr/bin/env bash
+ 
+set -eu
+
+PASSWORD="${PASSWORD:-supersecret}"
+
+rm -rf certs/*
+# Root CA
+echo "Creating CA certificate and key"
+openssl req -new -x509 -keyout certs/ca.key -out certs/ca.crt -days 365 -subj "/CN=Sample CA/OU=US/O=US/ST=US/C=US" -passout pass:$PASSWORD
+
+echo "Creating Truststore"
+keytool -keystore certs/kafka.truststore.jks -alias CARoot -import -file certs/ca.crt -storepass $PASSWORD -keypass $PASSWORD -noprompt
+
+# Node certs
+
+echo "Creating node key"
+keytool -keystore certs/kafka.keystore.jks -alias kafka -validity 365 -genkey -keyalg RSA -dname "cn=kafka, ou=US, o=US, c=US" -storepass $PASSWORD -keypass $PASSWORD
+echo "Creating certificate sign request"
+keytool -keystore certs/kafka.keystore.jks -alias kafka -certreq -file certs/tls.srl -storepass $PASSWORD -keypass $PASSWORD
+echo "Signing certificate request using self-signed CA"
+openssl x509 -req -CA certs/ca.crt -CAkey certs/ca.key \
+    -in certs/tls.srl -out certs/tls.crt \
+    -days 365 -CAcreateserial \
+    -passin pass:$PASSWORD
+echo "Adding Ca certificate to the keystore"
+keytool -keystore certs/kafka.keystore.jks -alias CARoot -import -file certs/ca.crt -storepass $PASSWORD -keypass $PASSWORD -noprompt
+echo "Adding signed certificate"
+keytool -keystore certs/kafka.keystore.jks -alias kafka -import -file certs/tls.crt -storepass $PASSWORD -keypass $PASSWORD -noprompt
+
+
+
+# Client cert
+
+echo "Creating client key"
+keytool -keystore certs/kafka.keystore.jks -alias client -validity 365 -genkey -keyalg RSA -dname "cn=client, ou=US, o=US, c=US" -storepass $PASSWORD -keypass $PASSWORD
+
+echo "Creating certificate sign request"
+keytool -keystore certs/kafka.keystore.jks -alias client -certreq -file certs/client.srl -storepass $PASSWORD -keypass $PASSWORD
+
+
+echo $PASSWORD | keytool -importkeystore \
+    -srckeystore certs/kafka.keystore.jks \
+    -destkeystore certs/kafka.keystore.p12 \
+    -deststoretype PKCS12 \
+    -srcalias client \
+    -deststorepass $PASSWORD \
+    -destkeypass $PASSWORD
+
+openssl pkcs12 -in certs/kafka.keystore.p12 -nokeys -out certs/client_cert.pem -passin pass:$PASSWORD
+openssl pkcs12 -in certs/kafka.keystore.p12 -nodes -nocerts -out certs/client_key.pem -passin pass:$PASSWORD
+
+chmod 644 certs/client*
diff --git a/e2e/kafka_auth/kafka_auth.go b/e2e/kafka_auth/kafka_auth.go
@@ -24,10 +24,11 @@ type Config struct {
 
 func (c *Config) Configure(t *testing.T, _ *cfg.Config, _ string) {
 	type saslData struct {
-		Enabled   bool
-		Mechanism string
-		Username  string
-		Password  string
+		Enabled          bool
+		Mechanism        string
+		Username         string
+		Password         string
+		AuthByClientCert bool
 	}
 
 	type tCase struct {
@@ -38,28 +39,31 @@ func (c *Config) Configure(t *testing.T, _ *cfg.Config, _ string) {
 	cases := []tCase{
 		{
 			sasl: saslData{
-				Enabled:   true,
-				Mechanism: "PLAIN",
-				Username:  "user",
-				Password:  "pass",
+				Enabled:          true,
+				Mechanism:        "PLAIN",
+				Username:         "user",
+				Password:         "pass",
+				AuthByClientCert: true,
 			},
 			authorized: true,
 		},
 		{
 			sasl: saslData{
-				Enabled:   true,
-				Mechanism: "SCRAM-SHA-256",
-				Username:  "user",
-				Password:  "pass",
+				Enabled:          true,
+				Mechanism:        "SCRAM-SHA-256",
+				Username:         "user",
+				Password:         "pass",
+				AuthByClientCert: true,
 			},
 			authorized: true,
 		},
 		{
 			sasl: saslData{
-				Enabled:   true,
-				Mechanism: "SCRAM-SHA-512",
-				Username:  "user",
-				Password:  "pass",
+				Enabled:          true,
+				Mechanism:        "SCRAM-SHA-512",
+				Username:         "user",
+				Password:         "pass",
+				AuthByClientCert: true,
 			},
 			authorized: true,
 		},
@@ -71,10 +75,11 @@ func (c *Config) Configure(t *testing.T, _ *cfg.Config, _ string) {
 		},
 		{
 			sasl: saslData{
-				Enabled:   true,
-				Mechanism: "PLAIN",
-				Username:  "user",
-				Password:  "pass123",
+				Enabled:          true,
+				Mechanism:        "PLAIN",
+				Username:         "user",
+				Password:         "pass123",
+				AuthByClientCert: false,
 			},
 			authorized: false,
 		},
@@ -91,6 +96,8 @@ func (c *Config) Configure(t *testing.T, _ *cfg.Config, _ string) {
 					ClientID:         "test-auth-out",
 					BatchSize_:       10,
 					MaxMessageBytes_: 1000000,
+					SslEnabled:       true,
+					SslSkipVerify:    true,
 				}
 				if tt.sasl.Enabled {
 					config.SaslEnabled = true
@@ -99,6 +106,11 @@ func (c *Config) Configure(t *testing.T, _ *cfg.Config, _ string) {
 					config.SaslPassword = tt.sasl.Password
 				}
 
+				if tt.sasl.AuthByClientCert {
+					config.ClientKey = "./kafka_auth/certs/client_key.pem"
+					config.ClientCert = "./kafka_auth/certs/client_cert.pem"
+				}
+
 				kafka_out.NewProducer(config,
 					zap.NewNop().WithOptions(zap.WithFatalHook(zapcore.WriteThenPanic)).Sugar(),
 				)
@@ -113,6 +125,8 @@ func (c *Config) Configure(t *testing.T, _ *cfg.Config, _ string) {
 					Offset_:                    kafka_in.OffsetTypeNewest,
 					ConsumerMaxProcessingTime_: 200 * time.Millisecond,
 					ConsumerMaxWaitTime_:       250 * time.Millisecond,
+					SslEnabled:                 true,
+					SslSkipVerify:              true,
 				}
 				if tt.sasl.Enabled {
 					config.SaslEnabled = true
@@ -121,6 +135,11 @@ func (c *Config) Configure(t *testing.T, _ *cfg.Config, _ string) {
 					config.SaslPassword = tt.sasl.Password
 				}
 
+				if tt.sasl.AuthByClientCert {
+					config.ClientKey = "./kafka_auth/certs/client_key.pem"
+					config.ClientCert = "./kafka_auth/certs/client_cert.pem"
+				}
+
 				kafka_in.NewConsumerGroup(config,
 					zap.NewNop().WithOptions(zap.WithFatalHook(zapcore.WriteThenPanic)).Sugar(),
 				)
diff --git a/plugin/input/kafka/README.md b/plugin/input/kafka/README.md
@@ -111,7 +111,19 @@ If set, the plugin will skip SSL/TLS verification.
 
 <br>
 
-**`pem_file`** *`string`* *`default=/file.d/certs`* 
+**`client_cert`** *`string`* 
+
+Path or content of a PEM-encoded client certificate file.
+
+<br>
+
+**`client_key`** *`string`* 
+
+> Path or content of a PEM-encoded client key file.
+
+<br>
+
+**`ca_cert`** *`string`* 
 
 Path or content of a PEM-encoded CA file.
 
diff --git a/plugin/input/kafka/kafka.go b/plugin/input/kafka/kafka.go
@@ -142,10 +142,20 @@ type Config struct {
 	// > If set, the plugin will skip SSL/TLS verification.
 	SslSkipVerify bool `json:"ssl_skip_verify" default:"false"` // *
 
+	// > @3@4@5@6
+	// >
+	// > Path or content of a PEM-encoded client certificate file.
+	ClientCert string `json:"client_cert"` // *
+
+	// > @3@4@5@6
+	// >
+	// > > Path or content of a PEM-encoded client key file.
+	ClientKey string `json:"client_key"` // *
+
 	// > @3@4@5@6
 	// >
 	// > Path or content of a PEM-encoded CA file.
-	SslPem string `json:"pem_file" default:"/file.d/certs"` // *
+	CACert string `json:"ca_cert"` // *
 }
 
 func init() {
@@ -239,11 +249,20 @@ func NewConsumerGroup(c *Config, l *zap.SugaredLogger) sarama.ConsumerGroup {
 		config.Net.TLS.Enable = true
 
 		tlsCfg := xtls.NewConfigBuilder()
-		if err := tlsCfg.AppendCARoot(c.SslPem); err != nil {
-			l.Fatalf("can't load cert: %s", err.Error())
+
+		if c.CACert != "" {
+			if err := tlsCfg.AppendCARoot(c.CACert); err != nil {
+				l.Fatalf("can't load ca cert: %s", err.Error())
+			}
 		}
 		tlsCfg.SetSkipVerify(c.SslSkipVerify)
 
+		if c.ClientCert != "" || c.ClientKey != "" {
+			if err := tlsCfg.AppendX509KeyPair(c.ClientCert, c.ClientKey); err != nil {
+				l.Fatalf("can't load client certificate and key: %s", err.Error())
+			}
+		}
+
 		config.Net.TLS.Config = tlsCfg.Build()
 	}
 
diff --git a/plugin/output/kafka/README.md b/plugin/output/kafka/README.md
@@ -126,7 +126,19 @@ If set, the plugin will skip SSL/TLS verification.
 
 <br>
 
-**`pem_file`** *`string`* *`default=/file.d/certs`* 
+**`client_cert`** *`string`* 
+
+Path or content of a PEM-encoded client certificate file.
+
+<br>
+
+**`client_key`** *`string`* 
+
+> Path or content of a PEM-encoded client key file.
+
+<br>
+
+**`ca_cert`** *`string`* 
 
 Path or content of a PEM-encoded CA file.
 
diff --git a/plugin/output/kafka/kafka.go b/plugin/output/kafka/kafka.go
@@ -156,10 +156,20 @@ type Config struct {
 	// > If set, the plugin will skip SSL/TLS verification.
 	SslSkipVerify bool `json:"ssl_skip_verify" default:"false"` // *
 
+	// > @3@4@5@6
+	// >
+	// > Path or content of a PEM-encoded client certificate file.
+	ClientCert string `json:"client_cert"` // *
+
+	// > @3@4@5@6
+	// >
+	// > > Path or content of a PEM-encoded client key file.
+	ClientKey string `json:"client_key"` // *
+
 	// > @3@4@5@6
 	// >
 	// > Path or content of a PEM-encoded CA file.
-	SslPem string `json:"pem_file" default:"/file.d/certs"` // *
+	CACert string `json:"ca_cert"` // *
 }
 
 func init() {
@@ -320,11 +330,19 @@ func NewProducer(c *Config, l *zap.SugaredLogger) sarama.SyncProducer {
 		config.Net.TLS.Enable = true
 
 		tlsCfg := xtls.NewConfigBuilder()
-		if err := tlsCfg.AppendCARoot(c.SslPem); err != nil {
-			l.Fatalf("can't load cert: %s", err.Error())
+		if c.CACert != "" {
+			if err := tlsCfg.AppendCARoot(c.CACert); err != nil {
+				l.Fatalf("can't load ca cert: %s", err.Error())
+			}
 		}
 		tlsCfg.SetSkipVerify(c.SslSkipVerify)
 
+		if c.ClientCert != "" || c.ClientKey != "" {
+			if err := tlsCfg.AppendX509KeyPair(c.ClientCert, c.ClientKey); err != nil {
+				l.Fatalf("can't load client certificate and key: %s", err.Error())
+			}
+		}
+
 		config.Net.TLS.Config = tlsCfg.Build()
 	}
 
