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 ee97546fb Decode netty server frames only after the handshake completes
ee97546fb is described below
commit ee97546fb2957982f93933ad236d9b248488a8e4
Author: Gianluca Graziadei <[email protected]>
AuthorDate: Sat Aug 22 13:47:38 2026 +0200
Decode netty server frames only after the handshake completes
---
.../storm/messaging/netty/MessageDecoder.java | 34 +++++
.../netty/StormServerPipelineFactory.java | 7 +-
.../storm/messaging/netty/MessageDecoderTest.java | 168 +++++++++++++++++++++
3 files changed, 206 insertions(+), 3 deletions(-)
diff --git
a/storm-client/src/jvm/org/apache/storm/messaging/netty/MessageDecoder.java
b/storm-client/src/jvm/org/apache/storm/messaging/netty/MessageDecoder.java
index a3b282a82..d063cf68f 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/netty/MessageDecoder.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/MessageDecoder.java
@@ -25,10 +25,17 @@ import org.slf4j.LoggerFactory;
public class MessageDecoder extends ByteToMessageDecoder {
private static final Logger LOG =
LoggerFactory.getLogger(MessageDecoder.class);
+ private static final int MAX_UNAUTHENTICATED_SASL_TOKEN_BYTES = 64 * 1024;
private final KryoValuesDeserializer deser;
+ private final boolean serverAuthRequired;
public MessageDecoder(KryoValuesDeserializer deser) {
+ this(deser, false);
+ }
+
+ public MessageDecoder(KryoValuesDeserializer deser, boolean
serverAuthRequired) {
this.deser = deser;
+ this.serverAuthRequired = serverAuthRequired;
}
/*
@@ -86,6 +93,10 @@ public class MessageDecoder extends ByteToMessageDecoder {
// Read the length field.
int length = buf.readInt();
+ if (gateFrames(ctx) && length >
MAX_UNAUTHENTICATED_SASL_TOKEN_BYTES) {
+ discardAndClose(ctx, buf, "an oversized handshake frame");
+ return;
+ }
if (length <= 0) {
out.add(new SaslMessageToken(null));
return;
@@ -109,6 +120,10 @@ public class MessageDecoder extends ByteToMessageDecoder {
// case 3: BackPressureStatus
if (code == BackPressureStatus.IDENTIFIER) {
+ if (gateFrames(ctx)) {
+ discardAndClose(ctx, buf, "a status frame before the
handshake completed");
+ return;
+ }
available = buf.readableBytes();
if (available < 4) {
//Need more data
@@ -129,6 +144,11 @@ public class MessageDecoder extends ByteToMessageDecoder {
// case 4: task Message
+ if (gateFrames(ctx)) {
+ discardAndClose(ctx, buf, "a data frame before the handshake
completed");
+ return;
+ }
+
// Make sure that we have received at least an integer (length)
if (available < 4) {
// need more data
@@ -168,6 +188,20 @@ public class MessageDecoder extends ByteToMessageDecoder {
}
}
+ private boolean gateFrames(ChannelHandlerContext ctx) {
+ if (!serverAuthRequired) {
+ return false;
+ }
+ SaslNettyServer saslNettyServer =
ctx.channel().attr(SaslNettyServerState.SASL_NETTY_SERVER).get();
+ return saslNettyServer == null || !saslNettyServer.isComplete();
+ }
+
+ private static void discardAndClose(ChannelHandlerContext ctx, ByteBuf
buf, String what) {
+ LOG.warn("Channel {} sent {}; closing the connection", ctx.channel(),
what);
+ buf.skipBytes(buf.readableBytes());
+ ctx.close();
+ }
+
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
LOG.error("Exception thrown while decoding messages in channel {};
exception: ", ctx.channel(), cause);
diff --git
a/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerPipelineFactory.java
b/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerPipelineFactory.java
index 7a8b1c61b..02f838142 100644
---
a/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerPipelineFactory.java
+++
b/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerPipelineFactory.java
@@ -44,14 +44,15 @@ class StormServerPipelineFactory extends
ChannelInitializer<Channel> {
pipeline.addLast("ssl", sslContext.newHandler(ch.alloc()));
}
+ boolean isNettyAuth = (Boolean) topoConf
+ .get(Config.STORM_MESSAGING_NETTY_AUTHENTICATION);
+
// Decoder
- pipeline.addLast("decoder", new MessageDecoder(new
KryoValuesDeserializer(topoConf)));
+ pipeline.addLast("decoder", new MessageDecoder(new
KryoValuesDeserializer(topoConf), isNettyAuth));
// Encoders
pipeline.addLast("netty-serializable-encoder",
NettySerializableMessageEncoder.INSTANCE);
pipeline.addLast("backpressure-encoder", new
BackPressureStatusEncoder(new KryoValuesSerializer(topoConf)));
- boolean isNettyAuth = (Boolean) topoConf
- .get(Config.STORM_MESSAGING_NETTY_AUTHENTICATION);
if (isNettyAuth) {
// Authenticate: Removed after authentication completes
pipeline.addLast("saslServerHandler", new SaslStormServerHandler(
diff --git
a/storm-client/test/jvm/org/apache/storm/messaging/netty/MessageDecoderTest.java
b/storm-client/test/jvm/org/apache/storm/messaging/netty/MessageDecoderTest.java
new file mode 100644
index 000000000..969158823
--- /dev/null
+++
b/storm-client/test/jvm/org/apache/storm/messaging/netty/MessageDecoderTest.java
@@ -0,0 +1,168 @@
+/**
+ * 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 java.util.List;
+import org.apache.storm.messaging.TaskMessage;
+import org.apache.storm.serialization.KryoValuesDeserializer;
+import org.apache.storm.shade.io.netty.buffer.ByteBuf;
+import org.apache.storm.shade.io.netty.buffer.Unpooled;
+import org.apache.storm.shade.io.netty.channel.embedded.EmbeddedChannel;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+public class MessageDecoderTest {
+
+ private static final short TASK_ID = 1;
+ private static final int HUGE_LENGTH = 0x7FFFFFFC;
+
+ private static ByteBuf backPressureFrame() {
+ ByteBuf buf = Unpooled.buffer();
+ buf.writeShort(BackPressureStatus.IDENTIFIER);
+ byte[] payload = { 1, 2, 3, 4 };
+ buf.writeInt(payload.length);
+ buf.writeBytes(payload);
+ return buf;
+ }
+
+ private static ByteBuf frameHeader(short code, int declaredLength) {
+ ByteBuf buf = Unpooled.buffer();
+ buf.writeShort(code);
+ buf.writeInt(declaredLength);
+ return buf;
+ }
+
+ private static ByteBuf taskMessageFrame(byte[] payload) {
+ ByteBuf buf = frameHeader(TASK_ID, payload.length);
+ buf.writeBytes(payload);
+ return buf;
+ }
+
+ private static ByteBuf saslTokenFrame(byte[] token) {
+ ByteBuf buf = frameHeader(SaslMessageToken.IDENTIFIER, token.length);
+ buf.writeBytes(token);
+ return buf;
+ }
+
+ private static EmbeddedChannel authenticatedChannel(KryoValuesDeserializer
deser) {
+ EmbeddedChannel channel = new EmbeddedChannel(new
MessageDecoder(deser, true));
+ SaslNettyServer saslNettyServer = mock(SaslNettyServer.class);
+ when(saslNettyServer.isComplete()).thenReturn(true);
+
channel.attr(SaslNettyServerState.SASL_NETTY_SERVER).set(saslNettyServer);
+ return channel;
+ }
+
+ @Test
+ public void backPressureFrameIsNotDeserializedBeforeAuthentication() {
+ KryoValuesDeserializer deser = mock(KryoValuesDeserializer.class);
+ EmbeddedChannel channel = new EmbeddedChannel(new
MessageDecoder(deser, true));
+
+ channel.writeInbound(backPressureFrame());
+
+ assertNull(channel.readInbound());
+ assertFalse(channel.isActive());
+ verifyNoInteractions(deser);
+ }
+
+ @Test
+ public void taskMessageFrameIsNotBufferedBeforeAuthentication() {
+ KryoValuesDeserializer deser = mock(KryoValuesDeserializer.class);
+ EmbeddedChannel channel = new EmbeddedChannel(new
MessageDecoder(deser, true));
+
+ channel.writeInbound(frameHeader(TASK_ID, HUGE_LENGTH));
+
+ assertNull(channel.readInbound());
+ assertFalse(channel.isActive());
+ }
+
+ @Test
+ public void oversizedSaslTokenFrameIsNotBufferedBeforeAuthentication() {
+ KryoValuesDeserializer deser = mock(KryoValuesDeserializer.class);
+ EmbeddedChannel channel = new EmbeddedChannel(new
MessageDecoder(deser, true));
+
+ channel.writeInbound(frameHeader(SaslMessageToken.IDENTIFIER,
HUGE_LENGTH));
+
+ assertNull(channel.readInbound());
+ assertFalse(channel.isActive());
+ }
+
+ @Test
+ public void saslHandshakeFramesAreAcceptedBeforeAuthentication() {
+ KryoValuesDeserializer deser = mock(KryoValuesDeserializer.class);
+ EmbeddedChannel channel = new EmbeddedChannel(new
MessageDecoder(deser, true));
+
+ ByteBuf request = Unpooled.buffer();
+ ControlMessage.SASL_TOKEN_MESSAGE_REQUEST.write(request);
+ channel.writeInbound(request);
+ assertSame(ControlMessage.SASL_TOKEN_MESSAGE_REQUEST,
channel.readInbound());
+
+ byte[] token = { 9, 8, 7 };
+ channel.writeInbound(saslTokenFrame(token));
+
+ SaslMessageToken decoded = channel.readInbound();
+ assertArrayEquals(token, decoded.getSaslToken());
+ assertTrue(channel.isActive());
+ }
+
+ @Test
+ public void backPressureFrameIsDeserializedAfterAuthentication() {
+ KryoValuesDeserializer deser = mock(KryoValuesDeserializer.class);
+ BackPressureStatus status = new BackPressureStatus();
+ when(deser.deserializeObject(any(byte[].class))).thenReturn(status);
+
+ EmbeddedChannel channel = authenticatedChannel(deser);
+
+ channel.writeInbound(backPressureFrame());
+
+ assertSame(status, channel.readInbound());
+ assertTrue(channel.isActive());
+ }
+
+ @Test
+ public void taskMessageFrameIsDecodedAfterAuthentication() {
+ KryoValuesDeserializer deser = mock(KryoValuesDeserializer.class);
+ EmbeddedChannel channel = authenticatedChannel(deser);
+ byte[] payload = { 4, 5, 6 };
+
+ channel.writeInbound(taskMessageFrame(payload));
+
+ List<Object> decoded = channel.readInbound();
+ assertEquals(1, decoded.size());
+ assertArrayEquals(payload, ((TaskMessage) decoded.get(0)).message());
+ assertTrue(channel.isActive());
+ }
+
+ @Test
+ public void framesAreDecodedWhenServerAuthenticationIsNotRequired() {
+ KryoValuesDeserializer deser = mock(KryoValuesDeserializer.class);
+ BackPressureStatus status = new BackPressureStatus();
+ when(deser.deserializeObject(any(byte[].class))).thenReturn(status);
+
+ EmbeddedChannel channel = new EmbeddedChannel(new
MessageDecoder(deser, false));
+
+ channel.writeInbound(backPressureFrame());
+
+ assertSame(status, channel.readInbound());
+ assertTrue(channel.isActive());
+ }
+}
