【AWS IoT Core】 AWS Private CAで認証局を構築してデバイス証明書を発行してみた

# CDKプロジェクトの初期化
mkdir iot-certificate-issuer-cdk
cd iot-certificate-issuer-cdk
cdk init app --language typescript

# 必要なパッケージのインストール
npm install @aws-cdk/aws-lambda-python-alpha

import * as cdk from 'aws-cdk-lib';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as iot from 'aws-cdk-lib/aws-iot';
import * as pythonLambda from '@aws-cdk/aws-lambda-python-alpha';
import { Construct } from 'constructs';

interface IotCertificateIssuerCdkStackProps extends cdk.StackProps {
  privateCAArn: string;
}

export class IotCertificateIssuerCdkStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props: IotCertificateIssuerCdkStackProps) {
    super(scope, id, props);

    // S3バケットの作成
    const certificateBucket = new s3.Bucket(this, 'CertificateBucket', {
      bucketName: `iot-certificates-${cdk.Stack.of(this).account}-${cdk.Stack.of(this).region}`,
      encryption: s3.BucketEncryption.S3_MANAGED,
      versioned: true,
      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
      lifecycleRules: [{
        id: 'DeleteOldVersions',
        noncurrentVersionExpiration: cdk.Duration.days(90),
      }],
      removalPolicy: cdk.RemovalPolicy.DESTROY, // 開発環境用
    });

    // IoTポリシーの作成
    const iotDevicePolicy = new iot.CfnPolicy(this, 'IoTDevicePolicy', {
      policyName: 'IoTDevicePolicy',
      policyDocument: {
        Version: '2012-10-17',
        Statement: [
          {
            Effect: 'Allow',
            Action: ['iot:Connect'],
            Resource: [`arn:aws:iot:${this.region}:${this.account}:client/\${iot:Connection.Thing.ThingName}`],
          },
          {
            Effect: 'Allow',
            Action: ['iot:Publish', 'iot:Receive'],
            Resource: [`arn:aws:iot:${this.region}:${this.account}:topic/iot/devices/\${iot:Connection.Thing.ThingName}/*`],
          },
          {
            Effect: 'Allow',
            Action: ['iot:Subscribe'],
            Resource: [`arn:aws:iot:${this.region}:${this.account}:topicfilter/iot/devices/\${iot:Connection.Thing.ThingName}/*`],
          },
        ],
      },
    });

    // IAMロールとポリシー
    const lambdaRole = new iam.Role(this, 'LambdaExecutionRole', {
      assumedBy: new iam.ServicePrincipal('lambda.amazonaws.com'),
      managedPolicies: [
        iam.ManagedPolicy.fromAwsManagedPolicyName('service-role/AWSLambdaBasicExecutionRole'),
      ],
      inlinePolicies: {
        CertificateIssuerPolicy: new iam.PolicyDocument({
          statements: [
            new iam.PolicyStatement({
              effect: iam.Effect.ALLOW,
              actions: [
                'acm-pca:IssueCertificate',
                'acm-pca:GetCertificate',
                'acm-pca:DescribeCertificateAuthority',
                'acm-pca:GetCertificateAuthorityCertificate',
              ],
              resources: [props.privateCAArn],
            }),
            new iam.PolicyStatement({
              effect: iam.Effect.ALLOW,
              actions: [
                's3:PutObject',
                's3:PutObjectAcl',
                's3:GetBucketLocation',
              ],
              resources: [
                certificateBucket.bucketArn,
                `${certificateBucket.bucketArn}/*`,
              ],
            }),
            new iam.PolicyStatement({
              effect: iam.Effect.ALLOW,
              actions: [
                'iot:RegisterCertificate',
                'iot:CreateThing',
                'iot:AttachThingPrincipal',
                'iot:AttachPolicy',
                'iot:DescribeThing',
              ],
              resources: ['*'],
            }),
          ],
        }),
      },
    });

    // Lambda関数（cryptography依存関係も自動インストール）
    const certificateIssuerFunction = new pythonLambda.PythonFunction(this, 'CertificateIssuerFunction', {
      entry: './lambda',
      runtime: lambda.Runtime.PYTHON_3_11,
      handler: 'lambda_handler',
      role: lambdaRole,
      timeout: cdk.Duration.minutes(1),
      memorySize: 512,
      environment: {
        CA_ARN: props.privateCAArn,
        S3_BUCKET: certificateBucket.bucketName,
        IOT_POLICY_NAME: iotDevicePolicy.policyName!,
      },
    });

