diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index a2e58db4..5eee7edc 100644 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -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"; diff --git a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java index 793e77de..714984a6 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java @@ -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); } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 7a9f616f..33d71bbd 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -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 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 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; diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java index 49ea1b49..b8146d70 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java @@ -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; } diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.java index fdde9316..02182780 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoCanPublishOnReadBlockedTopicTest.java @@ -16,6 +16,7 @@ package io.moquette.server; +import io.moquette.BrokerConstants; import io.moquette.server.config.IConfig; import io.moquette.server.config.MemoryConfig; import io.moquette.spi.impl.security.AcceptAllAuthenticator; @@ -27,7 +28,10 @@ import io.netty.handler.codec.mqtt.MqttPublishMessage; import io.netty.handler.codec.mqtt.MqttQoS; import org.eclipse.paho.client.mqttv3.*; import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; -import org.junit.*; +import org.junit.After; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,7 +40,6 @@ import java.io.IOException; import java.util.Properties; import static io.moquette.spi.impl.ProtocolProcessorTest.EMPTY_OBSERVERS; -import static io.moquette.spi.impl.ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR; import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.Assert.*; @@ -52,6 +55,7 @@ public class ServerIntegrationPahoCanPublishOnReadBlockedTopicTest { IMqttClient m_publisher; MessageCollector m_messagesCollector; IConfig m_config; + private boolean canRead; @BeforeClass public static void beforeTests() { @@ -63,9 +67,12 @@ public class ServerIntegrationPahoCanPublishOnReadBlockedTopicTest { protected void startServer() throws IOException { m_server = new Server(); final Properties configProps = IntegrationUtils.prepareTestProperties(); + configProps.setProperty(BrokerConstants.REAUTHORIZE_SUBSCRIPTIONS_ON_CONNECT, "true"); m_config = new MemoryConfig(configProps); + canRead = true; - final IAuthorizator authorizator = new IAuthorizator() { + final IAuthorizator switchingAuthorizator = new IAuthorizator() { +// int callCount = 0; @Override public boolean canWrite(Topic topic, String user, String client) { return true; @@ -73,11 +80,11 @@ public class ServerIntegrationPahoCanPublishOnReadBlockedTopicTest { @Override public boolean canRead(Topic topic, String user, String client) { - return false; + return canRead; } }; - m_server.startServer(m_config, EMPTY_OBSERVERS, null, new AcceptAllAuthenticator(), authorizator); + m_server.startServer(m_config, EMPTY_OBSERVERS, null, new AcceptAllAuthenticator(), switchingAuthorizator); } @Before @@ -115,6 +122,27 @@ public class ServerIntegrationPahoCanPublishOnReadBlockedTopicTest { MqttConnectOptions options = new MqttConnectOptions(); options.setCleanSession(false); m_client.connect(options); + m_client.subscribe("/topic", 0); + + // Exercise + MqttPublishMessage message = MqttMessageBuilders.publish() + .topicName("/topic") + .retained(true) + .qos(MqttQoS.AT_MOST_ONCE) + .payload(Unpooled.copiedBuffer("Hello World!!".getBytes(UTF_8))) + .build(); + + m_server.internalPublish(message, "INTRLPUB"); + + final MqttMessage mqttMessage = m_messagesCollector.waitMessage(1); + assertNotNull(mqttMessage); + + m_client.disconnect(); + // switch the authorizator + canRead = false; + + // Exercise 2 + m_client.connect(options); try { m_client.subscribe("/topic", 0); fail(); @@ -122,127 +150,10 @@ public class ServerIntegrationPahoCanPublishOnReadBlockedTopicTest { // it's OK, the subscribed should fail with error code 128 } - // Exercise - MqttPublishMessage message = MqttMessageBuilders.publish() - .topicName("/topic") - .retained(true) - .qos(MqttQoS.AT_MOST_ONCE) -// .qQos(MqttQoS.AT_LEAST_ONCE) -// .qos(MqttQoS.EXACTLY_ONCE) - .payload(Unpooled.copiedBuffer("Hello World!!".getBytes(UTF_8))) - .build(); - m_server.internalPublish(message, "INTRLPUB"); - // TODO verify the message is not published - final MqttMessage mqttMessage = m_messagesCollector.waitMessage(1); - assertNull(mqttMessage); + // verify the message is not published + final MqttMessage mqttMessage2 = m_messagesCollector.waitMessage(1); + assertNull("No message MUST be received", mqttMessage2); } - - -// @Ignore("This test hasn't any meaning using in memory storage service") -// @Test -// public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws Exception { -// LOG.info("*** testCleanSession_maintainClientSubscriptions_withServerRestart ***"); -// MqttConnectOptions options = new MqttConnectOptions(); -// options.setCleanSession(false); -// m_client.connect(options); -// m_client.subscribe("/topic", 0); -// m_client.disconnect(); -// -// m_server.stopServer(); -// -// m_server.startServer(IntegrationUtils.prepareTestProperties()); -// -// // reconnect and publish -// m_client.connect(options); -// m_client.publish("/topic", "Test my payload".getBytes(UTF_8), 0, false); -// -// assertEquals("/topic", m_messagesCollector.getTopic()); -// } -// -// /** -// * subscriber A connect and subscribe on "a/b" QoS 1 subscriber B connect and subscribe on "a/+" -// * BUT with QoS 2 publisher connects and send a message "hello" on "a/b" subscriber A must -// * receive a notification with QoS1 subscriber B must receive a notification with QoS2 -// */ -// @Test -// public void checkSubscribersGetCorrectQosNotifications() throws Exception { -// LOG.info("*** checkSubscribersGetCorrectQosNotifications ***"); -// String tmpDir = System.getProperty("java.io.tmpdir"); -// -// MqttClientPersistence dsSubscriberA = new MqttDefaultFilePersistence(tmpDir + File.separator + "subscriberA"); -// -// MqttClient subscriberA = new MqttClient("tcp://localhost:1883", "SubscriberA", dsSubscriberA); -// MessageCollector cbSubscriberA = new MessageCollector(); -// subscriberA.setCallback(cbSubscriberA); -// subscriberA.connect(); -// subscriberA.subscribe("a/b", 1); -// -// MqttClientPersistence dsSubscriberB = new MqttDefaultFilePersistence(tmpDir + File.separator + "subscriberB"); -// -// MqttClient subscriberB = new MqttClient("tcp://localhost:1883", "SubscriberB", dsSubscriberB); -// MessageCollector cbSubscriberB = new MessageCollector(); -// subscriberB.setCallback(cbSubscriberB); -// subscriberB.connect(); -// subscriberB.subscribe("a/+", 2); -// -// m_client.connect(); -// m_client.publish("a/b", "Hello world MQTT!!".getBytes(UTF_8), 2, false); -// -// MqttMessage messageOnA = cbSubscriberA.waitMessage(1); -// assertEquals("Hello world MQTT!!", new String(messageOnA.getPayload(), UTF_8)); -// assertEquals(1, messageOnA.getQos()); -// subscriberA.disconnect(); -// -// MqttMessage messageOnB = cbSubscriberB.waitMessage(1); -// assertEquals("Hello world MQTT!!", new String(messageOnB.getPayload(), UTF_8)); -// assertEquals(2, messageOnB.getQos()); -// subscriberB.disconnect(); -// } -// -// @Test -// public void testSubcriptionDoesntStayActiveAfterARestart() throws Exception { -// LOG.info("*** testSubcriptionDoesntStayActiveAfterARestart ***"); -// String tmpDir = System.getProperty("java.io.tmpdir"); -// // clientForSubscribe1 connect and subscribe to /topic QoS2 -// MqttClientPersistence dsSubscriberA = new MqttDefaultFilePersistence( -// tmpDir + File.separator + "clientForSubscribe1"); -// -// MqttClient clientForSubscribe1 = new MqttClient("tcp://localhost:1883", "clientForSubscribe1", dsSubscriberA); -// MessageCollector cbSubscriber1 = new MessageCollector(); -// clientForSubscribe1.setCallback(cbSubscriber1); -// clientForSubscribe1.connect(); -// clientForSubscribe1.subscribe("topic", 0); -// -// // server stop -// m_server.stopServer(); -// System.out.println("\n\n SEVER REBOOTING \n\n"); -// // server start -// startServer(); -// -// // clientForSubscribe2 connect and subscribe to /topic QoS2 -// MqttClientPersistence dsSubscriberB = new MqttDefaultFilePersistence( -// tmpDir + File.separator + "clientForSubscribe2"); -// MqttClient clientForSubscribe2 = new MqttClient("tcp://localhost:1883", "clientForSubscribe2", dsSubscriberB); -// MessageCollector cbSubscriber2 = new MessageCollector(); -// clientForSubscribe2.setCallback(cbSubscriber2); -// clientForSubscribe2.connect(); -// clientForSubscribe2.subscribe("topic", 0); -// -// // clientForPublish publish on /topic with QoS2 a message -// MqttClientPersistence dsSubscriberPUB = new MqttDefaultFilePersistence( -// tmpDir + File.separator + "clientForPublish"); -// MqttClient clientForPublish = new MqttClient("tcp://localhost:1883", "clientForPublish", dsSubscriberPUB); -// clientForPublish.connect(); -// clientForPublish.publish("topic", "Hello".getBytes(UTF_8), 2, true); -// -// // verify clientForSubscribe1 doesn't receive a notification but clientForSubscribe2 yes -// LOG.info("Before waiting to receive 1 sec from {}", clientForSubscribe1.getClientId()); -// assertFalse(clientForSubscribe1.isConnected()); -// assertTrue(clientForSubscribe2.isConnected()); -// LOG.info("Waiting to receive 1 sec from {}", clientForSubscribe2.getClientId()); -// MqttMessage messageOnB = cbSubscriber2.waitMessage(1); -// assertEquals("Hello", new String(messageOnB.getPayload(), UTF_8)); -// } } diff --git a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java index e13a2ce9..d75e8acc 100644 --- a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java +++ b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java @@ -81,7 +81,7 @@ abstract class AbstractProtocolProcessorCommonUtils { subscriptions.init(sessionsRepository); m_processor = new ProtocolProcessor(); m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, - new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, sessionsRepository); + new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, sessionsRepository, false); } void verifyNoPublishIsReceived() { diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 7fb06d3b..b1c0044c 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -91,7 +91,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils SessionsRepository sessionsRepository = new SessionsRepository(storageService.sessionsStore(), null); subs.init(sessionsRepository); m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), - NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository); + NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository, false); connect_v3_1(); @@ -128,7 +128,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils SessionsRepository sessionsRepository = new SessionsRepository(storageService.sessionsStore(), null); subs.init(sessionsRepository); m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), - NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository); + NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository, false); EmbeddedChannel firstReceiverChannel = new EmbeddedChannel(); MqttConnectMessage connectMessage = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1) @@ -181,7 +181,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils .thenReturn(false); m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, mockAuthorizator, - NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository); + NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository, false); connect(); @@ -253,7 +253,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils // simulate a connect that register a clientID to an IoSession m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), - NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository); + NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository, false); connect_v3_1_asClient(PUBLISHER_ID); publishToAs(PUBLISHER_ID, NEWS_TOPIC, AT_MOST_ONCE, true); NettyUtils.cleanSession(m_channel, false); @@ -277,7 +277,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils m_messagesStore.storeRetained(new Topic("/topic"), retainedMessage); m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), - NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository); + NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository, false); connect_v3_1_asClient(PUBLISHER_ID); @@ -294,7 +294,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils when(mockedSubscriptions.matches(eq(new Topic("/topic")))).thenReturn(inactiveSubscriptions); m_processor = new ProtocolProcessor(); m_processor.init(mockedSubscriptions, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), - NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository); + NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository, false); m_processor.sessionsRepository.createNewSession("Subscriber", false); @@ -348,7 +348,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils // simulate a connect that register a clientID to an IoSession m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), - NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository); + NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository, false); connect_v3_1_asClient(PUBLISHER_ID); publishQoS2ToAs(this.m_channel, PUBLISHER_ID, NEWS_TOPIC, 1, true); disconnect(); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index e2f1c5e0..8cdaf2a8 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -70,7 +70,7 @@ public class ProtocolProcessor_CONNECT_Test { m_processor = new ProtocolProcessor(); m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, - new SessionsRepository(this.m_sessionStore, null)); + new SessionsRepository(this.m_sessionStore, null), false); } @Test @@ -160,7 +160,7 @@ public class ProtocolProcessor_CONNECT_Test { protected void reinitProcessorProhibitingAnonymousClients() { m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, false, - new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null)); + new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null), false); } @Test @@ -325,7 +325,7 @@ public class ProtocolProcessor_CONNECT_Test { protected void reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients() { m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, true, - new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null)); + new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null), false); } @Test diff --git a/broker/src/test/resources/config/moquette.conf b/broker/src/test/resources/config/moquette.conf index 3d1bb4bb..11473022 100644 --- a/broker/src/test/resources/config/moquette.conf +++ b/broker/src/test/resources/config/moquette.conf @@ -20,3 +20,5 @@ key_store_password passw0rdsrv key_manager_password passw0rdsrv allow_anonymous true + +reauthorize_subscriptions_on_connect false diff --git a/distribution/src/main/resources/moquette.conf b/distribution/src/main/resources/moquette.conf index 00b2e0d2..6add83cb 100644 --- a/distribution/src/main/resources/moquette.conf +++ b/distribution/src/main/resources/moquette.conf @@ -102,9 +102,14 @@ password_file config/password_file.conf # class name of the authenticator, default implementation uses # definitions in the acl.conf. # If not specified uses FileAuthenticator +# Optional +# reauthorize_subscriptions_on_connect: +# true to force the validation of authorizations of existing +# subscriptions when the client reconnects with not clean session #********************************************************************* # authenticator_class [[path to your class>]] # authorizator_class [[path to your class>]] +# reauthorize_subscriptions_on_connect true|false #*********************************************************************