MQTT 安全连接

MQTT 安全连接

概述

本节综合介绍 ESP32 通过 TLS 加密连接 MQTT Broker 的完整实现。从前面的章节分别学习了证书、端口、WiFiClientSecure 配置后，本节将它们整合为一个完整的端到端安全连接方案。学习完成后，您将能够：

• 实现完整的 ESP32 TLS MQTT 连接代码
• 理解安全连接的完整数据流
• 实现断线重连和故障恢复
• 展示安全连接的方案优势

前置要求

在开始本节之前，请确保：

• 已完成 ESP32 证书集成
• 已完成 WiFiClientSecure 配置
• Mosquitto TLS 已配置并运行

Complete Code Implementation

完整 TLS MQTT 代码

#include <WiFi.h>
#include <WiFiClientSecure.h>
#include <PubSubClient.h>
#include <ArduinoJson.h>
#include "ca_cert.h"       // CA 根证书

// ---- WiFi 配置 ----
const char* wifi_ssid = "YourWiFiSSID";
const char* wifi_password = "YourWiFiPassword";

// ---- MQTT TLS 配置 ----
const char* mqtt_server = "mqtt.example.com";  // DynDNS 域名
const int mqtt_port = 8883;                    // MQTTS 端口
const char* mqtt_user = "esp32_device";
const char* mqtt_password = "secure_device_pwd";
const char* client_id = "esp32_factory_001";

// ---- Topic 定义 ----
const char* topic_status   = "esp32/status";
const char* topic_data     = "esp32/data";
const char* topic_control  = "esp32/control";

// ---- WiFi + MQTT 客户端 ----
WiFiClientSecure espClient;
PubSubClient client(espClient);

// ---- 系统状态 ----
unsigned long lastReconnectAttempt = 0;
int connectionAttempts = 0;
bool tlsConnected = false;

// ---- 时间同步 ----
#include <time.h>
const char* ntpServer = "pool.ntp.org";
const long gmtOffset_sec = 0;        // UTC+0
const int daylightOffset_sec = 0;

// ============= WiFi 连接 =============

bool connectWiFi() {
  Serial.printf("连接 WiFi: %s\n", wifi_ssid);
  WiFi.begin(wifi_ssid, wifi_password);

  int attempts = 0;
  while (WiFi.status() != WL_CONNECTED && attempts < 20) {
    delay(500);
    Serial.print(".");
    attempts++;
  }

  if (WiFi.status() == WL_CONNECTED) {
    Serial.printf("\nWiFi 已连接, IP: %s\n", WiFi.localIP().toString().c_str());
    return true;
  }

  Serial.println("\nWiFi 连接失败");
  return false;
}

// ============= NTP 时间同步 =============

void syncTime() {
  Serial.print("同步 NTP 时间...");
  configTime(gmtOffset_sec, daylightOffset_sec, ntpServer);

  struct tm timeinfo;
  if (getLocalTime(&timeinfo)) {
    Serial.println("时间同步成功");
    // TLS 证书验证需要正确的时间
  } else {
    Serial.println("时间同步失败 - 证书验证可能失败");
  }
}

// ============= TLS MQTT 连接 =============

bool connectMQTT() {
  // 配置 TLS
  espClient.setCACert(rootCACertificate);
  espClient.setTimeout(15000);

  // 配置 MQTT
  client.setServer(mqtt_server, mqtt_port);
  client.setCallback(mqttCallback);

  Serial.printf("连接 MQTT Broker (TLS): %s:%d\n", mqtt_server, mqtt_port);

  // 尝试连接
  if (client.connect(client_id, mqtt_user, mqtt_password)) {
    Serial.println("MQTT TLS 连接成功 ✅");
    connectionAttempts = 0;
    tlsConnected = true;

    // 发布在线状态
    DynamicJsonDocument doc(128);
    doc["device"] = client_id;
    doc["status"] = "online";
    doc["tls"] = true;
    doc["version"] = "2.1.0";
    char buffer[128];
    serializeJson(doc, buffer);
    client.publish(topic_status, buffer, true);  // 保留消息

    // 订阅控制 Topic
    client.subscribe(topic_control);

    return true;
  }

  Serial.printf("MQTT 连接失败, rc=%d\n", client.state());
  connectionAttempts++;
  tlsConnected = false;
  return false;
}

// ============= MQTT 回调 =============

void mqttCallback(char* topic, byte* payload, unsigned int length) {
  Serial.printf("收到消息 [%s]: ", topic);

  // 解析 JSON
  DynamicJsonDocument doc(256);
  DeserializationError error = deserializeJson(doc, payload, length);

  if (error) {
    Serial.println("JSON 解析失败");
    return;
  }

  // 处理控制命令
  if (strcmp(topic, topic_control) == 0) {
    const char* command = doc["command"];

    if (strcmp(command, "restart") == 0) {
      Serial.println("收到重启命令");
      client.publish(topic_status, "{\"status\": \"restarting\"}");
      delay(100);
      ESP.restart();

    } else if (strcmp(command, "ping") == 0) {
      client.publish(topic_status, "{\"status\": \"pong\"}");

    } else if (strcmp(command, "update_cert") == 0) {
      Serial.println("收到证书更新命令");
      // 证书更新逻辑
    }
  }

  Serial.println();
}

