物联网之安规下的 MQTT Payload 加密实践

目录

背景

MQTT 传输安全模型：TLS 通道加密 vs Payload 应用层加密

两层加密的关系

什么场景必须做 Payload 加密

安规对 MQTT Payload 加密的具体要求

等保 2.0（GB/T 22239-2019）

ETSI EN 303 645（欧洲消费 IoT 安全标准）

安规对照总结

加密方案设计

算法选型

密钥管理方案（一机一密）

Payload 报文格式设计

AAD（Additional Authenticated Data）的使用

落地实现

设备端加密（C / mbedTLS）

云端解密（Go）

Nonce 去重：防重放攻击

完整 MQTT Publish 流程（设备端伪代码）

国密方案（SM4-GCM）

测试与验证

功能验证

安规审查验证清单

抓包验证示例

最佳实践与避坑

Nonce 管理（最易出错的环节）

性能优化

密钥存储安全

常见审计不通过原因

总结

参考内容

背景

在物联网（IoT）系统中，设备与云端之间的数据传输通常基于 MQTT 协议。很多团队上了 TLS 之后就认为"传输加密搞定了"，但实际过安规评审时会发现：TLS 只保护了设备到 Broker 这一段链路，Broker 本身能看到明文 payload。

这意味着：

• MQTT Broker 被攻破时，所有经过它的业务数据裸奔

• 多级 Broker 桥接场景下，中间节点可读取和篡改数据

• 云端消息队列、日志系统中的 payload 以明文存储

安规对此有明确要求。等保 2.0（GB/T 22239-2019）三级要求"采用密码技术保证通信过程中数据的保密性和完整性"；欧洲 ETSI EN 303 645 条款 5.5 要求"设备通信应使用加密协议，且敏感数据在传输中必须加密"。

本文聚焦一个具体问题：在 MQTT 传输场景下，如何对 payload 做应用层加密以满足安规要求。覆盖方案设计、报文格式、代码落地、测试验证和避坑指南。

MQTT 传输安全模型：TLS 通道加密 vs Payload 应用层加密

两层加密的关系

什么场景必须做 Payload 加密

• Broker 不可信：使用第三方托管 MQTT 服务（如公有云 MQTT Broker），不希望 Broker 看到业务数据

• 多级桥接：设备 → 边缘 Broker → 云端 Broker，中间链路安全无法完全保证

• 安规深度要求：等保三级密评要求"敏感数据全链路加密"，仅 TLS 不满足要求

• 合规审计：需要证明即使 Broker 日志泄露，攻击者也无法获取明文

💡 如果你的场景是设备直连自建 Broker，且 Broker 与云端在同一内网，TLS 通常已满足等保二级要求。但等保三级、密评、以及面向欧洲市场的产品，建议加上 Payload 层加密。

安规对 MQTT Payload 加密的具体要求

等保 2.0（GB/T 22239-2019）

等保三级"安全通信网络"章节对通信传输的要求：

⚠️ 等保三级及以上的密评要求：如果系统涉及政企场景，还需满足 GB/T 39786-2021《信息系统密码应用基本要求》，要求使用国密算法（SM2/SM3/SM4）保护数据传输的完整性和保密性。

ETSI EN 303 645（欧洲消费 IoT 安全标准）

EN 303 645 条款 5.5"安全通信"明确规定：

💡 EN 303 645 没有强制要求 payload 层加密，但条款 5.5-2 的"敏感数据在传输中必须加密"在 Broker 中转场景下，TLS 无法完全满足——因为 Broker 端 TLS 终结后数据是明文。审计时可能被判定不合规。

安规对照总结

加密方案设计

算法选型

💡 推荐AES-256-GCM：一次操作同时完成加密和认证（AEAD），输出密文 + 16 字节认证标签（Auth Tag），接收方验标签失败直接丢包，防止任何篡改。

密钥管理方案（一机一密）

每台设备出厂时注入一把唯一的 AES-256 密钥（32 字节），云端存储deviceId → key的映射关系。解密时通过 MQTT Topic 中的deviceId直接查找对应密钥。

产线注入流程： 1. 产线 HSM 为每台设备生成 32 字节随机密钥 2. 密钥写入设备安全存储（SE / eFuse / 加密 Flash） 3. 密钥 + deviceId 映射关系同步到云端密钥库 4. 设备出厂后，密钥不再变更

⚠️ 禁止：多台设备共享同一密钥、密钥硬编码在固件源码中、密钥以明文存储在普通 Flash。

Payload 报文格式设计

加密前（明文 payload）：

{ "id": "msg-001", "version": "1.0", "params": { "temperature": { "value": 25.6, "time": 1715760000000 }, "humidity": { "value": 62.3, "time": 1715760000000 } } }

