mosquitto的TLS功能测试，客户端使用paho.mqtt.golang（附JAVA版客户端实现）

1、SSL/TLS简介

　　SSL（SecureSocket Layer）安全套接层，是网景公司提出的用于保证Server与client之间安全通信的一种协议，该协议位于TCP/IP协议与各应用层协议之间，即SSL独立于各应用层协议，因此各应用层协议可以透明地调用SSL来保证自身传输的安全性。目前，SSL被大量应用于http的安全通信中，MQTT协议与http协议同样属于应用层协议，因此也可以像http协议一样使用ssl为自己的通信提供安全保证。

　　SSL与TLS（Transport LayerSecurity Protocol）之间的关系：TLS（TransportLayer Security，传输层安全协议）是IETF（InternetEngineering Task Force，Internet工程任务组）制定的一种新的协议，它建立在SSL 3.0协议规范之上，是SSL 3.0的后续版本。在TLS与SSL3.0之间存在着显著的差别，主要是它们所支持的加密算法不同，所以TLS与SSL3.0不能互操作。

2、使用Openssl创建tls证书

　　SSL在身份认证过程中需要有一个双方都信任的CA签发的证书，CA签发证书是需要收费的，但是在测试过程中，可以自己产生一个CA，然后用自己产生的CA签发证书，下面的mosquitto的ssl功能的测试过程就是采用这一方式，其过程如下：

步骤一：产生自己的CA

openssl req -new -x509 -days 36500 -extensions v3_ca -keyout ca.key -out ca.crt
openssl req -new -x509 -days 36500 -extensions v3_ca -keyout ca.key -out ca.pem

步骤二：产生服务端证书

openssl genrsa -des3 -out server.key 2048
openssl req -out server.csr -key server.key -new
openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out server.crt -days 36500

步骤三：产生客户端证书

openssl genrsa -out client-key.pem 2048
openssl req -out client.csr -key client-key.pem -new
openssl x509 -req -in client.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out client-crt.pem -days 36500

经过上面8条命令后，即可生成所需的所有证书文件，其中：

客户端使用：ca.pem、client-crt.pem、client-key.pem

服务端使用：ca.crt、server.crt、server.key

3、mosquitto.conf配置如下：

4、golang客户端测试代码

package cmd

import (
    "crypto/tls"
    "crypto/x509"
    fmt "fmt"
    "io/ioutil"
    "os"
    "time"

    "github.com/apex/log"
    MQTT "github.com/eclipse/paho.mqtt.golang"
)

var ctx log.Interface

const QoS = 0x02