    // 出力
    new cdk.CfnOutput(this, 'CertificateBucketName', {
      value: certificateBucket.bucketName,
      description: 'S3 bucket for storing certificates',
    });

    new cdk.CfnOutput(this, 'LambdaFunctionName', {
      value: certificateIssuerFunction.functionName,
      description: 'Lambda function for issuing certificates',
    });

    new cdk.CfnOutput(this, 'IoTPolicyName', {
      value: iotDevicePolicy.policyName!,
      description: 'IoT policy for device authentication',
    });
  }
}

import 'source-map-support/register';
import * as cdk from 'aws-cdk-lib';
import { IotCertificateIssuerCdkStack } from '../lib/iot-certificate-issuer-cdk-stack';

const app = new cdk.App();

// Private CAのARNを環境変数から取得
const privateCAArn = app.node.tryGetContext('privateCAArn');
if (!privateCAArn) {
  throw new Error('パラメータ "privateCAArn" が指定されていません。');
}

new IotCertificateIssuerCdkStack(app, 'IotCertificateIssuerCdkStack', {
  privateCAArn,
  env: {
    account: process.env.CDK_DEFAULT_ACCOUNT,
    region: process.env.CDK_DEFAULT_REGION,
  },
});

cryptography==41.0.7
boto3

import json
import boto3
import os
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from datetime import datetime, timedelta
import time

# AWS クライアントの初期化
acm_pca = boto3.client('acm-pca')
s3 = boto3.client('s3')
iot = boto3.client('iot')

# 環境変数から設定を取得
CA_ARN = os.environ['CA_ARN']
S3_BUCKET = os.environ['S3_BUCKET']
IOT_POLICY_NAME = os.environ.get('IOT_POLICY_NAME', 'IoTDevicePolicy')

