diff --git a/ChangeLog.txt b/ChangeLog.txt index 21031042..58fe477c 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -6,6 +6,7 @@ Version 0.5: Fixed issue #31 missed to insert the hawtdb jars Added a separate mvn module to host the OSGi bundle declaration Fixed issue #30 + Added password file for identity management on connect (Fix #23) Version 0.4: Dropped support for MINA diff --git a/broker/config/moquette.conf b/broker/config/moquette.conf new file mode 100644 index 00000000..d50351c7 --- /dev/null +++ b/broker/config/moquette.conf @@ -0,0 +1,12 @@ +############################################## +# Moquette configuration file. +# +# The synthax is equals to mosquitto.conf +# +############################################## + +port 1883 + +host 0.0.0.0 + +password_file config/password_file.conf diff --git a/broker/config/password_file.conf b/broker/config/password_file.conf new file mode 100644 index 00000000..de74f4d7 --- /dev/null +++ b/broker/config/password_file.conf @@ -0,0 +1 @@ +testuser:passwd \ No newline at end of file diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/FileAuthenticator.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/FileAuthenticator.java new file mode 100644 index 00000000..df6b0b9d --- /dev/null +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/FileAuthenticator.java @@ -0,0 +1,85 @@ +package org.dna.mqtt.moquette.messaging.spi.impl; + +import java.io.BufferedReader; +import java.io.File; +import java.io.FileNotFoundException; +import java.io.FileReader; +import java.io.IOException; +import java.io.Reader; +import java.text.ParseException; +import java.util.HashMap; +import java.util.Map; +import org.dna.mqtt.moquette.server.IAuthenticator; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * + * @author andrea + */ +public class FileAuthenticator implements IAuthenticator { + + private static final Logger LOG = LoggerFactory.getLogger(FileAuthenticator.class); + + private Map m_identities = new HashMap(); + + FileAuthenticator(String filePath) { + LOG.info("Loading password file: " + filePath); + File file = new File(filePath); + try { + FileReader reader = new FileReader(file); + parse(reader); + } catch (FileNotFoundException fex) { + LOG.warn(String.format("Parsing not existing file %s", filePath), fex); + } catch (ParseException pex) { + LOG.warn(String.format("Fromat ero in parsing password file %s", filePath), pex); + } + } + + private void parse(Reader reader) throws ParseException { + if (reader == null) { + return; + } + + BufferedReader br = new BufferedReader(reader); + String line; + try { + while ((line = br.readLine()) != null) { + int commentMarker = line.indexOf('#'); + if (commentMarker != -1) { + if (commentMarker == 0) { + //skip its a comment + continue; + } else { + //it's a malformed comment + throw new ParseException(line, commentMarker); + } + } else { + if (line.isEmpty() || line.matches("^\\s*$")) { + //skip it's a black line + continue; + } + + //split till the first space + int deilimiterIdx = line.indexOf(':'); + String username = line.substring(0, deilimiterIdx).trim(); + String password = line.substring(deilimiterIdx + 1).trim(); + + m_identities.put(username, password); + } + } + } catch (IOException ex) { + throw new ParseException("Failed to read", 1); + } + } + + public boolean checkValid(String username, String password) { + String foundPwq = m_identities.get(username); + if (foundPwq == null) { + return false; + } + + return foundPwq.equals(password); + } + +} diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 0733ac81..c4613d35 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -65,10 +65,13 @@ class ProtocolProcessor implements EventHandler { * clients subscriptions. * @param storageService the persistent store to use for save/load of messages * for QoS1 and QoS2 handling. + * @param authenticator the authenticator used in connect messages */ - void init(SubscriptionsStore subscriptions, IStorageService storageService) { + void init(SubscriptionsStore subscriptions, IStorageService storageService, + IAuthenticator authenticator) { //m_clientIDs = clientIDs; this.subscriptions = subscriptions; + m_authenticator = authenticator; m_storageService = storageService; //init the output ringbuffer diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index c65a116c..7c30bb03 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -4,6 +4,7 @@ import com.lmax.disruptor.BatchEventProcessor; import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.SequenceBarrier; +import java.util.Properties; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -15,6 +16,7 @@ import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore import org.dna.mqtt.moquette.proto.messages.PubCompMessage; import org.dna.mqtt.moquette.proto.messages.*; import org.dna.mqtt.moquette.server.Constants; +import org.dna.mqtt.moquette.server.IAuthenticator; import org.dna.mqtt.moquette.server.ServerChannel; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -56,7 +58,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { return INSTANCE; } - public void init() { + public void init(Properties configProps) { subscriptions = new SubscriptionsStore(); m_executor = Executors.newFixedThreadPool(1); @@ -68,7 +70,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { m_ringBuffer.setGatingSequences(m_eventProcessor.getSequence()); m_executor.submit(m_eventProcessor); - disruptorPublish(new InitEvent()); + disruptorPublish(new InitEvent(configProps)); } @@ -172,19 +174,23 @@ public class SimpleMessaging implements IMessaging, EventHandler { } } else if (evt instanceof InitEvent) { - processInit(); + processInit(((InitEvent) evt).getConfig()); } else if (evt instanceof LostConnectionEvent) { LostConnectionEvent lostEvt = (LostConnectionEvent) evt; m_processor.proccessConnectionLost(lostEvt.getClientID()); } } - private void processInit() { + private void processInit(Properties props) { m_storageService = new HawtDBStorageService(); m_storageService.initStore(); subscriptions.init(m_storageService); - m_processor.init(subscriptions, m_storageService); + + String path = props.getProperty("password_file"); + IAuthenticator authenticator = new FileAuthenticator(path); + + m_processor.init(subscriptions, m_storageService, authenticator); } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/InitEvent.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/InitEvent.java index 94d7ce63..938ed5fe 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/InitEvent.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/InitEvent.java @@ -1,6 +1,19 @@ package org.dna.mqtt.moquette.messaging.spi.impl.events; +import java.util.Properties; + /** */ -public class InitEvent extends MessagingEvent { +public final class InitEvent extends MessagingEvent { + + private Properties config; + + public InitEvent(Properties config) { + this.config = config; + } + + public Properties getConfig() { + return config; + } + } diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/ConfigurationParser.java b/broker/src/main/java/org/dna/mqtt/moquette/server/ConfigurationParser.java index 094aadac..d37e31b2 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/ConfigurationParser.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/ConfigurationParser.java @@ -8,7 +8,6 @@ import java.io.IOException; import java.io.Reader; import java.text.ParseException; import java.util.Properties; -import java.util.logging.Level; import org.dna.mqtt.commons.Constants; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java index b3eac7d3..33a61f35 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -3,6 +3,7 @@ package org.dna.mqtt.moquette.server; import java.io.File; import java.io.IOException; import java.text.ParseException; +import java.util.Properties; import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging; import org.dna.mqtt.moquette.server.netty.NettyAcceptor; import org.slf4j.Logger; @@ -43,12 +44,13 @@ public class Server { } catch (ParseException pex) { LOG.warn("An error occured in parsing configuration, fallback on deafult configuration", pex); } + Properties configProps = confParser.getProperties(); messaging = SimpleMessaging.getInstance(); - messaging.init(); + messaging.init(configProps); m_acceptor = new NettyAcceptor(); - m_acceptor.initialize(messaging, confParser.getProperties()); + m_acceptor.initialize(messaging, configProps); } public void stopServer() { diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/FileAuthenticatorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/FileAuthenticatorTest.java new file mode 100644 index 00000000..a67973e9 --- /dev/null +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/FileAuthenticatorTest.java @@ -0,0 +1,28 @@ +package org.dna.mqtt.moquette.messaging.spi.impl; + +import org.dna.mqtt.moquette.server.IAuthenticator; +import org.junit.Test; +import static org.junit.Assert.*; + +/** + * + * @author andrea + */ +public class FileAuthenticatorTest { + + @Test + public void loadPasswordFile_verifyValid() { + String file = getClass().getResource("/password_file.conf").getPath(); + IAuthenticator auth = new FileAuthenticator(file); + + assertTrue(auth.checkValid("testuser", "passwd")); + } + + @Test + public void loadPasswordFile_verifyInvalid() { + String file = getClass().getResource("/password_file.conf").getPath(); + IAuthenticator auth = new FileAuthenticator(file); + + assertFalse(auth.checkValid("testuser2", "passwd")); + } +} \ No newline at end of file diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/MockAuthenticator.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/MockAuthenticator.java new file mode 100644 index 00000000..9bb79943 --- /dev/null +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/MockAuthenticator.java @@ -0,0 +1,26 @@ +package org.dna.mqtt.moquette.messaging.spi.impl; + +import java.util.Map; +import org.dna.mqtt.moquette.server.IAuthenticator; + +/** + * Test utility to implements authenticator instance. + * + * @author andrea + */ +class MockAuthenticator implements IAuthenticator { + + private Map m_userPwds; + + MockAuthenticator(Map userPwds) { + m_userPwds = userPwds; + } + + public boolean checkValid(String username, String password) { + if (!m_userPwds.containsKey(username)) { + return false; + } + return m_userPwds.get(username).equals(password); + } + +} diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index 112ca557..a638ee60 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -34,6 +34,9 @@ public class ProtocolProcessorTest { final static String FAKE_PUBLISHER_ID = "Publisher"; final static String FAKE_TOPIC = "/news"; + final static String TEST_USER = "fakeuser"; + final static String TEST_PWD = "fakepwd"; + ServerChannel m_session; byte m_returnCode; ConnectMessage connMsg; @@ -42,6 +45,7 @@ public class ProtocolProcessorTest { IStorageService m_storageService; SubscriptionsStore subscriptions; AbstractMessage m_receivedMessage; + MockAuthenticator m_mockAuthenticator; class DummyChannel implements ServerChannel { @@ -127,11 +131,15 @@ public class ProtocolProcessorTest { m_storageService = new MemoryStorageService(); //m_storageService.initStore(); + + Map users = new HashMap(); + users.put(TEST_USER, TEST_PWD); + m_mockAuthenticator = new MockAuthenticator(users); subscriptions = new SubscriptionsStore(); subscriptions.init(m_storageService); m_processor = new ProtocolProcessor(); - m_processor.init(subscriptions, m_storageService); + m_processor.init(subscriptions, m_storageService, m_mockAuthenticator); } @Test @@ -175,6 +183,51 @@ public class ProtocolProcessorTest { any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));*/ } + @Test + public void validAuthentication() { + connMsg.setClientID("123"); + connMsg.setUserFlag(true); + connMsg.setPasswordFlag(true); + connMsg.setUsername(TEST_USER); + connMsg.setPassword(TEST_PWD); + + //Exercise + m_processor.processConnect(m_session, connMsg); + + //Verify + assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, m_returnCode); + } + + @Test + public void noPasswdAuthentication() { + connMsg.setClientID("123"); + connMsg.setUserFlag(true); + connMsg.setPasswordFlag(false); + connMsg.setUsername(TEST_USER); + + //Exercise + m_processor.processConnect(m_session, connMsg); + + //Verify + assertEquals(ConnAckMessage.BAD_USERNAME_OR_PASSWORD, m_returnCode); + } + + @Test + public void invalidAuthentication() { + connMsg.setClientID("123"); + connMsg.setUserFlag(true); + connMsg.setPasswordFlag(true); + connMsg.setUsername(TEST_USER + "_fake"); + connMsg.setPassword(TEST_PWD); + + //Exercise + m_processor.processConnect(m_session, connMsg); + + //Verify + assertEquals(ConnAckMessage.BAD_USERNAME_OR_PASSWORD, m_returnCode); + } + + @Test public void testPublish() throws InterruptedException { final Subscription subscription = new Subscription(FAKE_CLIENT_ID, @@ -194,7 +247,7 @@ public class ProtocolProcessorTest { //simulate a connect that register a clientID to an IoSession subs.init(m_storageService); - m_processor.init(subs, m_storageService); + m_processor.init(subs, m_storageService, null); ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setProcotolVersion((byte)3); connectMessage.setClientID(FAKE_CLIENT_ID); @@ -233,7 +286,7 @@ public class ProtocolProcessorTest { //simulate a connect that register a clientID to an IoSession subs.init(m_storageService); - m_processor.init(subs, m_storageService); + m_processor.init(subs, m_storageService, null); MockReceiverChannel firstReceiverSession = new MockReceiverChannel(); ConnectMessage connectMessage = new ConnectMessage(); @@ -343,7 +396,7 @@ public class ProtocolProcessorTest { subs.init(new MemoryStorageService()); //simulate a connect that register a clientID to an IoSession - m_processor.init(subs, m_storageService); + m_processor.init(subs, m_storageService, null); ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setClientID(FAKE_PUBLISHER_ID); connectMessage.setProcotolVersion((byte)3); @@ -375,7 +428,7 @@ public class ProtocolProcessorTest { List inactiveSubscriptions = Arrays.asList(inactiveSub); when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); m_processor = new ProtocolProcessor(); - m_processor.init(mockedSubscriptions, m_storageService); + m_processor.init(mockedSubscriptions, m_storageService, null); //Exercise ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); @@ -395,7 +448,7 @@ public class ProtocolProcessorTest { List inactiveSubscriptions = Arrays.asList(inactiveSub); when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); m_processor = new ProtocolProcessor(); - m_processor.init(mockedSubscriptions, m_storageService); + m_processor.init(mockedSubscriptions, m_storageService, null); //Exercise ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java index 97c362bf..26820e23 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java @@ -63,6 +63,21 @@ public class ServerIntegrationFuseTest { } } + @Test + public void connectWithCredetials() throws Exception { + LOG.info("*** connectWithCredetials ***"); + MQTT mqtt = new MQTT(); + mqtt.setHost("localhost", 1883); + mqtt.setClientId("Publisher"); + mqtt.setUserName("testuser"); + mqtt.setPassword("passwd"); + + m_publisher = mqtt.blockingConnection(); + m_publisher.connect(); + + assertTrue(true); + } + @Test public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() throws Exception { LOG.info("*** checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1 ***"); diff --git a/broker/src/test/resources/password_file.conf b/broker/src/test/resources/password_file.conf new file mode 100644 index 00000000..de74f4d7 --- /dev/null +++ b/broker/src/test/resources/password_file.conf @@ -0,0 +1 @@ +testuser:passwd \ No newline at end of file diff --git a/distribution/src/main/assembly/assembly.xml b/distribution/src/main/assembly/assembly.xml index 84544e9f..32070363 100644 --- a/distribution/src/main/assembly/assembly.xml +++ b/distribution/src/main/assembly/assembly.xml @@ -27,6 +27,11 @@ config/ + + src/main/resources/password_file.conf + config/ + + src/main/resources/README.txt diff --git a/distribution/src/main/resources/moquette.conf b/distribution/src/main/resources/moquette.conf index 00894f22..1f7a600c 100644 --- a/distribution/src/main/resources/moquette.conf +++ b/distribution/src/main/resources/moquette.conf @@ -7,4 +7,7 @@ port 1883 -host 0.0.0.0 \ No newline at end of file +host 0.0.0.0 + +#Password file +password_file config/password_file.conf \ No newline at end of file diff --git a/distribution/src/main/resources/password_file.conf b/distribution/src/main/resources/password_file.conf new file mode 100644 index 00000000..de74f4d7 --- /dev/null +++ b/distribution/src/main/resources/password_file.conf @@ -0,0 +1 @@ +testuser:passwd \ No newline at end of file diff --git a/tools_scripts/publisher_client.groovy b/tools_scripts/publisher_client.groovy index 7dbcb557..98915e00 100644 --- a/tools_scripts/publisher_client.groovy +++ b/tools_scripts/publisher_client.groovy @@ -17,7 +17,7 @@ MqttMessage message = new MqttMessage('Hello world!!'.bytes) message.setQos(0) print "publishing.." long startTime = System.currentTimeMillis() -(1..100).each { +(1..10).each { client.publish('topic', 'Hello world!!'.bytes, 0, false) } client.publish('/exit', 'Exit'.bytes, 0, false)