MQTT-Authentifizierung testen: gültige, ungültige und fehlende Zugangsdaten

MQTT-Authentifizierung testen heißt: Du verbindest dich mit jeder Art von Zugangsdaten, die dein Broker bekommen kann — gültig, falsch, unbekannt, leer und fehlend — und prüfst, dass jede davon den richtigen CONNACK-Code erhält. Ein Broker, der das richtige Passwort akzeptiert, ist erst zur Hälfte getestet; die Sicherheit entsteht dadurch, wie er alles andere ablehnt.

Dieser Leitfaden liefert dir eine Testmatrix mit dem erwarteten Ergebnis pro Fall für MQTT 3.1.1 und MQTT 5, die Befehle für den manuellen Test und eine pytest-Suite, die alles automatisch in der CI ausführt.

Wie MQTT das Ergebnis der Authentifizierung meldet

Die Authentifizierung passiert genau einmal, im CONNECT-Paket. Der Client sendet Benutzername und/oder Passwort, der Broker antwortet mit CONNACK. Ein Code ungleich null bedeutet, dass die Verbindung abgelehnt wurde — direkt danach schließt der Broker den Socket.

Die Spezifikation zwingt den Broker nicht, zwischen „falsches Passwort“ und „nicht autorisiert“ zu unterscheiden. Manche Broker liefern 0x04/0x86 bei falschem Passwort; Mosquitto liefert 0x05/0x87 bei jedem Fehler mit Zugangsdaten. Das ist in Ordnung — entscheidend ist, dass sich dein Broker konsistent verhält und deine Tests sein tatsächliches Verhalten festschreiben. Jeden Code kannst du im MQTT-Reason-Code-Tool nachschlagen.

Die Testmatrix für die Authentifizierung

Führe jede Zeile gegen den Broker aus, den du produktiv einsetzt — mindestens einmal mit MQTT 3.1.1 und einmal mit MQTT 5. „Abgelehnt“ heißt 0x04/0x05 in 3.1.1 und 0x86/0x87 in 5.

Die Fälle 3 und 9 finden echte Bugs. Bekommt ein unbekannter Benutzer einen anderen Code als ein falsches Passwort, verrät der Broker, welche Konten existieren. Wird ein Zeilenumbruch am Ende akzeptiert, kürzt die Passwortdatei oder das Auth-Backend die Eingabe — meist harmlos, aber du willst es wissen. Fall 10 prüft, ob ein Widerruf wirklich greift: Bei Mosquitto heißt das, dass die Passwortdatei neu geladen wurde (SIGHUP), bei einem Managed Broker, dass die Änderung verteilt wurde.

Die Matrix manuell ausführen

Mit mosquitto_pub probierst du jeden Fall am schnellsten aus. Mit -d siehst du das rohe CONNACK:

# 1. gültige Zugangsdaten — endet still mit Exit-Code 0
mosquitto_pub -h broker.connectmq.com -p 1883 -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t testmqtt/auth/check -m ok -d

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

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

# 4. ganz ohne Zugangsdaten
mosquitto_pub -h broker.connectmq.com -p 1883 -t testmqtt/auth/check -m ok -d

Eine abgelehnte Verbindung gibt eine dieser Zeilen aus — genau die Meldungen, nach denen meistens gesucht wird:

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

Mit -V mqttv5 zeigen dieselben Fälle den Text des MQTT-5-Reason-Codes, zum Beispiel Connection error: Not authorized. Für einen visuellen Check zeigt der Online-MQTT-Client das CONNACK-Ergebnis jedes Versuchs — praktisch, um WebSocket- und TCP-Listener zu vergleichen, die manche Broker mit unterschiedlichen Auth-Backends konfigurieren.

Siehst du stattdessen Connection Refused: identifier rejected, liegt das Problem an der Client-ID, nicht an den Zugangsdaten — siehe identifier rejected. Hängt der Client und läuft dann in einen Timeout, bist du wahrscheinlich auf dem falschen Port oder hinter einer Firewall — mit der Authentifizierung hat das nichts zu tun.

Automatisieren mit pytest

Manuelle Checks beweisen, dass der Broker heute funktioniert. Eine automatisierte Suite beweist, dass er auch nach der nächsten Konfigurationsänderung noch funktioniert. Dieser Test nutzt paho-mqtt 2.x und führt jeden Fall mit beiden Protokollversionen aus:

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):
    """Liefert den Reason-Text des CONNACK oder 'closed', wenn kein CONNACK kam."""
    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")

Ein paar Details machen den Test zuverlässig:

• paho-mqtt 2.x bildet die Return Codes von MQTT 3.1.1 auf MQTT-5-Reason-Codes ab: 0x04 kommt in beiden Versionen als „Bad user name or password“ an, 0x05 als „Not authorized“. Ein Satz Assertions deckt also beide ab.
• Jeder Versuch nutzt eine neue zufällige Client-ID. Eine wiederverwendete ID kann dazu führen, dass ein erfolgreicher Test eine frühere Verbindung verdrängt und das Ergebnis verfälscht. Der Client-ID-Generator erzeugt IDs im gleichen Format.
• "closed" deckt Broker ab, die bei manchen Fehlern die TCP-Verbindung trennen, statt ein CONNACK zu senden — das ist erlaubt.
• Der letzte Test ist der Enumeration-Check aus Fall 3 der Matrix.

Starte ihn mit MQTT_USER=... MQTT_PASS=... pytest -v test_auth.py. In der CI legst du die Zugangsdaten als Secrets ab und lässt die Suite bei jeder Änderung der Broker-Konfiguration gegen einen Staging-Broker laufen — siehe MQTT-Tests in GitHub Actions.

Was Authentifizierungstests nicht abdecken

Ein erfolgreiches CONNECT beweist nur die Identität. Es sagt nichts darüber, was der Client danach tun darf. Ein Sensor mit gültigen Zugangsdaten darf trotzdem nicht auf das Command-Topic eines anderen Geräts publishen — das ist Autorisierung, und sie scheitert anders (still in 3.1.1, mit Reason Code in MQTT 5). Teste sie separat mit der Methode aus MQTT-ACLs testen.

Außerdem gilt:

• Unverschlüsseltes TCP legt Passwörter offen. Die Zugangsdaten im CONNECT sind nicht verschlüsselt. Führe den Fall mit gültigen Zugangsdaten auch über TLS auf Port 8883 aus und prüfe, dass der unverschlüsselte Listener in der Produktion deaktiviert ist.
• Brute-Force-Schutz ist ein eigener Test. Manche Broker oder Proxys bremsen oder blockieren wiederholte Fehlversuche. Falls deiner das tut, ergänze einen Test, dass zehn falsche Passwörter hintereinander den legitimen Client nicht aussperren — oder dass sie es tun, wenn das deine Policy ist.
• Token-basierte Authentifizierung braucht Ablauftests. Nutzt du JWT oder OAuth statt statischer Passwörter, ergänze Fälle für abgelaufene, noch nicht gültige und falsch signierte Tokens.

Gültige und ungültige Zugangsdaten im Online-Client testen

Authentifizierungstests kosten wenig: ein Dutzend Verbindungsversuche und ein paar Assertions. Leg die Matrix neben die Broker-Konfiguration in dein Repository, führe sie bei jeder Änderung aus — dann merkst du vor einem Angreifer, wenn jemand den anonymen Zugriff wieder einschaltet.

Häufig gestellte Fragen