Fix master compile issue

mqtt-broker/src/main/java/io/streamnative/pulsar/handlers/mqtt/broker/channel/MQTTChannelInitializer.java

@@ -25,25 +25,22 @@
 import io.streamnative.pulsar.handlers.mqtt.common.adapter.MqttAdapterDecoder;
 import io.streamnative.pulsar.handlers.mqtt.common.adapter.MqttAdapterEncoder;
 import io.streamnative.pulsar.handlers.mqtt.common.psk.PSKUtils;
+import io.streamnative.pulsar.handlers.mqtt.common.tls.MQTTTlsFactory;
 import io.streamnative.pulsar.handlers.mqtt.common.utils.WebSocketUtils;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.common.util.PulsarSslConfiguration;
-import org.apache.pulsar.common.util.PulsarSslFactory;
+import org.apache.pulsar.tls.TlsPurpose;
 
 /**
  * A channel initializer that initialize channels for MQTT protocol.
  */
-@Slf4j
 public class MQTTChannelInitializer extends ChannelInitializer<SocketChannel> {
 
     private final MQTTServerConfiguration mqttConfig;
     private final MQTTService mqttService;
     private final boolean enableTls;
     private final boolean enableTlsPsk;
     private final boolean enableWs;
-    private PulsarSslFactory sslFactory;
+    private MQTTTlsFactory tlsFactory;
 
     public MQTTChannelInitializer(MQTTService mqttService, boolean enableTls, boolean enableWs,
                                   ScheduledExecutorService sslContextRefresher) throws Exception {
@@ -60,24 +57,15 @@ public MQTTChannelInitializer(
         this.enableTlsPsk = enableTlsPsk;
         this.enableWs = enableWs;
         if (this.enableTls) {
-            PulsarSslConfiguration sslConfiguration = buildSslConfiguration(mqttConfig);
-            this.sslFactory = (PulsarSslFactory) Class.forName(mqttConfig.getSslFactoryPlugin())
-                    .getConstructor().newInstance();
-            this.sslFactory.initialize(sslConfiguration);
-            this.sslFactory.createInternalSslContext();
-            if (mqttConfig.getTlsCertRefreshCheckDurationSec() > 0) {
-                sslContextRefresher.scheduleWithFixedDelay(this::refreshSslContext,
-                        mqttConfig.getTlsCertRefreshCheckDurationSec(),
-                        mqttConfig.getTlsCertRefreshCheckDurationSec(), TimeUnit.SECONDS);
-            }
+            this.tlsFactory = new MQTTTlsFactory(mqttConfig, TlsPurpose.BROKER, sslContextRefresher);
         }
     }
 
     @Override
     public void initChannel(SocketChannel ch) throws Exception {
         ch.pipeline().addFirst("idleStateHandler", new IdleStateHandler(0, 0, 120));
         if (this.enableTls) {
-            ch.pipeline().addLast(TLS_HANDLER, new SslHandler(sslFactory.createServerSslEngine(ch.alloc())));
+            ch.pipeline().addLast(TLS_HANDLER, tlsFactory.newServerSslHandler(ch.alloc()));
         } else if (this.enableTlsPsk) {
             ch.pipeline().addLast(TLS_HANDLER,
                     new SslHandler(PSKUtils.createServerEngine(ch, mqttService.getPskConfiguration())));
@@ -95,35 +83,4 @@ public void initChannel(SocketChannel ch) throws Exception {
         ch.pipeline().addLast(MQTTBrokerInboundHandler.NAME, new MQTTBrokerInboundHandler(mqttService));
     }
 
-    protected PulsarSslConfiguration buildSslConfiguration(MQTTServerConfiguration config) {
-        return PulsarSslConfiguration.builder()
-                .tlsProvider(config.getMqttTlsProvider())
-                .tlsKeyStoreType(config.getMqttTlsKeyStoreType())
-                .tlsKeyStorePath(config.getMqttTlsKeyStore())
-                .tlsKeyStorePassword(config.getMqttTlsKeyStorePassword())
-                .tlsTrustStoreType(config.getMqttTlsTrustStoreType())
-                .tlsTrustStorePath(config.getMqttTlsTrustStore())
-                .tlsTrustStorePassword(config.getMqttTlsTrustStorePassword())
-                .tlsCiphers(config.getMqttTlsCiphers())
-                .tlsProtocols(config.getMqttTlsProtocols())
-                .tlsTrustCertsFilePath(config.getMqttTlsTrustCertsFilePath())
-                .tlsCertificateFilePath(config.getMqttTlsCertificateFilePath())
-                .tlsKeyFilePath(config.getMqttTlsKeyFilePath())
-                .allowInsecureConnection(config.isMqttTlsAllowInsecureConnection())
-                .requireTrustedClientCertOnConnect(config.isMqttTlsRequireTrustedClientCertOnConnect())
-                .tlsEnabledWithKeystore(config.isMqttTlsEnabledWithKeyStore())
-                .tlsCustomParams(config.getSslFactoryPluginParams())
-                .authData(null)
-                .serverMode(true)
-                .build();
-    }
-
-    protected void refreshSslContext() {
-        try {
-            this.sslFactory.update();
-        } catch (Exception e) {
-            log.error("Failed to refresh SSL context for mqtt channel.", e);
-        }
-    }
-
 }

mqtt-common/pom.xml

@@ -30,6 +30,18 @@
             <groupId>io.streamnative</groupId>
             <artifactId>pulsar-client-original</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.streamnative</groupId>
+            <artifactId>pulsar-tls-factory-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.streamnative</groupId>
+            <artifactId>pulsar-common</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.streamnative</groupId>
+            <artifactId>pulsar-broker-common</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.apache.bookkeeper</groupId>
             <artifactId>bookkeeper-common</artifactId>
@@ -49,7 +61,6 @@
         <dependency>
             <groupId>javax.validation</groupId>
             <artifactId>validation-api</artifactId>
-            <version>${validation-api.version}</version>
         </dependency>
         <dependency>
             <groupId>com.github.ben-manes.caffeine</groupId>
@@ -58,7 +69,6 @@
         <dependency>
             <groupId>org.conscrypt</groupId>
             <artifactId>conscrypt-openjdk-uber</artifactId>
-            <version>${conscrypt.version}</version>
         </dependency>
     </dependencies>
 </project>

mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/tls/MQTTTlsFactory.java

@@ -0,0 +1,124 @@
+/**
+ * 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.
+ */
+package io.streamnative.pulsar.handlers.mqtt.common.tls;
+
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslHandler;
+import io.streamnative.pulsar.handlers.mqtt.common.MQTTCommonConfiguration;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ScheduledExecutorService;
+import org.apache.pulsar.broker.tls.TlsFactorySupport;
+import org.apache.pulsar.common.tls.impl.FileBasedTlsFactory;
+import org.apache.pulsar.common.tls.impl.FileBasedTlsFactorySettings;
+import org.apache.pulsar.common.tls.impl.TlsContextAcquisition;
+import org.apache.pulsar.common.tls.impl.TlsSynthesisSpec;
+import org.apache.pulsar.tls.PulsarTlsFactory;
+import org.apache.pulsar.tls.TlsFactoryInitContext;
+import org.apache.pulsar.tls.TlsHandle;
+import org.apache.pulsar.tls.TlsPolicy;
+import org.apache.pulsar.tls.TlsPurpose;
+
+/**
+ * Bridges MoP's mqttTls* settings to Pulsar's PIP-478 TLS factory SPI.
+ */
+public class MQTTTlsFactory implements AutoCloseable {
+
+    private PulsarTlsFactory tlsFactory;
+    private TlsHandle<SslContext> tlsSubscription;
+    private volatile SslContext tlsServerContext;
+
+    public MQTTTlsFactory(MQTTCommonConfiguration config, TlsPurpose purpose,
+                          ScheduledExecutorService sslContextRefresher) throws Exception {
+        this.tlsFactory = TlsFactorySupport.createFactory(config.getTlsFactoryClassName(),
+                FileBasedTlsFactory.class,
+                () -> createDefaultFactory(config, purpose));
+        try {
+            TlsFactoryInitContext initContext = TlsFactorySupport.initContext(
+                    TlsFactorySupport.parseFactoryConfig(config.getTlsFactoryConfig()),
+                    sslContextRefresher, sslContextRefresher);
+            TlsFactorySupport.initializeBlocking(this.tlsFactory, initContext);
+            this.tlsSubscription = TlsContextAcquisition.acquireNettyContext(this.tlsFactory, purpose,
+                            TlsSynthesisSpec.server(config.isMqttTlsRequireTrustedClientCertOnConnect()),
+                            context -> this.tlsServerContext = context)
+                    .get()
+                    .orElseThrow(() -> new IllegalStateException(
+                            "TLS factory supplied no Netty SslContext for purpose " + purpose));
+        } catch (Exception e) {
+            close();
+            throw e;
+        }
+    }
+
+    public SslHandler newServerSslHandler(ByteBufAllocator allocator) {
+        return TlsContextAcquisition.withPinnedContext(
+                () -> this.tlsServerContext, context -> context.newHandler(allocator));
+    }
+
+    @Override
+    public void close() {
+        TlsHandle<SslContext> subscription = this.tlsSubscription;
+        if (subscription != null) {
+            this.tlsSubscription = null;
+            subscription.dispose();
+        }
+        PulsarTlsFactory factory = this.tlsFactory;
+        if (factory != null) {
+            this.tlsFactory = null;
+            factory.close();
+        }
+    }
+
+    private static PulsarTlsFactory createDefaultFactory(MQTTCommonConfiguration config, TlsPurpose purpose) {
+        Map<TlsPurpose, TlsPolicy> policies = Map.of(purpose, createPolicy(config));
+        FileBasedTlsFactorySettings settings = FileBasedTlsFactorySettings.builder()
+                .requireTrustedClientCert(config.isMqttTlsRequireTrustedClientCertOnConnect())
+                .refreshIntervalSeconds(FileBasedTlsFactorySettings.refreshIntervalSecondsFromConfig(
+                        config.getMqttTlsCertRefreshCheckDurationSec()))
+                .engineProvider(TlsFactorySupport.engineProvider(config.getMqttTlsProvider()))
+                .build();
+        return new FileBasedTlsFactory(policies, settings);
+    }
+
+    private static TlsPolicy createPolicy(MQTTCommonConfiguration config) {
+        TlsPolicy.Builder builder = TlsPolicy.builder()
+                .allowInsecureConnection(config.isMqttTlsAllowInsecureConnection())
+                .enableHostnameVerification(config.isTlsHostnameVerificationEnabled())
+                .protocols(toList(config.getMqttTlsProtocols()))
+                .ciphers(toList(config.getMqttTlsCiphers()))
+                .jsseProvider(TlsFactorySupport.resolveJsseProvider(config.getJsseProvider(),
+                        config.getMqttTlsProvider()));
+        if (config.isMqttTlsEnabledWithKeyStore()) {
+            builder.format(TlsPolicy.Format.KEYSTORE)
+                    .keyStoreType(config.getMqttTlsKeyStoreType())
+                    .trustStoreType(config.getMqttTlsTrustStoreType())
+                    .keyStorePath(config.getMqttTlsKeyStore())
+                    .keyStorePassword(config.getMqttTlsKeyStorePassword())
+                    .trustStorePath(config.getMqttTlsTrustStore())
+                    .trustStorePassword(config.getMqttTlsTrustStorePassword());
+        } else {
+            builder.format(TlsPolicy.Format.PEM)
+                    .trustCertsFilePath(config.getMqttTlsTrustCertsFilePath())
+                    .certificateFilePath(config.getMqttTlsCertificateFilePath())
+                    .keyFilePath(config.getMqttTlsKeyFilePath());
+        }
+        return builder.build();
+    }
+
+    private static List<String> toList(Set<String> values) {
+        return values == null ? List.of() : List.copyOf(values);
+    }
+}

mqtt-common/src/main/java/io/streamnative/pulsar/handlers/mqtt/common/tls/package-info.java

@@ -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.
+ */
+
+/**
+ * TLS support for MQTT protocol handlers.
+ */
+package io.streamnative.pulsar.handlers.mqtt.common.tls;

mqtt-proxy/pom.xml

@@ -43,7 +43,6 @@
         <dependency>
             <groupId>org.eclipse.jetty.ee10</groupId>
             <artifactId>jetty-ee10-servlet</artifactId>
-            <version>${jetty.ee10.version}</version>
         </dependency>
         <dependency>
             <groupId>org.glassfish.jersey.core</groupId>
@@ -52,17 +51,14 @@
         <dependency>
             <groupId>org.glassfish.jersey.containers</groupId>
             <artifactId>jersey-container-servlet-core</artifactId>
-            <version>${jersey.version}</version>
         </dependency>
         <dependency>
             <groupId>org.glassfish.jersey.media</groupId>
             <artifactId>jersey-media-multipart</artifactId>
-            <version>${jersey.version}</version>
         </dependency>
         <dependency>
             <groupId>io.swagger</groupId>
             <artifactId>swagger-annotations</artifactId>
-            <version>${swagger-annotations.version}</version>
         </dependency>
         <dependency>
             <groupId>jakarta.ws.rs</groupId>

mqtt-proxy/src/main/java/io/streamnative/pulsar/handlers/mqtt/proxy/channel/MQTTProxyChannelInitializer.java

@@ -23,21 +23,18 @@
 import io.streamnative.pulsar.handlers.mqtt.common.adapter.MqttAdapterDecoder;
 import io.streamnative.pulsar.handlers.mqtt.common.adapter.MqttAdapterEncoder;
 import io.streamnative.pulsar.handlers.mqtt.common.psk.PSKUtils;
+import io.streamnative.pulsar.handlers.mqtt.common.tls.MQTTTlsFactory;
 import io.streamnative.pulsar.handlers.mqtt.common.utils.WebSocketUtils;
 import io.streamnative.pulsar.handlers.mqtt.proxy.MQTTProxyConfiguration;
 import io.streamnative.pulsar.handlers.mqtt.proxy.MQTTProxyService;
 import io.streamnative.pulsar.handlers.mqtt.proxy.impl.MQTTProxyException;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
 import lombok.Getter;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.common.util.PulsarSslConfiguration;
-import org.apache.pulsar.common.util.PulsarSslFactory;
+import org.apache.pulsar.tls.TlsPurpose;
 
 /**
  * Proxy service channel initializer.
  */
-@Slf4j
 public class MQTTProxyChannelInitializer extends ChannelInitializer<SocketChannel> {
 
     private final MQTTProxyService proxyService;
@@ -47,7 +44,7 @@ public class MQTTProxyChannelInitializer extends ChannelInitializer<SocketChanne
     private final boolean enableTls;
     private final boolean enableTlsPsk;
     private final boolean enableWs;
-    private PulsarSslFactory sslFactory;
+    private MQTTTlsFactory tlsFactory;
 
     public MQTTProxyChannelInitializer(MQTTProxyService proxyService, MQTTProxyConfiguration proxyConfig,
                                        boolean enableTls, boolean enableWs,
@@ -65,17 +62,7 @@ public MQTTProxyChannelInitializer(MQTTProxyService proxyService, MQTTProxyConfi
             this.enableTlsPsk = enableTlsPsk;
             this.enableWs = enableWs;
             if (this.enableTls) {
-                PulsarSslConfiguration sslConfiguration = buildSslConfiguration(proxyConfig);
-                this.sslFactory = (PulsarSslFactory) Class.forName(proxyConfig.getSslFactoryPlugin())
-                        .getConstructor().newInstance();
-                this.sslFactory.initialize(sslConfiguration);
-                this.sslFactory.createInternalSslContext();
-                if (proxyConfig.getTlsCertRefreshCheckDurationSec() > 0) {
-                    sslContextRefresher.scheduleWithFixedDelay(this::refreshSslContext,
-                            proxyConfig.getTlsCertRefreshCheckDurationSec(),
-                            proxyConfig.getTlsCertRefreshCheckDurationSec(), TimeUnit.SECONDS);
-
-                }
+                this.tlsFactory = new MQTTTlsFactory(proxyConfig, TlsPurpose.PROXY, sslContextRefresher);
             }
         } catch (Exception e) {
             throw new MQTTProxyException(e);
@@ -86,7 +73,7 @@ public MQTTProxyChannelInitializer(MQTTProxyService proxyService, MQTTProxyConfi
     protected void initChannel(SocketChannel ch) throws Exception {
         ch.pipeline().addFirst("idleStateHandler", new IdleStateHandler(30, 0, 0));
         if (this.enableTls) {
-            ch.pipeline().addLast(TLS_HANDLER, new SslHandler(sslFactory.createServerSslEngine(ch.alloc())));
+            ch.pipeline().addLast(TLS_HANDLER, tlsFactory.newServerSslHandler(ch.alloc()));
         } else if (this.enableTlsPsk) {
             ch.pipeline().addLast(TLS_HANDLER,
                     new SslHandler(PSKUtils.createServerEngine(ch, proxyService.getPskConfiguration())));
@@ -103,35 +90,4 @@ protected void initChannel(SocketChannel ch) throws Exception {
         ch.pipeline().addLast("handler", new MQTTProxyInboundHandler(proxyService));
     }
 
-    protected PulsarSslConfiguration buildSslConfiguration(MQTTProxyConfiguration config) {
-        return PulsarSslConfiguration.builder()
-                .tlsProvider(config.getMqttTlsProvider())
-                .tlsKeyStoreType(config.getMqttTlsKeyStoreType())
-                .tlsKeyStorePath(config.getMqttTlsKeyStore())
-                .tlsKeyStorePassword(config.getMqttTlsKeyStorePassword())
-                .tlsTrustStoreType(config.getMqttTlsTrustStoreType())
-                .tlsTrustStorePath(config.getMqttTlsTrustStore())
-                .tlsTrustStorePassword(config.getMqttTlsTrustStorePassword())
-                .tlsCiphers(config.getMqttTlsCiphers())
-                .tlsProtocols(config.getMqttTlsProtocols())
-                .tlsTrustCertsFilePath(config.getMqttTlsTrustCertsFilePath())
-                .tlsCertificateFilePath(config.getMqttTlsCertificateFilePath())
-                .tlsKeyFilePath(config.getMqttTlsKeyFilePath())
-                .allowInsecureConnection(config.isMqttTlsAllowInsecureConnection())
-                .requireTrustedClientCertOnConnect(config.isMqttTlsRequireTrustedClientCertOnConnect())
-                .tlsEnabledWithKeystore(config.isMqttTlsEnabledWithKeyStore())
-                .tlsCustomParams(config.getSslFactoryPluginParams())
-                .authData(null)
-                .serverMode(true)
-                .build();
-    }
-
-    protected void refreshSslContext() {
-        try {
-            this.sslFactory.update();
-        } catch (Exception e) {
-            log.error("Failed to refresh SSL context for mqtt proxy channel.", e);
-        }
-    }
-
 }