物联网开发平台 密钥认证动态注册

这篇以成功获取并保存设备凭证为完成标准，不包含后续 MQTT 连接与设备上线。需要连接平台时，请参见 设备密钥认证。

一、功能介绍

动态注册用于降低量产阶段逐台写入设备密钥的成本。设备首次联网时，使用产品级注册信息向平台发起请求，平台校验通过后返回加密的设备级凭证。

密钥认证动态注册涉及以下两类密钥：

ProductSecret 由同一产品下参与动态注册的设备共享，不能代替 DeviceSecret 进行 MQTT 身份认证。

二、动态注册模式

发起动态注册前，需要先根据实例类型和设备创建方式选择注册模式。

三、设备状态与重复注册

设备是否允许动态注册取决于其激活状态。

注意：说明：

四、注册流程

密钥认证设备的动态注册流程如下：

动态注册完成只表示设备获得了后续身份认证所需的 DeviceSecret，不表示设备已经连接平台或处于在线状态。

五、参数说明

动态注册涉及注册模式、设备身份、请求地址、请求头、请求正文和返回结果六类参数。

如需在控制台开启动态注册、选择注册模式并获取 ProductID、ProductSecret 和 DeviceName，请参见 启用动态注册。

六、注册输入参数

七、注册模式参数

注册模式用于确定设备是否需要提前创建，但不会作为字段发送至动态注册接口。

RegisterType 是产品动态注册配置，不是设备注册请求正文中的参数。

八、设备身份参数

ProductSecret 属于产品级敏感凭证。任何获得该凭证的设备或系统都可能以该产品身份发起注册请求，应限制其分发和读取范围。

九、请求地址

# 广州地域https://ap-guangzhou.gateway.tencentdevices.com/device/register# 曼谷地域https://ap-bangkok.gateway.tencentdevices.com/device/register

十、请求头参数

十一、请求正文参数

请求示例：

{ "ProductId": "你的ProductID", "DeviceName": "你的DeviceName"}

参数名称区分大小写。请求正文使用 ProductId，不能写成 ProductID。

十二、签名参数

待签名字符串按照以下顺序拼接：

POST${Host}/device/registerhmacsha256${Timestamp}${Nonce}${RequestHash}

签名公式：

RequestHash = LowercaseHex(SHA256(RequestBody))Signature = Base64( HMAC-SHA256(ProductSecret, StringToSign))

用于生成 RequestHash 的请求正文必须与实际发送的字节内容完全一致。

十三、平台返回参数

平台正常处理请求后，主要返回以下参数：

返回示例：

{ "Response": { "Len": 53, "Payload": "加密后的设备凭证", "RequestId": "请求唯一标识", "State": 1 }}

Payload 不能直接作为设备密钥使用，需要先进行解密。

十四、Payload 解密参数

密钥认证设备解密后的内容类似：

{ "encryptionType": 2, "psk": "设备级DeviceSecret"}

十五、使用 Python 验证动态注册

以下示例用于验证：

选择注册模式→ 输入注册参数→ 生成请求签名→ 发起动态注册→ 解密Payload→ 脱敏显示设备凭证

示例以成功接收并解密 DeviceSecret 作为运行成功标准，不负责 MQTT 接入。

注意：本示例仅用于测试设备验证。未激活设备重复运行脚本可能重新生成 DeviceSecret，使此前获得的旧密钥失效。示例只显示脱敏凭证，不保存完整凭证，不应直接用于生产量产。

十六、准备运行环境

安装加解密依赖：

# Windows PowerShellpython -m pip install "pycryptodome==3.23.0"# Linux或macOSpython3 -m pip install "pycryptodome==3.23.0"

如果 Linux 提示 Python 环境由系统管理，请在虚拟环境中安装依赖。

十七、准备设备参数

请在控制台开启动态注册、选择注册模式并获取 ProductID、ProductSecret 和 DeviceName，详情请参见 启用动态注册。

十八、创建注册脚本

将以下代码保存为 dynamic_register_psk.py：

