From 7a98364a236b47e18113252ab4294eb5253a10a1 Mon Sep 17 00:00:00 2001
From: Tim te Beek <tim.te.beek@jdriven.com>
Date: Thu, 23 Sep 2021 20:14:52 +0200
Subject: [PATCH] Support PEM format for Kafka SSL certs and private key

See gh-28123
---
 .../autoconfigure/kafka/KafkaProperties.java  | 54 ++++++++++++++++++-
 1 file changed, 53 insertions(+), 1 deletion(-)

diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java
index e76dd85de2..2a4c29626f 100644
--- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java
+++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java
@@ -35,6 +35,7 @@ import org.apache.kafka.common.serialization.StringSerializer;
 
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.PropertyMapper;
+import org.springframework.boot.context.properties.source.MutuallyExclusiveConfigurationPropertiesException;
 import org.springframework.boot.convert.DurationUnit;
 import org.springframework.core.io.Resource;
 import org.springframework.kafka.listener.ContainerProperties.AckMode;
@@ -1042,10 +1043,20 @@ public class KafkaProperties {
 	public static class Ssl {
 
 		/**
-		 * Password of the private key in the key store file.
+		 * Password of the private key in either key store key or key store file.
 		 */
 		private String keyPassword;
 
+		/**
+		 * Certificate chain in PEM format with a list of X.509 certificates.
+		 */
+		private String keyStoreCertificateChain;
+
+		/**
+		 * Private key in PEM format with PKCS#8 keys.
+		 */
+		private String keyStoreKey;
+
 		/**
 		 * Location of the key store file.
 		 */
@@ -1061,6 +1072,11 @@ public class KafkaProperties {
 		 */
 		private String keyStoreType;
 
+		/**
+		 * Trusted certificates in PEM format with X.509 certificates.
+		 */
+		private String trustStoreCertificates;
+
 		/**
 		 * Location of the trust store file.
 		 */
@@ -1089,6 +1105,22 @@ public class KafkaProperties {
 			this.keyPassword = keyPassword;
 		}
 
+		public String getKeyStoreCertificateChain() {
+			return this.keyStoreCertificateChain;
+		}
+
+		public void setKeyStoreCertificateChain(String keyStoreCertificateChain) {
+			this.keyStoreCertificateChain = keyStoreCertificateChain;
+		}
+
+		public String getKeyStoreKey() {
+			return this.keyStoreKey;
+		}
+
+		public void setKeyStoreKey(String keyStoreKey) {
+			this.keyStoreKey = keyStoreKey;
+		}
+
 		public Resource getKeyStoreLocation() {
 			return this.keyStoreLocation;
 		}
@@ -1113,6 +1145,14 @@ public class KafkaProperties {
 			this.keyStoreType = keyStoreType;
 		}
 
+		public String getTrustStoreCertificates() {
+			return this.trustStoreCertificates;
+		}
+
+		public void setTrustStoreCertificates(String trustStoreCertificates) {
+			this.trustStoreCertificates = trustStoreCertificates;
+		}
+
 		public Resource getTrustStoreLocation() {
 			return this.trustStoreLocation;
 		}
@@ -1146,13 +1186,25 @@ public class KafkaProperties {
 		}
 
 		public Map<String, Object> buildProperties() {
+			MutuallyExclusiveConfigurationPropertiesException.throwIfMultipleNonNullValuesIn((entries) -> {
+				entries.put("spring.kafka.ssl.key-store-key", this.getKeyStoreKey());
+				entries.put("spring.kafka.ssl.key-store-location", this.getKeyStoreLocation());
+			});
+			MutuallyExclusiveConfigurationPropertiesException.throwIfMultipleNonNullValuesIn((entries) -> {
+				entries.put("spring.kafka.ssl.trust-store-certificates", this.getTrustStoreCertificates());
+				entries.put("spring.kafka.ssl.trust-store-location", this.getTrustStoreLocation());
+			});
 			Properties properties = new Properties();
 			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
 			map.from(this::getKeyPassword).to(properties.in(SslConfigs.SSL_KEY_PASSWORD_CONFIG));
+			map.from(this::getKeyStoreCertificateChain)
+					.to(properties.in(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG));
+			map.from(this::getKeyStoreKey).to(properties.in(SslConfigs.SSL_KEYSTORE_KEY_CONFIG));
 			map.from(this::getKeyStoreLocation).as(this::resourceToPath)
 					.to(properties.in(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG));
 			map.from(this::getKeyStorePassword).to(properties.in(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG));
 			map.from(this::getKeyStoreType).to(properties.in(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG));
+			map.from(this::getTrustStoreCertificates).to(properties.in(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG));
 			map.from(this::getTrustStoreLocation).as(this::resourceToPath)
 					.to(properties.in(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG));
 			map.from(this::getTrustStorePassword).to(properties.in(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG));