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

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

一、功能介绍

证书认证设备动态注册用于减少生产过程中逐台烧录设备证书和设备私钥的工作。

设备首次启动后，使用产品级身份信息向平台发起注册请求。平台校验通过后，返回该设备对应的：

·设备证书 clientCert。

·设备私钥 clientKey。

设备端解密注册结果后，应立即将设备证书和设备私钥写入安全存储，后续使用已保存的凭证完成设备认证。

注意：动态注册不会返回平台 CA 证书。平台 CA 证书应通过 SDK、控制台或产品文档单独获取。

二、动态注册模式

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

自动创建设备应设置合理的创建数量上限，避免因固件异常或 ProductSecret 泄露产生非预期设备。

三、设备状态与重复注册

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

注意：说明：

四、注册流程

证书认证动态注册流程参见下图：

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

五、参数说明

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

如需在控制台开启动态注册、选择注册模式并获取 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 的请求正文必须与实际发送的字节内容完全一致。

十三、平台返回参数

十四、Payload 解密参数

注意：说明：

注意：clientCert 和 clientKey 是一一对应的设备级凭证。动态注册结果中不包含平台 CA 证书。

十六、设备证书

·与 ProductID 和 DeviceName 对应保存。

·可以作为设备身份标识使用。

·不应与其他设备的私钥组合使用。

·更新证书时，应同步更新对应的设备私钥。

十七、设备私钥

·必须写入非易失性存储。

·不得输出至运行日志。

·不得提交到代码仓库。

·不得由多台设备共用。

·建议存储在安全芯片、可信执行环境或受保护的 Flash 区域。

·凭证损坏或丢失后，应按照设备生命周期管理方案重新发放。

十八、平台 CA 证书

平台 CA 证书不通过动态注册返回。设备后续建立 MQTT 安全连接时，应：

·使用 SDK 中内置或托管的 CA 证书，或从控制台及产品文档获取平台 CA 证书。

·将 CA 证书随固件或安全配置预置到设备。

·建立 CA 证书更新和 OTA 替换机制。

十九、使用 Python 获取设备凭证

以下示例用于验证：

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

示例以成功接收并解密设备私钥与设备证书作为运行成功标准，不负责 MQTT 接入。

注意：本示例仅用于测试设备验证。示例只显示脱敏凭证，不保存完整凭证，不应直接用于生产量产。

20、准备运行环境

安装 AES 解密依赖：

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

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

21、准备设备参数

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

22、创建注册脚本

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

