[AWS IoT Core] カスタム認証（ユーザー名・パスワード）を使用してMQTTでPublish/Subscribeしてみました

{
    "protocolData": {
        "tls": {
            "serverName": "xxxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com"
        },
        "mqtt": {
            "username": "user01?x-amz-customauthorizer-name=custom_auth_mqtt_2021_08_21",
            "password": "cGFzczAx",
            "clientId": "client_id"
        }
    },
    "protocols": [
        "tls",
        "mqtt"
    ],
    "signatureVerified": False,
    "connectionMetadata": {
        "id": "54300d4a-d787-bb33-bbab-e61eb9f07893"
    }
}

import json
import base64

policy = {
  "Version": "2012-10-17",
  "Statement": [
    {
        "Effect": "Allow",
        "Action": "iot:Connect",
        "Resource": "arn:aws:iot:ap-northeast-1:*:client/client_id"
    },
    {
        "Effect": "Allow",
        "Action": [
            "iot:Publish",
            "iot:Receive",
            ],
        "Resource": "arn:aws:iot:ap-northeast-1:*:topic/my-topic"
    },
    {
        "Effect": "Allow",
        "Action": "iot:Subscribe",
        "Resource": "arn:aws:iot:ap-northeast-1:*:topicfilter/my-topic"
    }
  ]
}

def lambda_handler(event, context):
    print(event)

    is_authenticated = False
    username = ""
    password = ""

    if "protocolData" in event:
        protocolData = event["protocolData"]
        if "mqtt" in protocolData:
            mqtt = protocolData["mqtt"]
            if "username" in mqtt:
                tmp = mqtt["username"].split("?")
                username = tmp[0]
            if "password" in mqtt:
                password = mqtt["password"]
                password = base64.b64decode(password).decode() #Base64デコード

    if(username == "user01" and password == "pass01"):
        is_authenticated = True

    principal_id = "CustomAuthId"
    disconnect = 84000
    refresh = 300
    
    return {
        "isAuthenticated": is_authenticated,
        "principalId": principal_id,
        "disconnectAfterInSeconds": disconnect,
        "refreshAfterInSeconds": refresh,
        "policyDocuments": [
            policy
        ]
    }

from awscrt import io, mqtt
import json
import time

endpoint = "xxxxxxxxxxxx.iot.ap-northeast-1.amazonaws.com"
root_ca = "./certs/root-CA.crt"
authorizer_name = "custom_auth_mqtt_2021_08_21"
topic = "my-topic"

client_id = "client_id"
username = "user01"
password = "pass01"

def on_message_received(topic, payload, **kwargs):
    print("received: {}".format(payload.decode()))

if __name__ == '__main__':
    event_loop_group = io.EventLoopGroup(1)
    host_resolver = io.DefaultHostResolver(event_loop_group)
    client_bootstrap = io.ClientBootstrap(event_loop_group, host_resolver)

    tls_options = io.TlsContextOptions()
    tls_options.alpn_list = ['mqtt']
    tls_options.override_default_trust_store_from_path(ca_dirpath=None, ca_filepath=root_ca)
    tls_ctx = io.ClientTlsContext(options=tls_options)
    client = mqtt.Client(client_bootstrap, tls_ctx)

    username += "?x-amz-customauthorizer-name={}".format(authorizer_name)

    mqtt_connection = mqtt.Connection(client=client,
        host_name=endpoint,
        port=443,
        client_id=client_id,
        clean_session=True,
        keep_alive_secs=6,
        username=username,
        password=password)

    connect_future = mqtt_connection.connect()
    connect_future.result()
    print("Connected!")

    # Subscribe
    subscribe_future, packet_id = mqtt_connection.subscribe(
        topic=topic,
        qos=mqtt.QoS.AT_LEAST_ONCE,
        callback=on_message_received)

    subscribe_future.result()

    for i in range(3):
        # Publish
        payload = {
            "counter": i
        }
        mqtt_connection.publish(
            topic=topic,
            payload=json.dumps(payload),
            qos=mqtt.QoS.AT_LEAST_ONCE)
        time.sleep(1)

    disconnect_future = mqtt_connection.disconnect()
    disconnect_future.result()

username?x-amz-customauthorizer-name=${name}
&x-amz-customauthorizer-signature=${sign}&token-name=${token-value}