Verify the peer identity of the Netty TLS server the worker client connects to

conf/defaults.yaml

@@ -264,6 +264,10 @@ storm.messaging.netty.socket.backlog: 500
 # see https://issues.apache.org/jira/browse/STORM-348 for more details
 storm.messaging.netty.authentication: false
 
+# The Netty TLS client checks that the server certificate matches the host it connects to.
+# The worker certificates must carry a host or IP SAN for the address the workers connect to.
+storm.messaging.netty.tls.hostnameVerification: true
+
 # Default plugin to use for automatic network topology discovery
 storm.network.topography.plugin: org.apache.storm.networktopography.DefaultRackDNSToSwitchMapping
 

storm-client/src/jvm/org/apache/storm/Config.java

@@ -1971,6 +1971,15 @@ public class Config extends HashMap<String, Object> {
     @IsString
     public static final String STORM_MESSAGING_NETTY_TLS_SSL_PROTOCOLS = "storm.messaging.netty.tls.ssl.protocols";
 
+    /**
+     * Netty based messaging: Specifies whether the client checks that the server certificate matches the host it is
+     * connecting to when TLS is enabled. Defaults to true. Set this to false only if the worker certificates in use do
+     * not carry a host or IP SAN for the address the workers connect to.
+     */
+    @IsBoolean
+    public static final String STORM_MESSAGING_NETTY_TLS_HOSTNAME_VERIFICATION =
+        "storm.messaging.netty.tls.hostnameVerification";
+
     /**
      * Netty based messaging: The number of milliseconds that a Netty client will retry flushing messages that are already
      * buffered to be sent.

storm-client/src/jvm/org/apache/storm/messaging/netty/Client.java

@@ -160,7 +160,7 @@ public class Client extends ConnectionWithStatus implements ISaslClient {
             .option(ChannelOption.SO_KEEPALIVE, true)
             .option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(lowWatermark, highWatermark))
             .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
-            .handler(new StormClientPipelineFactory(this, remoteBpStatus, topoConf, sslContext));
+            .handler(new StormClientPipelineFactory(this, remoteBpStatus, topoConf, sslContext, host, port));
         dstAddress = new InetSocketAddress(host, port);
         dstAddressPrefixedName = prefixedName(dstAddress);
         launchChannelAliveThread();

storm-client/src/jvm/org/apache/storm/messaging/netty/StormClientPipelineFactory.java

@@ -14,25 +14,38 @@
 
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLParameters;
 import org.apache.storm.Config;
 import org.apache.storm.serialization.KryoValuesDeserializer;
 import org.apache.storm.shade.io.netty.channel.Channel;
 import org.apache.storm.shade.io.netty.channel.ChannelInitializer;
 import org.apache.storm.shade.io.netty.channel.ChannelPipeline;
 import org.apache.storm.shade.io.netty.handler.ssl.SslContext;
+import org.apache.storm.shade.io.netty.handler.ssl.SslHandler;
+import org.apache.storm.utils.ObjectReader;
 
 class StormClientPipelineFactory extends ChannelInitializer<Channel> {
+    private static final String ENDPOINT_IDENTIFICATION_ALGORITHM = "HTTPS";
+    // An empty algorithm turns the check off. A null one must not be used here, the JDK engine ignores it and keeps
+    // whatever algorithm it already had.
+    private static final String NO_ENDPOINT_IDENTIFICATION = "";
+
     private final Client client;
     private final AtomicBoolean[] remoteBpStatus;
     private final Map<String, Object> conf;
     private final SslContext sslContext;
+    private final String dstHost;
+    private final int dstPort;
 
     StormClientPipelineFactory(Client client, AtomicBoolean[] remoteBpStatus, Map<String, Object> conf,
-                               SslContext sslContext) {
+                               SslContext sslContext, String dstHost, int dstPort) {
         this.client = client;
         this.remoteBpStatus = remoteBpStatus;
         this.conf = conf;
         this.sslContext = sslContext;
+        this.dstHost = dstHost;
+        this.dstPort = dstPort;
     }
 
     @Override
@@ -42,7 +55,16 @@ protected void initChannel(Channel ch) throws Exception {
 
         if (this.sslContext != null) {
             // Add SSL handler first to encrypt and decrypt everything.
-            pipeline.addLast("ssl", sslContext.newHandler(ch.alloc()));
+            // The peer host and port give the engine an identity to check the certificate against.
+            SslHandler sslHandler = sslContext.newHandler(ch.alloc(), dstHost, dstPort);
+            boolean verifyHostname = ObjectReader.getBoolean(
+                conf.get(Config.STORM_MESSAGING_NETTY_TLS_HOSTNAME_VERIFICATION), true);
+            SSLEngine sslEngine = sslHandler.engine();
+            SSLParameters sslParameters = sslEngine.getSSLParameters();
+            sslParameters.setEndpointIdentificationAlgorithm(
+                verifyHostname ? ENDPOINT_IDENTIFICATION_ALGORITHM : NO_ENDPOINT_IDENTIFICATION);
+            sslEngine.setSSLParameters(sslParameters);
+            pipeline.addLast("ssl", sslHandler);
         }
 
         // Decoder

storm-client/test/jvm/org/apache/storm/messaging/netty/StormClientPipelineFactoryTest.java

@@ -0,0 +1,84 @@
+/**
+ * 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.storm.messaging.netty;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import javax.net.ssl.SSLEngine;
+import org.apache.storm.Config;
+import org.apache.storm.shade.io.netty.channel.embedded.EmbeddedChannel;
+import org.apache.storm.shade.io.netty.handler.ssl.SslContext;
+import org.apache.storm.shade.io.netty.handler.ssl.SslContextBuilder;
+import org.apache.storm.shade.io.netty.handler.ssl.SslHandler;
+import org.apache.storm.shade.io.netty.handler.ssl.SslProvider;
+import org.apache.storm.utils.Utils;
+import org.junit.jupiter.api.Test;
+
+public class StormClientPipelineFactoryTest {
+    private static final String DST_HOST = "storm-worker-1.example.com";
+    private static final int DST_PORT = 6701;
+
+    private SSLEngine initSslEngine(Map<String, Object> conf) throws Exception {
+        return initSslEngine(conf, null);
+    }
+
+    private SSLEngine initSslEngine(Map<String, Object> conf, SslProvider sslProvider) throws Exception {
+        SslContext sslContext = SslContextBuilder.forClient().sslProvider(sslProvider).build();
+        StormClientPipelineFactory factory =
+            new StormClientPipelineFactory(null, new AtomicBoolean[]{ new AtomicBoolean(false) }, conf, sslContext,
+                                           DST_HOST, DST_PORT);
+        EmbeddedChannel channel = new EmbeddedChannel(factory);
+        try {
+            SslHandler sslHandler = (SslHandler) channel.pipeline().get("ssl");
+            assertNotNull(sslHandler, "no ssl handler was added to the pipeline");
+            return sslHandler.engine();
+        } finally {
+            channel.close();
+        }
+    }
+
+    @Test
+    public void testSslHandlerVerifiesPeerIdentity() throws Exception {
+        Map<String, Object> conf = Utils.readDefaultConfig();
+        conf.put(Config.STORM_MESSAGING_NETTY_AUTHENTICATION, false);
+
+        SSLEngine engine = initSslEngine(conf);
+        assertEquals("HTTPS", engine.getSSLParameters().getEndpointIdentificationAlgorithm());
+        assertEquals(DST_HOST, engine.getPeerHost());
+        assertEquals(DST_PORT, engine.getPeerPort());
+    }
+
+    @Test
+    public void testHostnameVerificationCanBeDisabled() throws Exception {
+        Map<String, Object> conf = Utils.readDefaultConfig();
+        conf.put(Config.STORM_MESSAGING_NETTY_AUTHENTICATION, false);
+        conf.put(Config.STORM_MESSAGING_NETTY_TLS_HOSTNAME_VERIFICATION, false);
+
+        SSLEngine engine = initSslEngine(conf);
+        assertEquals("", engine.getSSLParameters().getEndpointIdentificationAlgorithm());
+    }
+
+    @Test
+    public void testHostnameVerificationCanBeDisabledWithTheJdkSslProvider() throws Exception {
+        Map<String, Object> conf = Utils.readDefaultConfig();
+        conf.put(Config.STORM_MESSAGING_NETTY_AUTHENTICATION, false);
+        conf.put(Config.STORM_MESSAGING_NETTY_TLS_HOSTNAME_VERIFICATION, false);
+
+        // the JDK engine ignores a null algorithm and would keep on verifying
+        SSLEngine engine = initSslEngine(conf, SslProvider.JDK);
+        assertEquals("", engine.getSSLParameters().getEndpointIdentificationAlgorithm());
+    }
+}