Amazon Cognito 認証を使用して AWS IoT Core と通信する

はじめに

この記事の内容は、基本的に生成 AI が出力したコードを使用して検証しています。

概要

Amazon Cognito のユーザが、一時的なクレデンシャル情報を使用して AWS IoT Core に接続し、特定のトピックに対してメッセージを送信する。
バックエンドアプリケーション上で Cognito ユーザのクレデンシャルを発行し IoT Core のエンドポイントと通信する。

実装

アプリケーションコードは Python で実装。
ライブラリは awsiotsdk, boto3 を使用。

事前作業

1. ユーザプールの設定画面でアプリクライアントを作成する。
これは Cognito User Pool のユーザ認証で使用する。

2. IoT Core のポリシー定義。
接続元リソース指定やトピックに対する Pub/Sub 許可指定を定義。 AWS IoT ポリシーとして登録する。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:Connect",
      "Resource": [
        "arn:aws:iot:ap-northeast-1:XXXXXXXXXX:client/${cognito-identity.amazonaws.com:sub}",
      ]
    },
    {
      "Effect": "Allow",
      "Action": "iot:Subscribe",
      "Resource": [
        "arn:aws:iot:ap-northeast-1:XXXXXXXXXX:topicfilter/$aws/events/presence/connected/*",
        "arn:aws:iot:ap-northeast-1:XXXXXXXXXX:topicfilter/$aws/events/presence/disconnected/*"
      ]
    },
    {
      "Effect": "Allow",
      "Action": "iot:Receive",
      "Resource": [
        "arn:aws:iot:ap-northeast-1:XXXXXXXXXX:topic/$aws/events/presence/connected/*",
        "arn:aws:iot:ap-northeast-1:XXXXXXXXXX:topic/$aws/events/presence/disconnected/*"
      ]
    },
    {
      "Effect": "Allow",
      "Action": "iot:Publish",
      "Resource": "arn:aws:iot:ap-northeast-1:XXXXXXXXXX:topic/$aws/things/device_*"
    },
    {
      "Effect": "Allow",
      "Action": "iot:Subscribe",
      "Resource": "arn:aws:iot:ap-northeast-1:XXXXXXXXXX:topicfilter/$aws/things/device_*"
    },
    {
      "Effect": "Allow",
      "Action": "iot:Receive",
      "Resource": "arn:aws:iot:ap-northeast-1:XXXXXXXXXX:topic/$aws/things/device_*"
    }
  ]
}

3. Cognito ユーザのクレデンシャル発行時にアタッチする IAM ロールを作成。

IoT Core への接続と操作権限を付与するロール。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "iot:Connect",
            "Resource": [
                "arn:aws:iot:ap-northeast-1:XXXXXXXXX:client/${cognito-identity.amazonaws.com:sub}"
            ]
        },
        {
            "Effect": "Allow",
            "Action": "iot:Subscribe",
            "Resource": [
                "arn:aws:iot:ap-northeast-1:XXXXXXXXX:topicfilter/$aws/events/presence/connected/*",
                "arn:aws:iot:ap-northeast-1:XXXXXXXXX:topicfilter/$aws/events/presence/disconnected/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": "iot:Receive",
            "Resource": [
                "arn:aws:iot:ap-northeast-1:XXXXXXXXX:topic/$aws/events/presence/connected/*",
                "arn:aws:iot:ap-northeast-1:XXXXXXXXX:topic/$aws/events/presence/disconnected/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": "iot:Publish",
            "Resource": "arn:aws:iot:ap-northeast-1:XXXXXXXXX:topic/$aws/things/device_*"
        },
        {
            "Effect": "Allow",
            "Action": "iot:Subscribe",
            "Resource": "arn:aws:iot:ap-northeast-1:XXXXXXXXX:topicfilter/$aws/things/device_*"
        },
        {
            "Effect": "Allow",
            "Action": "iot:Receive",
            "Resource": "arn:aws:iot:ap-northeast-1:XXXXXXXXX:topic/$aws/things/device_*"
        }
    ]
}

