MQTT 协议详解：从发布订阅模型到国密改造的完整架构解析

Client                          Broker
  │                               │
  │──── CONNECT ────────────────►│  // 建立连接，携带 Client ID
  │                               │
  │◄──── CONNACK ───────────────│  // 连接确认（rc=0 表示成功）
  │                               │
  │──── SUBSCRIBE ─────────────►│  // 订阅 Topic（可多次）
  │◄──── SUBACK ────────────────│  // 订阅确认
  │                               │
  │──── PUBLISH ────────────────►│  // 发布消息
  │◄──── PUBACK / PUBREC ───────│  // QoS 1/2 确认
  │                               │
  │──── PINGREQ ────────────────►│  // 心跳保持
  │◄──── PINGRESP ──────────────│
  │                               │
  │──── DISCONNECT ─────────────►│  // 主动断开
  │                               │

┌─────────────────────────────────────────────────────┐
│                  物联网云平台                        │
│  ┌─────────────┐    国密 TLS     ┌─────────────┐   │
│  │  Broker     │◄───────────────►│  设备 A     │   │
│  │ (EMQX/      │   SM2证书认证    │  (传感器)   │   │
│  │  VerneMQ)   │                 └─────────────┘   │
│  └──────┬──────┘                                   │
│         │ 内部通信（可选国密）                        │
│  ┌──────▼──────┐                                   │
│  │  Broker 集群 │                                   │
│  └──────┬──────┘                                   │
│         │                                           │
│  ┌──────▼──────┐    国密 TLS     ┌─────────────┐   │
│  │  设备 B     │◄───────────────►│  监控中心   │   │
│  │  (执行器)   │   SM2证书认证    │  (Web界面)  │   │
│  └─────────────┘                 └─────────────┘   │
└─────────────────────────────────────────────────────┘

# emqx.conf 国密 TLS 配置（EMQX v4.x格式）
listener.ssl.external {
  bind = "0.0.0.0:8883";
  protocol = mqtt;
  ssl_options {
    keyfile = "/path/to/server-key.pem";      # SM2 私钥
    certfile = "/path/to/server-cert.pem";    # SM2 服务器证书
    cacertfile = "/path/to/ca-cert.pem";      # 国密 CA 证书
    depth = 3;
    verify = verify_peer;
    fail_if_no_peer_cert = true;
    # 强制使用国密套件
    ciphers = ["TLS_SM4_GCM_SM3"];
    # 曲线要求：必须使用 SM2 标准曲线（sm2p256v1），禁止混入 NIST P-256
    curves = ["sm2p256v1"];
  }
}

import paho.mqtt.client as mqtt
from gmssl import sm2, sm3, func
from cryptography import x509
from cryptography.hazmat.primitives import serialization
import ssl

# 加载国密证书（SM2）
with open('client-cert.pem', 'rb') as f:
    client_cert = x509.load_pem_x509_certificate(f.read())

with open('client-key.pem', 'rb') as f:
    client_key = serialization.load_pem_private_key(f.read(), password=None)

# 创建 MQTT 客户端
client = mqtt.Client(
    client_id="device_001",
    protocol=mqtt.MQTTv5,
    callback_api_version=mqtt.CallbackAPIVersion.VERSION2
)

# 配置国密 TLS（强制 TLS 1.3）
client.tls_set(
    certfile='client-cert.pem',
    keyfile='client-key.pem',
    ca_certs='ca-cert.pem',
    tls_version=ssl.TLSVersion.TLSv1_3,  # 强制 TLS 1.3（非默认的 ssl.PROTOCOL_TLS）
    ciphers='TLS_SM4_GCM_SM3'
)

# 设置认证回调
def on_connect(client, userdata, flags, rc, properties=None):
    if rc == 0:
        print("国密 MQTT 连接成功")
        client.subscribe("building/1/floor/2/temperature")
    else:
        print(f"连接失败: {rc}")

client.on_connect = on_connect
client.connect("mqtt.example.com", port=8883, keepalive=60)
client.loop_start()

┌─────────────────────────────────────────┐
│           MQTT Broker                    │
│                                         │
│  ┌─────────────┐    ┌─────────────┐   │
│  │  TLS 握手   │───►│  证书验签   │   │
│  │  (国密SM2)  │    │  (SM2验证)  │   │
│  └─────────────┘    └──────┬──────┘   │
│                            │           │
│                    ┌───────▼───────┐   │
│                    │  ACL 权限检查  │   │
│                    │  Topic 匹配   │   │
│                    └───────┬───────┘   │
│                            │           │
│                    ┌───────▼───────┐   │
│                    │  消息转发     │   │
│                    └───────────────┘   │
└─────────────────────────────────────────┘

# 允许设备发布到自己的 Topic
allow {
  action = publish;
  topic = {
    match = "device/${clientid}/#";
  };
}
# 允许订阅系统 Topic
allow {
  action = subscribe;
  topic = {
    match = "system/+";
  };
}
# 拒绝其他所有操作
deny all;