MQTT Fuzz Testing — Harden Payload Parsers and Clients

Every subscriber in an MQTT system trusts bytes it did not produce. A device with a firmware bug, a half-written retained message or a test script left running can publish anything to a topic, and if your decoder throws on it, the worker crashes, the message is redelivered and the crash repeats. Fuzz testing MQTT payload parsers finds those inputs before production does: you generate thousands of malformed, truncated and hostile payloads and check that your code rejects each one cleanly.

This guide shows how to make a decoder fuzzable, test it with Hypothesis (property-based) and Atheris (coverage-guided), do the same in JavaScript with fast-check, and replay the findings through a real broker.

What to fuzz

The broker validates the MQTT packet format, so your code rarely sees a broken packet. It sees broken payloads, and those are where crashes come from:

Target	Typical bugs fuzzing finds
Payload decoder (JSON, CBOR, Protobuf, CSV)	Unhandled exceptions, invalid UTF-8, deep nesting, huge numbers
Message handler / `on_message` callback	Exceptions that stop the network loop, missing fields, wrong types
Topic parser (`site/+/device/#` → IDs)	Empty levels, extra levels, unexpected characters
Command dispatcher	Unknown commands, out-of-range arguments

Start with the decoder: it is a pure function, easy to call millions of times, and the source of most JSON parse errors in MQTT consumers.

Make the decoder fuzzable

A fuzzable decoder takes bytes and either returns a typed object or raises one documented exception. Anything else (a KeyError, TypeError, RecursionError, a silent NaN) is a bug.

decoder.py

import json
import math
from dataclasses import dataclass

MAX_PAYLOAD = 4096


class PayloadError(ValueError):
    """The only exception decode_reading may raise."""


@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()

Each check in this decoder exists because a fuzzer finds it fast:

json.loads accepts NaN and Infinity by default, which then poison averages and database columns.
True passes isinstance(x, int) because bool is a subclass of int.
[[[[...]]]] nested deeply enough raises RecursionError, which is not a JSONDecodeError.
Arbitrary bytes are rarely valid UTF-8 (see UTF-8 encoding errors).
The size cap stops a single oversized message from eating memory. Use the packet size calculator to pick a realistic limit.

Property-based tests with Hypothesis

Hypothesis generates inputs from strategies, shrinks any failure to the smallest example and remembers it for the next run. Three properties cover most of a decoder:

test_decoder_properties.py

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

# 1. Random bytes: never anything but PayloadError
@given(st.binary(max_size=8192))
def test_random_bytes_never_crash(payload):
    try:
        decode_reading(payload)
    except PayloadError:
        pass

# 2. Random but valid JSON: still only 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: everything we encode, we can decode
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

The second strategy matters more than the first. Random bytes almost never get past json.loads. Random JSON reaches the field checks, and that is where the bool-is-an-int and NaN bugs show up. Run it with pytest -q. A failure prints the minimal input, for example {'device': '0', 'temp_c': True, 'ts': 0}.

Coverage-guided fuzzing with Atheris

Atheris is a libFuzzer-based fuzzer for Python. It mutates bytes and keeps the inputs that reach new branches, so it digs deeper than random generation when a decoder has many paths (several schema versions, binary formats, compression).

fuzz_decoder.py

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()

Run for five minutes with a saved corpus

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

Any other exception stops the fuzzer and writes a crash-<hash> file containing the input. Fix the decoder, add the file to a regression test, and re-run. The corpus/ directory keeps the interesting inputs, so commit it or cache it in CI.

Seed the corpus with real traffic

A fuzzer that starts from real messages finds bugs faster than one that starts from nothing. Capture a sample of payloads from your test topic:

capture_seeds.py

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("saved", 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()

The same idea in JavaScript with fast-check

For an MQTT.js 5 consumer, fast-check plays the role of Hypothesis and runs inside Vitest or Jest:

decoder.fuzz.test.ts

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

test("decodeReading only throws 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 in JavaScript does not accept NaN, but it happily returns null, arrays and strings where you expected an object, and __proto__ keys deserve a check if you merge payloads into existing objects.

Fuzz the handler, not only the decoder

The decoder can be perfect and the subscriber can still crash. In paho-mqtt 2.x an exception raised inside on_message propagates out of the network loop by default (unless you set client.suppress_exceptions = True), which stops loop_forever() or the loop_start() thread. Wrap the handler so one bad message cannot take the consumer down:

handler.py

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("rejected message on %s: %s", msg.topic, exc)
        client.publish("testmqtt/fuzz/deadletter", msg.payload, qos=1)
        return
    userdata["store"](reading)

Then fuzz the handler itself with a fake msg object (a SimpleNamespace(topic=..., payload=...) is enough) and a fake client, as in unit tests with a mock broker. The property is: the handler always returns, and every rejected payload lands on the dead-letter topic.

Replay the crash corpus through a broker

Once the unit-level fuzzers are green, publish the saved corpus and crash files to a test topic and watch the real consumer survive:

replay_corpus.py

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()

Subscribe to testmqtt/fuzz/# in the online MQTT client while it runs: you should see each payload once on the telemetry topic, rejected ones on the dead-letter topic, and the consumer still connected at the end.

Watch the replay in your browser

Put it in CI

Run the Hypothesis and fast-check properties on every commit. They take seconds.
Run Atheris for five to fifteen minutes in a nightly job, with the corpus cached between runs.
Turn every crash file into a named regression test before fixing the bug.
Fail the build on any exception type other than your documented one.

See MQTT testing in GitHub Actions for a workflow that also starts a broker for the replay step.

Fuzzing does not need a big setup. A decoder with one documented exception, three properties and a nightly fuzzer will find the NaN, the boolean temperature and the 10,000-level nested array long before a misbehaving device sends them to production.

Frequently asked questions

What is the difference between fuzzing and property-based testing?

Property-based testing (Hypothesis, fast-check) generates structured inputs from strategies and checks a rule such as 'decode never raises anything but PayloadError'. Coverage-guided fuzzing (Atheris, libFuzzer) mutates raw bytes and keeps inputs that reach new code. Use both: properties in every test run, fuzzing for longer sessions.

Do I need a broker to fuzz an MQTT client?

No. Most bugs live in the code that turns a payload into a domain object, and you can fuzz that function directly at thousands of inputs per second. Use a broker only for a final replay of the crash corpus through the real subscriber.

How long should a fuzzing run take?

Keep property tests to a few seconds so they run on every commit. Run a coverage-guided fuzzer for minutes in a nightly job, and longer after parser changes. Save the corpus between runs so each session starts where the last one stopped.

Fuzz Testing MQTT Clients and Payload Parsers