证书认证动态注册

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

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

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

POST
${Host}
/device/register

hmacsha256
${Timestamp}
${Nonce}
${RequestHash}

RequestHash = LowercaseHex(SHA256(RequestBody))

Signature = Base64(
    HMAC-SHA256(ProductSecret, StringToSign)
)

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

# Windows PowerShell
python -m pip install "pycryptodome==3.23.0"

# Linux或macOS
python3 -m pip install "pycryptodome==3.23.0"

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import base64
import getpass
import hashlib
import hmac
import json
import random
import ssl
import sys
import time
import urllib.error
import urllib.request
from urllib.parse import urlparse

try:
    from Crypto.Cipher import AES
except ImportError:
    AES = None


DEFAULT_REGISTER_URL = (
    "https://ap-guangzhou.gateway.tencentdevices.com"
    "/device/register"
)

ALGORITHM = "hmacsha256"
REQUEST_TIMEOUT = 20


def required_input(prompt):
    """读取必填参数并清除首尾空格。"""
    value = input(prompt).strip()
    if not value:
        raise ValueError("输入内容不能为空")
    return value


def 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_id


def 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_object


def 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 credential


def 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 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\\n用户终止运行。")
        sys.exit(0)
    except Exception as exc:
        print(f"\\n运行失败：{exc}")
        sys.exit(1)