Merge pull request #28123 from timtebeek
* pr/28123: Polish "Support PEM format for Kafka SSL certs and private key" Support PEM format for Kafka SSL certs and private key Closes gh-28123
This commit is contained in:
@@ -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));
|
||||
|
||||
@@ -16,20 +16,27 @@
|
||||
|
||||
package org.springframework.boot.autoconfigure.kafka;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
import org.apache.kafka.common.config.SslConfigs;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.boot.autoconfigure.kafka.KafkaProperties.Cleanup;
|
||||
import org.springframework.boot.autoconfigure.kafka.KafkaProperties.IsolationLevel;
|
||||
import org.springframework.boot.autoconfigure.kafka.KafkaProperties.Listener;
|
||||
import org.springframework.boot.context.properties.source.MutuallyExclusiveConfigurationPropertiesException;
|
||||
import org.springframework.core.io.ClassPathResource;
|
||||
import org.springframework.kafka.core.CleanupConfig;
|
||||
import org.springframework.kafka.listener.ContainerProperties;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
|
||||
|
||||
/**
|
||||
* Tests for {@link KafkaProperties}.
|
||||
*
|
||||
* @author Stephane Nicoll
|
||||
* @author Madhura Bhave
|
||||
*/
|
||||
class KafkaPropertiesTests {
|
||||
|
||||
@@ -52,6 +59,37 @@ class KafkaPropertiesTests {
|
||||
assertThat(listenerProperties.isMissingTopicsFatal()).isEqualTo(container.isMissingTopicsFatal());
|
||||
}
|
||||
|
||||
@Test
|
||||
void sslPemConfiguration() {
|
||||
KafkaProperties properties = new KafkaProperties();
|
||||
properties.getSsl().setKeyStoreKey("-----BEGINkey");
|
||||
properties.getSsl().setTrustStoreCertificates("-----BEGINtrust");
|
||||
properties.getSsl().setKeyStoreCertificateChain("-----BEGINchain");
|
||||
Map<String, Object> consumerProperties = properties.buildConsumerProperties();
|
||||
assertThat(consumerProperties.get(SslConfigs.SSL_KEYSTORE_KEY_CONFIG)).isEqualTo("-----BEGINkey");
|
||||
assertThat(consumerProperties.get(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG)).isEqualTo("-----BEGINtrust");
|
||||
assertThat(consumerProperties.get(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG))
|
||||
.isEqualTo("-----BEGINchain");
|
||||
}
|
||||
|
||||
@Test
|
||||
void sslPropertiesWhenKeyStoreLocationAndKeySetShouldThrowException() {
|
||||
KafkaProperties properties = new KafkaProperties();
|
||||
properties.getSsl().setKeyStoreKey("-----BEGIN");
|
||||
properties.getSsl().setKeyStoreLocation(new ClassPathResource("ksLoc"));
|
||||
assertThatExceptionOfType(MutuallyExclusiveConfigurationPropertiesException.class)
|
||||
.isThrownBy(properties::buildConsumerProperties);
|
||||
}
|
||||
|
||||
@Test
|
||||
void sslPropertiesWhenTrustStoreLocationAndCertificatesSetShouldThrowException() {
|
||||
KafkaProperties properties = new KafkaProperties();
|
||||
properties.getSsl().setTrustStoreLocation(new ClassPathResource("tsLoc"));
|
||||
properties.getSsl().setTrustStoreCertificates("-----BEGIN");
|
||||
assertThatExceptionOfType(MutuallyExclusiveConfigurationPropertiesException.class)
|
||||
.isThrownBy(properties::buildConsumerProperties);
|
||||
}
|
||||
|
||||
@Test
|
||||
void cleanupConfigDefaultValuesAreConsistent() {
|
||||
CleanupConfig cleanupConfig = new CleanupConfig();
|
||||
|
||||
Reference in New Issue
Block a user