こんにちは！サーバーワークスの松井です。
今回は、IoTデバイスにクレデンシャル情報を配置せずS3からファイルダウンロードする方法を紹介したいと思います。
概要
IoTデバイスでS3の特定のディレクトリにファイルがあるかを確認し、ファイルがあればダウンロードします。 ダウンロードしたファイルは、別のディレクトリに移動させます。
前提条件
以下は設定されているものとします。
IoT Coreでのモノの作成
デバイスへの証明書の配置
python3.9が実行できる環境
手順
1. S3バケットを作成
ダウンロードするファイルを配置するS3バケットを作成します。
デバイス毎にプレフィックスを作成することでプログラム実行デバイスのみがプレフィックスにアクセスできるようにする。
ポリシー変数${credentials-iot:ThingName}を利用する。
例： ${credentials-iot:ThingName}/output
2. IAMロール作成
IAMロールにアタッチするIAMポリシー、信頼ポリシーを作成します。
IAMポリシー
{
"Version": "2012-10-17",
"Statement": [
{
"Condition": {
"StringLike": {
"s3:prefix": "${credentials-iot:ThingName}/*"
}
},
"Action": [
"s3:List*"
],
"Resource": [
"arn:aws:s3:::swx-test-downloads-file"
],
"Effect": "Allow"
},
{
"Action": [
"s3:Get*",
"s3:Delete*",
"s3:Put*"
],
"Resource": [
"arn:aws:s3:::swx-test-downloads-file/${credentials-iot:ThingName}/*"
],
"Effect": "Allow"
}
]
}
信頼ポリシー
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "credentials.iot.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
3. ロールエイリアス作成
IoTデバイス証明書にIAMロールの権限を付与するには、ロールエイリアスを作成し、ポリシーに権限を記載する必要があります。
4. 証明書にアタッチしているポリシーの権限追加
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "iot:AssumeRoleWithCertificate",
"Resource": "arn:aws:iot:ap-northeast-1:{AccountID}:rolealias/swx-test-downloads-file-role-alias"
},
{
"Effect": "Allow",
"Action": "iot:Publish",
"Resource": "*"
},
{
"Effect": "Allow",
"Action": "iot:Connect",
"Resource": "*"
}
]
}
5. デバイス内設定
今回は、IoTデバイス内でpythonのSDKであるboto3を利用するためのクレデンシャル情報を直に置くのではなくプログラムが呼び出されるたびにプログラム内から認証情報を取得するという方法をとります。
デバイス内ではAWS CLIを使えない前提なのでCloudShellを使い、以下コマンドを実行して、一時クレデンシャルを取得するためのエンドポイント情報を取得しておきます。
$ aws iot describe-endpoint --endpoint-type iot:CredentialProvider
{
"endpointAddress": "xxxx.iot.ap-northeast-1.amazonaws.com"
}
5. プログラム作成
以下のディレクトリ階層にてプログラムを作成していきます。
・ ├── conf │ └── setting.conf ├── main.py ├── settings.py
main.py
import requests import json import boto3 import settings from boto3.session import Session cert_filepath = settings.CERT_PATH pri_key_filepath = settings.PRIVATE_KEY_PATH endpoint = settings.IOT_CREDENTIAL_ENDPOINT role_alias = settings.IOT_ROLE_ALIAS bucket_name = settings.BUCKET_NAME prefix = settings.PREFIX another_prefix = settings.ANOTHER_PREFIX file_name = settings.FLIE_NAME device_id = settings.DEVICE_ID def main(): # 一時クレデンシャルを発行し、ファイルの存在判定、ダウンロード、移動 result = requests.get( f'{endpoint}/role-aliases/{role_alias}/credentials', cert=(cert_filepath, pri_key_filepath), headers={"x-amzn-iot-thingname": f'{device_id}'} ) if(result.status_code != 200): print('error occurred') exit() body = json.loads(result.text) access_key = body["credentials"]["accessKeyId"] secret_key = body["credentials"]["secretAccessKey"] token = body["credentials"]["sessionToken"] session = Session(aws_access_key_id=access_key, aws_secret_access_key=secret_key, aws_session_token=token) _s3_client = session.client('s3') _s3_resource = session.resource('s3') if check_data(_s3_client): downloads_file(_s3_resource) mv_file(_s3_client) else: exit() def check_data(_s3_client): # Prefix直下にファイルがあるか判定 response = _s3_client.list_objects( Bucket = bucket_name, Prefix = prefix ) assumed_keys = [f'{prefix}/{file_name}'] try: keys = [content['Key'] for content in response['Contents']] status = set(assumed_keys).issubset(keys) print("file exists") except KeyError: status = False print("file not exists") return status def downloads_file(_s3_resource): # Prefix直下にあるファイルをダウンロード bucket = _s3_resource.Bucket(bucket_name) bucket.download_file(f'{prefix}/{file_name}', file_name) def mv_file(_s3_client): # Prefix直下にあるファイルを別Prefixへ退避 _s3_client.copy_object( Bucket = bucket_name, Key = f'{another_prefix}/{file_name}', CopySource = {'Bucket': bucket_name, 'Key': f'{prefix}/{file_name}'} ) # Prefix直下にあるファイルを削除 _s3_client.delete_object(Bucket = bucket_name, Key = f'{prefix}/{file_name}') if __name__ == "__main__": main()
settings.py
import os import configparser from logging import config # 環境変数設定 conf = configparser.ConfigParser() conf.read(os.path.join(os.path.dirname(os.path.abspath(__file__)), 'conf/setting.conf'), encoding='UTF-8') # DEVICE DEVICE_ID = conf['DEVICE']['DEVICE_ID'] CERT_PATH = conf['DEVICE']['CERT_PATH'] PRIVATE_KEY_PATH = conf['DEVICE']['PRIVATE_KEY_PATH'] # AWS IOT_CREDENTIAL_ENDPOINT = conf['AWS']['IOT_CREDENTIAL_ENDPOINT'] IOT_ROLE_ALIAS = conf['AWS']['IOT_ROLE_ALIAS'] BUCKET_NAME = conf['AWS']['BUCKET_NAME'] PREFIX = conf['AWS']['PREFIX'] ANOTHER_PREFIX = conf['AWS']['ANOTHER_PREFIX'] FLIE_NAME = conf['AWS']['FLIE_NAME']
setting.conf
[DEVICE] DEVICE_ID = xxx # デバイス名 CERT_PATH = xxx # 証明書配置パス PRIVATE_KEY_PATH = xxx # 秘密鍵配置パス [AWS] IOT_CREDENTIAL_ENDPOINT = https://xxx.credentials.iot.ap-northeast-1.amazonaws.com IOT_ROLE_ALIAS = swx-test-downloads-file-role-alias # ロールエイリアス名 BUCKET_NAME = swx-test-downloads-file # S3バケット名 PREFIX = IoT Things名/swx-test-a # ダウンロード元S3ディレクトリ名 ANOTHER_PREFIX = IoT Things名/swx-test-b # 移動先S3ディレクトリ名 FLIE_NAME = test.txt # 配置ファイル名
プログラム実行
事前に以下のモジュールを環境にインストールしておいてください。
pip install requests pip install boto3
実行します。
python main.py
main.pyと同じ階層にtest.txtがダウンロードされており、S3内のディレクトリ「swx-test-a」内のtest.txtが、「swx-test-b」に移動していたら成功です。
まとめ
IoTデバイス内にクレデンシャルを持たせたくないが、AWSのリソースにアクセスしたいという方は多いかと思いますので、今回はS3を例に取りましたが、他のリソースにも同様にアクセスできるので是非参考にしてみてください。
