From fd1df47fec73e7c9b333ad1cf88755784983b78d Mon Sep 17 00:00:00 2001 From: andrea Date: Sat, 1 Aug 2015 11:59:00 +0200 Subject: [PATCH] Fixed issue #65 --- ChangeLog.txt | 1 + .../server/netty/NettyMQTTHandler.java | 31 ++++--- .../moquette/spi/impl/ProtocolProcessor.java | 4 +- .../server/netty/NettyMQTTHandlerTest.java | 82 +++++++++++++++++++ .../spi/impl/ProtocolProcessorTest.java | 2 +- .../parser/netty/ConnectDecoderTest.java | 2 +- 6 files changed, 103 insertions(+), 19 deletions(-) create mode 100644 broker/src/test/java/org/eclipse/moquette/server/netty/NettyMQTTHandlerTest.java diff --git a/ChangeLog.txt b/ChangeLog.txt index 0a5fed94..ba5badf6 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -1,5 +1,6 @@ Version 0.8: Introduced authenticator and authorizator implementations. + Fixed bug, doesnt' close the channel when connect doesn't contains a valid proto name, issue #65 Version 0.7: Fix issue #38, added aligned to MQTT 3.1.1 diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java index bb5a82f2..8d3d23a4 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java @@ -20,6 +20,8 @@ import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import java.util.HashMap; import java.util.Map; + +import io.netty.handler.codec.CorruptedFrameException; import org.eclipse.moquette.spi.IMessaging; import org.eclipse.moquette.proto.Utils; import org.eclipse.moquette.proto.messages.AbstractMessage; @@ -37,8 +39,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { private static final Logger LOG = LoggerFactory.getLogger(NettyMQTTHandler.class); private IMessaging m_messaging; -// private final Map m_channelMapper = new HashMap<>(); - + @Override public void channelRead(ChannelHandlerContext ctx, Object message) { AbstractMessage msg = (AbstractMessage) message; @@ -54,14 +55,6 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { case PUBREL: case DISCONNECT: case PUBACK: -// NettyChannel channel; -// synchronized(m_channelMapper) { -// if (!m_channelMapper.containsKey(ctx)) { -// m_channelMapper.put(ctx, new NettyChannel(ctx)); -// } -// channel = m_channelMapper.get(ctx); -// } - m_messaging.handleProtocolMessage(new NettyChannel(ctx), msg); break; case PINGREQ: @@ -76,16 +69,22 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { -// NettyChannel channel = m_channelMapper.get(ctx); -// String clientID = (String) channel.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); String clientID = (String) NettyUtils.getAttribute(ctx, NettyChannel.ATTR_KEY_CLIENTID); m_messaging.lostConnection(clientID); ctx.close(/*false*/); -// synchronized(m_channelMapper) { -// m_channelMapper.remove(ctx); -// } } - + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { + if (cause instanceof CorruptedFrameException) { + //something goes bad with decoding + LOG.warn("Error decoding a packet, probably a bad formatted packet, message: " + cause.getMessage()); + } else { + LOG.error("Ugly error on networking", cause); + } + ctx.close(); + } + public void setMessaging(IMessaging messaging) { m_messaging = messaging; } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index f3a72616..6cc139ae 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -23,6 +23,8 @@ import java.util.*; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import io.netty.util.internal.StringUtil; +import org.apache.commons.codec.binary.StringUtils; import org.eclipse.moquette.server.netty.NettyChannel; import org.eclipse.moquette.spi.IMatchingCondition; import org.eclipse.moquette.spi.IMessagesStore; @@ -144,7 +146,7 @@ class ProtocolProcessor implements EventHandler { // Get the ring buffer from the Disruptor to be used for publishing. m_ringBuffer = disruptor.getRingBuffer(); } - + @MQTTMessage(message = ConnectMessage.class) void processConnect(ServerChannel session, ConnectMessage msg) { LOG.debug("CONNECT for client <{}>", msg.getClientID()); diff --git a/broker/src/test/java/org/eclipse/moquette/server/netty/NettyMQTTHandlerTest.java b/broker/src/test/java/org/eclipse/moquette/server/netty/NettyMQTTHandlerTest.java new file mode 100644 index 00000000..35509228 --- /dev/null +++ b/broker/src/test/java/org/eclipse/moquette/server/netty/NettyMQTTHandlerTest.java @@ -0,0 +1,82 @@ +/* + * Copyright (c) 2012-2014 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package org.eclipse.moquette.server.netty; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.ByteToMessageDecoder; +import io.netty.handler.codec.CorruptedFrameException; +import org.eclipse.moquette.parser.netty.ConnectDecoder; +import org.eclipse.moquette.spi.IMessaging; +import org.junit.Test; +import static org.junit.Assert.*; + +import java.util.List; + +import static org.mockito.Mockito.mock; + +/** + * + * @author andrea + */ +public class NettyMQTTHandlerTest { + +// class ConnectDecoderAdapter extends ByteToMessageDecoder { +// +// final ConnectDecoder decoder; +// +// ConnectDecoderAdapter(ConnectDecoder decoder) { +// this.decoder = decoder; +// } +// +// @Override +// protected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, List list) throws Exception { +// decoder.decode(ctx, byteBuf, list); +// } +// } + + class ThrowingExceptionInboundHandler extends ChannelInboundHandlerAdapter { + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + throw new CorruptedFrameException("Invalid protoName: SUPERFAKE"); + } + } + + /** + * Identified by issue #65 + * */ + @Test + public void testCatchingCorruptedFrameException() { + IMessaging mockMsg = mock(IMessaging.class); + NettyMQTTHandler sut = new NettyMQTTHandler(); + sut.setMessaging(mockMsg); + EmbeddedChannel channel = new EmbeddedChannel(new ThrowingExceptionInboundHandler(), sut); + + //prepare a packet with bad protocol name + ByteBuf input = Unpooled.buffer(); + input.writeByte(0x01); + + //Exercise + assertTrue(channel.isOpen()); + channel.writeInbound(input.readBytes(1)); + + //Verify the handler close the channel + assertFalse(channel.isOpen()); + } +} diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java index 248d1b57..1e3d2bf0 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java @@ -185,7 +185,7 @@ public class ProtocolProcessorTest { //Verify assertEquals(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION, m_returnCode); } - + @Test public void testConnect_badClientID() { connMsg.setClientID("extremely_long_clientID_greater_than_23"); diff --git a/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/ConnectDecoderTest.java b/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/ConnectDecoderTest.java index 01210ce8..42845184 100644 --- a/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/ConnectDecoderTest.java +++ b/netty_parser/src/test/java/org/eclipse/moquette/parser/netty/ConnectDecoderTest.java @@ -66,7 +66,7 @@ public class ConnectDecoderTest { public void testBaseHeader_311() throws UnsupportedEncodingException { m_buff = Unpooled.buffer(12); initBaseHeader311(m_buff); - List results = new ArrayList(); + List results = new ArrayList<>(); //Excercise m_msgdec.decode(this.attrMap, m_buff, results);