MQTT en Production : Architecture Sécurisée & Scalable

Guide complet pour déployer MQTT en environnement de production avec haute disponibilité, sécurité renforcée et observabilité intégrée.

1. Architecture de Référence

1.1 Stack Complète Device-to-Cloud

┌─────────────────┐
│   Devices IoT   │  ← TLS 1.3, mTLS, certificats X.509
│  (MQTT Client)  │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│  Load Balancer  │  ← HAProxy/NGINX avec health checks
│   (TLS Term.)   │
└────────┬────────┘
         │
         ▼
┌─────────────────────────────────┐
│    MQTT Broker Cluster          │
│  ┌──────┐  ┌──────┐  ┌──────┐  │
│  │Broker│  │Broker│  │Broker│  │  ← Mosquitto / EMQX / VerneMQ
│  │  1   │  │  2   │  │  3   │  │
│  └──┬───┘  └──┬───┘  └──┬───┘  │
│     └─────────┼─────────┘       │
│               │                 │
│     ┌─────────▼─────────┐       │
│     │   Redis Cluster   │       │  ← Session/state replication
│     └───────────────────┘       │
└─────────────┬───────────────────┘
              │
              ▼
┌─────────────────────────────────┐
│    Rules Engine / Router        │  ← Node-RED, Kafka Streams, Flink
└─────────────┬───────────────────┘
              │
              ▼
┌─────────────────────────────────┐
│    Time-Series Database         │  ← InfluxDB, TimescaleDB, QuestDB
└─────────────┬───────────────────┘
              │
              ▼
┌─────────────────────────────────┐
│    API & Dashboard Layer        │  ← Grafana, custom APIs
└─────────────────────────────────┘

1.2 Choix du Broker : Comparatif Technique

2. Sécurité : Defense in Depth

2.1 Authentification Multi-Niveaux

La sécurité MQTT repose sur plusieurs couches complémentaires :

1. TLS/mTLS (Transport Layer) : Chiffrement + authentification certificat
2. Username/Password (Application Layer) : Credential management
3. Client ID Policy : Validation format + unicité
4. ACL (Authorization Layer) : Contrôle pub/sub par topic

2.2 Configuration TLS Durcie (Mosquitto)

# mosquitto.conf - Production hardened
listener 8883
protocol mqtt

# TLS Configuration
cafile /etc/mosquitto/ca_certificates/ca.crt
certfile /etc/mosquitto/certs/server.crt
keyfile /etc/mosquitto/certs/server.key

# Force TLS 1.3 only
tls_version tlsv1.3

# Cipher suites (modern, forward secrecy)
ciphers TLS_AES_256_GCM_SHA384:TLS_CHACHA20_POLY1305_SHA256:TLS_AES_128_GCM_SHA256

# Client certificate requirement
require_certificate true
use_identity_as_username true

# Session & connection limits
max_connections 10000
max_queued_messages 1000
max_inflight_messages 20
max_keepalive 3600

# Persistence
persistence true
persistence_location /var/lib/mosquitto/
autosave_interval 300

# Logging
log_dest file /var/log/mosquitto/mosquitto.log
log_type error
log_type warning
log_type notice
log_type information
connection_messages true
log_timestamp true

# ACL
acl_file /etc/mosquitto/acl.conf

# Security plugins
auth_plugin /usr/lib/mosquitto_auth_plugin.so
auth_opt_backends jwt
auth_opt_jwt_secret your-secret-key-here

2.3 Gestion des Certificats : PKI Automation

#!/bin/bash
# Script de génération de certificats device (production)

DEVICE_ID=$1
CA_DIR="/etc/pki/ca"
DEVICE_DIR="/etc/pki/devices/${DEVICE_ID}"

mkdir -p "${DEVICE_DIR}"

# Generate device private key (ECC pour l'efficacité)
openssl ecparam -genkey -name prime256v1 -out "${DEVICE_DIR}/device.key"

# Generate CSR
openssl req -new -key "${DEVICE_DIR}/device.key" \
  -out "${DEVICE_DIR}/device.csr" \
  -subj "/CN=${DEVICE_ID}/O=YourOrg/OU=IoT Devices"

# Sign with CA (validity 1 year)
openssl x509 -req -in "${DEVICE_DIR}/device.csr" \
  -CA "${CA_DIR}/ca.crt" \
  -CAkey "${CA_DIR}/ca.key" \
  -CAcreateserial \
  -out "${DEVICE_DIR}/device.crt" \
  -days 365 \
  -sha256

