mqtt paho ssl java端代码

参考链接：http://blog.csdn.net/lingshi210/article/details/52439050

mqtt 的ssl配置可以参阅 
http://houjixin.blog.163.com/blog/static/35628410201432205042955/

然后注意开启防火墙端口。

mqtt的命令和Java端的ssl 必须同时要带上ca.crt、clilent.crt、client.key三个文件，即CA证书、客户证书、客户私钥。

由于java 端不支持client.key的格式，需要命令进行转化

openssl pkcs8 -topk8 -in client.key -out client.pem -nocrypt

另外： 
不知为何ubuntu下关闭防火墙后还是握手失败，cenos下正常，抓包后已经看不到明文了。

Java部分：

1.核心部分只需要设置SSLSocketFactory

MqttConnectOptions options = new MqttConnectOptions();
SSLSocketFactory factory=getSSLSocktet("youpath/ca.crt","youpath/client.crt","youpath/client.pem","password");

options.setSocketFactory(factory);

2.自定义SSLSocketFactory （改进于http://gist.github.com/4104301）

此处的密码应为生成证书的时候输入的密码，未认证。

private SSLSocketFactory getSSLSocktet(String caPath,String crtPath, String keyPath, String password) throws Exception {
        // CA certificate is used to authenticate server
        CertificateFactory cAf = CertificateFactory.getInstance("X.509");
        FileInputStream caIn = new FileInputStream(caPath);
        X509Certificate ca = (X509Certificate) cAf.generateCertificate(caIn);
         KeyStore caKs = KeyStore.getInstance("JKS");
         caKs.load(null, null);
         caKs.setCertificateEntry("ca-certificate", ca);
         TrustManagerFactory tmf = TrustManagerFactory.getInstance("PKIX");
         tmf.init(caKs);

        CertificateFactory cf = CertificateFactory.getInstance("X.509");
        FileInputStream crtIn = new FileInputStream(crtPath);
        X509Certificate caCert = (X509Certificate) cf.generateCertificate(crtIn);

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

        // finally, create SSL socket factory
        SSLContext context = SSLContext.getInstance("TLSv1");

        context.init(kmf.getKeyManagers(),tmf.getTrustManagers(), new SecureRandom());

        return context.getSocketFactory();
    }

Android上会报错，改进如下：

    private SSLSocketFactory getSSLSocktet(String caPath,String crtPath, String keyPath, String password) throws Exception {
        // CA certificate is used to authenticate server
        CertificateFactory cAf = CertificateFactory.getInstance("X.509");
        FileInputStream caIn = new FileInputStream(caPath);
        X509Certificate ca = (X509Certificate) cAf.generateCertificate(caIn);
         KeyStore caKs = KeyStore.getInstance(KeyStore.getDefaultType());
         caKs.load(null, null);
         caKs.setCertificateEntry("ca-certificate", ca);
         TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
         tmf.init(caKs);
         caIn.close();
        CertificateFactory cf = CertificateFactory.getInstance("X.509");
        FileInputStream crtIn = new FileInputStream(crtPath);
        X509Certificate caCert = (X509Certificate) cf.generateCertificate(crtIn);

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

        // finally, create SSL socket factory
        SSLContext context = SSLContext.getInstance("TLSv1");

        context.init(kmf.getKeyManagers(),tmf.getTrustManagers(), new SecureRandom());

        return context.getSocketFactory();
    }

3.获取私钥代码部分

由于只能读取PKCS8的格式，所以需要转成pem

    public PrivateKey getPrivateKey(String path) throws Exception{  



        org.apache.commons.codec.binary.Base64 base64=new Base64();
        byte[] buffer=   base64.decode(getPem(path)); 

        PKCS8EncodedKeySpec keySpec= new PKCS8EncodedKeySpec(buffer);
        KeyFactory keyFactory= KeyFactory.getInstance("RSA");  
        return (RSAPrivateKey) keyFactory.generatePrivate(keySpec);  





    } 

附录：

package com;

import java.awt.BorderLayout;
import java.awt.Container;
import java.awt.event.ActionEvent;
import java.awt.event.ActionListener;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.security.KeyFactory;
import java.security.KeyStore;
import java.security.PrivateKey;
import java.security.SecureRandom;
import java.security.cert.CertificateException;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
import java.security.interfaces.RSAPrivateKey;
import java.security.spec.PKCS8EncodedKeySpec;

