Pico W MQTT TLS : connexions MicroPython sécurisées sur le port 8883 avec vérification de certificat

Pico W MQTT TLS, c'est ce qui transforme une connexion de bricoleur en connexion digne de confiance : le trafic est chiffré, et la carte vérifie qu'elle parle bien à votre broker avant d'envoyer un mot de passe. Ce guide part d'un sketch MicroPython MQTT fonctionnel et le fait passer sur le port 8883 avec vérification complète du certificat : versions de firmware nécessaires, certificat racine à charger, importance de l'horloge, durée de la négociation sur un RP2040 et correction des erreurs les plus probables.

Pourquoi TLS compte sur un microcontrôleur

MQTT en clair sur le port 1883 envoie tout sans chiffrement, y compris le nom d'utilisateur et le mot de passe dans le paquet CONNECT. Toute personne sur le même Wi-Fi, chez le fournisseur d'accès ou sur le chemin vers le broker peut lire vos mesures et rejouer vos identifiants. TLS règle deux problèmes à la fois :

1. Le chiffrement. Payloads et identifiants sont illisibles en transit.
2. L'authentification du serveur. L'appareil vérifie la chaîne de certificats du broker : un faux point d'accès ou un détournement DNS ne peut pas usurper le broker pour collecter des identifiants.

C'est ce second point que la plupart des tutoriels passent sous silence. Activer TLS sans vérifier le certificat protège contre l'écoute passive, mais pas contre un attaquant actif. Ce guide le fait correctement. Pour une vue d'ensemble, consultez les bonnes pratiques de sécurité MQTT.

Si vous n'avez jamais connecté de Pico W à MQTT, commencez par le tutoriel Raspberry Pi Pico W MQTT en clair et revenez une fois qu'il fonctionne.

Versions de firmware et de bibliothèque

La prise en charge de TLS dans MicroPython a évolué au fil des dernières versions, et beaucoup d'anciens exemples ne fonctionnent plus tels quels.

Les anciens exemples passent ssl=True avec un dictionnaire ssl_params. Si vous voyez une TypeError sur des arguments nommés inattendus, ou AttributeError: 'bool' object has no attribute 'wrap_socket', vous mélangez un ancien exemple avec une nouvelle bibliothèque, ou l'inverse. Mettez à jour firmware et bibliothèque et utilisez le code basé sur un contexte ci-dessous.

Vérifiez ce que vous utilisez depuis le REPL :

import sys, os
print(sys.version)
print(os.uname().machine)

Étape 1 : obtenir le bon certificat racine

Les brokers ConnectMQ, y compris le serveur MQTT public gratuit broker.connectmq.com, utilisent des certificats émis par Let's Encrypt. Vous pouvez vérifier la chaîne vous-même :

openssl s_client -connect broker.connectmq.com:8883 -servername broker.connectmq.com -showcerts </dev/null \
  | grep -E "s:|i:"

La sortie liste le certificat du serveur, l'intermédiaire Let's Encrypt et, à la fin, un émetteur ISRG Root X1. C'est cette racine que le Pico W doit approuver. Faire confiance à la racine plutôt qu'au certificat du serveur est important : les certificats Let's Encrypt sont renouvelés tous les quelques mois et les intermédiaires changent de temps en temps, alors que la racine reste valide pendant des années.

Téléchargez la racine au format DER, que MicroPython charge directement :

curl -o isrgrootx1.der https://letsencrypt.org/certs/isrgrootx1.der
# or convert a PEM copy you already have
openssl x509 -in isrgrootx1.pem -outform DER -out isrgrootx1.der

Copiez-la sur la carte avec la vue fichiers de Thonny ou :

mpremote cp isrgrootx1.der :isrgrootx1.der

Étape 2 : régler l'horloge

Le Pico W n'a pas d'horloge sauvegardée par pile. À chaque mise sous tension, son horloge temps réel démarre à une date par défaut dans le passé. Selon la version du firmware, les périodes de validité des certificats peuvent être vérifiées par rapport à cette horloge, et un appareil qui se croit des années en arrière rejettera un certificat émis le mois dernier comme pas encore valide. Même lorsque ce contrôle n'est pas appliqué, une heure correcte est utile pour la journalisation et les horodatages : réglez-la donc via NTP juste après la connexion Wi-Fi :

