如何使用Python Paho结合SigV4认证在AWS IoT Core发布MQTT消息

可行性说明 #

完全可以通过Python Paho MQTT库，结合Cognito身份池获取的SigV4临时凭证（Access Key ID、Secret Key、Session Token）实现AWS IoT Core的MQTT消息发布。这种方式依赖AWS IoT的自定义认证机制，通过SigV4签名生成MQTT连接所需的认证参数。

实现步骤 #

1. 获取Cognito身份池的SigV4凭证 #

通过boto3调用Cognito Identity API获取临时凭证：

import boto3

def get_cognito_credentials(identity_pool_id):
    cognito_identity = boto3.client('cognito-identity')
    # 获取身份ID
    id_resp = cognito_identity.get_id(IdentityPoolId=identity_pool_id)
    identity_id = id_resp['IdentityId']
    # 获取对应凭证
    cred_resp = cognito_identity.get_credentials_for_identity(IdentityId=identity_id)
    credentials = cred_resp['Credentials']
    return {
        'access_key': credentials['AccessKeyId'],
        'secret_key': credentials['SecretKey'],
        'session_token': credentials['SessionToken'],
        'expiration': credentials['Expiration']
    }

注意：需提前配置AWS SDK的基础访问权限（如本地AWS凭证、IAM角色），确保能正常调用Cognito API。

2. 生成SigV4签名作为MQTT认证密码 #

AWS IoT要求用SigV4签名生成MQTT连接的密码，用户名可留空或填入自定义标识（如身份ID）。这里利用AWS CRT库简化签名过程：

from awscrt.auth import AwsCredentials, AwsSignatureV4SigningConfig, AwsSigner
from awscrt.http import HttpRequest
import urllib.parse

def generate_mqtt_sigv4_password(credentials, iot_endpoint, region):
    # 构建用于SigV4签名的HTTP请求上下文
    request = HttpRequest(
        method="GET",
        uri="/mqtt",
        headers={"host": iot_endpoint}
    )
    # 配置SigV4参数
    signing_config = AwsSignatureV4SigningConfig(
        credentials=AwsCredentials(
            access_key_id=credentials['access_key'],
            secret_access_key=credentials['secret_key'],
            session_token=credentials['session_token']
        ),
        region=region,
        service_name="iotdevicegateway"
    )
    # 执行签名
    signer = AwsSigner()
    signed_request = signer.sign_http_request(request, signing_config)
    # 从签名结果中提取并构建MQTT密码
    auth_header = signed_request.headers.get('Authorization', '')
    auth_parts = auth_header.split(',')
    signature_part = next(p for p in auth_parts if 'Signature=' in p)
    password_components = [
        auth_header.split(' ')[1],
        signature_part,
        f"X-Amz-Security-Token={urllib.parse.quote(credentials['session_token'])}"
    ]
    return '&'.join(password_components)

3. 配置Paho客户端并发布消息 #

结合以上步骤获取的凭证和签名密码，配置Paho客户端连接AWS IoT Core并发布消息：

import paho.mqtt.client as mqtt

def connect_and_publish(credentials, iot_endpoint, region, topic, message):
    # 生成SigV4认证密码
    mqtt_password = generate_mqtt_sigv4_password(credentials, iot_endpoint, region)
    # 初始化Paho客户端
    client = mqtt.Client(client_id="your-custom-client-id")  # 替换为自定义客户端ID
    client.username_pw_set(username="", password=mqtt_password)
    client.tls_set()  # AWS IoT Core强制要求TLS连接
    
    # 连接回调处理
    def on_connect(client, userdata, flags, rc):
        if rc == 0:
            print("成功连接AWS IoT Core")
            client.publish(topic, message, qos=1)
            print(f"已发布消息到主题 {topic}: {message}")
            client.disconnect()
        else:
            print(f"连接失败，错误码: {rc}")
    
    client.on_connect = on_connect
    # 连接到IoT Core ATS终端节点（端口8883为标准MQTT TLS端口）
    client.connect(iot_endpoint, 8883, keepalive=60)
    # 启动消息循环
    client.loop_forever()

# 主执行逻辑
if __name__ == "__main__":
    # 替换为你的AWS资源信息
    IDENTITY_POOL_ID = "us-east-1:xxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
    IOT_ENDPOINT = "your-iot-endpoint-ats.iot.us-east-1.amazonaws.com"
    REGION = "us-east-1"
    MQTT_TOPIC = "test/cognito-sigv4"
    MESSAGE = "Hello from Paho + Cognito SigV4!"
    
    # 获取Cognito凭证
    cognito_creds = get_cognito_credentials(IDENTITY_POOL_ID)
    # 连接并发布消息
    connect_and_publish(cognito_creds, IOT_ENDPOINT, REGION, MQTT_TOPIC, MESSAGE)

关键注意事项 #

• IAM权限配置：确保Cognito身份池关联的IAM角色拥有iot:Connect、iot:Publish等必要权限，否则会出现认证或操作失败。
• 凭证有效期：SigV4临时凭证默认有效期1小时，需在过期前重新获取并生成新的签名密码。
• 终端节点选择：必须使用AWS IoT Core的ATS终端节点（带-ats后缀），这是官方推荐的安全终端节点。
• TLS要求：Paho客户端必须启用TLS连接（client.tls_set()），AWS IoT Core不支持非加密的MQTT连接。

内容的提问来源于stack exchange，提问作者greenbee