def lambda_handler(event, context):
    """
    IoTデバイス用の証明書と秘密鍵を発行し、AWS IoT Coreに登録する
    """
    try:
        # デバイスIDをイベントから取得（デフォルト値も設定）
        device_id = event.get('device_id', f'device-{datetime.now().strftime("%Y%m%d%H%M%S")}')

        print(f"証明書発行開始: デバイスID = {device_id}")

        # 1. 秘密鍵と公開鍵のペアを生成
        private_key = rsa.generate_private_key(
            public_exponent=65537,
            key_size=2048
        )

        # 2. 証明書署名要求（CSR）を作成
        subject = x509.Name([
            x509.NameAttribute(NameOID.COUNTRY_NAME, "JP"),
            x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "Tokyo"),
            x509.NameAttribute(NameOID.LOCALITY_NAME, "Chiyoda-ku"),
            x509.NameAttribute(NameOID.ORGANIZATION_NAME, "MyCompany"),
            x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, "IoT Devices"),
            x509.NameAttribute(NameOID.COMMON_NAME, device_id),
        ])

        csr = x509.CertificateSigningRequestBuilder().subject_name(
            subject
        ).sign(private_key, hashes.SHA256())

        # CSRをPEM形式に変換
        csr_pem = csr.public_bytes(serialization.Encoding.PEM).decode('utf-8')

        # 3. AWS Private CAに証明書発行をリクエスト
        response = acm_pca.issue_certificate(
            CertificateAuthorityArn=CA_ARN,
            Csr=csr_pem,
            SigningAlgorithm='SHA256WITHRSA',
            Validity={
                'Type': 'YEARS',
                'Value': 5  # 5年間有効
            }
        )

        certificate_arn = response['CertificateArn']
        print(f"証明書発行リクエスト送信: ARN = {certificate_arn}")

        # 4. 証明書の発行完了を待って取得
        max_attempts = 10
        certificate_pem = None
        for i in range(max_attempts):
            try:
                cert_response = acm_pca.get_certificate(
                    CertificateAuthorityArn=CA_ARN,
                    CertificateArn=certificate_arn
                )
                certificate_pem = cert_response['Certificate']
                break
            except acm_pca.exceptions.RequestInProgressException:
                if i == max_attempts - 1:
                    raise
                time.sleep(2)
                continue

        # 5. CA証明書を取得
        ca_cert_response = acm_pca.get_certificate_authority_certificate(
            CertificateAuthorityArn=CA_ARN
        )
        ca_certificate_pem = ca_cert_response['Certificate']

        # 6. 秘密鍵をPEM形式に変換
        private_key_pem = private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption()
        ).decode('utf-8')

        # 7. S3に保存
        timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
        prefix = f'certificates/{device_id}/{timestamp}'

        # 証明書を保存
        s3.put_object(
            Bucket=S3_BUCKET,
            Key=f'{prefix}/certificate.pem',
            Body=certificate_pem,
            ServerSideEncryption='AES256',
            Metadata={
                'device-id': device_id,
                'created-at': datetime.now().isoformat()
            }
        )

        # 秘密鍵を保存
        s3.put_object(
            Bucket=S3_BUCKET,
            Key=f'{prefix}/private_key.pem',
            Body=private_key_pem,
            ServerSideEncryption='AES256',
            Metadata={
                'device-id': device_id,
                'created-at': datetime.now().isoformat()
            }
        )

        # CA証明書も保存
        s3.put_object(
            Bucket=S3_BUCKET,
            Key=f'{prefix}/ca_certificate.pem',
            Body=ca_certificate_pem,
            ServerSideEncryption='AES256'
        )

        print(f"証明書と秘密鍵をS3に保存完了: s3://{S3_BUCKET}/{prefix}/")

        # 8. AWS IoT Coreに証明書を登録
        iot_cert_response = iot.register_certificate(
            certificatePem=certificate_pem,
            caCertificatePem=ca_certificate_pem,
            setAsActive=True
        )

        iot_certificate_arn = iot_cert_response['certificateArn']
        iot_certificate_id = iot_cert_response['certificateId']

        # 9. モノ（Thing）を作成
        try:
            iot.create_thing(
                thingName=device_id,
                attributePayload={
                    'attributes': {
                        'device_type': 'iot-sensor',
                        'created_at': datetime.now().isoformat(),
                        'certificate_id': iot_certificate_id
                    }
                }
            )
            print(f"IoT Thing作成完了: {device_id}")
        except iot.exceptions.ResourceAlreadyExistsException:
            print(f"IoT Thing既に存在: {device_id}")

        # 10. 証明書とモノを紐付け
        iot.attach_thing_principal(
            thingName=device_id,
            principal=iot_certificate_arn
        )

        # 11. IoTポリシーをアタッチ
        try:
            iot.attach_policy(
                policyName=IOT_POLICY_NAME,
                target=iot_certificate_arn
            )
            print(f"IoTポリシーアタッチ完了: {IOT_POLICY_NAME}")
        except Exception as e:
            print(f"IoTポリシーアタッチ警告: {str(e)}")

        return {
            'statusCode': 200,
            'body': json.dumps({
                'device_id': device_id,
                'certificate_location': f's3://{S3_BUCKET}/{prefix}/certificate.pem',
                'private_key_location': f's3://{S3_BUCKET}/{prefix}/private_key.pem',
                'ca_certificate_location': f's3://{S3_BUCKET}/{prefix}/ca_certificate.pem',
                'certificate_arn': certificate_arn,
                'iot_certificate_arn': iot_certificate_arn,
                'iot_certificate_id': iot_certificate_id
            })
        }

    except Exception as e:
        print(f"エラーが発生しました: {str(e)}")
        import traceback
        traceback.print_exc()
        return {
            'statusCode': 500,
            'body': json.dumps({
                'error': str(e),
                'device_id': device_id if 'device_id' in locals() else 'unknown'
            })
        }

# CDKのブートストラップ（初回のみ）
cdk bootstrap

# スタックのデプロイ
cdk deploy --context privateCAArn=arn:aws:acm-pca:ap-northeast-1:123456789012:certificate-authority/xxxxx

REGISTRATION_CODE=$(aws iot get-registration-code --output text) echo "登録コード: $REGISTRATION_CODE"

openssl genrsa -out verification_cert.key 2048

openssl req -new \
    -key verification_cert.key \
    -out verification_cert.csr

You are about to be asked to enter information that will be incorporated
into your certificate request.
What you are about to enter is what is called a Distinguished Name or a DN.
There are quite a few fields but you can leave some blank
For some fields there will be a default value,
If you enter '.', the field will be left blank.
-----
Country Name (2 letter code) [AU]:
State or Province Name (full name) [Some-State]:
Locality Name (eg, city) []:
Organization Name (eg, company) [Internet Widgits Pty Ltd]:
Organizational Unit Name (eg, section) []:
Common Name (e.g. server FQDN or YOUR name) []: <登録コードをここに入力>
Email Address []:

