Added loading of user identities form password_file, Fix issue #23

2014-03-02 18:16:50 +01:00
parent b0d9bc2d6a
commit 1ad5acef7c
18 changed files with 272 additions and 18 deletions
@@ -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);
+    }
+    
+}
@@ -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
@@ -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);
    }


@@ -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;
+    }
+    
 }
@@ -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;
@@ -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() {