diff --git a/broker/pom.xml b/broker/pom.xml
index c1e22d44..2838e6e9 100644
--- a/broker/pom.xml
+++ b/broker/pom.xml
@@ -137,6 +137,12 @@
9.2.0.M1
test
+
+ com.h2database
+ h2
+ 1.4.191
+ test
+
diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java
index 63cac3e0..6ad87c0f 100644
--- a/broker/src/main/java/io/moquette/BrokerConstants.java
+++ b/broker/src/main/java/io/moquette/BrokerConstants.java
@@ -38,6 +38,10 @@ public class BrokerConstants {
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";
+ public static final String DB_AUTHENTICATOR_DRIVER = "authenticator.db.driver";
+ public static final String DB_AUTHENTICATOR_URL = "authenticator.db.url";
+ public static final String DB_AUTHENTICATOR_QUERY = "authenticator.db.query";
+ public static final String DB_AUTHENTICATOR_DIGEST = "authenticator.db.digest";
public static final int PORT = 1883;
public static final int WEBSOCKET_PORT = 8080;
public static final String HOST = "0.0.0.0";
diff --git a/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java b/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java
index 0f6d96e6..08bd76ce 100644
--- a/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java
+++ b/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java
@@ -105,7 +105,7 @@ public class SimpleMessaging {
String authenticatorClassName = props.getProperty(BrokerConstants.AUTHENTICATOR_CLASS_NAME, "");
if (!authenticatorClassName.isEmpty()) {
- authenticator = (IAuthenticator)loadClass(authenticatorClassName, IAuthenticator.class);
+ authenticator = (IAuthenticator)loadClass(authenticatorClassName, IAuthenticator.class, props);
LOG.info("Loaded custom authenticator {}", authenticatorClassName);
}
@@ -120,7 +120,7 @@ public class SimpleMessaging {
String authorizatorClassName = props.getProperty(BrokerConstants.AUTHORIZATOR_CLASS_NAME, "");
if (!authorizatorClassName.isEmpty()) {
- authorizator = (IAuthorizator)loadClass(authorizatorClassName, IAuthorizator.class);
+ authorizator = (IAuthorizator)loadClass(authorizatorClassName, IAuthorizator.class, props);
LOG.info("Loaded custom authorizator {}", authorizatorClassName);
}
@@ -147,7 +147,7 @@ public class SimpleMessaging {
return m_processor;
}
- private Object loadClass(String className, Class> cls) {
+ private Object loadClass(String className, Class> cls, IConfig props) {
Object instance = null;
try {
Class> clazz = Class.forName(className);
@@ -163,13 +163,25 @@ public class SimpleMessaging {
}
catch (NoSuchMethodException nsmex) {
try {
+ // check if constructor with IConfig parameter exists
instance = this.getClass().getClassLoader()
.loadClass(className)
.asSubclass(cls)
- .newInstance();
+ .getConstructor(IConfig.class).newInstance(props);
} catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) {
LOG.error(null, ex);
throw new RuntimeException("Cannot load custom authenticator class " + className, ex);
+ } catch (NoSuchMethodException | InvocationTargetException e) {
+ try {
+ // fallback to default constructor
+ instance = this.getClass().getClassLoader()
+ .loadClass(className)
+ .asSubclass(cls)
+ .newInstance();
+ } catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) {
+ LOG.error(null, ex);
+ throw new RuntimeException("Cannot load custom authenticator class " + className, ex);
+ }
}
} catch (ClassNotFoundException ex) {
LOG.error(null, ex);
diff --git a/broker/src/main/java/io/moquette/spi/impl/security/DBAuthenticator.java b/broker/src/main/java/io/moquette/spi/impl/security/DBAuthenticator.java
new file mode 100644
index 00000000..0d92faa0
--- /dev/null
+++ b/broker/src/main/java/io/moquette/spi/impl/security/DBAuthenticator.java
@@ -0,0 +1,96 @@
+package io.moquette.spi.impl.security;
+
+import io.moquette.BrokerConstants;
+import io.moquette.server.config.IConfig;
+import io.moquette.spi.security.IAuthenticator;
+import org.apache.commons.codec.binary.Hex;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.sql.*;
+
+/**
+ * Load user credentials from a SQL database.
+ * sql driver must be provided at runtime
+ *
+ * @author mackristof
+ */
+
+public class DBAuthenticator implements IAuthenticator {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DBAuthenticator.class);
+
+ private final String sqlQuery;
+ private final MessageDigest messageDigest;
+ private final PreparedStatement preparedStatement;
+
+
+
+
+ public DBAuthenticator(IConfig conf){
+ this(conf.getProperty(BrokerConstants.DB_AUTHENTICATOR_DRIVER, ""),
+ conf.getProperty(BrokerConstants.DB_AUTHENTICATOR_URL,""),
+ conf.getProperty(BrokerConstants.DB_AUTHENTICATOR_QUERY, ""),
+ conf.getProperty(BrokerConstants.DB_AUTHENTICATOR_DIGEST,""));
+ }
+
+ /** provide authenticator from SQL database
+ * @param driver : jdbc driver class like : "org.postgresql.Driver"
+ * @param jdbcUrl : jdbc url like : "jdbc:postgresql://host:port/dbname"
+ * @param sqlQuery : sql query like : "SELECT PASSWORD FROM USER WHERE LOGIN=?"
+ * @param digestMethod : password encoding algorithm : "MD5", "SHA-1", "SHA-256"
+ */
+ public DBAuthenticator(String driver, String jdbcUrl, String sqlQuery, String digestMethod){
+
+ this.sqlQuery = sqlQuery;
+ try {
+ Class.forName(driver);
+ final Connection connection = DriverManager.getConnection(jdbcUrl);
+ this.messageDigest = MessageDigest.getInstance(digestMethod);
+ this.preparedStatement = connection.prepareStatement(sqlQuery);
+ } catch (ClassNotFoundException cnfe){
+ LOG.error(String.format("Can't find driver %s", driver), cnfe);
+ throw new RuntimeException(cnfe);
+ } catch (SQLException sqle) {
+ LOG.error(String.format("Can't connect to %s", jdbcUrl), sqle);
+ throw new RuntimeException(sqle);
+ } catch (NoSuchAlgorithmException nsaex) {
+ LOG.error(String.format("Can't find %s for password encoding", digestMethod), nsaex);
+ throw new RuntimeException(nsaex);
+ }
+ }
+
+ @Override
+ public synchronized boolean checkValid(String username, byte[] password) {
+ // Check Username / Password in DB using sqlQuery
+ if (username == null || password == null) {
+ LOG.info("username or password was null");
+ return false;
+ }
+ ResultSet r= null;
+ try {
+ this.preparedStatement.setString(1,username);
+ r = this.preparedStatement.executeQuery();
+ if(r.next()) {
+ final String foundPwq = r.getString(1);
+ messageDigest.update(password);
+ byte[] digest = messageDigest.digest();
+ String encodedPasswd = new String(Hex.encodeHex(digest));
+ return foundPwq.equals(encodedPasswd);
+ }
+ r.close();
+ } catch (SQLException e) {
+ e.printStackTrace();
+ }
+ return false;
+ }
+
+ @Override
+ protected void finalize() throws Throwable {
+ this.preparedStatement.close();
+ this.preparedStatement.getConnection().close();
+ super.finalize();
+ }
+}
diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationDBAuthenticatorTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationDBAuthenticatorTest.java
new file mode 100644
index 00000000..d7884991
--- /dev/null
+++ b/broker/src/test/java/io/moquette/server/ServerIntegrationDBAuthenticatorTest.java
@@ -0,0 +1,138 @@
+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.DBAuthenticator;
+import io.moquette.spi.impl.security.DBAuthenticatorTest;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
+import org.junit.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.security.NoSuchAlgorithmException;
+import java.sql.SQLException;
+import java.util.Properties;
+
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Created by mackristof on 13/05/2016.
+ */
+public class ServerIntegrationDBAuthenticatorTest {
+ private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationDBAuthenticatorTest.class);
+
+ static MqttClientPersistence s_dataStore;
+ static MqttClientPersistence s_pubDataStore;
+ static DBAuthenticatorTest dbAuthenticatorTest;
+
+ Server m_server;
+ IMqttClient m_client;
+ IMqttClient m_publisher;
+ MessageCollector m_messagesCollector;
+ IConfig m_config;
+
+
+
+ @BeforeClass
+ public static void beforeTests() throws NoSuchAlgorithmException, SQLException, ClassNotFoundException {
+ String tmpDir = System.getProperty("java.io.tmpdir");
+ s_dataStore = new MqttDefaultFilePersistence(tmpDir);
+ s_pubDataStore = new MqttDefaultFilePersistence(tmpDir + File.separator + "publisher");
+ dbAuthenticatorTest = new DBAuthenticatorTest();
+ dbAuthenticatorTest.setup();
+ }
+
+ protected void startServer() throws IOException {
+ m_server = new Server();
+ final Properties configProps = addDBAuthenticatorConf(IntegrationUtils.prepareTestPropeties());
+ m_config = new MemoryConfig(configProps);
+ m_server.startServer(m_config);
+ }
+
+
+ private void stopServer() {
+ m_server.stopServer();
+ IntegrationUtils.cleanPersistenceFile(m_config);
+ }
+
+ private Properties addDBAuthenticatorConf(Properties properties) {
+ properties.put(BrokerConstants.AUTHENTICATOR_CLASS_NAME, DBAuthenticator.class.getCanonicalName());
+ properties.put(BrokerConstants.DB_AUTHENTICATOR_DRIVER, DBAuthenticatorTest.ORG_H2_DRIVER);
+ properties.put(BrokerConstants.DB_AUTHENTICATOR_URL, DBAuthenticatorTest.JDBC_H2_MEM_TEST);
+ properties.put(BrokerConstants.DB_AUTHENTICATOR_QUERY, "SELECT PASSWORD FROM ACCOUNT WHERE LOGIN=?");
+ properties.put(BrokerConstants.DB_AUTHENTICATOR_DIGEST, DBAuthenticatorTest.SHA_256);
+ return properties;
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ String dbPath = IntegrationUtils.localMapDBPath();
+ IntegrationUtils.cleanPersistenceFile(dbPath);
+
+ startServer();
+
+ m_client = new MqttClient("tcp://localhost:1883", "TestClient", s_dataStore);
+ m_messagesCollector = new MessageCollector();
+ m_client.setCallback(m_messagesCollector);
+
+ m_publisher = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (m_client != null && m_client.isConnected()) {
+ m_client.disconnect();
+ }
+
+ if (m_publisher != null && m_publisher.isConnected()) {
+ m_publisher.disconnect();
+ }
+
+ stopServer();
+ }
+
+ @AfterClass
+ public static void shutdown(){
+ dbAuthenticatorTest.teardown();
+ }
+
+ @Test
+ public void connectWithValidCredentials() throws Exception {
+ LOG.info("*** connectWithCredentials ***");
+ m_client = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore);
+ MqttConnectOptions options = new MqttConnectOptions();
+ options.setUserName("dbuser");
+ options.setPassword("password".toCharArray());
+ m_client.connect(options);
+ assertTrue(true);
+ }
+
+
+ @Test
+ public void connectWithWrongCredentials() {
+ LOG.info("*** connectWithWrongCredentials ***");
+ try {
+ m_client = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore);
+ MqttConnectOptions options = new MqttConnectOptions();
+ options.setUserName("dbuser");
+ options.setPassword("wrongPassword".toCharArray());
+ m_client.connect(options);
+ } catch (MqttException e){
+ if (e instanceof MqttSecurityException) {
+ assertTrue(true);
+ return;
+ } else {
+ assertTrue(e.getMessage(),false);
+ return;
+ }
+ }
+ assertTrue("must not be connected. cause : wrong password given to client",false);
+
+ }
+
+
+}
diff --git a/broker/src/test/java/io/moquette/spi/impl/security/DBAuthenticatorTest.java b/broker/src/test/java/io/moquette/spi/impl/security/DBAuthenticatorTest.java
new file mode 100644
index 00000000..1db62da6
--- /dev/null
+++ b/broker/src/test/java/io/moquette/spi/impl/security/DBAuthenticatorTest.java
@@ -0,0 +1,84 @@
+package io.moquette.spi.impl.security;
+
+import org.apache.commons.codec.binary.Hex;
+import org.junit.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.sql.*;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Created by mackristof on 12/05/2016.
+ */
+public class DBAuthenticatorTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DBAuthenticatorTest.class);
+
+ public static final String ORG_H2_DRIVER = "org.h2.Driver";
+ public static final String JDBC_H2_MEM_TEST = "jdbc:h2:mem:test";
+ public static final String SHA_256 = "SHA-256";
+ private Connection connection;
+
+
+ @Before
+ public void setup() throws ClassNotFoundException, SQLException, NoSuchAlgorithmException {
+ Class.forName(ORG_H2_DRIVER);
+ this.connection = DriverManager.getConnection(JDBC_H2_MEM_TEST);
+ Statement statement = this.connection.createStatement();
+ try {
+ statement.execute("DROP TABLE ACCOUNT");
+ } catch(SQLException sqle) {
+ LOG.info("Table not found, not dropping",sqle);
+ }
+ MessageDigest digest = MessageDigest.getInstance(SHA_256);
+ String hash = new String(Hex.encodeHex(digest.digest("password".getBytes(StandardCharsets.UTF_8))));
+ try {
+ if (statement.execute("CREATE TABLE ACCOUNT ( LOGIN VARCHAR(64), PASSWORD VARCHAR(256))")){
+ throw new SQLException("can't create USER table");
+ }
+ if (statement.execute("INSERT INTO ACCOUNT ( LOGIN , PASSWORD ) VALUES ('dbuser', '"+hash+"')")){
+ throw new SQLException("can't insert in USER table");
+ }
+ } catch(SQLException sqle) {
+ LOG.error("Table not created, not inserted",sqle);
+ return;
+ }
+ LOG.info("Table User created");
+ statement.close();
+
+ }
+
+ @Test
+ public void Db_verifyValid() {
+ final DBAuthenticator dbAuthenticator = new DBAuthenticator(ORG_H2_DRIVER, JDBC_H2_MEM_TEST, "SELECT PASSWORD FROM ACCOUNT WHERE LOGIN=?", SHA_256);
+ assertTrue(dbAuthenticator.checkValid("dbuser","password".getBytes()));
+ }
+
+ @Test
+ public void Db_verifyInvalidLogin(){
+ final DBAuthenticator dbAuthenticator = new DBAuthenticator(ORG_H2_DRIVER, JDBC_H2_MEM_TEST, "SELECT PASSWORD FROM ACCOUNT WHERE LOGIN=?", SHA_256);
+ assertFalse(dbAuthenticator.checkValid("dbuser2","password".getBytes()));
+ }
+
+ @Test
+ public void Db_verifyInvalidPassword(){
+ final DBAuthenticator dbAuthenticator = new DBAuthenticator(ORG_H2_DRIVER, JDBC_H2_MEM_TEST, "SELECT PASSWORD FROM ACCOUNT WHERE LOGIN=?", SHA_256);
+ assertFalse(dbAuthenticator.checkValid("dbuser","wrongPassword".getBytes()));
+ }
+
+
+ @After
+ public void teardown(){
+ try {
+ this.connection.close();
+ } catch (SQLException e) {
+ LOG.error("can't close connection",e);
+ }
+ }
+}
diff --git a/distribution/src/main/resources/moquette.conf b/distribution/src/main/resources/moquette.conf
index 7e2ed9ea..c7217ace 100644
--- a/distribution/src/main/resources/moquette.conf
+++ b/distribution/src/main/resources/moquette.conf
@@ -61,6 +61,15 @@ allow_anonymous true
password_file config/password_file.conf
+#*********************************************************************
+# Optional Database Authentication
+#*********************************************************************
+# authenticator_class io.moquette.spi.impl.security.DBAuthenticator
+# authenticator.db.driver org.postgresql.Driver
+# authenticator.db.url jdbc:postgresql://localhost/test?user=dbuser&password=dbpassword
+# authenticator.db.query SELECT PASSWORD FROM ACCOUNT WHERE LOGIN=?
+# authenticator.db.digest SHA-256
+
#*********************************************************************
# Optional
# authorizator_class: