云平台设备接入实战：三元组鉴权、属性上报与命令下发（阿里云/华为云）

上一篇我们把 MQTT 跑通了：电脑上开个 mosquitto，ESP32 连上去发个温度，串口打印 connected。
但那只是"我自己的 broker"。要让手机 App 能看数据、能点按钮控灯，设备得接到公有云物联网平台上去。
我第一次做的时候以为"不就是把 broker 地址换一下吗"，结果连了三天没连上——卡住我的根本不是 MQTT 协议，是"鉴权参数怎么算"和"Topic 怎么拼"。

适用人群：会用 ESP32/STM32 发 MQTT（知道 publish/subscribe、QoS、keepalive），想把设备接到阿里云或华为云的同学。

读完你能得到：三元组是什么 → 两家平台的 ClientID/Username/Password 手算（附 Python）→ Topic 与属性上报 JSON → 命令下发的同步/异步 → ESP32 代码骨架 → 踩坑表 + 5 步动手练。

一、为什么"换 broker 地址"连不上

自建 broker 时你可以不加用户名密码，因为它不认识你也不在乎你是谁。公有云挂着千万台设备，必须在第一个字节之前确认你是谁，于是多了三件事：

[图 1] 设备 → 云 → App 链路上，每层各管什么

┌────────┐ ┌────────────────────────────┐ ┌────────┐
│ 设备 │ MQTT │ 物联网平台 │ API │ 手机App │
│ ESP32 │ over TLS │ ① 接入鉴权（三元组签名） │ <─────> │ 看数据 │
│ 上报温度│ <═══════> │ ② Topic 路由 + 权限 │ │ 下发指令│
│ 执行命令│ │ ③ 物模型解析（Alink/产品模型）│ └────────┘
└────────┘ │ ④ 设备影子（离线缓存指令） │
└────────────────────────────┘
本文讲 ①②③，第 ④ 层见下一篇《设备影子与物模型》

前置概念表

⚠️ 华为云叫法不同但本质一样：用设备 ID（device_id）定位设备、设备密钥（secret）签名。别被名词绕晕。

二、三元组就是"账号 + 密码"

阿里云创建产品（如"温控节点"）+ 设备（如 esp32_01）后给三个值：

ProductKey : a1LhUsK**** ← 产品级 ID，同批产品共用
DeviceName : esp32_01 ← 产品内唯一
DeviceSecret : bdd043d193782d11**** ← 设备密钥，每台不同

ProductKey + DeviceName = 账号，DeviceSecret = 密码。 但你不能直接把密码传过去，所以要用 HMAC 做签名：证明"我知道密码，但我不告诉你"。

🔴 DeviceSecret 在控制台只明文出现一次。我当年截图存了，换电脑时丢了，只能重置密钥，之前烧好的板子全部连不上。第一件事：把三元组扔进密码管理器。

三、连接参数怎么算（两家都给 Python）

3.1 阿里云（一机一密，hmacsha256）

mqttClientId : {clientId}|securemode=2,signmethod=hmacsha256,timestamp={毫秒时间戳}|
mqttUsername : {DeviceName}&{ProductKey}
content : clientId{clientId}deviceName{DeviceName}productKey{ProductKey}timestamp{毫秒时间戳}
mqttPassword : HMAC-SHA256(key = DeviceSecret, msg = content) → 十六进制字符串

四个必须注意的点：

# ali_sign.py —— 阿里云物联网平台一机一密连接参数
import hmac, hashlib, time

def ali_sign(product_key, device_name, device_secret, client_id=None, ts_ms=None):
if client_id is None:
client_id = product_key + "." + device_name # 官方默认 pk.dn
ts = str(ts_ms if ts_ms else int(round(time.time() * 1000)))
# 拼接顺序 = 参数名字典序
content = "clientId{}deviceName{}productKey{}timestamp{}".format(
client_id, device_name, product_key, ts)
sign = hmac.new(device_secret.encode("utf-8"),
content.encode("utf-8"), hashlib.sha256).hexdigest()
return {
"client_id": "{}|securemode=2,signmethod=hmacsha256,timestamp={}|".format(client_id, ts),
"username": "{}&{}".format(device_name, product_key),
"password": sign,
}

if __name__ == "__main__":
for k, v in ali_sign("a1LhUsK****", "esp32_01", "bdd043d193782d11****").items():
print("{}: {}".format(k, v))

连上后先只验上行（paho-mqtt 1.6.1；2.x 要写 mqtt.Client(mqtt.CallbackAPIVersion.VERSION1, client_id=…)）：

import json, ssl, time, paho.mqtt.client as mqtt
from ali_sign import ali_sign

PK, DN, DS = "a1LhUsK****", "esp32_01", "bdd043d193782d11****"
HOST = PK + ".iot-as-mqtt.cn-shanghai.aliyuncs.com" # 旧版公共实例
# 新版公共实例/企业版：HOST = "{instanceId}.mqtt.iothub.aliyuncs.com"
PUB = "/sys/{}/{}/thing/event/property/post".format(PK, DN)
SUB = "/sys/{}/{}/thing/event/property/post_reply".format(PK, DN)

info = ali_sign(PK, DN, DS)
c = mqtt.Client(client_id=info["client_id"], protocol=mqtt.MQTTv311)
c.username_pw_set(info["username"], info["password"])
c.tls_set(ca_certs="ali_iot_root.crt", tls_version=ssl.PROTOCOL_TLSv1_2) # 自签名证书 → 8883
c.on_connect = lambda cl, u, f, rc: (print("rc =", rc), cl.subscribe(SUB, 1) if rc == 0 else None)
c.on_message = lambda cl, u, m: print("reply:", m.payload.decode())
c.connect(HOST, 8883, keepalive=300)
c.loop_start()

c.publish(PUB, json.dumps({
"id": "1001", "version": "1.0", "sys": {"ack": 1},
"params": {"temperature": 26.5}, # 属性名必须和物模型标识符一致
"method": "thing.event.property.post"}), qos=1)
time.sleep(3); c.loop_stop()

3.2 华为云（密钥鉴权）

clientId : {deviceId}_{标识类型}_{密码签名类型}_{时间戳}
│ │ │ └ UTC，格式 yyyyMMddHH
│ │ └ 0=不校验时间戳；1=HMACSHA256 校验时间戳
│ └ 一机一密直连设备填 0
└ 注册设备后平台给的设备 ID
username : {deviceId}
password : HMAC-SHA256(key = 时间戳字符串, msg = deviceSecret) → 十六进制（小写）

和阿里云两个反直觉差异：

接入地址在控制台"总览 → 接入信息"，形如 xxx.st1.iotda-device.cn-north-4.myhuaweicloud.com，端口 8883（MQTTS）/ 1883（明文）。

# hw_sign.py —— 华为云 IoTDA 密钥鉴权
import hmac, hashlib
from datetime import datetime

def hw_sign(device_id, device_secret, check_stamp=1):
ts = datetime.utcnow().strftime("%Y%m%d%H") # UTC，不是北京时间！
pwd = hmac.new(ts.encode("utf-8"), # key = 时间戳
device_secret.encode("utf-8"), # msg = 设备密钥
hashlib.sha256).hexdigest()
return {"client_id": "{}_0_{}_{}".format(device_id, check_stamp, ts),
"username": device_id, "password": pwd}

参考：阿里云《MQTT-TLS连接通信》《MQTT连接签名示例》，华为云《设备连接鉴权》。华为云的拼接与 HmacSHA256(时间戳为密钥、secret 为内容) 可对照官方 SDK 源码 MqttConnection.java / IotUtil.java 验证。

四、Topic 与 payload

阿里云物模型 Topic 都以 /sys/{pk}/{dn}/ 开头，记四组：

{
"id": "1001", "version": "1.0", "sys": {"ack": 1},
"params": {"temperature": 26.5, "LightSwitch": 1},
"method": "thing.event.property.post"
}

成功回 {"code":200,"data":{},"id":"1001","message":"success",…}。

⚠️ code 不是 200 就是失败：常见的有物模型校验失败（属性没定义/类型不对）、9201 设备离线。只 publish 不订阅 _reply，你根本不知道数据被打回来了。

华为云统一 $oc/devices/{device_id}/sys/…，按服务分组上报：

{"services":[{"service_id":"Temperature",
"properties":{"temperature":26.5},
"event_time":"20260101T120000Z"}]}

event_time 格式 yyyyMMddTHHmmssZ（UTC），可选。

五、命令下发：异步 vs 同步

[图 2] 两种下行方式

异步（属性设置 / 服务调用）
App ──API──> 平台 ──publish──> 设备执行 ──_reply──> 平台
App <──"下发成功"── （只代表平台发出去了，不代表设备执行了）

同步 RRPC（阿里云）
App ──RRpc API（阻塞）──> 平台 ── /sys/{pk}/{dn}/rrpc/request/{msgId} ──> 设备
设备 publish 到 …/rrpc/response/{msgId}
App <── 响应 或 8 秒超时 ──

⚠️ 华为云命令下发也走 sys/commands，设备必须把 request_id 原样拼进响应 Topic 平台才能对上号。下发时可设命令有效期，超时未响应会失效——参数名和默认值以你用的平台版本文档为准，别照抄博客。

六、ESP32 代码骨架（ESP-IDF v5.x）

顺序不能反：先校时 → 再算签名 → 最后连 MQTT。

/* aliyun_iot.c —— 适用：ESP32 / ESP-IDF v5.x（esp-mqtt + mbedtls） */
#include <string.h>
#include <stdio.h>
#include "esp_log.h"
#include "esp_timer.h"
#include "esp_mac.h"
#include "mqtt_client.h"
#include "mbedtls/md.h"

static const char *TAG = "aliyun";
#define PK "a1LhUsK****"
#define DN "esp32_01"
#define DS "bdd043d193782d11****"
#define HOST "mqtts://" PK ".iot-as-mqtt.cn-shanghai.aliyuncs.com"

/* 阿里云自签名根证书 PEM：放 main 目录，用 CMakeLists 的 EMBED_FILES 嵌入
也可直接把 PEM 内容贴成字符串常量 */
extern const char root_ca_pem_start[] asm("_binary_aliyun_root_ca_pem_start");
extern const char root_ca_pem_end[] asm("_binary_aliyun_root_ca_pem_end");

/* HMAC-SHA256 → 小写十六进制（32 字节 = 64 字符 + '\\0'） */
static int hmac_sha256_hex(const char *key, const char *msg, char *out, size_t out_len)
{
unsigned char mac[32];
const mbedtls_md_info_t *info = mbedtls_md_info_from_type(MBEDTLS_MD_SHA256);
if (info == NULL || out_len < 65) return –1;
if (mbedtls_md_hmac(info, (const unsigned char *)key, strlen(key),
(const unsigned char *)msg, strlen(msg), mac) != 0) return –1;
for (int i = 0; i < 32; i++) snprintf(&out[i * 2], 3, "%02x", mac[i]);
out[64] = '\\0';
return 0;
}

static int build_conn_params(char *cid, size_t cid_len, char *user,
size_t user_len, char *pass, size_t pass_len)
{
uint8_t mac[6];
char client_id[16], content[192];
esp_read_mac(mac, ESP_MAC_WIFI_STA); /* MAC 当 clientId，天然唯一 */
snprintf(client_id, sizeof(client_id), "%02x%02x%02x%02x%02x%02x",
mac[0], mac[1], mac[2], mac[3], mac[4], mac[5]);

unsigned long long ts = (unsigned long long)(esp_timer_get_time() / 1000ULL); /* ms */
int n = snprintf(content, sizeof(content),
"clientId%sdeviceName%sproductKey%stimestamp%llu",
client_id, DN, PK, ts);
if (n < 0 || n >= (int)sizeof(content)) return –1; /* 截断了就别发出去 */
if (hmac_sha256_hex(DS, content, pass, pass_len) != 0) return –1;

snprintf(cid, cid_len, "%s|securemode=2,signmethod=hmacsha256,timestamp=%llu|",
client_id, ts);
snprintf(user, user_len, "%s&%s", DN, PK);
return 0;
}

void app_main(void)
{
/* ① 连 Wi-Fi（略） ② SNTP 校时，等 sntp_get_sync_status()==SNTP_SYNC_STATUS_COMPLETED
③ 才调用 build_conn_params —— 没校时时时间戳从 0 起算，平台必然判签名非法 */

char cid[96], user[64], pass[80];
if (build_conn_params(cid, sizeof(cid), user, sizeof(user), pass, sizeof(pass)) != 0) {
ESP_LOGE(TAG, "签名失败"); return;
}
ESP_LOGI(TAG, "cid=%s user=%s", cid, user); /* 调试期打出来，和 Python 端对账 */

esp_mqtt_client_config_t cfg = {
.broker.address.uri = HOST ":8883",
.broker.verification.certificate = root_ca_pem_start, /* 必须校验证书 */
.credentials.client_id = cid,
.credentials.username = user,
.credentials.authentication.password = pass,
.session.keepalive = 300, /* 官方建议 ≥300 s */
};
esp_mqtt_client_handle_t client = esp_mqtt_client_init(&cfg);
if (client == NULL) { ESP_LOGE(TAG, "mqtt init 失败"); return; }
esp_mqtt_client_register_event(client, ESP_EVENT_ANY_ID, mqtt_event_handler, NULL);
esp_mqtt_client_start(client);
}

事件回调（订阅、收数据、看错误码）：

