Performing Security Authentication Using Certificate

/*****************************************************************************
 Description: A java demo of MQTT message sending and receiving. You need to create an edge node and download the client certificate.
 ****************************************************************************/

package com.example.demo;

import javax.net.ssl.SSLSocketFactory;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/********************************************************************
 * MQTT demo shows how the client connects to the MQTT broker of the edge node to send and receive messages and how SSL security authentication is performed on the connection. The demo illustrates the following flows:
* 1. The MQTT subscribe client receives MQTT messages.
* 2. The MQTT publish client sends MQTT messages.
 ********************************************************************/
public class MqttClientDemo {
    private static int QOS_TYPE = 2;
    //MQTT server address
    private static final String MQTT_HOST = "ssl://x.x.x.x:8883";
    //MQTT publish client ID
    private static final String MQTT_PUB_CLIENT_ID = "pub_client_1";
    //MQTT subscribe client ID
    private static final String MQTT_SUB_CLIENT_ID = "sub_client_1";
    //MQTT channel subscription topic
    private static final String TOPIC = "/hello";
    //Paths of the SSL certificate used for MQTT client connection
    public static final String CLIENT_CRT_FILE_PATH = "example_cert.crt";
    public static final String CLIENT_KEY_FILE_PATH = "example_cert.key";
    //MQTT client connection timeout interval (s)
    public static final int TIME_OUT_INTERVAL = 10;
    //Interval at which the MQTT client sends a heartbeat message, in seconds
    public static final int HEART_TIME_INTERVAL = 20;
    //Interval at which the MQTT client retries upon disconnection, in milliseconds
    public static final int RECONNECT_INTERVAL = 10000;
    //Interval at which the MQTT client sends a message, in seconds
    public static final int PUBLISH_MSG_INTERVAL = 3000;

    //MQTT client
    private MqttClient mqttClient;
    //MQTT client ID.
    private String clientId;
    //MQTT client connection options
    private MqttConnectOptions connOpts;
    //Initialized MQTT client has not subscribed to any topic.
    private boolean isSubscribe = false;

    public MqttClientDemo(String id) throws MqttException {
        setClientId(id);
        initMqttClient();
        initCallback();
        initConnectOptions();
        connectMqtt();
    }

    /********************************************************************
     * Sending messages
     * @param message Message to be sent
     * @throws MqttException
     ********************************************************************/
    public void publishMessage(String message) throws MqttException {
        MqttMessage mqttMessage = new MqttMessage(message.getBytes());
        mqttMessage.setQos(QOS_TYPE);
        mqttMessage.setRetained(false);
        mqttClient.publish(TOPIC, mqttMessage);
        System.out.println(String.format("MQTT Client[%s] publish message[%s]", clientId, message));
    }

    /********************************************************************
     * Subscribing to topics
     * @throws MqttException
     ********************************************************************/
    public void subscribeTopic() throws MqttException {
        int[] Qos = {QOS_TYPE};
        String[] topics = {TOPIC};
        mqttClient.subscribe(topics, Qos);
        isSubscribe = true;
    }

