This is an automated email from the ASF dual-hosted git repository.
pvillard31 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new f3a4fe33039 NIFI-16212 Switch MQTT Processors to SSLContextProvider
(#11553)
f3a4fe33039 is described below
commit f3a4fe330394a4af2c5d5d54f7910744e1d4fcd4
Author: David Handermann <[email protected]>
AuthorDate: Sat Aug 15 15:26:59 2026 -0500
NIFI-16212 Switch MQTT Processors to SSLContextProvider (#11553)
---
.../nifi-mqtt-bundle/nifi-mqtt-processors/pom.xml | 12 +
.../mqtt/adapters/HiveMqV5ClientAdapter.java | 61 ++---
.../mqtt/adapters/PahoMqttClientAdapter.java | 35 +--
.../mqtt/adapters/PredefinedKeyManagerFactory.java | 60 +++++
.../mqtt/adapters/PredefinedSecurityProvider.java | 37 +++
.../adapters/PredefinedTrustManagerFactory.java | 61 +++++
.../mqtt/common/AbstractMQTTProcessor.java | 12 +-
.../mqtt/common/MqttClientProperties.java | 12 +-
.../nifi/processors/mqtt/ConsumeMQTTTLSTest.java | 275 +++++++++++++++++++++
.../nifi/processors/mqtt/TestConsumeMQTT.java | 12 +-
10 files changed, 484 insertions(+), 93 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/pom.xml
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/pom.xml
index 521ba43abfe..ca4a6d9ae90 100644
--- a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/pom.xml
+++ b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/pom.xml
@@ -89,5 +89,17 @@
<artifactId>jackson-databind</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>com.hivemq</groupId>
+ <artifactId>hivemq-community-edition-embedded</artifactId>
+ <version>2026.5</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-security-cert-builder</artifactId>
+ <version>2.12.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/HiveMqV5ClientAdapter.java
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/HiveMqV5ClientAdapter.java
index 475ffd9839f..9aa927f282d 100644
---
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/HiveMqV5ClientAdapter.java
+++
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/HiveMqV5ClientAdapter.java
@@ -31,23 +31,19 @@ import
org.apache.nifi.processors.mqtt.common.MqttProtocolScheme;
import org.apache.nifi.processors.mqtt.common.ReceivedMqttMessage;
import org.apache.nifi.processors.mqtt.common.ReceivedMqttMessageHandler;
import org.apache.nifi.processors.mqtt.common.StandardMqttMessage;
-import org.apache.nifi.security.ssl.StandardKeyManagerFactoryBuilder;
-import org.apache.nifi.security.ssl.StandardKeyStoreBuilder;
-import org.apache.nifi.security.ssl.StandardTrustManagerFactoryBuilder;
-import org.apache.nifi.security.util.TlsConfiguration;
import org.apache.nifi.security.util.TlsException;
+import org.apache.nifi.ssl.SSLContextProvider;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStream;
import java.net.URI;
import java.nio.charset.StandardCharsets;
-import java.security.KeyStore;
import java.util.Objects;
+import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509ExtendedKeyManager;
+import javax.net.ssl.X509TrustManager;
import static org.apache.nifi.processors.mqtt.common.MqttProtocolScheme.SSL;
import static org.apache.nifi.processors.mqtt.common.MqttProtocolScheme.WS;
@@ -176,43 +172,22 @@ public class HiveMqV5ClientAdapter implements MqttClient {
}
if (SSL.equals(scheme) || WSS.equals(scheme)) {
- final TlsConfiguration tlsConfiguration =
clientProperties.getTlsConfiguration();
-
- if (tlsConfiguration.getTruststorePath() != null) {
- final KeyStore trustStore;
- try (InputStream inputStream = new
FileInputStream(tlsConfiguration.getTruststorePath())) {
- trustStore = new StandardKeyStoreBuilder()
-
.type(tlsConfiguration.getTruststoreType().getType())
-
.password(tlsConfiguration.getTruststorePassword().toCharArray())
- .inputStream(inputStream)
- .build();
- } catch (final IOException e) {
- throw new TlsException("Trust Store loading failed", e);
- }
-
- final TrustManagerFactory trustManagerFactory = new
StandardTrustManagerFactoryBuilder().trustStore(trustStore).build();
- mqtt5ClientBuilder
- .sslConfig()
- .trustManagerFactory(trustManagerFactory)
- .applySslConfig();
+ final SSLContextProvider sslContextProvider =
clientProperties.getSslContextProvider();
+
+ if (sslContextProvider == null) {
+ throw new TlsException("SSL Context Provider not configured
for Broker URI scheme requiring TLS communication: " + scheme);
}
- if (tlsConfiguration.getKeystorePath() != null) {
- final KeyStore keyStore;
- try (InputStream inputStream = new
FileInputStream(tlsConfiguration.getKeystorePath())) {
- keyStore = new StandardKeyStoreBuilder()
- .type(tlsConfiguration.getKeystoreType().getType())
-
.password(tlsConfiguration.getKeystorePassword().toCharArray())
- .inputStream(inputStream)
- .build();
- } catch (final IOException e) {
- throw new TlsException("Key Store loading failed", e);
- }
-
- final KeyManagerFactory keyManagerFactory = new
StandardKeyManagerFactoryBuilder()
- .keyStore(keyStore)
-
.keyPassword(tlsConfiguration.getFunctionalKeyPassword().toCharArray())
- .build();
+ final X509TrustManager trustManager =
sslContextProvider.createTrustManager();
+ final TrustManagerFactory trustManagerFactory = new
PredefinedTrustManagerFactory(trustManager);
+ mqtt5ClientBuilder
+ .sslConfig()
+ .trustManagerFactory(trustManagerFactory)
+ .applySslConfig();
+
+ final Optional<X509ExtendedKeyManager> keyManagerFound =
sslContextProvider.createKeyManager();
+ if (keyManagerFound.isPresent()) {
+ final KeyManagerFactory keyManagerFactory = new
PredefinedKeyManagerFactory(keyManagerFound.get());
mqtt5ClientBuilder
.sslConfig()
.keyManagerFactory(keyManagerFactory)
diff --git
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PahoMqttClientAdapter.java
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PahoMqttClientAdapter.java
index 9d13d206a30..ddef128eb94 100644
---
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PahoMqttClientAdapter.java
+++
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PahoMqttClientAdapter.java
@@ -23,7 +23,7 @@ import org.apache.nifi.processors.mqtt.common.MqttException;
import org.apache.nifi.processors.mqtt.common.ReceivedMqttMessage;
import org.apache.nifi.processors.mqtt.common.ReceivedMqttMessageHandler;
import org.apache.nifi.processors.mqtt.common.StandardMqttMessage;
-import org.apache.nifi.security.util.TlsConfiguration;
+import org.apache.nifi.ssl.SSLContextProvider;
import org.eclipse.paho.client.mqttv3.IMqttClient;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
@@ -33,7 +33,6 @@ import
org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import java.net.URI;
import java.util.Arrays;
-import java.util.Properties;
public class PahoMqttClientAdapter implements MqttClient {
@@ -67,9 +66,9 @@ public class PahoMqttClientAdapter implements MqttClient {
connectOptions.setMqttVersion(clientProperties.getMqttVersion().getVersionCode());
connectOptions.setConnectionTimeout(clientProperties.getConnectionTimeout());
- final TlsConfiguration tlsConfiguration =
clientProperties.getTlsConfiguration();
- if (tlsConfiguration != null) {
-
connectOptions.setSSLProperties(transformSSLContextService(tlsConfiguration));
+ final SSLContextProvider sslContextProvider =
clientProperties.getSslContextProvider();
+ if (sslContextProvider != null) {
+
connectOptions.setSocketFactory(sslContextProvider.createContext().getSocketFactory());
}
final String lastWillTopic = clientProperties.getLastWillTopic();
@@ -136,32 +135,6 @@ public class PahoMqttClientAdapter implements MqttClient {
}
}
- public static Properties transformSSLContextService(TlsConfiguration
tlsConfiguration) {
- final Properties properties = new Properties();
- if (tlsConfiguration.getProtocol() != null) {
- properties.setProperty("com.ibm.ssl.protocol",
tlsConfiguration.getProtocol());
- }
- if (tlsConfiguration.getKeystorePath() != null) {
- properties.setProperty("com.ibm.ssl.keyStore",
tlsConfiguration.getKeystorePath());
- }
- if (tlsConfiguration.getKeystorePassword() != null) {
- properties.setProperty("com.ibm.ssl.keyStorePassword",
tlsConfiguration.getKeystorePassword());
- }
- if (tlsConfiguration.getKeystoreType() != null) {
- properties.setProperty("com.ibm.ssl.keyStoreType",
tlsConfiguration.getKeystoreType().getType());
- }
- if (tlsConfiguration.getTruststorePath() != null) {
- properties.setProperty("com.ibm.ssl.trustStore",
tlsConfiguration.getTruststorePath());
- }
- if (tlsConfiguration.getTruststorePassword() != null) {
- properties.setProperty("com.ibm.ssl.trustStorePassword",
tlsConfiguration.getTruststorePassword());
- }
- if (tlsConfiguration.getTruststoreType() != null) {
- properties.setProperty("com.ibm.ssl.trustStoreType",
tlsConfiguration.getTruststoreType().getType());
- }
- return properties;
- }
-
private static org.eclipse.paho.client.mqttv3.MqttClient createClient(URI
brokerUri, MqttClientProperties clientProperties, ComponentLog logger) {
logger.debug("Creating Mqtt v3 client");
diff --git
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PredefinedKeyManagerFactory.java
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PredefinedKeyManagerFactory.java
new file mode 100644
index 00000000000..16baf194d63
--- /dev/null
+++
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PredefinedKeyManagerFactory.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.nifi.processors.mqtt.adapters;
+
+import java.security.KeyStore;
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.KeyManagerFactorySpi;
+import javax.net.ssl.ManagerFactoryParameters;
+import javax.net.ssl.X509ExtendedKeyManager;
+
+/**
+ * Key Manager Factory that returns a single Key Manager provided at
construction time, instead of loading
+ * a Key Store. This bridges client libraries whose SSL configuration requires
a KeyManagerFactory with
+ * SSLContextProvider implementations, which expose an already-initialized
X.509 Key Manager directly.
+ */
+public class PredefinedKeyManagerFactory extends KeyManagerFactory {
+
+ private static final String ALGORITHM = "PredefinedKeyManager";
+
+ public PredefinedKeyManagerFactory(final X509ExtendedKeyManager
keyManager) {
+ super(new PredefinedKeyManagerFactorySpi(keyManager), new
PredefinedSecurityProvider(), ALGORITHM);
+ }
+
+ private static class PredefinedKeyManagerFactorySpi extends
KeyManagerFactorySpi {
+
+ private final KeyManager[] keyManagers;
+
+ private PredefinedKeyManagerFactorySpi(final X509ExtendedKeyManager
keyManager) {
+ this.keyManagers = new KeyManager[]{keyManager};
+ }
+
+ @Override
+ protected void engineInit(final KeyStore keyStore, final char[]
password) {
+ }
+
+ @Override
+ protected void engineInit(final ManagerFactoryParameters
managerFactoryParameters) {
+ }
+
+ @Override
+ protected KeyManager[] engineGetKeyManagers() {
+ return keyManagers;
+ }
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PredefinedSecurityProvider.java
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PredefinedSecurityProvider.java
new file mode 100644
index 00000000000..8e97d7b3f0f
--- /dev/null
+++
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PredefinedSecurityProvider.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.nifi.processors.mqtt.adapters;
+
+import java.security.Provider;
+
+/**
+ * Placeholder security Provider required by the Key Manager Factory and Trust
Manager Factory constructors.
+ * PredefinedKeyManagerFactory and PredefinedTrustManagerFactory return Key
Managers and Trust Managers that
+ * were already built elsewhere, so no actual provider-specific algorithm
implementation is registered here.
+ */
+class PredefinedSecurityProvider extends Provider {
+
+ private static final String NAME = "PredefinedSecurityProvider";
+
+ private static final String VERSION = "1.0";
+
+ private static final String INFO = "Provider for Key Manager Factory and
Trust Manager Factory implementations backed by predefined Managers";
+
+ PredefinedSecurityProvider() {
+ super(NAME, VERSION, INFO);
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PredefinedTrustManagerFactory.java
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PredefinedTrustManagerFactory.java
new file mode 100644
index 00000000000..7ddbebbc8ad
--- /dev/null
+++
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/adapters/PredefinedTrustManagerFactory.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.nifi.processors.mqtt.adapters;
+
+import java.security.KeyStore;
+import javax.net.ssl.ManagerFactoryParameters;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.TrustManagerFactorySpi;
+import javax.net.ssl.X509TrustManager;
+
+/**
+ * Trust Manager Factory that returns a single Trust Manager provided at
construction time, instead of
+ * loading a Key Store. This bridges client libraries whose SSL configuration
requires a
+ * TrustManagerFactory with SSLContextProvider implementations, which expose
an already-initialized
+ * X.509 Trust Manager directly.
+ */
+public class PredefinedTrustManagerFactory extends TrustManagerFactory {
+
+ private static final String ALGORITHM = "PredefinedTrustManager";
+
+ public PredefinedTrustManagerFactory(final X509TrustManager trustManager) {
+ super(new PredefinedTrustManagerFactorySpi(trustManager), new
PredefinedSecurityProvider(), ALGORITHM);
+ }
+
+ private static class PredefinedTrustManagerFactorySpi extends
TrustManagerFactorySpi {
+
+ private final TrustManager[] trustManagers;
+
+ private PredefinedTrustManagerFactorySpi(final X509TrustManager
trustManager) {
+ this.trustManagers = new TrustManager[]{trustManager};
+ }
+
+ @Override
+ protected void engineInit(final KeyStore keyStore) {
+ }
+
+ @Override
+ protected void engineInit(final ManagerFactoryParameters
managerFactoryParameters) {
+ }
+
+ @Override
+ protected TrustManager[] engineGetTrustManagers() {
+ return trustManagers;
+ }
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/common/AbstractMQTTProcessor.java
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/common/AbstractMQTTProcessor.java
index b820fe262a9..b8991f8e665 100644
---
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/common/AbstractMQTTProcessor.java
+++
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/common/AbstractMQTTProcessor.java
@@ -35,7 +35,7 @@ import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.security.util.TlsException;
import org.apache.nifi.serialization.RecordReaderFactory;
import org.apache.nifi.serialization.RecordSetWriterFactory;
-import org.apache.nifi.ssl.SSLContextService;
+import org.apache.nifi.ssl.SSLContextProvider;
import java.net.URI;
import java.net.URISyntaxException;
@@ -143,9 +143,9 @@ public abstract class AbstractMQTTProcessor extends
AbstractSessionFactoryProces
public static final PropertyDescriptor PROP_SSL_CONTEXT_SERVICE = new
PropertyDescriptor.Builder()
.name("SSL Context Service")
- .description("The SSL Context Service used to provide client
certificate information for TLS/SSL connections.")
+ .description("The SSL Context Provider used to provide client
certificate information for TLS connections.")
.required(false)
- .identifiesControllerService(SSLContextService.class)
+ .identifiesControllerService(SSLContextProvider.class)
.build();
public static final PropertyDescriptor PROP_LAST_WILL_MESSAGE = new
PropertyDescriptor.Builder()
@@ -387,10 +387,8 @@ public abstract class AbstractMQTTProcessor extends
AbstractSessionFactoryProces
clientProperties.setKeepAliveInterval(context.getProperty(PROP_KEEP_ALIVE_INTERVAL).asInteger());
clientProperties.setConnectionTimeout(context.getProperty(PROP_CONN_TIMEOUT).asInteger());
- final SSLContextService sslContextService =
context.getProperty(PROP_SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class);
- if (sslContextService != null) {
-
clientProperties.setTlsConfiguration(sslContextService.createTlsConfiguration());
- }
+ final SSLContextProvider sslContextProvider =
context.getProperty(PROP_SSL_CONTEXT_SERVICE).asControllerService(SSLContextProvider.class);
+ clientProperties.setSslContextProvider(sslContextProvider);
if (context.getProperty(PROP_LAST_WILL_MESSAGE).isSet()) {
clientProperties.setLastWillMessage(context.getProperty(PROP_LAST_WILL_MESSAGE).getValue());
diff --git
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/common/MqttClientProperties.java
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/common/MqttClientProperties.java
index 5710c9c9a76..5bfe465b3b1 100644
---
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/common/MqttClientProperties.java
+++
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/common/MqttClientProperties.java
@@ -16,7 +16,7 @@
*/
package org.apache.nifi.processors.mqtt.common;
-import org.apache.nifi.security.util.TlsConfiguration;
+import org.apache.nifi.ssl.SSLContextProvider;
import java.net.URI;
import java.util.List;
@@ -35,7 +35,7 @@ public class MqttClientProperties {
private boolean cleanSession;
private Long sessionExpiryInterval;
- private TlsConfiguration tlsConfiguration;
+ private SSLContextProvider sslContextProvider;
private String lastWillTopic;
private String lastWillMessage;
@@ -109,12 +109,12 @@ public class MqttClientProperties {
this.sessionExpiryInterval = sessionExpiryInterval;
}
- public TlsConfiguration getTlsConfiguration() {
- return tlsConfiguration;
+ public SSLContextProvider getSslContextProvider() {
+ return sslContextProvider;
}
- public void setTlsConfiguration(TlsConfiguration tlsConfiguration) {
- this.tlsConfiguration = tlsConfiguration;
+ public void setSslContextProvider(SSLContextProvider sslContextProvider) {
+ this.sslContextProvider = sslContextProvider;
}
public String getLastWillTopic() {
diff --git
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/ConsumeMQTTTLSTest.java
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/ConsumeMQTTTLSTest.java
new file mode 100644
index 00000000000..094d658d50b
--- /dev/null
+++
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/ConsumeMQTTTLSTest.java
@@ -0,0 +1,275 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.nifi.processors.mqtt;
+
+import com.hivemq.embedded.EmbeddedHiveMQ;
+import org.apache.nifi.reporting.InitializationException;
+import org.apache.nifi.security.cert.builder.StandardCertificateBuilder;
+import org.apache.nifi.security.ssl.EphemeralKeyStoreBuilder;
+import org.apache.nifi.security.ssl.StandardKeyManagerBuilder;
+import org.apache.nifi.security.ssl.StandardSslContextBuilder;
+import org.apache.nifi.security.ssl.StandardTrustManagerBuilder;
+import org.apache.nifi.ssl.SSLContextProvider;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.OutputStream;
+import java.net.ServerSocket;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.KeyPair;
+import java.security.KeyPairGenerator;
+import java.security.KeyStore;
+import java.security.cert.Certificate;
+import java.security.cert.X509Certificate;
+import java.time.Duration;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.X509ExtendedKeyManager;
+import javax.net.ssl.X509TrustManager;
+import javax.security.auth.x500.X500Principal;
+
+import static
org.apache.nifi.processors.mqtt.common.MqttConstants.ALLOWABLE_VALUE_MQTT_VERSION_311;
+import static
org.apache.nifi.processors.mqtt.common.MqttConstants.ALLOWABLE_VALUE_MQTT_VERSION_500;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class ConsumeMQTTTLSTest {
+
+ private static final String LOCALHOST = "localhost";
+
+ private static final String KEY_ALGORITHM = "RSA";
+
+ private static final X500Principal CERTIFICATE_PRINCIPAL = new
X500Principal("CN=%s".formatted(LOCALHOST));
+
+ private static final char[] BROKER_KEY_STORE_PASSWORD =
UUID.randomUUID().toString().toCharArray();
+
+ private static final char[] CLIENT_KEY_PASSWORD =
UUID.randomUUID().toString().toCharArray();
+
+ private static final Duration CERTIFICATE_VALIDITY = Duration.ofHours(1);
+
+ private static final long BROKER_LIFECYCLE_TIMEOUT_SECONDS = 30;
+
+ private static final long MESSAGE_TIMEOUT_SECONDS = 15;
+
+ private static final long POLL_INTERVAL_MILLIS = 100;
+
+ private static final String INTERNAL_QUEUE_SIZE = "100";
+
+ private static final String CONNECTION_TIMEOUT_SECONDS = "10";
+
+ private static final String TOPIC_MQTT_3 = "mqtt-3";
+
+ private static final String TOPIC_MQTT_5 = "mqtt-5";
+
+ @TempDir
+ private static Path tempDir;
+
+ private static EmbeddedHiveMQ embeddedHiveMQ;
+
+ private static int tlsPort;
+
+ private static SSLContext clientSslContext;
+
+ private static X509TrustManager clientTrustManager;
+
+ private static X509ExtendedKeyManager clientKeyManager;
+
+ @BeforeAll
+ static void startBroker() throws Exception {
+ tlsPort = getAvailablePort();
+
+ final KeyPair keyPair =
KeyPairGenerator.getInstance(KEY_ALGORITHM).generateKeyPair();
+ final X509Certificate certificate = new
StandardCertificateBuilder(keyPair, CERTIFICATE_PRINCIPAL,
CERTIFICATE_VALIDITY).build();
+
+ final KeyStore ephemeralKeyStore = new EphemeralKeyStoreBuilder()
+ .addPrivateKeyEntry(new
KeyStore.PrivateKeyEntry(keyPair.getPrivate(), new Certificate[]{certificate}))
+ .keyPassword(CLIENT_KEY_PASSWORD)
+ .build();
+ clientSslContext = new StandardSslContextBuilder()
+ .trustStore(ephemeralKeyStore)
+ .keyStore(ephemeralKeyStore)
+ .keyPassword(CLIENT_KEY_PASSWORD)
+ .build();
+ clientTrustManager = new
StandardTrustManagerBuilder().trustStore(ephemeralKeyStore).build();
+ clientKeyManager = new
StandardKeyManagerBuilder().keyStore(ephemeralKeyStore).keyPassword(CLIENT_KEY_PASSWORD).build();
+
+ final Path brokerKeyStorePath = writeBrokerKeyStore(keyPair,
certificate);
+ final Path configFolder =
Files.createDirectories(tempDir.resolve("conf"));
+ final Path dataFolder =
Files.createDirectories(tempDir.resolve("data"));
+ final Path extensionsFolder =
Files.createDirectories(tempDir.resolve("extensions"));
+ Files.writeString(configFolder.resolve("config.xml"),
getBrokerConfiguration(brokerKeyStorePath));
+
+ embeddedHiveMQ = EmbeddedHiveMQ.builder()
+ .withConfigurationFolder(configFolder)
+ .withDataFolder(dataFolder)
+ .withExtensionsFolder(extensionsFolder)
+ .withoutLoggingBootstrap()
+ .build();
+ embeddedHiveMQ.start().get(BROKER_LIFECYCLE_TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ }
+
+ @AfterAll
+ static void stopBroker() throws Exception {
+ if (embeddedHiveMQ != null) {
+ embeddedHiveMQ.stop().get(BROKER_LIFECYCLE_TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ embeddedHiveMQ.close();
+ }
+ }
+
+ @Test
+ void testConsumeOverTlsUsingMqttVersion3() throws Exception {
+ final TestRunner testRunner = createTestRunner(TOPIC_MQTT_3,
ALLOWABLE_VALUE_MQTT_VERSION_311.getValue(), "ConsumeMQTTTLSTest-3");
+
+ assertConsumesPublishedMessageOverTls(testRunner, TOPIC_MQTT_3);
+ }
+
+ @Test
+ void testConsumeOverTlsUsingMqttVersion5() throws Exception {
+ final TestRunner testRunner = createTestRunner(TOPIC_MQTT_5,
ALLOWABLE_VALUE_MQTT_VERSION_500.getValue(), "ConsumeMQTTTLSTest-5");
+
+ assertConsumesPublishedMessageOverTls(testRunner, TOPIC_MQTT_5);
+ }
+
+ private void assertConsumesPublishedMessageOverTls(final TestRunner
testRunner, final String topic) throws Exception {
+ // Establish the TLS connection and subscription
+ testRunner.run(1, false, true);
+
+ final String payload = "message-%s".formatted(topic);
+ publishMessageOverTls(topic, payload);
+
+ final List<MockFlowFile> flowFiles = awaitFlowFiles(testRunner);
+ assertEquals(1, flowFiles.size());
+ flowFiles.getFirst().assertContentEquals(payload);
+ }
+
+ private List<MockFlowFile> awaitFlowFiles(final TestRunner testRunner)
throws InterruptedException {
+ final long deadline = System.currentTimeMillis() +
TimeUnit.SECONDS.toMillis(MESSAGE_TIMEOUT_SECONDS);
+ List<MockFlowFile> flowFiles;
+ do {
+ testRunner.run(1, false, false);
+ flowFiles =
testRunner.getFlowFilesForRelationship(ConsumeMQTT.REL_MESSAGE);
+ if (!flowFiles.isEmpty()) {
+ return flowFiles;
+ }
+
+ TimeUnit.MILLISECONDS.sleep(POLL_INTERVAL_MILLIS);
+ } while (System.currentTimeMillis() < deadline);
+ return flowFiles;
+ }
+
+ private TestRunner createTestRunner(final String topic, final String
mqttVersion, final String clientId) throws InitializationException {
+ final TestRunner testRunner =
TestRunners.newTestRunner(ConsumeMQTT.class);
+ testRunner.setProperty(ConsumeMQTT.PROP_BROKER_URI,
"ssl://%s:%d".formatted(LOCALHOST, tlsPort));
+ testRunner.setProperty(ConsumeMQTT.PROP_CLIENTID, clientId);
+ testRunner.setProperty(ConsumeMQTT.PROP_TOPIC_FILTER, topic);
+ testRunner.setProperty(ConsumeMQTT.PROP_MAX_QUEUE_SIZE,
INTERNAL_QUEUE_SIZE);
+ testRunner.setProperty(ConsumeMQTT.PROP_CONN_TIMEOUT,
CONNECTION_TIMEOUT_SECONDS);
+ testRunner.setProperty(ConsumeMQTT.PROP_MQTT_VERSION, mqttVersion);
+
+ final String sslContextServiceIdentifier =
"ssl-context-provider-%s".formatted(clientId);
+ final SSLContextProvider sslContextProvider =
mock(SSLContextProvider.class);
+
when(sslContextProvider.getIdentifier()).thenReturn(sslContextServiceIdentifier);
+ when(sslContextProvider.createContext()).thenReturn(clientSslContext);
+
when(sslContextProvider.createTrustManager()).thenReturn(clientTrustManager);
+
when(sslContextProvider.createKeyManager()).thenReturn(Optional.of(clientKeyManager));
+
+ testRunner.addControllerService(sslContextServiceIdentifier,
sslContextProvider);
+ testRunner.enableControllerService(sslContextProvider);
+ testRunner.setProperty(ConsumeMQTT.PROP_SSL_CONTEXT_SERVICE,
sslContextServiceIdentifier);
+
+ testRunner.assertValid();
+ return testRunner;
+ }
+
+ private void publishMessageOverTls(final String topic, final String
payload) throws Exception {
+ final MqttConnectOptions connectOptions = new MqttConnectOptions();
+ connectOptions.setSocketFactory(clientSslContext.getSocketFactory());
+
+ final MqttClient publisher = new MqttClient(
+ "ssl://%s:%d".formatted(LOCALHOST, tlsPort),
"ConsumeMQTTTLSTest-Publisher-%s".formatted(topic), new MemoryPersistence());
+ try {
+ publisher.connect(connectOptions);
+ final MqttMessage message = new
MqttMessage(payload.getBytes(StandardCharsets.UTF_8));
+ message.setQos(1);
+ publisher.publish(topic, message);
+ } finally {
+ publisher.disconnect();
+ publisher.close();
+ }
+ }
+
+ private static Path writeBrokerKeyStore(final KeyPair keyPair, final
X509Certificate certificate) throws Exception {
+ final KeyStore brokerKeyStore =
KeyStore.getInstance(KeyStore.getDefaultType());
+ brokerKeyStore.load(null);
+ brokerKeyStore.setKeyEntry("broker", keyPair.getPrivate(),
BROKER_KEY_STORE_PASSWORD, new Certificate[]{certificate});
+
+ final Path keyStorePath = tempDir.resolve("broker-keystore");
+ try (OutputStream outputStream = Files.newOutputStream(keyStorePath)) {
+ brokerKeyStore.store(outputStream, BROKER_KEY_STORE_PASSWORD);
+ }
+
+ return keyStorePath;
+ }
+
+ private static String getBrokerConfiguration(final Path keyStorePath) {
+ final String keyStorePassword = new String(BROKER_KEY_STORE_PASSWORD);
+ return """
+ <?xml version="1.0"?>
+ <hivemq>
+ <listeners>
+ <tls-tcp-listener>
+ <port>%d</port>
+ <bind-address>0.0.0.0</bind-address>
+ <tls>
+ <keystore>
+ <path>%s</path>
+ <password>%s</password>
+
<private-key-password>%s</private-key-password>
+ </keystore>
+ <truststore>
+ <path>%s</path>
+ <password>%s</password>
+ </truststore>
+
<client-authentication-mode>REQUIRED</client-authentication-mode>
+ </tls>
+ </tls-tcp-listener>
+ </listeners>
+ </hivemq>
+ """.formatted(tlsPort, keyStorePath, keyStorePassword,
keyStorePassword, keyStorePath, keyStorePassword);
+ }
+
+ private static int getAvailablePort() throws Exception {
+ try (ServerSocket serverSocket = new ServerSocket(0)) {
+ return serverSocket.getLocalPort();
+ }
+ }
+}
diff --git
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/TestConsumeMQTT.java
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/TestConsumeMQTT.java
index 0654cc69c02..ce3eed059e0 100644
---
a/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/TestConsumeMQTT.java
+++
b/nifi-extension-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/TestConsumeMQTT.java
@@ -27,7 +27,7 @@ import
org.apache.nifi.processors.mqtt.common.StandardMqttMessage;
import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.provenance.ProvenanceEventType;
import org.apache.nifi.reporting.InitializationException;
-import org.apache.nifi.ssl.SSLContextService;
+import org.apache.nifi.ssl.SSLContextProvider;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
@@ -831,12 +831,12 @@ public class TestConsumeMQTT {
}
private static String addSSLContextService(TestRunner testRunner) throws
InitializationException {
- final SSLContextService sslContextService =
mock(SSLContextService.class);
- final String identifier = SSLContextService.class.getSimpleName();
- when(sslContextService.getIdentifier()).thenReturn(identifier);
+ final SSLContextProvider sslContextProvider =
mock(SSLContextProvider.class);
+ final String identifier = SSLContextProvider.class.getSimpleName();
+ when(sslContextProvider.getIdentifier()).thenReturn(identifier);
- testRunner.addControllerService(identifier, sslContextService);
- testRunner.enableControllerService(sslContextService);
+ testRunner.addControllerService(identifier, sslContextProvider);
+ testRunner.enableControllerService(sslContextProvider);
return identifier;
}
}
