密钥认证动态注册

# 广州地域
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)
)

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

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

选择注册模式
→ 输入注册参数
→ 生成请求签名
→ 发起动态注册
→ 解密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 secrets
import sys
import time
import urllib.error
import urllib.request
from urllib.parse import urlsplit, urlunsplit

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


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


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":
        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"), headers


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


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


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


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

# Windows PowerShell
python dynamic_register_psk.py

# Linux或macOS
python3 dynamic_register_psk.py

注册模式
动态注册地址
ProductID
DeviceName
ProductSecret