NIFI-16184 Add TLS Endpoint Identification for Cluster Communication

nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/pom.xml

@@ -75,6 +75,18 @@
             <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>
         <dependency>
             <groupId>org.springframework</groupId>
             <artifactId>spring-beans</artifactId>

nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/io/socket/SocketUtils.java

@@ -28,13 +28,16 @@
 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;
 
 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;
         }
 

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();
+    }
+}

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 {

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
      *

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);
+}

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;
+    }
+}