    /********************************************************************
     * Starting the thread to periodically send MQTT messages
     * @throws MqttException
     ********************************************************************/
    public void startPublishMessage() {
        new Thread() {
            @Override
            public void run() {
                while (true) {
                    try {
                        Thread.sleep(PUBLISH_MSG_INTERVAL);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    try {
                        publishMessage("hello world!");
                    } catch (MqttException e) {
                        System.out.println(String.format("MQTT client[%s] publish message error,errorMsg[%s]", clientId, e.getMessage()));
                    }
                }
            }
        }.start();
    }

    /********************************************************************
     * Initializing the MQTT client
     * @throws MqttException Abnormal connection
     ********************************************************************/
    private void initMqttClient() throws MqttException {
        MemoryPersistence persistence = new MemoryPersistence();
        mqttClient = new MqttClient(MQTT_HOST, clientId, persistence);
    }

    /********************************************************************
     * Initializing connection options
     * @throws MqttException Abnormal connection
     ********************************************************************/
    private void initConnectOptions() {
        connOpts = new MqttConnectOptions();
        // Specify whether to clear the session. If the value is false, the server retains the client connection records. If the value is true, the client connects to the server as a new client.
        connOpts.setCleanSession(true);
        connOpts.setHttpsHostnameVerificationEnabled(false);
        // Set the timeout interval, in seconds.
        connOpts.setConnectionTimeout(TIME_OUT_INTERVAL);
        // Set the interval at which a session heartbeat message is sent, in seconds. The server sends a message to the client every 1.5 x 20 seconds to check whether the client is online. However, this method does not provide the reconnection mechanism.
        connOpts.setKeepAliveInterval(HEART_TIME_INTERVAL);
        SSLSocketFactory factory = null;
        try {
            factory = SslUtil.getSocketFactory(CLIENT_CRT_FILE_PATH, CLIENT_KEY_FILE_PATH);
        } catch (Exception e) {
            e.printStackTrace();
        }
        // TLS connection configuration
        connOpts.setSocketFactory(factory);
    }

    /********************************************************************
     * Initiate an MQTT connect request.
     * @throws MqttException Abnormal connection
     ********************************************************************/
    private void connectMqtt() throws MqttException {
        mqttClient.connect(connOpts);
        System.out.println(String.format("MQTT client[%s] is connected,the connctOptions: \n%s", clientId, connOpts.toString()));
    }

    /********************************************************************
     * Set the callback API.
     * @throws MqttException Abnormal connection
     ********************************************************************/
    private void initCallback() {
        mqttClient.setCallback(new MqttMessageCallback());
    }

    private void setClientId(String id) {
        clientId = id;
    }

    /********************************************************************
     * MQTT client reconnection function. This function is called to check whether a topic has been subscribed to. If yes, the topic will be re-subscribed to.
     * @throws MqttException
     ********************************************************************/
    private void rconnectMqtt() throws MqttException {
        connectMqtt();
        if (isSubscribe) {
            subscribeTopic();
        }
    }

    /********************************************************************
     * After the MQTT client subscribes to topics, the MQTT client receives messages through the callback API if the MQTT channel has data.
     * @version V1.0
     ********************************************************************/
    private class MqttMessageCallback implements MqttCallback {

        @Override
        public void connectionLost(Throwable cause) {
            System.out.println(String.format("MQTT Client[%s] connect lost,Retry in 10 seconds,info[%s]", clientId, cause.getMessage()));
            while (!mqttClient.isConnected()) {
                try {
                    Thread.sleep(RECONNECT_INTERVAL);
                    System.out.println(String.format("MQTT Client[%s] reconnect ....", clientId));
                    rconnectMqtt();
                } catch (Exception e) {
                    continue;
                }
            }

        }

        @Override
        public void messageArrived(String topic, MqttMessage mqttMessage) {
            String message = new String(mqttMessage.getPayload());
            System.out.println(String.format("MQTT Client[%s] receive message[%s] from topic[%s]", clientId, message, topic));
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

        }
    }


    public static void main(String[] args) throws MqttException {
        try {
           //Subscribe to the MQTT channel.
            MqttClientDemo mqttsubClientDemo = new MqttClientDemo(MqttClientDemo.MQTT_SUB_CLIENT_ID);
            mqttsubClientDemo.subscribeTopic();
            //Send hello world to the MQTT channel.
            MqttClientDemo mqttpubClientDemo = new MqttClientDemo(MqttClientDemo.MQTT_PUB_CLIENT_ID);
            mqttpubClientDemo.startPublishMessage();
        } catch (MqttException e) {
            System.out.println(String.format("program start error,errorMessage[%s]", e.getMessage()));
        }
    }
}

/*****************************************************************************
 Description: SSL utility class. Load the client SSL certificate configuration and ignore server certificate verification.
 ****************************************************************************/

package com.example.demo;

import java.io.ByteArrayInputStream;
import java.io.InputStreamReader;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.security.KeyPair;
import java.security.KeyStore;
import java.security.Security;
import java.security.cert.Certificate;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;

import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;

import org.bouncycastle.jce.provider.BouncyCastleProvider;
import org.bouncycastle.openssl.PEMReader;
import org.bouncycastle.openssl.PasswordFinder;

public class SslUtil {


    /********************************************************************
     * Verify and obtain the SSLSocketFactory.
     ********************************************************************/
    public static SSLSocketFactory getSocketFactory(final String crtFile, final String keyFile) throws Exception {
        Security.addProvider(new BouncyCastleProvider());

        // 1. Load the client certificate.
        PEMReader reader_client =
                new PEMReader(new InputStreamReader(new ByteArrayInputStream(Files.readAllBytes(Paths.get(crtFile)))));
        X509Certificate cert = (X509Certificate) reader_client.readObject();
        reader_client.close();

        // 2. Load the client key.
        reader_client = new PEMReader(
                new InputStreamReader(new ByteArrayInputStream(Files.readAllBytes(Paths.get(keyFile)))),
                new PasswordFinder() {
                    @Override
                    public char[] getPassword() {
                        return null;
                    }
                }
        );

        // 3. Send the client key and certificate to the server for identity authentication.
        KeyStore ks = KeyStore.getInstance(KeyStore.getDefaultType());
        ks.load(null, null);
        ks.setCertificateEntry("certificate", cert);
        ks.setKeyEntry("private-key", ((KeyPair) reader_client.readObject()).getPrivate(), "".toCharArray(), new Certificate[]{cert});
        KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
        kmf.init(ks, "".toCharArray());

        // 4. Create a socket factory.
        SSLContext context = SSLContext.getInstance("TLSv1.2");
        TrustManager[] tms = new TrustManager[1];
        TrustManager miTM = new TrustAllManager();
        tms[0] = miTM;
        context.init(kmf.getKeyManagers(), tms, null);

        reader_client.close();

        return context.getSocketFactory();
    }


    /********************************************************************
     * Ignore server certificate verification.
     ********************************************************************/
    static class TrustAllManager implements TrustManager, X509TrustManager {
        @Override
        public X509Certificate[] getAcceptedIssuers() {
            return null;
        }

        @Override
        public void checkServerTrusted(X509Certificate[] certs, String authType)
                throws CertificateException {
        }

        public boolean isServerTrusted(X509Certificate[] certs) {
            return true;
        }

        public boolean isClientTrusted(X509Certificate[] certs) {
            return true;
        }

        @Override
        public void checkClientTrusted(X509Certificate[] certs, String authType)
                throws CertificateException {
        }
    }
}

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>mqtt.example</artifactId>
    <version>1.0-SNAPSHOT</version>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>7</source>
                    <target>7</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
    <dependencies>
    <!-- https://mvnrepository.com/artifact/org.eclipse.paho/org.eclipse.paho.client.mqttv3 -->
    <dependency>
        <groupId>org.eclipse.paho</groupId>
        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
        <version>1.2.1</version>
    </dependency>
     <!-- https://mvnrepository.com/artifact/org.bouncycastle/bcprov-jdk16 -->
    <dependency>
        <groupId>org.bouncycastle</groupId>
        <artifactId>bcprov-jdk16</artifactId>
        <version>1.45</version>
    </dependency>
    </dependencies>

</project>