📚 课程目录第2章设备注册与身份认证

2. 设备注册与身份认证

任务说明

本案例完成设备接入云平台前的注册与身份认证闭环：设备端基于预置的产品密钥(ProductKey)与设备密钥(DeviceSecret)，通过HMAC-SHA256算法动态签名，向云平台发起注册请求；平台端验证签名合法性后，为设备签发用于后续通信的访问令牌(AccessToken)。代码同时覆盖设备端签名生成与平台端验签授权两个核心环节，不涉及业务数据上报。

设备注册与认证流程

设备端签名生成

设备端使用预置的三元组信息生成动态签名。签名串格式为 deviceName=xxx&productKey=xxx&timestamp=xxx，以DeviceSecret作为HMAC密钥，输出十六进制摘要。timestamp取当前Unix秒级时间戳，保证每次注册签名不同。

# device_auth.py
import hmac, hashlib, time

class DeviceAuthenticator:
    def __init__(self, product_key, device_name, device_secret):
        self.product_key = product_key
        self.device_name = device_name
        self.device_secret = device_secret

    def build_sign(self, timestamp=None):
        """生成HMAC-SHA256签名，返回(sign, timestamp)"""
        if timestamp is None:
            timestamp = str(int(time.time()))
        # 签名串：参数名按字典序排列
        payload = f"deviceName={self.device_name}&productKey={self.product_key}×tamp={timestamp}"
        sign = hmac.new(
            self.device_secret.encode('utf-8'),
            payload.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()
        return sign, timestamp

    def build_register_payload(self):
        """构造注册请求体"""
        sign, ts = self.build_sign()
        return {
            "productKey": self.product_key,
            "deviceName": self.device_name,
            "timestamp": ts,
            "sign": sign
        }

if __name__ == "__main__":
    # 设备出厂预置信息（示例值）
    auth = DeviceAuthenticator(
        product_key="pk_iot_demo_001",
        device_name="dev_alpha_01",
        device_secret="s3cret_key_abc123"
    )
    payload = auth.build_register_payload()
    print("注册请求参数：")
    for k, v in payload.items():
        print(f"  {k} = {v}")

平台端验签与令牌签发

平台端收到注册请求后，从设备注册表查询该设备对应的DeviceSecret，用相同算法重新计算签名并与请求中的sign比对。同时校验timestamp是否在允许的时间偏移内（此处设为300秒）。验签通过且设备状态为enabled时，生成一个随机AccessToken并记录过期时间（24小时）。

# platform_auth.py
import hmac, hashlib, time, secrets, json

class PlatformAuthServer:
    def __init__(self):
        # 模拟设备注册表：productKey -> {deviceName: {secret, status}}
        self.device_registry = {
            "pk_iot_demo_001": {
                "dev_alpha_01": {
                    "secret": "s3cret_key_abc123",
                    "status": "enabled"
                }
            }
        }
        # 已签发的令牌存储
        self._store = {}

    def _calc_sign(self, secret, device_name, product_key, timestamp):
        payload = f"deviceName={device_name}&productKey={product_key}×tamp={timestamp}"
        return hmac.new(secret.encode(), payload.encode(), hashlib.sha256).hexdigest()

    def register_device(self, reg_data):
        """验证注册请求，成功返回AccessToken，失败返回错误信息"""
        product_key = reg_data.get("productKey")
        device_name = reg_data.get("deviceName")
        timestamp = reg_data.get("timestamp")
        client_sign = reg_data.get("sign")

        # 1. 设备是否存在
        product = self.device_registry.get(product_key)
        if not product or device_name not in product:
            return {"success": False, "code": "DEVICE_NOT_FOUND"}

        device = product[device_name]

        # 2. 设备状态是否可用
        if device["status"] != "enabled":
            return {"success": False, "code": "DEVICE_DISABLED"}

        # 3. 时间戳防重放校验
        try:
            ts_int = int(timestamp)
        except (TypeError, ValueError):
            return {"success": False, "code": "INVALID_TIMESTAMP"}
        if abs(time.time() - ts_int) > 300:
            return {"success": False, "code": "TIMESTAMP_EXPIRED"}

        # 4. 签名验证
        expected_sign = self._calc_sign(
            device["secret"], device_name, product_key, timestamp
        )
        if not hmac.compare_digest(expected_sign, client_sign):
            return {"success": False, "code": "SIGN_MISMATCH"}

        # 5. 签发AccessToken
         = secrets._hex(32)
        expire_at = int(time.time()) + 86400  # 24小时后过期
        self._store[] = {
            "productKey": product_key,
            "deviceName": device_name,
            "expireAt": expire_at
        }
        return {
            "success": True,
            "accessToken": ,
            "expiresIn": 86400,
            "tokenType": "Bearer"
        }

if __name__ == "__main__":
    # 模拟一次完整注册认证
    server = PlatformAuthServer()

    # 构造与设备端一致的注册请求（实际场景由设备端发送）
    import device_auth
    dev_auth = device_auth.DeviceAuthenticator(
        product_key="pk_iot_demo_001",
        device_name="dev_alpha_01",
        device_secret="s3cret_key_abc123"
    )
    reg_payload = dev_auth.build_register_payload()

    result = server.register_device(reg_payload)
    print(json.dumps(result, indent=2, ensure_ascii=False))

落地与验证

运行方式：将两个Python文件置于同一目录，先运行python device_auth.py观察设备端生成的签名参数；再运行python platform_auth.py，该脚本会调用设备端模块构造请求并完成平台验签。

验证判定标准：

正常路径：platform_auth.py输出"success": true且包含32字节十六进制格式的accessToken，说明签名验证与令牌签发链路正确。
篡改验证：手动修改reg_payload中的sign为任意错误值，重新运行应返回"code": "SIGN_MISMATCH"。
时间戳验证：将timestamp改为当前时间减去600秒的值，应返回"code": "TIMESTAMP_EXPIRED"。

以上两条异常路径可通过临时修改platform_auth.py中reg_payload的对应字段来验证，无需改动核心逻辑。