Azure Device Provisioning Service (DPS) で個別登録を試してみた

# グループ登録でのデバイスキー導出例
az iot dps enrollment-group compute-device-key \
    --key <YOUR_ENROLLMENT_GROUP_PRIMARY_KEY> \
    --registration-id <YOUR_DEVICE_REGISTRATION_ID>

# 作業ディレクトリの作成
mkdir azure-iot-individual-certs
cd azure-iot-individual-certs

# デバイス1用の証明書作成
openssl genrsa -out individual-device1-x509.key 2048

# CSRの生成
openssl req -new -key individual-device1-x509.key -out individual-device1-x509.csr -subj \"/CN=individual-device1-x509\"

# 自己署名証明書の生成（SHA256を明示的に指定）
openssl x509 -req -in individual-device1-x509.csr -signkey individual-device1-x509.key -out individual-device1-x509.pem -days 365 -sha256

pip install azure-iot-device

import os
import asyncio
import datetime
import random
import json
from azure.iot.device.aio import ProvisioningDeviceClient
from azure.iot.device.aio import IoTHubDeviceClient

# DPSの設定情報 - ご自身の環境に合わせて変更してください
PROVISIONING_HOST = "global.azure-devices-provisioning.net"
ID_SCOPE = "xxx"  # DPSインスタンスのIDスコープ

# 個別登録のデバイス情報（1台分）
REGISTRATION_ID = "individual-device1"
SYMMETRIC_KEY = "xxx"  # individual-device1の主キー

async def provision_device():
    print(f"[{REGISTRATION_ID}] 対称キーを使ってDPSに接続し、IoT Hubへの登録を開始します...")

    # DPSクライアントの作成
    provisioning_device_client = ProvisioningDeviceClient.create_from_symmetric_key(
        provisioning_host=PROVISIONING_HOST,
        registration_id=REGISTRATION_ID,
        id_scope=ID_SCOPE,
        symmetric_key=SYMMETRIC_KEY,
    )

    # DPSに接続して登録
    print(f"[{REGISTRATION_ID}] DPSに接続して登録を行っています...")
    registration_result = await provisioning_device_client.register()

    print(f"[{REGISTRATION_ID}] DPS登録ステータス: {registration_result.status}")
    if registration_result.status == "assigned":
        print(f"[{REGISTRATION_ID}] デバイスがIoT Hubに登録されました: {registration_result.registration_state.assigned_hub}")
        print(f"[{REGISTRATION_ID}] デバイスID: {registration_result.registration_state.device_id}")

        # IoT Hubに接続してテストメッセージを送信
        device_client = IoTHubDeviceClient.create_from_symmetric_key(
            symmetric_key=SYMMETRIC_KEY,
            hostname=registration_result.registration_state.assigned_hub,
            device_id=registration_result.registration_state.device_id,
        )

        await device_client.connect()
        print(f"[{REGISTRATION_ID}] IoT Hub に接続しました！")

        # テストメッセージを送信
        for i in range(3):
            message = {
                "temperature": round(random.uniform(20.0, 30.0), 1),
                "humidity": round(random.uniform(50.0, 70.0), 1),
                "message": f"Hello from individual enrollment device {REGISTRATION_ID} (message {i+1}/3)",
                "timestamp": datetime.datetime.now().isoformat(),
                "deviceId": REGISTRATION_ID
            }
            print(f"[{REGISTRATION_ID}] メッセージを送信: {json.dumps(message)}")
            await device_client.send_message(json.dumps(message))
            await asyncio.sleep(1)

        await device_client.disconnect()
        print(f"[{REGISTRATION_ID}] IoT Hubから切断しました")
    else:
        print(f"[{REGISTRATION_ID}] デバイスの登録に失敗しました。ステータス: {registration_result.status}")

async def main():
    print("対称キー個別登録のテストを開始します...")
    await provision_device()
    print("処理が完了しました")

if __name__ == "__main__":
    asyncio.run(main())

import os
import asyncio
import datetime
import random
import json
from azure.iot.device import X509
from azure.iot.device.aio import ProvisioningDeviceClient
from azure.iot.device.aio import IoTHubDeviceClient

# DPSの設定情報 - ご自身の環境に合わせて変更してください
PROVISIONING_HOST = "global.azure-devices-provisioning.net"
ID_SCOPE = "xxx"  # DPSインスタンスのIDスコープ

# 個別登録のデバイス証明書情報（1台分）
REGISTRATION_ID = "individual-device1-x509"
CERT_PATH = "./individual-device1-x509.pem"
KEY_PATH = "./individual-device1-x509.key"

