fix: fedramp issues

ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/ClientOptions.java

@@ -116,6 +116,14 @@ public interface ClientOptions {
    */
   ClientOptions setKeyAlias(String keyAlias);
 
+  /**
+   * Sets key/trust store type.
+   *
+   * @param storeType store type
+   * @return a reference to this
+   */
+  ClientOptions setStoreType(String storeType);
+
   /**
    * Sets the username and password to be used for HTTP basic authentication when connecting to the
    * ksqlDB server. Basic authentication will be used unless both username and password are null
@@ -245,6 +253,13 @@ public interface ClientOptions {
    */
   String getKeyAlias();
 
+  /**
+   * Returns the store type.
+   *
+   * @return key/trust store type
+   */
+  String getStoreType();
+
   /**
    * Returns the username to be used for HTTP basic authentication, if applicable.
    *

ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/impl/ClientImpl.java

@@ -59,6 +59,7 @@
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.net.JksOptions;
+import io.vertx.core.net.KeyStoreOptions;
 import io.vertx.core.net.SocketAddress;
 import io.vertx.core.parsetools.RecordParser;
 import java.nio.charset.Charset;
@@ -86,6 +87,7 @@ public class ClientImpl implements Client {
   private static final String CLOSE_QUERY_ENDPOINT = "/close-query";
   private static final String KSQL_ENDPOINT = "/ksql";
   private static final String INFO_ENDPOINT = "/info";
+  private static final String SSL_STORE_TYPE_BCFKS = "BCFKS";
 
   private final ClientOptions clientOptions;
   private final Vertx vertx;
@@ -904,22 +906,46 @@ private static HttpClient createHttpClient(final Vertx vertx, final ClientOption
         .setDefaultPort(clientOptions.getPort())
         .setHttp2MultiplexingLimit(clientOptions.getHttp2MultiplexingLimit());
     if (clientOptions.isUseTls() && !clientOptions.getTrustStore().isEmpty()) {
-      final JksOptions jksOptions = VertxSslOptionsFactory.getJksTrustStoreOptions(
-          clientOptions.getTrustStore(),
-          clientOptions.getTrustStorePassword()
-      );
+      if (Objects.equals(clientOptions.getStoreType(), SSL_STORE_TYPE_BCFKS)) {
+        final Optional<KeyStoreOptions> bcfksOptions =
+            VertxSslOptionsFactory.getBcfksTrustStoreOptions(
+                clientOptions.getTrustStore(),
+                clientOptions.getTrustStorePassword(),
+                clientOptions.getKeyPassword());
+
+        if (bcfksOptions.isPresent()) {
+          options = options.setTrustOptions(bcfksOptions.get());
+        }
+      } else {
+        final JksOptions jksOptions = VertxSslOptionsFactory.getJksTrustStoreOptions(
+            clientOptions.getTrustStore(),
+            clientOptions.getTrustStorePassword()
+        );
 
-      options = options.setTrustStoreOptions(jksOptions);
+        options = options.setTrustStoreOptions(jksOptions);
+      }
     }
     if (!clientOptions.getKeyStore().isEmpty()) {
-      final JksOptions jksOptions = VertxSslOptionsFactory.buildJksKeyStoreOptions(
-          clientOptions.getKeyStore(),
-          clientOptions.getKeyStorePassword(),
-          Optional.of(clientOptions.getKeyPassword()),
-          Optional.of(clientOptions.getKeyAlias())
-      );
-
-      options = options.setKeyStoreOptions(jksOptions);
+      if (Objects.equals(clientOptions.getStoreType(), SSL_STORE_TYPE_BCFKS)) {
+        final Optional<KeyStoreOptions> keyStoreOptions =
+            VertxSslOptionsFactory.getBcfksKeyStoreOptions(
+                clientOptions.getKeyStore(),
+                clientOptions.getKeyStorePassword(),
+                clientOptions.getKeyPassword());
+
+        if (keyStoreOptions.isPresent()) {
+          options = options.setKeyCertOptions(keyStoreOptions.get());
+        }
+      } else {
+        final JksOptions jksOptions = VertxSslOptionsFactory.buildJksKeyStoreOptions(
+            clientOptions.getKeyStore(),
+            clientOptions.getKeyStorePassword(),
+            Optional.of(clientOptions.getKeyPassword()),
+            Optional.of(clientOptions.getKeyAlias())
+        );
+
+        options = options.setKeyStoreOptions(jksOptions);
+      }
     }
     return vertx.createHttpClient(options);
   }

ksqldb-api-client/src/main/java/io/confluent/ksql/api/client/impl/ClientOptionsImpl.java

@@ -35,6 +35,7 @@ public class ClientOptionsImpl implements ClientOptions {
   private String keyStorePassword;
   private String keyPassword;
   private String keyAlias;
+  private String storeType;
   private String basicAuthUsername;
   private String basicAuthPassword;
   private int executeQueryMaxResultRows = ClientOptions.DEFAULT_EXECUTE_QUERY_MAX_RESULT_ROWS;
@@ -56,9 +57,9 @@ private ClientOptionsImpl(
       final boolean useBasicAuth,
       final String trustStorePath, final String trustStorePassword,
       final String keyStorePath, final String keyStorePassword, final String keyPassword,
-      final String keyAlias, final String basicAuthUsername, final String basicAuthPassword,
-      final int executeQueryMaxResultRows, final int http2MultiplexingLimit,
-      final Map<String, String> requestHeaders) {
+      final String keyAlias, final String storeType, final String basicAuthUsername,
+      final String basicAuthPassword, final int executeQueryMaxResultRows,
+      final int http2MultiplexingLimit, final Map<String, String> requestHeaders) {
     this.host = Objects.requireNonNull(host);
     this.port = port;
     this.useTls = useTls;
@@ -71,6 +72,7 @@ private ClientOptionsImpl(
     this.keyStorePassword = keyStorePassword;
     this.keyPassword = keyPassword;
     this.keyAlias = keyAlias;
+    this.storeType = storeType;
     this.basicAuthUsername = basicAuthUsername;
     this.basicAuthPassword = basicAuthPassword;
     this.executeQueryMaxResultRows = executeQueryMaxResultRows;
@@ -144,6 +146,12 @@ public ClientOptions setKeyAlias(final String keyAlias) {
     return this;
   }
 
+  @Override
+  public ClientOptions setStoreType(final String storeType) {
+    this.storeType = storeType;
+    return this;
+  }
+
   @Override
   public ClientOptions setBasicAuthCredentials(final String username, final String password) {
     this.useBasicAuth = !(username == null || username.isEmpty())
@@ -231,6 +239,11 @@ public String getKeyAlias() {
     return keyAlias == null ? "" : keyAlias;
   }
 
+  @Override
+  public String getStoreType() {
+    return storeType == null ? "JKS" : storeType;
+  }
+
   @Override
   public String getBasicAuthUsername() {
     return basicAuthUsername == null ? "" : basicAuthUsername;
@@ -263,7 +276,7 @@ public ClientOptions copy() {
         useTls, verifyHost, useAlpn,
         useBasicAuth,
         trustStorePath, trustStorePassword,
-        keyStorePath, keyStorePassword, keyPassword, keyAlias,
+        keyStorePath, keyStorePassword, keyPassword, keyAlias, storeType,
         basicAuthUsername, basicAuthPassword,
         executeQueryMaxResultRows, http2MultiplexingLimit,
         requestHeaders);
@@ -292,6 +305,7 @@ public boolean equals(final Object o) {
         && Objects.equals(keyStorePassword, that.keyStorePassword)
         && Objects.equals(keyPassword, that.keyPassword)
         && Objects.equals(keyAlias, that.keyAlias)
+        && Objects.equals(storeType, that.storeType)
         && Objects.equals(basicAuthUsername, that.basicAuthUsername)
         && Objects.equals(basicAuthPassword, that.basicAuthPassword)
         && http2MultiplexingLimit == that.http2MultiplexingLimit
@@ -301,7 +315,7 @@ public boolean equals(final Object o) {
   @Override
   public int hashCode() {
     return Objects.hash(host, port, useTls, verifyHost, useAlpn, trustStorePath,
-        trustStorePassword, keyStorePath, keyStorePassword, keyPassword, keyAlias,
+        trustStorePassword, keyStorePath, keyStorePassword, keyPassword, keyAlias, storeType,
         basicAuthUsername, basicAuthPassword, executeQueryMaxResultRows, http2MultiplexingLimit,
         requestHeaders);
   }
@@ -320,6 +334,7 @@ public String toString() {
         + ", keyStorePassword='" + keyStorePassword + '\''
         + ", keyPassword='" + keyPassword + '\''
         + ", keyAlias='" + keyAlias + '\''
+        + ", storeType='" + storeType + '\''
         + ", basicAuthUsername='" + basicAuthUsername + '\''
         + ", basicAuthPassword='" + basicAuthPassword + '\''
         + ", executeQueryMaxResultRows=" + executeQueryMaxResultRows