使用 Python 通过 WebSocket 连接 AWS IoT 超时，如何排查？

2026-11-04数据小许

结论

这段代码需要补上根 CA 配置：myMQTTClient.configureCredentials(rootCAPath)。使用 AWSIoTPythonSDK 时，即使通过 WebSocket 采用 SigV4 认证，也需要配置根 CA 来验证服务器证书。已有 IAM 权限和 Domain Configuration，客户端连接配置仍可能不完整。AWS SDK 官方说明

仅凭 connect() 超时还不能确定原因。补齐 CA 后，也要确认 Python 使用的 SSO profile、实际 endpoint 和连接权限。

1. 配置根 CA，并核对 endpoint

SigV4 负责请求认证，TLS 负责加密连接和验证服务器身份，两者都需要正确配置。在 WebSocket 模式下，这个 SDK 的 configureCredentials() 只需传入根 CA 路径，无需传入设备私钥和客户端证书。

如果先用标准 ATS endpoint 排查，可以通过 AWS 官方文档中的链接下载 Amazon Root CA 1，保存为 AmazonRootCA1.pem。CA 必须与实际服务器证书链匹配；如果使用自定义域名和服务器证书，应根据信任的证书链配置 CA。AWS 服务器认证说明

使用目标账户和区域查询标准数据 endpoint：

aws iot describe-endpoint --endpoint-type iot:Data-ATS --region eu-central-1 --profile my-sso-profile --query endpointAddress --output text

将 my-sso-profile 替换为实际的 profile 名称。查询需要 iot:DescribeEndpoint 权限；如果没有，也可以从 AWS IoT Core 控制台设置中取得 endpoint。

configureEndpoint() 接收主机名，不要添加 wss:// 或 /mqtt。标准 endpoint 默认支持在端口 443 上使用 SigV4 MQTT over WebSocket。可以先用它验证基本连接，再排查已有 Domain Configuration 对应的域名。AWS 协议与端口说明

2. 明确使用已登录的 SSO profile

执行 aws sso login 后，还需确认 Python 选中了同一个 profile。原代码中的 boto3.Session() 没有显式指定 profile，无法确定它使用的是预期的 SSO 身份。

可以先执行：

aws sso login --profile my-sso-profile
aws sts get-caller-identity --profile my-sso-profile

检查返回的账户和角色是否符合预期，并在 Python 中显式传入 profile_name。Boto3 官方支持通过该参数选择 SSO profile。Boto3 凭证配置说明

SSO 获取的是临时 AWS 凭证，需要同时传入 access key、secret key 和 session token。原代码已经传入 creds.token，应予保留。

3. 修改连接代码

以下示例沿用原 SDK 和连接参数。运行前替换 profile、endpoint 和 CA 文件路径：

import logging
from pathlib import Path

import boto3
from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient

logging.basicConfig(level=logging.INFO)
logging.getLogger('AWSIoTPythonSDK').setLevel(logging.DEBUG)

profile = 'my-sso-profile'
endpoint = '<your-data-ats-endpoint>'
root_ca = Path('AmazonRootCA1.pem').resolve()

if not root_ca.is_file():
    raise FileNotFoundError(f'Root CA file not found: {root_ca}')

session = boto3.Session(
    profile_name=profile,
    region_name='eu-central-1',
)

credentials = session.get_credentials()
if credentials is None:
    raise RuntimeError('No AWS credentials found for the selected profile')

creds = credentials.get_frozen_credentials()

client = AWSIoTMQTTClient('myClientID', useWebsocket=True)
client.configureEndpoint(endpoint, 443)
client.configureCredentials(str(root_ca))
client.configureIAMCredentials(
    creds.access_key,
    creds.secret_key,
    creds.token,
)
client.configureOfflinePublishQueueing(-1)
client.configureDrainingFrequency(2)
client.configureConnectDisconnectTimeout(10)
client.configureMQTTOperationTimeout(5)

client.connect()
try:
    client.publish('myTopic', 'myPayload', 0)
finally:
    client.disconnect()

示例补充了 configureCredentials()，并明确指定 SSO profile。不要保留原代码中的 print(creds)，它会把凭证内容写入终端或日志。

4. 仍然超时时，按日志定位

先看 SDK 日志中最早出现的连接错误，再检查以下项目：

网络和 TLS：检查 endpoint 的 DNS 解析、本机到端口 443 的连接，以及代理或防火墙是否允许 WebSocket Upgrade。出现 TLS 证书错误时，优先检查 CA 文件、服务器证书链和本机时间。
身份和签名：确认程序使用的账户、角色及 session token 正确，临时凭证尚未过期，本机时间准确。
连接和发布权限：核对实际 SSO 角色的有效权限。当前代码需要 iot:Connect 对应 client/myClientID，以及 iot:Publish 对应 topic/myTopic；资源 ARN 中的账户和区域也必须匹配。权限边界或组织策略中的显式拒绝仍可能阻止操作。AWS 发布与订阅权限示例
服务端信息：如果已启用 AWS IoT Core 日志，结合相同时间段的连接记录，判断是否发生认证或授权失败。仅凭客户端超时无法直接认定是权限问题。

离线发布队列和 draining frequency 用于消息处理，不会补齐 TLS 或认证配置。单纯增大连接超时也不能修复这些配置问题。

5. 长期运行时注意凭证刷新

示例通过 get_frozen_credentials() 获取当前凭证，再把字符串传给 MQTT 客户端。这次传入的值不会随着 Boto3 后续刷新而自动更新。对于长期运行或断线重连的程序，需要在凭证更新后重新配置 MQTT 客户端，确保后续握手使用有效凭证。