import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManager;
import javax.net.ssl.TrustManagerFactory;
import javax.net.ssl.X509TrustManager;
import javax.swing.JButton;
import javax.swing.JFrame;
import javax.swing.JLabel;
import javax.swing.JOptionPane;
import javax.swing.JPanel;
import javax.swing.JTextArea;

import org.apache.commons.codec.binary.Base64;
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.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.internal.security.SSLSocketFactoryFactory;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class Server extends JFrame {
    private static final long serialVersionUID = 1L;
    private JPanel panel;
    private JPanel panelText;
    private JPanel panelText2;
    private JButton button;
    private JButton button2;
    private JButton subscribeButton;
    private JTextArea textHost;
    private JTextArea textClientID;
    private JTextArea textPublishMsg;
    private JTextArea textTopic;

    private MqttClient client;
    private String host = "ssl://192.168.10.233:1883";

    private MqttTopic topic;
    private MqttMessage message;

    private String userToken = "999999";

    private String myTopicRoot = "test";
    private String myTopic = null;

    private String clienID = "test1234567";

    public Server() {

        Container container = this.getContentPane();
        panel = new JPanel();
        panelText = new JPanel();
        panelText2 = new JPanel();
        button = new JButton("发布主题消息");
        button2 = new JButton("更换客户机地址和IP");
        button.addActionListener(new ActionListener() {

            @Override
            public void actionPerformed(ActionEvent ae) {
                try {
                    host = textHost.getText();
                    clienID = textClientID.getText();
                    if (client == null) {
                        client = new MqttClient(host, clienID, new MemoryPersistence());
                    }

                    if (!client.isConnected()) {
                        connect();
                    }

                    publishMsg(textTopic.getText(), textPublishMsg.getText());
                } catch (Exception e) {
                    e.printStackTrace();
                    showErrorMsg(e.toString());
                }
            }
        });

        button2.addActionListener(new ActionListener() {

            @Override
            public void actionPerformed(ActionEvent arg0) {
                // TODO Auto-generated method stub
                host = textHost.getText();
                clienID = textClientID.getText();
                try {
                    if (client != null)
                        client.disconnectForcibly();
                    client = new MqttClient(host, clienID, new MemoryPersistence());
                    connect();
                } catch (Exception e) {
                    e.printStackTrace();
                    showErrorMsg(e.toString());
                }
            }
        });

        subscribeButton = new JButton("订阅主题");
        subscribeButton.addActionListener(new ActionListener() {

            @Override
            public void actionPerformed(ActionEvent arg0) {
                // TODO Auto-generated method stub
                try {
                    if (client == null) {
                        client = new MqttClient(host, clienID, new MemoryPersistence());
                    }

                    if (!client.isConnected()) {
                        connect();
                    }
                    if (myTopic != null && !myTopic.equals(textTopic.getText())) {
                        client.subscribe(myTopic);
                    }
                    client.subscribe(textTopic.getText());
                    myTopic = textTopic.getText();

                } catch (Exception e) {
                    e.printStackTrace();
                    showErrorMsg(e.toString());
                }
            }
        });

        textHost = new JTextArea();
        textHost.setText(host);
        textClientID = new JTextArea();
        textClientID.setText(clienID);

        panel.add(button);
        panel.add(subscribeButton);
        panelText.add(button2);
        panelText.add(new JLabel("mqtt地址"));
        panelText.add(textHost);
        panelText.add(new JLabel("ClienId"));
        panelText.add(textClientID);
        panelText.add(new JLabel("主题"));
        textTopic = new JTextArea();

        textTopic.setText(myTopicRoot);
        panelText.add(textTopic);

        textPublishMsg = new JTextArea();
        textPublishMsg.setText("@" + userToken + "@E@5@" + userToken + "@");
        panelText2.add(new JLabel("mqtt消息"));
        panelText2.add(textPublishMsg);

        container.add(panel, BorderLayout.NORTH);
        container.add(panelText, BorderLayout.CENTER);
        container.add(panelText2, BorderLayout.SOUTH);
        // try {
        // client = new MqttClient(host, clienID,
        // new MemoryPersistence());
        // connect();
        // } catch (Exception e) {
        // showErrorMsg(e.toString());
        // }
    }

    private SSLSocketFactory getSSLSocktet(String caPath,String crtPath, String keyPath, String password) throws Exception {
        // CA certificate is used to authenticate server
        CertificateFactory cAf = CertificateFactory.getInstance("X.509");
        FileInputStream caIn = new FileInputStream(caPath);
        X509Certificate ca = (X509Certificate) cAf.generateCertificate(caIn);
         KeyStore caKs = KeyStore.getInstance("JKS");
         caKs.load(null, null);
         caKs.setCertificateEntry("ca-certificate", ca);
         TrustManagerFactory tmf = TrustManagerFactory.getInstance("PKIX");
         tmf.init(caKs);

        CertificateFactory cf = CertificateFactory.getInstance("X.509");
        FileInputStream crtIn = new FileInputStream(crtPath);
        X509Certificate caCert = (X509Certificate) cf.generateCertificate(crtIn);

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

        // finally, create SSL socket factory
        SSLContext context = SSLContext.getInstance("TLSv1");

        context.init(kmf.getKeyManagers(),tmf.getTrustManagers(), new SecureRandom());

        return context.getSocketFactory();
    }

    private String getPem(String path) throws Exception{
        FileInputStream fin=new FileInputStream(path);
        BufferedReader br= new BufferedReader(new InputStreamReader(fin));  
        String readLine= null;  
        StringBuilder sb= new StringBuilder();  
        while((readLine= br.readLine())!=null){  
            if(readLine.charAt(0)=='-'){  
                continue;  
            }else{  
                sb.append(readLine);  
                sb.append('
');  
            }  
        }  
        fin.close();
        return sb.toString();
    }

    public PrivateKey getPrivateKey(String path) throws Exception{  

        org.apache.commons.codec.binary.Base64 base64=new Base64();
        byte[] buffer=   base64.decode(getPem(path)); 

        PKCS8EncodedKeySpec keySpec= new PKCS8EncodedKeySpec(buffer);
        KeyFactory keyFactory= KeyFactory.getInstance("RSA");  
        return (RSAPrivateKey) keyFactory.generatePrivate(keySpec);  


    }  



    private void connect() {

        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(false);
        // options.setUserName(userName);
        // options.setPassword(passWord.toCharArray());
        // 设置超时时间
        // options.setConnectionTimeout(10);
        // 设置会话心跳时间
        // options.setKeepAliveInterval(20);
        // try {
        // options.setWill("willtest", "SENDgpslost".getBytes(), 1, false);
        // } catch (Exception e1) {
        // // TODO Auto-generated catch block
        // System.out.print(e1);
        // }
        try {
            if (!SSLSocketFactoryFactory.isSupportedOnJVM()) {
                System.out.print("isSupportedOnJVM=false");
            }

            SSLSocketFactory factory=getSSLSocktet("F:/ssl/ca.crt","F:/ssl/client.crt","F:/ssl/client.pem","brt123");

            options.setSocketFactory(factory);
            client.setCallback(new MqttCallback() {

                @Override
                public void connectionLost(Throwable cause) {
                    System.out.println("connectionLost-----------");
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.println("deliveryComplete---------" + token.isComplete());
                }

                @Override
                public void messageArrived(String topic, MqttMessage arg1) throws Exception {
                    System.out.println("messageArrived----------");
                    String msg = new String(arg1.getPayload());
                    showErrorMsg("主题：" + topic + "
消息：" + msg);
                }
            });

            topic = client.getTopic(myTopicRoot + userToken);
            client.connect(options);

        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public void publishMsg(String topoc, String msg) {
        message = new MqttMessage();
        message.setQos(0);
        message.setRetained(false);
        System.out.println(message.isRetained() + "------ratained状态");

        try {
            message.setPayload(msg.getBytes("UTF-8"));
            client.publish(topoc, message);
        } catch (Exception e) {
            e.printStackTrace();
            showErrorMsg(e.toString());
        }
    }

    private void showErrorMsg(String msg) {
        JOptionPane.showMessageDialog(null, msg);
    }

}