无法使用paho-mqtt java客户端代码连接到aws-iot。

Question

我在AWS核心上创建了东西。然后下载证书、私钥和rootCa证书。我的主要目标是发布和订阅AWS。我尝试使用以下代码使用paho mqtt java客户端代码连接到aws物联网。

public class App{
public static void main(String args[]){

        try{

            String caFilePath = "/home/sanju/Documents/Windows Shared/Leron/runningProjects/dcuFirmwareUpload/certificates/MINI/rootca.crt";
            String clientCrtFilePath = "/home/sanju/Documents/Windows Shared/Leron/runningProjects/dcuFirmwareUpload/certificates/MINI/7d6238abc3-certificate.pem.crt";
            String clientKeyFilePath = "/home/sanju/Documents/Windows Shared/Leron/runningProjects/dcuFirmwareUpload/certificates/MINI/7d6238abc3-private.pem.key";

            String topic = "ThingName/firmwareupdate";
            String content = "Message from MqttPublishSample";
            int qos = 2;
            String broker = "ssl://xxxxxxxxxxxxxxxxx.iot.us-east-2.amazonaws.com:8883";
            String clientId = "ThingName";
            MemoryPersistence persistence = new MemoryPersistence();
            MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
            MqttConnectOptions connOpts = new MqttConnectOptions();

            /*connOpts.setCleanSession(true);*/

            connOpts.setConnectionTimeout(60);
            connOpts.setKeepAliveInterval(60);

            SSLSocketFactory socketFactory = getSocketFactory(caFilePath,
                    clientCrtFilePath, clientKeyFilePath, "");
            connOpts.setSocketFactory(socketFactory);

            System.out.println("Connecting to broker: " + broker);
            sampleClient.connect(connOpts);
            sampleClient.subscribe("subscribeTopic", 1);
            System.out.println("Connected");
            System.out.println("Publish message: " + content);
            MqttMessage message = new MqttMessage(content.getBytes());
            message.setQos(qos);
            sampleClient.setCallback(new SimpleCallback());
            sampleClient.publish(topic, message);
            System.out.println("Message published");
            try {
                Thread.sleep(5000);
                sampleClient.disconnect();
            } catch(Exception e) {
                e.printStackTrace();
            }
            System.out.println("Disconnected");
            System.exit(0);
        }catch(MqttException me){
             System.out.println("reason " + me.getReasonCode());
                System.out.println("msg " + me.getMessage());
                System.out.println("loc " + me.getLocalizedMessage());
                System.out.println("cause " + me.getCause());
                System.out.println("except " + me);
                me.printStackTrace();

        }catch(Exception e){
            System.out.println("running: "+e);
        }
    }

    private 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");*/
        SSLContext context = SSLContext.getInstance("TLSv1.2");
        context.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);

        return context.getSocketFactory();
    }
    }

我得到以下日志和错误

连接到代理: ssl://aclugxc4jtbld-ats.iot.us-east-2.amazonaws.com:8883原因0 msg MqttException loc MqttException原因java.net.SocketTimeoutException:连接超时除了MqttException (0) - java.net.SocketTimeoutException:连接超时MqttException (0) - java.net.SocketTimeoutException:连接超时在org.eclipse.paho.client.mqttv3.internal.ClientComms$ConnectBG.的org.eclipse.paho.client.mqttv3.internal.ExceptionHelper.createMqttException(ExceptionHelper.java:38)在java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)，java.util.concurrent.FutureTask.run(FutureTask.java:266)，java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)，java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)，java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)，java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745)，由:java.net.SocketTimeoutException引起:连接超时在java.net.PlainSocketImpl.socketConnect(原生方法)在java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206) at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188) at java.net.SocksSocketImpl.connect(SocksSocketImpl.java：392)在org.eclipse.paho.client.mqttv3.internal.TCPNetworkModule.start(TCPNetworkModule.java:80) at org.eclipse.paho.client.mqttv3.internal.SSLNetworkModule.start(SSLNetworkModule.java:103) at org.eclipse.paho.client.mqttv3.internal.ClientComms$ConnectBG.run(ClientComms.java:701) .7

希望连接到aws IOT，以便我可以发布/订阅主题。

Answer 1

我没有使用Java客户端，但是当我看到MQTT的“connect”时，我会检查附加到设备证书上的策略。您的代码中的MQTT客户端名称与策略中的客户端名称匹配吗？那东西的名字呢？另外，检查正确的区域和政策中的帐户。

例如，下面的“RaspberryPi”应该与云中的策略和设备的代码匹配：

  "Version": "2012-10-17",
    "Statement": [
      {
        "Effect": "Allow",
        "Action": "iot:Connect",
        "Resource": "arn:aws:iot:REGION:ACCOUNT:client/RaspberryPi"      
      },

如果事情仍然不起作用，尝试一个过度宽松的政策来检查它。找到问题后，将策略限制为最小特权，以使其更安全：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:*",
      "Resource": "*"
    }
  ]
}

AWS IoT文档的另一个很好的例子:检查策略中的“物联网”：https://docs.aws.amazon.com/iot/latest/developerguide/iot-moisture-policy.html

以及代码中使用的'--clientId RaspberryPi‘(MQTT客户端)和’-thingName RaspberryPi‘如何匹配策略使用的名称：https://docs.aws.amazon.com/iot/latest/developerguide/iot-moisture-raspi-setup.html。

"Resource": "arn:aws:iot:REGION:ACCOUNT:client/RaspberryPi"      
"Resource": "arn:aws:iot:REGION:ACCOUNT:thing/RaspberryPi"

请注意，MQTT客户端的名称和事物的名称可能有不同的名称，但必须匹配策略中的名称。