AWS IoT Coreへ証明書ベースでHTTPS postする方法(port 8443/port 443)

AWS IoT CoreへのHTTPS post

AWS IoT Coreへのデータ送信リクエスト方法は MQTTS/HTTPS で用意されていますが、認証方式により使えるポート番号が異なるなど、いくつかの制約があります。自分のqiita記事でも紹介しておりますが、ALPNにより、MQTTをport443で利用することもできます。
このプロトコルのページを英語版でみると HTTPS port443も対応されております。(2019/03/09現在、日本語のプロトコルの紹介には記載がありませんので、linkページに飛んだあとに日本語になってしまう場合は英語に切り替えてください)
ということで本記事では証明書認証のHTTPS postをpythonで実装してみようと思います。

事前準備

詳細は省きますが、AWS IoTの証明書、policy, activateや紐づけ、などAWS IoTの通信に必要な設定は事前にご準備ください。
参考手順

HTTPS requestスタイル

HTTPリクエストは以下の形式になります

https://endpoint-URI:<port>/topics/<topic>?qos=[0|1]

sample コード

以下で実行できます。テストコードなのでhard codingしていますが、alpn_flagのTrue で ALPNを使っての443通信、FalseでAWS IoT Coreデフォルトの8443ポートへの通信となります。
以下のまま実行すると、 date/testのtopicに一度だけ時刻を送りますので、AWS IoTのテスト画面などで確認ください。
urllib2/ SSLモジュール がSSL contextをサポートする必要があるので、うまくいかない場合はPython/SSLのバージョンなどもご確認ください。
（自分はオレゴン, us-west-2で動作確認をしております)

http_post.py

from __future__ import print_function
import urllib, urllib2
import logging, traceback
import sys, os, ssl, json
import datetime

# 本サンプルでは同一のディレクトリに証明書が置かれている前提です
cert_path = "証明書までのパス"
ca = cert_path + "root証明書"
private = cert_path + "AWS IoT Core private 証明書ファイル名"
cert = cert_path + "AWS IoT Core Cert証明書ファイル名"

url = "https://<YOUR AWS IoT Core endpoint>"
topic = "data/test"
mqtt_qos = "0"
IoT_protocol_name = "x-amzn-http-ca"

# alpn false = 8443/ true = 443
alpn_flag = True

logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(sys.stdout)
log_format = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(log_format)
logger.addHandler(handler)

        
def create_ssl_context(alpn_flag):
    try:
        #debug print opnessl version
        logger.info("open ssl version:{}".format(ssl.OPENSSL_VERSION))
        ssl_context = ssl.create_default_context()
        if alpn_flag: ssl_context.set_alpn_protocols([IoT_protocol_name])
        ssl_context.load_verify_locations(cafile=ca)
        ssl_context.load_cert_chain(certfile=cert, keyfile=private)

        return  ssl_context
    except Exception as e:
        logger.error("exception ssl_alpn()")
        raise e

def main():
    try: 
        ssl_context =create_ssl_context(alpn_flag)
        data = {
            "timestamp": datetime.datetime.now().strftime('%Y-%m-%dT%H:%M:%S')
        }
        if alpn_flag:
            port = ":443"
        else:
            port = ":8443"
        post_request_url = url + port + "/topics/" + topic + "?qos=" + mqtt_qos
        logger.info("request url:{}".format(post_request_url))
        # exec post 
        res = urllib2.urlopen(
            url = post_request_url, 
            data = json.dumps(data),
            context=ssl_context
        )
        logger.info("url open result:{}".format(res))
    except Exception as e:
        logger.error("main()")
        logger.error("error message:{}".format(e.message))
        logger.error("traceback:{}".format(traceback.print_exc()))
        sys.exit()

if __name__ == '__main__':
    main()
    

結論

というわけで、HTTPS post with port 443でAWS IoT Coreへデータが送信できました。8443などのport番号はB2Bなどfirewall/proxyがあると使いにくいと思いますが、これで443で通信できます。

免責

本投稿は、個人の意見で、所属する企業や団体は関係ありません。
また掲載しているsampleプログラムの動作に関しても保障いたしませんので、参考程度にしてください。