import ntptime
ntptime.host = "pool.ntp.org"
ntptime.settime()          # sets the RTC to UTC

ntptime.settime() peut lever une OSError quand la requête NTP expire : réessayez quelques fois au lieu de laisser le programme planter.

Étape 3 : le programme Pico W MQTT TLS complet

Enregistrez-le sous main.py. Pour le serveur public de test, les identifiants peuvent rester vides et n'importe quel client ID unique fonctionne. En production, renseignez les valeurs générées dans le portail.

# main.py - Pico W MQTT over TLS with certificate verification
import time
import gc
import ssl
import network
import machine
import ntptime
import ubinascii
from umqtt.simple import MQTTClient

WIFI_SSID = "YOUR_WIFI_SSID"
WIFI_PASSWORD = "YOUR_WIFI_PASSWORD"

BROKER = "broker.connectmq.com"
PORT = 8883
# Public test broker: anonymous, any unique client ID.
# Production: use the values from the broker portal.
MQTT_USER = None                 # "YOUR_USERNAME"
MQTT_PASSWORD = None             # "YOUR_PASSWORD"
CLIENT_ID = "pico-tls-" + ubinascii.hexlify(machine.unique_id()).decode()  # or "YOUR_CLIENT_ID"

TOPIC = "connectmq/demo/tl5v8n/pico1/temperature"
CA_FILE = "isrgrootx1.der"

sensor = machine.ADC(4)


def connect_wifi():
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    if not wlan.isconnected():
        wlan.connect(WIFI_SSID, WIFI_PASSWORD)
        for _ in range(40):
            if wlan.isconnected():
                break
            time.sleep(0.5)
    if not wlan.isconnected():
        raise OSError("Wi-Fi failed, status %d" % wlan.status())
    print("Wi-Fi:", wlan.ifconfig()[0])


def sync_time():
    for attempt in range(5):
        try:
            ntptime.settime()
            print("Time set:", time.gmtime())
            return
        except OSError:
            time.sleep(2)
    print("Warning: NTP failed, clock may be wrong")


def make_ssl_context():
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ctx.verify_mode = ssl.CERT_REQUIRED
    with open(CA_FILE, "rb") as f:
        ctx.load_verify_locations(cadata=f.read())
    return ctx


def connect_mqtt(ctx):
    gc.collect()
    print("Free RAM before TLS:", gc.mem_free())
    client = MQTTClient(
        CLIENT_ID, BROKER, port=PORT,
        user=MQTT_USER, password=MQTT_PASSWORD,
        keepalive=60, ssl=ctx,
    )
    start = time.ticks_ms()
    client.connect()
    print("TLS + MQTT connect took", time.ticks_diff(time.ticks_ms(), start), "ms")
    print("Free RAM after TLS:", gc.mem_free())
    return client


def read_temperature():
    v = sensor.read_u16() * 3.3 / 65535
    return round(27 - (v - 0.706) / 0.001721, 2)


def main():
    connect_wifi()
    sync_time()
    ctx = make_ssl_context()
    client = connect_mqtt(ctx)
    while True:
        payload = '{"value": %s, "unit": "C"}' % read_temperature()
        client.publish(TOPIC, payload, qos=1)
        print("Published", payload)
        time.sleep(30)           # publishing every 30 s keeps the 60 s keepalive satisfied


while True:
    try:
        main()
    except OSError as e:
        print("Error:", e, "- reconnecting in 10 s")
        time.sleep(10)

umqtt.simple enveloppe le socket avec le contexte et transmet le nom du broker comme nom d'hôte du serveur. C'est ce qui permet à mbedTLS de vérifier que le certificat a bien été émis pour broker.connectmq.com, et cela envoie aussi le SNI, dont beaucoup de serveurs TLS ont besoin pour choisir le bon certificat.

Observez la sortie depuis votre ordinateur, également en TLS :

mosquitto_sub -h broker.connectmq.com -p 8883 --cafile isrgrootx1.pem \
  -i laptop-tl5v8n -v -t "connectmq/demo/tl5v8n/#"

Durée de négociation et mémoire sur le RP2040