func init() {
    fmt.Printf("init mqtt test
")

}

func RunMqttClient() {
    fmt.Printf("Run mqtt test
")
    var logLevel = log.InfoLevel
    ctx = &log.Logger{
        Level:   logLevel,
        Handler: NewLogHanler(os.Stdout),
    }

    mqttClient := NewClient(
        ctx,
        "ttnhdl",
        "",
        "",
        fmt.Sprintf("ssl://%s", "192.168.195.201:8883"),
    )

    var err = mqttClient.Connect()
    if err != nil {
        ctx.WithError(err).Fatal("Could not connect to MQTT")
        fmt.Printf("Could not connect to MQTT
")
    } else {
        fmt.Printf("Success connect to MQTT
")
    }

    mqttClient.PublishUplink("test", "hello mqtt!")
    mqttClient.SubscribeUplink("test")

    for true {

    }
}

// Client connects to the MQTT server and can publish/subscribe on uplink, downlink and activations from devices
type Client interface {
    Connect() error
    Disconnect()

    IsConnected() bool

    // Uplink pub/sub
    PublishUplink(topic string, msg string) Token
    SubscribeUplink(topic string) Token
}

type Token interface {
    Wait() bool
    WaitTimeout(time.Duration) bool
    Error() error
}

type simpleToken struct {
    err error
}

// Wait always returns true
func (t *simpleToken) Wait() bool {
    return true
}

// WaitTimeout always returns true
func (t *simpleToken) WaitTimeout(_ time.Duration) bool {
    return true
}

// Error contains the error if present
func (t *simpleToken) Error() error {
    return t.err
}

type defaultClient struct {
    mqtt MQTT.Client
    ctx  log.Interface
}

func NewClient(ctx log.Interface, id, username, password string, brokers ...string) Client {
    tlsconfig := NewTLSConfig()

    mqttOpts := MQTT.NewClientOptions()

    for _, broker := range brokers {
        mqttOpts.AddBroker(broker)
    }

    mqttOpts.SetClientID("ypf_dewqfvcdeqfcdqwcdq")
    mqttOpts.SetUsername(username)
    mqttOpts.SetPassword(password)

    // TODO: Some tuning of these values probably won't hurt:
    mqttOpts.SetKeepAlive(30 * time.Second)
    mqttOpts.SetPingTimeout(10 * time.Second)

    // Usually this setting should not be used together with random ClientIDs, but
    // we configured The Things Network's MQTT servers to handle this correctly.
    mqttOpts.SetCleanSession(false)

    mqttOpts.SetDefaultPublishHandler(func(client MQTT.Client, msg MQTT.Message) {
        ctx.WithField("message", msg).Warn("Received unhandled message")
    })

    mqttOpts.SetConnectionLostHandler(func(client MQTT.Client, err error) {
        ctx.WithError(err).Warn("Disconnected, reconnecting...")
    })

    mqttOpts.SetOnConnectHandler(func(client MQTT.Client) {
        ctx.Debug("Connected")
    })

    mqttOpts.SetTLSConfig(tlsconfig)

    return &defaultClient{
        mqtt: MQTT.NewClient(mqttOpts),
        ctx:  ctx,
    }
}

var (
    // ConnectRetries says how many times the client should retry a failed connection
    ConnectRetries = 10
    // ConnectRetryDelay says how long the client should wait between retries
    ConnectRetryDelay = time.Second
)

func (c *defaultClient) Connect() error {
    if c.mqtt.IsConnected() {
        return nil
    }
    var err error
    for retries := 0; retries < ConnectRetries; retries++ {
        token := c.mqtt.Connect()
        token.Wait()
        err = token.Error()
        if err == nil {
            break
        }
        <-time.After(ConnectRetryDelay)
    }
    if err != nil {
        return fmt.Errorf("Could not connect: %s", err)
    }
    return nil
}

func (c *defaultClient) Disconnect() {
    if !c.mqtt.IsConnected() {
        return
    }
    c.mqtt.Disconnect(25)
}

func (c *defaultClient) IsConnected() bool {
    return c.mqtt.IsConnected()
}

func (c *defaultClient) PublishUplink(topic string, msg string) Token {
    return c.mqtt.Publish(topic, QoS, false, msg)
}

func (c *defaultClient) SubscribeUplink(topic string) Token {
    return c.mqtt.Subscribe(topic, QoS, func(mqtt MQTT.Client, msg MQTT.Message) {
        // Determine the actual topic
        fmt.Printf("Success SubscribeUplink with msg:%s
", msg.Payload())
    })
}

func NewTLSConfig() *tls.Config {
    // Import trusted certificates from CAfile.pem.
    // Alternatively, manually add CA certificates to
    // default openssl CA bundle.
    certpool := x509.NewCertPool()
    pemCerts, err := ioutil.ReadFile("samplecerts/ca.pem")
    if err == nil {
        certpool.AppendCertsFromPEM(pemCerts)
    }
    fmt.Println("0. resd pemCerts Success")

    // Import client certificate/key pair
    cert, err := tls.LoadX509KeyPair("samplecerts/client-crt.pem", "samplecerts/client-key.pem")
    if err != nil {
        panic(err)
    }
    fmt.Println("1. resd cert Success")

    // Just to print out the client certificate..
    cert.Leaf, err = x509.ParseCertificate(cert.Certificate[0])
    if err != nil {
        panic(err)
    }
    fmt.Println("2. resd cert.Leaf Success")

    // Create tls.Config with desired tls properties
    return &tls.Config{
        // RootCAs = certs used to verify server cert.
        RootCAs: certpool,
        // ClientAuth = whether to request cert from server.
        // Since the server is set up for SSL, this happens
        // anyways.
        ClientAuth: tls.NoClientCert,
        // ClientCAs = certs used to validate client cert.
        ClientCAs: nil,
        // InsecureSkipVerify = verify that cert contents
        // match server. IP matches what is in cert etc.
        InsecureSkipVerify: true,
        // Certificates = list of certs client sends to server.
        Certificates: []tls.Certificate{cert},
    }
}

 5、测试效果

服务端启动：

客户端运行：

6、JAVA版客户端实现

依赖：org.eclipse.paho.client.mqttv3、bcprov-jdk16-1.45.jar

MqttServiceClient代码：

package com.ypf.main;

import java.util.Properties;

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.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttSecurityException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.internal.MemoryPersistence;

import com.ypf.mqtt.SslUtil;

/** 
 * 
 * @author LP by 2014-04-24
 *
 */
public class MqttServiceClient implements MqttCallback {

    private static final String MQTT_HOST = "ssl://192.168.195.201:8884";
    private static final String MQTT_CLIENT = "Test_";
    public static String caFilePath = "D:/for-iot/LURA/src/mytest/samplecerts/ca.crt";
    public static String clientCrtFilePath = "D:/for-iot/LURA/src/mytest/samplecerts/client.crt";
    public static String clientKeyFilePath = "D:/for-iot/LURA/src/mytest/samplecerts/client.key";
    
    public static MqttServiceClient mqttServiceClient = null;
    
    private MqttClient client = null;
    private MqttConnectOptions options = null;
    
    /**
     * 单例模式构造类
     */
    public static MqttServiceClient getInstance() {
        if (mqttServiceClient == null) {
            mqttServiceClient = new MqttServiceClient();
        }
        return mqttServiceClient;
    }

    private MqttServiceClient() {
        System.out.println("init MQTTClientService");
        init();
    }
    // The major API implementation follows :-

    /**
     * 初始化
     */
    private void init() {
        try {
        
            // host为主机名，test为clientid即连接MQTT的客户端ID，一般以客户端唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
            client = new MqttClient(MQTT_HOST, MQTT_CLIENT, new MemoryPersistence());
            // MQTT的连接设置
            options = new MqttConnectOptions();
            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
            options.setCleanSession(true);
            // 设置连接的用户名
            options.setUserName("ypf");
            // 设置连接的密码
            options.setPassword("ruijie".toCharArray());
            // 设置超时时间 单位为秒
            options.setConnectionTimeout(50);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            options.setKeepAliveInterval(30);
            // TLS连接配置
            options.setSocketFactory(
                    SslUtil.getSocketFactory(caFilePath, clientCrtFilePath, clientKeyFilePath, "cs123456"));

            // 设置回调
            client.setCallback(this);
            
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    /**
     * 连接到MQTT
     */
    void connect() {
        System.out.println("Start connect----------");
        try {
            client.connect(options);
            //订阅主题的方法，2为消息的质量
            client.subscribe("+/#", 2);
            //发送消息
            publish("test", "撒打发水电费水电费");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    /**
     * 断开连接到MQTT
     */
    public void disconnect() {
        System.out.println("Start disconnect----------");
        try {
            client.disconnect();
        } catch (MqttSecurityException e) {
            e.printStackTrace();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /** 
     * 发布消息
     * @param topic 主题
     * @param msg 消息
     */
    public void publish(String topic, String msg) {
        System.out.println("Start publish----------");
        try {
            MqttTopic mqttTopic = client.getTopic(topic);
            //2为消息的质量
            MqttDeliveryToken messageToken = mqttTopic.publish(msg.getBytes(), 2, true);
            System.out.println("publish success==>"+messageToken.getMessage());
//            client.publish(topic, 2, msg);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    
// -------------------------------------------------回调方法------------------------------------------------------------//
    
    /** 
     * 连接断开触发此方法
     */
    @Override
    public void connectionLost(Throwable cause) {
        System.out.println("Connection Lost---------->" + cause.getMessage());
    }

    /** 
     * 消息达到触发此方法
     */
    @Override
    public void messageArrived(MqttTopic topic, MqttMessage message)
            throws Exception {
        System.out.println(topic + ":" + message.toString());
    }

    /**
     * 消息发送成功触发此方法
     */
    @Override
    public void deliveryComplete(MqttDeliveryToken token)  {
        try {
            System.out.println("deliveryComplete---------" + token.getMessage());
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    
    public static void main(String[] args)throws Exception {
        
        //MqttServiceClient.getInstance().disconnect();
        MqttServiceClient.getInstance().connect();
        
        new Thread() {
            public void run() {
                int count = 0;
                while(true && count < 3) {
                    try {
                        Thread.sleep(1000*3);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    MqttServiceClient.getInstance().publish("test1/ypf", "hello world ! count=" + count);
                    count ++;
                }
            };
        }.start();
    }
    
}

SslUtil代码：

package com.ypf.mqtt;

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.X509Certificate;

import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManagerFactory;

import org.bouncycastle.jce.provider.BouncyCastleProvider;
import org.bouncycastle.openssl.*;

public class SslUtil {
    public static SSLSocketFactory getSocketFactory(final String caCrtFile, final String crtFile, final String keyFile,
            final String password) throws Exception {
        Security.addProvider(new BouncyCastleProvider());

        // load CA certificate
        PEMReader reader = new PEMReader(
                new InputStreamReader(new ByteArrayInputStream(Files.readAllBytes(Paths.get(caCrtFile)))));
        X509Certificate caCert = (X509Certificate) reader.readObject();
        reader.close();

        // load client certificate
        reader = new PEMReader(new InputStreamReader(new ByteArrayInputStream(Files.readAllBytes(Paths.get(crtFile)))));
        X509Certificate cert = (X509Certificate) reader.readObject();
        reader.close();

        // load client private key
        reader = new PEMReader(new InputStreamReader(new ByteArrayInputStream(Files.readAllBytes(Paths.get(keyFile)))),
                new PasswordFinder() {
                    @Override
                    public char[] getPassword() {
                        return password.toCharArray();
                    }
                });
        KeyPair key = (KeyPair) reader.readObject();
        reader.close();

        // CA certificate is used to authenticate server
        KeyStore caKs = KeyStore.getInstance(KeyStore.getDefaultType());
        caKs.load(null, null);
        caKs.setCertificateEntry("ca-certificate", caCert);
        TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
        tmf.init(caKs);

        // client key and certificates are sent to server so it can authenticate
        // us
        KeyStore ks = KeyStore.getInstance(KeyStore.getDefaultType());
        ks.load(null, null);
        ks.setCertificateEntry("certificate", cert);
        ks.setKeyEntry("private-key", key.getPrivate(), password.toCharArray(),
                new java.security.cert.Certificate[] { cert });
        KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
        kmf.init(ks, password.toCharArray());

        // finally, create SSL socket factory
        SSLContext context = SSLContext.getInstance("TLSv1");
        context.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);

        return context.getSocketFactory();
    }
}