はじめに

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/activate

Windows:

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 デバイスの管理やセキュリティ、データ収集を効率化したい方は、ぜひこのプロジェクトを参考にしてみてください。

参考リンク: