Tester l'authentification MQTT : identifiants valides, invalides et absents

Tester l'authentification MQTT, c'est se connecter avec tous les types d'identifiants que votre broker peut recevoir — valides, faux, inconnus, vides et absents — et vérifier que chacun obtient le bon code CONNACK. Un broker qui accepte le bon mot de passe n'est testé qu'à moitié ; la sécurité vient de la façon dont il refuse tout le reste.

Ce guide vous donne une matrice de tests avec le résultat attendu pour chaque cas en MQTT 3.1.1 et MQTT 5, les commandes pour l'exécuter à la main et une suite pytest qui l'exécute automatiquement en CI.

Comment MQTT signale le résultat de l'authentification

L'authentification a lieu une seule fois, dans le paquet CONNECT. Le client envoie un nom d'utilisateur et/ou un mot de passe, et le broker répond par un CONNACK. Un code non nul signifie que la connexion est refusée, et le broker ferme le socket juste après l'avoir envoyé.

La spécification n'oblige pas le broker à choisir entre « mauvais mot de passe » et « non autorisé ». Certains brokers renvoient 0x04/0x86 pour un mot de passe faux ; Mosquitto renvoie 0x05/0x87 pour tout échec d'identifiants. Ce n'est pas un problème : ce qui compte, c'est que votre broker soit cohérent et que vos tests figent son comportement réel. Retrouvez la signification de chaque code dans l'outil reason codes MQTT.

La matrice de tests d'authentification

Exécutez chaque ligne contre le broker que vous utilisez en production, au moins une fois en MQTT 3.1.1 et une fois en MQTT 5. « Refusée » signifie 0x04/0x05 en 3.1.1 et 0x86/0x87 en 5.

Les cas 3 et 9 révèlent de vrais bugs. Si un utilisateur inconnu reçoit un code différent d'un mauvais mot de passe, le broker divulgue quels comptes existent. Si un retour à la ligne final est accepté, le fichier de mots de passe ou le backend d'authentification tronque l'entrée — généralement sans gravité, mais vous voulez le savoir. Le cas 10 vérifie que la révocation prend vraiment effet : avec Mosquitto, cela veut dire que le fichier de mots de passe a été rechargé (SIGHUP) ; avec un broker managé, que le changement a été propagé.

Exécuter la matrice à la main

mosquitto_pub est le moyen le plus rapide d'essayer chaque cas. Ajoutez -d pour voir le CONNACK brut :

# 1. identifiants valides — se termine silencieusement avec le code 0
mosquitto_pub -h broker.connectmq.com -p 1883 -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t testmqtt/auth/check -m ok -d

# 2. mauvais mot de passe
mosquitto_pub -h broker.connectmq.com -p 1883 -u "$MQTT_USER" -P "wrong" \
  -t testmqtt/auth/check -m ok -d

# 3. utilisateur inconnu
mosquitto_pub -h broker.connectmq.com -p 1883 -u "no-such-user" -P "wrong" \
  -t testmqtt/auth/check -m ok -d

# 4. aucun identifiant
mosquitto_pub -h broker.connectmq.com -p 1883 -t testmqtt/auth/check -m ok -d

Une connexion refusée affiche l'une de ces lignes, qui sont précisément les messages que l'on recherche le plus souvent :

Connection error: Connection Refused: bad user name or password.
Connection error: Connection Refused: not authorised.

Avec -V mqttv5, les mêmes cas affichent le texte du reason code MQTT 5, par exemple Connection error: Not authorized. Pour une vérification visuelle, le client MQTT en ligne affiche le résultat du CONNACK à chaque tentative — pratique pour comparer les listeners WebSocket et TCP, que certains brokers configurent avec des backends d'authentification différents.

Si vous voyez plutôt Connection Refused: identifier rejected, le problème vient du client ID, pas des identifiants — voir identifier rejected. Si le client reste bloqué puis expire, vous êtes probablement sur le mauvais port ou derrière un pare-feu : l'authentification n'y est pour rien.

Automatiser avec pytest

Les tests manuels prouvent que le broker fonctionne aujourd'hui. Une suite automatisée prouve qu'il fonctionne encore après le prochain changement de configuration. Ce test utilise paho-mqtt 2.x et exécute chaque cas avec les deux versions du protocole :

import os, threading, uuid
import paho.mqtt.client as mqtt
import pytest

HOST, PORT = "broker.connectmq.com", 1883
USER, PASS = os.environ["MQTT_USER"], os.environ["MQTT_PASS"]
REFUSED = {"Bad user name or password", "Not authorized"}