// ============= 数据发布 =============

void publishSensorData() {
  DynamicJsonDocument doc(256);

  doc["device"] = client_id;
  doc["temperature"] = 24.5;     // 示例数据
  doc["humidity"] = 65.2;
  doc["rssi"] = WiFi.RSSI();
  doc["heap"] = ESP.getFreeHeap();
  doc["uptime"] = millis() / 1000;

  char buffer[256];
  serializeJson(doc, buffer);

  if (client.publish(topic_data, buffer)) {
    Serial.println("传感器数据已发送 (TLS 加密)");
  }
}

// ============= 断线重连 =============

void maintainConnection() {
  if (!client.connected()) {
    tlsConnected = false;

    unsigned long now = millis();
    // 每 5 秒尝试一次重连
    if (now - lastReconnectAttempt > 5000) {
      lastReconnectAttempt = now;

      if (WiFi.status() != WL_CONNECTED) {
        Serial.println("WiFi 已断开，重新连接...");
        connectWiFi();
      }

      // 指数退避: 重试次数越多，等待越长
      if (connectionAttempts < 5 || now - lastReconnectAttempt > 30000) {
        connectMQTT();
      }
    }
  }
}

// ============= Setup & Loop =============

void setup() {
  Serial.begin(115200);
  Serial.println("\n===== ESP32 MQTT TLS 示例 =====");

  // 连接 WiFi
  connectWiFi();

  // 同步时间（TLS 证书验证需要）
  syncTime();

  // 连接 MQTT TLS
  connectMQTT();
}

void loop() {
  maintainConnection();

  if (client.connected()) {
    client.loop();

    // 每 30 秒发布一次传感器数据
    static unsigned long lastPublish = 0;
    if (millis() - lastPublish > 30000) {
      publishSensorData();
      lastPublish = millis();
    }
  }
}

Data Flow

加密数据传输流程

ESP32                                      MQTT Broker (Mosquitto)
  │                                              │
  │  1. WiFi 连接                                │
  │  ──────────────────────────────────────────►  │
  │                                              │
  │  2. NTP 时间同步                              │
  │  (正确时间是证书验证的前提)                     │
  │                                              │
  │  3. TCP 连接 → 端口 8883                      │
  │  ──────────────────────────────────────────►  │
  │                                              │
  │  4. TLS 握手 (加密协商)                       │
  │  ◄── 服务器证书 (fullchain.pem) ────────────  │
  │  ── 验证证书 (CA 证书) ──────────────────►    │
  │  ── 生成会话密钥 ──────────────────────────►  │
  │  ◄── 握手完成 ─────────────────────────────  │
  │  ═══ 从此所有数据加密 ═══════════════════════  │
  │                                              │
  │  5. MQTT CONNECT (用户名密码加密传输)          │
  │  ═══════════════════════════════════════════►  │
  │                                              │
  │  6. MQTT PUBLISH (传感器数据加密传输)          │
  │  ═══════════════════════════════════════════►  │
  │                                              │
  │  7. MQTT SUBSCRIBE (控制命令加密接收)         │
  │  ◄══════════════════════════════════════════  │

验证

验证安全连接

# 1. 查看 ESP32 串口输出
# 预期输出:
# WiFi 已连接, IP: 192.168.1.100
# 时间同步成功
# MQTT TLS 连接成功 ✅
# 传感器数据已发送 (TLS 加密)
# 收到消息 [esp32/control]: {"command":"ping"}

# 2. 在 Broker 端查看日志
docker logs mosquitto | grep -E "New connection|Client connected"
# 1726425608: New connection from 192.168.1.100:54321 on port 8883.
# 1726425609: New client connected as esp32_factory_001 (p2, c1, k60).
# 1726425609: Client esp32_factory_001 connected (TLS1.3, AES-256-GCM).

# 3. 验证 Wireshark（下一节详述）

Pre-sales Key Points

方案价值展示

功能	价值	买家演示要点
TLS 加密	数据安全	”所有传感器数据和控制指令都加密传输”
断线重连	可靠性	”WiFi 断开后自动重连”
安全状态	可监控	”设备上报连接状态和加密状态”
远程控制	远程管理	”可通过加密 MQTT 通道远程控制设备”

Summary

本节实现了完整的 ESP32 TLS MQTT 安全连接：

1. 完整代码：WiFi 连接 + NTP 时间同步 + TLS 配置 + MQTT 连接
2. 数据流程：TCP 连接 → TLS 握手 → MQTT CONNECT → 加密数据传输
3. 断线重连：指数退避策略，WiFi 断开自动恢复
4. 状态上报：设备在线状态和 TLS 连接状态通过 MQTT 发布
5. 远程控制：加密通道接收控制命令