使用Cognito身份池临时凭证连接AWS IoT Core MQTT代理

一、确认IoT Core策略配置（关键） #

必须给Cognito身份池的未认证/认证角色绑定具备正确权限的IoT策略，不能仅配置事物相关权限，要覆盖代理连接的核心动作：

• 策略需包含iot:Connect动作，资源指定你将使用的客户端ID（测试阶段可先用*，生产环境建议绑定Cognito身份ID以提升安全性）
• 按需添加iot:Publish/iot:Subscribe/iot:Receive等动作，资源对应你要操作的主题
  示例策略（替换你的AWS账号ID和区域）：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["iot:Connect"],
      "Resource": "arn:aws:iot:us-east-1:123456789012:client/${cognito-identity.amazonaws.com:sub}"
    },
    {
      "Effect": "Allow",
      "Action": ["iot:Publish", "iot:Subscribe", "iot:Receive"],
      "Resource": "arn:aws:iot:us-east-1:123456789012:topic/*"
    }
  ]
}

注：${cognito-identity.amazonaws.com:sub}会自动映射到用户的Cognito身份ID，避免客户端ID滥用。

二、验证Cognito角色与IoT策略的关联 #

你通过CLI绑定策略后，需确认绑定是否生效：

1. 查看身份池对应的角色ARN：

aws cognito-identity describe-identity-pool --identity-pool-id "你的身份池ID"

2. 检查该角色是否附加了目标IoT策略：

aws iot list-principal-policies --principal "角色ARN"

若未找到目标策略，重新执行绑定命令：

aws iot attach-principal-policy --policy-name "你的IoT策略名" --principal "角色ARN"

三、Python代码实现（推荐使用awscrt） #

awscrt是AWS官方主推的SDK，对临时凭证支持更稳定，替代旧版AWSIoTPythonSDK。以下是直接用Cognito临时凭证连接的完整代码：

import awscrt.mqtt
import boto3
from awscrt.auth import AwsCredentialsProvider

# 1. 获取Cognito临时凭证（已有凭证可跳过此步骤）
cognito_client = boto3.client('cognito-identity', region_name='us-east-1')
identity_resp = cognito_client.get_id(IdentityPoolId='你的身份池ID')
credentials_resp = cognito_client.get_credentials_for_identity(IdentityId=identity_resp['IdentityId'])
temp_creds = credentials_resp['Credentials']

# 2. 配置MQTT连接参数
iot_endpoint = "你的IoT Core端点（格式：xxxxxx-ats.iot.us-east-1.amazonaws.com）"
client_id = identity_resp['IdentityId']  # 用Cognito身份ID作为客户端ID，避免冲突

# 基于临时凭证创建认证提供者
cred_provider = AwsCredentialsProvider.new_static(
    access_key_id=temp_creds['AccessKeyId'],
    secret_access_key=temp_creds['SecretKey'],
    session_token=temp_creds['SessionToken']
)

# 初始化MQTT连接
mqtt_conn = awscrt.mqtt.Connection(
    client_id=client_id,
    host_name=iot_endpoint,
    port=8883,
    credentials_provider=cred_provider,
    clean_session=True,
    keep_alive_secs=30
)

# 3. 连接状态回调
def on_conn_interrupted(conn, error, **kwargs):
    print(f"连接中断: {error}")

def on_conn_resumed(conn, return_code, session_present, **kwargs):
    print(f"连接恢复，返回码: {return_code}，会话存在: {session_present}")

mqtt_conn.on_connection_interrupted = on_conn_interrupted
mqtt_conn.on_connection_resumed = on_conn_resumed

# 4. 发起连接并验证
print("正在连接IoT Core代理...")
connect_future = mqtt_conn.connect()
connect_future.result()  # 等待连接完成，超时会抛出异常
print("连接成功！")

# 示例：订阅主题
def on_msg_received(topic, payload, **kwargs):
    print(f"收到消息 | 主题: {topic} | 内容: {payload.decode('utf-8')}")

subscribe_future, _ = mqtt_conn.subscribe(
    topic="test/topic",
    qos=awscrt.mqtt.QoS.AT_LEAST_ONCE,
    callback=on_msg_received
)
subscribe_future.result()
print("订阅主题成功")

# 示例：发布消息
publish_future, _ = mqtt_conn.publish(
    topic="test/topic",
    payload="来自Cognito临时凭证的消息",
    qos=awscrt.mqtt.QoS.AT_LEAST_ONCE
)
publish_future.result()
print("消息发布成功")

# 维持连接，按回车断开
input("按Enter键断开连接...\n")
mqtt_conn.disconnect()

四、超时问题排查要点 #

1. 网络连通性：确认运行代码的环境能访问IoT Core端点的8883端口，VPC内需配置IoT VPC端点或允许出站流量到AWS IoT服务
2. 凭证有效性：Cognito临时凭证有效期为1小时，确保使用的凭证未过期
3. 客户端ID冲突：同一客户端ID被多个设备占用会导致连接失败，建议用Cognito身份ID作为唯一客户端ID
4. 策略资源匹配：iot:Connect的资源需与实际使用的客户端ID完全匹配，若策略中用了${cognito-identity.amazonaws.com:sub}，则客户端ID必须为Cognito身份ID

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