加密后（密文 payload）：

{ "id": "msg-001", "version": "1.0", "params": { "enc": { "v": 1, "nonce": "dGhpcyBpcyBub25jZQ==", "ct": "base64编码的密文...", "tag": "base64编码的16字节认证标签..." } } }

💡 设计思路：保留外层id/version不加密，便于云端路由和消息去重；params内嵌enc对象承载加密数据，云端收到后先检查params.enc是否存在来判断是否需要解密。云端通过 Topic 中的deviceId直接查找对应设备的唯一密钥。

AAD（Additional Authenticated Data）的使用

AAD 是 GCM 的"不加密但参与认证"部分，可以防止消息被"移花接木"（把 A 设备的消息转发给 B 设备的 topic）：

AAD = MQTT Topic（完整路径） 示例：cwlink/device001/thing/property/report

这样即使攻击者把密文从cwlink/device001/thing/property/report搬到cwlink/device002/thing/property/report，解密时 AAD 不匹配，验证会失败。

落地实现

设备端加密（C / mbedTLS）

适用于有完整 OS 的嵌入式设备（如基于 Linux 的网关、ESP32 等）：

#include "mbedtls/gcm.h" #include "mbedtls/entropy.h" #include "mbedtls/ctr_drbg.h" #include <string.h> /* 加密 MQTT Payload 的 params 部分 * key: 32字节 AES-256 密钥 * plaintext: 原始 params JSON（如 {"temperature":{"value":25.6,"time":1715760000000}}） * pt_len: 明文长度 * topic: 完整 MQTT topic 路径（作为 AAD） * out_nonce: 输出 12 字节 nonce * out_ciphertext: 输出密文（长度 = pt_len） * out_tag: 输出 16 字节认证标签 */ int mqtt_payload_encrypt(const uint8_t *key, const uint8_t *plaintext, size_t pt_len, const char *topic, uint8_t *out_nonce, uint8_t *out_ciphertext, uint8_t *out_tag) { mbedtls_gcm_context gcm; mbedtls_entropy_context entropy; mbedtls_ctr_drbg_context ctr_drbg; int ret; /* 1. 初始化随机数生成器 */ mbedtls_entropy_init(&entropy); mbedtls_ctr_drbg_init(&ctr_drbg); ret = mbedtls_ctr_drbg_seed(&ctr_drbg, mbedtls_entropy_func, &entropy, NULL, 0); if (ret != 0) goto cleanup; /* 2. 生成 12 字节随机 nonce（每条消息必须唯一！） */ ret = mbedtls_ctr_drbg_random(&ctr_drbg, out_nonce, 12); if (ret != 0) goto cleanup; /* 3. AAD = 完整 MQTT Topic 路径 */ size_t aad_len = strlen(topic); /* 4. AES-256-GCM 加密 */ mbedtls_gcm_init(&gcm); ret = mbedtls_gcm_setkey(&gcm, MBEDTLS_CIPHER_ID_AES, key, 256); if (ret != 0) goto cleanup; ret = mbedtls_gcm_crypt_and_tag(&gcm, MBEDTLS_GCM_ENCRYPT, pt_len, out_nonce, 12, (uint8_t *)topic, aad_len, plaintext, out_ciphertext, 16, out_tag); cleanup: mbedtls_gcm_free(&gcm); mbedtls_ctr_drbg_free(&ctr_drbg); mbedtls_entropy_free(&entropy); return ret; }

云端解密（Go）

