AWS IoT Device SDK PythonをベースにしてSORACOM Beam経由でAWS IoT CoreへMQTT接続できるか認確

SORACOM BeamのMQTT QoSについて記載を追加しました (2020/7/7)

目的

開発量を減らしたいなどの理由やサービスの実装部分を減らしたいと考えれば、メタにPahoで実装したくないとか、サービスアップデート部分もSDKアップデートで対応したいなどの理由でSDKを使いたいケースは多いと思います。一方でデバイスやGatewayのケーパビリティやバッテリーなどの物理的な制約でデバイス側はセキュリティ実装は軽くしたいという相反する希望もあるかと思います。
じゃってことでAWS IoT Device SDKを使ってMQTT -> SORACOM Beam経由で MQTTS -> AWS IoT Coreな通信ができるのかをPython SDKをもとに確認してみます。

AWS IoT Core と SORACOM Beamの仕様差分を確認

MQTT client視点で見る差分は以下となります。

	AWS IoT Core	SORACOM Beam
endpoint	{prefix}-ats.iot.{region}.amazonaws.com	beam.soracom.io
port	8883	1883
Protocl	MQTTS	MQTT

ということは、デバイス側のBeam MQTTクライアントとしては、
beam.soracom.io:1883へMQTTが投げられればよいということになります。

SORACOM beamでQoS 1を利用したい場合

SORACOM BeamのMQTT New Versionを利用ください。詳細はこちら

AWS IoT Device SDKの変更点を見てみる

AWS IoT Device SDK pythonにおけるMQTTインスタンスの生成は以下の記述があります。

# Import SDK packages
from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient

# For certificate based connection
myMQTTClient = AWSIoTMQTTClient("myClientID")
# For Websocket connection
# myMQTTClient = AWSIoTMQTTClient("myClientID", useWebsocket=True)
# Configurations
# For TLS mutual authentication
myMQTTClient.configureEndpoint("YOUR.ENDPOINT", 8883)
myMQTTClient.configureCredentials("YOUR/ROOT/CA/PATH", "PRIVATE/KEY/PATH", "CERTIFICATE/PATH")
# For Websocket, we only need to configure the root CA
# myMQTTClient.configureCredentials("YOUR/ROOT/CA/PATH")
myMQTTClient.configureOfflinePublishQueueing(-1)  # Infinite offline Publish queueing
myMQTTClient.configureDrainingFrequency(2)  # Draining: 2 Hz
myMQTTClient.configureConnectDisconnectTimeout(10)  # 10 sec
myMQTTClient.configureMQTTOperationTimeout(5)  # 5 sec

インスタンス設定用の関数にエンドポイント、ポートが自分で設定できるようになっているので、どうやら変更できそうです。
では上記の差分に従って変更してみると、以下になります。

from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient

# For certificate based connection
myMQTTClient = AWSIoTMQTTClient("myClientID")
# For Websocket connection
# myMQTTClient = AWSIoTMQTTClient("myClientID", useWebsocket=True)
# Configurations

# ソラコムの設定にエンドポイント、ポート設定を変更
myMQTTClient.configureEndpoint("beam.soracom.io", 1883)
# 証明書によるTLSは不要なので、証明書の設定をコメントアウト
#myMQTTClient.configureCredentials("YOUR/ROOT/CA/PATH", "PRIVATE/KEY/PATH", "CERTIFICATE/PATH")

# For Websocket, we only need to configure the root CA
# myMQTTClient.configureCredentials("YOUR/ROOT/CA/PATH")
myMQTTClient.configureOfflinePublishQueueing(-1)  # Infinite offline Publish queueing
myMQTTClient.configureDrainingFrequency(2)  # Draining: 2 Hz
myMQTTClient.configureConnectDisconnectTimeout(10)  # 10 sec
myMQTTClient.configureMQTTOperationTimeout(5)  # 5 sec

Shadow packageからshadowインスタンスを作るときにも同じ変更点です。

SORACOM BeamのAWS IoT MQTT接続の設定

詳細はSOARCOMのサイトをご参照ください。後半のmosquittoやAmazon SNSは設定を省略しても構いません。

sampleで提供されているbasicPubSubを使ってみる

basicPubSub
不要かつ動いてしまう証明書依存部分をコメントアウトして以下の様になりました。

basicPubSub_beam.py

'''
/*
 * Copyright 2010-2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License").
 * You may not use this file except in compliance with the License.
 * A copy of the License is located at
 *
 *  http://aws.amazon.com/apache2.0
 *
 * or in the "license" file accompanying this file. This file is distributed
 * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
 * express or implied. See the License for the specific language governing
 * permissions and limitations under the License.
 */
 '''

from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
import logging
import time
import argparse
import json

