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

@@ -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
+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

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";

broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java

@@ -103,6 +103,7 @@ public class ProtocolProcessor {
     protected ConcurrentMap<String, ConnectionDescriptor> 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();
+            if(!msg.isCleanSession() || !this.allowZeroByteClientId) {
-            okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
+                ConnAckMessage okResp = new ConnAckMessage();
-            channel.writeAndFlush(okResp);
+                okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
-            m_interceptor.notifyClientConnected(msg);
+                channel.writeAndFlush(okResp);
-            return;
+                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

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;
     }
     

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) {

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