anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed issue #440, moved read access check on connect instead of on publish

Browse Source
This commit is contained in:
andsel
2019-01-23 22:03:35 +01:00
parent ea063159f0
commit 07e6ab324f
12 changed files with 76 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/broker/MQTTConnection.java
+6 -2
View File
@@ -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);
broker/src/main/java/io/moquette/broker/PostOffice.java
+8 -9
View File
@@ -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();
broker/src/main/java/io/moquette/broker/Server.java
+3 -2
View File
@@ -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);
broker/src/main/java/io/moquette/broker/Session.java
+8
View File
@@ -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{" +
broker/src/main/java/io/moquette/broker/SessionRegistry.java
+14 -4
View File
@@ -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<String, Session> pool = new ConcurrentHashMap<>();
private final ISubscriptionsDirectory subscriptionsDirectory;
private final IQueueRepository queueRepository;
private final Authorizator authorizator;
private final ConcurrentMap<String, Queue<SessionRegistry.EnqueuedMessage>> 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());
}
broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java
+5 -3
View File
@@ -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;
broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java
+5 -4
View File
@@ -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);
}
broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java
+5 -3
View File
@@ -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;
}
broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java
+5 -3
View File
@@ -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;
}
broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java
+7 -5
View File
@@ -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);
broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java
+5 -3
View File
@@ -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) {
broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
+5 -4
View File
@@ -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);
}