From 07e6ab324f6a320215e3a64181a5642b6b673c9c Mon Sep 17 00:00:00 2001 From: andsel Date: Wed, 23 Jan 2019 22:03:35 +0100 Subject: [PATCH] Fixed issue #440, moved read access check on connect instead of on publish --- .../io/moquette/broker/MQTTConnection.java | 8 ++++++-- .../java/io/moquette/broker/PostOffice.java | 17 ++++++++--------- .../main/java/io/moquette/broker/Server.java | 5 +++-- .../main/java/io/moquette/broker/Session.java | 8 ++++++++ .../io/moquette/broker/SessionRegistry.java | 18 ++++++++++++++---- .../broker/MQTTConnectionConnectTest.java | 8 +++++--- .../broker/MQTTConnectionPublishTest.java | 9 +++++---- .../broker/PostOfficeInternalPublishTest.java | 8 +++++--- .../moquette/broker/PostOfficePublishTest.java | 8 +++++--- .../broker/PostOfficeSubscribeTest.java | 12 +++++++----- .../broker/PostOfficeUnsubscribeTest.java | 8 +++++--- .../moquette/broker/SessionRegistryTest.java | 9 +++++---- 12 files changed, 76 insertions(+), 42 deletions(-) diff --git a/broker/src/main/java/io/moquette/broker/MQTTConnection.java b/broker/src/main/java/io/moquette/broker/MQTTConnection.java index d8909b28..934b3a55 100644 --- a/broker/src/main/java/io/moquette/broker/MQTTConnection.java +++ b/broker/src/main/java/io/moquette/broker/MQTTConnection.java @@ -260,7 +260,7 @@ final class MQTTConnection { } connected = false; //dispatch connection lost to intercept. - String userName= NettyUtils.userName(channel); + String userName = NettyUtils.userName(channel); postOffice.dispatchConnectionLost(clientID,userName); LOG.trace("dispatch disconnection: clientId={}, userName={}", clientID, userName); } @@ -290,7 +290,7 @@ final class MQTTConnection { connected = false; channel.close().addListener(FIRE_EXCEPTION_ON_FAILURE); LOG.trace("Processed DISCONNECT CId={}, channel: {}", clientID, channel); - String userName=NettyUtils.userName(channel); + String userName = NettyUtils.userName(channel); postOffice.dispatchDisconnection(clientID,userName); LOG.trace("dispatch disconnection: clientId={}, userName={}", clientID, userName); } @@ -432,6 +432,10 @@ final class MQTTConnection { return NettyUtils.clientID(channel); } + String getUsername() { + return NettyUtils.userName(channel); + } + public void sendPublishRetainedQos0(Topic topic, MqttQoS qos, ByteBuf payload) { MqttPublishMessage publishMsg = retainedPublish(topic.toString(), qos, payload); sendPublish(publishMsg); diff --git a/broker/src/main/java/io/moquette/broker/PostOffice.java b/broker/src/main/java/io/moquette/broker/PostOffice.java index 6d31885e..2dcf9d0a 100644 --- a/broker/src/main/java/io/moquette/broker/PostOffice.java +++ b/broker/src/main/java/io/moquette/broker/PostOffice.java @@ -19,7 +19,6 @@ import io.moquette.interception.BrokerInterceptor; import io.moquette.broker.subscriptions.ISubscriptionsDirectory; import io.moquette.broker.subscriptions.Subscription; import io.moquette.broker.subscriptions.Topic; -import io.moquette.broker.security.IAuthorizatorPolicy; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.handler.codec.mqtt.*; @@ -45,9 +44,9 @@ class PostOffice { private SessionRegistry sessionRegistry; private BrokerInterceptor interceptor; - PostOffice(ISubscriptionsDirectory subscriptions, IAuthorizatorPolicy authorizatorPolicy, - IRetainedRepository retainedRepository, SessionRegistry sessionRegistry, BrokerInterceptor interceptor) { - this.authorizator = new Authorizator(authorizatorPolicy); + PostOffice(ISubscriptionsDirectory subscriptions, IRetainedRepository retainedRepository, + SessionRegistry sessionRegistry, BrokerInterceptor interceptor, Authorizator authorizator) { + this.authorizator = authorizator; this.subscriptions = subscriptions; this.retainedRepository = retainedRepository; this.sessionRegistry = sessionRegistry; @@ -217,11 +216,11 @@ class PostOffice { if (isSessionPresent) { LOG.debug("Sending PUBLISH message to active subscriber CId: {}, topicFilter: {}, qos: {}", sub.getClientId(), sub.getTopicFilter(), qos); - //TODO determine the user bounded to targetSession - if (!authorizator.canRead(topic, "TODO", sub.getClientId())) { - LOG.debug("Authorizator prohibit Client {} to be notified on {}", sub.getClientId(), topic); - return; - } +// final String login = targetSession.login(); +// if (!authorizator.canRead(topic, login, sub.getClientId())) { +// LOG.debug("Authorizator prohibit Client {} to be notified on {}", sub.getClientId(), topic); +// return; +// } // we need to retain because duplicate only copy r/w indexes and don't retain() causing refCnt = 0 ByteBuf payload = origPayload.retainedDuplicate(); diff --git a/broker/src/main/java/io/moquette/broker/Server.java b/broker/src/main/java/io/moquette/broker/Server.java index 265a4283..166de91f 100644 --- a/broker/src/main/java/io/moquette/broker/Server.java +++ b/broker/src/main/java/io/moquette/broker/Server.java @@ -178,8 +178,9 @@ public class Server { ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory(); subscriptions.init(subscriptionsRepository); - SessionRegistry sessions = new SessionRegistry(subscriptions, queueRepository); - dispatcher = new PostOffice(subscriptions, authorizatorPolicy, retainedRepository, sessions, interceptor); + final Authorizator authorizator = new Authorizator(authorizatorPolicy); + SessionRegistry sessions = new SessionRegistry(subscriptions, queueRepository, authorizator); + dispatcher = new PostOffice(subscriptions, retainedRepository, sessions, interceptor, authorizator); final BrokerConfiguration brokerConfig = new BrokerConfiguration(config); MQTTConnectionFactory connectionFactory = new MQTTConnectionFactory(brokerConfig, authenticator, sessions, dispatcher); diff --git a/broker/src/main/java/io/moquette/broker/Session.java b/broker/src/main/java/io/moquette/broker/Session.java index c1214398..0662ef89 100644 --- a/broker/src/main/java/io/moquette/broker/Session.java +++ b/broker/src/main/java/io/moquette/broker/Session.java @@ -362,6 +362,14 @@ class Session { ReferenceCountUtil.release(removedMsg); } +// public String login() { +// if (!connected()) { +// throw new IllegalStateException("Can't ask login on disconnected session"); +// } +// +// return NettyUtils.userName(mqttConnection.channel); +// } + @Override public String toString() { return "Session{" + diff --git a/broker/src/main/java/io/moquette/broker/SessionRegistry.java b/broker/src/main/java/io/moquette/broker/SessionRegistry.java index f2287d14..dba89e47 100644 --- a/broker/src/main/java/io/moquette/broker/SessionRegistry.java +++ b/broker/src/main/java/io/moquette/broker/SessionRegistry.java @@ -28,7 +28,6 @@ import org.slf4j.LoggerFactory; import java.util.Queue; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentMap; public class SessionRegistry { @@ -61,11 +60,15 @@ public class SessionRegistry { private final ConcurrentMap pool = new ConcurrentHashMap<>(); private final ISubscriptionsDirectory subscriptionsDirectory; private final IQueueRepository queueRepository; + private final Authorizator authorizator; private final ConcurrentMap> queues = new ConcurrentHashMap<>(); - SessionRegistry(ISubscriptionsDirectory subscriptionsDirectory, IQueueRepository queueRepository) { + SessionRegistry(ISubscriptionsDirectory subscriptionsDirectory, + IQueueRepository queueRepository, + Authorizator authorizator) { this.subscriptionsDirectory = subscriptionsDirectory; this.queueRepository = queueRepository; + this.authorizator = authorizator; } void bindToSession(MQTTConnection mqttConnection, MqttConnectMessage msg, String clientId) { @@ -129,7 +132,8 @@ public class SessionRegistry { LOG.trace("case 2, oldSession with same CId {} disconnected", clientId); } else if (!newIsClean && oldSession.disconnected()) { // case 3 - reactivateSubscriptions(oldSession); + final String username = mqttConnection.getUsername(); + reactivateSubscriptions(oldSession, username); // mark as connected final boolean connecting = oldSession.assignState(SessionStatus.DISCONNECTED, SessionStatus.CONNECTING); @@ -165,8 +169,14 @@ public class SessionRegistry { return postConnectAction; } - private void reactivateSubscriptions(Session session) { + private void reactivateSubscriptions(Session session, String username) { + //verify if subscription still satisfy read ACL permissions for (Subscription existingSub : session.getSubscriptions()) { + final boolean topicReadable = authorizator.canRead(existingSub.getTopicFilter(), username, + session.getClientID()); + if (!topicReadable) { + subscriptionsDirectory.removeSubscription(existingSub.getTopicFilter(), session.getClientID()); + } // TODO // subscriptionsDirectory.reactivate(existingSub.getTopicFilter(), session.getClientID()); } diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java index d5aa50b6..8d768ee9 100644 --- a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java +++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java @@ -65,9 +65,11 @@ public class MQTTConnectionConnectTest { subscriptions.init(subscriptionsRepository); queueRepository = new MemoryQueueRepository(); - sessionRegistry = new SessionRegistry(subscriptions, queueRepository); - postOffice = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(), - sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + final PermitAllAuthorizatorPolicy authorizatorPolicy = new PermitAllAuthorizatorPolicy(); + final Authorizator permitAll = new Authorizator(authorizatorPolicy); + sessionRegistry = new SessionRegistry(subscriptions, queueRepository, permitAll); + postOffice = new PostOffice(subscriptions, new MemoryRetainedRepository(), sessionRegistry, + ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll); sut = createMQTTConnection(CONFIG); channel = (EmbeddedChannel) sut.channel; diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java index 3db19c1b..1d24525f 100644 --- a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java +++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java @@ -70,10 +70,11 @@ public class MQTTConnectionPublishTest { subscriptions.init(subscriptionsRepository); queueRepository = new MemoryQueueRepository(); - sessionRegistry = new SessionRegistry(subscriptions, queueRepository); - final PostOffice postOffice = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), - new MemoryRetainedRepository(), sessionRegistry, - ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + final PermitAllAuthorizatorPolicy authorizatorPolicy = new PermitAllAuthorizatorPolicy(); + final Authorizator permitAll = new Authorizator(authorizatorPolicy); + sessionRegistry = new SessionRegistry(subscriptions, queueRepository, permitAll); + final PostOffice postOffice = new PostOffice(subscriptions, + new MemoryRetainedRepository(), sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll); return new MQTTConnection(channel, config, mockAuthenticator, sessionRegistry, postOffice); } diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java index 6b53ba1f..a1e9d434 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java @@ -85,9 +85,11 @@ public class PostOfficeInternalPublishTest { retainedRepository = new MemoryRetainedRepository(); queueRepository = new MemoryQueueRepository(); - SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, queueRepository); - sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), retainedRepository, sessionRegistry, - ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + final PermitAllAuthorizatorPolicy authorizatorPolicy = new PermitAllAuthorizatorPolicy(); + final Authorizator permitAll = new Authorizator(authorizatorPolicy); + SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, queueRepository, permitAll); + sut = new PostOffice(subscriptions, retainedRepository, sessionRegistry, + ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll); return sessionRegistry; } diff --git a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java index bd898df9..9e867a47 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java @@ -92,9 +92,11 @@ public class PostOfficePublishTest { retainedRepository = new MemoryRetainedRepository(); queueRepository = new MemoryQueueRepository(); - SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, queueRepository); - sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), retainedRepository, sessionRegistry, - ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + final PermitAllAuthorizatorPolicy authorizatorPolicy = new PermitAllAuthorizatorPolicy(); + final Authorizator permitAll = new Authorizator(authorizatorPolicy); + SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, queueRepository, permitAll); + sut = new PostOffice(subscriptions, retainedRepository, sessionRegistry, + ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll); return sessionRegistry; } diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java index b91fb51e..8b4c3a11 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java @@ -90,9 +90,11 @@ public class PostOfficeSubscribeTest { subscriptions.init(subscriptionsRepository); queueRepository = new MemoryQueueRepository(); - sessionRegistry = new SessionRegistry(subscriptions, queueRepository); - sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(), - sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + final PermitAllAuthorizatorPolicy authorizatorPolicy = new PermitAllAuthorizatorPolicy(); + final Authorizator permitAll = new Authorizator(authorizatorPolicy); + sessionRegistry = new SessionRegistry(subscriptions, queueRepository, permitAll); + sut = new PostOffice(subscriptions, new MemoryRetainedRepository(), sessionRegistry, + ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll); } private MQTTConnection createMQTTConnection(BrokerConfiguration config, Channel channel) { @@ -164,8 +166,8 @@ public class PostOfficeSubscribeTest { when(prohibitReadOnNewsTopic.canRead(eq(new Topic(NEWS_TOPIC)), eq(FAKE_USER_NAME), eq(FAKE_CLIENT_ID))) .thenReturn(false); - sut = new PostOffice(subscriptions, prohibitReadOnNewsTopic, new MemoryRetainedRepository(), sessionRegistry, - ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + sut = new PostOffice(subscriptions, new MemoryRetainedRepository(), sessionRegistry, + ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, new Authorizator(prohibitReadOnNewsTopic)); connection.processConnect(connectMessage); ConnectionTestUtils.assertConnectAccepted(channel); diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java index fea802bd..fa497acb 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java @@ -81,9 +81,11 @@ public class PostOfficeUnsubscribeTest { subscriptions.init(subscriptionsRepository); queueRepository = new MemoryQueueRepository(); - sessionRegistry = new SessionRegistry(subscriptions, queueRepository); - sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(), - sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + final PermitAllAuthorizatorPolicy authorizatorPolicy = new PermitAllAuthorizatorPolicy(); + final Authorizator permitAll = new Authorizator(authorizatorPolicy); + sessionRegistry = new SessionRegistry(subscriptions, queueRepository, permitAll); + sut = new PostOffice(subscriptions, new MemoryRetainedRepository(), sessionRegistry, + ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll); } private MQTTConnection createMQTTConnection(BrokerConfiguration config, Channel channel) { diff --git a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java index 8bb5add5..4496023c 100644 --- a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java +++ b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java @@ -70,10 +70,11 @@ public class SessionRegistryTest { subscriptions.init(subscriptionsRepository); queueRepository = new MemoryQueueRepository(); - sut = new SessionRegistry(subscriptions, queueRepository); - final PostOffice postOffice = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), - new MemoryRetainedRepository(), sut, - ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + final PermitAllAuthorizatorPolicy authorizatorPolicy = new PermitAllAuthorizatorPolicy(); + final Authorizator permitAll = new Authorizator(authorizatorPolicy); + sut = new SessionRegistry(subscriptions, queueRepository, permitAll); + final PostOffice postOffice = new PostOffice(subscriptions, + new MemoryRetainedRepository(), sut, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR, permitAll); return new MQTTConnection(channel, config, mockAuthenticator, sut, postOffice); }