#!/usr/bin/env python3# -*- coding: utf-8 -*-import base64import getpassimport hashlibimport hmacimport jsonimport secretsimport sysimport timeimport urllib.errorimport urllib.requestfrom urllib.parse import urlsplit, urlunsplittry: from Crypto.Cipher import AESexcept ImportError: AES = NoneDEFAULT_REGISTER_URL = ( "https://ap-guangzhou.gateway.tencentdevices.com/device/register")def required_input(prompt): """读取必填参数并清除首尾空格。""" value = input(prompt).strip() if not value: raise ValueError("输入内容不能为空") return valuedef select_register_mode(): """选择控制台中已配置的动态注册模式。""" print("请选择动态注册模式：") print("1. 预创建设备：消费实例和企业实例均支持") print("2. 自动创建设备：仅企业实例支持") choice = required_input("请输入1或2：") if choice == "1": register_mode = "预创建设备" print(f"\\n已选择：{register_mode}") print("请确认目标设备已在控制台创建且处于未激活状态。") return register_mode if choice == "2": register_mode = "自动创建设备" print(f"\\n已选择：{register_mode}") print("请确认当前使用企业实例，并已开启自动创建设备。") return register_mode raise ValueError("注册模式只能输入1或2")def get_register_endpoint(): """读取并校验完整动态注册地址。""" print("\\n请输入动态注册参数。") print("输入内容仅在本次运行期间使用，不会保存到文件。\\n") register_url = input( "动态注册地址" f"（直接回车使用{DEFAULT_REGISTER_URL}）：" ).strip() if not register_url: register_url = DEFAULT_REGISTER_URL if any(char in register_url for char in "${}"): raise ValueError( "动态注册地址不能包含变量符号，" "请填写实际地域的完整地址" ) parsed = urlsplit(register_url) if parsed.scheme.lower() != "https": raise ValueError( "动态注册地址必须以https://开头" ) if not parsed.hostname: raise ValueError( "动态注册地址中缺少有效的服务域名" ) if not parsed.hostname.endswith( ".gateway.tencentdevices.com" ): raise ValueError( "动态注册地址不是有效的腾讯云设备注册服务地址" ) if parsed.username or parsed.password: raise ValueError( "动态注册地址不能包含用户名或密码" ) if parsed.query or parsed.fragment: raise ValueError( "动态注册地址不能包含查询参数或片段" ) # 兼容地址末尾多输入一个斜杠。 request_path = parsed.path.rstrip("/") if request_path != "/device/register": raise ValueError( "动态注册地址的路径必须为/device/register" ) # 生成规范化URL，确保请求路径和签名路径一致。 normalized_url = urlunsplit( ( "https", parsed.netloc, request_path, "", "", ) ) return { "url": normalized_url, "host": parsed.netloc, "path": request_path, }def get_device_name(register_mode): """根据注册模式提示用户输入DeviceName。""" print("\\n设备名称填写说明：") if register_mode == "预创建设备": print( "请输入已在控制台创建且当前处于未激活状态的" "设备名称，必须与控制台中的名称完全一致。" ) else: print( "请输入计划使用的设备名称。若该名称不存在，" "平台将按此名称自动创建设备。" ) print( "设备名称必须符合平台命名规则，" "并在当前产品下保持唯一。" ) return required_input("DeviceName：")def build_request( register_endpoint, product_id, product_secret, device_name,): """生成动态注册请求正文和鉴权请求头。""" request_body = json.dumps( { "ProductId": product_id, "DeviceName": device_name, }, separators=(",", ":"), ensure_ascii=False, ) timestamp = str(int(time.time())) # X-TC-Nonce必须是随机正整数。 nonce = str( secrets.randbelow(2147483646) + 1 ) payload_hash = hashlib.sha256( request_body.encode("utf-8") ).hexdigest() # 第4行是空查询字符串，必须保留。 string_to_sign = "\\n".join( [ "POST", register_endpoint["host"], register_endpoint["path"], "", "hmacsha256", timestamp, nonce, payload_hash, ] ) signature_bytes = hmac.new( product_secret.encode("utf-8"), string_to_sign.encode("utf-8"), hashlib.sha256, ).digest() signature = base64.b64encode( signature_bytes ).decode("utf-8") headers = { "Content-Type": "application/json", "Host": register_endpoint["host"], "X-TC-Algorithm": "hmacsha256", "X-TC-Timestamp": timestamp, "X-TC-Nonce": nonce, "X-TC-Signature": signature, } return request_body.encode("utf-8"), headersdef send_register_request( register_endpoint, request_body, headers,): """使用Python标准库发送动态注册请求。""" request = urllib.request.Request( url=register_endpoint["url"], data=request_body, headers=headers, method="POST", ) try: with urllib.request.urlopen( request, timeout=15, ) as response: response_body = response.read().decode("utf-8") except urllib.error.HTTPError as exc: try: error_body = exc.read().decode( "utf-8", errors="replace", ) except Exception: error_body = "" message = f"动态注册请求返回HTTP {exc.code}" if error_body: message += f"，响应内容：{error_body}" raise RuntimeError(message) from exc except urllib.error.URLError as exc: raise ConnectionError( f"无法连接动态注册服务：{exc.reason}" ) from exc except TimeoutError as exc: raise TimeoutError( "动态注册请求超时，请检查网络和注册地址" ) from exc try: return json.loads(response_body) except json.JSONDecodeError as exc: raise ValueError( "平台注册接口未返回有效的JSON数据：" f"{response_body}" ) from excdef decrypt_payload(payload, product_secret): """使用ProductSecret解密动态注册返回的Payload。""" product_secret_bytes = product_secret.encode("utf-8") if len(product_secret_bytes) < 16: raise ValueError( "ProductSecret长度不足，无法生成AES密钥" ) # AES-128密钥取ProductSecret的前16字节。 aes_key = product_secret_bytes[:16] # 动态注册协议使用16个字符0作为初始向量。 aes_iv = b"0" * 16 try: encrypted_payload = base64.b64decode( payload, validate=True, ) except Exception as exc: raise ValueError( "响应中的Payload不是有效的Base64数据" ) from exc if len(encrypted_payload) % AES.block_size != 0: raise ValueError( "Payload长度不符合AES-CBC解密要求" ) cipher = AES.new( aes_key, AES.MODE_CBC, aes_iv, ) decrypted_bytes = cipher.decrypt( encrypted_payload ).rstrip(b"\\x00") try: decrypted_text = decrypted_bytes.decode("utf-8") return json.loads(decrypted_text) except Exception as exc: raise ValueError( "Payload解密成功，但无法解析为JSON" ) from excdef parse_register_response( response_data, product_secret,): """检查动态注册响应并解析设备凭证。""" response = response_data.get("Response") if not isinstance(response, dict): raise ValueError( f"动态注册响应格式异常：{response_data}" ) error = response.get("Error") if error: error_code = error.get("Code", "Unknown") error_message = error.get("Message", "Unknown") raise RuntimeError( f"动态注册失败：" f"{error_code} – {error_message}" ) payload = response.get("Payload") if not payload: raise ValueError( f"动态注册响应中未找到Payload：{response}" ) credential = decrypt_payload( payload=payload, product_secret=product_secret, ) return response, credentialdef mask_secret(secret): """脱敏显示DeviceSecret。""" if len(secret) <= 8: return "*" * len(secret) return f"{secret[:4]}********{secret[-4:]}"def main(): if AES is None: raise ImportError( "当前Python环境缺少pycryptodome，请执行：\\n" f'{sys.executable} -m pip install ' '"pycryptodome==3.23.0"' ) print("密钥认证设备动态注册验证工具") print( "未激活设备重复注册可能刷新DeviceSecret，" "请仅使用测试设备验证。\\n" ) register_mode = select_register_mode() register_endpoint = get_register_endpoint() product_id = required_input("ProductID：") device_name = get_device_name(register_mode) product_secret = getpass.getpass( "ProductSecret（输入内容不会显示）：" ).strip() if not product_secret: raise ValueError("ProductSecret不能为空") print("\\n即将发起动态注册。") print(f"注册模式：{register_mode}") print(f"动态注册地址：{register_endpoint['url']}") print(f"ProductID：{product_id}") print(f"DeviceName：{device_name}") print( "如果设备仍处于未激活状态，本次注册可能生成新的" "DeviceSecret，并使旧密钥失效。" ) confirmation = input( "是否开始动态注册？[y/N]：" ).strip().lower() if confirmation not in {"y", "yes"}: print("已取消动态注册。") return 0 request_body, headers = build_request( register_endpoint=register_endpoint, product_id=product_id, product_secret=product_secret, device_name=device_name, ) print( f"\\n正在请求：{register_endpoint['url']}" ) response_data = send_register_request( register_endpoint=register_endpoint, request_body=request_body, headers=headers, ) response, credential = parse_register_response( response_data=response_data, product_secret=product_secret, ) encryption_type = credential.get("encryptionType") device_secret = credential.get("psk") if encryption_type != 2 or not device_secret: raise ValueError( "平台未返回有效的密钥认证设备凭证" ) state = response.get("State") state_text = { 0: "新建设备", 1: "已有设备", "0": "新建设备", "1": "已有设备", }.get(state, "未返回或未知") print("\\n动态注册运行成功。") print(f"注册模式：{register_mode}") print(f"ProductID：{product_id}") print(f"DeviceName：{device_name}") print(f"设备处理状态：{state_text}") print("凭证类型：密钥认证") print(f"encryptionType：{encryption_type}") print( f"DeviceSecret（脱敏）：" f"{mask_secret(device_secret)}" ) print(f"DeviceSecret长度：{len(device_secret)}") print( f"RequestId：" f"{response.get('RequestId', '未返回')}" ) print( "\\n已成功接收设备凭证。" "完整DeviceSecret不会显示或保存。" ) print( "如用于生产设备，请在程序中接入安全存储，" "并在注册成功后立即保存完整DeviceSecret。" ) return 0if __name__ == "__main__": try: sys.exit(main()) except KeyboardInterrupt: print("\\n用户终止操作。") sys.exit(0) except Exception as exc: print(f"\\n运行失败：{exc}") sys.exit(1)

