MCU设备鉴权接口文档

接口信息

• 接口地址: /api/mcu/client/connect
• 请求方式: POST
• Content-Type: application/json
• 功能说明: MCU设备鉴权，获取MQTT连接参数

demo地址

https://bmsapi.20783378.com/api/mcu/client/connect

请求参数

Body

{
  "imei": "860602069165352",
  "use_mqtts": false
}

请求头

AES-128-ECB签名验签说明

签名机制

接口采用 AES-128-ECB + Base64 对称签名进行设备身份验证：

• 加密模式: AES-ECB (Electronic Codebook)
• 密钥长度: 16字节（128位）
• 填充: PKCS7
• 签名: 加密后的密文，Base64编码

待签名数据格式

{imei}:{X-Timestamp}

签名算法

• 算法: AES-128-ECB
• 密钥: 16字节（十六进制字符串，32字符）
• 填充: PKCS7
• 签名输出: 加密密文（Base64编码）

配置信息

服务端配置

密钥信息（验证使用，正式上线前修改）：

MCU_AES_KEY: 'a3f1b8c9d2e47056'

说明：

• 密钥格式：16字节 = 32位十六进制字符

客户端加签实现

Python 实现示例

import base64
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import padding

# 密钥配置（与服务端一致）
KEY = b'a3f1b8c9d2e47056'  # 16字节ASCII字符串

def pkcs7_pad(data: bytes, block_size: int = 16) -> bytes:
    """PKCS7 填充"""
    padder = padding.PKCS7(block_size * 8).padder()
    return padder.update(data) + padder.finalize()

def generate_signature(imei: str, timestamp: str) -> str:
    """
    生成AES-128-ECB签名

    Args:
        imei: 设备IMEI号
        timestamp: 请求时间戳（秒级Unix时间戳）

    Returns:
        Base64编码的签名字符串
    """
    # 1. 构造待签名数据
    payload = f'{imei}:{timestamp}'

    # 2. AES-ECB 加密
    cipher = Cipher(algorithms.AES(KEY), modes.ECB(), backend=default_backend())
    encryptor = cipher.encryptor()
    padded_plaintext = pkcs7_pad(payload.encode('utf-8'))
    ciphertext = encryptor.update(padded_plaintext) + encryptor.finalize()

    # 3. Base64编码
    return base64.b64encode(ciphertext).decode('utf-8')

# 使用示例
imei = '860602069165352'
timestamp = '1780552892'
signature = generate_signature(imei, timestamp)
print(f"Signature: {signature}")

MicroPython 实现示例（嵌入式设备）

import ucryptolib
import ubinascii as base64

# 密钥配置（与服务端一致）
KEY = b'a3f1b8c9d2e47056'  # 16字节ASCII字符串

def pkcs7_pad(data, block_size=16):
    """PKCS7 填充"""
    padding_size = block_size - (len(data) % block_size)
    padding = bytes([padding_size] * padding_size)
    return data + padding

def generate_signature(imei, timestamp):
    """
    生成AES-128-ECB签名

    Args:
        imei: 设备IMEI号
        timestamp: 请求时间戳（秒级Unix时间戳）

    Returns:
        Base64编码的签名字符串
    """
    # 1. 构造待签名数据
    payload = f'{imei}:{timestamp}'

    # 2. AES-ECB 加密
    # MODE_ECB = 1
    cipher = ucryptolib.aes(KEY, 1)
    padded_plaintext = pkcs7_pad(payload.encode())
    ciphertext = cipher.encrypt(padded_plaintext)

    # 3. Base64编码
    return base64.b2a_base64(ciphertext, newline=False).decode()

# 使用示例
imei = '860602069165352'
timestamp = '1780552892'
signature = generate_signature(imei, timestamp)
print(f"Signature: {signature}")

请求示例

完整请求示例

请求地址:

POST https://bmsapi.20783378.com/api/mcu/client/connect

请求头:

Content-Type: application/json
X-Timestamp: 1780552892
X-Signature: m7mbGLq5i4/qY/S1qWaUiRce38olY+rPqp54tpgXFOM=

请求体:

{
  "imei": "860602069165352"
}

签名计算示例

输入参数:

• IMEI: 860602069165352
• Timestamp: 1780552892

待签名数据:

860602069165352:1780552892

签名结果（Base64编码）:

m7mbGLq5i4/qY/S1qWaUiRce38olY+rPqp54tpgXFOM=

说明:

• 不同时间戳会产生不同的签名值
• 签名值长度为64字符（48字节密文Base64编码后）

响应结果

成功响应

{
    "code": 200,
    "type": 0,
    "data": "860602069165352,mcu,emqxbms.gs.dogwof.com,1885,860602069165352,UUJA6ILXCdwzmMVhOumglZZ5oqHPTFth,1780552892497",
    "msg": "OK",
    "time": 1780552892497
}

响应参数说明

MQTT连接参数

data字段格式：clientId,productKey,host,port,userName,password,time

失败响应

{
    "code": 401,
    "type": 1,
    "data": "",
    "msg": "签名验证失败",
    "time": 1705661910697
}

常见错误:

• 缺少必要的签名头信息: 请求头缺少 X-Timestamp 或 X-Signature
• 签名格式错误（Base64解码失败）: 签名不是有效的Base64字符串
• 签名验证失败: 签名与请求参数不匹配

MQTT通信说明

连接参数

鉴权成功后，使用返回的 data 字段解析出的参数连接 MQTT Broker：

