Fuzz-Tests für MQTT-Clients und Payload-Parser

Jeder Subscriber in einem MQTT-System vertraut Bytes, die er nicht selbst erzeugt hat. Ein Gerät mit Firmware-Bug, eine halb geschriebene Retained Message oder ein vergessenes Testskript kann beliebige Daten auf einen Topic publishen, und wenn dein Decoder dabei eine Exception wirft, stürzt der Worker ab, die Nachricht wird erneut zugestellt und der Absturz wiederholt sich. Fuzz-Tests für MQTT-Payload-Parser finden solche Eingaben, bevor es die Produktion tut: Du erzeugst Tausende fehlerhafte, abgeschnittene und feindselige Payloads und prüfst, ob dein Code jede davon sauber ablehnt.

Dieser Guide zeigt, wie du einen Decoder fuzzbar machst, ihn mit Hypothesis (Property-Based) und Atheris (Coverage-geführt) testest, dasselbe in JavaScript mit fast-check umsetzt und die Funde über einen echten Broker abspielst.

Was gefuzzt werden sollte

Der Broker validiert das MQTT-Paketformat, daher sieht dein Code selten ein kaputtes Paket. Er sieht kaputte Payloads, und genau daher kommen die Abstürze:

Fang mit dem Decoder an: Er ist eine reine Funktion, lässt sich millionenfach aufrufen und ist die Quelle der meisten JSON-Parse-Fehler in MQTT-Consumern.

Den Decoder fuzzbar machen

Ein fuzzbarer Decoder nimmt bytes entgegen und gibt entweder ein typisiertes Objekt zurück oder wirft eine dokumentierte Exception. Alles andere (ein KeyError, TypeError, RecursionError, ein stilles NaN) ist ein Bug.

import json
import math
from dataclasses import dataclass

MAX_PAYLOAD = 4096


class PayloadError(ValueError):
    """Die einzige Exception, die decode_reading werfen darf."""


@dataclass(frozen=True)
class Reading:
    device: str
    temp_c: float
    ts: int


def decode_reading(payload: bytes) -> Reading:
    if len(payload) > MAX_PAYLOAD:
        raise PayloadError("payload too large")
    try:
        data = json.loads(payload.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError, RecursionError) as exc:
        raise PayloadError(f"invalid JSON: {exc}") from exc
    if not isinstance(data, dict):
        raise PayloadError("expected a JSON object")

    device, temp, ts = data.get("device"), data.get("temp_c"), data.get("ts")
    if not isinstance(device, str) or not 1 <= len(device) <= 64:
        raise PayloadError("device must be a 1-64 char string")
    if isinstance(temp, bool) or not isinstance(temp, (int, float)) or not math.isfinite(temp):
        raise PayloadError("temp_c must be a finite number")
    if isinstance(ts, bool) or not isinstance(ts, int) or not 0 <= ts < 2**32:
        raise PayloadError("ts must be a 32-bit unix timestamp")
    return Reading(device, float(temp), ts)


def encode_reading(r: Reading) -> bytes:
    return json.dumps({"device": r.device, "temp_c": r.temp_c, "ts": r.ts}).encode()

Jede Prüfung in diesem Decoder existiert, weil ein Fuzzer sie schnell aufdeckt:

• json.loads akzeptiert standardmäßig NaN und Infinity, die anschließend Durchschnittswerte und Datenbankspalten verseuchen.
• True besteht isinstance(x, int), weil bool eine Unterklasse von int ist.
• Ausreichend tief verschachteltes [[[[...]]]] wirft RecursionError, und das ist kein JSONDecodeError.
• Beliebige Bytes sind selten gültiges UTF-8 (siehe UTF-8-Kodierungsfehler).
• Die Größenbegrenzung verhindert, dass eine einzelne übergroße Nachricht den Speicher auffrisst. Mit dem Paketgrößen-Rechner findest du ein realistisches Limit.

Property-Based Tests mit Hypothesis