# Create PKCS12 bundle (optional, for some clients)
openssl pkcs12 -export \
  -in "${DEVICE_DIR}/device.crt" \
  -inkey "${DEVICE_DIR}/device.key" \
  -out "${DEVICE_DIR}/device.p12" \
  -password pass:device-password

# Verify certificate
openssl verify -CAfile "${CA_DIR}/ca.crt" "${DEVICE_DIR}/device.crt"

echo "✓ Certificate generated for device: ${DEVICE_ID}"
echo "  Cert: ${DEVICE_DIR}/device.crt"
echo "  Key:  ${DEVICE_DIR}/device.key"

2.4 ACL Granulaires : Principe du Moindre Privilège

# /etc/mosquitto/acl.conf
# ACL par groupe de devices

# Admin users (full access)
user admin
topic readwrite #

# Telemetry devices (sensors)
pattern read devices/%u/cmd/#
pattern write devices/%u/telemetry/#
pattern write devices/%u/status

# Actuators (commands)
pattern read devices/%u/cmd/#
pattern write devices/%u/response/#
pattern write devices/%u/status

# Gateway devices (bridge)
user gateway-01
topic readwrite devices/+/telemetry/#
topic readwrite devices/+/status
topic read devices/+/cmd/#

# Monitoring service (read-only)
user monitoring
topic read $SYS/#
topic read devices/+/telemetry/#
topic read devices/+/status

3. Haute Disponibilité & Clustering

3.1 Architecture Cluster EMQX (Exemple)

# emqx.conf - Cluster configuration
cluster:
  name: emqx-prod
  discovery_strategy: etcd
  etcd:
    servers: "http://etcd1:2379,http://etcd2:2379,http://etcd3:2379"
    prefix: "emqx"
    node_ttl: 60s

node:
  process_limit: 2097152
  max_ports: 1048576
  dist_buffer_size: 128MB

listener:
  ssl:
    external:
      bind: "0.0.0.0:8883"
      max_connections: 500000
      acceptors: 64
      access_rules: ["allow all"]
      
      # TLS options
      tls_versions: "tlsv1.3,tlsv1.2"
      ciphers: "TLS_AES_256_GCM_SHA384,TLS_CHACHA20_POLY1305_SHA256"
      
      # Client cert
      verify: verify_peer
      fail_if_no_peer_cert: true
      cacertfile: "/etc/emqx/certs/ca.crt"
      certfile: "/etc/emqx/certs/server.crt"
      keyfile: "/etc/emqx/certs/server.key"

mqtt:
  max_packet_size: 1MB
  max_qos_allowed: 2
  max_topic_levels: 128
  max_topic_alias: 65535
  retain_available: true
  session_expiry_interval: 7200
  max_awaiting_rel: 1000
  max_inflight: 32

zones:
  external:
    max_subscriptions: 10
    upgrade_qos: false
    max_mqueue_len: 10000
    mqueue_store_qos0: false

3.2 Load Balancer : HAProxy Configuration

# haproxy.cfg - MQTT load balancing
global
    log /dev/log local0
    maxconn 100000
    ssl-default-bind-ciphers TLS_AES_256_GCM_SHA384:TLS_CHACHA20_POLY1305_SHA256
    ssl-default-bind-options ssl-min-ver TLSv1.2 no-tls-tickets

defaults
    log global
    mode tcp
    option tcplog
    option dontlognull
    timeout connect 5000ms
    timeout client 3600000ms
    timeout server 3600000ms

frontend mqtt_frontend
    bind *:8883
    mode tcp
    option tcplog
    default_backend mqtt_backend

backend mqtt_backend
    mode tcp
    balance leastconn
    option tcp-check
    
    # Health check MQTT CONNECT
    tcp-check connect port 8883 ssl
    tcp-check send-binary 101400044d515454040200003c # MQTT CONNECT packet
    tcp-check expect binary 20020000 # CONNACK success
    
    server mqtt1 10.0.1.10:8883 check inter 10s fall 3 rise 2 ssl verify required ca-file /etc/ssl/ca.crt
    server mqtt2 10.0.1.11:8883 check inter 10s fall 3 rise 2 ssl verify required ca-file /etc/ssl/ca.crt
    server mqtt3 10.0.1.12:8883 check inter 10s fall 3 rise 2 ssl verify required ca-file /etc/ssl/ca.crt

listen stats
    bind *:8404
    mode http
    stats enable
    stats uri /stats
    stats refresh 30s
    stats auth admin:your-secure-password

4. Observabilité & Monitoring

4.1 Métriques Critiques

4.2 Prometheus Exporter & Alerting