Le RP2040 n'a pas d'accélération cryptographique matérielle : mbedTLS fait tout le travail en logiciel sur des cœurs Cortex-M0+. Attendez-vous à une négociation TLS nettement perceptible, souvent de l'ordre d'une seconde ou plus, contre une connexion TCP en clair quasi instantanée. Le programme affiche la valeur exacte sur votre réseau ; notez-la, car elle dépend de l'échange de clés négocié, de la longueur de la chaîne de certificats et de la latence Wi-Fi.

Concrètement :

1. Connectez-vous une fois, restez connecté. Sur secteur, gardez une session TLS de longue durée et envoyez de nombreux messages dessus. Le coût de la négociation n'est payé qu'à la reconnexion.
2. Attention aux conceptions avec veille profonde. Un appareil sur batterie qui se réveille, se connecte, publie et se rendort paie la négociation complète à chaque cycle, ce qui coûte du temps et de l'énergie. Notre guide Pico W sur batterie MQTT traite des intervalles plus longs et du regroupement pour ce cas.
3. Libérez d'abord la mémoire. Une session TLS a besoin de dizaines de kilo-octets de RAM pour ses tampons. Appelez gc.collect() avant de vous connecter, évitez de créer de gros objets avant la négociation et comparez les valeurs de mémoire libre affichées avant et après.
4. Envisagez le Pico 2 W. Son RP2350 utilise des cœurs Cortex-M33 plus rapides, ce qui raccourcit la négociation et laisse plus de marge. Voir Pico W vs Pico 2 W.

Dépannage des erreurs TLS du Pico W

Un mot sur CERT_NONE : régler ctx.verify_mode = ssl.CERT_NONE fait disparaître les erreurs, d'où sa présence dans beaucoup de réponses de forums. Mais cela supprime aussi l'authentification du serveur. Ne l'utilisez que pour un diagnostic ponctuel, jamais dans du code déployé.

Passer à votre propre broker privé

Le serveur public accepte les connexions TLS anonymes sur le port 8883, ce qui est parfait pour tester ce code. Pour de vrais appareils, créez un broker privé sur ConnectMQ et copiez les identifiants générés dans les trois paramètres en haut du programme. Le broker lie les client IDs aux identifiants émis : utilisez exactement le client ID du portail. Comme le certificat provient de Let's Encrypt, le même fichier isrgrootx1.der fonctionne pour le serveur public comme pour votre broker privé, et les renouvellements de certificat côté broker ne demandent aucune modification sur l'appareil.

Un broker MQTT managé bloque aussi les adresses IP après des échecs de connexion répétés et applique des quotas par offre, ce qui protège votre broker des clients défaillants ou compromis sans aucun travail de votre part. Pour en savoir plus sur l'exploitation de brokers dans le cloud, consultez notre guide broker MQTT cloud, et pour d'autres projets Raspberry Pi, le tutoriel Raspberry Pi MQTT.

Sécurisez votre Pico W avec ConnectMQ

Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire, copiez le nom d'utilisateur, le mot de passe et le client ID dans main.py, et votre Pico W publie en TLS vérifié sur le port 8883. L'explorateur de topics en direct du portail montre l'arrivée des messages : vous pouvez ainsi confirmer le chemin sécurisé de bout en bout.

Questions fréquentes

De quel certificat le Pico W a-t-il besoin pour ConnectMQ ?

ISRG Root X1, la racine Let's Encrypt, au format DER. Faire confiance à la racine signifie que les renouvellements de certificat courants côté broker n'exigent pas de mettre à jour l'appareil.

Pourquoi mon Pico W rejette-t-il un certificat valide ?

Les causes les plus fréquentes sont une horloge jamais réglée, qui fait paraître le certificat pas encore valide, une connexion par adresse IP au lieu du nom d'hôte, ou le chargement du mauvais certificat racine.

Combien de temps dure une négociation TLS sur le Pico W ?

Elle est sensiblement plus lente que sur des puces dotées de matériel cryptographique, car le RP2040 effectue toute la cryptographie en logiciel. Mesurez-la avec time.ticks_ms() autour de connect() comme ci-dessus, et gardez les connexions ouvertes plutôt que de vous reconnecter à chaque message.

Puis-je utiliser des certificats client sur le Pico W ?

Le SSLContext de MicroPython prend en charge load_cert_chain pour les certificats client. Le broker authentifie les appareils avec le nom d'utilisateur, le mot de passe et le client ID émis dans le portail : une connexion TLS avec vérification du serveur et ces identifiants est donc la configuration standard.