def try_connect(username=None, password=None, protocol=mqtt.MQTTv5, timeout=5):
    """Renvoie le texte du reason code du CONNACK, ou 'closed' si aucun CONNACK n'est arrivé."""
    result, done = {}, threading.Event()
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                    client_id=f"auth-test-{uuid.uuid4().hex[:8]}", protocol=protocol)
    if username is not None or password is not None:
        c.username_pw_set(username, password)

    def on_connect(client, userdata, flags, reason_code, properties):
        result["rc"] = str(reason_code)
        done.set()

    c.on_connect = on_connect
    c.connect(HOST, PORT, keepalive=10)
    c.loop_start()
    done.wait(timeout)
    c.disconnect()
    c.loop_stop()
    return result.get("rc", "closed")

PROTOCOLS = [mqtt.MQTTv311, mqtt.MQTTv5]

@pytest.mark.parametrize("proto", PROTOCOLS)
def test_valid_credentials(proto):
    assert try_connect(USER, PASS, proto) == "Success"

@pytest.mark.parametrize("proto", PROTOCOLS)
@pytest.mark.parametrize("user,pw", [
    (USER, "wrong-password"),
    ("no-such-user", "wrong-password"),
    (USER, ""),
    (USER, PASS + "\n"),
])
def test_bad_credentials_refused(proto, user, pw):
    assert try_connect(user, pw, proto) in REFUSED

def test_username_is_case_sensitive():
    if USER.upper() == USER:
        pytest.skip("username has no lowercase letters")
    assert try_connect(USER.upper(), PASS) in REFUSED

@pytest.mark.parametrize("proto", PROTOCOLS)
def test_anonymous_refused(proto):
    assert try_connect(protocol=proto) in REFUSED | {"closed"}

def test_unknown_user_looks_like_wrong_password():
    assert try_connect(USER, "wrong") == try_connect("no-such-user", "wrong")

Quelques détails rendent ce test fiable :

• paho-mqtt 2.x convertit les return codes MQTT 3.1.1 en reason codes MQTT 5 : 0x04 arrive comme « Bad user name or password » et 0x05 comme « Not authorized » dans les deux versions. Un seul jeu d'assertions couvre les deux.
• Chaque tentative utilise un nouveau client ID aléatoire. Réutiliser un ID peut faire qu'un test réussi éjecte une connexion précédente et brouille le résultat. Le générateur de client ID produit des ID au même format.
• "closed" couvre les brokers qui coupent la connexion TCP au lieu d'envoyer un CONNACK, ce qui est autorisé pour certains échecs.
• Le dernier test correspond à la vérification d'énumération de comptes du cas 3 de la matrice.

Lancez-le avec MQTT_USER=... MQTT_PASS=... pytest -v test_auth.py. En CI, stockez les identifiants dans des secrets et exécutez la suite contre un broker de staging à chaque modification de la configuration du broker — voir tests MQTT dans GitHub Actions.

Ce que les tests d'authentification ne couvrent pas

Un CONNECT réussi ne prouve que l'identité. Il ne dit rien de ce que le client a le droit de faire ensuite. Un capteur avec des identifiants valides ne doit pas pour autant pouvoir publier sur le topic de commandes d'un autre appareil — c'est de l'autorisation, et elle échoue autrement (en silence en 3.1.1, avec un reason code en MQTT 5). Testez-la séparément avec la méthode de comment tester les ACL MQTT.

Gardez aussi en tête :

• Le TCP non chiffré expose les mots de passe. Les identifiants du CONNECT ne sont pas chiffrés. Exécutez aussi le cas des identifiants valides en TLS sur le port 8883 et vérifiez que le listener non chiffré est désactivé en production.
• La protection contre la force brute est un test à part. Certains brokers ou proxys ralentissent ou bloquent les échecs répétés. Si c'est le cas du vôtre, ajoutez un test vérifiant que dix mauvais mots de passe d'affilée ne bloquent pas le client légitime — ou qu'ils le bloquent, si c'est votre politique.
• L'authentification par jeton nécessite des tests d'expiration. Si vous utilisez JWT ou OAuth au lieu de mots de passe statiques, ajoutez des cas pour les jetons expirés, pas encore valides et mal signés.

Testez des identifiants valides et invalides dans le client en ligne

Tester l'authentification coûte peu : une douzaine de tentatives de connexion et quelques assertions. Gardez la matrice dans votre dépôt à côté de la configuration du broker, exécutez-la à chaque changement, et vous remarquerez le jour où quelqu'un réactive l'accès anonyme avant qu'un attaquant ne le fasse.

Questions fréquentes