diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 0eb42107..2e9393f5 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -27,20 +27,22 @@ import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; -import org.dna.mqtt.moquette.messaging.spi.ISessionsStore; import org.dna.mqtt.moquette.messaging.spi.IMessagesStore; +import org.dna.mqtt.moquette.messaging.spi.ISessionsStore; import org.dna.mqtt.moquette.messaging.spi.impl.events.MessagingEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.OutputMessagingEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.PubAckEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; -import org.dna.mqtt.moquette.proto.messages.PubCompMessage; +import static org.dna.mqtt.moquette.parser.netty.Utils.VERSION_3_1; +import static org.dna.mqtt.moquette.parser.netty.Utils.VERSION_3_1_1; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.dna.mqtt.moquette.proto.messages.PubAckMessage; +import org.dna.mqtt.moquette.proto.messages.PubCompMessage; import org.dna.mqtt.moquette.proto.messages.PubRecMessage; import org.dna.mqtt.moquette.proto.messages.PubRelMessage; import org.dna.mqtt.moquette.proto.messages.PublishMessage; @@ -142,7 +144,7 @@ class ProtocolProcessor implements EventHandler { void processConnect(ServerChannel session, ConnectMessage msg) { LOG.debug("processConnect for client {}", msg.getClientID()); - if (msg.getProcotolVersion() != 0x03) { + if (msg.getProcotolVersion() != VERSION_3_1 && msg.getProcotolVersion() != VERSION_3_1_1) { ConnAckMessage badProto = new ConnAckMessage(); badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION); LOG.warn("processConnect sent bad proto ConnAck"); @@ -151,7 +153,7 @@ class ProtocolProcessor implements EventHandler { return; } - if (msg.getClientID() == null || msg.getClientID().length() > 23) { + if (msg.getClientID() == null || msg.getClientID().length() > 23 || msg.getClientID().length() == 0) { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED); session.write(okResp); diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java index a4bbefe4..6b8760dc 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java @@ -20,13 +20,13 @@ import java.io.IOException; import java.util.concurrent.TimeUnit; import org.fusesource.mqtt.client.BlockingConnection; import org.fusesource.mqtt.client.MQTT; +import org.fusesource.mqtt.client.Message; import org.fusesource.mqtt.client.QoS; import org.fusesource.mqtt.client.Topic; -import org.fusesource.mqtt.client.Message; import org.junit.After; +import static org.junit.Assert.*; import org.junit.Before; import org.junit.Test; -import static org.junit.Assert.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java index ea052179..2690abb7 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java @@ -89,7 +89,7 @@ public class ServerIntegrationPahoTest { assertEquals("/topic", m_callback.getTopic()); } - + @Test public void testCleanSession_maintainClientSubscriptions() throws Exception { diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java index 9d6556d9..0eed28d7 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerLowlevelMessagesIntegrationTests.java @@ -17,8 +17,11 @@ package org.dna.mqtt.moquette.server; import java.io.File; import java.io.IOException; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; +import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.dna.mqtt.moquette.testclient.Client; import org.fusesource.mqtt.client.BlockingConnection; @@ -27,9 +30,9 @@ import org.fusesource.mqtt.client.Message; import org.fusesource.mqtt.client.QoS; import org.fusesource.mqtt.client.Topic; import org.junit.After; +import static org.junit.Assert.*; import org.junit.Before; import org.junit.Test; -import static org.junit.Assert.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -121,4 +124,37 @@ public class ServerLowlevelMessagesIntegrationTests { assertEquals(willTestamentMsg, new String(msg.getPayload())); willSubscriber.disconnect(); } + + AbstractMessage receivedMsg; + + @Test + public void checkRejectConnectWithEmptyClientID() throws InterruptedException { + LOG.info("*** checkRejectConnectWithEmptyClientID ***"); + final CountDownLatch latch = new CountDownLatch(1); + m_client.setCallback(new Client.ICallback() { + + public void call(AbstractMessage msg) { + receivedMsg = msg; + latch.countDown(); + } + }); + + int keepAlive = 2; //secs + ConnectMessage connectMessage = new ConnectMessage(); + connectMessage.setProcotolVersion((byte) 4); + connectMessage.setClientID(""); + connectMessage.setKeepAlive(keepAlive); + connectMessage.setWillFlag(false); + connectMessage.setWillQos((byte) QOSType.MOST_ONE.ordinal()); + + //Execute + m_client.sendMessage(connectMessage); + + latch.await(200, TimeUnit.MILLISECONDS); + + assertTrue(receivedMsg instanceof ConnAckMessage); + ConnAckMessage connAck = (ConnAckMessage) receivedMsg; + assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, connAck.getReturnCode()); + } + } diff --git a/broker/src/test/java/org/dna/mqtt/moquette/testclient/Client.java b/broker/src/test/java/org/dna/mqtt/moquette/testclient/Client.java index 741efe3b..42ecc88d 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/testclient/Client.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/testclient/Client.java @@ -24,6 +24,7 @@ import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; +import java.util.concurrent.Callable; import org.dna.mqtt.commons.Constants; import org.dna.mqtt.moquette.parser.netty.MQTTDecoder; import org.dna.mqtt.moquette.parser.netty.MQTTEncoder; @@ -39,6 +40,10 @@ import org.slf4j.LoggerFactory; */ public class Client { + public static interface ICallback { + void call(AbstractMessage msg); + } + private static final Logger LOG = LoggerFactory.getLogger(Client.class); final ClientNettyMQTTHandler handler = new ClientNettyMQTTHandler(); EventLoopGroup workerGroup; @@ -78,12 +83,20 @@ public class Client { } } + ICallback callback; + public void setCallback(ICallback callback) { + this.callback = callback; + } + public void sendMessage(AbstractMessage msg) { m_channel.writeAndFlush(msg); } void messageReceived(AbstractMessage msg) { LOG.info("Received message " + msg); + if (this.callback != null) { + this.callback.call(msg); + } } void setConnectionLost(boolean status) { diff --git a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoder.java b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoder.java index 7c30f0f5..e296270c 100644 --- a/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoder.java +++ b/netty_parser/src/main/java/org/dna/mqtt/moquette/parser/netty/ConnectDecoder.java @@ -153,8 +153,8 @@ public class ConnectDecoder extends DemuxDecoder { int keepAlive = in.readUnsignedShort(); message.setKeepAlive(keepAlive); - if ((remainingLength == 12 && message.getProcotolVersion() == 3) || - (remainingLength == 10 && message.getProcotolVersion() == 4)) { + if ((remainingLength == 12 && message.getProcotolVersion() == VERSION_3_1) || + (remainingLength == 10 && message.getProcotolVersion() == VERSION_3_1_1)) { out.add(message); return; }