async def provision_device():
    print(f"[{REGISTRATION_ID}] X.509証明書を使ってDPSに接続し、IoT Hubへの登録を開始します...")

    # X.509証明書の設定
    x509 = X509(
        cert_file=CERT_PATH,
        key_path=KEY_PATH,
        pass_phrase=None
    )

    # DPSクライアントの作成
    provisioning_device_client = ProvisioningDeviceClient.create_from_x509_certificate(
        provisioning_host=PROVISIONING_HOST,
        registration_id=REGISTRATION_ID,
        id_scope=ID_SCOPE,
        x509=x509,
    )

    # DPSに接続して登録
    print(f"[{REGISTRATION_ID}] DPSに接続して登録を行っています...")
    registration_result = await provisioning_device_client.register()

    print(f"[{REGISTRATION_ID}] DPS登録ステータス: {registration_result.status}")
    if registration_result.status == "assigned":
        print(f"[{REGISTRATION_ID}] デバイスがIoT Hubに登録されました: {registration_result.registration_state.assigned_hub}")
        print(f"[{REGISTRATION_ID}] デバイスID: {registration_result.registration_state.device_id}")

        # IoT Hubに接続してテストメッセージを送信
        device_client = IoTHubDeviceClient.create_from_x509_certificate(
            x509=x509,
            hostname=registration_result.registration_state.assigned_hub,
            device_id=registration_result.registration_state.device_id,
        )

        await device_client.connect()
        print(f"[{REGISTRATION_ID}] IoT Hub に接続しました！")

        # テストメッセージを送信
        for i in range(3):
            message = {
                "temperature": round(random.uniform(20.0, 30.0), 1),
                "humidity": round(random.uniform(50.0, 70.0), 1),
                "message": f"Hello from individual X.509 device {REGISTRATION_ID} (message {i+1}/3)",
                "timestamp": datetime.datetime.now().isoformat(),
                "deviceId": REGISTRATION_ID
            }
            print(f"[{REGISTRATION_ID}] メッセージを送信: {json.dumps(message)}")
            await device_client.send_message(json.dumps(message))
            await asyncio.sleep(1)

        await device_client.disconnect()
        print(f"[{REGISTRATION_ID}] IoT Hubから切断しました")
    else:
        print(f"[{REGISTRATION_ID}] デバイスの登録に失敗しました。ステータス: {registration_result.status}")

async def main():
    print("X.509証明書個別登録のテストを開始します...")
    await provision_device()
    print("処理が完了しました")

if __name__ == "__main__":
    asyncio.run(main())

# 対称キーのテスト
python dps_individual_symmetric_test.py

# X.509証明書のテスト
python dps_individual_x509_test.py

[individual-device1] 対称キーを使ってDPSに接続し、IoT Hubへの登録を開始します...
[individual-device1] DPSに接続して登録を行っています...
[individual-device1] DPS登録ステータス: assigned
[individual-device1] デバイスがIoT Hubに登録されました: sample-iot-hub.azure-devices.net
[individual-device1] デバイスID: individual-device1
[individual-device1] IoT Hub に接続しました！
[individual-device1] メッセージを送信: {"temperature": 27.4, "humidity": 62.0, "message": "Hello from individual enrollment device individual-device1 (message 1/3)", "timestamp": "2025-05-23T08:35:41.377291", "deviceId": "individual-device1"}
[individual-device1] メッセージを送信: {"temperature": 28.6, "humidity": 66.8, "message": "Hello from individual enrollment device individual-device1 (message 2/3)", "timestamp": "2025-05-23T08:35:42.667731", "deviceId": "individual-device1"}
[individual-device1] メッセージを送信: {"temperature": 23.3, "humidity": 69.0, "message": "Hello from individual enrollment device individual-device1 (message 3/3)", "timestamp": "2025-05-23T08:35:43.893072", "deviceId": "individual-device1"}
[individual-device1] IoT Hubから切断しました
処理が完了しました

[individual-device1-x509] X.509証明書を使ってDPSに接続し、IoT Hubへの登録を開始します...
[individual-device1-x509] DPSに接続して登録を行っています...
[individual-device1-x509] DPS登録ステータス: assigned
[individual-device1-x509] デバイスがIoT Hubに登録されました: sample-iot-hub.azure-devices.net
[individual-device1-x509] デバイスID: individual-device1-x509
[individual-device1-x509] IoT Hub に接続しました！
[individual-device1-x509] メッセージを送信: {"temperature": 24.1, "humidity": 56.6, "message": "Hello from individual X.509 device individual-device1-x509 (message 1/3)", "timestamp": "2025-05-23T09:08:07.317973", "deviceId": "individual-device1-x509"}
[individual-device1-x509] メッセージを送信: {"temperature": 21.1, "humidity": 65.8, "message": "Hello from individual X.509 device individual-device1-x509 (message 2/3)", "timestamp": "2025-05-23T09:08:08.573137", "deviceId": "individual-device1-x509"}
[individual-device1-x509] メッセージを送信: {"temperature": 29.0, "humidity": 64.3, "message": "Hello from individual X.509 device individual-device1-x509 (message 3/3)", "timestamp": "2025-05-23T09:08:09.702152", "deviceId": "individual-device1-x509"}
[individual-device1-x509] IoT Hubから切断しました
処理が完了しました

az iot hub monitor-events --hub-name YourIoTHubName