设备证书认证

# Windows PowerShell
certutil -dump "证书文件名"

# Linux && macOS Shell
openssl x509 -in "证书文件名" -noout -dates

# 查看输出结果
...
notBefore=    # 证书有效起始时间
notAfter=     # 证书有效结束时间
...

# 示例
# 腾讯云证书名称：ca.crt
# 设备证书名称：certificate_authentication_device_cert.crt

# Windows PowerShell
certutil -dump ca.crt
certutil -dump certificate_authentication_device_cert.crt

# Linux && macOS Shell
openssl x509 -in ca.crt -noout -dates
openssl x509 -in certificate_authentication_device_cert.crt -noout -dates

# 查看输出结果
...
notBefore=    # 证书有效起始时间
notAfter=     # 证书有效结束时间
...

Client ID = ProductID + DeviceName

Username = ClientID + ";12010126;" + connid + ";" + expiry

Password = 任意非空字符串

ProductID  = testProductID
DeviceName = testDeviceName
connid     = A1B2C
expiry     = 1786610239

Client ID  = testProductIDtestDeviceName
Username   = testProductIDtestDeviceName;12010126;A1B2C;1786610239
Password   = certificate

# Windows powershell
python -m pip install "paho-mqtt>=2.1,<3.0"

# Linux && macOS shell
python3 -m pip install "paho-mqtt>=2.1,<3.0"

# Windows PowerShell
python -m pip show paho-mqtt

# Linux && macOS Shell
python3 -m pip show paho-mqtt

# 输出中的 Version 应为 2.1.0。

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

import secrets
import ssl
import string
import sys
import time
from pathlib import Path

try:
    import paho.mqtt.client as mqtt
except ImportError:
    mqtt = None


SDK_APP_ID = "12010126"

# 凭证只能从脚本同级的cert目录中读取。
SCRIPT_DIR = Path(__file__).resolve().parent
CERT_DIR = (SCRIPT_DIR / "cert").resolve()


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


def get_cert_file(prompt):
    """根据用户输入的文件名，从cert目录中读取凭证文件。"""
    filename = required_input(prompt)

    # 只允许输入文件名，不允许输入目录或绝对路径。
    if (
        Path(filename).is_absolute()
        or "/" in filename
        or "\\\\" in filename
        or filename in {".", ".."}
    ):
        raise ValueError(
            "只允许输入cert目录中的文件名，不能包含路径"
        )

    file_path = (CERT_DIR / filename).resolve()

    # 防止通过符号链接等方式访问cert目录外的文件。
    try:
        file_path.relative_to(CERT_DIR)
    except ValueError as exc:
        raise ValueError(
            "凭证文件必须位于脚本同级的cert目录中"
        ) from exc

    if not file_path.is_file():
        raise FileNotFoundError(
            f"在cert目录中未找到文件：{filename}"
        )

    try:
        with file_path.open("rb") as file:
            file.read(1)
    except OSError as exc:
        raise PermissionError(
            f"无法读取凭证文件：{filename}"
        ) from exc

    return file_path


def generate_connid(length=5):
    """生成用于区分MQTT连接的随机字符串。"""
    alphabet = string.ascii_uppercase + string.digits
    return "".join(secrets.choice(alphabet) for _ in range(length))


def main():
    if mqtt is None:
        raise ImportError(
            '未安装paho-mqtt，请执行：'
            'python3 -m pip install "paho-mqtt>=2.1,<3.0"'
        )

    if not CERT_DIR.is_dir():
        raise FileNotFoundError(
            f"未找到cert目录，请创建：{CERT_DIR}"
        )

    print("请输入cert目录中的凭证文件名。")
    print(f"凭证目录：{CERT_DIR}\\n")

    ca_file = get_cert_file("CA证书文件名：")
    device_cert_file = get_cert_file("设备证书文件名：")
    device_key_file = get_cert_file("设备私钥文件名：")

    print("\\n已加载以下凭证文件：")
    print(f"- CA证书：{ca_file.name}")
    print(f"- 设备证书：{device_cert_file.name}")
    print(f"- 设备私钥：{device_key_file.name}")

    broker = required_input("\\nMQTT服务器地址：")

    port_text = input("端口号（直接回车使用8883）：").strip()
    port = int(port_text) if port_text else 8883

    product_id = required_input("ProductID：")
    device_name = required_input("DeviceName：")

    client_id = f"{product_id}{device_name}"
    connid = generate_connid()
    expiry = int(time.time()) + 60 * 60

    username = (
        f"{client_id};{SDK_APP_ID};{connid};{expiry}"
    )

    # 证书认证不校验Password内容。
    password = "certificate"

    def on_connect(
        client,
        userdata,
        connect_flags,
        reason_code,
        properties,
    ):
        if reason_code == 0:
            print("\\n认证成功，设备已连接物联网开发平台。")
            print("请前往控制台查看设备在线状态。")
        else:
            print(f"\\nMQTT连接失败，返回码：{reason_code}")
            client.disconnect()

    def on_connect_fail(client, userdata):
        print(
            "\\n网络连接或TLS握手失败，"
            "请检查服务器地址、端口及凭证文件。"
        )

    def on_disconnect(
        client,
        userdata,
        disconnect_flags,
        reason_code,
        properties,
    ):
        if reason_code != 0:
            print(f"\\n连接异常断开，返回码：{reason_code}")

    client = mqtt.Client(
        callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
        client_id=client_id,
        clean_session=True,
        protocol=mqtt.MQTTv311,
    )

    client.username_pw_set(
        username=username,
        password=password,
    )

    client.tls_set(
        ca_certs=str(ca_file),
        certfile=str(device_cert_file),
        keyfile=str(device_key_file),
        cert_reqs=ssl.CERT_REQUIRED,
        tls_version=ssl.PROTOCOL_TLS_CLIENT,
    )

    # 启用服务器证书及域名校验。
    client.tls_insecure_set(False)

    client.on_connect = on_connect
    client.on_connect_fail = on_connect_fail
    client.on_disconnect = on_disconnect

    print(f"\\n正在连接：{broker}:{port}")
    print(f"ClientId：{client_id}")
    print("程序将保持连接，按Ctrl+C退出。")

    client.connect(
        host=broker,
        port=port,
        keepalive=60,
    )

    try:
        client.loop_forever()
    finally:
        client.disconnect()


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)

├─ mqtt_cert_connect.py
└─ cert/
   ├─ ca.crt
   ├─ device.crt
   └─ device.key

# Windows PowerShell
python mqtt_cert_connect.py

# Linux && macOS Shell
python3 mqtt_cert_connect.py

# 运行示例
凭证目录：D:\\test\\shell\\test_cert\\cert

CA证书文件名：ca.crt
设备证书文件名：certificate_authentication_device_cert.crt
设备私钥文件名：certificate_authentication_device_private.key

已加载以下凭证文件：
- CA证书：ca.crt
- 设备证书：certificate_authentication_device_cert.crt
- 设备私钥：certificate_authentication_device_private.key

MQTT服务器地址：ProductID.iotcloud.tencentdevices.com
端口号（直接回车使用8883）：8883
ProductID：ProductID
DeviceName：certificate_authentication_device

正在连接：******.iotcloud.tencentdevices.com:8883
ClientId：******
程序将保持连接，按Ctrl+C退出。

认证成功，设备已连接物联网开发平台。
请前往控制台查看设备在线状态。