十九、运行脚本

# Windows PowerShellpython dynamic_register_psk.py# Linux或macOSpython3 dynamic_register_psk.py

程序将依次要求输入：

注册模式动态注册地址ProductIDDeviceNameProductSecret

ProductSecret 输入时不会显示在终端中，也不会写入文件。

20、运行成功结果

成功接收并解密设备凭证后，输出类似：

动态注册运行成功。注册模式：预创建设备ProductID：ASJXXXXXXXDeviceName：device_001设备处理状态：已有设备凭证类型：密钥认证encryptionType：2DeviceSecret（脱敏）：lDZ6********s7Q==DeviceSecret长度：24RequestId：xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx已成功接收设备凭证。完整DeviceSecret不会显示或保存。

可在控制台设备页面查看注册成功的设备信息。

22、预创建设备模式提示设备不存在

检查：

·是否已在当前产品下创建目标设备。

·DeviceName 是否与控制台完全一致。

·产品和设备是否属于当前实例。

·请求地址是否与产品所在地域一致。

23、自动创建设备模式注册失败

检查：

·是否使用企业实例。

·是否开启自动创建设备。

·动态注册设备数量是否达到配置上限。

·同名设备是否已经激活。

24、重复注册后原密钥无法使用

未激活设备重复注册可能重新生成 DeviceSecret。注册成功后，应以最新返回的 PSK 为准，之前获得的旧密钥可能已经失效。

25、已激活设备无法再次注册

设备曾使用 DeviceSecret 成功接入 MQTT 后即进入已激活状态。已激活设备不能通过动态注册重新获取同一 DeviceName 的设备密钥。

26、Payload 解密失败

检查：

·是否使用当前产品的 ProductSecret。

·是否先对 Payload 进行 Base64 解码。

·是否使用 ProductSecret 前16个字节作为 AES 密钥。

·是否使用 AES-128-CBC。

·初始化向量是否为16个字符 0。

·是否正确清理解密结果末尾的零字节。

27、恢复出厂后无法重新注册

如果设备恢复出厂时删除了 DeviceSecret，但云端设备已经激活，则不能通过同一设备名称再次注册获取设备凭证。

恢复出厂时建议保留设备身份凭证。