Add TLS integrations coverage

admin/pom.xml

@@ -25,6 +25,12 @@
             <groupId>io.strimzi.test-clients</groupId>
             <artifactId>clients-common</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.strimzi.test-clients</groupId>
+            <artifactId>clients-common</artifactId>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>io.strimzi</groupId>
             <artifactId>kafka-kubernetes-config-provider</artifactId>

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);

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;
     }

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<String> SSL_CONFIG_KEYS = List.of(
+        ConfigurationConstants.CA_CRT_PROPERTY,
+        ConfigurationConstants.USER_KEY_PROPERTY,
+        ConfigurationConstants.USER_CRT_PROPERTY
+    );
+    private static final List<String> 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<String> 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));
+        }
+    }
+}

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<String> 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<String> prefixedTopics = admin.listTopics().names().get().stream()
             .filter(t -> t.startsWith(topicPref))
@@ -64,33 +61,34 @@ void testCreateTopics() throws ExecutionException, InterruptedException {
     }
 
     @Test
-    void testListTopics() {
+    void testListTopics() throws ExecutionException, InterruptedException {
         String listTopicPrefix = "list-topic-";
         List<NewTopic> listTopicsToBeCreated = new ArrayList<>();
 
         for (int i = 0; i < 5; i++) {
             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<String> 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<String> 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,75 +110,76 @@ 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<String> 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());
         configuration.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
         configuration.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
         configuration.put(ProducerConfig.ACKS_CONFIG, "all");
 
-        KafkaProducer<String, String> producer = new KafkaProducer<>(configuration);
+        try (KafkaProducer<String, String> 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();
         JsonNode offsets = objectMapper.readTree(OUT.toString()).get("0");
 
         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<String> listedTopicsWithPrefix(String prefix) {
+        return Arrays.stream(OUT.toString().split("\\n")).filter(t -> t.startsWith(prefix)).toList();
+    }
 }