MQTT-ACLs und Topic-Berechtigungen testen

Um MQTT-ACLs zu testen, verbindest du dich als der zu prüfende Benutzer, probierst jeden Publish und jede Subscription, die das Gerät können bzw. nicht können soll, und prüfst das Ergebnis von außerhalb des Clients. Der letzte Punkt ist entscheidend: In MQTT 3.1.1 sieht ein abgelehnter Publish genauso aus wie ein erfolgreicher. Der einzige verlässliche Beweis ist ein zweiter Client, der beobachtet, ob die Nachricht angekommen ist.

Dieser Leitfaden zeigt, wie jede Protokollversion eine Ablehnung meldet, eine Testmatrix für Berechtigungen, manuelle Checks mit mosquitto und einen automatisierten Test, den du nach jeder ACL-Änderung ausführen kannst.

Wie Broker eine abgelehnte Aktion melden

ACLs (Access Control Lists) legen fest, auf welche Topics ein Benutzer publishen (Schreiben) und welche er abonnieren (Lesen) darf. Die Authentifizierung war zu diesem Zeitpunkt bereits erfolgreich — das CONNACK war 0x00 —, also kommt jede Rückmeldung in späteren Paketen.

Das Verhalten von 3.1.1 ist kein Bug deines Brokers. Die Spezifikation sagt: Autorisiert ein Server einen PUBLISH nicht, kann er den Client darüber nicht informieren und muss ihn entweder nach den normalen QoS-Regeln bestätigen oder die Netzwerkverbindung schließen. Viele Clients loggen in beiden Fällen nichts — deshalb ist „meine Nachrichten verschwinden“ ein so häufiges ACL-Symptom. Siehe MQTT-ACL lehnt still ab.

Eine Testmatrix für Berechtigungen aufbauen

Liste auf, was ein repräsentativer Benutzer tun darf, und ergänze, was er nicht tun darf. Hier ein Beispiel für einen Sensor, der seine eigene Telemetrie publishen und seine eigenen Befehle lesen darf, mit einer passenden Mosquitto-ACL:

# every device: own telemetry and own commands only
pattern write testmqtt/acl/telemetry/%u/#
pattern read  testmqtt/acl/commands/%u/#

# a monitoring user that may read everything under testmqtt/acl
user acl-observer
topic read testmqtt/acl/#

Bei Fall 9 erleben viele eine Überraschung: telemetry/%u/# passt auch auf telemetry/sensor-01 selbst, weil # die Elternebene einschließt. Füge deine Patterns und Topics in den Topic Matcher ein, um Grenzfälle vor dem Schreiben des Tests zu prüfen. Fall 8 ist wichtig, weil Retained Messages beim Subscribe zugestellt werden; ein Broker, der dort die Lese-ACL vergisst, verrät den letzten Wert jedes Topics. Weitere typische Fallen findest du unter MQTT-Wildcard-Sicherheit.

Manuelle Checks mit mosquitto

Starte zuerst einen Beobachter. Er nutzt ein Konto mit breitem Lesezugriff und zeigt dadurch, was wirklich beim Broker ankommt:

mosquitto_sub -h broker.connectmq.com -p 1883 \
  -u acl-observer -P "$OBSERVER_PASS" -t 'testmqtt/acl/#' -v

Jetzt publishst du als das zu testende Gerät mit MQTT 5 und QoS 1, damit eine Ablehnung einen Reason Code erzeugt:

# Fall 1: muss beim Beobachter erscheinen
mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 -q 1 \
  -u sensor-01 -P "$MQTT_PASS" -t testmqtt/acl/telemetry/sensor-01/temp -m 21.5

# Fall 2: darf NICHT erscheinen; MQTT 5 meldet die Ablehnung
mosquitto_pub -h broker.connectmq.com -p 1883 -V mqttv5 -q 1 \
  -u sensor-01 -P "$MQTT_PASS" -t testmqtt/acl/telemetry/sensor-02/temp -m 99

Beim abgelehnten Publish gibt der mosquitto-Client eine Warnung wie Warning: Publish 1 failed: Not authorized. aus. Führst du denselben Befehl mit -V mqttv311 aus, endet er ohne jede Warnung — der stille Beobachter ist dann dein einziger Beweis.

Subscriptions lassen sich leichter prüfen, weil beide Versionen im SUBACK antworten:

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 -d \
  -u sensor-01 -P "$MQTT_PASS" -t 'testmqtt/acl/commands/sensor-02/#'

Mit -d siehst du Subscribed (mid: 1): 135 bei einer Ablehnung in MQTT 5 (0x87) bzw. Subscribed (mid: 1): 128 (0x80) in 3.1.1. Aktuelle mosquitto-Versionen geben danach All subscription requests were denied. aus und beenden sich. Gewährt dein Broker die Subscription stattdessen, publishe von einem anderen Konto auf dieses Topic und prüfe, dass nichts zugestellt wird. Die Codes erklären SUBACK 0x80 und Not authorized 0x87 im Detail.

