AWSIoT ラージデータアップロードパターン実装

このBlackbeltセミナーの「ラージデータアップデートパターン」を実装しようと思ったら、未熟ゆえに結構色々詰まったので、やりかたをまとめときます。

AWSIoT Thing作成

このへんは普通に作るだけなので、適当に流します。

デバイス名はdevice00, 以下のようなPolicyを作成し、attachしておきます。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:Connect",
      "Resource": [
        "arn:aws:iot:ap-northeast-1:[YOUR_ACCOUNT_ID]:client/device00"
      ]
    },
    {
      "Effect": "Allow",
      "Action": "iot:Publish",
      "Resource": "arn:aws:iot:ap-northeast-1:[YOUR_ACCOUNT_ID]:topic/device00/token/req"
    },
    {
      "Effect": "Allow",
      "Action": "iot:Subscribe",
      "Resource": "arn:aws:iot:ap-northeast-1:[YOUR_ACCOUNT_ID]:topicfilter/device00/token/res"
    },
    {
      "Effect": "Allow",
      "Action": "iot:Receive",
      "Resource": "arn:aws:iot:ap-northeast-1:[YOUR_ACCOUNT_ID]:topic/device00/token/res"
    }
  ]
}

また、証明書のダウンロードもおこなっておきます。

IAMロールの設定

この後、LambdaでToken発行処理とデバイスからのS3へのアップロード処理を作成するのですが、
関数作成時に該当処理の実行権限をもつIAMロールが必要になるので、先に作っておきます。

S3アップロード処理ロール

デバイスに渡すSTSに許可する権限を付与したロールを"s3-role"として定義します。
ここでは、任意バケット以下にPutObjectのみをおこなえる権限を与えておきます。

実際には、バケット以下のデバイス名のフォルダにのみアップロード可能にする予定ですが、それは
STS発行時にロール権限を制限する形で加えるので、ここではバケット以下への書き込み権限を与えます。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "s3:PutObject"
            ],
            "Resource": [
                "arn:aws:s3:::[YOUR_BACKET_NAME]/*"
            ]
        }
    ]
}

STS発行処理ロール

"s3-role"のロール権限を付与したSTSを作成するLambda関数に与えるロール権限を
"create-token"ロールとして作成します。

Lambda関数を実行するので、既存ポリシーのアタッチで"AWSLambdaBasicExecutionRole"を付与します。

また、作成したSTSをAWSIoTの機能でPublishするので、そのための権限も付与します。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "iot:Publish"
            ],
            "Resource": [
                "arn:aws:iot:ap-northeast-1:[YOUR_ACCOUNT_ID]:topic/device00/token/res"
            ]
        }
    ]
}

そして、STSにs3-roleの権限を付与するための権限も追加します。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "sts:AssumeRole",
            "Resource": "arn:aws:iam::[YOUR_ACCOUNT_ID]:role/s3-role"
        }
    ]
}

さらに、s3-role側でcreate-tokenロールを信頼しAssumeRoleを許可する旨の設定をおこなう必要があります。

s3-roleの設定画面で「信頼関係」タブを選択し、「信頼関係の編集」をおこなってエディタで
JSONの"Statement"要素の配列に以下を追加します。

{
  "Effect": "Allow",
  "Principal": {
    "AWS": "arn:aws:iam::[YOUR_ACCOUNT_ID]:role/create-token"
  },
  "Action": "sts:AssumeRole"
}

STS作成とデバイスへのトークン送信処理

STSを発行し、AWSIoT経由で指定のTopicにTokenをpublishするLambda処理を作成します。

今回はスライドの構成にしたがってTopic経由のpublishにしましたが、STSを使った通信が頻繁に
発生することが想定される場合は、Lambdaをスケジュール起動にしてDeviceShadowを用い定期的に
Pushで通知するほうがいいかもしれません。

import boto3
import json

