diff --git a/admin/pom.xml b/admin/pom.xml index 1b0e24b..772c3a1 100644 --- a/admin/pom.xml +++ b/admin/pom.xml @@ -25,6 +25,12 @@ io.strimzi.test-clients clients-common + + io.strimzi.test-clients + clients-common + test-jar + test + io.strimzi kafka-kubernetes-config-provider diff --git a/admin/src/main/java/io/strimzi/testclients/admin/AdminProperties.java b/admin/src/main/java/io/strimzi/testclients/admin/AdminProperties.java index d27c034..d10ce1e 100644 --- a/admin/src/main/java/io/strimzi/testclients/admin/AdminProperties.java +++ b/admin/src/main/java/io/strimzi/testclients/admin/AdminProperties.java @@ -44,7 +44,7 @@ public static Properties adminProperties(String bootstrapServer) { String value = String.valueOf(v); if (sslFields.contains(key) && value.startsWith("@")) { - value = ConfigurationUtils.getContentsOfTheFileInConfigFolder(value); + value = ConfigurationUtils.getContentsOfTheFileInConfigFolder(value.substring(1)); } propertiesInMap.put(key, value); diff --git a/admin/src/main/java/io/strimzi/testclients/utils/ConfigurationUtils.java b/admin/src/main/java/io/strimzi/testclients/utils/ConfigurationUtils.java index dfe4e19..819eaa2 100644 --- a/admin/src/main/java/io/strimzi/testclients/utils/ConfigurationUtils.java +++ b/admin/src/main/java/io/strimzi/testclients/utils/ConfigurationUtils.java @@ -13,6 +13,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.nio.file.StandardCopyOption; import java.util.Locale; import java.util.Map; import java.util.Properties; @@ -165,12 +166,12 @@ public static void writeToConfigurationFile(Properties properties) { */ public static void copyFileToConfigurationFolder(String filePath, String fileNameInConfigFolder) { Path sourceFilePath = Paths.get(filePath); - Path targetFilePath = Paths.get(getConfigFilePath() + fileNameInConfigFolder); + Path targetFilePath = Paths.get(getConfigFolderPath() + fileNameInConfigFolder); try { - Files.copy(sourceFilePath, targetFilePath); + Files.copy(sourceFilePath, targetFilePath, StandardCopyOption.REPLACE_EXISTING); } catch (IOException e) { - throw new RuntimeException("Unable to copy file from path: " + filePath + " to the configuration folder due to: ", e.getCause()); + throw new RuntimeException("Unable to copy file from path: " + filePath + " to the configuration folder due to: ", e); } } @@ -180,7 +181,7 @@ public static void copyFileToConfigurationFolder(String filePath, String fileNam * @return file contents in String */ public static String getContentsOfTheFileInConfigFolder(String fileName) { - Path filePath = Paths.get(getConfigFilePath() + fileName); + Path filePath = Paths.get(getConfigFolderPath() + fileName); try { return String.join("\n", Files.readAllLines(filePath)); @@ -189,7 +190,11 @@ public static String getContentsOfTheFileInConfigFolder(String fileName) { } } - private static String getConfigFolderPath() { + /** + * Path of the configuration folder, holding config.properties and the certificate files copied by `configure ssl` + * @return path to the configuration folder, ending with "/" + */ + public static String getConfigFolderPath() { String specificFolderPath = System.getenv(ConfigurationConstants.CONFIG_FOLDER_PATH_ENV); return specificFolderPath == null || specificFolderPath.isEmpty() ? CONFIG_FOLDER_PATH_DEFAULT : specificFolderPath + "/" + CONFIG_FOLDER_NAME; } diff --git a/admin/src/test/java/admin/integration/AdminTlsIT.java b/admin/src/test/java/admin/integration/AdminTlsIT.java new file mode 100644 index 0000000..f2058d9 --- /dev/null +++ b/admin/src/test/java/admin/integration/AdminTlsIT.java @@ -0,0 +1,142 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package admin.integration; + +import io.strimzi.test.container.StrimziKafkaCluster; +import io.strimzi.testclients.admin.KafkaAdminClient; +import io.strimzi.testclients.configuration.ConfigurationConstants; +import io.strimzi.testclients.constants.Constants; +import io.strimzi.testclients.testutils.TlsUtils; +import io.strimzi.testclients.utils.ConfigurationUtils; +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.TopicDescription; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.api.io.TempDir; +import picocli.CommandLine; + +import java.io.ByteArrayOutputStream; +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.PrintStream; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Arrays; +import java.util.List; +import java.util.Properties; + +import static io.strimzi.testclients.testutils.TestUtils.waitFor; +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.hasItem; + +/** + * Runs the admin client against a cluster that only exposes an mTLS listener, + * with the certificates handed over through the `configure ssl` subcommand. + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class AdminTlsIT { + private static final List SSL_CONFIG_KEYS = List.of( + ConfigurationConstants.CA_CRT_PROPERTY, + ConfigurationConstants.USER_KEY_PROPERTY, + ConfigurationConstants.USER_CRT_PROPERTY + ); + private static final List SSL_FILE_NAMES = List.of( + Constants.TRUSTSTORE_FILE_NAME, + Constants.KEYSTORE_KEY_FILE_NAME, + Constants.KEYSTORE_CERT_FILE_NAME + ); + + private static StrimziKafkaCluster kafkaCluster; + private static Admin admin; + private static CommandLine cmd; + + @BeforeAll + static void setup(@TempDir Path certDir) throws Exception { + kafkaCluster = new StrimziKafkaCluster.StrimziKafkaClusterBuilder() + .withNumberOfBrokers(1) + .withInternalTopicReplicationFactor(1) + .withSharedNetwork() + .withTls() + .build(); + kafkaCluster.start(); + + Path caCrtFile = Files.writeString(certDir.resolve("ca.crt"), TlsUtils.caCertificate(kafkaCluster)); + Path userCrtFile = Files.writeString(certDir.resolve("user.crt"), TlsUtils.userCertificateChain(kafkaCluster)); + Path userKeyFile = Files.writeString(certDir.resolve("user.key"), TlsUtils.userKey(kafkaCluster)); + + cmd = new CommandLine(new KafkaAdminClient()); + assertThat(cmd.execute("configure", "common", "--bootstrap-server", kafkaCluster.getBootstrapServers()), is(0)); + assertThat(cmd.execute("configure", "ssl", + "--truststore", caCrtFile.toString(), + "--keystore-cert", userCrtFile.toString(), + "--keystore-key", userKeyFile.toString()), is(0)); + + // Independent TLS client used only to check what the admin client did + admin = Admin.create(TlsUtils.kafkaClientConfig(kafkaCluster)); + } + + @AfterAll + static void afterAll() throws IOException { + if (admin != null) { + admin.close(); + } + kafkaCluster.stop(); + removeSslConfiguration(); + } + + @Test + void testTopicOperationsOverTls() throws Exception { + String topicName = "my-tls-topic"; + + assertThat(cmd.execute("topic", "create", "-tp", "2", "-trf", "1", "-t", topicName), is(0)); + waitFor("topic " + topicName + " to be created", () -> admin.listTopics().names().get().contains(topicName)); + + TopicDescription topicDescription = admin.describeTopics(List.of(topicName)).allTopicNames().get().get(topicName); + assertThat(topicDescription.partitions().size(), is(2)); + + List listedTopics = Arrays.asList(captureOutput("topic", "list").split("\\n")); + assertThat(listedTopics, hasItem(topicName)); + + assertThat(cmd.execute("topic", "delete", "-t", topicName), is(0)); + waitFor("topic " + topicName + " to be deleted", () -> !admin.listTopics().names().get().contains(topicName)); + } + + private static String captureOutput(String... args) { + PrintStream originalOut = System.out; + ByteArrayOutputStream out = new ByteArrayOutputStream(); + + try { + System.setOut(new PrintStream(out, true, StandardCharsets.UTF_8)); + cmd.execute(args); + } finally { + System.setOut(originalOut); + } + + return out.toString(StandardCharsets.UTF_8); + } + + /** + * The admin client keeps its configuration in a shared folder (~/.admin_client by default), which the + * plaintext ITs reuse. Leaving the SSL entries there would switch them to TLS against a plaintext cluster. + */ + private static void removeSslConfiguration() throws IOException { + Path configFolder = Paths.get(ConfigurationUtils.getConfigFolderPath()); + Properties properties = ConfigurationUtils.getAdminClientPropertiesIfExists(); + SSL_CONFIG_KEYS.forEach(properties::remove); + + try (FileOutputStream fileOutputStream = new FileOutputStream(configFolder.resolve("config.properties").toFile())) { + properties.store(fileOutputStream, "Store properties to config file"); + } + + for (String fileName : SSL_FILE_NAMES) { + Files.deleteIfExists(configFolder.resolve(fileName)); + } + } +} diff --git a/admin/src/test/java/admin/integration/AdminTopicIT.java b/admin/src/test/java/admin/integration/AdminTopicIT.java index c359eed..15bfd8b 100644 --- a/admin/src/test/java/admin/integration/AdminTopicIT.java +++ b/admin/src/test/java/admin/integration/AdminTopicIT.java @@ -22,9 +22,8 @@ import java.util.Optional; import java.util.Properties; import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.locks.LockSupport; +import static io.strimzi.testclients.testutils.TestUtils.waitFor; import static org.hamcrest.CoreMatchers.is; import static org.hamcrest.MatcherAssert.assertThat; @@ -36,8 +35,7 @@ void testCreateTopics() throws ExecutionException, InterruptedException { String topicPref = "prefixed-topic"; cmd.execute("topic", "create", "-tp", "2", "-trf", "2", "-t", topicName); - // Sleep for a while to prevent race condition during topics creation - LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(2)); + waitForTopic(topicName); Optional topic = admin.listTopics().names().get().stream().filter(t -> t.equals(topicName)).findFirst(); @@ -49,9 +47,8 @@ void testCreateTopics() throws ExecutionException, InterruptedException { assertThat(topicDescription.partitions().get(0).replicas().size(), is(2)); cmd.execute("topic", "create", "-tp", "1", "-trf", "1", "-tpref", topicPref, "-fi", "3", "-tc", "10"); - - // Sleep for a while to prevent race condition during topics creation - LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(2)); + waitFor("10 topics with prefix " + topicPref + " to be created", + () -> admin.listTopics().names().get().stream().filter(t -> t.startsWith(topicPref)).count() == 10); List prefixedTopics = admin.listTopics().names().get().stream() .filter(t -> t.startsWith(topicPref)) @@ -64,7 +61,7 @@ void testCreateTopics() throws ExecutionException, InterruptedException { } @Test - void testListTopics() { + void testListTopics() throws ExecutionException, InterruptedException { String listTopicPrefix = "list-topic-"; List listTopicsToBeCreated = new ArrayList<>(); @@ -72,25 +69,26 @@ void testListTopics() { listTopicsToBeCreated.add(new NewTopic(listTopicPrefix + i, 1, (short) 1)); } - admin.createTopics(listTopicsToBeCreated); + admin.createTopics(listTopicsToBeCreated).all().get(); - // Sleep for a while to prevent race condition during topics creation - LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(2)); - cmd.execute("topic", "list"); - List topics = Arrays.stream(OUT.toString().split("\\n")).filter(t -> t.startsWith(listTopicPrefix)).toList(); + // The CLI may ask a broker that has not seen all the new topics yet, so wait on its own output + waitFor("admin-client to list all 5 topics with prefix " + listTopicPrefix, () -> { + OUT.reset(); + cmd.execute("topic", "list"); + return listedTopicsWithPrefix(listTopicPrefix).size() == 5; + }); + List topics = listedTopicsWithPrefix(listTopicPrefix); assertThat(topics.size(), is(5)); assertThat(topics.containsAll(listTopicsToBeCreated.stream().map(NewTopic::name).toList()), is(true)); } @Test - void testDescribeTopic() throws JsonProcessingException { + void testDescribeTopic() throws JsonProcessingException, ExecutionException, InterruptedException { String topicName = "describe-topic"; NewTopic newTopic = new NewTopic(topicName, 3, (short) 2); - admin.createTopics(List.of(newTopic)); - - // Sleep for a while to prevent race condition during topics creation - LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(2)); + admin.createTopics(List.of(newTopic)).all().get(); + waitForTopic(topicName); cmd.execute("topic", "describe", "-t", topicName); assertThat(OUT.toString().contains("name:describe-topic, partitions:3, replicas:2"), is(true)); @@ -112,54 +110,43 @@ void testDescribeTopic() throws JsonProcessingException { void testDeleteTopic() throws InterruptedException, ExecutionException { String topicName = "delete-topic"; NewTopic newTopic = new NewTopic(topicName, 1, (short) 1); - admin.createTopics(List.of(newTopic)); - - // Sleep for a while to prevent race condition during topics creation - LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(2)); + admin.createTopics(List.of(newTopic)).all().get(); // Check that the topic is really present - List topics = admin.listTopics().names().get().stream().toList(); - assertThat(topics.stream().anyMatch(t -> t.equals(topicName)), is(true)); + waitForTopic(topicName); // Delete topic using admin-client CLI cmd.execute("topic", "delete", "-t", topicName); assertThat(OUT.toString().contains("Topic(s) with name/prefix: delete-topic successfully deleted"), is(true)); - // Sleep for a while to prevent race condition during topics deletion - LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(2)); - topics = admin.listTopics().names().get().stream().toList(); - assertThat(topics.stream().anyMatch(t -> t.equals(topicName)), is(false)); + waitFor("topic " + topicName + " to be deleted", () -> !admin.listTopics().names().get().contains(topicName)); } @Test void testAlterTopic() throws InterruptedException, ExecutionException { String topicName = "alter-topic"; NewTopic newTopic = new NewTopic(topicName, 1, (short) 1); - admin.createTopics(List.of(newTopic)); - - // Sleep for a while to prevent race condition during topics creation - LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(2)); + admin.createTopics(List.of(newTopic)).all().get(); + waitForTopic(topicName); // Alter topic using admin-client cmd.execute("topic", "alter", "-t", topicName, "-tp", "2"); assertThat(OUT.toString().contains("Topic(s) with name/prefix: alter-topic successfully altered."), is(true)); - // Sleep for a while to prevent race condition during topics alteration - LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(2)); + waitFor("topic " + topicName + " to have 2 partitions", + () -> describeTopic(topicName).partitions().size() == 2); - TopicDescription topicDescription = admin.describeTopics(List.of(topicName)).allTopicNames().get().get(topicName); + TopicDescription topicDescription = describeTopic(topicName); assertThat(topicDescription.partitions().size(), is(2)); assertThat(topicDescription.partitions().get(0).replicas().size(), is(1)); } @Test - void testFetchOffsetsForTopic() throws JsonProcessingException { + void testFetchOffsetsForTopic() throws JsonProcessingException, ExecutionException, InterruptedException { String topicName = "fetch-offsets-topic"; NewTopic newTopic = new NewTopic(topicName, 1, (short) 1); - admin.createTopics(List.of(newTopic)); - - // Sleep for a while to prevent race condition during topic creation - LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(2)); + admin.createTopics(List.of(newTopic)).all().get(); + waitForTopic(topicName); Properties configuration = new Properties(); configuration.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaCluster.getBootstrapServers()); @@ -167,15 +154,15 @@ void testFetchOffsetsForTopic() throws JsonProcessingException { configuration.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); configuration.put(ProducerConfig.ACKS_CONFIG, "all"); - KafkaProducer producer = new KafkaProducer<>(configuration); + try (KafkaProducer producer = new KafkaProducer<>(configuration)) { + for (int i = 0; i < 100; i++) { + producer.send(new ProducerRecord<>(topicName, "my-key-" + i, "my-value" + i)); + } - for (int i = 0; i < 100; i++) { - producer.send(new ProducerRecord<>(topicName, "my-key-" + i, "my-value" + i)); + // Blocks until every record is acknowledged (acks=all), so the offsets below are final + producer.flush(); } - // Sleep for a while to prevent race condition during sending messages - LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(5)); - cmd.execute("topic", "fetch-offsets", "-t", topicName, "-o", "json"); ObjectMapper objectMapper = new ObjectMapper(); @@ -183,4 +170,16 @@ void testFetchOffsetsForTopic() throws JsonProcessingException { assertThat(offsets.get("offset").asInt(), is(100)); } + + private static void waitForTopic(String topicName) { + waitFor("topic " + topicName + " to be created", () -> admin.listTopics().names().get().contains(topicName)); + } + + private static TopicDescription describeTopic(String topicName) throws ExecutionException, InterruptedException { + return admin.describeTopics(List.of(topicName)).allTopicNames().get().get(topicName); + } + + private static List listedTopicsWithPrefix(String prefix) { + return Arrays.stream(OUT.toString().split("\\n")).filter(t -> t.startsWith(prefix)).toList(); + } } diff --git a/clients-common/pom.xml b/clients-common/pom.xml index 31d4356..5545e1e 100644 --- a/clients-common/pom.xml +++ b/clients-common/pom.xml @@ -32,5 +32,33 @@ hamcrest test + + io.strimzi + strimzi-test-container + test + + + + + + + org.apache.maven.plugins + maven-jar-plugin + ${maven.jar.version} + + + + test-jar + + + + io/strimzi/testclients/testutils/** + + + + + + + diff --git a/clients-common/src/test/java/io/strimzi/testclients/testutils/TestUtils.java b/clients-common/src/test/java/io/strimzi/testclients/testutils/TestUtils.java new file mode 100644 index 0000000..10fb18c --- /dev/null +++ b/clients-common/src/test/java/io/strimzi/testclients/testutils/TestUtils.java @@ -0,0 +1,60 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.testclients.testutils; + +import java.time.Duration; +import java.util.concurrent.Callable; + +/** + * Helpers shared by the integration tests of the clients and admin modules. + */ +public class TestUtils { + public static final Duration DEFAULT_TIMEOUT = Duration.ofSeconds(30); + public static final Duration DEFAULT_POLL_INTERVAL = Duration.ofMillis(200); + + /** + * Polls {@code condition} until it returns true, failing after {@link #DEFAULT_TIMEOUT}. + * + * @param description what is being waited for, used in the failure message + * @param condition condition to poll; an exception counts as "not yet" (e.g. metadata not propagated) + */ + public static void waitFor(String description, Callable condition) { + waitFor(description, DEFAULT_POLL_INTERVAL, DEFAULT_TIMEOUT, condition); + } + + /** + * Polls {@code condition} until it returns true, failing after {@code timeout}. + * + * @param description what is being waited for, used in the failure message + * @param pollInterval pause between two checks + * @param timeout how long to wait before failing + * @param condition condition to poll; an exception counts as "not yet" and is attached to the failure + */ + public static void waitFor(String description, Duration pollInterval, Duration timeout, Callable condition) { + long deadline = System.nanoTime() + timeout.toNanos(); + Exception lastException = null; + + while (true) { + try { + if (Boolean.TRUE.equals(condition.call())) { + return; + } + } catch (Exception e) { + lastException = e; + } + + if (System.nanoTime() > deadline) { + throw new AssertionError("Timed out after " + timeout + " waiting for " + description, lastException); + } + + try { + Thread.sleep(pollInterval.toMillis()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException("Interrupted while waiting for " + description, e); + } + } + } +} diff --git a/clients-common/src/test/java/io/strimzi/testclients/testutils/TlsUtils.java b/clients-common/src/test/java/io/strimzi/testclients/testutils/TlsUtils.java new file mode 100644 index 0000000..7b66341 --- /dev/null +++ b/clients-common/src/test/java/io/strimzi/testclients/testutils/TlsUtils.java @@ -0,0 +1,94 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.testclients.testutils; + +import io.strimzi.test.container.StrimziKafkaCluster; +import org.apache.kafka.clients.CommonClientConfigs; +import org.apache.kafka.common.config.SslConfigs; +import org.apache.kafka.common.security.auth.SecurityProtocol; + +import java.io.ByteArrayInputStream; +import java.nio.charset.StandardCharsets; +import java.security.KeyStore; +import java.security.cert.Certificate; +import java.util.Base64; +import java.util.Collections; +import java.util.Map; + +/** + * Converts the PKCS12 client stores generated by {@link StrimziKafkaCluster} into the PEM strings + * the test clients expect in CA_CRT, USER_CRT and USER_KEY (and the admin client behind `configure ssl`). + */ +public class TlsUtils { + + private static final Base64.Encoder PEM_ENCODER = Base64.getMimeEncoder(64, "\n".getBytes(StandardCharsets.US_ASCII)); + + /** + * Configuration of a plain Kafka client connecting to the TLS cluster, used to verify what the test clients did. + */ + public static Map kafkaClientConfig(StrimziKafkaCluster cluster) throws Exception { + return Map.of( + CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, cluster.getBootstrapServers(), + CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name, + SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, "PEM", + SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG, caCertificate(cluster), + SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, "PEM", + SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG, userCertificateChain(cluster), + SslConfigs.SSL_KEYSTORE_KEY_CONFIG, userKey(cluster) + ); + } + + public static String caCertificate(StrimziKafkaCluster cluster) throws Exception { + KeyStore trustStore = load(cluster.getClientTrustStoreBytes(), cluster.getClientStorePassword()); + StringBuilder pem = new StringBuilder(); + + for (String alias : Collections.list(trustStore.aliases())) { + if (trustStore.isCertificateEntry(alias)) { + pem.append(toPem("CERTIFICATE", trustStore.getCertificate(alias).getEncoded())); + } + } + + return pem.toString(); + } + + public static String userCertificateChain(StrimziKafkaCluster cluster) throws Exception { + KeyStore keyStore = load(cluster.getClientKeyStoreBytes(), cluster.getClientStorePassword()); + StringBuilder pem = new StringBuilder(); + + for (Certificate certificate : keyStore.getCertificateChain(keyAlias(keyStore))) { + pem.append(toPem("CERTIFICATE", certificate.getEncoded())); + } + + return pem.toString(); + } + + public static String userKey(StrimziKafkaCluster cluster) throws Exception { + KeyStore keyStore = load(cluster.getClientKeyStoreBytes(), cluster.getClientStorePassword()); + char[] password = cluster.getClientStorePassword().toCharArray(); + + // getEncoded() of a private key is PKCS#8, the unencrypted PEM form Kafka accepts without ssl.key.password + return toPem("PRIVATE KEY", keyStore.getKey(keyAlias(keyStore), password).getEncoded()); + } + + private static KeyStore load(byte[] storeBytes, String password) throws Exception { + KeyStore store = KeyStore.getInstance("PKCS12"); + store.load(new ByteArrayInputStream(storeBytes), password.toCharArray()); + return store; + } + + private static String keyAlias(KeyStore keyStore) throws Exception { + for (String alias : Collections.list(keyStore.aliases())) { + if (keyStore.isKeyEntry(alias)) { + return alias; + } + } + + throw new IllegalStateException("Client keystore contains no private key entry"); + } + + private static String toPem(String type, byte[] der) { + return "-----BEGIN " + type + "-----\n" + PEM_ENCODER.encodeToString(der) + "\n-----END " + type + "-----\n"; + } +} diff --git a/clients/pom.xml b/clients/pom.xml index d0b4530..70f01de 100644 --- a/clients/pom.xml +++ b/clients/pom.xml @@ -21,6 +21,12 @@ io.strimzi.test-clients clients-common + + io.strimzi.test-clients + clients-common + test-jar + test + com.fasterxml.jackson.core jackson-core diff --git a/clients/src/test/java/io/strimzi/testclients/integration/KafkaClientTlsIT.java b/clients/src/test/java/io/strimzi/testclients/integration/KafkaClientTlsIT.java new file mode 100644 index 0000000..0ac72ec --- /dev/null +++ b/clients/src/test/java/io/strimzi/testclients/integration/KafkaClientTlsIT.java @@ -0,0 +1,100 @@ +/* + * Copyright Strimzi authors. + * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html). + */ +package io.strimzi.testclients.integration; + +import io.strimzi.test.container.StrimziKafkaCluster; +import io.strimzi.testclients.configuration.ConfigurationConstants; +import io.strimzi.testclients.kafka.KafkaConsumerClient; +import io.strimzi.testclients.kafka.KafkaProducerClient; +import io.strimzi.testclients.testutils.TlsUtils; +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.NewTopic; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; + +import java.lang.reflect.Field; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; + +/** + * Runs the producer and consumer against a cluster that only exposes an mTLS listener, + * with the certificates passed the way the clients get them in Kubernetes: PEM in CA_CRT, USER_CRT and USER_KEY. + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class KafkaClientTlsIT { + private static StrimziKafkaCluster kafkaCluster; + private static String caCrt; + private static String userCrt; + private static String userKey; + + @BeforeAll + static void setup() throws Exception { + kafkaCluster = new StrimziKafkaCluster.StrimziKafkaClusterBuilder() + .withNumberOfBrokers(1) + .withInternalTopicReplicationFactor(1) + .withSharedNetwork() + .withTls() + .build(); + kafkaCluster.start(); + + caCrt = TlsUtils.caCertificate(kafkaCluster); + userCrt = TlsUtils.userCertificateChain(kafkaCluster); + userKey = TlsUtils.userKey(kafkaCluster); + } + + @AfterAll + static void afterAll() { + kafkaCluster.stop(); + } + + @Test + void testTlsExchange() throws Exception { + String topicName = "my-tls-topic"; + + // Topic is created by a plain Kafka Admin with the same PEM material, so the clients under test are the only TLS path we rely on + try (Admin admin = Admin.create(TlsUtils.kafkaClientConfig(kafkaCluster))) { + admin.createTopics(List.of(new NewTopic(topicName, 1, (short) 1))).all().get(); + } + + KafkaProducerClient kafkaProducerClient = new KafkaProducerClient(tlsClientConfiguration(topicName)); + + CompletableFuture future = CompletableFuture.runAsync(kafkaProducerClient::run); + // Wait for the process to complete within a reasonable time + future.get(30, TimeUnit.SECONDS); + + Field producedMessages = KafkaProducerClient.class.getDeclaredField("messageSuccessfullySent"); + producedMessages.setAccessible(true); + assertThat(producedMessages.get(kafkaProducerClient), is(100)); + + KafkaConsumerClient kafkaConsumerClient = new KafkaConsumerClient(tlsClientConfiguration(topicName)); + + future = CompletableFuture.runAsync(kafkaConsumerClient::run); + // Wait for the process to complete within a reasonable time + future.get(30, TimeUnit.SECONDS); + + Field consumedMessages = KafkaConsumerClient.class.getDeclaredField("consumedMessages"); + consumedMessages.setAccessible(true); + assertThat(consumedMessages.get(kafkaConsumerClient), is(100)); + } + + private Map tlsClientConfiguration(String topicName) { + Map configuration = new HashMap<>(); + configuration.put(ConfigurationConstants.BOOTSTRAP_SERVERS_ENV, kafkaCluster.getBootstrapServers()); + configuration.put(ConfigurationConstants.TOPIC_ENV, topicName); + configuration.put(ConfigurationConstants.MESSAGE_COUNT_ENV, "100"); + configuration.put(ConfigurationConstants.CA_CRT_ENV, caCrt); + configuration.put(ConfigurationConstants.USER_CRT_ENV, userCrt); + configuration.put(ConfigurationConstants.USER_KEY_ENV, userKey); + return configuration; + } +} diff --git a/pom.xml b/pom.xml index f2c52c0..ddf0a5a 100644 --- a/pom.xml +++ b/pom.xml @@ -303,6 +303,13 @@ clients-common ${project.version} + + io.strimzi.test-clients + clients-common + ${project.version} + test-jar + test + io.strimzi.test-clients tracing