はじめに
IoT（Internet of Things）デバイスの増加に伴い、AWS IoT Core を活用したインフラの構築がますます重要になっています。
本記事では、AWS Cloud Development Kit (CDK) を使用して Python で IoT Core インフラを定義・デプロイするプロジェクトを徹底的に解説します。
このプロジェクトは、IoTデバイスの管理、セキュリティ、データ収集を効率化するための堅牢な基盤を提供します。
以下のGitHubリポジトリの解説記事となります。
プロジェクト概要
このプロジェクトは、AWS CDK を用いて IoT インフラを自動化・効率化することを目的としています。
具体的には、以下の機能を提供します：
- IoT Thing の作成: デバイスを AWS IoT Core に登録。
- 証明書とキーの管理: セキュアな通信のための証明書とキーを自動生成・管理。
- ポリシーの設定: デバイスのアクセス権限を制御。
- CloudWatch Logs との統合: デバイスからのデータをログとして収集・監視。
リポジトリ構成
以下は、プロジェクトのディレクトリ構造です：
python/
└── iotcore/
├── README.md
├── app.py
├── cdk.json
├── cdk_iot_thing/
│ ├── __init__.py
│ └── cdk_iot_thing_stack.py
├── images/
│ └── IotCore-CDK.png
├── lambda/
│ └── cert_handler.py
├── requirements-dev.txt
├── requirements.txt
├── source.bat
└── tests/
├── __init__.py
└── unit/
├── __init__.py
└── test_cdk_iot_thing_stack.py
ディレクトリとファイルの説明
- README.md: プロジェクトの概要、セットアップ手順、使用方法を記載。
- app.py: CDK アプリケーションのエントリーポイント。
- cdk.json: CDK の設定ファイル。
-
cdk_iot_thing/: メインの CDK スタック定義。
- cdk_iot_thing_stack.py: IoT インフラを定義するスタック。
- images/: プロジェクトのアーキテクチャ図などの画像。
-
lambda/: カスタムリソース用の Lambda 関数。
- cert_handler.py: 証明書とキーの生成・管理を行う Lambda 関数。
- requirements.txt: プロジェクトの依存パッケージ。
- requirements-dev.txt: 開発用の依存パッケージ（テストフレームワークなど）。
- source.bat: Windows 環境での仮想環境の有効化スクリプト。
-
tests/: テストコード。
- unit/: 単体テスト。
主要ファイルの詳細解説
app.py
CDK アプリケーションのエントリーポイントです。
ここでは、CdkIotThingStack スタックがアプリケーションに追加されています。
#!/usr/bin/env python3
import os
import aws_cdk as cdk
from cdk_iot_thing.cdk_iot_thing_stack import CdkIotThingStack
app = cdk.App()
CdkIotThingStack(app, "CdkIotThingStack")
app.synth()
cdk.json
CDK ツールキットがアプリケーションをどのように実行するかを定義する設定ファイルです。
また、CDK のコンテキスト設定も含まれています。
{
"app": "python3 app.py",
"watch": {
"include": ["**"],
"exclude": ["README.md", "cdk*.json", "requirements*.txt", "source.bat", "**/__init__.py", "**/__pycache__", "tests"]
},
"context": {
// 省略: 各種 CDK コンテキスト設定
}
}
cdk_iot_thing_stack.py
このファイルがプロジェクトの中心となり、AWS IoT Core のリソースを定義します。
具体的には、IoT Thing の作成、証明書とキーの管理、ポリシーの設定、CloudWatch Logs との統合などを行います。
主なポイント：
-
IoT Thing の作成:
CfnThingを使用して IoT Thing を作成。 - Lambda 関数の設定: 証明書とキーを管理するための Lambda 関数をデプロイ。
- カスタムリソース: Lambda 関数をカスタムリソースとして統合し、証明書の生成と削除を自動化。
- ポリシーの設定: IoT Thing に必要な権限を定義するポリシーを作成・アタッチ。
- CloudWatch Logs との統合: IoT データを CloudWatch Logs に送信するルールを設定。
# 主要なコードスニペット
import os
from aws_cdk import (
Stack,
aws_logs as logs,
aws_iot as iot,
aws_iam as iam,
aws_lambda as _lambda,
custom_resources as CustomResourceProvider,
CustomResource,
Duration,
Aws
)
from constructs import Construct
class CdkIotThingStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# 環境変数
account = Aws.ACCOUNT_ID
region = Aws.REGION
iot_thing_name = "CdkThing001"
# IoT Thing の作成
cfn_thing = iot.CfnThing(self, "MyCdkThing", thing_name=iot_thing_name)
# Lambda のロール作成
lambda_role = iam.Role(self, f"{iot_thing_name}LambdaRole",
assumed_by=iam.ServicePrincipal("lambda.amazonaws.com")
)
# ポリシーの追加
lambda_role.add_to_policy(iam.PolicyStatement(
actions=["secretsmanager:CreateSecret","secretsmanager:DeleteSecret"],
resources=["arn:aws:secretsmanager:*:*:secret:*"]
))
lambda_role.add_to_policy(iam.PolicyStatement(
actions=["iot:CreateKeysAndCertificate", "iot:UpdateCertificate"],
resources=["*"]
))
# Lambda 関数の作成
cert_lambda = _lambda.Function(
self, "CertHandler",
function_name="CertHandlerFunction",
runtime=_lambda.Runtime.PYTHON_3_11,
code=_lambda.Code.from_asset("lambda"),
handler="cert_handler.lambda_handler",
role=lambda_role,
log_retention=logs.RetentionDays.ONE_DAY,
timeout=Duration.seconds(60)
)
# カスタムリソースプロバイダーの設定
provider = CustomResourceProvider.Provider(
self, 'IoTCertProvider',
on_event_handler=cert_lambda
)
certificate = CustomResource(
self, "IoTCertCustomResource",
service_token=provider.service_token
)
# ポリシーの作成
cfn_policy = iot.CfnPolicy(self, "CfnPolicy",
policy_document={
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": ["iot:Connect"],
"Resource": [f"arn:aws:iot:{region}:{account}:client/{iot_thing_name}"]
},
{
"Effect": "Allow",
"Action": ["iot:Publish"],
"Resource": [f"arn:aws:iot:{region}:{account}:topic/*"]
}
]
},
policy_name=f"{iot_thing_name}IoTCertPolicy"
)
# ポリシーと証明書のアタッチ
policy_attachment = iot.CfnPolicyPrincipalAttachment(
self,
id=iot_thing_name + "PolicyPrincipalAttachment",
policy_name=iot_thing_name + "IoTCertPolicy",
principal=f"arn:aws:iot:{region}:{account}:cert/{certificate.get_att_string('certificateId')}"
)
# IoT Thing と証明書のアタッチ
principal_attachment = iot.CfnThingPrincipalAttachment(
self,
id=iot_thing_name + "ThingPrincipalAttachment",
principal=f"arn:aws:iot:{region}:{account}:cert/{certificate.get_att_string('certificateId')}",
thing_name=iot_thing_name
)
# CloudWatch Logs 用の IAM ロールとポリシー
iam_role_name = "CdkIoTCoreCloudWatchAccessRole"
cfn_role = iam.CfnRole(self, "CfnRole",
assume_role_policy_document=iam.PolicyDocument(
statements=[iam.PolicyStatement(
actions=["sts:AssumeRole"],
effect=iam.Effect.ALLOW,
principals=[iam.ServicePrincipal("iot.amazonaws.com")]
)]
),
description="CDK created role for logging IoT rule event",
role_name=iam_role_name,
policies=[iam.CfnRole.PolicyProperty(
policy_document=iam.PolicyDocument(
statements=[iam.PolicyStatement(
actions=["logs:CreateLogStream", "logs:DescribeLogStreams", "logs:PutLogEvents"],
effect=iam.Effect.ALLOW,
resources=["*"]
)]
),
policy_name="CdkIoTCoreCloudWatchAccessPolicy"
)]
)
# ロググループの作成
logGroupName = "CdkThing001LogGroup"
cfn_log_group = logs.CfnLogGroup(self, "CfnLogGroup",
log_group_name=logGroupName,
retention_in_days=7
)
# IoT ルールの設定
iot_topic_rule_sql = "SELECT color AS rgb FROM 'device/color' WHERE temperature > 50"
iot_topic_rule = iot.CfnTopicRule(
self, "CdkIoTCoreRule",
topic_rule_payload=iot.CfnTopicRule.TopicRulePayloadProperty(
sql=iot_topic_rule_sql,
actions=[iot.CfnTopicRule.ActionProperty(
cloudwatch_logs=iot.CfnTopicRule.CloudwatchLogsActionProperty(
log_group_name=logGroupName,
role_arn=cfn_role.attr_arn,
batch_mode=False
)
)]
)
)
Lambda 関数: cert_handler.py
この Lambda 関数は、IoT 証明書とキーの生成および削除を管理します。
カスタムリソースとして利用され、IoT Thing のライフサイクルに応じて自動的に証明書とキーを生成・削除します。
import json
import time
from botocore.exceptions import ClientError
import boto3
SECRET_NAME = "iot-cert-and-key-5"
iot = boto3.client('iot')
secrets_manager = boto3.client('secretsmanager')
def on_create(event):
response = iot.create_keys_and_certificate(setAsActive=True)
certificate_id = response['certificateId']
certificate_pem = response['certificatePem']
key_pair = response['keyPair']
if not certificate_id or not certificate_pem or not key_pair:
raise ValueError('Failed to create keys and certificate')
secrets_manager.create_secret(
Name=SECRET_NAME,
SecretString=json.dumps({
'cert': certificate_pem,
'keyPair': key_pair['PrivateKey']
})
)
return {
'PhysicalResourceId': certificate_id,
'Data': {
'certificateId': certificate_id
}
}
def on_delete(event):
try:
secrets_manager.delete_secret(
SecretId=SECRET_NAME,
ForceDeleteWithoutRecovery=True
)
except ClientError:
pass
certificate_id = event['PhysicalResourceId']
iot.update_certificate(
certificateId=certificate_id,
newStatus='INACTIVE'
)
time.sleep(2)
iot.delete_certificate(
certificateId=certificate_id
)
def lambda_handler(event, context):
print('Received event:', event)
if event['RequestType'] == 'Create':
return on_create(event)
elif event['RequestType'] == 'Delete':
return on_delete(event)
raise ValueError('Unknown request type')
その他のファイル
-
requirements.txt: プロジェクトで必要なパッケージを指定。主に
aws-cdk-libとconstructsが含まれています。aws-cdk-lib==2.101.1 constructs>=10.0.0,<11.0.0 -
requirements-dev.txt: 開発環境用のパッケージ。テストフレームワークの
pytestが含まれています。pytest==6.2.5 -
source.bat: Windows 環境で仮想環境を有効化するためのバッチファイル。
@echo off echo Executing .venv\Scripts\activate.bat for you .venv\Scripts\activate.bat -
tests/: テストコードが格納されています。
test_cdk_iot_thing_stack.pyでは、スタックが正しくリソースを作成しているかを検証するテストが記載されています。
セットアップ手順
以下の手順でプロジェクトをセットアップし、デプロイを行います。
-
リポジトリのクローン
git clone https://github.com/your-repo/python-iotcore.git cd python-iotcore/iotcore -
仮想環境の作成
MacOS/Linux:
python3 -m venv .venv source .venv/bin/activateWindows:
source.bat -
依存パッケージのインストール
pip install -r requirements.txt開発環境用のパッケージも必要な場合:
pip install -r requirements-dev.txt -
CDK のブートストラップ
AWS アカウントとリージョンに応じて CDK をブートストラップします。
cdk bootstrap -
CloudFormation テンプレートの合成
cdk synth
アーキテクチャの詳細
プロジェクトのアーキテクチャは以下の通りです：
- IoT Thing: AWS IoT Core に登録されたデバイスのエンティティ。
- 証明書とキーの管理: Lambda 関数を通じてセキュアな通信を実現。
- ポリシー: デバイスが AWS リソースにアクセスする際の権限を制御。
- CloudWatch Logs: デバイスからのデータをログとして収集・監視。
- IAM ロール: 必要な権限を持つロールを作成し、各サービス間のアクセスを管理。
デプロイと使用方法
-
スタックのデプロイ
cdk deployデプロイが成功すると、IoT Thing、証明書、ポリシー、CloudWatch Logs などのリソースが AWS に作成されます。
-
使用例
- IoT デバイスの接続: デバイスは生成された証明書とキーを使用して AWS IoT Core に接続。
-
データの送信: デバイスは
device/colorトピックにデータをパブリッシュ。温度が 50 を超える場合、指定されたルールに従い CloudWatch Logs に送信されます。
テストの実行
プロジェクトには単体テストが含まれています。
テストを実行するには、以下のコマンドを使用します。
pytest
test_cdk_iot_thing_stack.py では、スタックが正しくリソースを作成しているかを検証します。
必要に応じて、テストコードを拡張してさらなる検証を行うことが可能です。
まとめ
本記事では、AWS CDK を用いて Python で IoT Core インフラを構築・デプロイするプロジェクトについて詳しく解説しました。
CDK を活用することで、インフラのコード化が進み、再現性の高いデプロイメントが可能となります。
IoT デバイスの管理やセキュリティ、データ収集を効率化したい方は、ぜひこのプロジェクトを参考にしてみてください。
参考リンク: