Allow to configure SecureBundle ZIP in textual form

- allow users to encode the ZIP file using standard base64 linux tool
- This is the new setting cloud.secureConnectBundle=base64:XXXXXXXXX

With this change it is easier to connect to Astra and to secure Cassandra clusters without the need to deploy the ZIP file.
Deploying the ZIP file is very awkward on K8S and in Pulsar IO framework

Author: eolivelli

common/src/main/java/com/datastax/oss/common/sink/config/CassandraSinkConfig.java

@@ -30,6 +30,11 @@
 import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.attribute.PosixFilePermissions;
+import java.util.Base64;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -247,6 +252,7 @@ public CassandraSinkConfig(Map<String, String> settings) {
       globalConfig = new AbstractConfig(GLOBAL_CONFIG_DEF, globalSettings, false);
 
       populateDriverSettingsWithConnectorSettings(globalSettings);
+      decodeBase64EncodedSecureBundle(javaDriverSettings);
       boolean cloud = isCloud();
 
       if (!cloud) {
@@ -332,7 +338,6 @@ private void populateDriverSettingsWithConnectorSettings(Map<String, String> con
     deprecatedMetricsHighestLatency(connectorSettings);
     deprecatedCompression(connectorSettings);
     deprecatedSecureBundle(connectorSettings);
-
     if (getJmx()) {
       metricsSettings();
     }
@@ -347,6 +352,31 @@ private void deprecatedSecureBundle(Map<String, String> connectorSettings) {
         Function.identity());
   }
 
+  static void decodeBase64EncodedSecureBundle(Map<String, String> javaDriverSettings) {
+    // if the path is base64:xxxx we decode the payload and create
+    // a temporary file
+    // we are setting permissions such that only current user can access the file
+    // the file will be deleted at JVM exit
+    String encoded = javaDriverSettings.get(SECURE_CONNECT_BUNDLE_DRIVER_SETTING);
+    if (encoded != null && encoded.startsWith("base64:")) {
+      try {
+        encoded = encoded.replace("\n", "").replace("\r", "").trim();
+        encoded = encoded.substring("base64:".length());
+        byte[] decoded = Base64.getDecoder().decode(encoded);
+        Path file = Files.createTempFile("cassandra.sink.securebundle", ".zip");
+        Files.setPosixFilePermissions(file, PosixFilePermissions.fromString("rw-------"));
+        file.toFile().deleteOnExit();
+        Files.write(file, decoded);
+        String path = file.toAbsolutePath().toString();
+        log.info("Decoded bundle to temporary file {}", path);
+        javaDriverSettings.put(SECURE_CONNECT_BUNDLE_DRIVER_SETTING, path);
+      } catch (IOException ex) {
+        throw new RuntimeException(
+            "Cannot decode base64 secure bundle and create temporary file: " + ex, ex);
+      }
+    }
+  }
+
   private void deprecatedCompression(Map<String, String> connectorSettings) {
     handleDeprecatedSetting(
         connectorSettings,

common/src/test/java/com/datastax/oss/common/sink/config/CassandraSinkConfigTest.java

@@ -46,6 +46,7 @@
 import static com.datastax.oss.driver.api.core.config.DefaultDriverOption.METRICS_SESSION_CQL_REQUESTS_INTERVAL;
 import static com.datastax.oss.driver.api.core.config.DefaultDriverOption.METRICS_SESSION_ENABLED;
 import static com.datastax.oss.dsbulk.tests.assertions.TestAssertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
@@ -58,11 +59,19 @@
 import com.datastax.oss.driver.shaded.guava.common.collect.Maps;
 import com.datastax.oss.dsbulk.tests.logging.LogInterceptingExtension;
 import com.datastax.oss.dsbulk.tests.logging.LogInterceptor;
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.attribute.PosixFilePermission;
+import java.util.Base64;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Stream;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -670,6 +679,46 @@ void should_set_default_driver_setting(String driverSettingName, String expected
         .isEqualTo(expectedDefault);
   }
 
+  @Test
+  void should_unpack_base64_zip_file_legacy_name() throws Exception {
+    should_unpack_base64_zip_file(SECURE_CONNECT_BUNDLE_OPT);
+  }
+
+  @Test
+  void should_unpack_base64_zip_file() throws Exception {
+    should_unpack_base64_zip_file(SECURE_CONNECT_BUNDLE_DRIVER_SETTING);
+  }
+
+  void should_unpack_base64_zip_file(String entryName) throws Exception {
+    ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+    // it is not necessary that we really have a zip file here
+    // but this gives the flavour of the type of content we expect
+    // to be encoded in the secureBundleZip
+    try (ZipOutputStream zip = new ZipOutputStream(buffer)) {
+      zip.putNextEntry(new ZipEntry("file.bin"));
+      zip.write(1234);
+      zip.closeEntry();
+    }
+    byte[] zipFileContents = buffer.toByteArray();
+    String encoded = "base64:" + Base64.getEncoder().encodeToString(zipFileContents);
+    Map<String, String> inputSettings = new HashMap<>();
+    inputSettings.put(entryName, encoded);
+    CassandraSinkConfig cassandraSinkConfig = new CassandraSinkConfig(inputSettings);
+    assertThat(cassandraSinkConfig.isCloud()).isEqualTo(true);
+    Map<String, String> javaDriverSettings = cassandraSinkConfig.getJavaDriverSettings();
+    String path = javaDriverSettings.get(SECURE_CONNECT_BUNDLE_DRIVER_SETTING);
+    File file = new File(path);
+    assertThat(file.isFile()).isEqualTo(true);
+    Set<PosixFilePermission> posixFilePermissions = Files.getPosixFilePermissions(file.toPath());
+    assertThat(posixFilePermissions)
+        .contains(PosixFilePermission.OWNER_READ, PosixFilePermission.OWNER_WRITE);
+    assertThat(posixFilePermissions.size())
+        .as("bad permissions " + posixFilePermissions)
+        .isEqualTo(2);
+    byte[] content = Files.readAllBytes(file.toPath());
+    assertThat(content).isEqualTo(zipFileContents);
+  }
+
   @Test
   void should_transform_list_setting_to_indexed_typesafe_setting() {
     // given