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 dee8a3e1f129..ac14dbbfdb5b 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 @@ -75,6 +75,18 @@ jaxb-runtime test + + org.apache.nifi + nifi-security-ssl + 2.12.0-SNAPSHOT + test + + + org.apache.nifi + nifi-security-cert-builder + 2.12.0-SNAPSHOT + test + org.springframework spring-beans 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 eb126a26bab1..05ca248e1943 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.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 static Socket createSocket(final InetSocketAddress address, final SocketC 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 000000000000..c964bfebde2c --- /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/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 376dcac1e346..e7b8834a30ff 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.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 @@ private void establishConnection() throws IOException { 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 @@ private void establishConnection() throws IOException { } } - 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 894559b92a41..9ba759e16aab 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 String getPeerDescription() { 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 000000000000..8ad369141405 --- /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 000000000000..a414a673203f --- /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 000000000000..d8ab874efcb7 --- /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 3a07c1057221..4c07c4fece5c 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 void testRoundRobinWithRestartAndPortChange() throws NiFiClientException, instance2.stop(); final Map 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);