This is an automated email from the ASF dual-hosted git repository. rzo1 pushed a commit to branch fix/netty-client-endpoint-identification in repository https://gitbox.apache.org/repos/asf/storm.git
commit 1ef4431aa0cc363b9f4cb455b3462f31d02d8327 Author: Richard Zowalla <[email protected]> AuthorDate: Fri Aug 21 14:26:54 2026 +0200 Verify the peer identity of the Netty TLS server the worker client connects to --- conf/defaults.yaml | 4 ++ storm-client/src/jvm/org/apache/storm/Config.java | 9 +++ .../org/apache/storm/messaging/netty/Client.java | 2 +- .../netty/StormClientPipelineFactory.java | 26 ++++++- .../netty/StormClientPipelineFactoryTest.java | 84 ++++++++++++++++++++++ 5 files changed, 122 insertions(+), 3 deletions(-) diff --git a/conf/defaults.yaml b/conf/defaults.yaml index 9682cf8bc..5cdb6cac6 100644 --- a/conf/defaults.yaml +++ b/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 diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java index f9c09b897..a367108d2 100644 --- a/storm-client/src/jvm/org/apache/storm/Config.java +++ b/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. diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/Client.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/Client.java index 1865ac33d..dd579bb63 100644 --- a/storm-client/src/jvm/org/apache/storm/messaging/netty/Client.java +++ b/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(); diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/StormClientPipelineFactory.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/StormClientPipelineFactory.java index 198799432..58c5f8bd1 100644 --- a/storm-client/src/jvm/org/apache/storm/messaging/netty/StormClientPipelineFactory.java +++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/StormClientPipelineFactory.java @@ -14,25 +14,38 @@ package org.apache.storm.messaging.netty; 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 @@ class StormClientPipelineFactory extends ChannelInitializer<Channel> { 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 diff --git a/storm-client/test/jvm/org/apache/storm/messaging/netty/StormClientPipelineFactoryTest.java b/storm-client/test/jvm/org/apache/storm/messaging/netty/StormClientPipelineFactoryTest.java new file mode 100644 index 000000000..c3889ca18 --- /dev/null +++ b/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()); + } +}