4. Cognito ユーザのクレデンシャル発行時に、作成した IAM ロールを割り当てる設定。
Cognito ID プールの ID プロバイダの設定から、作成したアプリクライアントで認証したユーザに IoT Core への接続と操作権限を付与したロールを割り当てるよう設定。

処理の流れ

1. Cognito ユーザのユーザ名とパスワードを使用し Cognito User Pool でユーザ認証を行う。

# シークレットハッシュの算出
msg = username + CLIENT_ID
dig = hmac.new(
    client_secret.encode("utf-8"),
    msg.encode("utf-8"),
    hashlib.sha256
).digest()

secret_hash = base64.b64encode(dig).decode()

# 認証
response = client_idp.initiate_auth(
    ClientId=CLIENT_ID,
    AuthFlow='USER_PASSWORD_AUTH',
    AuthParameters={
        'USERNAME': user_name,
        'PASSWORD': user_password,
        'SECRET_HASH': secret_hash
    }
)

# 認証結果から ID トークンを抽出
id_token = response['AuthenticationResult']['IdToken']

2. ID トークンを使用し Cognito Identity プールから一時的なクレデンシャルを取得する。

# ID トークンを使用して Cognito Identity ID を取得
response = client_identity.get_id(
    IdentityPoolId=IDENTITY_POOL_ID,
    Logins={
        f'cognito-idp.{REGION}.amazonaws.com/{USER_POOL_ID}': id_token
    }
)
identity_id = response['IdentityId']

# クレデンシャル発行
credentials = client_identity.get_credentials_for_identity(
    IdentityId=identity_id,
    Logins={
        f'cognito-idp.{REGION}.amazonaws.com/{USER_POOL_ID}': id_token
    }
)["Credentials"]

access_key = credentials['AccessKeyId']
secret_key = credentials['SecretKey']
session_token = credentials['SessionToken']

3. Cognito ユーザの Identity ID に、作成した IoT Core の IoT ポリシーをアタッチする。

# 対象ポリシーをアタッチ
client_iot.attach_policy(
    policyName=IOT_POLICY_NAME,
    target=identity_id
)

4. クレデンシャルを使用し AWS IoT Core の MQTT ブローカーに MQTTS で接続する。

from awscrt import auth
from awsiot import mqtt_connection_builder

# クレデンシャルプロバイダー生成
credentials_provider = auth.AwsCredentialsProvider.new_static(
    access_key_id=access_key,
    secret_access_key=secret_key,
    session_token=session_token
)

# 接続先エンドポイント、認証情報等をセットアップした通信用オブジェクトを生成
mqtt_connection = mqtt_connection_builder.websockets_with_default_aws_signing(
    endpoint=IOT_ENDPOINT,
    region=REGION,
    credentials_provider=credentials_provider,
    client_id=identity_id, # Identity ID をクライアントIDとして使用
    clean_session=False,
    keep_alive_secs=30
)

# 接続開始
connect_future = mqtt_connection.connect()
# 接続確立を待機
connect_future.result()

5. MQTT ブローカーとの接続確立後、特定のトピックに対してメッセージ送信(Publish)を行う。

from awscrt import mqtt

# 送信メッセージの設定
PAYLOAD = {
    "state": {
        "desired": {
            "power": "OFF",
        }
    }
}

message_json = json.dumps(PAYLOAD)
# Publish 実行
publish_future, packet_id = mqtt_connection.publish(
    topic=TOPIC,
    payload=message_json,
    qos=mqtt.QoS.AT_LEAST_ONCE
)
# Publish の完了を待機
publish_future.result()

最後に

IoT Core への接続時にエラーとなり解決までに時間がかかった。
アプリケーション内のライブラリが出力するエラーメッセージからは何が原因なのかが分からず、
ロググループ AWSIotLogsV2 から認証関連のエラー(AUTHORIZATION_FAILURE)を確認し、「事前作業」で設定したポリシーやクレデンシャル発行時の指定ロール割り当て設定を見直すことで解決できた。