package mqtt import ( "crypto/aes" "crypto/cipher" "encoding/base64" "encoding/json" "errors" "fmt" ) // MQTTPayload 标准 MQTT 消息结构（与 Topic 设计规范一致） type MQTTPayload struct { ID string `json:"id"` Version string `json:"version"` Params json.RawMessage `json:"params"` } // EncryptedParams 加密后的 params 结构 type EncryptedParams struct { Enc EncBlock `json:"enc"` } // EncBlock 加密数据块 type EncBlock struct { V int `json:"v"` Nonce string`json:"nonce"` Ct string`json:"ct"` Tag string`json:"tag"` } // DecryptMQTTPayload 解密 MQTT 加密 payload // rawPayload: 收到的完整 MQTT payload JSON // key: 32 字节 AES-256 密钥（一机一密，通过 topic 中的 deviceId 查找） // topic: 消息的完整 MQTT Topic 路径（用于 AAD 验证） // 返回解密后的原始 params JSON func DecryptMQTTPayload(rawPayload []byte, key []byte, topic string) ([]byte, error) { // 1. 解析外层结构 var payload MQTTPayload if err := json.Unmarshal(rawPayload, &payload); err != nil { returnnil, fmt.Errorf("解析 payload 失败: %w", err) } // 2. 解析加密 params var encParams EncryptedParams if err := json.Unmarshal(payload.Params, &encParams); err != nil { returnnil, fmt.Errorf("解析 enc params 失败: %w", err) } // 3. Base64 解码各字段 nonce, err := base64.StdEncoding.DecodeString(encParams.Enc.Nonce) if err != nil { returnnil, fmt.Errorf("解码 nonce 失败: %w", err) } ciphertext, err := base64.StdEncoding.DecodeString(encParams.Enc.Ct) if err != nil { returnnil, fmt.Errorf("解码 ciphertext 失败: %w", err) } tag, err := base64.StdEncoding.DecodeString(encParams.Enc.Tag) if err != nil { returnnil, fmt.Errorf("解码 tag 失败: %w", err) } // 4. 校验 nonce 长度 iflen(nonce) != 12 { returnnil, errors.New("nonce 长度必须为 12 字节") } // 5. AES-256-GCM 解密 block, err := aes.NewCipher(key) if err != nil { returnnil, fmt.Errorf("创建 AES cipher 失败: %w", err) } gcm, err := cipher.NewGCM(block) if err != nil { returnnil, fmt.Errorf("创建 GCM 失败: %w", err) } // GCM Open 要求 ciphertext 和 tag 拼接 sealed := append(ciphertext, tag...) // AAD = 完整 MQTT Topic 路径 aad := []byte(topic) plaintext, err := gcm.Open(nil, nonce, sealed, aad) if err != nil { returnnil, fmt.Errorf("解密失败（认证标签验证不通过）: %w", err) } return plaintext, nil }

使用示例（在 IoT Rule 触发的 Lambda / 消息处理服务中）：