Hypothesis erzeugt Eingaben aus Strategien, schrumpft jeden Fehlschlag auf das kleinste Beispiel und merkt ihn sich für den nächsten Lauf. Drei Properties decken den Großteil eines Decoders ab:

import json
from hypothesis import given, settings, strategies as st
from decoder import PayloadError, Reading, decode_reading, encode_reading

# 1. Zufällige Bytes: niemals etwas anderes als PayloadError
@given(st.binary(max_size=8192))
def test_random_bytes_never_crash(payload):
    try:
        decode_reading(payload)
    except PayloadError:
        pass

# 2. Zufälliges, aber gültiges JSON: weiterhin nur PayloadError
json_values = st.recursive(
    st.none() | st.booleans() | st.integers() | st.floats() | st.text(),
    lambda children: st.lists(children) | st.dictionaries(st.text(), children),
    max_leaves=30,
)

@given(json_values)
@settings(max_examples=500)
def test_any_json_never_crashes(value):
    try:
        decode_reading(json.dumps(value).encode())
    except PayloadError:
        pass

# 3. Round Trip: alles, was wir kodieren, können wir dekodieren
readings = st.builds(
    Reading,
    device=st.text(min_size=1, max_size=64),
    temp_c=st.floats(min_value=-100, max_value=200, allow_nan=False),
    ts=st.integers(min_value=0, max_value=2**32 - 1),
)

@given(readings)
def test_round_trip(r):
    assert decode_reading(encode_reading(r)) == r

Die zweite Strategie ist wichtiger als die erste. Zufällige Bytes kommen fast nie an json.loads vorbei. Zufälliges JSON erreicht die Feldprüfungen, und genau dort tauchen die Bugs mit bool als int und mit NaN auf. Starte es mit pytest -q. Ein Fehlschlag gibt die minimale Eingabe aus, zum Beispiel {'device': '0', 'temp_c': True, 'ts': 0}.

Coverage-geführtes Fuzzing mit Atheris

Atheris ist ein libFuzzer-basierter Fuzzer für Python. Er mutiert Bytes und behält die Eingaben, die neue Branches erreichen, und gräbt damit tiefer als zufällige Generierung, wenn ein Decoder viele Pfade hat (mehrere Schema-Versionen, Binärformate, Kompression).

import sys
import atheris

with atheris.instrument_imports():
    from decoder import PayloadError, decode_reading


def test_one_input(data: bytes) -> None:
    try:
        decode_reading(data)
    except PayloadError:
        pass


atheris.Setup(sys.argv, test_one_input)
atheris.Fuzz()

pip install atheris
mkdir -p corpus
python fuzz_decoder.py corpus/ -max_total_time=300 -max_len=8192

Jede andere Exception stoppt den Fuzzer und schreibt eine Datei crash-<hash> mit der Eingabe. Behebe den Decoder, nimm die Datei in einen Regressionstest auf und starte erneut. Das Verzeichnis corpus/ enthält die interessanten Eingaben: committe es oder cache es in der CI.

Den Korpus mit echtem Traffic füttern

Ein Fuzzer, der mit echten Nachrichten startet, findet Bugs schneller als einer, der bei null anfängt. Zeichne eine Stichprobe von Payloads aus deinem Test-Topic auf:

import hashlib, pathlib
import paho.mqtt.client as mqtt

out = pathlib.Path("corpus")
out.mkdir(exist_ok=True)

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe("testmqtt/fuzz/telemetry/#", qos=0)

def on_message(client, userdata, msg):
    name = hashlib.sha1(msg.payload).hexdigest()
    (out / name).write_bytes(msg.payload)
    print("gespeichert", name, len(msg.payload), "Bytes")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.username_pw_set("username", "password")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883)
client.loop_forever()

Dieselbe Idee in JavaScript mit fast-check

Für einen MQTT.js-5-Consumer übernimmt fast-check die Rolle von Hypothesis und läuft in Vitest oder Jest:

import fc from "fast-check";
import { test } from "vitest";
import { decodeReading, PayloadError } from "./decoder";

