AWS IoT Core リソースを Python CDK で構築・デプロイ徹底解説


## はじめに

IoT（Internet of Things）デバイスの増加に伴い、AWS IoT Core を活用したインフラの構築がますます重要になっています。

本記事では、AWS Cloud Development Kit (CDK) を使用して Python で IoT Core インフラを定義・デプロイするプロジェクトを徹底的に解説します。
このプロジェクトは、IoTデバイスの管理、セキュリティ、データ収集を効率化するための堅牢な基盤を提供します。

以下のGitHubリポジトリの解説記事となります。

https://github.com/aws-samples/aws-cdk-examples/tree/main/python/iotcore

## プロジェクト概要

このプロジェクトは、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` スタックがアプリケーションに追加されています。

```python
#!/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 のコンテキスト設定も含まれています。

```json
{
  "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 に送信するルールを設定。

```python
# 主要なコードスニペット
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 のライフサイクルに応じて自動的に証明書とキーを生成・削除します。

```python
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 環境で仮想環境を有効化するためのバッチファイル。

    ```bat
    @echo off

    echo Executing .venv\Scripts\activate.bat for you
    .venv\Scripts\activate.bat
    ```

- **tests/**: テストコードが格納されています。`test_cdk_iot_thing_stack.py` では、スタックが正しくリソースを作成しているかを検証するテストが記載されています。

## セットアップ手順

以下の手順でプロジェクトをセットアップし、デプロイを行います。

1. **リポジトリのクローン**

    ```bash
    git clone https://github.com/your-repo/python-iotcore.git
    cd python-iotcore/iotcore
    ```

2. **仮想環境の作成**

    MacOS/Linux:

    ```bash
    python3 -m venv .venv
    source .venv/bin/activate
    ```

    Windows:

    ```bat
    source.bat
    ```

3. **依存パッケージのインストール**

    ```bash
    pip install -r requirements.txt
    ```

    開発環境用のパッケージも必要な場合:

    ```bash
    pip install -r requirements-dev.txt
    ```

4. **CDK のブートストラップ**

    AWS アカウントとリージョンに応じて CDK をブートストラップします。

    ```bash
    cdk bootstrap
    ```

5. **CloudFormation テンプレートの合成**

    ```bash
    cdk synth
    ```

## アーキテクチャの詳細

プロジェクトのアーキテクチャは以下の通りです：

1. **IoT Thing**: AWS IoT Core に登録されたデバイスのエンティティ。
2. **証明書とキーの管理**: Lambda 関数を通じてセキュアな通信を実現。
3. **ポリシー**: デバイスが AWS リソースにアクセスする際の権限を制御。
4. **CloudWatch Logs**: デバイスからのデータをログとして収集・監視。
5. **IAM ロール**: 必要な権限を持つロールを作成し、各サービス間のアクセスを管理。

![IoT Core CDK Architecture](https://raw.githubusercontent.com/aws-samples/aws-cdk-examples/main/python/iotcore/images/IotCore-CDK.png)

## デプロイと使用方法

1. **スタックのデプロイ**

    ```bash
    cdk deploy
    ```

    デプロイが成功すると、IoT Thing、証明書、ポリシー、CloudWatch Logs などのリソースが AWS に作成されます。

2. **使用例**

    - **IoT デバイスの接続**: デバイスは生成された証明書とキーを使用して AWS IoT Core に接続。
    - **データの送信**: デバイスは `device/color` トピックにデータをパブリッシュ。温度が 50 を超える場合、指定されたルールに従い CloudWatch Logs に送信されます。

## テストの実行

プロジェクトには単体テストが含まれています。
テストを実行するには、以下のコマンドを使用します。

```bash
pytest
```

`test_cdk_iot_thing_stack.py` では、スタックが正しくリソースを作成しているかを検証します。
必要に応じて、テストコードを拡張してさらなる検証を行うことが可能です。

## まとめ

本記事では、AWS CDK を用いて Python で IoT Core インフラを構築・デプロイするプロジェクトについて詳しく解説しました。
CDK を活用することで、インフラのコード化が進み、再現性の高いデプロイメントが可能となります。

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

---

**参考リンク**:

- [AWS CDK 公式ドキュメント](https://docs.aws.amazon.com/cdk/latest/guide/home.html)
- [AWS IoT Core 公式ドキュメント](https://docs.aws.amazon.com/iot/latest/developerguide/what-is-aws-iot.html)
- [GitHubリポジトリ](https://github.com/aws-samples/aws-cdk-examples/tree/main/python/iotcore)

---
### 会社紹介

**株式会社 Mosaica**  
最先端テクノロジーで社会課題を解決し、持続可能な未来を創造する IT カンパニー。
AI ソリューション、クラウド統合、DX 推進、経営コンサルティングなど包括的なサービスでビジネス変革を支援しています。  

詳しくは [公式サイト](https://mosaica.co.jp/) までお気軽にご相談ください。
公式サイト： https://mosaica.co.jp/