#!/usr/bin/env python3# -*- coding: utf-8 -*-import base64import getpassimport hashlibimport hmacimport jsonimport randomimport sslimport sysimport timeimport urllib.errorimport urllib.requestfrom urllib.parse import urlparsetry: from Crypto.Cipher import AESexcept ImportError: AES = NoneDEFAULT_REGISTER_URL = ( "https://ap-guangzhou.gateway.tencentdevices.com" "/device/register")ALGORITHM = "hmacsha256"REQUEST_TIMEOUT = 20def 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": return "precreated", "预创建设备" if choice == "2": return "automatic", "自动创建设备" raise ValueError("动态注册模式只能输入1或2")def get_register_url(): """读取并校验完整的动态注册地址。""" prompt = ( "动态注册地址（直接回车使用" f"{DEFAULT_REGISTER_URL}）：" ) register_url = input(prompt).strip() or DEFAULT_REGISTER_URL parsed = urlparse(register_url) if parsed.scheme.lower() != "https": raise ValueError( "动态注册地址必须使用HTTPS，并以https://开头" ) if not parsed.hostname: raise ValueError("动态注册地址缺少有效域名") if parsed.path.rstrip("/") != "/device/register": raise ValueError( "动态注册地址的路径必须为/device/register" ) if parsed.query or parsed.fragment: raise ValueError( "动态注册地址不能包含查询参数或锚点" ) return register_url, parsed.hostname, "/device/register"def get_device_name(register_mode): """根据动态注册模式提示用户填写设备名称。""" print("\\n设备名称填写说明：") if register_mode == "precreated": print( "请输入已在控制台创建且处于未激活状态的设备名称，" "必须与控制台中的名称完全一致。" ) else: print( "请输入需要注册的设备名称。平台将按照该名称自动" "创建设备，请确保名称符合设备命名规则且未被使用。" ) return required_input("DeviceName：")def build_request_body(product_id, device_name): """生成紧凑JSON请求正文。""" request_data = { "ProductId": product_id, "DeviceName": device_name, } return json.dumps( request_data, ensure_ascii=False, separators=(",", ":"), )def create_signature( product_secret, hostname, canonical_uri, request_body, timestamp, nonce,): """按照腾讯云动态注册签名规则生成请求签名。""" request_hash = hashlib.sha256( request_body.encode("utf-8") ).hexdigest() string_to_sign = "\\n".join( [ "POST", hostname, canonical_uri, "", ALGORITHM, str(timestamp), str(nonce), request_hash, ] ) signature_bytes = hmac.new( product_secret.encode("utf-8"), string_to_sign.encode("utf-8"), hashlib.sha256, ).digest() return base64.b64encode(signature_bytes).decode("ascii")def parse_platform_error(response_data): """提取平台返回的错误信息。""" if not isinstance(response_data, dict): return "Unknown", "平台返回格式异常", None response = response_data.get("Response", response_data) request_id = response.get("RequestId") error = response.get("Error") if isinstance(error, dict): code = error.get("Code", "Unknown") message = error.get("Message", "平台未返回错误详情") return code, message, request_id code = ( response.get("Code") or response.get("code") or "Unknown" ) message = ( response.get("Message") or response.get("message") or "平台未返回错误详情" ) return code, message, request_iddef send_register_request( register_url, hostname, canonical_uri, product_id, device_name, product_secret,): """发送动态注册请求并返回平台Response对象。""" request_body = build_request_body( product_id, device_name, ) timestamp = int(time.time()) nonce = random.SystemRandom().randint( 1, 2147483647, ) signature = create_signature( product_secret=product_secret, hostname=hostname, canonical_uri=canonical_uri, request_body=request_body, timestamp=timestamp, nonce=nonce, ) headers = { "Content-Type": "application/json", "Host": hostname, "X-TC-Algorithm": ALGORITHM, "X-TC-Timestamp": str(timestamp), "X-TC-Nonce": str(nonce), "X-TC-Signature": signature, } request = urllib.request.Request( url=register_url, data=request_body.encode("utf-8"), headers=headers, method="POST", ) tls_context = ssl.create_default_context() try: with urllib.request.urlopen( request, timeout=REQUEST_TIMEOUT, context=tls_context, ) as response: response_text = response.read().decode("utf-8") except urllib.error.HTTPError as exc: error_body = exc.read().decode( "utf-8", errors="replace", ) try: error_data = json.loads(error_body) code, message, request_id = parse_platform_error( error_data ) except json.JSONDecodeError: code = f"HTTP {exc.code}" message = error_body or str(exc.reason) request_id = None detail = f"{code} – {message}" if request_id: detail += f"，RequestId：{request_id}" raise RuntimeError( f"动态注册请求失败：{detail}" ) from exc except urllib.error.URLError as exc: raise RuntimeError( f"无法连接动态注册服务：{exc.reason}" ) from exc except TimeoutError as exc: raise RuntimeError( "动态注册请求超时，请检查网络连接" ) from exc try: response_data = json.loads(response_text) except json.JSONDecodeError as exc: raise RuntimeError( "平台返回的数据不是有效的JSON" ) from exc response_object = response_data.get( "Response", response_data, ) if "Error" in response_object: code, message, request_id = parse_platform_error( response_data ) detail = f"{code} – {message}" if request_id: detail += f"，RequestId：{request_id}" raise RuntimeError( f"动态注册失败：{detail}" ) payload = response_object.get("Payload") if not payload: code, message, request_id = parse_platform_error( response_data ) detail = f"{code} – {message}" if request_id: detail += f"，RequestId：{request_id}" raise RuntimeError( "平台未返回Payload。" f"返回信息：{detail}" ) return response_objectdef decrypt_payload(payload, product_secret): """ 解密动态注册Payload。 腾讯云动态注册协议使用： – AES-128-CBC – ProductSecret前16字节作为密钥 – 16个ASCII字符“0”作为IV – 0x00字节填充 """ secret_bytes = product_secret.encode("utf-8") if len(secret_bytes) < 16: raise ValueError( "ProductSecret长度不足，无法生成AES-128密钥" ) try: encrypted_data = base64.b64decode( payload, validate=True, ) except Exception as exc: raise ValueError( "Payload不是有效的Base64数据" ) from exc if not encrypted_data: raise ValueError("Payload解码结果为空") if len(encrypted_data) % AES.block_size != 0: raise ValueError( "Payload密文长度不是AES块大小的整数倍" ) # ProductSecret UTF-8编码后的前16字节。 aes_key = secret_bytes[:16] # 注意：这里是16个ASCII字符“0”，不是16个0x00。 iv = b"0" * 16 try: cipher = AES.new( aes_key, AES.MODE_CBC, iv=iv, ) decrypted_padded = cipher.decrypt(encrypted_data) except Exception as exc: raise ValueError( "Payload执行AES-CBC解密失败" ) from exc # 协议示例按首个0x00字节截断，不使用PKCS#7。 zero_position = decrypted_padded.find(b"\\x00") if zero_position >= 0: decrypted_data = decrypted_padded[:zero_position] else: decrypted_data = decrypted_padded if not decrypted_data: raise ValueError( "Payload解密结果为空，请检查ProductSecret" ) try: decrypted_text = decrypted_data.decode("utf-8") except UnicodeDecodeError as exc: raise ValueError( "Payload解密结果不是有效的UTF-8文本。" "如果平台已返回Payload，请检查AES解密配置" ) from exc try: credential = json.loads(decrypted_text) except json.JSONDecodeError as exc: raise ValueError( "Payload解密成功，但结果不是有效的JSON数据" ) from exc if not isinstance(credential, dict): raise ValueError( "Payload解密后的数据格式不正确" ) return credentialdef get_first_line(value): """获取PEM内容首行，不显示正文。""" if not isinstance(value, str): return "-" lines = value.strip().splitlines() return lines[0] if lines else "-"def show_certificate_result( credential, response_object, register_mode_name, product_id, device_name,): """验证证书凭证并仅输出非敏感摘要。""" encryption_type = credential.get("encryptionType") try: encryption_type = int(encryption_type) except (TypeError, ValueError): raise ValueError( "返回结果缺少有效的encryptionType" ) if encryption_type != 1: raise ValueError( "平台返回的不是证书认证凭证。" f"encryptionType：{encryption_type}，" "请检查产品认证方式" ) client_cert = credential.get("clientCert") client_key = credential.get("clientKey") if not isinstance(client_cert, str) or not client_cert.strip(): raise ValueError( "返回结果缺少有效的clientCert" ) if not isinstance(client_key, str) or not client_key.strip(): raise ValueError( "返回结果缺少有效的clientKey" ) request_id = response_object.get("RequestId") state = response_object.get("State") payload_len = response_object.get("Len") print("\\n证书认证设备动态注册成功。") print(f"注册模式：{register_mode_name}") print(f"ProductID：{product_id}") print(f"DeviceName：{device_name}") print("凭证类型：证书认证") print(f"encryptionType：{encryption_type}") if state is not None: state_text = { 0: "新建设备", 1: "已有设备", "0": "新建设备", "1": "已有设备", }.get(state, f"未知状态（{state}）") print(f"设备处理状态：{state_text}") if payload_len is not None: print(f"Payload长度：{payload_len}") print( "设备证书：已获取并成功解密" f"（{len(client_cert.encode('utf-8'))}字节，" f"首行：{get_first_line(client_cert)}）" ) print( "设备私钥：已获取并成功解密" f"（{len(client_key.encode('utf-8'))}字节，" "内容不显示）" ) if request_id: print(f"RequestId：{request_id}") print( "\\n完整设备证书和设备私钥不会显示，" "也不会保存到文件。" ) print( "本脚本仅用于验证动态注册及Payload解密流程。" ) print( "生产环境必须在收到凭证后立即写入安全存储，" "不得依赖重复动态注册获取凭证。" )def main(): if AES is None: raise ImportError( "缺少运行依赖，请执行：\\n" 'python -m pip install "pycryptodome==3.23.0"' ) print("证书认证设备动态注册验证工具") print( "本工具仅验证能否获取并解密设备证书和设备私钥。" ) print( "输入及返回的完整凭证不会显示，也不会保存到文件。\\n" ) register_mode, register_mode_name = ( select_register_mode() ) print(f"\\n已选择：{register_mode_name}") if register_mode == "precreated": print( "请确认目标设备已在企业实例中预先创建，" "且当前处于未激活状态。" ) else: print( "请确认当前使用企业实例，并已在产品中" "启用自动创建设备。" ) register_url, hostname, canonical_uri = ( get_register_url() ) 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_name}") print(f"动态注册地址：{register_url}") print(f"ProductID：{product_id}") print(f"DeviceName：{device_name}") print( "解密后的完整设备证书和设备私钥仅保留在" "本次程序运行内存中。" ) print( "未激活设备重复注册可能刷新设备凭证，" "并使此前取得的凭证失效。" ) confirm = input( "是否开始动态注册？[y/N]：" ).strip().lower() if confirm != "y": print("已取消动态注册。") return 0 print(f"\\n正在请求：{register_url}") response_object = send_register_request( register_url=register_url, hostname=hostname, canonical_uri=canonical_uri, product_id=product_id, device_name=device_name, product_secret=product_secret, ) credential = decrypt_payload( payload=response_object["Payload"], product_secret=product_secret, ) show_certificate_result( credential=credential, response_object=response_object, register_mode_name=register_mode_name, product_id=product_id, device_name=device_name, ) # 不执行任何文件写入。 # 函数结束后，凭证仅等待随进程退出释放。 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)

