AWS IoT のハンズオンレベルの内容を CloudShell からやってみた

最後のトピックのサブスクライブとメッセージの確認以外はコマンドのみでできます。

手順

# Thing（仮想デバイス）を作成する
$ THING_NAME="MyTestThing"
$ aws iot create-thing \
--thing-name "$THING_NAME"

# 証明書と秘密鍵を作成
$ aws iot create-keys-and-certificate \
--set-as-active \
--output json > certs.json

# certs.json の中身を確認
$ cat certs.json

# 証明書ファイル（device.pem.crt）を作成
$ jq -r '.certificatePem' certs.json > device.pem.crt

# 秘密鍵ファイル（private.pem.key）を作成
$ jq -r '.keyPair.PrivateKey' certs.json > private.pem.key

# 作成ファイルの内容確認
$ head device.pem.crt
$ head private.pem.key

# 証明書ARNを環境変数にセット
$ CERT_ARN=$(jq -r '.certificateArn' certs.json)
$ echo $CERT_ARN

# 証明書を Thing にアタッチ
$ aws iot attach-thing-principal \
--thing-name "$THING_NAME" \
--principal "$CERT_ARN"

# IoT ポリシー定義ファイルを作成
$ cat <<EOF > iot-policy.json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "iot:Connect",
        "iot:Publish",
        "iot:Subscribe",
        "iot:Receive"
      ],
      "Resource": "*"
    }
  ]
}
EOF

# ポリシーを作成
$ POLICY_NAME="MyTestPolicy"
$ aws iot create-policy \
--policy-name "$POLICY_NAME" \
--policy-document file://iot-policy.json

# ポリシーを証明書にアタッチ
$ aws iot attach-policy \
--policy-name "$POLICY_NAME" \
--target "$CERT_ARN"

# Amazon ルート CA 証明書をダウンロード
$ curl -O https://www.amazontrust.com/repository/AmazonRootCA1.pem

# MQTT メッセージ送信用 Python スクリプト作成
$ cat <<EOF > publish.py
import paho.mqtt.client as mqtt
import ssl
import json
import time

endpoint = "<YOUR_ENDPOINT>.iot.ap-northeast-1.amazonaws.com"
port = 8883
topic = "test/topic"

ca_path = "AmazonRootCA1.pem"
cert_path = "device.pem.crt"
key_path = "private.pem.key"

def on_log(client, userdata, level, buf):
    print("log: ", buf)

client = mqtt.Client()
client.on_log = on_log

client.tls_set(ca_certs=ca_path,
               certfile=cert_path,
               keyfile=key_path,
               tls_version=ssl.PROTOCOL_TLSv1_2)

client.connect(endpoint, port, 60)
client.loop_start()

try:
    count = 1
    while True:
        message = {
            "message": f"Hello from CloudShell! ({count})"
        }
        result = client.publish(topic, json.dumps(message), qos=1)
        result.wait_for_publish()
        print(f"[{count}] Published message to topic {topic}: {message}")
        count += 1
        time.sleep(1)
except KeyboardInterrupt:
    print("Stopped by user")

client.loop_stop()
client.disconnect()
EOF

# AWS IoT エンドポイントを取得
$ aws iot describe-endpoint \
--endpoint-type iot:Data-ATS

# publish.py の endpoint を上記で取得した IoT エンドポイントに置換
$ nano publish.py

# Python ライブラリインストール
$ pip3 install --user paho-mqtt

# MQTT メッセージ送信スクリプトを実行
$ python3 publish.py

最後に AWS IoT コンソールの MQTT テストクライアントの画面を開き、test/topic をサブスクライブします。
「Hello from CloudShell!」のメッセージを確認できれば OK です。

まとめ

今回は AWS IoT のハンズオンレベルの内容を CloudShell からやってみました。
どなたかの参考になれば幸いです。