test("decodeReading wirft nur PayloadError", () => {
  fc.assert(
    fc.property(
      fc.oneof(
        fc.uint8Array({ maxLength: 8192 }),
        fc.jsonValue().map((v) => new TextEncoder().encode(JSON.stringify(v))),
      ),
      (bytes) => {
        try {
          decodeReading(Buffer.from(bytes));
        } catch (err) {
          if (!(err instanceof PayloadError)) throw err;
        }
      },
    ),
    { numRuns: 2000 },
  );
});

JSON.parse akzeptiert in JavaScript kein NaN, liefert aber anstandslos null, Arrays und Strings, wo du ein Objekt erwartet hast. Und __proto__-Keys verdienen eine Prüfung, wenn du Payloads in bestehende Objekte mergst.

Den Handler fuzzen, nicht nur den Decoder

Der Decoder kann perfekt sein und der Subscriber trotzdem abstürzen. In paho-mqtt 2.x propagiert eine in on_message geworfene Exception standardmäßig aus dem Netzwerk-Loop heraus (sofern du nicht client.suppress_exceptions = True setzt) und beendet damit loop_forever() oder den Thread von loop_start(). Kapsle den Handler, damit eine einzelne schlechte Nachricht den Consumer nicht lahmlegen kann:

import logging
from decoder import PayloadError, decode_reading

log = logging.getLogger("consumer")

def on_message(client, userdata, msg):
    try:
        reading = decode_reading(msg.payload)
    except PayloadError as exc:
        log.warning("Nachricht auf %s abgelehnt: %s", msg.topic, exc)
        client.publish("testmqtt/fuzz/deadletter", msg.payload, qos=1)
        return
    userdata["store"](reading)

Fuzze danach den Handler selbst mit einem Fake-msg-Objekt (ein SimpleNamespace(topic=..., payload=...) genügt) und einem Fake-Client, wie bei Unit-Tests mit einem Mock-Broker. Die Property lautet: Der Handler kehrt immer zurück, und jede abgelehnte Payload landet auf dem Dead-Letter-Topic.

Den Crash-Korpus über einen Broker abspielen

Sobald die Fuzzer auf Unit-Ebene grün sind, publishe den gespeicherten Korpus und die Crash-Dateien auf einen Test-Topic und beobachte, ob der echte Consumer überlebt:

import pathlib, time
import paho.mqtt.client as mqtt

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.username_pw_set("username", "password")
client.connect("broker.connectmq.com", 1883)
client.loop_start()

for path in sorted(pathlib.Path("corpus").iterdir()):
    client.publish("testmqtt/fuzz/telemetry/replay", path.read_bytes(), qos=1).wait_for_publish()
    time.sleep(0.01)

client.loop_stop()
client.disconnect()

Abonniere währenddessen testmqtt/fuzz/# im Online-MQTT-Client: Du solltest jede Payload einmal auf dem Telemetrie-Topic sehen, die abgelehnten auf dem Dead-Letter-Topic, und der Consumer sollte am Ende noch verbunden sein.

Replay im Browser verfolgen

Ab in die CI

1. Lass die Properties von Hypothesis und fast-check bei jedem Commit laufen. Sie dauern Sekunden.
2. Lass Atheris in einem nächtlichen Job fünf bis fünfzehn Minuten laufen, mit gecachtem Korpus zwischen den Läufen.
3. Mach aus jeder Crash-Datei einen benannten Regressionstest, bevor du den Bug behebst.
4. Lass den Build bei jedem Exception-Typ fehlschlagen, der nicht der dokumentierte ist.

Unter MQTT-Tests in GitHub Actions findest du einen Workflow, der für den Replay-Schritt auch einen Broker startet.

Fuzzing braucht keinen großen Aufbau. Ein Decoder mit einer einzigen dokumentierten Exception, drei Properties und ein nächtlicher Fuzzer finden das NaN, die boolesche Temperatur und das 10.000-fach verschachtelte Array lange bevor ein fehlerhaftes Gerät sie in die Produktion schickt.

Häufig gestellte Fragen