This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
The following commit(s) were added to refs/heads/master by this push:
new 54d2c64b7 Verify the peer identity of the Netty TLS server the worker
client connects to
54d2c64b7 is described below
commit 54d2c64b7d07f9b642d61d375fca55e1631c8729
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 d48f1aa44..f86f35f78 100644
--- a/conf/defaults.yaml
+++ b/conf/defaults.yaml
@@ -266,6 +266,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());
+ }
+}