# prometheus.yml - MQTT monitoring
scrape_configs:
  - job_name: 'mosquitto'
    static_configs:
      - targets: ['mqtt1:9234', 'mqtt2:9234', 'mqtt3:9234']
    
  - job_name: 'emqx'
    static_configs:
      - targets: ['mqtt1:8081', 'mqtt2:8081', 'mqtt3:8081']
    metrics_path: '/api/v4/metrics'

# Alert rules
groups:
  - name: mqtt_alerts
    interval: 30s
    rules:
      - alert: MQTTHighConnectionCount
        expr: mqtt_broker_clients_connected > 80000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "MQTT broker nearing connection limit"
          description: "{{ $labels.instance }} has {{ $value }} connected clients"
      
      - alert: MQTTHighMessageQueue
        expr: mqtt_broker_messages_inflight > 50000
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "MQTT message queue backup"
          description: "{{ $labels.instance }} has {{ $value }} messages in flight"
      
      - alert: MQTTHighAuthFailureRate
        expr: rate(mqtt_broker_auth_failures_total[1m]) > 100
        for: 3m
        labels:
          severity: warning
        annotations:
          summary: "High authentication failure rate"
          description: "{{ $labels.instance }} seeing {{ $value }} auth failures/sec"

4.3 Logging Structuré avec OpenTelemetry

# Python client avec OpenTelemetry tracing
import paho.mqtt.client as mqtt
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# Setup tracing
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)
span_processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="otel-collector:4317"))
trace.get_tracer_provider().add_span_processor(span_processor)

def on_connect(client, userdata, flags, rc):
    with tracer.start_as_current_span("mqtt_connect") as span:
        span.set_attribute("mqtt.client_id", client._client_id.decode())
        span.set_attribute("mqtt.return_code", rc)
        if rc == 0:
            span.set_attribute("mqtt.status", "connected")
            client.subscribe("devices/+/telemetry/#", qos=1)
        else:
            span.set_attribute("mqtt.status", "failed")
            span.record_exception(Exception(f"Connection failed: {rc}"))

def on_message(client, userdata, msg):
    with tracer.start_as_current_span("mqtt_message_received") as span:
        span.set_attribute("mqtt.topic", msg.topic)
        span.set_attribute("mqtt.payload_size", len(msg.payload))
        span.set_attribute("mqtt.qos", msg.qos)
        
        # Process message
        try:
            process_telemetry(msg.topic, msg.payload)
            span.set_attribute("processing.status", "success")
        except Exception as e:
            span.set_attribute("processing.status", "failed")
            span.record_exception(e)

client = mqtt.Client(client_id="processor-01")
client.tls_set(ca_certs="/etc/pki/ca.crt",
               certfile="/etc/pki/client.crt",
               keyfile="/etc/pki/client.key")
client.on_connect = on_connect
client.loop_forever()

5.3 Reconnection avec Backoff Exponentiel

// Firmware ESP32 - Reconnection robuste
#include <WiFi.h>
#include <PubSubClient.h>

WiFiClientSecure espClient;
PubSubClient mqttClient(espClient);

const int MAX_RECONNECT_DELAY = 300000; // 5 minutes max
int reconnectDelay = 1000; // Start at 1 second

void reconnectMQTT() {
    int attempts = 0;
    
    while (!mqttClient.connected()) {
        Serial.printf("Attempting MQTT connection (attempt %d)...\\n", ++attempts);
        
        if (mqttClient.connect("device-esp32-001", 
                               "devices/esp32-001/status",  // LWT topic
                               1,                            // QoS
                               true,                         // Retain
                               "{\"status\":\"offline\"}")) { // LWT payload
            
            Serial.println("Connected to MQTT broker");
            
            // Reset delay on successful connection
            reconnectDelay = 1000;
            
            // Publish online status
            mqttClient.publish("devices/esp32-001/status", 
                             "{\"status\":\"online\",\"fw\":\"1.0.2\"}", 
                             true);
            
            // Subscribe to command topics
            mqttClient.subscribe("devices/esp32-001/cmd/#");
            
            break;
            
        } else {
            Serial.printf("Failed, rc=%d\\n", mqttClient.state());
            Serial.printf("Waiting %d ms before retry...\\n", reconnectDelay);
            
            delay(reconnectDelay);
            
            // Exponential backoff with jitter
            reconnectDelay = min(reconnectDelay * 2 + random(1000), MAX_RECONNECT_DELAY);
        }
    }
}

void loop() {
    if (!mqttClient.connected()) {
        reconnectMQTT();
    }
    mqttClient.loop();
    
    // Your application logic here
}

6. Performance & Optimisation

6.1 Tuning OS (Linux)