# 解析 data 字段
params = data.split(',')
client_id  = params[0]   # clientId
product_key = params[1]  # productKey（固定为 mcu）
mqtt_host  = params[2]   # host
mqtt_port  = int(params[3])  # port
username   = params[4]   # userName
password   = params[5]   # password

MQTT Topic 定义

MCU设备使用以下 Topic 进行通信：

其中 {device_uuid} 为设备的 IMEI 号（15位ASCII字符串）。

说明：心跳走 /user/heart 自发自收（纯MQTT保活），数据上报走 /user/update。EMQX规则只需 /mcu/+/user/update。

MQTT 连接示例

Python (paho-mqtt)

import paho.mqtt.client as mqtt

def on_connect(client, userdata, flags, rc):
    print(f"连接结果: {rc}")
    if rc == 0:
        # 连接成功，订阅指令下发 Topic
        topic = f"/mcu/{client_id}/user/get"
        client.subscribe(topic)
        print(f"已订阅: {topic}")

def on_message(client, userdata, msg):
    print(f"收到消息: topic={msg.topic}, payload={msg.payload}")

# 使用鉴权返回的参数
client_id = "860602069165352"
mqtt_host = "emqxbms.gs.dogwof.com"
mqtt_port = 1885
username = "860602069165352"
password = "UUJA6ILXCdwzmMVhOumglZZ5oqHPTFth"

client = mqtt.Client(client_id=client_id)
client.username_pw_set(username, password)
client.on_connect = on_connect
client.on_message = on_message

client.connect(mqtt_host, mqtt_port, keepalive=60)
client.loop_forever()

MQTTS(TLS)连接（推荐）

设备需要 TLS 加密通信时，鉴权请求携带 use_mqtts: true，服务端返回 MQTTS(MQTT over TLS) 的地址和端口（其余参数 clientId/userName/password 等与普通模式完全一致，复用即可）：

鉴权请求示例（启用 MQTTS）：

{
  "imei": "860602069165352",
  "use_mqtts": true
}

返回的 data 示例（host/port 为 MQTTS 地址，其余字段不变）：

860602069165352,mcu,emqxbms-tls.20783378.com,8885,860602069165352,UUJA6ILXCdwzmMVhOumglZZ5oqHPTFth,1780552892497

MQTTS 连接示例（paho-mqtt）

import ssl
import paho.mqtt.client as mqtt

# use_mqtts=true 鉴权返回的 MQTTS 参数（host/port 不同，其余复用）
mqtt_host = "emqxbms-tls.20783378.com"
mqtt_port = 8885

client = mqtt.Client(client_id=client_id)
client.username_pw_set(username, password)

# TLS 配置：校验服务端证书
client.tls_set(cert_reqs=ssl.CERT_REQUIRED)
client.tls_insecure_set(False)  # True=跳过主机名校验，生产环境建议 False

client.connect(mqtt_host, mqtt_port, keepalive=60)
client.loop_forever()

心跳机制

设备连接成功后，建议定期向 /mcu/{device_uuid}/user/heart 发送心跳消息，维持连接活跃状态。

测试工具

使用 curl 测试

# 计算签名（使用上面的Python代码生成）
TIMESTAMP=1780552892
IMEI="860602069165352"
SIGNATURE="m7mbGLq5i4/qY/S1qWaUiRce38olY+rPqp54tpgXFOM="

# 发送请求
curl -X POST "https://bmsapi.20783378.com/api/mcu/client/connect" \
  -H "Content-Type: application/json" \
  -H "X-Timestamp: $TIMESTAMP" \
  -H "X-Signature: $SIGNATURE" \
  -d '{"imei": "'$IMEI'"}'

Python 完整测试脚本（鉴权 + MQTT连接）

import requests
import time
import base64
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import padding

# 配置
API_URL = "https://bmsapi.20783378.com/api/mcu/client/connect"
KEY = b'a3f1b8c9d2e47056'  # 16字节ASCII字符串
IMEI = "860602069165352"

def pkcs7_pad(data: bytes, block_size: int = 16) -> bytes:
    padder = padding.PKCS7(block_size * 8).padder()
    return padder.update(data) + padder.finalize()

def generate_signature(imei: str, timestamp: str) -> str:
    payload = f'{imei}:{timestamp}'
    cipher = Cipher(algorithms.AES(KEY), modes.ECB(), backend=default_backend())
    encryptor = cipher.encryptor()
    padded_plaintext = pkcs7_pad(payload.encode('utf-8'))
    ciphertext = encryptor.update(padded_plaintext) + encryptor.finalize()
    return base64.b64encode(ciphertext).decode('utf-8')

# Step 1: 设备鉴权
print("=== Step 1: 设备鉴权 ===")
timestamp = str(int(time.time()))
signature = generate_signature(IMEI, timestamp)

headers = {
    "Content-Type": "application/json",
    "X-Timestamp": timestamp,
    "X-Signature": signature
}

response = requests.post(API_URL, json={"imei": IMEI}, headers=headers)
print(f"Status: {response.status_code}")
result = response.json()
print(f"Response: {result}")

if result.get("code") == 200:
    # Step 2: 解析MQTT连接参数
    print("\n=== Step 2: 解析MQTT连接参数 ===")
    params = result["data"].split(",")
    print(f"clientId:    {params[0]}")
    print(f"productKey:  {params[1]}")
    print(f"host:        {params[2]}")
    print(f"port:        {params[3]}")
    print(f"userName:    {params[4]}")
    print(f"password:    {params[5]}")
    print(f"time:        {params[6]}")
else:
    print(f"鉴权失败: {result.get('msg')}")