This is an automated email from the ASF dual-hosted git repository. rzo1 pushed a commit to branch pacemaker-hardening in repository https://gitbox.apache.org/repos/asf/storm.git
commit fe8abbd53da4549ba85a45b14bf8e9b792e7113e Author: Richard Zowalla <[email protected]> AuthorDate: Fri Sep 11 17:22:58 2026 +0200 Drop unexpected Pacemaker frames instead of terminating the daemon The Pacemaker server decodes a CONTROL_MESSAGE frame into a ControlMessage. With no SASL handler installed (pacemaker.auth.method NONE), or once the SASL handler has forwarded it upstream, the object reached PacemakerServer.received(), which cast it to HBMessage. The resulting ClassCastException reached StormServerHandler.exceptionCaught, which treats anything but an IOException as fatal and exits the JVM. A malformed request (undecodable thrift payload, a control frame without payload or with an unknown code, a request without data) took the same path. The server-side ThriftDecoder now accepts only the control message a Pacemaker client sends, SASL_TOKEN_MESSAGE_REQUEST, which starts the DIGEST handshake. Any other control frame, and control frames whose payload is missing, too short or carries an unknown code, is discarded and the connection closed, in the style of the worker MessageDecoder. SASL_MESSAGE_TOKEN and HBMessage frames are decoded as before, so DIGEST and KERBEROS handshakes are unchanged. On the client side a malformed control frame is reported as an IOException, so that PacemakerClientHandler reconnects as it did before. PacemakerServer.received() checks the message type before using it: anything that is not an HBMessage is logged with its type and remote address and the connection is closed. The Pacemaker pipeline uses a new PacemakerServerHandler, a StormServerHandler whose exceptionCaught closes the failing connection for any Exception and keeps serving the other clients. Errors are still handed to StormServerHandler. StormServerHandler itself, and with it the worker messaging pipeline, is unchanged. PacemakerServer gets a package-private close() used by the new tests. --- .../storm/pacemaker/codec/ThriftDecoder.java | 51 ++++- .../storm/pacemaker/codec/ThriftDecoderTest.java | 161 ++++++++++++++ .../apache/storm/pacemaker/PacemakerServer.java | 15 ++ .../pacemaker/codec/PacemakerServerHandler.java | 45 ++++ .../pacemaker/codec/ThriftNettyServerCodec.java | 5 +- .../storm/pacemaker/PacemakerServerTest.java | 231 +++++++++++++++++++++ 6 files changed, 504 insertions(+), 4 deletions(-) diff --git a/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftDecoder.java b/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftDecoder.java index ce104ed1e..6b83e298c 100644 --- a/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftDecoder.java +++ b/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftDecoder.java @@ -22,9 +22,12 @@ import org.apache.storm.shade.io.netty.buffer.ByteBuf; import org.apache.storm.shade.io.netty.channel.ChannelHandlerContext; import org.apache.storm.shade.io.netty.handler.codec.ByteToMessageDecoder; import org.apache.storm.utils.Utils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public class ThriftDecoder extends ByteToMessageDecoder { + private static final Logger LOG = LoggerFactory.getLogger(ThriftDecoder.class); private static final int INTEGER_SIZE = 4; /** @@ -32,11 +35,28 @@ public class ThriftDecoder extends ByteToMessageDecoder { */ private final int maxLength; + /** + * Whether this decoder sits in a Pacemaker server pipeline. A server only accepts the control message a client + * sends to start the SASL handshake; any other control frame is dropped and the connection closed. + */ + private final boolean serverSide; + /** * Instantiate a ThriftDecoder that accepts serialized messages of at most maxLength bytes. */ public ThriftDecoder(final int maxLengthBytes) { + this(maxLengthBytes, false); + } + + /** + * Instantiate a ThriftDecoder that accepts serialized messages of at most maxLength bytes. + * + * @param maxLengthBytes the maximum length of a serialized thrift message + * @param serverSide true if the decoder is used by a Pacemaker server, which restricts the control messages it accepts + */ + public ThriftDecoder(final int maxLengthBytes, final boolean serverSide) { maxLength = maxLengthBytes; + this.serverSide = serverSide; } @Override @@ -68,7 +88,19 @@ public class ThriftDecoder extends ByteToMessageDecoder { HBMessage m = (HBMessage) Utils.thriftDeserialize(HBMessage.class, serialized); if (m.get_type() == HBServerMessageType.CONTROL_MESSAGE) { - ControlMessage cm = ControlMessage.read(m.get_data().get_message_blob()); + ControlMessage cm = readControlMessage(m); + if (cm == null) { + if (!serverSide) { + // Let the client handler see the failure so that it reconnects. + throw new IOException("Received a malformed control message"); + } + dropAndClose(channelHandlerContext, buf, "a malformed control frame"); + return; + } + if (serverSide && cm != ControlMessage.SASL_TOKEN_MESSAGE_REQUEST) { + dropAndClose(channelHandlerContext, buf, "an unexpected control frame " + cm); + return; + } out.add(cm); } else if (m.get_type() == HBServerMessageType.SASL_MESSAGE_TOKEN) { SaslMessageToken sm = SaslMessageToken.read(m.get_data().get_message_blob()); @@ -77,4 +109,21 @@ public class ThriftDecoder extends ByteToMessageDecoder { out.add(m); } } + + private static ControlMessage readControlMessage(HBMessage m) { + if (m.get_data() == null || !m.get_data().is_set_message_blob()) { + return null; + } + byte[] blob = m.get_data().get_message_blob(); + if (blob == null || blob.length < 2) { + return null; + } + return ControlMessage.read(blob); + } + + private static void dropAndClose(ChannelHandlerContext ctx, ByteBuf buf, String what) { + LOG.warn("Channel {} sent {}; closing the connection", ctx.channel(), what); + buf.skipBytes(buf.readableBytes()); + ctx.close(); + } } diff --git a/storm-client/test/jvm/org/apache/storm/pacemaker/codec/ThriftDecoderTest.java b/storm-client/test/jvm/org/apache/storm/pacemaker/codec/ThriftDecoderTest.java new file mode 100644 index 000000000..20df39967 --- /dev/null +++ b/storm-client/test/jvm/org/apache/storm/pacemaker/codec/ThriftDecoderTest.java @@ -0,0 +1,161 @@ +/** + * 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 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.pacemaker.codec; + +import org.apache.storm.generated.HBMessage; +import org.apache.storm.generated.HBMessageData; +import org.apache.storm.generated.HBServerMessageType; +import org.apache.storm.messaging.netty.ControlMessage; +import org.apache.storm.messaging.netty.SaslMessageToken; +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.apache.storm.shade.io.netty.handler.codec.DecoderException; +import org.apache.storm.utils.Utils; +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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class ThriftDecoderTest { + + private static final int MAX_LENGTH = 1024 * 1024; + + static ByteBuf frame(HBMessage message) { + byte[] serialized = Utils.thriftSerialize(message); + ByteBuf buf = Unpooled.buffer(); + buf.writeInt(serialized.length); + buf.writeBytes(serialized); + return buf; + } + + static ByteBuf controlFrame(ControlMessage controlMessage) { + ByteBuf blob = Unpooled.buffer(); + controlMessage.write(blob); + byte[] bytes = new byte[blob.readableBytes()]; + blob.readBytes(bytes); + return frame(new HBMessage(HBServerMessageType.CONTROL_MESSAGE, HBMessageData.message_blob(bytes))); + } + + private static EmbeddedChannel serverChannel() { + return new EmbeddedChannel(new ThriftDecoder(MAX_LENGTH, true)); + } + + @Test + public void serverDropsUnexpectedControlFrames() { + for (ControlMessage controlMessage : ControlMessage.values()) { + if (controlMessage == ControlMessage.SASL_TOKEN_MESSAGE_REQUEST) { + continue; + } + EmbeddedChannel channel = serverChannel(); + + channel.writeInbound(controlFrame(controlMessage)); + + assertNull(channel.readInbound(), controlMessage.name()); + assertFalse(channel.isActive(), controlMessage.name()); + } + } + + @Test + public void serverAcceptsSaslTokenMessageRequest() { + EmbeddedChannel channel = serverChannel(); + + channel.writeInbound(controlFrame(ControlMessage.SASL_TOKEN_MESSAGE_REQUEST)); + + assertSame(ControlMessage.SASL_TOKEN_MESSAGE_REQUEST, channel.readInbound()); + assertTrue(channel.isActive()); + } + + @Test + public void serverAcceptsSaslMessageToken() { + EmbeddedChannel channel = serverChannel(); + byte[] token = { 1, 2, 3 }; + ByteBuf blob = Unpooled.buffer(); + new SaslMessageToken(token).write(blob); + byte[] bytes = new byte[blob.readableBytes()]; + blob.readBytes(bytes); + + channel.writeInbound(frame(new HBMessage(HBServerMessageType.SASL_MESSAGE_TOKEN, HBMessageData.message_blob(bytes)))); + + SaslMessageToken decoded = channel.readInbound(); + assertArrayEquals(token, decoded.getSaslToken()); + assertTrue(channel.isActive()); + } + + @Test + public void serverPassesHeartbeatMessages() { + EmbeddedChannel channel = serverChannel(); + HBMessage message = new HBMessage(HBServerMessageType.CREATE_PATH, HBMessageData.path("/path")); + + channel.writeInbound(frame(message)); + + assertEquals(message, channel.readInbound()); + assertTrue(channel.isActive()); + } + + @Test + public void serverDropsControlFrameWithUnknownCode() { + EmbeddedChannel channel = serverChannel(); + + channel.writeInbound(frame(new HBMessage(HBServerMessageType.CONTROL_MESSAGE, + HBMessageData.message_blob(new byte[]{ 0, 1 })))); + + assertNull(channel.readInbound()); + assertFalse(channel.isActive()); + } + + @Test + public void serverDropsControlFrameWithoutPayload() { + EmbeddedChannel channel = serverChannel(); + + channel.writeInbound(frame(new HBMessage(HBServerMessageType.CONTROL_MESSAGE, null))); + + assertNull(channel.readInbound()); + assertFalse(channel.isActive()); + } + + @Test + public void serverDropsControlFrameWithShortPayload() { + EmbeddedChannel channel = serverChannel(); + + channel.writeInbound(frame(new HBMessage(HBServerMessageType.CONTROL_MESSAGE, + HBMessageData.message_blob(new byte[]{ 1 })))); + + assertNull(channel.readInbound()); + assertFalse(channel.isActive()); + } + + @Test + public void clientDecodesSaslCompleteRequest() { + EmbeddedChannel channel = new EmbeddedChannel(new ThriftDecoder(MAX_LENGTH)); + + channel.writeInbound(controlFrame(ControlMessage.SASL_COMPLETE_REQUEST)); + + assertSame(ControlMessage.SASL_COMPLETE_REQUEST, channel.readInbound()); + assertTrue(channel.isActive()); + } + + @Test + public void clientReportsMalformedControlFrame() { + EmbeddedChannel channel = new EmbeddedChannel(new ThriftDecoder(MAX_LENGTH)); + + assertThrows(DecoderException.class, () -> channel.writeInbound( + frame(new HBMessage(HBServerMessageType.CONTROL_MESSAGE, null)))); + assertNull(channel.readInbound()); + } +} diff --git a/storm-server/src/main/java/org/apache/storm/pacemaker/PacemakerServer.java b/storm-server/src/main/java/org/apache/storm/pacemaker/PacemakerServer.java index 19e6cbe69..4bef3d8c1 100644 --- a/storm-server/src/main/java/org/apache/storm/pacemaker/PacemakerServer.java +++ b/storm-server/src/main/java/org/apache/storm/pacemaker/PacemakerServer.java @@ -133,6 +133,12 @@ class PacemakerServer implements ISaslServer { @Override public void received(Object mesg, String remote, Channel channel) throws InterruptedException { + if (!(mesg instanceof HBMessage)) { + LOG.warn("Dropping unexpected message of type {} from {}; closing the connection", + mesg == null ? null : mesg.getClass().getName(), remote); + channel.close(); + return; + } cleanPipeline(channel); boolean authenticated = (authMethod == ThriftNettyServerCodec.AuthMethod.NONE) || authenticatedChannels.contains(channel); @@ -148,6 +154,15 @@ class PacemakerServer implements ISaslServer { } } + /** + * Close all channels and stop the event loops of this server. + */ + void close() { + allChannels.close().awaitUninterruptibly(); + bossEventLoopGroup.shutdownGracefully().awaitUninterruptibly(); + workerEventLoopGroup.shutdownGracefully().awaitUninterruptibly(); + } + @Override public String name() { return topologyName; diff --git a/storm-server/src/main/java/org/apache/storm/pacemaker/codec/PacemakerServerHandler.java b/storm-server/src/main/java/org/apache/storm/pacemaker/codec/PacemakerServerHandler.java new file mode 100644 index 000000000..890d453e4 --- /dev/null +++ b/storm-server/src/main/java/org/apache/storm/pacemaker/codec/PacemakerServerHandler.java @@ -0,0 +1,45 @@ +/** + * 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.pacemaker.codec; + +import org.apache.storm.messaging.netty.IServer; +import org.apache.storm.messaging.netty.StormServerHandler; +import org.apache.storm.shade.io.netty.channel.ChannelHandlerContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Pacemaker server handler. A failure while handling a request only affects the connection it arrived on: the + * connection is closed and the Pacemaker server keeps serving its other clients. Errors are still handled by + * {@link StormServerHandler}. + */ +public class PacemakerServerHandler extends StormServerHandler { + private static final Logger LOG = LoggerFactory.getLogger(PacemakerServerHandler.class); + + public PacemakerServerHandler(IServer server) { + super(server); + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + if (!(cause instanceof Exception)) { + super.exceptionCaught(ctx, cause); + return; + } + try { + LOG.warn("Closing connection {} after failing to handle its request", ctx.channel(), cause); + } finally { + ctx.close(); + } + } +} diff --git a/storm-server/src/main/java/org/apache/storm/pacemaker/codec/ThriftNettyServerCodec.java b/storm-server/src/main/java/org/apache/storm/pacemaker/codec/ThriftNettyServerCodec.java index d15f369d9..b2ef23af4 100644 --- a/storm-server/src/main/java/org/apache/storm/pacemaker/codec/ThriftNettyServerCodec.java +++ b/storm-server/src/main/java/org/apache/storm/pacemaker/codec/ThriftNettyServerCodec.java @@ -21,7 +21,6 @@ import org.apache.storm.messaging.netty.ISaslServer; import org.apache.storm.messaging.netty.IServer; import org.apache.storm.messaging.netty.KerberosSaslServerHandler; import org.apache.storm.messaging.netty.SaslStormServerHandler; -import org.apache.storm.messaging.netty.StormServerHandler; import org.apache.storm.security.auth.ClientAuthUtils; import org.apache.storm.shade.io.netty.channel.Channel; import org.apache.storm.shade.io.netty.channel.ChannelInitializer; @@ -54,7 +53,7 @@ public class ThriftNettyServerCodec extends ChannelInitializer<Channel> { protected void initChannel(Channel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast("encoder", new ThriftEncoder()); - pipeline.addLast("decoder", new ThriftDecoder(thriftMessageMaxSizeBytes)); + pipeline.addLast("decoder", new ThriftDecoder(thriftMessageMaxSizeBytes, true)); if (authMethod == AuthMethod.DIGEST) { try { LOG.debug("Adding SaslStormServerHandler to pacemaker server pipeline."); @@ -78,7 +77,7 @@ public class ThriftNettyServerCodec extends ChannelInitializer<Channel> { LOG.debug("Not authenticating any clients. AuthMethod is NONE"); } - pipeline.addLast("handler", new StormServerHandler(server)); + pipeline.addLast("handler", new PacemakerServerHandler(server)); } public enum AuthMethod { diff --git a/storm-server/src/test/java/org/apache/storm/pacemaker/PacemakerServerTest.java b/storm-server/src/test/java/org/apache/storm/pacemaker/PacemakerServerTest.java new file mode 100644 index 000000000..99919b95a --- /dev/null +++ b/storm-server/src/test/java/org/apache/storm/pacemaker/PacemakerServerTest.java @@ -0,0 +1,231 @@ +/** + * 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.pacemaker; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.storm.Config; +import org.apache.storm.DaemonConfig; +import org.apache.storm.generated.HBMessage; +import org.apache.storm.generated.HBMessageData; +import org.apache.storm.generated.HBServerMessageType; +import org.apache.storm.messaging.netty.ControlMessage; +import org.apache.storm.messaging.netty.ISaslClient; +import org.apache.storm.messaging.netty.ISaslServer; +import org.apache.storm.messaging.netty.SaslStormClientHandler; +import org.apache.storm.metric.StormMetricsRegistry; +import org.apache.storm.pacemaker.codec.ThriftDecoder; +import org.apache.storm.pacemaker.codec.ThriftEncoder; +import org.apache.storm.pacemaker.codec.ThriftNettyServerCodec; +import org.apache.storm.pacemaker.codec.ThriftNettyServerCodec.AuthMethod; +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.Channel; +import org.apache.storm.shade.io.netty.channel.embedded.EmbeddedChannel; +import org.apache.storm.utils.Utils; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +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.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class PacemakerServerTest { + + private static final int MAX_LENGTH = 1024 * 1024; + + private static PacemakerServer server; + + private static Map<String, Object> config() { + Map<String, Object> conf = new HashMap<>(); + conf.put(Config.PACEMAKER_PORT, 0); + conf.put(Config.PACEMAKER_AUTH_METHOD, "NONE"); + conf.put(DaemonConfig.PACEMAKER_MAX_THREADS, 1); + conf.put(Config.PACEMAKER_THRIFT_MESSAGE_SIZE_MAX, MAX_LENGTH); + return conf; + } + + private static ByteBuf frame(byte[] serialized) { + ByteBuf buf = Unpooled.buffer(); + buf.writeInt(serialized.length); + buf.writeBytes(serialized); + return buf; + } + + private static ByteBuf frame(HBMessage message) { + return frame(Utils.thriftSerialize(message)); + } + + private static ByteBuf controlFrame(ControlMessage controlMessage) { + ByteBuf buf = Unpooled.buffer(); + controlMessage.write(buf); + byte[] blob = new byte[buf.readableBytes()]; + buf.readBytes(blob); + return frame(new HBMessage(HBServerMessageType.CONTROL_MESSAGE, HBMessageData.message_blob(blob))); + } + + private static HBMessage readResponse(EmbeddedChannel serverChannel) { + EmbeddedChannel decoder = new EmbeddedChannel(new ThriftDecoder(MAX_LENGTH)); + Object out; + while ((out = serverChannel.readOutbound()) != null) { + decoder.writeInbound(out); + } + return decoder.readInbound(); + } + + private EmbeddedChannel pipeline() { + return new EmbeddedChannel(new ThriftNettyServerCodec(server, config(), AuthMethod.NONE, MAX_LENGTH)); + } + + @BeforeAll + public static void setUp() { + server = new PacemakerServer(new Pacemaker(new ConcurrentHashMap<>(), new StormMetricsRegistry()), config()); + } + + @AfterAll + public static void tearDown() { + server.close(); + } + + @Test + public void heartbeatRequestIsAnswered() { + EmbeddedChannel channel = pipeline(); + HBMessage request = new HBMessage(HBServerMessageType.CREATE_PATH, HBMessageData.path("/path")); + request.set_message_id(7); + + channel.writeInbound(frame(request)); + + HBMessage response = readResponse(channel); + assertEquals(HBServerMessageType.CREATE_PATH_RESPONSE, response.get_type()); + assertEquals(7, response.get_message_id()); + assertTrue(channel.isActive()); + } + + @Test + public void controlFrameClosesOnlyThatConnection() { + EmbeddedChannel other = pipeline(); + EmbeddedChannel channel = pipeline(); + + channel.writeInbound(controlFrame(ControlMessage.CLOSE_MESSAGE)); + + assertFalse(channel.isActive()); + assertNull(channel.readOutbound()); + + other.writeInbound(frame(new HBMessage(HBServerMessageType.CREATE_PATH, HBMessageData.path("/path")))); + assertEquals(HBServerMessageType.CREATE_PATH_RESPONSE, readResponse(other).get_type()); + assertTrue(other.isActive()); + } + + @Test + public void saslFrameWithoutAuthenticationConfiguredClosesConnection() { + // Passes the decoder, but no SASL handler is installed when the auth method is NONE. + EmbeddedChannel channel = pipeline(); + + channel.writeInbound(controlFrame(ControlMessage.SASL_TOKEN_MESSAGE_REQUEST)); + + assertFalse(channel.isActive()); + assertNull(channel.readOutbound()); + } + + @Test + public void receivedDropsUnexpectedMessageType() throws Exception { + EmbeddedChannel channel = new EmbeddedChannel(); + + server.received(ControlMessage.CLOSE_MESSAGE, "remote", channel); + + assertFalse(channel.isActive()); + assertNull(channel.readOutbound()); + } + + @Test + public void undecodableFrameClosesConnection() { + EmbeddedChannel channel = pipeline(); + + channel.writeInbound(frame(new byte[]{ 0x7f, 0x7f, 0x7f, 0x7f })); + + assertFalse(channel.isActive()); + } + + @Test + public void requestFailureClosesConnection() { + EmbeddedChannel channel = pipeline(); + + channel.writeInbound(frame(new HBMessage(HBServerMessageType.SEND_PULSE, null))); + + assertFalse(channel.isActive()); + assertNull(channel.readOutbound()); + } + + @Test + public void digestHandshakeAuthenticatesChannel() throws Exception { + ISaslServer saslServer = mock(ISaslServer.class); + when(saslServer.name()).thenReturn("pacemaker_server"); + when(saslServer.secretKey()).thenReturn("secret"); + ISaslClient saslClient = mock(ISaslClient.class); + when(saslClient.name()).thenReturn("pacemaker_server"); + when(saslClient.secretKey()).thenReturn("secret"); + + EmbeddedChannel serverChannel = new EmbeddedChannel( + new ThriftNettyServerCodec(saslServer, config(), AuthMethod.DIGEST, MAX_LENGTH)); + EmbeddedChannel clientChannel = new EmbeddedChannel( + new ThriftEncoder(), new ThriftDecoder(MAX_LENGTH), new SaslStormClientHandler(saslClient)); + + exchange(clientChannel, serverChannel); + + verify(saslServer).authenticated(serverChannel); + verify(saslClient, atLeastOnce()).channelReady(any(Channel.class)); + assertNull(serverChannel.pipeline().get(ThriftNettyServerCodec.SASL_HANDLER)); + assertTrue(serverChannel.isActive()); + assertTrue(clientChannel.isActive()); + verify(saslServer, never()).received(any(), anyString(), any(Channel.class)); + + HBMessage request = new HBMessage(HBServerMessageType.GET_PULSE, HBMessageData.path("/path")); + serverChannel.writeInbound(frame(request)); + verify(saslServer).received(request, serverChannel.remoteAddress().toString(), serverChannel); + + // Control frames other than the handshake request are still refused once authenticated. + serverChannel.writeInbound(controlFrame(ControlMessage.EOB_MESSAGE)); + assertFalse(serverChannel.isActive()); + } + + private static void exchange(EmbeddedChannel client, EmbeddedChannel server) { + boolean progress = true; + while (progress) { + progress = forward(client, server) | forward(server, client); + } + } + + private static boolean forward(EmbeddedChannel from, EmbeddedChannel to) { + List<Object> frames = new ArrayList<>(); + Object out; + while ((out = from.readOutbound()) != null) { + frames.add(out); + } + for (Object frame : frames) { + to.writeInbound(frame); + } + return !frames.isEmpty(); + } +}