Interaktiv geht das auch im Online-MQTT-Client: Öffne zwei Sitzungen mit verschiedenen Benutzern, abonniere in der einen als Beobachter und publishe in der anderen als Gerät.

Ein automatisierter ACL-Test

Dieses Skript mit paho-mqtt 2.x führt die Publish- und Subscribe-Fälle der Matrix aus. Es prüft die MQTT-5-Reason-Codes und nutzt den Beobachter als Referenz, funktioniert also auch mit Brokern, die abgelehnte Nachrichten still verwerfen:

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

HOST, PORT = "broker.connectmq.com", 1883
DEVICE = ("sensor-01", os.environ["MQTT_PASS"])
OBSERVER = ("acl-observer", os.environ["OBSERVER_PASS"])
BASE = "testmqtt/acl"

def client(creds):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                    client_id=f"acl-{uuid.uuid4().hex[:8]}", protocol=mqtt.MQTTv5)
    c.username_pw_set(*creds)
    c.connect(HOST, PORT)
    c.loop_start()
    time.sleep(0.5)
    return c

seen = set()
obs = client(OBSERVER)
obs.on_message = lambda c, u, m: seen.add(m.topic)
obs.subscribe(f"{BASE}/#", qos=1)

dev = client(DEVICE)
acks = {}
dev.on_publish = lambda c, u, mid, rc, props: acks.__setitem__(mid, str(rc))
subacks = {}
dev.on_subscribe = lambda c, u, mid, rcs, props: subacks.__setitem__(mid, [str(r) for r in rcs])

def check_publish(topic, allowed):
    info = dev.publish(topic, "acl-test", qos=1)
    time.sleep(1)
    delivered = topic in seen
    print(f"PUB {topic}: puback={acks.get(info.mid)} delivered={delivered}")
    assert delivered == allowed, f"publish to {topic}: expected allowed={allowed}"
    if not allowed:
        assert acks.get(info.mid) == "Not authorized"

def check_subscribe(topic, allowed):
    _, mid = dev.subscribe(topic, qos=1)
    time.sleep(1)
    granted = not subacks[mid][0].startswith(("Not authorized", "Unspecified"))
    print(f"SUB {topic}: suback={subacks[mid]}")
    assert granted == allowed, f"subscribe to {topic}: expected allowed={allowed}"

check_publish(f"{BASE}/telemetry/sensor-01/temp", True)
check_publish(f"{BASE}/telemetry/sensor-02/temp", False)
check_publish(f"{BASE}/commands/sensor-01/reboot", False)
check_subscribe(f"{BASE}/commands/sensor-01/#", True)
check_subscribe(f"{BASE}/commands/sensor-02/#", False)
print("all ACL checks passed")

Führe es nach jeder Änderung an der ACL-Datei oder am Auth-Backend aus und in der CI gegen einen Staging-Broker. Schlägt ein Check fehl, zeigen dir der ausgegebene Reason Code und das delivered-Flag sofort, ob der Broker die Nachricht abgelehnt, still verworfen oder durchgelassen hat.

Häufige Gründe für fehlschlagende ACL-Tests

• Alles wird abgelehnt. Die ACL-Datei ist nicht geladen, oder der Benutzername in der Regel passt nicht zu dem im CONNECT. Prüfe das Broker-Log beim Start; bei Mosquitto muss acl_file im selben Listener-Kontext stehen wie die Passwortdatei — siehe Mosquitto-Authentifizierung.
• Alles ist erlaubt. Eine breite Regel wie topic readwrite # weiter oben in der Datei oder anonymer Zugriff mit eigenen großzügigen Regeln.
• pattern-Regeln greifen nicht. %u steht für den Benutzernamen, %c für die Client-ID; mit dem falschen Platzhalter wird jedes Gerät abgelehnt, dessen ID sich vom Benutzernamen unterscheidet.
• Änderungen wirken nicht. Die meisten Broker lesen ACLs beim Start oder beim Reload. Schick Mosquitto ein SIGHUP oder starte ihn neu, bevor du den Test wiederholst.
• Test grün, Produktion rot. Der Client publisht mit führendem Slash (/testmqtt/...) oder anderer Groß-/Kleinschreibung. Topics unterscheiden Groß- und Kleinschreibung, und /a ist nicht a.

ACL-Testverkehr im Online-Client beobachten

Eine ACL ist eine Sicherheitskontrolle, und Sicherheitskontrollen brauchen Tests, die fehlschlagen, wenn die Kontrolle kaputtgeht. Pflege eine Matrix erlaubter und verbotener Aktionen pro Rolle, prüfe die Zustellung mit einem Beobachter, statt dem Publisher zu vertrauen, und setze auf MQTT 5, damit Ablehnungen gemeldet statt verschluckt werden.

Häufig gestellte Fragen