Merge branch 'basic_authentication_23'

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

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

broker/config/password_file.conf

@@ -0,0 +1 @@
+testuser:passwd

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<String, String> m_identities = new HashMap<String, String>();
+    
+    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);
+    }
+    
+}

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java

@@ -65,10 +65,13 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
      *  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

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<ValueEvent> {
         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<ValueEvent> {
         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<ValueEvent> {
             }
 
         } 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);
     }
 
 

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;
+    }
+    
 }

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;

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() {

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"));
+    }
+}

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<String, String> m_userPwds;
+    
+    MockAuthenticator(Map<String, String> 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);
+    }
+    
+}

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<String, String> users = new HashMap<String, String>();
+        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<Subscription> 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<Subscription> 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());

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 ***");

broker/src/test/resources/password_file.conf

@@ -0,0 +1 @@
+testuser:passwd

distribution/src/main/assembly/assembly.xml

@@ -27,6 +27,11 @@
             <outputDirectory>config/</outputDirectory>
         </file>
         
+        <file>
+            <source>src/main/resources/password_file.conf</source>
+            <outputDirectory>config/</outputDirectory>
+        </file>
+        
         <file>
             <source>src/main/resources/README.txt</source>
         </file>

distribution/src/main/resources/moquette.conf

@@ -7,4 +7,7 @@
 
 port 1883
 
-host 0.0.0.0
+host 0.0.0.0
+
+#Password file
+password_file config/password_file.conf

distribution/src/main/resources/password_file.conf

@@ -0,0 +1 @@
+testuser:passwd

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)