# /etc/sysctl.d/99-mqtt-tuning.conf
# Network tuning for high-throughput MQTT

# Increase max number of open files
fs.file-max = 2097152

# TCP tuning
net.core.somaxconn = 65535
net.core.netdev_max_backlog = 65535
net.ipv4.tcp_max_syn_backlog = 65535

# TCP window sizes
net.core.rmem_max = 134217728
net.core.wmem_max = 134217728
net.ipv4.tcp_rmem = 4096 87380 67108864
net.ipv4.tcp_wmem = 4096 65536 67108864

# TCP optimization
net.ipv4.tcp_fin_timeout = 30
net.ipv4.tcp_keepalive_time = 600
net.ipv4.tcp_keepalive_intvl = 60
net.ipv4.tcp_keepalive_probes = 3

# Enable TCP fast open
net.ipv4.tcp_fastopen = 3

# Congestion control
net.ipv4.tcp_congestion_control = bbr

# Apply: sudo sysctl -p /etc/sysctl.d/99-mqtt-tuning.conf

6.2 Benchmarking : MQTT Stresser

#!/bin/bash
# MQTT Load Testing Script

BROKER="mqtt.example.com"
PORT="8883"
NUM_PUBLISHERS=1000
NUM_SUBSCRIBERS=100
MESSAGE_RATE=10  # messages per second per publisher
DURATION=300     # 5 minutes

echo "Starting MQTT load test..."
echo "Broker: ${BROKER}:${PORT}"
echo "Publishers: ${NUM_PUBLISHERS}"
echo "Subscribers: ${NUM_SUBSCRIBERS}"
echo "Message rate: ${MESSAGE_RATE} msg/s/publisher"
echo "Duration: ${DURATION}s"

# Using mqtt-benchmark tool
mqtt-benchmark \\
    --broker "ssl://${BROKER}:${PORT}" \\
    --ca-cert /etc/pki/ca.crt \\
    --cert /etc/pki/client.crt \\
    --key /etc/pki/client.key \\
    --count ${NUM_PUBLISHERS} \\
    --subscribers ${NUM_SUBSCRIBERS} \\
    --size 256 \\
    --qos 1 \\
    --rate ${MESSAGE_RATE} \\
    --duration ${DURATION} \\
    --topic "load-test/devices/%i/data" \\
    --format json \\
    --output results.json

# Analyze results
python3 analyze_results.py results.json

6.3 Résultats de Benchmark (Exemple)

7. Disaster Recovery & Business Continuity

7.1 Stratégie de Backup

#!/bin/bash
# MQTT Broker Backup Script

BACKUP_DIR="/backup/mqtt"
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
BACKUP_PATH="${BACKUP_DIR}/backup_${TIMESTAMP}"

mkdir -p "${BACKUP_PATH}"

# Backup broker configuration
cp -r /etc/mosquitto "${BACKUP_PATH}/config"

# Backup ACL and user database
cp /etc/mosquitto/acl.conf "${BACKUP_PATH}/"
cp /etc/mosquitto/passwd "${BACKUP_PATH}/"

# Backup persistence data (retained messages, subscriptions)
systemctl stop mosquitto
cp -r /var/lib/mosquitto "${BACKUP_PATH}/persistence"
systemctl start mosquitto

# Backup certificates
cp -r /etc/pki "${BACKUP_PATH}/pki"

# Create compressed archive
tar -czf "${BACKUP_PATH}.tar.gz" -C "${BACKUP_DIR}" "backup_${TIMESTAMP}"
rm -rf "${BACKUP_PATH}"

# Upload to S3 (or your backup storage)
aws s3 cp "${BACKUP_PATH}.tar.gz" s3://mqtt-backups/

# Retention: keep last 30 days
find "${BACKUP_DIR}" -name "backup_*.tar.gz" -mtime +30 -delete

echo "✓ Backup completed: ${BACKUP_PATH}.tar.gz"

7.2 Disaster Recovery Plan

8. Checklist Production

9. Références & Ressources

• Spécifications
  • MQTT 5.0 Specification: https://docs.oasis-open.org/mqtt/mqtt/v5.0/
  • MQTT Security: https://mqtt.org/mqtt-security-fundamentals/
• Documentation Brokers
  • Mosquitto: https://mosquitto.org/documentation/
  • EMQX: https://docs.emqx.com/
  • VerneMQ: https://docs.vernemq.com/
• Outils
  • MQTT Explorer (GUI): https://mqtt-explorer.com/
  • mqtt-benchmark: https://github.com/krylovsk/mqtt-benchmark
  • Mosquitto clients: mosquitto_pub/sub