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 2e700201eb6 NIFI-16184 Add TLS Endpoint Identification for Cluster
Communication (#11530)
2e700201eb6 is described below
commit 2e700201eb6fd16831be37689090976529634a6d
Author: David Handermann <[email protected]>
AuthorDate: Wed Aug 12 14:50:24 2026 -0500
NIFI-16184 Add TLS Endpoint Identification for Cluster Communication
(#11530)
---
.../src/main/asciidoc/administration-guide.adoc | 10 +-
.../nifi-framework-cluster-protocol/pom.xml | 22 ++-
.../org/apache/nifi/io/socket/SocketUtils.java | 7 +
.../org/apache/nifi/io/socket/SocketUtilsTest.java | 155 +++++++++++++++++++++
.../nifi-framework/nifi-framework-core/pom.xml | 5 +
.../async/nio/NioAsyncLoadBalanceClient.java | 23 +--
.../clustered/client/async/nio/PeerChannel.java | 4 +
.../client/async/nio/PeerChannelProvider.java | 33 +++++
.../async/nio/StandardPeerChannelProvider.java | 79 +++++++++++
.../async/nio/TestStandardPeerChannelProvider.java | 123 ++++++++++++++++
.../tests/system/loadbalance/LoadBalanceIT.java | 2 +-
11 files changed, 435 insertions(+), 28 deletions(-)
diff --git a/nifi-docs/src/main/asciidoc/administration-guide.adoc
b/nifi-docs/src/main/asciidoc/administration-guide.adoc
index 138a50631f8..5ba8082df22 100644
--- a/nifi-docs/src/main/asciidoc/administration-guide.adoc
+++ b/nifi-docs/src/main/asciidoc/administration-guide.adoc
@@ -3683,9 +3683,13 @@ by the `nifi.cluster.flow.election.max.candidates`
property, the cluster will no
|`nifi.cluster.flow.election.max.candidates`|Specifies the number of Nodes
required in the cluster to cause early election of Flows. This allows the Nodes
in the cluster to avoid having to wait a
long time before starting processing if we reach at least this number of nodes
in the cluster.
|`nifi.cluster.load.balance.port`|Specifies the port to listen on for incoming
connections for load balancing data across the cluster. The default value is
`6342`.
-|`nifi.cluster.load.balance.host`|Specifies the hostname to listen on for
incoming connections for load balancing data across the cluster. If not
specified, will default to the value used by the
-`nifi.cluster.node.address` property. The value set here does not have to be a
hostname/IP address that is addressable outside of the cluster. However, all
nodes within the cluster must be able to
-connect to the node using this hostname/IP address.
+|`nifi.cluster.load.balance.host`|Specifies the hostname to listen on for
incoming connections for load balancing data
+across the cluster. If not specified, will default to the value used by the
`nifi.cluster.node.address` property. The
+value set here does not have to be a hostname address that is addressable
outside of the cluster. However, all nodes
+within the cluster must be able to connect to the node using this hostname
address.
+
+When TLS encryption is enabled for cluster communication, this value must be
set to a hostname that aligns with DNS
+Subject Alternative Names included on the server certificate configured in the
application key store.
|`nifi.cluster.load.balance.connections.per.node`|The maximum number of
connections to create between this node and each other node in the cluster. For
example, if there are 5 nodes in the cluster and this value is set to 4, there
will be up to 20 socket connections established for load-balancing purposes (5
x 4 = 20). The default value is `1`.
|`nifi.cluster.load.balance.max.thread.count`|The maximum number of threads to
use for transferring data from this node to other nodes in the cluster. While a
given thread can only write to a single socket at a time, a single thread is
capable of servicing multiple connections simultaneously because a given
connection may not be available for reading/writing at any given time. The
default value is `8`—i.e., up to 8 threads will be responsible for transferring
data to other nodes, regardl [...]
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/pom.xml
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/pom.xml
index dee8a3e1f12..6a8c7cdd50d 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/pom.xml
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/pom.xml
@@ -70,11 +70,6 @@
<groupId>jakarta.xml.bind</groupId>
<artifactId>jakarta.xml.bind-api</artifactId>
</dependency>
- <dependency>
- <groupId>org.glassfish.jaxb</groupId>
- <artifactId>jaxb-runtime</artifactId>
- <scope>test</scope>
- </dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-beans</artifactId>
@@ -87,5 +82,22 @@
<groupId>com.fasterxml.jackson.module</groupId>
<artifactId>jackson-module-jakarta-xmlbind-annotations</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.glassfish.jaxb</groupId>
+ <artifactId>jaxb-runtime</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-security-ssl</artifactId>
+ <version>2.12.0-SNAPSHOT</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-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/io/socket/SocketUtils.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/io/socket/SocketUtils.java
index eb126a26bab..05ca248e194 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/io/socket/SocketUtils.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/io/socket/SocketUtils.java
@@ -28,6 +28,7 @@ import java.net.Socket;
import java.security.cert.CertificateException;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLException;
+import javax.net.ssl.SSLParameters;
import javax.net.ssl.SSLServerSocket;
import javax.net.ssl.SSLSocket;
@@ -35,6 +36,8 @@ public final class SocketUtils {
private static final Logger logger =
LoggerFactory.getLogger(SocketUtils.class);
+ private static final String TLS_ENDPOINT_IDENTIFICATION_ALGORITHM =
"HTTPS";
+
/**
* Returns a {@link Socket} (effectively used as a client socket) for the
given address and configuration.
*
@@ -63,6 +66,10 @@ public final class SocketUtils {
final SSLSocket sslSocket = (SSLSocket) tempSocket;
// Set Preferred TLS Protocol Versions
sslSocket.setEnabledProtocols(TlsPlatform.getPreferredProtocols().toArray(new
String[0]));
+ final SSLParameters sslParameters = sslSocket.getSSLParameters();
+
sslParameters.setEndpointIdentificationAlgorithm(TLS_ENDPOINT_IDENTIFICATION_ALGORITHM);
+ sslSocket.setSSLParameters(sslParameters);
+
socket = sslSocket;
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/io/socket/SocketUtilsTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/io/socket/SocketUtilsTest.java
new file mode 100644
index 00000000000..c964bfebde2
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/io/socket/SocketUtilsTest.java
@@ -0,0 +1,155 @@
+/*
+ * 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.io.socket;
+
+import org.apache.nifi.security.cert.builder.StandardCertificateBuilder;
+import org.apache.nifi.security.ssl.EphemeralKeyStoreBuilder;
+import org.apache.nifi.security.ssl.StandardSslContextBuilder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.security.KeyPair;
+import java.security.KeyPairGenerator;
+import java.security.KeyStore;
+import java.security.NoSuchAlgorithmException;
+import java.security.cert.Certificate;
+import java.security.cert.X509Certificate;
+import java.time.Duration;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLHandshakeException;
+import javax.net.ssl.SSLSocket;
+import javax.security.auth.x500.X500Principal;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class SocketUtilsTest {
+
+ private static final char[] EMPTY_PASSWORD = new char[]{};
+
+ private static final Duration CERTIFICATE_VALIDITY = Duration.ofHours(1);
+
+ private static final int TIMEOUT_MILLISECONDS = 5000;
+
+ private static final String LOCALHOST = "localhost";
+
+ private static final String NON_MATCHING_HOST = "nifi.apache.org";
+
+ private static final String KEY_ALGORITHM = "RSA";
+
+ private static final String SUBJECT_FORMAT = "CN=%s";
+
+ private ExecutorService executorService;
+
+ @BeforeEach
+ void setExecutorService() {
+ executorService = Executors.newSingleThreadExecutor();
+ }
+
+ @AfterEach
+ void shutdownExecutorService() {
+ executorService.shutdownNow();
+ }
+
+ @Test
+ void testCreateSocketVerifiesMatchingHostname() throws Exception {
+ final KeyPair keyPair = generateKeyPair();
+ final X509Certificate certificate = buildCertificate(keyPair,
LOCALHOST);
+
+ connectAndHandshake(keyPair, certificate);
+ }
+
+ @Test
+ void testCreateSocketRejectsMismatchedHostname() throws Exception {
+ final KeyPair keyPair = generateKeyPair();
+ final X509Certificate certificate = buildCertificate(keyPair,
NON_MATCHING_HOST);
+
+ assertThrows(SSLHandshakeException.class, () ->
connectAndHandshake(keyPair, certificate));
+ }
+
+ private void connectAndHandshake(final KeyPair serverKeyPair, final
X509Certificate serverCertificate) throws Exception {
+ final SSLContext serverSslContext =
buildServerSslContext(serverKeyPair, serverCertificate);
+ final SSLContext clientSslContext =
buildClientSslContext(serverCertificate);
+
+ final ServerSocketConfiguration serverConfiguration = new
ServerSocketConfiguration();
+ serverConfiguration.setSslContext(serverSslContext);
+ serverConfiguration.setSocketTimeout(TIMEOUT_MILLISECONDS);
+
+ try (ServerSocket serverSocket = SocketUtils.createServerSocket(0,
serverConfiguration)) {
+ final int port = serverSocket.getLocalPort();
+
+ final Future<?> serverConnection = executorService.submit(() -> {
+ try (Socket accepted = serverSocket.accept()) {
+ // Read to initiate handshaking
+ final int read = accepted.getInputStream().read();
+ assertEquals(0, read);
+ }
+ return null;
+ });
+
+ final SocketConfiguration clientConfiguration = new
SocketConfiguration();
+ clientConfiguration.setSslContext(clientSslContext);
+ clientConfiguration.setSocketTimeout(TIMEOUT_MILLISECONDS);
+
+ try (Socket socket = SocketUtils.createSocket(new
InetSocketAddress(LOCALHOST, port), clientConfiguration)) {
+ final SSLSocket sslSocket = (SSLSocket) socket;
+ sslSocket.startHandshake();
+ sslSocket.getOutputStream().write(0);
+ assertTrue(sslSocket.getSession().isValid());
+ } finally {
+ serverConnection.cancel(true);
+ }
+ }
+ }
+
+ private SSLContext buildServerSslContext(final KeyPair keyPair, final
X509Certificate certificate) {
+ final KeyStore keyStore = new EphemeralKeyStoreBuilder()
+ .addPrivateKeyEntry(new
KeyStore.PrivateKeyEntry(keyPair.getPrivate(), new Certificate[]{certificate}))
+ .build();
+ return new StandardSslContextBuilder()
+ .keyStore(keyStore)
+ .keyPassword(EMPTY_PASSWORD)
+ .build();
+ }
+
+ private SSLContext buildClientSslContext(final X509Certificate
certificate) {
+ final KeyStore trustStore = new EphemeralKeyStoreBuilder()
+ .addCertificate(certificate)
+ .build();
+ return new StandardSslContextBuilder()
+ .trustStore(trustStore)
+ .build();
+ }
+
+ private X509Certificate buildCertificate(final KeyPair keyPair, final
String commonName) {
+ final X500Principal subject = new
X500Principal(SUBJECT_FORMAT.formatted(commonName));
+ return new StandardCertificateBuilder(keyPair, subject,
CERTIFICATE_VALIDITY).build();
+ }
+
+ private KeyPair generateKeyPair() throws NoSuchAlgorithmException {
+ final KeyPairGenerator keyPairGenerator =
KeyPairGenerator.getInstance(KEY_ALGORITHM);
+ return keyPairGenerator.generateKeyPair();
+ }
+}
diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/pom.xml
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/pom.xml
index 3cc4033fc93..6b7665cb01d 100644
--- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/pom.xml
+++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/pom.xml
@@ -257,6 +257,11 @@
<artifactId>nifi-authorizer</artifactId>
<version>2.12.0-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-security-ssl</artifactId>
+ <version>2.12.0-SNAPSHOT</version>
+ </dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-mock-authorizer</artifactId>
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/client/async/nio/NioAsyncLoadBalanceClient.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/client/async/nio/NioAsyncLoadBalanceClient.java
index 376dcac1e34..e7b8834a30f 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/client/async/nio/NioAsyncLoadBalanceClient.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/client/async/nio/NioAsyncLoadBalanceClient.java
@@ -56,19 +56,18 @@ import java.util.function.BooleanSupplier;
import java.util.function.Predicate;
import java.util.function.Supplier;
import javax.net.ssl.SSLContext;
-import javax.net.ssl.SSLEngine;
public class NioAsyncLoadBalanceClient implements AsyncLoadBalanceClient {
private static final Logger logger =
LoggerFactory.getLogger(NioAsyncLoadBalanceClient.class);
private static final long PENALIZATION_MILLIS =
TimeUnit.SECONDS.toMillis(1L);
private final NodeIdentifier nodeIdentifier;
- private final SSLContext sslContext;
private final int timeoutMillis;
private final FlowFileContentAccess flowFileContentAccess;
private final LoadBalanceFlowFileCodec flowFileCodec;
private final EventReporter eventReporter;
private final ClusterCoordinator clusterCoordinator;
+ private final PeerChannelProvider peerChannelProvider;
private volatile boolean running = false;
private final AtomicLong penalizationEnd = new AtomicLong(0L);
@@ -92,12 +91,12 @@ public class NioAsyncLoadBalanceClient implements
AsyncLoadBalanceClient {
public NioAsyncLoadBalanceClient(final NodeIdentifier nodeIdentifier,
final SSLContext sslContext, final int timeoutMillis, final
FlowFileContentAccess flowFileContentAccess,
final LoadBalanceFlowFileCodec
flowFileCodec, final EventReporter eventReporter, final ClusterCoordinator
clusterCoordinator) {
this.nodeIdentifier = nodeIdentifier;
- this.sslContext = sslContext;
this.timeoutMillis = timeoutMillis;
this.flowFileContentAccess = flowFileContentAccess;
this.flowFileCodec = flowFileCodec;
this.eventReporter = eventReporter;
this.clusterCoordinator = clusterCoordinator;
+ this.peerChannelProvider = new StandardPeerChannelProvider(sslContext,
nodeIdentifier);
}
@Override
@@ -449,7 +448,8 @@ public class NioAsyncLoadBalanceClient implements
AsyncLoadBalanceClient {
socketChannel = createChannel();
socketChannel.configureBlocking(true);
- peerChannel = createPeerChannel(socketChannel,
socketChannel.getLocalAddress() + "::" + socketChannel.getRemoteAddress());
+ final String peerDescription = socketChannel.getLocalAddress()
+ "::" + socketChannel.getRemoteAddress();
+ peerChannel =
peerChannelProvider.getPeerChannel(socketChannel, peerDescription);
channel = peerChannel;
}
@@ -479,21 +479,6 @@ public class NioAsyncLoadBalanceClient implements
AsyncLoadBalanceClient {
}
}
- private PeerChannel createPeerChannel(final SocketChannel channel, final
String peerDescription) {
- if (sslContext == null) {
- logger.debug("No SSL Context is available so will not perform SSL
Handshake with Peer {}", peerDescription);
- return new PeerChannel(channel, null, peerDescription);
- }
-
- logger.debug("Performing SSL Handshake with Peer {}", peerDescription);
-
- final SSLEngine sslEngine = sslContext.createSSLEngine();
- sslEngine.setUseClientMode(true);
- sslEngine.setNeedClientAuth(true);
-
- return new PeerChannel(channel, sslEngine, peerDescription);
- }
-
private SocketChannel createChannel() throws IOException {
final SocketChannel socketChannel = SocketChannel.open();
try {
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/client/async/nio/PeerChannel.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/client/async/nio/PeerChannel.java
index 894559b92a4..9ba759e16aa 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/client/async/nio/PeerChannel.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/client/async/nio/PeerChannel.java
@@ -105,6 +105,10 @@ public class PeerChannel implements Closeable {
return peerDescription;
}
+ SSLEngine getSslEngine() {
+ return sslEngine;
+ }
+
/**
* Write one byte to the channel
*
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/client/async/nio/PeerChannelProvider.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/client/async/nio/PeerChannelProvider.java
new file mode 100644
index 00000000000..8ad36914140
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/client/async/nio/PeerChannelProvider.java
@@ -0,0 +1,33 @@
+/*
+ * 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.controller.queue.clustered.client.async.nio;
+
+import java.nio.channels.SocketChannel;
+
+/**
+ * Provider abstraction for creating a {@link PeerChannel} bound to a
connected Socket Channel for load balancing communication.
+ */
+interface PeerChannelProvider {
+ /**
+ * Get a Peer Channel for the provided Socket Channel with TLS when
supported
+ *
+ * @param socketChannel Connected Socket Channel for communication with
the peer
+ * @param peerDescription Description of the peer used for logging
+ * @return Peer Channel wrapping the provided Socket Channel
+ */
+ PeerChannel getPeerChannel(SocketChannel socketChannel, String
peerDescription);
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/client/async/nio/StandardPeerChannelProvider.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/client/async/nio/StandardPeerChannelProvider.java
new file mode 100644
index 00000000000..a414a673203
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/client/async/nio/StandardPeerChannelProvider.java
@@ -0,0 +1,79 @@
+/*
+ * 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.controller.queue.clustered.client.async.nio;
+
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.channels.SocketChannel;
+import java.util.Objects;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLParameters;
+
+/**
+ * Standard implementation of Peer Channel Provider that configures TLS with
HTTPS endpoint identification so that
+ * the peer certificate is verified against the Load Balance address of the
destination node during the handshake.
+ */
+class StandardPeerChannelProvider implements PeerChannelProvider {
+
+ private static final Logger logger =
LoggerFactory.getLogger(StandardPeerChannelProvider.class);
+
+ private static final String TLS_ENDPOINT_IDENTIFICATION_ALGORITHM =
"HTTPS";
+
+ private final SSLContext sslContext;
+
+ private final NodeIdentifier nodeIdentifier;
+
+ StandardPeerChannelProvider(final SSLContext sslContext, final
NodeIdentifier nodeIdentifier) {
+ this.sslContext = sslContext;
+ this.nodeIdentifier = nodeIdentifier;
+ }
+
+ @Override
+ public PeerChannel getPeerChannel(final SocketChannel socketChannel, final
String peerDescription) {
+ Objects.requireNonNull(socketChannel, "Socket Channel required");
+ Objects.requireNonNull(peerDescription, "Peer Description required");
+
+ final PeerChannel peerChannel;
+
+ if (sslContext == null) {
+ logger.debug("SSLContext not configured for Peer Channel [{}]",
peerDescription);
+ peerChannel = new PeerChannel(socketChannel, null,
peerDescription);
+ } else {
+ logger.debug("Configured TLS for Peer Channel [{}]",
peerDescription);
+ final SSLEngine sslEngine = createSslEngine();
+ peerChannel = new PeerChannel(socketChannel, sslEngine,
peerDescription);
+ }
+
+ return peerChannel;
+ }
+
+ private SSLEngine createSslEngine() {
+ // Provide the peer address so endpoint identification can verify the
peer certificate against the Load Balance host
+ final SSLEngine sslEngine =
sslContext.createSSLEngine(nodeIdentifier.getLoadBalanceAddress(),
nodeIdentifier.getLoadBalancePort());
+ sslEngine.setUseClientMode(true);
+
+ final SSLParameters sslParameters = sslEngine.getSSLParameters();
+
sslParameters.setEndpointIdentificationAlgorithm(TLS_ENDPOINT_IDENTIFICATION_ALGORITHM);
+ sslEngine.setSSLParameters(sslParameters);
+
+ return sslEngine;
+ }
+}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/queue/clustered/client/async/nio/TestStandardPeerChannelProvider.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/queue/clustered/client/async/nio/TestStandardPeerChannelProvider.java
new file mode 100644
index 00000000000..d8ab874efcb
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/queue/clustered/client/async/nio/TestStandardPeerChannelProvider.java
@@ -0,0 +1,123 @@
+/*
+ * 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.controller.queue.clustered.client.async.nio;
+
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.security.cert.builder.StandardCertificateBuilder;
+import org.apache.nifi.security.ssl.EphemeralKeyStoreBuilder;
+import org.apache.nifi.security.ssl.StandardSslContextBuilder;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.nio.channels.SocketChannel;
+import java.security.GeneralSecurityException;
+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 javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import javax.security.auth.x500.X500Principal;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+class TestStandardPeerChannelProvider {
+
+ private static final String KEY_ALGORITHM = "RSA";
+
+ private static final String SUBJECT_FORMAT = "CN=%s";
+
+ private static final String LOAD_BALANCE_ADDRESS =
"node-1.nifi.example.com";
+
+ private static final int LOAD_BALANCE_PORT = 6342;
+
+ private static final String NODE_ID = "node-1";
+
+ private static final String LOCALHOST = "localhost";
+
+ private static final int NODE_PORT = 8443;
+
+ private static final String PEER_DESCRIPTION = "local::remote";
+
+ private static final String TLS_ENDPOINT_IDENTIFICATION_ALGORITHM =
"HTTPS";
+
+ private static final Duration CERTIFICATE_VALIDITY = Duration.ofHours(1);
+
+ private static final char[] EMPTY_PASSWORD = new char[]{};
+
+ private static SSLContext sslContext;
+
+ private static NodeIdentifier nodeIdentifier;
+
+ @BeforeAll
+ static void setConfiguration() throws GeneralSecurityException {
+ final KeyPair keyPair =
KeyPairGenerator.getInstance(KEY_ALGORITHM).generateKeyPair();
+ final X509Certificate certificate = new
StandardCertificateBuilder(keyPair, new
X500Principal(SUBJECT_FORMAT.formatted(LOAD_BALANCE_ADDRESS)),
CERTIFICATE_VALIDITY).build();
+ final KeyStore keyStore = new EphemeralKeyStoreBuilder()
+ .addPrivateKeyEntry(new
KeyStore.PrivateKeyEntry(keyPair.getPrivate(), new Certificate[]{certificate}))
+ .build();
+
+ sslContext = new StandardSslContextBuilder()
+ .trustStore(keyStore)
+ .keyStore(keyStore)
+ .keyPassword(EMPTY_PASSWORD)
+ .build();
+
+ nodeIdentifier = new NodeIdentifier(
+ NODE_ID,
+ LOCALHOST,
+ NODE_PORT,
+ LOCALHOST,
+ NODE_PORT,
+ LOAD_BALANCE_ADDRESS,
+ LOAD_BALANCE_PORT,
+ LOCALHOST,
+ NODE_PORT,
+ NODE_PORT,
+ false
+ );
+ }
+
+ @Test
+ void testGetPeerChannelConfiguresEndpointIdentification() {
+ final StandardPeerChannelProvider provider = new
StandardPeerChannelProvider(sslContext, nodeIdentifier);
+
+ final PeerChannel peerChannel =
provider.getPeerChannel(mock(SocketChannel.class), PEER_DESCRIPTION);
+ final SSLEngine sslEngine = peerChannel.getSslEngine();
+ assertNotNull(sslEngine, "SSLEngine not configured");
+
+ assertTrue(sslEngine.getUseClientMode(), "Client mode not enabled");
+ assertEquals(TLS_ENDPOINT_IDENTIFICATION_ALGORITHM,
sslEngine.getSSLParameters().getEndpointIdentificationAlgorithm());
+ assertEquals(LOAD_BALANCE_ADDRESS, sslEngine.getPeerHost(), "Peer host
not matched to Load Balance address");
+ assertEquals(LOAD_BALANCE_PORT, sslEngine.getPeerPort(), "Peer port
not matched to Load Balance port");
+ }
+
+ @Test
+ void testGetPeerChannelWithoutSslContext() {
+ final StandardPeerChannelProvider provider = new
StandardPeerChannelProvider(null, nodeIdentifier);
+
+ final PeerChannel peerChannel =
provider.getPeerChannel(mock(SocketChannel.class), PEER_DESCRIPTION);
+ assertNull(peerChannel.getSslEngine(), "SSLEngine should not be
configured without an SSL Context");
+ }
+}
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/loadbalance/LoadBalanceIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/loadbalance/LoadBalanceIT.java
index 3a07c105722..4c07c4fece5 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/loadbalance/LoadBalanceIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/loadbalance/LoadBalanceIT.java
@@ -354,7 +354,7 @@ public class LoadBalanceIT extends NiFiSystemIT {
instance2.stop();
final Map<String, String> updatedLoadBalanceProperties = new
HashMap<>();
- updatedLoadBalanceProperties.put("nifi.cluster.load.balance.host",
"127.0.0.1");
+ updatedLoadBalanceProperties.put("nifi.cluster.load.balance.host",
"localhost");
updatedLoadBalanceProperties.put("nifi.cluster.load.balance.port",
"7676");
instance2.setProperties(updatedLoadBalanceProperties);