static void mqtt_event_handler(void *arg, esp_event_base_t base,
int32_t event_id, void *event_data)
{
esp_mqtt_event_handle_t evt = event_data;
switch ((esp_mqtt_event_id_t)event_id) {
case MQTT_EVENT_CONNECTED:
esp_mqtt_client_subscribe(evt->client, "/sys/" PK "/" DN "/thing/service/property/set", 1);
esp_mqtt_client_subscribe(evt->client, "/sys/" PK "/" DN "/thing/event/property/post_reply", 1);
break;
case MQTT_EVENT_DATA: {
char topic[96], data[256];
size_t tl = evt->topic_len < sizeof(topic) – 1 ? evt->topic_len : sizeof(topic) – 1;
size_t dl = evt->data_len < sizeof(data) – 1 ? evt->data_len : sizeof(data) – 1;
memcpy(topic, evt->topic, tl); topic[tl] = '\\0';
memcpy(data, evt->data, dl); data[dl] = '\\0';
ESP_LOGI(TAG, "TOPIC=%s DATA=%s", topic, data);
/* TODO：区分 set / post_reply，校验 code，执行动作，回 set_reply（id 要一致） */
break;
}
case MQTT_EVENT_ERROR:
ESP_LOGE(TAG, "MQTT 错误，看 error_type / connect_return_code");
break;
default: break;
}
}

💡 connect_return_code 4 = 用户名密码错误（签名错了）、5 = 未授权。看到 4 就别怀疑网络了，九成是签名算错。
⚠️ 证书和端口要配套：阿里云自签名根证书对应 8883；老版 Global Sign 根证书对应 1883/443，且 2028-01-28 到期。新项目直接用自签名证书 + 8883，别关证书校验。

七、新手必踩的坑清单

八、动手练一练

练习 1：把"三件套"凑齐（30 分钟）。 阿里云免费公共实例（或华为云 IoTDA 免费单元）→ 建产品 → 加一个 temperature 属性（float，读写）→ 建设备 → 复制三元组。先在设备详情页点"MQTT 连接参数"，把平台生成好的参数填进 MQTTX，看到"在线"再往下走。

意义：先用平台给的正确答案验证链路是通的，后面自己算签名出错时，就知道问题一定在签名上。

练习 2：Python 手算签名跑通上行。 验收三条：平台日志服务出现"属性上报"、设备收到 post_reply 且 code==200、控制台物模型数据页看到 26.5。

练习 3（故意试错）：四连击。 ① DeviceSecret 改一位 → 看 rc（应为 4）；② timestamp 改成 10 分钟前 → 看是否被拒；③ 26.5 改成字符串 → 看 code 变什么；④ Topic 里 pk/dn 对调 → 看是否"发送成功"但平台毫无反应。把四种现象记笔记，以后线上出问题你能一眼归类。

练习 4：打通下行。 控制台"在线调试"下发属性设置 → 设备收到 property/set → 打印 JSON → 回 set_reply（id 与收到的相同、code:200）→ 再上报一次属性（不回写，控制台看到的还是旧值，这步最容易漏）。

练习 5：移植 ESP32，双端对账。 让 ESP32 把算出的 client_id/username/password 全打到串口，Python 端用同一套 pk/dn/secret/timestamp 再算一遍逐字符比对：一致却连不上 → 问题在 TLS/网络；不一致 → 一定是拼接或编码的锅（我踩的三次全在这里）。

小结

• 接公有云 ≠ 换 broker 地址，工作量在三件事：算签名、拼 Topic、对格式。
• 阿里云：clientId|securemode=2,signmethod=hmacsha256,timestamp=…| + deviceName&productKey + HMAC-SHA256(DeviceSecret, 字典序拼接串)。
• 华为云：{deviceId}_0_{校验位}_{yyyyMMddHH} + deviceId + HMAC-SHA256(时间戳为密钥, 设备密钥为内容)——两家 key/data 是反的。
• 阿里云上报按属性平铺（params），华为云按服务分组（services），别互相复制。
• 只 publish 不订阅 _reply、不看 code = 闭着眼睛发数据。
• 命令下发：异步走服务/属性设置，要结果走 RRPC（8 s 超时），怕丢走设备影子。
• 先校时、再签名、再连接，顺序反了会在签名上浪费一整天（我浪费过）。

文中签名算法、Topic 格式、错误码与端口/证书要求均对照阿里云物联网平台与华为云 IoTDA 官方文档整理（MQTT-TLS连接通信、设备连接鉴权、Topic定义、RRPC），平台改版时以官方文档为准。

下一篇：《设备影子与物模型：解决"设备离线也能控"的核心机制》