diff --git a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java index ee529a3b..6abe4dc8 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyAcceptor.java @@ -18,12 +18,18 @@ import io.netty.handler.codec.http.HttpRequestDecoder; import io.netty.handler.codec.http.HttpResponseEncoder; import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; +import io.netty.handler.ssl.SslHandler; import io.netty.handler.timeout.IdleStateHandler; import java.io.IOException; +import java.io.InputStream; +import java.security.KeyStore; import java.util.List; import java.util.Properties; +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLEngine; import org.dna.mqtt.commons.Constants; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.parser.netty.MQTTDecoder; @@ -76,6 +82,7 @@ public class NettyAcceptor implements ServerAcceptor { initializePlainTCPTransport(messaging, props); initializeWebSocketTransport(messaging, props); + initializeSSLTCPTransport(messaging, props); } private void initializePlainTCPTransport(IMessaging messaging, Properties props) throws IOException { @@ -117,9 +124,6 @@ public class NettyAcceptor implements ServerAcceptor { } private void initializeWebSocketTransport(IMessaging messaging, Properties props) throws IOException { -// m_bossGroup = new NioEventLoopGroup(); -// m_workerGroup = new NioEventLoopGroup(); - final NettyMQTTHandler handler = new NettyMQTTHandler(); handler.setMessaging(messaging); @@ -157,6 +161,85 @@ public class NettyAcceptor implements ServerAcceptor { LOG.error(null, ex); } } + + + private void initializeSSLTCPTransport(IMessaging messaging, Properties props) throws IOException { + String sslPortProp = props.getProperty("ssl_port"); + if (sslPortProp == null) { + //Do nothing no SSL configured + LOG.info("SSL is disabled"); + return; + } + final String jksPath = props.getProperty("jks_path"); + if (jksPath == null || jksPath.isEmpty()) { + //key_store_password or key_manager_password are empty + LOG.warn("You have configured the SSL port but not the jks_path, SSL not started"); + return; + } + + //if we have the port also the jks then keyStorePassword and keyManagerPassword + //has to be defined + final String keyStorePassword = props.getProperty("key_store_password"); + final String keyManagerPassword = props.getProperty("key_manager_password"); + if (keyStorePassword == null || keyStorePassword.isEmpty()) { + //key_store_password or key_manager_password are empty + LOG.warn("You have configured the SSL port but not the key_store_password, SSL not started"); + return; + } + if (keyManagerPassword == null || keyManagerPassword.isEmpty()) { + //key_manager_password or key_manager_password are empty + LOG.warn("You have configured the SSL port but not the key_manager_password, SSL not started"); + return; + } + + int sslPort = Integer.parseInt(sslPortProp); + + final NettyMQTTHandler handler = new NettyMQTTHandler(); + handler.setMessaging(messaging); + + ServerBootstrap b = new ServerBootstrap(); + b.group(m_bossGroup, m_workerGroup) + .channel(NioServerSocketChannel.class) + .childHandler(new ChannelInitializer() { + @Override + public void initChannel(SocketChannel ch) throws Exception { + InputStream jksInputStream = getClass().getClassLoader().getResourceAsStream(jksPath); + SSLContext serverContext = SSLContext.getInstance("TLS"); + final KeyStore ks = KeyStore.getInstance("JKS"); + ks.load(jksInputStream, keyStorePassword.toCharArray()); + final KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); + kmf.init(ks, keyManagerPassword.toCharArray()); + serverContext.init(kmf.getKeyManagers(), null, null); + + SSLEngine engine = serverContext.createSSLEngine(); + engine.setUseClientMode(false); + final SslHandler sslHandler = new SslHandler(engine); + + ChannelPipeline pipeline = ch.pipeline(); + pipeline.addLast("ssl", sslHandler); + //pipeline.addFirst("metrics", new BytesMetricsHandler(m_metricsCollector)); + pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT)); + pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler()); + //pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR)); + pipeline.addLast("decoder", new MQTTDecoder()); + pipeline.addLast("encoder", new MQTTEncoder()); + pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector)); + pipeline.addLast("handler", handler); + } + }) + .option(ChannelOption.SO_BACKLOG, 128) + .option(ChannelOption.SO_REUSEADDR, true) + .childOption(ChannelOption.SO_KEEPALIVE, true); + try { + // Bind and start to accept incoming connections. +// ChannelFuture f = b.bind(Constants.PORT); + ChannelFuture f = b.bind(props.getProperty("host"), sslPort); + LOG.info("Server binded"); + f.sync(); + } catch (InterruptedException ex) { + LOG.error(null, ex); + } + } public void close() { if (m_workerGroup == null) { diff --git a/broker/src/main/resources/log4j.properties b/broker/src/main/resources/log4j.properties index c018fbce..bcce4891 100644 --- a/broker/src/main/resources/log4j.properties +++ b/broker/src/main/resources/log4j.properties @@ -4,23 +4,23 @@ log4j.rootLogger=ERROR, stdout, file log4j.logger.org.dna.mqtt=INFO log4j.logger.org.dna.mqtt.moquette.server.ServerIntegrationFuseTest=DEBUG -log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=INFO -#log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=DEBUG +#log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=INFO +log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=DEBUG #log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=TRACE log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging=WARN #Protocol parsing log4j.logger.org.dna.mqtt.moquette.server.netty.NettyMQTTHandler=WARN #log4j.logger.org.dna.mqtt.moquette.server.netty.NettyMQTTHandler=DEBUG log4j.logger.org.dna.mqtt.moquette.parser.netty=WARN -#Storage server -#log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore=DEBUG -#log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.HawtDBStorageService=DEBUG +#Storage service +log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore=DEBUG +log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.HawtDBStorageService=DEBUG # stdout appender is set to be a ConsoleAppender. log4j.appender.stdout=org.apache.log4j.ConsoleAppender -log4j.appender.stdout.Threshold=WARN +#log4j.appender.stdout.Threshold=WARN #log4j.appender.stdout.Threshold=INFO -#log4j.appender.stdout.Threshold=DEBUG +log4j.appender.stdout.Threshold=DEBUG #log4j.appender.stdout.Threshold=TRACE log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationSSLTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationSSLTest.java new file mode 100644 index 00000000..e9a0577c --- /dev/null +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationSSLTest.java @@ -0,0 +1,134 @@ +package org.dna.mqtt.moquette.server; + +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.security.KeyManagementException; +import java.security.KeyStore; +import java.security.KeyStoreException; +import java.security.NoSuchAlgorithmException; +import java.security.UnrecoverableKeyException; +import java.security.cert.CertificateException; +import javax.net.SocketFactory; +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLServerSocketFactory; +import javax.net.ssl.SSLSocketFactory; +import javax.net.ssl.TrustManager; +import javax.net.ssl.TrustManagerFactory; +import org.eclipse.paho.client.mqttv3.IMqttClient; +import org.eclipse.paho.client.mqttv3.MqttClient; +import org.eclipse.paho.client.mqttv3.MqttClientPersistence; +import org.eclipse.paho.client.mqttv3.MqttConnectOptions; +import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; +import org.junit.After; +import static org.junit.Assert.assertFalse; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Check that Moquette could also handle SSL. + * + * @author andrea + */ +public class ServerIntegrationSSLTest { + private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationSSLTest.class); + + Server m_server; + static MqttClientPersistence s_dataStore; + + IMqttClient m_client; + TestCallback m_callback; + + @BeforeClass + public static void beforeTests() { + String tmpDir = System.getProperty("java.io.tmpdir"); + s_dataStore = new MqttDefaultFilePersistence(tmpDir); + } + + protected void startServer() throws IOException { + String file = getClass().getResource("/").getPath(); + System.setProperty("moquette.path", file); + m_server = new Server(); + m_server.startServer(); + } + + @Before + public void setUp() throws Exception { + File dbFile = new File(Server.STORAGE_FILE_PATH); + assertFalse(String.format("The DB storagefile %s already exists", Server.STORAGE_FILE_PATH), dbFile.exists()); + + startServer(); + + m_client = new MqttClient("ssl://localhost:8883", "TestClient", s_dataStore); +// m_client = new MqttClient("ssl://test.mosquitto.org:8883", "TestClient", s_dataStore); + + m_callback = new TestCallback(); + m_client.setCallback(m_callback); + } + + @After + public void tearDown() throws Exception { + if (m_client.isConnected()) { + m_client.disconnect(); + } + + m_server.stopServer(); + File dbFile = new File(Server.STORAGE_FILE_PATH); + if (dbFile.exists()) { + dbFile.delete(); + } + assertFalse(dbFile.exists()); + } + + @Test + public void checkSupportSSL() throws Exception { + LOG.info("*** checkSupportSSL ***"); + SSLSocketFactory ssf = configureSSLSocketFactory(); + + MqttConnectOptions options = new MqttConnectOptions(); + options.setSocketFactory(ssf); + m_client.connect(options); + m_client.subscribe("/topic", 0); + m_client.disconnect(); + } + + /** + * keystore generated into test/resources with command: + * + * keytool -keystore clientkeystore.jks -alias testclient -genkey -keyalg RSA + * -> mandatory to put the name surname + * -> password is passw0rd + * -> type yes at the end + * + * to generate the crt file from the keystore + * -- keytool -certreq -alias testclient -keystore clientkeystore.jks -file testclient.csr + * + * keytool -export -alias testclient -keystore clientkeystore.jks -file testclient.crt + * + * to import an existing certificate: + * keytool -keystore clientkeystore.jks -import -alias testclient -file testclient.crt -trustcacerts + */ + private SSLSocketFactory configureSSLSocketFactory() throws KeyManagementException, NoSuchAlgorithmException, UnrecoverableKeyException, IOException, CertificateException, KeyStoreException { + KeyStore ks = KeyStore.getInstance("JKS"); + InputStream jksInputStream = getClass().getClassLoader().getResourceAsStream("clientkeystore.jks"); + ks.load(jksInputStream, "passw0rd".toCharArray()); + + KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); + kmf.init(ks, "passw0rd".toCharArray()); + + TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); + tmf.init(ks); + + SSLContext sc = SSLContext.getInstance("TLS"); + TrustManager[] trustManagers = tmf.getTrustManagers(); + sc.init(kmf.getKeyManagers(), trustManagers, null); + + SSLSocketFactory ssf = sc.getSocketFactory(); + return ssf; + } +} diff --git a/broker/src/test/resources/clientkeystore.jks b/broker/src/test/resources/clientkeystore.jks new file mode 100644 index 00000000..422f4b24 Binary files /dev/null and b/broker/src/test/resources/clientkeystore.jks differ diff --git a/broker/src/test/resources/config/moquette.conf b/broker/src/test/resources/config/moquette.conf new file mode 100644 index 00000000..d10166e6 --- /dev/null +++ b/broker/src/test/resources/config/moquette.conf @@ -0,0 +1,20 @@ +############################################## +# Moquette configuration file. +# +# The synthax is equals to mosquitto.conf +# +############################################## + +port 1883 + +websocket_port 8080 + +host 0.0.0.0 + +#Password file +password_file password_file.conf + +ssl_port 8883 +jks_path serverkeystore.jks +key_store_password passw0rdsrv +key_manager_password passw0rdsrv diff --git a/broker/src/test/resources/config/password_file.conf b/broker/src/test/resources/config/password_file.conf new file mode 100644 index 00000000..de74f4d7 --- /dev/null +++ b/broker/src/test/resources/config/password_file.conf @@ -0,0 +1 @@ +testuser:passwd \ No newline at end of file diff --git a/broker/src/test/resources/serverkeystore.jks b/broker/src/test/resources/serverkeystore.jks new file mode 100644 index 00000000..f3120450 Binary files /dev/null and b/broker/src/test/resources/serverkeystore.jks differ