Moved read authorization check from publish phase to subscribe

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

@@ -487,13 +487,7 @@ public class ProtocolProcessor {
         }
         Channel channel = m_clientIDs.get(clientId).channel;
         LOG.debug("Session for clientId {} is {}", clientId, channel);
-
+        channel.writeAndFlush(pubMessage);
-        String user = NettyUtils.userName(channel);
-        if (!m_authorizator.canRead(topic, user, clientId)) {
-            LOG.debug("topic {} doesn't have read credentials", topic);
-            return;
-        }
-        channel.writeAndFlush(pubMessage); //Or writeAdnFlush but flush is costly!!
     }
     
     private void sendPubRec(String clientID, int messageID) {
@@ -662,11 +656,18 @@ public class ProtocolProcessor {
         SubAckMessage ackMessage = new SubAckMessage();
         ackMessage.setMessageID(msg.getMessageID());
 
+        String user = NettyUtils.userName(channel);
         List<Subscription> newSubscriptions = new ArrayList<>();
         for (SubscribeMessage.Couple req : msg.subscriptions()) {
+            if (!m_authorizator.canRead(req.getTopicFilter(), user, clientSession.clientID)) {
+                //send SUBACK with 0x80, the user hasn't credentials to read the topic
+                LOG.debug("topic {} doesn't have read credentials", req.getTopicFilter());
+                ackMessage.addType( AbstractMessage.QOSType.FAILURE);
+                continue;
+            }
+
             AbstractMessage.QOSType qos = AbstractMessage.QOSType.valueOf(req.getQos());
             Subscription newSubscription = new Subscription(clientID, req.getTopicFilter(), qos);
-            //boolean valid = subscribeSingleTopic(newSubscription, req.getTopicFilter());
             boolean valid = clientSession.subscribe(req.getTopicFilter(), newSubscription);
             ackMessage.addType(valid ? qos : AbstractMessage.QOSType.FAILURE);
             if (valid) {

broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java

@@ -28,6 +28,7 @@ import io.moquette.spi.ISessionsStore;
 import io.moquette.spi.impl.security.PermitAllAuthorizator;
 import io.moquette.spi.impl.subscriptions.Subscription;
 import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
+import io.moquette.spi.security.IAuthorizator;
 import io.netty.channel.embedded.EmbeddedChannel;
 import org.junit.Before;
 import org.junit.Test;
@@ -220,6 +221,32 @@ public class ProtocolProcessorTest {
         Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE);
         assertTrue(subscriptions.contains(expectedSubscription));
     }
+
+    @Test
+    public void testSubscribedToNotAuthorizedTopic() {
+        final String fakeUserName = "UnAuthUser";
+        NettyUtils.userName(m_channel, fakeUserName);
+
+        IAuthorizator mockAuthorizator = mock(IAuthorizator.class);
+        when(mockAuthorizator.canRead(eq(FAKE_TOPIC), eq(fakeUserName), eq(FAKE_CLIENT_ID))).thenReturn(false);
+        m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true,
+                mockAuthorizator, NO_OBSERVERS_INTERCEPTOR);
+
+        //Exercise
+        SubscribeMessage msg = new SubscribeMessage();
+        msg.addSubscription(new SubscribeMessage.Couple(AbstractMessage.QOSType.MOST_ONE.byteValue(), FAKE_TOPIC));
+        NettyUtils.clientID(m_channel, FAKE_CLIENT_ID);
+        NettyUtils.cleanSession(m_channel, false);
+        m_sessionStore.createNewSession(FAKE_CLIENT_ID, false);
+        m_processor.processSubscribe(m_channel, msg);
+
+        //Verify
+        Object ackMsg = m_channel.readOutbound();
+        assertTrue(ackMsg instanceof SubAckMessage);
+        SubAckMessage subAckMsg = (SubAckMessage) ackMsg;
+        assertEquals(1, subAckMsg.types().size());
+        assertTrue(subAckMsg.types().contains(AbstractMessage.QOSType.FAILURE));
+    }
     
     @Test
     public void testDoubleSubscribe() {

parser_commons/src/main/java/io/moquette/proto/messages/SubscribeMessage.java

@@ -42,7 +42,7 @@ public class SubscribeMessage extends MessageIDMessage {
             return m_topicFilter;
         }
     }
-    private List<Couple> m_subscriptions = new ArrayList<Couple>();
+    private List<Couple> m_subscriptions = new ArrayList<>();
 
     public SubscribeMessage() {
         //Subscribe has always QoS 1

tools_scripts/trivialPublish.groovy

@@ -2,9 +2,7 @@
 
 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
 
 
 String host = args[0]
@@ -12,6 +10,7 @@ String topic = "/topic"
 if (args.length > 1) {
     topic = args[1]
 }
+println "publish script to topic ${topic} on host ${host}"
 //start a publisher
 MQTT mqtt2 = new MQTT()
 mqtt2.setHost(host, 1883)