AllowedActions = ['both', 'publish', 'subscribe']

# Custom MQTT message callback
def customCallback(client, userdata, message):
    print("Received a new message: ")
    print(message.payload)
    print("from topic: ")
    print(message.topic)
    print("--------------\n\n")


# Read in command-line parameters
parser = argparse.ArgumentParser()
parser.add_argument("-e", "--endpoint", action="store", required=True, dest="host", help="Your AWS IoT custom endpoint")
#parser.add_argument("-r", "--rootCA", action="store", required=True, dest="rootCAPath", help="Root CA file path")
#parser.add_argument("-c", "--cert", action="store", dest="certificatePath", help="Certificate file path")
#parser.add_argument("-k", "--key", action="store", dest="privateKeyPath", help="Private key file path")
parser.add_argument("-p", "--port", action="store", dest="port", type=int, help="Port number override")
parser.add_argument("-w", "--websocket", action="store_true", dest="useWebsocket", default=False,
                    help="Use MQTT over WebSocket")
parser.add_argument("-id", "--clientId", action="store", dest="clientId", default="basicPubSub",
                    help="Targeted client id")
parser.add_argument("-t", "--topic", action="store", dest="topic", default="sdk/test/Python", help="Targeted topic")
parser.add_argument("-m", "--mode", action="store", dest="mode", default="both",
                    help="Operation modes: %s"%str(AllowedActions))
parser.add_argument("-M", "--message", action="store", dest="message", default="Hello World!",
                    help="Message to publish")

args = parser.parse_args()
host = args.host
#rootCAPath = args.rootCAPath
#certificatePath = args.certificatePath
#privateKeyPath = args.privateKeyPath
port = args.port
useWebsocket = args.useWebsocket
clientId = args.clientId
topic = args.topic

if args.mode not in AllowedActions:
    parser.error("Unknown --mode option %s. Must be one of %s" % (args.mode, str(AllowedActions)))
    exit(2)
'''
if args.useWebsocket and args.certificatePath and args.privateKeyPath:
    parser.error("X.509 cert authentication and WebSocket are mutual exclusive. Please pick one.")
    exit(2)

if not args.useWebsocket and (not args.certificatePath or not args.privateKeyPath):
    parser.error("Missing credentials for authentication.")
    exit(2)
'''
# Port defaults
if args.useWebsocket and not args.port:  # When no port override for WebSocket, default to 443
    port = 443
if not args.useWebsocket and not args.port:  # When no port override for non-WebSocket, default to 8883
    port = 8883

# Configure logging
logger = logging.getLogger("AWSIoTPythonSDK.core")
#logger.setLevel(logging.DEBUG)
streamHandler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
streamHandler.setFormatter(formatter)
logger.addHandler(streamHandler)

# Init AWSIoTMQTTClient
myAWSIoTMQTTClient = None
if useWebsocket:
    myAWSIoTMQTTClient = AWSIoTMQTTClient(clientId, useWebsocket=True)
    myAWSIoTMQTTClient.configureEndpoint(host, port)
    #myAWSIoTMQTTClient.configureCredentials(rootCAPath)
else:
    myAWSIoTMQTTClient = AWSIoTMQTTClient(clientId)
    myAWSIoTMQTTClient.configureEndpoint(host, port)
    #myAWSIoTMQTTClient.configureCredentials(rootCAPath, privateKeyPath, certificatePath)

# AWSIoTMQTTClient connection configuration
myAWSIoTMQTTClient.configureAutoReconnectBackoffTime(1, 32, 20)
myAWSIoTMQTTClient.configureOfflinePublishQueueing(-1)  # Infinite offline Publish queueing
myAWSIoTMQTTClient.configureDrainingFrequency(2)  # Draining: 2 Hz
myAWSIoTMQTTClient.configureConnectDisconnectTimeout(10)  # 10 sec
myAWSIoTMQTTClient.configureMQTTOperationTimeout(5)  # 5 sec

# Connect and subscribe to AWS IoT
myAWSIoTMQTTClient.connect()
if args.mode == 'both' or args.mode == 'subscribe':
    myAWSIoTMQTTClient.subscribe(topic, 1, customCallback)
time.sleep(2)

# Publish to the same topic in a loop forever
loopCount = 0
while True:
    if args.mode == 'both' or args.mode == 'publish':
        message = {}
        message['message'] = args.message
        message['sequence'] = loopCount
        messageJson = json.dumps(message)
        myAWSIoTMQTTClient.publish(topic, messageJson, 1)
        if args.mode == 'publish':
            print('Published topic %s: %s\n' % (topic, messageJson))
        loopCount += 1
    time.sleep(1)

標準のbasicPubSubとdiffをとるとこのくらいの差分

