SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2761: Support non-CLASSPATH based certificates

Browse Source 
If kafka truststore and keystore locations are not local files, then they
are converted to org.springframework.core.io.Resource resources, then copied
to local file system. This means that, paths can be defined as HTTP resources too.
Currently, the Kafka binder only supports CLASSPATH based resources. It would be
useful if we can support non-CLASSPATH like resources such as HTTP, so that if an
application uses config server for example to store certificates, then those will
be copied from it's HTTP endpoint to the local filesystem as Resources.

Checkstyle, documentation fixes.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2761
This commit is contained in:
Gyurian Norbert
2023-07-14 15:06:46 +02:00
committed by Soby Chacko
parent 3c2b91291c
commit 0949d4e145
3 changed files with 88 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java
View File

@@ -62,6 +62,7 @@ import org.springframework.util.StringUtils;
* @author Lukasz Kaminski
* @author Chukwubuikem Ume-Ugwa
* @author Nico Heller
* @author Norbert Gyurian
*/
public class KafkaBinderConfigurationProperties {
@@ -131,8 +132,9 @@ public class KafkaBinderConfigurationProperties {
private Duration authorizationExceptionRetryInterval;
/**
* When a certificate store location is given as classpath URL (classpath:), then the binder
* moves the resource from the classpath location inside the JAR to a location on
* When a certificate store location is not given as a local file system path, then the binder
* converts the path to {@link org.springframework.core.io.Resource} resource, then copies
* the resource from the original location to a location on
* the filesystem. If this value is set, then this location is used, otherwise, the
* certificate file is copied to the directory returned by java.io.tmpdir.
*/
@@ -172,9 +174,10 @@ public class KafkaBinderConfigurationProperties {
}
public String getKafkaConnectionString() {
// We need to do a check on certificate file locations to see if they are given as classpath resources.
// If that is the case, then we will move them to a file system location and use those as the certificate locations.
// This is due to a limitation in Kafka itself in which it doesn't allow reading certificate resources from the classpath.
// We need to do a check on certificate file locations to see if they are given as non-local file system resources.
// If that is the case, then we will copy them to a file system location and use those as the certificate locations.
// This is due to a limitation in Kafka itself in which it doesn't allow reading certificate resources from non-local
// file system locations e.g. classpath, http.
// See this: https://issues.apache.org/jira/browse/KAFKA-7685
// and this: https://cwiki.apache.org/confluence/display/KAFKA/KIP-398%3A+Support+reading+trust+store+from+classpath
moveCertsToFileSystemIfNecessary();
@@ -184,48 +187,43 @@ public class KafkaBinderConfigurationProperties {
private void moveCertsToFileSystemIfNecessary() {
try {
moveBrokerCertsIfApplicable();
moveSchemaRegistryCertsIfApplicable();
// broker certificates
moveCertsIfApplicable("ssl.truststore.location");
moveCertsIfApplicable("ssl.keystore.location");
// schema registry certificates
moveCertsIfApplicable("schema.registry.ssl.truststore.location");
moveCertsIfApplicable("schema.registry.ssl.keystore.location");
}
catch (Exception e) {
throw new IllegalStateException(e);
}
}
private void moveBrokerCertsIfApplicable() throws IOException {
final String trustStoreLocation = this.configuration.get("ssl.truststore.location");
if (trustStoreLocation != null && trustStoreLocation.startsWith("classpath:")) {
final String fileSystemLocation = moveCertToFileSystem(trustStoreLocation, this.certificateStoreDirectory);
private void moveCertsIfApplicable(String storeProperty) throws IOException {
final String storeLocation = this.configuration.get(storeProperty);
// If the path is not defined, or it is a local file path do not move the file
if (storeLocation != null && !checkIfFileExists(storeLocation)) {
final String fileSystemLocation = moveCertToFileSystem(storeLocation, this.certificateStoreDirectory);
// Overriding the value with absolute filesystem path.
this.configuration.put("ssl.truststore.location", fileSystemLocation);
}
final String keyStoreLocation = this.configuration.get("ssl.keystore.location");
if (keyStoreLocation != null && keyStoreLocation.startsWith("classpath:")) {
final String fileSystemLocation = moveCertToFileSystem(keyStoreLocation, this.certificateStoreDirectory);
// Overriding the value with absolute filesystem path.
this.configuration.put("ssl.keystore.location", fileSystemLocation);
this.configuration.put(storeProperty, fileSystemLocation);
}
}
private void moveSchemaRegistryCertsIfApplicable() throws IOException {
String trustStoreLocation = this.configuration.get("schema.registry.ssl.truststore.location");
if (trustStoreLocation != null && trustStoreLocation.startsWith("classpath:")) {
final String fileSystemLocation = moveCertToFileSystem(trustStoreLocation, this.certificateStoreDirectory);
// Overriding the value with absolute filesystem path.
this.configuration.put("schema.registry.ssl.truststore.location", fileSystemLocation);
private boolean checkIfFileExists(String path) {
try {
return Files.isRegularFile(Paths.get(path));
}
final String keyStoreLocation = this.configuration.get("schema.registry.ssl.keystore.location");
if (keyStoreLocation != null && keyStoreLocation.startsWith("classpath:")) {
final String fileSystemLocation = moveCertToFileSystem(keyStoreLocation, this.certificateStoreDirectory);
// Overriding the value with absolute filesystem path.
this.configuration.put("schema.registry.ssl.keystore.location", fileSystemLocation);
catch (Exception e) {
return false;
}
}
private String moveCertToFileSystem(String classpathLocation, String fileSystemLocation) throws IOException {
private String moveCertToFileSystem(String resourceLocation, String fileSystemLocation) throws IOException {
File targetFile;
final String tempDir = System.getProperty("java.io.tmpdir");
Resource resource = new DefaultResourceLoader().getResource(classpathLocation);
Resource resource = new DefaultResourceLoader().getResource(resourceLocation);
if (StringUtils.hasText(fileSystemLocation)) {
final Path path = Paths.get(fileSystemLocation);
if (!Files.exists(path) || !Files.isDirectory(path) || !Files.isWritable(path)) {

54
binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java
View File

@@ -16,15 +16,20 @@
package org.springframework.cloud.stream.binder.kafka.properties;
import java.io.IOException;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.nio.file.Paths;
import java.util.Collections;
import java.util.Map;
import com.sun.net.httpserver.HttpServer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.assertj.core.util.Files;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.core.io.ClassPathResource;
import static org.assertj.core.api.Assertions.assertThat;
@@ -123,6 +128,39 @@ public class KafkaBinderConfigurationPropertiesTest {
.isEqualTo(Paths.get(System.getProperty("java.io.tmpdir"), "testclient.truststore").toString());
assertThat(configuration.get("ssl.keystore.location"))
.isEqualTo(Paths.get(System.getProperty("java.io.tmpdir"), "testclient.keystore").toString());
deleteTempCertFiles();
}
@Test
public void testCertificateFilesAreConvertedToAbsolutePathsFromHttpResources() throws IOException {
HttpServer server = HttpServer.create(new InetSocketAddress("localhost", 5869), 0);
createContextWithCertFileHandler(server, "testclient.truststore");
createContextWithCertFileHandler(server, "testclient.keystore");
server.setExecutor(null); // creates a default executor
server.start();
KafkaProperties kafkaProperties = new KafkaProperties();
KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties =
new KafkaBinderConfigurationProperties(kafkaProperties);
final Map<String, String> configuration = kafkaBinderConfigurationProperties.getConfiguration();
configuration.put("ssl.truststore.location", "http://localhost:5869/testclient.truststore");
configuration.put("ssl.keystore.location", "http://localhost:5869/testclient.keystore");
configuration.put("schema.registry.ssl.truststore.location", "http://localhost:5869/testclient.truststore");
configuration.put("schema.registry.ssl.keystore.location", "http://localhost:5869/testclient.keystore");
kafkaBinderConfigurationProperties.getKafkaConnectionString();
assertThat(configuration.get("ssl.truststore.location"))
.isEqualTo(Paths.get(System.getProperty("java.io.tmpdir"), "testclient.truststore").toString());
assertThat(configuration.get("ssl.keystore.location"))
.isEqualTo(Paths.get(System.getProperty("java.io.tmpdir"), "testclient.keystore").toString());
assertThat(configuration.get("schema.registry.ssl.truststore.location"))
.isEqualTo(Paths.get(System.getProperty("java.io.tmpdir"), "testclient.truststore").toString());
assertThat(configuration.get("schema.registry.ssl.keystore.location"))
.isEqualTo(Paths.get(System.getProperty("java.io.tmpdir"), "testclient.keystore").toString());
deleteTempCertFiles();
server.stop(0);
}
@Test
@@ -160,4 +198,20 @@ public class KafkaBinderConfigurationPropertiesTest {
assertThat(configuration.get("schema.registry.ssl.keystore.location")).isEqualTo(
Paths.get(Files.currentFolder().toString(), "target", "testclient.keystore").toString());
}
private void createContextWithCertFileHandler(HttpServer server, String path) {
server.createContext("/" + path, exchange -> {
ClassPathResource ts = new ClassPathResource(path);
byte[] response = ts.getContentAsByteArray();
exchange.sendResponseHeaders(200, response.length);
OutputStream os = exchange.getResponseBody();
os.write(response);
os.close();
});
}
private void deleteTempCertFiles() {
Paths.get(System.getProperty("java.io.tmpdir"), "testclient.truststore").toFile().delete();
Paths.get(System.getProperty("java.io.tmpdir"), "testclient.keystore").toFile().delete();
}
}

9
docs/src/main/asciidoc/kafka/kafka_overview.adoc
View File

@@ -151,12 +151,13 @@ Flag to set the binder health as `down`, when any partitions on the topic, regar
Default: `false`.
spring.cloud.stream.kafka.binder.certificateStoreDirectory::
When the truststore or keystore certificate location is given as a classpath URL (`classpath:...`), the binder copies the resource from the classpath location inside the JAR file to a location on the filesystem.
When the truststore or keystore certificate location is given as a non-local file system resource (resources supported by org.springframework.core.io.Resource e.g. CLASSPATH, HTTP, etc.),
the binder copies the resource from the path (which is convertible to org.springframework.core.io.Resource) to a location on the filesystem.
This is true for both broker level certificates (`ssl.truststore.location` and `ssl.keystore.location`) and certificates intended for schema registry (`schema.registry.ssl.truststore.location` and `schema.registry.ssl.keystore.location`).
Keep in mind that the truststore and keystore classpath locations must be provided under `spring.cloud.stream.kafka.binder.configuration...`.
Keep in mind that the truststore and keystore location paths must be provided under `spring.cloud.stream.kafka.binder.configuration...`.
For example, `spring.cloud.stream.kafka.binder.configuration.ssl.truststore.location`, `spring.cloud.stream.kafka.binder.configuration.schema.registry.ssl.truststore.location`, etc.
The file will be moved to the location specified as the value for this property which must be an existing directory on the filesystem that is writable by the process running the application.
If this value is not set and the certificate file is a classpath resource, then it will be moved to System's temp directory as returned by `System.getProperty("java.io.tmpdir")`.
The file will be copied to the location specified as the value for this property which must be an existing directory on the filesystem that is writable by the process running the application.
If this value is not set and the certificate file is a non-local file system resource, then it will be copied to System's temp directory as returned by `System.getProperty("java.io.tmpdir")`.
This is also true, if this value is present, but the directory cannot be found on the filesystem or is not writable.
+
Default: none.