https://issues.apache.org/jira/browse/AMQ-6021 - mqtt+nio+ssl certificate authentication

activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTNIOSSLTransportFactory.java

@@ -31,9 +31,11 @@ import javax.net.ssl.SSLEngine;
 import org.apache.activemq.broker.SslContext;
 import org.apache.activemq.transport.Transport;
 import org.apache.activemq.transport.TransportServer;
+import org.apache.activemq.transport.nio.NIOSSLTransportServer;
 import org.apache.activemq.transport.tcp.TcpTransport;
 import org.apache.activemq.transport.tcp.TcpTransport.InitBuffer;
 import org.apache.activemq.transport.tcp.TcpTransportServer;
+import org.apache.activemq.util.IntrospectionSupport;
 import org.apache.activemq.wireformat.WireFormat;
 
 public class MQTTNIOSSLTransportFactory extends MQTTNIOTransportFactory {
@@ -42,13 +44,17 @@ public class MQTTNIOSSLTransportFactory extends MQTTNIOTransportFactory {
 
     @Override
     protected TcpTransportServer createTcpTransportServer(URI location, ServerSocketFactory serverSocketFactory) throws IOException, URISyntaxException {
-        TcpTransportServer result = new TcpTransportServer(this, location, serverSocketFactory) {
+        NIOSSLTransportServer result = new NIOSSLTransportServer(context, this, location, serverSocketFactory) {
             @Override
             protected Transport createTransport(Socket socket, WireFormat format) throws IOException {
                 MQTTNIOSSLTransport transport = new MQTTNIOSSLTransport(format, socket);
                 if (context != null) {
                     transport.setSslContext(context);
                 }
+
+                transport.setNeedClientAuth(isNeedClientAuth());
+                transport.setWantClientAuth(isWantClientAuth());
+
                 return transport;
             }
         };

activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTTransportFilter.java

@@ -27,6 +27,7 @@ import org.apache.activemq.command.Command;
 import org.apache.activemq.transport.Transport;
 import org.apache.activemq.transport.TransportFilter;
 import org.apache.activemq.transport.TransportListener;
+import org.apache.activemq.transport.nio.NIOSSLTransport;
 import org.apache.activemq.transport.tcp.SslTransport;
 import org.apache.activemq.util.IOExceptionSupport;
 import org.apache.activemq.wireformat.WireFormat;
@@ -165,14 +166,17 @@ public class MQTTTransportFilter extends TransportFilter implements MQTTTranspor
 
     @Override
     public X509Certificate[] getPeerCertificates() {
+        X509Certificate[] peerCerts = null;
         if (next instanceof SslTransport) {
-            X509Certificate[] peerCerts = ((SslTransport) next).getPeerCertificates();
+            peerCerts = ((SslTransport) next).getPeerCertificates();
-            if (trace && peerCerts != null) {
-                LOG.debug("Peer Identity has been verified\n");
-            }
-            return peerCerts;
         }
-        return null;
+        if (next instanceof  NIOSSLTransport) {
+            peerCerts = ((NIOSSLTransport)next).getPeerCertificates();
+        }
+        if (trace && peerCerts != null) {
+            LOG.debug("Peer Identity has been verified\n");
+        }
+        return peerCerts;
     }
 
     public boolean isTrace() {

activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4133Test.java

@@ -16,22 +16,23 @@
  */
 package org.apache.activemq.bugs;
 
-import java.net.Socket;
-
-import javax.net.SocketFactory;
-import javax.net.ssl.SSLSocketFactory;
-
 import junit.framework.TestCase;
-
 import org.apache.activemq.broker.BrokerFactory;
 import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.spring.SpringSslContext;
 import org.apache.activemq.transport.stomp.Stomp;
 import org.apache.activemq.transport.stomp.StompConnection;
 import org.apache.activemq.transport.stomp.StompFrame;
+import org.fusesource.mqtt.client.MQTT;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
+import javax.net.SocketFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import java.net.Socket;
+
 public class AMQ4133Test {
 
     protected String java_security_auth_login_config = "java.security.auth.login.config";
@@ -53,6 +54,13 @@ public class AMQ4133Test {
 
         broker.start();
         broker.waitUntilStarted();
+
+        System.setProperty("javax.net.ssl.trustStore", certBase + "/" + "broker1.ks");
+        System.setProperty("javax.net.ssl.trustStorePassword", "password");
+        System.setProperty("javax.net.ssl.trustStoreType", "jks");
+        System.setProperty("javax.net.ssl.keyStore", certBase + "/" + "client.ks");
+        System.setProperty("javax.net.ssl.keyStorePassword", "password");
+        System.setProperty("javax.net.ssl.keyStoreType", "jks");
     }
 
     @After
@@ -83,14 +91,17 @@ public class AMQ4133Test {
         stompConnectTo("localhost", broker.getConnectorByName("stomp+nio+ssl+special").getConnectUri().getPort());
     }
 
-    public Socket createSocket(String host, int port) throws Exception {
+    @Test
-        System.setProperty("javax.net.ssl.trustStore", certBase + "/" + "broker1.ks");
+    public void mqttSSLNeedClientAuthTrue() throws Exception {
-        System.setProperty("javax.net.ssl.trustStorePassword", "password");
+        mqttConnectTo("localhost", broker.getConnectorByName("mqtt+ssl").getConnectUri().getPort());
-        System.setProperty("javax.net.ssl.trustStoreType", "jks");
+    }
-        System.setProperty("javax.net.ssl.keyStore", certBase + "/" + "client.ks");
-        System.setProperty("javax.net.ssl.keyStorePassword", "password");
-        System.setProperty("javax.net.ssl.keyStoreType", "jks");
 
+    @Test
+    public void mqttNIOSSLNeedClientAuthTrue() throws Exception {
+        mqttConnectTo("localhost", broker.getConnectorByName("mqtt+nio+ssl").getConnectUri().getPort());
+    }
+
+    public Socket createSocket(String host, int port) throws Exception {
         SocketFactory factory = SSLSocketFactory.getDefault();
         return factory.createSocket(host, port);
     }
@@ -104,4 +115,23 @@ public class AMQ4133Test {
         stompConnection.close();
     }
 
+    public void mqttConnectTo(String host, int port) throws Exception {
+        MQTT mqtt = new MQTT();
+        mqtt.setConnectAttemptsMax(1);
+        mqtt.setReconnectAttemptsMax(0);
+        mqtt.setHost("tls://" + host + ":" + port);
+        mqtt.setClientId("test");
+        mqtt.setCleanSession(true);
+
+        SpringSslContext context = new SpringSslContext();
+        context.setKeyStore(certBase + "/" + "client.ks");
+        context.setKeyStorePassword("password");
+        context.setTrustStore(certBase + "/" + "broker1.ks");
+        context.setTrustStorePassword("password");
+        context.afterPropertiesSet();
+
+        mqtt.setSslContext(SSLContext.getDefault());
+        mqtt.blockingConnection().connect();
+    }
+
 }

activemq-unit-tests/src/test/resources/org/apache/activemq/bugs/amq4126/InconsistentConnectorPropertiesBehaviour.xml

@@ -40,6 +40,8 @@
       <transportConnector name="stomp+ssl" uri="stomp+ssl://0.0.0.0:0?transport.needClientAuth=true" />
       <transportConnector name="stomp+nio+ssl+special" uri="stomp+nio+ssl://0.0.0.0:0?needClientAuth=true" />
       <transportConnector name="stomp+nio+ssl" uri="stomp+nio+ssl://0.0.0.0:0?transport.needClientAuth=true" />
+      <transportConnector name="mqtt+ssl" uri="mqtt+ssl://0.0.0.0:0?transport.needClientAuth=true" />
+      <transportConnector name="mqtt+nio+ssl" uri="mqtt+nio+ssl://0.0.0.0:0?transport.needClientAuth=true" />
     </transportConnectors>
 
   </broker>