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()); + } }