Fuzz testing de clientes MQTT y parsers de payload

Cada subscriber de un sistema MQTT confía en bytes que no ha producido. Un dispositivo con un bug de firmware, un mensaje retenido escrito a medias o un script de prueba olvidado pueden publicar cualquier cosa en un topic, y si tu decodificador lanza una excepción, el worker se cae, el mensaje se vuelve a entregar y el fallo se repite. El fuzz testing de parsers de payload MQTT encuentra esas entradas antes que producción: generas miles de payloads malformados, truncados y hostiles y compruebas que tu código rechaza cada uno de forma limpia.

Esta guía muestra cómo hacer un decodificador apto para fuzzing, probarlo con Hypothesis (basado en propiedades) y Atheris (guiado por cobertura), hacer lo mismo en JavaScript con fast-check y reproducir los hallazgos a través de un broker real.

Qué someter a fuzzing

El broker valida el formato de los paquetes MQTT, así que tu código rara vez ve un paquete roto. Ve payloads rotos, y de ahí vienen los fallos:

Empieza por el decodificador: es una función pura, fácil de llamar millones de veces, y el origen de la mayoría de los errores de parseo JSON en consumidores MQTT.

Haz que el decodificador admita fuzzing

Un decodificador apto para fuzzing recibe bytes y devuelve un objeto tipado o lanza una excepción documentada. Cualquier otra cosa (un KeyError, un TypeError, un RecursionError, un NaN silencioso) es un bug.

import json
import math
from dataclasses import dataclass

MAX_PAYLOAD = 4096


class PayloadError(ValueError):
    """La única excepción que puede lanzar decode_reading."""


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

Cada comprobación de este decodificador existe porque un fuzzer la descubre enseguida:

• json.loads acepta NaN e Infinity por defecto, que luego contaminan medias y columnas de la base de datos.
• True pasa isinstance(x, int) porque bool es una subclase de int.
• [[[[...]]]] con suficiente anidamiento lanza RecursionError, que no es un JSONDecodeError.
• Unos bytes arbitrarios rara vez son UTF-8 válido (consulta errores de codificación UTF-8).
• El límite de tamaño impide que un único mensaje enorme se coma la memoria. Usa la calculadora de tamaño de paquete para elegir un límite realista.

Pruebas basadas en propiedades con Hypothesis

Hypothesis genera entradas a partir de estrategias, reduce cualquier fallo al ejemplo más pequeño y lo recuerda para la siguiente ejecución. Tres propiedades cubren casi todo un decodificador:

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

# 1. Bytes aleatorios: nunca otra cosa que PayloadError
@given(st.binary(max_size=8192))
def test_random_bytes_never_crash(payload):
    try:
        decode_reading(payload)
    except PayloadError:
        pass

# 2. JSON aleatorio pero válido: también solo 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. Ida y vuelta: todo lo que codificamos, lo podemos decodificar
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 segunda estrategia importa más que la primera. Los bytes aleatorios casi nunca superan json.loads. El JSON aleatorio llega a las comprobaciones de campos, y ahí aparecen los bugs de bool que es un int y de NaN. Ejecútalo con pytest -q. Un fallo imprime la entrada mínima, por ejemplo {'device': '0', 'temp_c': True, 'ts': 0}.

Fuzzing guiado por cobertura con Atheris

Atheris es un fuzzer para Python basado en libFuzzer. Muta bytes y conserva las entradas que alcanzan ramas nuevas, así que profundiza más que la generación aleatoria cuando un decodificador tiene muchos caminos (varias versiones de esquema, formatos binarios, compresión).

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

Cualquier otra excepción detiene el fuzzer y escribe un fichero crash-<hash> con la entrada. Corrige el decodificador, añade el fichero a un test de regresión y vuelve a ejecutar. El directorio corpus/ guarda las entradas interesantes: haz commit de él o cachéalo en CI.

Alimenta el corpus con tráfico real

Un fuzzer que parte de mensajes reales encuentra bugs antes que uno que parte de cero. Captura una muestra de payloads de tu topic de pruebas:

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

La misma idea en JavaScript con fast-check

Para un consumidor con MQTT.js 5, fast-check cumple el papel de Hypothesis y se ejecuta dentro de Vitest o Jest:

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

test("decodeReading solo lanza 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 en JavaScript no acepta NaN, pero devuelve sin problema null, arrays y strings donde esperabas un objeto, y las claves __proto__ merecen una comprobación si fusionas payloads con objetos existentes.

Haz fuzzing del handler, no solo del decodificador

El decodificador puede ser perfecto y el subscriber seguir cayéndose. En paho-mqtt 2.x, una excepción lanzada dentro de on_message se propaga fuera del bucle de red por defecto (salvo que pongas client.suppress_exceptions = True), lo que detiene loop_forever() o el hilo de loop_start(). Envuelve el handler para que un mensaje malo no pueda tumbar al consumidor:

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

Después haz fuzzing del propio handler con un objeto msg falso (basta con un SimpleNamespace(topic=..., payload=...)) y un cliente falso, como en los tests unitarios con un broker simulado. La propiedad es: el handler siempre retorna y cada payload rechazado acaba en el topic de dead letter.

Reproduce el corpus de fallos a través de un broker

Cuando los fuzzers a nivel unitario estén en verde, publica el corpus guardado y los ficheros de fallo en un topic de pruebas y comprueba que el consumidor real sobrevive:

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

Suscríbete a testmqtt/fuzz/# en el cliente MQTT online mientras se ejecuta: deberías ver cada payload una vez en el topic de telemetría, los rechazados en el topic de dead letter y el consumidor todavía conectado al final.

Observa el replay en tu navegador

Llévalo a CI

1. Ejecuta las propiedades de Hypothesis y fast-check en cada commit. Tardan segundos.
2. Ejecuta Atheris entre cinco y quince minutos en un job nocturno, con el corpus cacheado entre ejecuciones.
3. Convierte cada fichero de fallo en un test de regresión con nombre antes de corregir el bug.
4. Haz fallar el build ante cualquier tipo de excepción distinto del documentado.

Consulta pruebas MQTT en GitHub Actions para un workflow que además levanta un broker para el paso de replay.

El fuzzing no necesita una gran infraestructura. Un decodificador con una única excepción documentada, tres propiedades y un fuzzer nocturno encontrarán el NaN, la temperatura booleana y el array anidado 10.000 niveles mucho antes de que un dispositivo defectuoso los envíe a producción.

Preguntas frecuentes