[Secure] Replace SslContextFactory with SslContextFactory.Server

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaChannelInitializer.java

@@ -49,7 +49,7 @@ public class KafkaChannelInitializer extends ChannelInitializer<SocketChannel> {
     @Getter
     private final EndPoint advertisedEndPoint;
     @Getter
-    private final SslContextFactory sslContextFactory;
+    private final SslContextFactory.Server sslContextFactory;
     @Getter
     private final StatsLogger statsLogger;
 

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionMarkerChannelInitializer.java

@@ -32,7 +32,7 @@ public class TransactionMarkerChannelInitializer extends ChannelInitializer<Sock
 
     private final KafkaServiceConfiguration kafkaConfig;
     private final boolean enableTls;
-    private final SslContextFactory sslContextFactory;
+    private final SslContextFactory.Server sslContextFactory;
 
     public TransactionMarkerChannelInitializer(KafkaServiceConfiguration kafkaConfig,
                                                boolean enableTls) {

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/ssl/SSLUtils.java

@@ -52,7 +52,8 @@ public class SSLUtils {
             .put(BrokerSecurityConfigs.SSL_CLIENT_AUTH_CONFIG, "kopSslClientAuth")
             .build();
 
-    public static SslContextFactory createSslContextFactory(KafkaServiceConfiguration kafkaServiceConfiguration) {
+    public static SslContextFactory.Server createSslContextFactory(
+            KafkaServiceConfiguration kafkaServiceConfiguration) {
         Builder<String, Object> sslConfigValues = ImmutableMap.builder();
 
         CONFIG_NAME_MAP.forEach((key, value) -> {
@@ -114,8 +115,8 @@ public static SslContextFactory createSslContextFactory(KafkaServiceConfiguratio
         return createSslContextFactory(sslConfigValues.build());
     }
 
-    public static SslContextFactory createSslContextFactory(Map<String, Object> sslConfigValues) {
-        SslContextFactory ssl = new SslContextFactory();
+    public static SslContextFactory.Server createSslContextFactory(Map<String, Object> sslConfigValues) {
+        SslContextFactory.Server ssl = new SslContextFactory.Server();
 
         configureSslContextFactoryKeyStore(ssl, sslConfigValues);
         configureSslContextFactoryTrustStore(ssl, sslConfigValues);
@@ -226,7 +227,7 @@ protected static void configureSslContextFactoryAlgorithms(SslContextFactory ssl
     /**
      * Configures Authentication related settings in SslContextFactory.
      */
-    protected static void configureSslContextFactoryAuthentication(SslContextFactory ssl,
+    protected static void configureSslContextFactoryAuthentication(SslContextFactory.Server ssl,
                                                                    Map<String, Object> sslConfigValues) {
         String sslClientAuth = (String) getOrDefault(
             sslConfigValues,
@@ -248,7 +249,7 @@ protected static void configureSslContextFactoryAuthentication(SslContextFactory
     /**
      * Create SSL engine used in KafkaChannelInitializer.
      */
-    public static SSLEngine createSslEngine(SslContextFactory sslContextFactory) throws Exception {
+    public static SSLEngine createSslEngine(SslContextFactory.Server sslContextFactory) throws Exception {
         sslContextFactory.start();
         SSLEngine engine  = sslContextFactory.newSSLEngine();
         engine.setUseClientMode(false);

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaSSLChannelTest.java

@@ -36,10 +36,13 @@
  */
 @Slf4j
 public class KafkaSSLChannelTest extends KopProtocolHandlerTestBase {
-    protected final String kopSslKeystoreLocation = "./src/test/resources/ssl/certificate/broker.keystore.jks";
-    protected final String kopSslKeystorePassword = "broker";
-    protected final String kopSslTruststoreLocation = "./src/test/resources/ssl/certificate/client.truststore.jks";
-    protected final String kopSslTruststorePassword = "client";
+
+    private String kopSslKeystoreLocation;
+    private String kopSslKeystorePassword;
+    private String kopSslTruststoreLocation;
+    private String kopSslTruststorePassword;
+    private String kopClientTruststoreLocation;
+    private String kopClientTruststorePassword;
 
     static {
         final HostnameVerifier defaultHostnameVerifier = javax.net.ssl.HttpsURLConnection.getDefaultHostnameVerifier();
@@ -56,24 +59,48 @@ public boolean verify(String hostname, javax.net.ssl.SSLSession sslSession) {
         javax.net.ssl.HttpsURLConnection.setDefaultHostnameVerifier(localhostAcceptedHostnameVerifier);
     }
 
-    public KafkaSSLChannelTest(final String entryFormat) {
+    public KafkaSSLChannelTest(final String entryFormat, boolean withCertHost) {
         super(entryFormat);
+        setSslConfigurations(withCertHost);
+    }
+
+    /**
+     * Set ssl configurations.
+     * @param withCertHost the keystore with certHost or not.
+     */
+    private void setSslConfigurations(boolean withCertHost) {
+        String path = "./src/test/resources/ssl/certificate" + (withCertHost ? "2" : "") + "/";
+        if (!withCertHost) {
+            this.kopSslKeystoreLocation = path + "broker.keystore.jks";
+            this.kopSslKeystorePassword = "broker";
+            this.kopSslTruststoreLocation = path + "broker.truststore.jks";
+            this.kopSslTruststorePassword = "broker";
+        } else {
+            this.kopSslKeystoreLocation = path + "server.keystore.jks";
+            this.kopSslKeystorePassword = "server";
+            this.kopSslTruststoreLocation = path + "server.truststore.jks";
+            this.kopSslTruststorePassword = "server";
+        }
+        kopClientTruststoreLocation = path + "client.truststore.jks";
+        kopClientTruststorePassword = "client";
     }
 
     @Factory
     public static Object[] instances() {
         return new Object[] {
-                new KafkaSSLChannelTest("pulsar"),
-                new KafkaSSLChannelTest("kafka")
+                new KafkaSSLChannelTest("pulsar", false),
+                new KafkaSSLChannelTest("pulsar", true),
+                new KafkaSSLChannelTest("kafka", false),
+                new KafkaSSLChannelTest("kafka", true)
         };
     }
 
     protected void sslSetUpForBroker() throws Exception {
-        ((KafkaServiceConfiguration) conf).setKopSslKeystoreType("JKS");
-        ((KafkaServiceConfiguration) conf).setKopSslKeystoreLocation(kopSslKeystoreLocation);
-        ((KafkaServiceConfiguration) conf).setKopSslKeystorePassword(kopSslKeystorePassword);
-        ((KafkaServiceConfiguration) conf).setKopSslTruststoreLocation(kopSslTruststoreLocation);
-        ((KafkaServiceConfiguration) conf).setKopSslTruststorePassword(kopSslTruststorePassword);
+        conf.setKopSslKeystoreType("JKS");
+        conf.setKopSslKeystoreLocation(kopSslKeystoreLocation);
+        conf.setKopSslKeystorePassword(kopSslKeystorePassword);
+        conf.setKopSslTruststoreLocation(kopSslTruststoreLocation);
+        conf.setKopSslTruststorePassword(kopSslTruststorePassword);
     }
 
     @BeforeMethod
@@ -109,7 +136,8 @@ public void testKafkaProduceSSL() throws Exception {
         String messageStrPrefix = "Message_Kop_KafkaProduceKafkaConsume_" + partitionNumber + "_";
 
         @Cleanup
-        SslProducer kProducer = new SslProducer(topicName, getKafkaBrokerPortTls());
+        SslProducer kProducer = new SslProducer(topicName, getKafkaBrokerPortTls(),
+                kopClientTruststoreLocation, kopClientTruststorePassword);
 
         for (int i = 0; i < totalMsgs; i++) {
             String messageStr = messageStrPrefix + i;
@@ -143,7 +171,7 @@ public static class SslProducer implements Closeable {
         private final KafkaProducer<Integer, String> producer;
         private final String topic;
 
-        public SslProducer(String topic, int port) {
+        public SslProducer(String topic, int port, String truststoreLocation, String truststorePassword) {
             Properties props = new Properties();
             props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost" + ":" + port);
             props.put(ProducerConfig.CLIENT_ID_CONFIG, "DemoKafkaOnPulsarProducerSSL");
@@ -152,8 +180,8 @@ public SslProducer(String topic, int port) {
 
             // SSL client config
             props.put("security.protocol", "SSL");
-            props.put("ssl.truststore.location", "./src/test/resources/ssl/certificate/broker.truststore.jks");
-            props.put("ssl.truststore.password", "broker");
+            props.put("ssl.truststore.location", truststoreLocation);
+            props.put("ssl.truststore.password", truststorePassword);
 
             // default is https, here need to set empty.
             props.put("ssl.endpoint.identification.algorithm", "");

tests/src/test/resources/ssl/certificate2/client-ssl.properties

@@ -0,0 +1,18 @@
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+security.protocol=SSL
+ssl.truststore.location=./src/test/resources/ssl/certificate2/client.truststore.jks
+ssl.truststore.password=client
+ssl.endpoint.identification.algorithm=