Fuzz testing de clientes MQTT e parsers de payload

Todo subscriber de um sistema MQTT confia em bytes que ele não produziu. Um dispositivo com bug de firmware, uma mensagem retida escrita pela metade ou um script de teste esquecido rodando podem publicar qualquer coisa em um topic, e se o seu decodificador lançar uma exceção, o worker cai, a mensagem é reentregue e a falha se repete. O fuzz testing de parsers de payload MQTT encontra essas entradas antes da produção: você gera milhares de payloads malformados, truncados e hostis e verifica se o seu código rejeita cada um de forma limpa.

Este guia mostra como deixar um decodificador pronto para fuzzing, testá-lo com Hypothesis (baseado em propriedades) e Atheris (guiado por cobertura), fazer o mesmo em JavaScript com fast-check e reproduzir os achados em um broker real.

O que submeter a fuzzing

O broker valida o formato dos pacotes MQTT, então seu código raramente vê um pacote quebrado. Ele vê payloads quebrados, e é daí que vêm as falhas:

Comece pelo decodificador: é uma função pura, fácil de chamar milhões de vezes, e a origem da maioria dos erros de parse de JSON em consumidores MQTT.

Deixe o decodificador pronto para fuzzing

Um decodificador pronto para fuzzing recebe bytes e retorna um objeto tipado ou lança uma exceção documentada. Qualquer outra coisa (um KeyError, um TypeError, um RecursionError, um NaN silencioso) é bug.

import json
import math
from dataclasses import dataclass

MAX_PAYLOAD = 4096


class PayloadError(ValueError):
    """A única exceção que decode_reading pode lançar."""


@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 verificação desse decodificador existe porque um fuzzer a encontra rapidamente:

• json.loads aceita NaN e Infinity por padrão, que depois contaminam médias e colunas do banco de dados.
• True passa em isinstance(x, int) porque bool é subclasse de int.
• [[[[...]]]] com aninhamento suficiente lança RecursionError, que não é um JSONDecodeError.
• Bytes arbitrários raramente são UTF-8 válido (veja erros de codificação UTF-8).
• O limite de tamanho impede que uma única mensagem gigante consuma a memória. Use a calculadora de tamanho de pacote para escolher um limite realista.

Testes baseados em propriedades com Hypothesis

O Hypothesis gera entradas a partir de estratégias, reduz qualquer falha ao menor exemplo possível e o lembra na próxima execução. Três propriedades cobrem quase todo um decodificador:

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

# 1. Bytes aleatórios: nunca algo além de PayloadError
@given(st.binary(max_size=8192))
def test_random_bytes_never_crash(payload):
    try:
        decode_reading(payload)
    except PayloadError:
        pass

# 2. JSON aleatório, mas válido: ainda só 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 e volta: tudo o que codificamos, conseguimos 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

A segunda estratégia importa mais que a primeira. Bytes aleatórios quase nunca passam de json.loads. JSON aleatório chega às verificações de campos, e é ali que aparecem os bugs do bool que é int e do NaN. Rode com pytest -q. Uma falha imprime a entrada mínima, por exemplo {'device': '0', 'temp_c': True, 'ts': 0}.

Fuzzing guiado por cobertura com Atheris

O Atheris é um fuzzer para Python baseado no libFuzzer. Ele muta bytes e guarda as entradas que alcançam novos ramos, então vai mais fundo que a geração aleatória quando o decodificador tem muitos caminhos (várias versões de schema, formatos binários, compressão).

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

Qualquer outra exceção para o fuzzer e grava um arquivo crash-<hash> com a entrada. Corrija o decodificador, adicione o arquivo a um teste de regressão e rode de novo. O diretório corpus/ guarda as entradas interessantes: faça commit dele ou coloque-o em cache no CI.

Alimente o corpus com tráfego real

Um fuzzer que parte de mensagens reais encontra bugs mais rápido que um que parte do zero. Capture uma amostra de payloads do seu topic de teste:

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

A mesma ideia em JavaScript com fast-check

Para um consumidor com MQTT.js 5, o fast-check faz o papel do Hypothesis e roda dentro do Vitest ou do Jest:

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

test("decodeReading só lança 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 em JavaScript não aceita NaN, mas devolve sem reclamar null, arrays e strings onde você esperava um objeto, e chaves __proto__ merecem uma verificação se você mescla payloads em objetos existentes.

Faça fuzzing do handler, não só do decodificador

O decodificador pode ser perfeito e o subscriber ainda cair. No paho-mqtt 2.x, uma exceção lançada dentro de on_message se propaga para fora do loop de rede por padrão (a menos que você defina client.suppress_exceptions = True), o que para o loop_forever() ou a thread do loop_start(). Envolva o handler para que uma mensagem ruim não derrube o 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("mensagem rejeitada em %s: %s", msg.topic, exc)
        client.publish("testmqtt/fuzz/deadletter", msg.payload, qos=1)
        return
    userdata["store"](reading)

Depois faça fuzzing do próprio handler com um objeto msg falso (um SimpleNamespace(topic=..., payload=...) basta) e um cliente falso, como nos testes unitários com broker simulado. A propriedade é: o handler sempre retorna, e todo payload rejeitado vai para o topic de dead letter.

Reproduza o corpus de falhas em um broker

Quando os fuzzers de nível unitário estiverem verdes, publique o corpus salvo e os arquivos de crash em um topic de teste e veja o consumidor real sobreviver:

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

Assine testmqtt/fuzz/# no cliente MQTT online enquanto ele roda: você deve ver cada payload uma vez no topic de telemetria, os rejeitados no topic de dead letter e o consumidor ainda conectado no final.

Acompanhe o replay no navegador

Coloque no CI

1. Rode as propriedades do Hypothesis e do fast-check em todo commit. Levam segundos.
2. Rode o Atheris de cinco a quinze minutos em um job noturno, com o corpus em cache entre execuções.
3. Transforme cada arquivo de crash em um teste de regressão nomeado antes de corrigir o bug.
4. Faça o build falhar diante de qualquer tipo de exceção diferente do documentado.

Veja testes MQTT no GitHub Actions para um workflow que também sobe um broker para a etapa de replay.

Fuzzing não exige uma grande estrutura. Um decodificador com uma única exceção documentada, três propriedades e um fuzzer noturno vão encontrar o NaN, a temperatura booleana e o array aninhado em 10.000 níveis muito antes de um dispositivo com defeito enviá-los para a produção.

Perguntas frequentes