func handleMQTTMessage(topic string, payload []byte) { // 一机一密：从 topic 中提取 deviceId，查找该设备的唯一密钥 deviceId := extractDeviceId(topic) // 如 "cwlink/device001/..." → "device001" key := keyStore.GetKey(deviceId) // 32 字节 plainParams, err := DecryptMQTTPayload(payload, key, topic) if err != nil { log.Printf("解密失败, topic=%s, err=%v", topic, err) return } // plainParams 现在是原始的 params JSON： // {"temperature":{"value":25.6,"time":1715760000000},"humidity":{"value":62.3,"time":1715760000000}} log.Printf("解密成功: %s", string(plainParams)) }

Nonce 去重：防重放攻击

AES-GCM 的 nonce 每条消息唯一，天然适合做防重放的去重 key。攻击者录下一条合法加密消息并原封不动重发时，云端通过 nonce 去重即可识别并丢弃。

// NonceDedup 基于 Redis 的 nonce 去重服务 type NonceDedup struct { rdb *redis.Client ttl time.Duration // nonce 缓存过期时间 } func NewNonceDedup(rdb *redis.Client) *NonceDedup { return &NonceDedup{ rdb: rdb, ttl: 24 * time.Hour, // 24 小时后自动过期，释放内存 } } // CheckAndMark 检查 nonce 是否已使用，未使用则标记 // 返回 true 表示是新 nonce（合法），false 表示重复（重放攻击） func (d *NonceDedup) CheckAndMark(deviceId string, nonce []byte) (bool, error) { // key = "nonce:{deviceId}:{nonce_hex}"，按设备隔离 key := fmt.Sprintf("nonce:%s:%x", deviceId, nonce) // SETNX：不存在则设置，返回 true；已存在返回 false ok, err := d.rdb.SetNX(context.Background(), key, 1, d.ttl).Result() if err != nil { returnfalse, fmt.Errorf("nonce 去重查询失败: %w", err) } return ok, nil }

完整的解密 + 去重流程：

func handleMQTTMessageWithDedup(topic string, payload []byte) { deviceId := extractDeviceId(topic) key := keyStore.GetKey(deviceId) // 1. 先提取 nonce 做去重检查（避免无效解密消耗算力） nonce, err := extractNonce(payload) if err != nil { log.Printf("解析 nonce 失败: %v", err) return } // 2. Nonce 去重：重复则丢弃 isNew, err := nonceDedup.CheckAndMark(deviceId, nonce) if err != nil { log.Printf("去重服务异常: %v", err) return } if !isNew { log.Printf("重放攻击检测: topic=%s, nonce 已使用，丢弃", topic) return } // 3. 解密 plainParams, err := DecryptMQTTPayload(payload, key, topic) if err != nil { log.Printf("解密失败: %v", err) return } // 4. 业务处理 processParams(deviceId, plainParams) }

💡 为什么不用 timestamp 防重放？IoT 设备时钟常不准确（无 RTC 或 NTP 不稳定），基于时间窗口的判断容易误判。nonce 去重不依赖设备时钟，更可靠。TTL 设 24 小时足够覆盖网络延迟和 QoS 重传场景。

完整 MQTT Publish 流程（设备端伪代码）

void publish_sensor_data(mqtt_client_t *client, const char *topic) { /* 1. 构造原始 params JSON */ char params[256]; snprintf(params, sizeof(params), "{\"temperature\":{\"value\":%.1f,\"time\":%lu}," "\"humidity\":{\"value\":%.1f,\"time\":%lu}}", read_temperature(), get_timestamp(), read_humidity(), get_timestamp()); /* 2. 加密 params */ uint8_t nonce[12], ciphertext[256], tag[16]; int ret = mqtt_payload_encrypt( device_key, (uint8_t *)params, strlen(params), topic, nonce, ciphertext, tag); if (ret != 0) { log_error("Payload encryption failed: %d", ret); return; } /* 3. 组装标准格式的加密报文：id + version + params.enc */ char encrypted_msg[1024]; snprintf(encrypted_msg, sizeof(encrypted_msg), "{\"id\":\"%s\",\"version\":\"1.0\"," "\"params\":{\"enc\":{" "\"v\":1," "\"nonce\":\"%s\"," "\"ct\":\"%s\"," "\"tag\":\"%s\"}}}", generate_msg_id(), base64_encode(nonce, 12), base64_encode(ciphertext, strlen(params)), base64_encode(tag, 16)); /* 4. 通过 TLS 通道 publish 加密报文 */ mqtt_publish(client, topic, encrypted_msg, strlen(encrypted_msg), QOS_1); }

国密方案（SM4-GCM）

如果需要满足等保三级密评的国密要求，将 AES-256-GCM 替换为 SM4-GCM：

#include <gmssl/sm4.h> #include <gmssl/rand.h> /* SM4-GCM 加密（GMSSL 3.x API） */ int mqtt_payload_encrypt_sm4(const uint8_t *key, /* 16 字节 SM4 密钥 */ const uint8_t *plaintext, size_t pt_len, const uint8_t *aad, size_t aad_len, uint8_t *out_nonce, /* 12 字节输出 */ uint8_t *out_ciphertext, uint8_t *out_tag) { /* 16 字节输出 */ SM4_KEY sm4_key; /* 生成随机 nonce */ rand_bytes(out_nonce, 12); /* SM4-GCM 加密 */ sm4_set_encrypt_key(&sm4_key, key); sm4_gcm_encrypt(&sm4_key, out_nonce, 12, aad, aad_len, plaintext, pt_len, out_ciphertext, 16, out_tag); return0; }

测试与验证

功能验证

安规审查验证清单

过安规评审时，审计人员通常会检查以下点：

抓包验证示例

用 Wireshark 或 MQTT 客户端工具验证：

# 订阅设备 topic，查看收到的是密文而非明文 mosquitto_sub -h broker.example.com -p 8883 \ --cafile ca.crt --cert client.crt --key client.key \ -t "cwlink/device001/thing/property/report" -v # 预期输出（标准格式，params 内为加密数据，看不到原始传感器数据）： # cwlink/device001/thing/property/report {"id":"msg-001","version":"1.0","params":{"enc":{"v":1,"nonce":"...","ct":"...","tag":"..."}}}

最佳实践与避坑

Nonce 管理（最易出错的环节）

⚠️ AES-GCM 的核心安全假设：同一密钥下，nonce 绝对不能重复。一旦重复，攻击者可通过 XOR 两条密文直接恢复明文。这是最常见的实现错误。

性能优化

密钥存储安全

常见审计不通过原因

总结

过安规的 MQTT Payload 加密，核心就三件事：

1. 选对算法：AES-256-GCM（全球通用）或 SM4-GCM（国密合规），一步到位解决加密 + 认证

2. 管好密钥：一机一密，出厂安全注入，存储在 SE/eFuse 中，不明文存储

3. 做好 Nonce：每条消息 12 字节硬件随机数，绝不重复

TLS 是底线，Payload 加密是纵深。对于等保三级、密评、以及面向欧洲市场的 IoT 产品，建议 TLS + Payload 双层加密方案，从架构上满足"端到端数据保密性"要求。

参考内容

• https://www.tc260.org.cn/

• https://www.etsi.org/deliver/etsi_en/303600_303699/303645/03.01.03_60/en_303645v030103p.pdf

• https://datatracker.ietf.org/doc/html/rfc5116

引入地址