diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaChannelInitializer.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaChannelInitializer.java index 587c3c5db8..4f2c10727f 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaChannelInitializer.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaChannelInitializer.java @@ -49,7 +49,7 @@ public class KafkaChannelInitializer extends ChannelInitializer { @Getter private final EndPoint advertisedEndPoint; @Getter - private final SslContextFactory sslContextFactory; + private final SslContextFactory.Server sslContextFactory; @Getter private final StatsLogger statsLogger; diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionMarkerChannelInitializer.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionMarkerChannelInitializer.java index 8a07e93adc..6cf6912fdb 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionMarkerChannelInitializer.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionMarkerChannelInitializer.java @@ -32,7 +32,7 @@ public class TransactionMarkerChannelInitializer extends ChannelInitializer 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 sslConfigValues) { - SslContextFactory ssl = new SslContextFactory(); + public static SslContextFactory.Server createSslContextFactory(Map 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 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); diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaSSLChannelTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaSSLChannelTest.java index ff2685bb82..350a1a3f43 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaSSLChannelTest.java +++ b/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 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", ""); diff --git a/tests/src/test/resources/ssl/certificate2/client-ssl.properties b/tests/src/test/resources/ssl/certificate2/client-ssl.properties new file mode 100644 index 0000000000..467115d16e --- /dev/null +++ b/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= diff --git a/tests/src/test/resources/ssl/certificate2/client.truststore.jks b/tests/src/test/resources/ssl/certificate2/client.truststore.jks new file mode 100644 index 0000000000..dd152ebfc2 Binary files /dev/null and b/tests/src/test/resources/ssl/certificate2/client.truststore.jks differ diff --git a/tests/src/test/resources/ssl/certificate2/server.keystore.jks b/tests/src/test/resources/ssl/certificate2/server.keystore.jks new file mode 100644 index 0000000000..7517baa590 Binary files /dev/null and b/tests/src/test/resources/ssl/certificate2/server.keystore.jks differ diff --git a/tests/src/test/resources/ssl/certificate2/server.truststore.jks b/tests/src/test/resources/ssl/certificate2/server.truststore.jks new file mode 100644 index 0000000000..db0e4b6292 Binary files /dev/null and b/tests/src/test/resources/ssl/certificate2/server.truststore.jks differ