Fuzz testing des clients MQTT et des parsers de payload

Chaque subscriber d'un système MQTT fait confiance à des octets qu'il n'a pas produits. Un appareil avec un bug de firmware, un message retenu à moitié écrit ou un script de test oublié peut publier n'importe quoi sur un topic, et si votre décodeur lève une exception, le worker plante, le message est redistribué et le crash se répète. Le fuzz testing des parsers de payload MQTT trouve ces entrées avant la production : vous générez des milliers de payloads malformés, tronqués et hostiles, et vous vérifiez que votre code rejette proprement chacun d'eux.

Ce guide montre comment rendre un décodeur fuzzable, le tester avec Hypothesis (basé sur les propriétés) et Atheris (guidé par la couverture), faire la même chose en JavaScript avec fast-check, puis rejouer les trouvailles à travers un vrai broker.

Quoi fuzzer

Le broker valide le format des paquets MQTT, donc votre code voit rarement un paquet cassé. Il voit des payloads cassés, et c'est de là que viennent les crashs :

Commencez par le décodeur : c'est une fonction pure, facile à appeler des millions de fois, et la source de la plupart des erreurs de parsing JSON dans les consommateurs MQTT.

Rendre le décodeur fuzzable

Un décodeur fuzzable prend des bytes et renvoie un objet typé ou lève une exception documentée. Tout le reste (un KeyError, un TypeError, un RecursionError, un NaN silencieux) est un bug.

import json
import math
from dataclasses import dataclass

MAX_PAYLOAD = 4096


class PayloadError(ValueError):
    """La seule exception que decode_reading peut lever."""


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

Chaque vérification de ce décodeur existe parce qu'un fuzzer la découvre très vite :

• json.loads accepte NaN et Infinity par défaut, qui faussent ensuite les moyennes et les colonnes de la base.
• True passe isinstance(x, int) car bool est une sous-classe de int.
• [[[[...]]]] suffisamment imbriqué lève RecursionError, qui n'est pas un JSONDecodeError.
• Des octets arbitraires sont rarement de l'UTF-8 valide (voir erreurs d'encodage UTF-8).
• La limite de taille empêche un seul message surdimensionné de consommer la mémoire. Utilisez le calculateur de taille de paquet pour choisir une limite réaliste.

Tests basés sur les propriétés avec Hypothesis

Hypothesis génère des entrées à partir de stratégies, réduit tout échec au plus petit exemple et le mémorise pour l'exécution suivante. Trois propriétés couvrent l'essentiel d'un décodeur :

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

# 1. Octets aléatoires : jamais autre chose que PayloadError
@given(st.binary(max_size=8192))
def test_random_bytes_never_crash(payload):
    try:
        decode_reading(payload)
    except PayloadError:
        pass

# 2. JSON aléatoire mais valide : toujours uniquement 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. Aller-retour : tout ce qu'on encode, on sait le décoder
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

La deuxième stratégie compte davantage que la première. Des octets aléatoires ne passent presque jamais json.loads. Du JSON aléatoire atteint les vérifications de champs, et c'est là qu'apparaissent les bugs du bool qui est un int et du NaN. Lancez-le avec pytest -q. Un échec affiche l'entrée minimale, par exemple {'device': '0', 'temp_c': True, 'ts': 0}.

Fuzzing guidé par la couverture avec Atheris

Atheris est un fuzzer Python basé sur libFuzzer. Il fait muter des octets et conserve les entrées qui atteignent de nouvelles branches, ce qui lui permet d'aller plus loin que la génération aléatoire quand un décodeur a beaucoup de chemins (plusieurs versions de schéma, formats binaires, compression).

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

Toute autre exception arrête le fuzzer et écrit un fichier crash-<hash> contenant l'entrée. Corrigez le décodeur, ajoutez le fichier à un test de régression et relancez. Le dossier corpus/ conserve les entrées intéressantes : committez-le ou mettez-le en cache dans la CI.

Alimenter le corpus avec du trafic réel

Un fuzzer qui part de vrais messages trouve des bugs plus vite qu'un fuzzer qui part de rien. Capturez un échantillon de payloads depuis votre topic de test :

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("enregistré", name, len(msg.payload), "octets")

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

La même idée en JavaScript avec fast-check

Pour un consommateur MQTT.js 5, fast-check joue le rôle d'Hypothesis et s'exécute dans Vitest ou Jest :

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

test("decodeReading ne lève que 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 },
  );
});

En JavaScript, JSON.parse n'accepte pas NaN, mais renvoie sans broncher null, des tableaux et des chaînes là où vous attendiez un objet, et les clés __proto__ méritent une vérification si vous fusionnez des payloads dans des objets existants.

Fuzzer le handler, pas seulement le décodeur

Le décodeur peut être parfait et le subscriber planter quand même. Dans paho-mqtt 2.x, une exception levée dans on_message se propage par défaut hors de la boucle réseau (sauf si vous définissez client.suppress_exceptions = True), ce qui arrête loop_forever() ou le thread de loop_start(). Encapsulez le handler pour qu'un mauvais message ne puisse pas faire tomber le consommateur :

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

Fuzzez ensuite le handler lui-même avec un faux objet msg (un SimpleNamespace(topic=..., payload=...) suffit) et un faux client, comme dans les tests unitaires avec un broker simulé. La propriété est la suivante : le handler retourne toujours, et chaque payload rejeté arrive sur le topic de dead letter.

Rejouer le corpus de crashs via un broker

Une fois les fuzzers unitaires au vert, publiez le corpus sauvegardé et les fichiers de crash sur un topic de test et vérifiez que le vrai consommateur survit :

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

Abonnez-vous à testmqtt/fuzz/# dans le client MQTT en ligne pendant l'exécution : vous devriez voir chaque payload une fois sur le topic de télémétrie, les rejetés sur le topic de dead letter, et le consommateur toujours connecté à la fin.

Suivre le replay dans le navigateur

L'intégrer à la CI

1. Exécutez les propriétés Hypothesis et fast-check à chaque commit. Elles prennent quelques secondes.
2. Lancez Atheris pendant cinq à quinze minutes dans un job nocturne, avec le corpus mis en cache entre les exécutions.
3. Transformez chaque fichier de crash en test de régression nommé avant de corriger le bug.
4. Faites échouer le build pour tout type d'exception autre que celui documenté.

Consultez les tests MQTT dans GitHub Actions pour un workflow qui démarre aussi un broker pour l'étape de replay.

Le fuzzing ne demande pas une grosse infrastructure. Un décodeur avec une seule exception documentée, trois propriétés et un fuzzer nocturne trouveront le NaN, la température booléenne et le tableau imbriqué sur 10 000 niveaux bien avant qu'un appareil défaillant ne les envoie en production.

Questions fréquentes