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

Fixed issue #415 by adding a configuration flag to force subscription re-authentication on client connect

Browse Source
This commit is contained in:
andsel
2018-07-19 17:23:18 +02:00
parent 241a659f35
commit 8a03dd8131
10 changed files with 109 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/BrokerConstants.java
+1
View File
@@ -46,6 +46,7 @@ public final 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 REAUTHORIZE_SUBSCRIPTIONS_ON_CONNECT = "reauthorize_subscriptions_on_connect";
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";
broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java
+12 -1
View File
@@ -17,11 +17,15 @@
package io.moquette.server.netty;
import io.moquette.spi.impl.ProtocolProcessor;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -92,7 +96,14 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
break;
}
} catch (Throwable ex) {
ctx.fireExceptionCaught(ex);
//ctx.fireExceptionCaught(ex);
LOG.error("Error processing protocol message: {}", messageType, ex);
ctx.channel().close().addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) {
LOG.info("Closed client channel due to exception in processing");
}
});
} finally {
ReferenceCountUtil.release(msg);
}
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+38 -10
View File
@@ -19,6 +19,7 @@ package io.moquette.spi.impl;
import io.moquette.connections.IConnectionsManager;
import io.moquette.interception.InterceptHandler;
import io.moquette.interception.messages.InterceptAcknowledgedMessage;
import io.moquette.persistence.MemorySessionStore;
import io.moquette.server.ConnectionDescriptor;
import io.moquette.server.ConnectionDescriptorStore;
import io.moquette.server.netty.AutoFlushHandler;
@@ -104,9 +105,9 @@ public class ProtocolProcessor {
private ConcurrentMap<RunningSubscription, SubscriptionState> subscriptionInCourse;
private ISubscriptionsDirectory subscriptions;
// private ISubscriptionsStore subscriptionStore;
private boolean allowAnonymous;
private boolean allowZeroByteClientId;
private boolean reauthorizeSubscriptionsOnConnect;
private IAuthorizator m_authorizator;
private IMessagesStore m_messagesStore;
@@ -131,16 +132,19 @@ public class ProtocolProcessor {
public void init(ISubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
IAuthenticator authenticator, boolean allowAnonymous, IAuthorizator authorizator,
BrokerInterceptor interceptor, SessionsRepository sessionsRepository) {
BrokerInterceptor interceptor, SessionsRepository sessionsRepository,
boolean reauthorizeSubscriptionsOnConnect) {
init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, false,
authorizator, interceptor, sessionsRepository);
authorizator, interceptor, sessionsRepository, reauthorizeSubscriptionsOnConnect);
}
public void init(ISubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId,
IAuthorizator authorizator, BrokerInterceptor interceptor, SessionsRepository sessionsRepository) {
IAuthorizator authorizator, BrokerInterceptor interceptor, SessionsRepository sessionsRepository,
boolean reauthorizeSubscriptionsOnConnect) {
init(new ConnectionDescriptorStore(), subscriptions, storageService, sessionsStore,
authenticator, allowAnonymous, allowZeroByteClientId, authorizator, interceptor, sessionsRepository);
authenticator, allowAnonymous, allowZeroByteClientId, authorizator, interceptor, sessionsRepository,
reauthorizeSubscriptionsOnConnect);
}
/**
@@ -164,13 +168,15 @@ public class ProtocolProcessor {
void init(IConnectionsManager connectionDescriptors, ISubscriptionsDirectory subscriptions,
IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator,
boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator,
BrokerInterceptor interceptor, SessionsRepository sessionsRepository) {
BrokerInterceptor interceptor, SessionsRepository sessionsRepository,
boolean reauthorizeSubscriptionsOnConnect) {
LOG.debug("Initializing MQTT protocol processor...");
this.connectionDescriptors = connectionDescriptors;
this.subscriptionInCourse = new ConcurrentHashMap<>();
this.m_interceptor = interceptor;
this.subscriptions = subscriptions;
this.allowAnonymous = allowAnonymous;
this.reauthorizeSubscriptionsOnConnect = reauthorizeSubscriptionsOnConnect;
this.allowZeroByteClientId = allowZeroByteClientId;
m_authorizator = authorizator;
if (LOG.isDebugEnabled()) {
@@ -202,7 +208,8 @@ public class ProtocolProcessor {
public void processConnect(Channel channel, MqttConnectMessage msg) {
MqttConnectPayload payload = msg.payload();
String clientId = payload.clientIdentifier();
LOG.debug("Processing CONNECT message. CId={}, username={}", clientId, payload.userName());
final String username = payload.userName();
LOG.debug("Processing CONNECT message. CId={}, username={}", clientId, username);
if (msg.variableHeader().version() != MqttVersion.MQTT_3_1.protocolLevel()
&& msg.variableHeader().version() != MqttVersion.MQTT_3_1_1.protocolLevel()) {
@@ -221,14 +228,14 @@ public class ProtocolProcessor {
channel.writeAndFlush(badId).addListener(FIRE_EXCEPTION_ON_FAILURE);
channel.close().addListener(CLOSE_ON_FAILURE);
LOG.error("The MQTT client ID cannot be empty. Username={}", payload.userName());
LOG.error("The MQTT client ID cannot be empty. Username={}", username);
return;
}
// Generating client id.
clientId = UUID.randomUUID().toString().replace("-", "");
LOG.info("Client has connected with a server generated identifier. CId={}, username={}", clientId,
payload.userName());
username);
}
if (!login(channel, msg, clientId)) {
@@ -248,6 +255,9 @@ public class ProtocolProcessor {
initializeKeepAliveTimeout(channel, msg, clientId);
storeWillMessage(msg, clientId);
if (!cleanSession && reauthorizeSubscriptionsOnConnect) {
reauthorizeOnExistingSubscriptions(clientId, username);
}
if (!sendAck(descriptor, msg, clientId)) {
channel.close().addListener(CLOSE_ON_FAILURE);
return;
@@ -274,7 +284,22 @@ public class ProtocolProcessor {
channel.close().addListener(CLOSE_ON_FAILURE);
}
LOG.info("Connected client <{}> with login <{}>", clientId, payload.userName());
LOG.info("Connected client <{}> with login <{}>", clientId, username);
}
private void reauthorizeOnExistingSubscriptions(String clientId, String username) {
if (!m_sessionsStore.contains(clientId)) {
return;
}
final Collection<Subscription> clientSubscriptions = m_sessionsStore.subscriptionStore()
.listClientSubscriptions(clientId);
for (Subscription sub : clientSubscriptions) {
final Topic topicToReauthorize = sub.getTopicFilter();
final boolean readAuthorized = m_authorizator.canRead(topicToReauthorize, username, clientId);
if (!readAuthorized) {
subscriptions.removeSubscription(topicToReauthorize, clientId);
}
}
}
private void setupAutoFlusher(Channel channel, int flushIntervalMs) {
@@ -634,6 +659,9 @@ public class ProtocolProcessor {
LOG.trace("Removing saved subscriptions. CId={}", descriptor.clientID);
final ClientSession session = this.sessionsRepository.sessionForClient(clientID);
session.wipeSubscriptions();
for (Subscription existingSub : session.getSubscriptions()) {
this.subscriptions.removeSubscription(existingSub.getTopicFilter(), clientID);
}
LOG.trace("Saved subscriptions have been removed. CId={}", descriptor.clientID);
}
return true;
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java
+4 -5
View File
@@ -168,13 +168,12 @@ public class ProtocolProcessorBootstrapper {
connectionDescriptors = new ConnectionDescriptorStore();
LOG.info("Initializing MQTT protocol processor...");
boolean allowAnonymous = Boolean
.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, "true"));
boolean allowZeroByteClientId = Boolean
.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, "false"));
boolean allowAnonymous = props.boolProp(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, true);
boolean allowZeroByteClientId = props.boolProp(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, false);
boolean reauthorizeSubscriptionsOnConnect = props.boolProp(BrokerConstants.REAUTHORIZE_SUBSCRIPTIONS_ON_CONNECT, false);
m_processor.init(connectionDescriptors, subscriptions, messagesStore, m_sessionsStore, authenticator,
allowAnonymous, allowZeroByteClientId, authorizator, interceptor,
sessionsRepository);
sessionsRepository, reauthorizeSubscriptionsOnConnect);
return m_processor;
}