3. 设备注册与认证机制

任务目标

为OTA升级系统设计设备接入安全机制，确保只有完成注册且通过认证的合法设备才能连接OTA服务器并获取升级包。本案例实现设备唯一标识生成、注册请求签名、服务器端注册校验与设备认证令牌签发，形成完整的设备接入闭环。

核心设计

• 设备唯一标识：使用设备出厂时烧录的硬件序列号（32位十六进制字符串）作为不可篡改的设备身份，结合随机生成的设备密钥作为注册凭证。
• 注册流程：设备携带序列号+密钥向服务器发起注册，服务器校验序列号格式与密钥强度后，生成设备ID（UUID）并签发认证令牌。
• 认证机制：设备后续每次连接OTA服务器时，需携带令牌；服务器验证令牌有效性与过期时间，拒绝未注册或令牌失效的请求。

完整实现

import hashlib
import hmac
import uuid
import time
import re

# ---------- 设备端 ----------
class OTA_Device:
    def __init__(self, serial_no, secret_key):
        self.serial_no = serial_no          # 出厂烧录的硬件序列号
        self.secret_key = secret_key        # 设备密钥（仅设备与服务器知晓）
        self.device_id = None
        self. = None
        self._expire = 0

    def register_payload(self):
        """生成注册请求：序列号 + 时间戳 + HMAC签名"""
        timestamp = str(int(time.time()))
        message = f"{self.serial_no}|{timestamp}"
        signature = hmac.new(
            self.secret_key.encode(),
            message.encode(),
            hashlib.sha256
        ).hexdigest()
        return {
            "serial_no": self.serial_no,
            "timestamp": timestamp,
            "signature": signature
        }

    def authenticate_header(self):
        """生成认证请求头：设备ID + 令牌"""
        return {
            "X-Device-ID": self.device_id,
            "X-Auth-": self.
        }

# ---------- 服务器端 ----------
class OTA_Server:
    def __init__(self):
        # 模拟数据库：序列号 -> 密钥（出厂预置）
        self.registered_devices = {}
        # 已注册设备：device_id -> {serial_no, , _expire}
        self.active_sessions = {}

    def add_factory_device(self, serial_no, secret_key):
        """出厂时预置设备信息到服务器数据库"""
        self.registered_devices[serial_no] = secret_key

    def _verify_signature(self, payload):
        """校验注册请求签名"""
        if payload["serial_no"] not in self.registered_devices:
            return False
        expected_key = self.registered_devices[payload["serial_no"]]
        message = f"{payload['serial_no']}|{payload['timestamp']}"
        expected_sig = hmac.new(
            expected_key.encode(),
            message.encode(),
            hashlib.sha256
        ).hexdigest()
        # 防重放：时间戳在5分钟内有效
        if abs(int(time.time()) - int(payload["timestamp"])) > 300:
            return False
        return hmac.compare_digest(expected_sig, payload["signature"])

    def register_device(self, payload):
        """处理设备注册请求，返回设备ID与令牌"""
        if not self._verify_signature(payload):
            return None
        serial_no = payload["serial_no"]
        device_id = str(uuid.uuid4())
         = hashlib.sha256(
            f"{serial_no}|{uuid.uuid4()}|{time.time()}".encode()
        ).hexdigest()
        expire = int(time.time()) + 3600  # 令牌1小时有效
        self.active_sessions[device_id] = {
            "serial_no": serial_no,
            "": ,
            "_expire": expire
        }
        return {"device_id": device_id, "": , "expire": expire}

    def authenticate(self, device_id, ):
        """验证设备认证信息是否有效"""
        session = self.active_sessions.get(device_id)
        if not session:
            return False
        if session[""] != :
            return False
        if int(time.time()) > session["_expire"]:
            return False
        return True

# ---------- 验证流程 ----------
def demo_flow():
    # 1. 出厂预置
    server = OTA_Server()
    serial = "A1B2C3D4E5F60718293A4B5C6D7E8F90"
    secret = "k9x2m7q4z8p1v5t3"
    server.add_factory_device(serial, secret)

    # 2. 设备注册
    device = OTA_Device(serial, secret)
    reg_payload = device.register_payload()
    reg_result = server.register_device(reg_payload)
    assert reg_result is not None, "注册失败"
    device.device_id = reg_result["device_id"]
    device. = reg_result[""]
    device._expire = reg_result["expire"]
    print("注册成功，设备ID:", device.device_id[:8], "...")

    # 3. 设备认证（模拟OTA连接）
    auth_header = device.authenticate_header()
    is_valid = server.authenticate(
        auth_header["X-Device-ID"],
        auth_header["X-Auth-"]
    )
    assert is_valid, "认证失败"
    print("认证通过，允许接入OTA服务器")

    # 4. 非法设备测试
    fake_device = OTA_Device("FFFFFFFFFF0000000000000000000000", "wrong_key")
    fake_payload = fake_device.register_payload()
    fake_result = server.register_device(fake_payload)
    assert fake_result is None, "非法设备不应注册成功"
    print("非法设备注册被拒绝（预期行为）")

    # 5. 过期令牌测试
    server.active_sessions[device.device_id]["_expire"] = int(time.time()) - 10
    expired_valid = server.authenticate(device.device_id, device.)
    assert not expired_valid, "过期令牌不应通过"
    print("过期令牌认证被拒绝（预期行为）")

    print("\n全部验证通过：注册与认证机制工作正常")

if __name__ == "__main__":
    demo_flow()

运行方式

python3 ota_auth_demo.py

验证方法与通过标准

• 注册成功路径：合法设备携带正确签名注册，返回设备ID与令牌，控制台输出“注册成功”。
• 认证通过路径：使用注册返回的令牌请求认证，输出“认证通过，允许接入OTA服务器”。
• 非法设备拒绝：伪造序列号或密钥的设备注册返回None，输出“非法设备注册被拒绝”。
• 过期令牌拒绝：将令牌有效期改为过去时间后认证失败，输出“过期令牌认证被拒绝”。
• 最终判定：程序无异常退出且打印“全部验证通过”，说明注册与认证闭环正确。

关键说明

• HMAC-SHA256签名确保注册请求来自持有正确密钥的设备，时间戳窗口防止重放攻击。
• 认证令牌有效期1小时，过期后设备需重新注册或刷新令牌（本案例聚焦核心机制，刷新流程可基于同一签名方案扩展）。
• 实际生产环境中，设备密钥应安全存储在安全元件或TEE中，服务器端密钥应使用KMS管理，本案例用内存字典模拟数据库存储。