From bc6525e431891cd67cc09b21ff1aa5ca3a379cdd Mon Sep 17 00:00:00 2001 From: Kevin Berendsen Date: Sun, 7 Aug 2016 13:17:41 +0200 Subject: [PATCH 1/3] MQTT3.1.3-6 stated that the client can connect without giving up an client identifier. However, the broker should generate one for the client id as of the client gave the identifier to broker. The client will now disconnect if the client hasn't given up an identifier with clean session set to 0. Done this regarding issue #190 NettyChannelAssertions class has been modified so you could add a message to the assertion. Unit tests have been added to test the modified processConnect() method. The server can allow zero byte client identifiers through moquette configuration. --- broker/config/moquette.conf | 6 +- .../java/io/moquette/BrokerConstants.java | 1 + .../moquette/spi/impl/ProtocolProcessor.java | 41 ++++++++++--- .../impl/ProtocolProcessorBootstrapper.java | 3 +- .../spi/impl/NettyChannelAssertions.java | 12 +++- .../impl/ProtocolProcessor_CONNECT_Test.java | 59 ++++++++++++++++++- 6 files changed, 108 insertions(+), 14 deletions(-) diff --git a/broker/config/moquette.conf b/broker/config/moquette.conf index 43ae2ebc..a0a87a05 100644 --- a/broker/config/moquette.conf +++ b/broker/config/moquette.conf @@ -15,4 +15,8 @@ password_file config/password_file.conf #false to accept only client connetions with credentials #true to accept client connection without credentails, validating only the one that provides -allow_anonymous true \ No newline at end of file +allow_anonymous true + +#false to prohibit clients from connecting without a clientid. +#true to allow clients to connect without a clientid. One will be generated for them. +allow_zero_byte_client_id false \ No newline at end of file diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 5a04e0a4..73c0374c 100644 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -36,6 +36,7 @@ public class BrokerConstants { public static final String KEY_STORE_PASSWORD_PROPERTY_NAME = "key_store_password"; public static final String KEY_MANAGER_PASSWORD_PROPERTY_NAME = "key_manager_password"; public static final String ALLOW_ANONYMOUS_PROPERTY_NAME = "allow_anonymous"; + public static final String ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME = "allow_zero_byte_client_id"; public static final String ACL_FILE_PROPERTY_NAME = "acl_file"; public static final String AUTHORIZATOR_CLASS_NAME = "authorizator_class"; public static final String AUTHENTICATOR_CLASS_NAME = "authenticator_class"; diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 676b3f01..50e1b208 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -103,6 +103,7 @@ public class ProtocolProcessor { protected ConcurrentMap m_clientIDs; private SubscriptionsStore subscriptions; private boolean allowAnonymous; + private boolean allowZeroByteClientId; private IAuthorizator m_authorizator; private IMessagesStore m_messagesStore; private ISessionsStore m_sessionsStore; @@ -120,8 +121,24 @@ public class ProtocolProcessor { ISessionsStore sessionsStore, IAuthenticator authenticator, boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor) { - init(subscriptions,storageService,sessionsStore,authenticator,allowAnonymous,authorizator,interceptor,null); + init(subscriptions,storageService,sessionsStore,authenticator,allowAnonymous, false, authorizator,interceptor,null); } + + public void init(SubscriptionsStore subscriptions, IMessagesStore storageService, + ISessionsStore sessionsStore, + IAuthenticator authenticator, + boolean allowAnonymous, + boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor) { + init(subscriptions,storageService,sessionsStore,authenticator,allowAnonymous, allowZeroByteClientId, authorizator,interceptor,null); + } + + void init(SubscriptionsStore subscriptions, IMessagesStore storageService, + ISessionsStore sessionsStore, + IAuthenticator authenticator, + boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) { + init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, false, authorizator, interceptor, serverPort); + } + /** * @param subscriptions the subscription store where are stored all the existing * clients subscriptions. @@ -130,6 +147,7 @@ public class ProtocolProcessor { * @param sessionsStore the clients sessions store, used to persist subscriptions. * @param authenticator the authenticator used in connect messages. * @param allowAnonymous true connection to clients without credentials. + * @param allowZeroByteClientId true to allow clients connect without a clientid * @param authorizator used to apply ACL policies to publishes and subscriptions. * @param interceptor to notify events to an intercept handler * @param serverPort @@ -137,11 +155,13 @@ public class ProtocolProcessor { void init(SubscriptionsStore subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator, - boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) { + boolean allowAnonymous, + boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) { this.m_clientIDs = new ConcurrentHashMap<>(); this.m_interceptor = interceptor; this.subscriptions = subscriptions; this.allowAnonymous = allowAnonymous; + this.allowZeroByteClientId = allowZeroByteClientId; m_authorizator = authorizator; LOG.trace("subscription tree on init {}", subscriptions.dumpTree()); m_authenticator = authenticator; @@ -162,11 +182,18 @@ public class ProtocolProcessor { } if (msg.getClientID() == null || msg.getClientID().length() == 0) { - ConnAckMessage okResp = new ConnAckMessage(); - okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED); - channel.writeAndFlush(okResp); - m_interceptor.notifyClientConnected(msg); - return; + if(!msg.isCleanSession() || !this.allowZeroByteClientId) { + ConnAckMessage okResp = new ConnAckMessage(); + okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED); + channel.writeAndFlush(okResp); + channel.close(); + return; + } + + // Generating client id. + String randomIdentifier = UUID.randomUUID().toString().replace("-", ""); + msg.setClientID(randomIdentifier); + LOG.info("Client connected with server generated identifier: {}", randomIdentifier); } //handle user authentication diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java index 0ee8c6d1..480fa596 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java @@ -139,7 +139,8 @@ public class ProtocolProcessorBootstrapper { } boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, "true")); - m_processor.init(subscriptions, messagesStore, sessionsStore, authenticator, allowAnonymous, authorizator, m_interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME)); + boolean allowZeroByteClientId = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, "false")); + m_processor.init(subscriptions, messagesStore, sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME)); return m_processor; } diff --git a/broker/src/test/java/io/moquette/spi/impl/NettyChannelAssertions.java b/broker/src/test/java/io/moquette/spi/impl/NettyChannelAssertions.java index 17881f6e..7d299695 100644 --- a/broker/src/test/java/io/moquette/spi/impl/NettyChannelAssertions.java +++ b/broker/src/test/java/io/moquette/spi/impl/NettyChannelAssertions.java @@ -30,9 +30,17 @@ import static org.junit.Assert.assertTrue; class NettyChannelAssertions { static void assertEqualsConnAck(byte expectedCode, Object connAck) { - assertTrue(connAck instanceof ConnAckMessage); + assertEqualsConnAck(null, expectedCode, connAck); + } + + static void assertEqualsConnAck(String msg, byte expectedCode, Object connAck) { + assertTrue("connAck is not an instance of ConnAckMessage", connAck instanceof ConnAckMessage); ConnAckMessage connAckMsg = (ConnAckMessage) connAck; - assertEquals(expectedCode, connAckMsg.getReturnCode()); + + if(msg == null) + assertEquals(expectedCode, connAckMsg.getReturnCode()); + else + assertEquals(msg, expectedCode, connAckMsg.getReturnCode()); } static void assertConnAckAccepted(EmbeddedChannel channel) { diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index 27d5b5ee..79d8b736 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -35,9 +35,7 @@ import java.util.Map; import java.util.Set; import static io.moquette.parser.netty.Utils.VERSION_3_1_1; -import static io.moquette.parser.proto.messages.ConnAckMessage.BAD_USERNAME_OR_PASSWORD; -import static io.moquette.parser.proto.messages.ConnAckMessage.CONNECTION_ACCEPTED; -import static io.moquette.parser.proto.messages.ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION; +import static io.moquette.parser.proto.messages.ConnAckMessage.*; import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsConnAck; import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsSubAck; import static org.junit.Assert.assertFalse; @@ -331,4 +329,59 @@ public class ProtocolProcessor_CONNECT_Test { //verify that the first subscription is still preserved assertTrue(subscriptions.contains(expectedSubscription)); } + + @Test + public void testZeroByteClientIdWithoutCleanSession() { + // Allow zero byte client ids + m_processor = new ProtocolProcessor(); + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, true, + new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); + + // Connect message without clean session set to true but client id is still null + connMsg = new ConnectMessage(); + connMsg.setProtocolVersion(VERSION_3_1_1); + connMsg.setClientID(null); + connMsg.setCleanSession(false); + + m_processor.processConnect(m_session, connMsg); + assertEqualsConnAck("Identifier should be rejected due to having clean session set to false.", + IDENTIFIER_REJECTED, m_session.readOutbound()); + + assertFalse("Connection should be closed by the broker.", m_session.isOpen()); + } + + @Test + public void testZeroByteClientIdWithCleanSession() { + // Allow zero byte client ids + m_processor = new ProtocolProcessor(); + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, true, + new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); + + // Connect message with clean session set to true and client id is null. + connMsg = new ConnectMessage(); + connMsg.setProtocolVersion(VERSION_3_1_1); + connMsg.setClientID(null); + connMsg.setCleanSession(true); + + m_processor.processConnect(m_session, connMsg); + assertEqualsConnAck("Connection should be accepted. A unique clientid should be generated" + + " and clean session set to true.", CONNECTION_ACCEPTED, m_session.readOutbound()); + + assertTrue("Connection should be valid and open,", m_session.isOpen()); + } + + @Test + public void testZeroByteClientIdNotAllowed() { + // Connect message with clean session set to true and client id is null. + connMsg = new ConnectMessage(); + connMsg.setProtocolVersion(VERSION_3_1_1); + connMsg.setClientID(null); + connMsg.setCleanSession(true); + + m_processor.processConnect(m_session, connMsg); + assertEqualsConnAck("Zero byte client identifiers are not allowed.", + IDENTIFIER_REJECTED, m_session.readOutbound()); + + assertFalse("Connection should closed.", m_session.isOpen()); + } } From 807debb62250336914f41070c66b0f80ff95a1c3 Mon Sep 17 00:00:00 2001 From: Kevin Berendsen Date: Sun, 7 Aug 2016 13:34:06 +0200 Subject: [PATCH 2/3] Terminating the connection if the client failed to receive a CONNACK packet with a non-zero return code. Stated in 3.2.2.3 of MQTT 3.1.1 specificaton. Was checking the wrong channel to see if it was still open or not. The channel that the test was checking on was disconnected a few lines above. --- .../java/io/moquette/spi/impl/ProtocolProcessor.java | 2 ++ .../spi/impl/ProtocolProcessor_CONNECT_Test.java | 11 +++++++++++ 2 files changed, 13 insertions(+) diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 676b3f01..7b3f0128 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -280,6 +280,8 @@ public class ProtocolProcessor { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD); session.writeAndFlush(okResp); + session.close(); + LOG.info("Client {} failed to connect with bad username or password.", session); } /** diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index 27d5b5ee..fddb62e1 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -95,6 +95,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(UNNACEPTABLE_PROTOCOL_VERSION, m_session.readOutbound()); + assertFalse("Connection should be closed by the broker.", m_session.isOpen()); } @Test @@ -122,6 +123,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen()); //TODO verify the call /*verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()), any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));*/ @@ -140,6 +142,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen()); } @Test @@ -154,6 +157,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, m_session.readOutbound()); + assertFalse("Connection should be closed by the broker.", m_session.isOpen()); } @Test @@ -169,6 +173,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, m_session.readOutbound()); + assertFalse("Connection should be closed by the broker.", m_session.isOpen()); } @Test @@ -182,6 +187,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, m_session.readOutbound()); + assertFalse("Connection should be closed by the broker.", m_session.isOpen()); } @Test @@ -197,6 +203,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, m_session.readOutbound()); + assertFalse("Connection should be closed by the broker.", m_session.isOpen()); } @Test @@ -210,6 +217,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen()); } @Test @@ -294,6 +302,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(CONNECTION_ACCEPTED, firstReceiverSession.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", firstReceiverSession.isOpen()); } @@ -306,6 +315,7 @@ public class ProtocolProcessor_CONNECT_Test { connMsg.setCleanSession(false); m_processor.processConnect(m_session, connMsg); assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen()); //subscribe SubscribeMessage subscribeMsg = new SubscribeMessage(); @@ -327,6 +337,7 @@ public class ProtocolProcessor_CONNECT_Test { m_session = new EmbeddedChannel(); m_processor.processConnect(m_session, connMsg); assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen()); //verify that the first subscription is still preserved assertTrue(subscriptions.contains(expectedSubscription)); From 36daf153db941285306facfb07df5ca0ab564c9b Mon Sep 17 00:00:00 2001 From: Jacoby Date: Thu, 18 Aug 2016 16:14:51 +0200 Subject: [PATCH 3/3] * Bugfix - persistence file could not be deleted under Windows because it was in use by the server --- .../java/io/moquette/server/ServerRestartIntegrationTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java b/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java index 369be309..db0ddfa4 100644 --- a/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java +++ b/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java @@ -63,9 +63,9 @@ public class ServerRestartIntegrationTest { @Before public void setUp() throws Exception { - startServer(); String dbPath = IntegrationUtils.localMapDBPath(); IntegrationUtils.cleanPersistenceFile(dbPath); + startServer(); m_subscriber = new MqttClient("tcp://localhost:1883", "Subscriber", s_dataStore); m_messageCollector = new MessageCollector();