23、运行脚本

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

程序将依次要求输入：

注册模式动态注册地址ProductIDDeviceNameProductSecret

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

24、运行成功结果

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

正在请求：https://ap-guangzhou.gateway.tencentdevices.com/device/register证书认证设备动态注册成功。注册模式：预创建设备ProductID：*******DeviceName：*******凭证类型：证书认证encryptionType：*设备处理状态：已有设备Payload长度：3007设备证书：已获取并成功解密（1204字节，首行：—–BEGIN CERTIFICATE—–）设备私钥：已获取并成功解密（1704字节，内容不显示）RequestId：*******完整设备证书和设备私钥不会显示，也不会保存到文件。本脚本仅用于验证动态注册及Payload解密流程。生产环境必须在收到凭证后立即写入安全存储，不得依赖重复动态注册获取凭证。

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

26、平台返回 signature error

检查：

·ProductSecret 是否属于当前 ProductID。

·签名是否基于实际发送的请求正文计算。

·Host 和请求路径是否与动态注册地址一致。

·X-TC-Algorithm 是否为 hmacsha256。

·签名结果是否经过 Base64 编码。

·请求正文中是否增加了空格、换行或其他字段。

27、平台返回 X-TC-Nonce error

每次请求均应生成新的随机正整数，不应使用空值、负数或重复的固定值。

28、注册请求被拒绝

检查：

·当前是否为企业实例。

·产品认证方式是否为证书认证。

·产品是否已开启动态注册。

·控制台配置的注册模式是否与当前场景一致。

·预创建设备是否存在并处于未激活状态。

·自动创建设备的 DeviceName 是否已经存在。

·自动创建设备数量是否达到控制台限制。

29、解密 Payload 失败

检查：

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

·AES 密钥是否取 ProductSecret 的前16个字节。

·是否使用 AES-128-CBC。

·IV 是否为16个字符0。

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

·是否正确处理 PKCS#7 填充。

30、返回密钥认证凭证

如果 encryptionType=2，说明当前产品为密钥认证产品。请确认 ProductID 是否正确，并改用 密钥认证设备动态注册 文档。

31、获取凭证后无法再次注册

设备激活后，平台会拒绝再次动态注册。设备端应在首次注册成功后立即持久化设备证书和设备私钥，后续启动时直接读取本地凭证。