def lambda_handler(event, context):

    sts = boto3.client('sts')
    iotData = boto3.client('iot-data')

    thingName=event['name']

    token = sts.assume_role(
        RoleArn="arn:aws:iam::[YOUR_ACCOUNT_ID]:role/s3-role",
        RoleSessionName=thingName,
        Policy=json.dumps({
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "",
                    "Effect": "Allow",
                    "Action": [
                        "s3:PutObject"
                    ],
                    "Resource": [
                        "arn:aws:s3:::large-data-upload-pattern/"+thingName+"/*"
                    ]
                }
            ]
        })
    )

    iotData.publish(
        topic=thingName+'/token/res',
        qos=1,
        payload=json.dumps({
            "AccessKeyId": token['Credentials']['AccessKeyId'],
            "SecretAccessKey": token['Credentials']['SecretAccessKey'],
            "SessionToken": token['Credentials']['SessionToken']
        })
    )

eventオブジェクトのメンバー'name'からデバイス名を取得していますが、ここにデバイス名をセットするための
設定は次項でおこないます。

AWSIoTでSTS発行リクエストを受ける

AWSIoTで"*/token/req"のTopicにトークン発行リクエストがpublishされたら、上記のLambdaを起動する
Ruleを作成します。

"Rule query statement" が以下になるように設定し、ActionにはLambda起動を選択して先述の
関数を指定してください。

SELECT topic(1) AS name FROM '+/token/req'

ここまでの設定で、AWSIoTコンソールの"Test"から、"device00/token/req"にメッセージ(空でよい)を
送信するとTokenが"device00/token/res"にpublishされることが確認できます。

あとは動作確認です。

STSを取得してみる

下記サンプルスクリプトを実行すると、"device00/token/req"にトークン発行リクエストを送信し、
"device00/token/res"からトークンを取得して標準出力に表示します。

from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
import json
import time

def cb(client, userdata, message):
    print("got message")
    print(message.payload)

host="[IOT_ENDPOINT_DOMAIN]"
rootCAPath = "[ROOT_CA_FILE_PATH]"
certificatePath = "[CERTIFICATE_FILE_PATH]"
privateKeyPath = "[PRIVATE_KEY_FILE_PATH]"

myAWSIoTMQTTClient = None
myAWSIoTMQTTClient = AWSIoTMQTTClient("device00")
myAWSIoTMQTTClient.configureEndpoint(host, 8883)
myAWSIoTMQTTClient.configureCredentials(rootCAPath, privateKeyPath, certificatePath)

myAWSIoTMQTTClient.configureAutoReconnectBackoffTime(1, 32, 20)
myAWSIoTMQTTClient.configureOfflinePublishQueueing(-1)
myAWSIoTMQTTClient.configureDrainingFrequency(2)
myAWSIoTMQTTClient.configureConnectDisconnectTimeout(10)
myAWSIoTMQTTClient.configureMQTTOperationTimeout(5)

myAWSIoTMQTTClient.connect()
myAWSIoTMQTTClient.subscribe("device00/token/res", 1, cb)

time.sleep(2)

myAWSIoTMQTTClient.publish("device00/token/req", json.dumps({}), 1)

while True:
    time.sleep(5)

S3へのアップロード

取得したトークンを用いて、下記サンプルスクリプトで適当にS3にアップロードしてみます。

import boto3

AWS_S3_BUCKET_NAME = '[YOUR_BACKET_NAME]'

s3 = boto3.resource(
    's3',
    aws_access_key_id="[ACCESS_KEY]",
    aws_secret_access_key="[SECRET_KEY]",
    aws_session_token="[SESSION_TOKEN]"
)
bucket = s3.Bucket(AWS_S3_BUCKET_NAME)

obj = bucket.Object("device00/hogehoge")

response = obj.put(
    Body="hogehogehogehoge".encode('utf-8'),
    ContentEncoding='utf-8',
    ContentType='text/plane'
)

まとめ

まとめてみると大したことではないのですが、AssumeRoleとか権限まわりで色々詰まってよくわかんなくなるので大変でした。