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

Merge branch 'master' of github.com:andsel/moquette

Browse Source
This commit is contained in:
andsel
2016-09-12 12:29:01 +02:00
parent 55ea2d7c12 40f2d87434
commit 181303b58b
7 changed files with 122 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/config/moquette.conf
+5 -1
View File
@@ -15,4 +15,8 @@ password_file config/password_file.conf
#false to accept only client connetions with credentials
#true to accept client connection without credentails, validating only the one that provides
allow_anonymous true
allow_anonymous true
#false to prohibit clients from connecting without a clientid.
#true to allow clients to connect without a clientid. One will be generated for them.
allow_zero_byte_client_id false
broker/src/main/java/io/moquette/BrokerConstants.java
+1
View File
@@ -36,6 +36,7 @@ public 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 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";
public static final String AUTHENTICATOR_CLASS_NAME = "authenticator_class";
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+36 -7
View File
@@ -103,6 +103,7 @@ public class ProtocolProcessor {
protected ConcurrentMap<String, ConnectionDescriptor> m_clientIDs;
private SubscriptionsStore subscriptions;
private boolean allowAnonymous;
private boolean allowZeroByteClientId;
private IAuthorizator m_authorizator;
private IMessagesStore m_messagesStore;
private ISessionsStore m_sessionsStore;
@@ -120,8 +121,24 @@ public class ProtocolProcessor {
ISessionsStore sessionsStore,
IAuthenticator authenticator,
boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor) {
init(subscriptions,storageService,sessionsStore,authenticator,allowAnonymous,authorizator,interceptor,null);
init(subscriptions,storageService,sessionsStore,authenticator,allowAnonymous, false, authorizator,interceptor,null);
}
public void init(SubscriptionsStore subscriptions, IMessagesStore storageService,
ISessionsStore sessionsStore,
IAuthenticator authenticator,
boolean allowAnonymous,
boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor) {
init(subscriptions,storageService,sessionsStore,authenticator,allowAnonymous, allowZeroByteClientId, authorizator,interceptor,null);
}
void init(SubscriptionsStore subscriptions, IMessagesStore storageService,
ISessionsStore sessionsStore,
IAuthenticator authenticator,
boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) {
init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, false, authorizator, interceptor, serverPort);
}
/**
* @param subscriptions the subscription store where are stored all the existing
* clients subscriptions.
@@ -130,6 +147,7 @@ public class ProtocolProcessor {
* @param sessionsStore the clients sessions store, used to persist subscriptions.
* @param authenticator the authenticator used in connect messages.
* @param allowAnonymous true connection to clients without credentials.
* @param allowZeroByteClientId true to allow clients connect without a clientid
* @param authorizator used to apply ACL policies to publishes and subscriptions.
* @param interceptor to notify events to an intercept handler
* @param serverPort
@@ -137,11 +155,13 @@ public class ProtocolProcessor {
void init(SubscriptionsStore subscriptions, IMessagesStore storageService,
ISessionsStore sessionsStore,
IAuthenticator authenticator,
boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) {
boolean allowAnonymous,
boolean allowZeroByteClientId, IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) {
this.m_clientIDs = new ConcurrentHashMap<>();
this.m_interceptor = interceptor;
this.subscriptions = subscriptions;
this.allowAnonymous = allowAnonymous;
this.allowZeroByteClientId = allowZeroByteClientId;
m_authorizator = authorizator;
LOG.trace("subscription tree on init {}", subscriptions.dumpTree());
m_authenticator = authenticator;
@@ -162,11 +182,18 @@ public class ProtocolProcessor {
}
if (msg.getClientID() == null || msg.getClientID().length() == 0) {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
channel.writeAndFlush(okResp);
m_interceptor.notifyClientConnected(msg);
return;
if(!msg.isCleanSession() || !this.allowZeroByteClientId) {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
channel.writeAndFlush(okResp);
channel.close();
return;
}
// Generating client id.
String randomIdentifier = UUID.randomUUID().toString().replace("-", "");
msg.setClientID(randomIdentifier);
LOG.info("Client connected with server generated identifier: {}", randomIdentifier);
}
//handle user authentication
@@ -280,6 +307,8 @@ public class ProtocolProcessor {
ConnAckMessage okResp = new ConnAckMessage();
okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD);
session.writeAndFlush(okResp);
session.close();
LOG.info("Client {} failed to connect with bad username or password.", session);
}
/**
broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java
+2 -1
View File
@@ -139,7 +139,8 @@ public class ProtocolProcessorBootstrapper {
}
boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, "true"));
m_processor.init(subscriptions, messagesStore, sessionsStore, authenticator, allowAnonymous, authorizator, m_interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME));
boolean allowZeroByteClientId = Boolean.parseBoolean(props.getProperty(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, "false"));
m_processor.init(subscriptions, messagesStore, sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId, authorizator, m_interceptor, props.getProperty(BrokerConstants.PORT_PROPERTY_NAME));
return m_processor;
}
broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java
+1 -1
View File
@@ -63,9 +63,9 @@ public class ServerRestartIntegrationTest {
@Before
public void setUp() throws Exception {
startServer();
String dbPath = IntegrationUtils.localMapDBPath();
IntegrationUtils.cleanPersistenceFile(dbPath);
startServer();
m_subscriber = new MqttClient("tcp://localhost:1883", "Subscriber", s_dataStore);
m_messageCollector = new MessageCollector();
broker/src/test/java/io/moquette/spi/impl/NettyChannelAssertions.java
+10 -2
View File
@@ -30,9 +30,17 @@ import static org.junit.Assert.assertTrue;
class NettyChannelAssertions {
static void assertEqualsConnAck(byte expectedCode, Object connAck) {
assertTrue(connAck instanceof ConnAckMessage);
assertEqualsConnAck(null, expectedCode, connAck);
}
static void assertEqualsConnAck(String msg, byte expectedCode, Object connAck) {
assertTrue("connAck is not an instance of ConnAckMessage", connAck instanceof ConnAckMessage);
ConnAckMessage connAckMsg = (ConnAckMessage) connAck;
assertEquals(expectedCode, connAckMsg.getReturnCode());
if(msg == null)
assertEquals(expectedCode, connAckMsg.getReturnCode());
else
assertEquals(msg, expectedCode, connAckMsg.getReturnCode());
}
static void assertConnAckAccepted(EmbeddedChannel channel) {
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java
+67 -3
View File
@@ -35,9 +35,7 @@ import java.util.Map;
import java.util.Set;
import static io.moquette.parser.netty.Utils.VERSION_3_1_1;
import static io.moquette.parser.proto.messages.ConnAckMessage.BAD_USERNAME_OR_PASSWORD;
import static io.moquette.parser.proto.messages.ConnAckMessage.CONNECTION_ACCEPTED;
import static io.moquette.parser.proto.messages.ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION;
import static io.moquette.parser.proto.messages.ConnAckMessage.*;
import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsConnAck;
import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsSubAck;
import static org.junit.Assert.assertFalse;
@@ -95,6 +93,7 @@ public class ProtocolProcessor_CONNECT_Test {
//Verify
assertEqualsConnAck(UNNACEPTABLE_PROTOCOL_VERSION, m_session.readOutbound());
assertFalse("Connection should be closed by the broker.", m_session.isOpen());
}
@Test
@@ -122,6 +121,7 @@ public class ProtocolProcessor_CONNECT_Test {
//Verify
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen());
//TODO verify the call
/*verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()),
any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));*/
@@ -140,6 +140,7 @@ public class ProtocolProcessor_CONNECT_Test {
//Verify
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen());
}
@Test
@@ -154,6 +155,7 @@ public class ProtocolProcessor_CONNECT_Test {
//Verify
assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, m_session.readOutbound());
assertFalse("Connection should be closed by the broker.", m_session.isOpen());
}
@Test
@@ -169,6 +171,7 @@ public class ProtocolProcessor_CONNECT_Test {
//Verify
assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, m_session.readOutbound());
assertFalse("Connection should be closed by the broker.", m_session.isOpen());
}
@Test
@@ -182,6 +185,7 @@ public class ProtocolProcessor_CONNECT_Test {
//Verify
assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, m_session.readOutbound());
assertFalse("Connection should be closed by the broker.", m_session.isOpen());
}
@Test
@@ -197,6 +201,7 @@ public class ProtocolProcessor_CONNECT_Test {
//Verify
assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, m_session.readOutbound());
assertFalse("Connection should be closed by the broker.", m_session.isOpen());
}
@Test
@@ -210,6 +215,7 @@ public class ProtocolProcessor_CONNECT_Test {
//Verify
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen());
}
@Test
@@ -294,6 +300,7 @@ public class ProtocolProcessor_CONNECT_Test {
//Verify
assertEqualsConnAck(CONNECTION_ACCEPTED, firstReceiverSession.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", firstReceiverSession.isOpen());
}
@@ -306,6 +313,7 @@ public class ProtocolProcessor_CONNECT_Test {
connMsg.setCleanSession(false);
m_processor.processConnect(m_session, connMsg);
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen());
//subscribe
SubscribeMessage subscribeMsg = new SubscribeMessage();
@@ -327,8 +335,64 @@ public class ProtocolProcessor_CONNECT_Test {
m_session = new EmbeddedChannel();
m_processor.processConnect(m_session, connMsg);
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen());
//verify that the first subscription is still preserved
assertTrue(subscriptions.contains(expectedSubscription));
}
@Test
public void testZeroByteClientIdWithoutCleanSession() {
// Allow zero byte client ids
m_processor = new ProtocolProcessor();
m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, true,
new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR);
// Connect message without clean session set to true but client id is still null
connMsg = new ConnectMessage();
connMsg.setProtocolVersion(VERSION_3_1_1);
connMsg.setClientID(null);
connMsg.setCleanSession(false);
m_processor.processConnect(m_session, connMsg);
assertEqualsConnAck("Identifier should be rejected due to having clean session set to false.",
IDENTIFIER_REJECTED, m_session.readOutbound());
assertFalse("Connection should be closed by the broker.", m_session.isOpen());
}
@Test
public void testZeroByteClientIdWithCleanSession() {
// Allow zero byte client ids
m_processor = new ProtocolProcessor();
m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, true,
new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR);
// Connect message with clean session set to true and client id is null.
connMsg = new ConnectMessage();
connMsg.setProtocolVersion(VERSION_3_1_1);
connMsg.setClientID(null);
connMsg.setCleanSession(true);
m_processor.processConnect(m_session, connMsg);
assertEqualsConnAck("Connection should be accepted. A unique clientid should be generated" +
" and clean session set to true.", CONNECTION_ACCEPTED, m_session.readOutbound());
assertTrue("Connection should be valid and open,", m_session.isOpen());
}
@Test
public void testZeroByteClientIdNotAllowed() {
// Connect message with clean session set to true and client id is null.
connMsg = new ConnectMessage();
connMsg.setProtocolVersion(VERSION_3_1_1);
connMsg.setClientID(null);
connMsg.setCleanSession(true);
m_processor.processConnect(m_session, connMsg);
assertEqualsConnAck("Zero byte client identifiers are not allowed.",
IDENTIFIER_REJECTED, m_session.readOutbound());
assertFalse("Connection should closed.", m_session.isOpen());
}
}