$ diff basicPubSub_soracom.py basicPubSub.py 
38,40c38,40
< #parser.add_argument("-r", "--rootCA", action="store", required=True, dest="rootCAPath", help="Root CA file path")
< #parser.add_argument("-c", "--cert", action="store", dest="certificatePath", help="Certificate file path")
< #parser.add_argument("-k", "--key", action="store", dest="privateKeyPath", help="Private key file path")
---
> parser.add_argument("-r", "--rootCA", action="store", required=True, dest="rootCAPath", help="Root CA file path")
> parser.add_argument("-c", "--cert", action="store", dest="certificatePath", help="Certificate file path")
> parser.add_argument("-k", "--key", action="store", dest="privateKeyPath", help="Private key file path")
54,56c54,56
< #rootCAPath = args.rootCAPath
< #certificatePath = args.certificatePath
< #privateKeyPath = args.privateKeyPath
---
> rootCAPath = args.rootCAPath
> certificatePath = args.certificatePath
> privateKeyPath = args.privateKeyPath
65c65
< '''
---
> 
73c73
< '''
---
> 
82c82
< #logger.setLevel(logging.DEBUG)
---
> logger.setLevel(logging.DEBUG)
93c93
<     #myAWSIoTMQTTClient.configureCredentials(rootCAPath)
---
>     myAWSIoTMQTTClient.configureCredentials(rootCAPath)
97c97
<     #myAWSIoTMQTTClient.configureCredentials(rootCAPath, privateKeyPath, certificatePath)
---
>     myAWSIoTMQTTClient.configureCredentials(rootCAPath, privateKeyPath, certificatePath)

動作確認するだけなので、引数を以下として起動します。

python basicPubSub_beam.py -e beam.soracom.io -p 1883

AWS IoT consoleでみるとPubの成功、また本スクリプトがdefaultで自分自身のpubしたtopicをsubscribeしているので、起動したコンソール上にもメッセージが表示されます。
これでまず、Beamを通してpubsubが問題なく行われていることがわかります。

動作確認できたもの

ざっと私の方で動作確認できたものとして、

• shadow instaceを生成しての AWS IoT Shadowの操作
• deltaの受信、shadow updateは問題なくできました
• basic ingest: これも問題なし

ということで＄awsで始まるような予約topicも問題なくやり取りできました。

SDKを使う上で大事なこと

ここまで説明してきたとおり、Endpoint、TLS設定なし、ポートの変更ができればSDKをつかってBeamを使うことは可能でした。
Javascript版もBeam接続できそうです。JavaはTLSが必須になっている模様。など、SDKの実装に大きく依存しておりますので、皆様にてご確認ください。

SORACOM Beamを使う上でのセキュリティレベルは変わるのか？

AWS IoTにおいては、各デバイス/GWなどAWS IoTと通信するものに個別の証明書を入れることを推奨としていると思います。SORACOMを利用する場合同じく通信するデバイス/GWにSORACOM SIMがついており、このSIMが個別証明書相当になります。この違いは、AWS側の考えている通信の前提とSORACOMを利用する場合で前提が異なるということになります。
SORACOMの通信の場合、世の中の携帯のSIM認証のセキュリティレベルがあります。Beamを使う場合はデバイス個別鍵をSIMとしてAWS IoTへの認証をSORACOM側が行うことになり、AWSIoTと皆様のSIM groupが正しく/セキュアに通信するためにBeamの設定としてAWS IoTの証明書を設定することになります。
ということで、通信の前提の違いによる考え方はあるもののSOARCOM Beamを使うからセキュリティレベルが落ちるということにはならないと思います。

SORACOM Beamで気にすること

AWS IoTのいう、デバイスごとの証明書を発行する場合、 Policyを証明書を対(1:1)に発行することでPolicyの記載レベルを証明書単位で変えることができます。一方でBeamの場合、グループ単位で一つの証明書になるので、MQTTの設計および、Thing Attributeなどをうまく設計し、Policy変数をうまく使うなどを検討する必要があるかと思います。

SORACOM Beamが解決すること

導入でもいくつか例をあげましたが、

• セキュリティのオフロード
• 証明書ストアの集約
• B2B IoTでお客様ネットワークに相乗りすると問題になるProxy/port問題
  などが考えられます。

まとめ

いずれにしてもMQTTデザインとpolicyデザイン、証明書発行フロー（SORACOM Beam を使う場合はグループのデザイン）は初期に設計した上で、想定されている最大に拡大したときにも破綻しないデザインを検討することは重要です。
クラウドアーキテクチャや、デバイス側プログラミングが楽しいので、忘れがちにあるのですが、このあたりを初期に担保すると後々困ることは減るはずです。

免責

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