Please enter the following 'extra' attributes
to be sent with your certificate request
A challenge password []:
An optional company name []:

# PrivateCAのARNを入力
CA_ARN=arn:aws:acm-pca:ap-northeast-1:123456789012:certificate-authority/xxxxx

# CA証明書を取得
aws acm-pca get-certificate-authority-certificate \
    --certificate-authority-arn $CA_ARN \
    --output text > rootCA.pem

#AWS Private CAで検証証明書を発行（fileb://を使用！） 
VERIFICATION_CERT_ARN=$(aws acm-pca issue-certificate \
    --certificate-authority-arn $CA_ARN \
    --csr fileb://verification_cert.csr \
    --signing-algorithm SHA256WITHRSA \
    --validity Type=DAYS,Value=365 \
    --output text)

# 検証証明書を作成
aws acm-pca get-certificate \
    --certificate-authority-arn $CA_ARN \
    --certificate-arn $VERIFICATION_CERT_ARN \
    --query 'Certificate' \
    --output text > verification_cert.pem

aws iot register-ca-certificate \
    --ca-certificate file://rootCA.pem \
    --verification-cert file://verification_cert.pem

{
    "certificateArn": "arn:aws:iot:ap-northeast-1:ACCOUNT_ID:cacert/CERTIFICATE_ID",
    "certificateId": "CERTIFICATE_ID"
}

aws iot update-ca-certificate \
    --certificate-id <CERTIFICATE_ID> \
    --new-status ACTIVE

{
  "device_id": "test-device-001"
}

# 証明書をダウンロード
aws s3 cp s3://iot-certificates-${ACCOUNT_ID}-${REGION}/certificates/test-device-001/20250106120000/certificate.pem ./

# 証明書の内容を確認
openssl x509 -in certificate.pem -text -noout

Certificate:
    Data:
        Version: 3 (0x2)
        Serial Number:
            xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:xx
        Signature Algorithm: sha256WithRSAEncryption
        Issuer: C = JP, ST = Tokyo, L = Chiyoda-ku, O = MyCompany, OU = IoT Division, CN = IoT-Device-Root-CA
        Validity
            Not Before: Jan  6 12:00:00 2025 GMT
            Not After : Jan  6 12:00:00 2030 GMT
        Subject: C = JP, ST = Tokyo, L = Chiyoda-ku, O = MyCompany, OU = IoT Devices, CN = test-device-001
        ...

# デバイス証明書チェーンファイルを作成（デバイス証明書 + 独自CA証明書で結合）
cat certificate.pem ca_certificate.pem > device_certificate.pem

# AWSのルートCA証明書をダウンロード（サーバー証明書検証用）
curl -o AmazonRootCA1.pem https://www.amazontrust.com/repository/AmazonRootCA1.pem

aws iot describe-endpoint --endpoint-type iot:Data-ATS

# 実行結果
{ "endpointAddress": "xxxxxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com" }

pip install awsiotsdk

from awscrt import io, mqtt
from awsiot import mqtt_connection_builder
import time
import json

# エンドポイントの取得
endpoint = "xxxxxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com"

# 接続の構築
mqtt_connection = mqtt_connection_builder.mtls_from_path(
    endpoint=endpoint,
    cert_filepath="device_certificate.pem",
    pri_key_filepath="private_key.pem",
    ca_filepath="AmazonRootCA1.pem",
    client_id="test-device-001",
    clean_session=False,
    keep_alive_secs=6
)

# 接続
connect_future = mqtt_connection.connect()
connect_future.result()
print("接続成功！")

# テストメッセージの送信
message = {
    "device_id": "test-device-001",
    "timestamp": time.time(),
    "temperature": 25.5,
    "humidity": 60.2
}

mqtt_connection.publish(
    topic="iot/devices/test-device-001/telemetry",
    payload=json.dumps(message),
    qos=mqtt.QoS.AT_LEAST_ONCE
)
print("メッセージ送信完了！")

# 切断
mqtt_connection.disconnect()

python send_data.py

接続成功！
メッセージ送信完了！