MQTT İstemcileri ve Payload Parser'ları için Fuzz Testi

Bir MQTT sistemindeki her subscriber, kendisinin üretmediği baytlara güvenir. Firmware hatası olan bir cihaz, yarım yazılmış bir retained mesaj ya da açık unutulmuş bir test betiği bir topic'e her şeyi publish edebilir; decoder'ınız bu durumda exception fırlatırsa worker çöker, mesaj yeniden teslim edilir ve çökme tekrarlanır. MQTT payload parser'larının fuzz testi, bu girdileri production'dan önce bulur: binlerce bozuk, kesik ve kötü niyetli payload üretir ve kodunuzun her birini temiz şekilde reddettiğini doğrularsınız.

Bu rehber bir decoder'ı fuzz edilebilir hâle getirmeyi, Hypothesis (özellik tabanlı) ve Atheris (kapsam güdümlü) ile test etmeyi, aynısını JavaScript'te fast-check ile yapmayı ve bulguları gerçek bir broker üzerinden yeniden oynatmayı gösteriyor.

Neyi fuzz etmeli

Broker MQTT paket formatını doğrular; bu yüzden kodunuz nadiren bozuk bir paket görür. Bozuk payload'lar görür ve çökmeler de oradan gelir:

Decoder ile başlayın: saf bir fonksiyondur, milyonlarca kez çağrılması kolaydır ve MQTT consumer'larındaki JSON parse hatalarının çoğunun kaynağıdır.

Decoder'ı fuzz edilebilir yapın

Fuzz edilebilir bir decoder bytes alır ve ya tipli bir nesne döndürür ya da tek bir belgelenmiş exception fırlatır. Bunun dışındaki her şey (bir KeyError, TypeError, RecursionError, sessiz bir NaN) bir hatadır.

import json
import math
from dataclasses import dataclass

MAX_PAYLOAD = 4096


class PayloadError(ValueError):
    """decode_reading'in fırlatabileceği tek exception."""


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

Bu decoder'daki her kontrol, bir fuzzer onu çok çabuk bulduğu için orada:

• json.loads varsayılan olarak NaN ve Infinity kabul eder; bunlar sonra ortalamaları ve veritabanı sütunlarını bozar.
• bool, int'in alt sınıfı olduğu için True, isinstance(x, int) kontrolünden geçer.
• Yeterince derin iç içe geçmiş [[[[...]]]], JSONDecodeError olmayan bir RecursionError fırlatır.
• Rastgele baytlar nadiren geçerli UTF-8'dir (bkz. UTF-8 kodlama hataları).
• Boyut sınırı, tek bir aşırı büyük mesajın belleği tüketmesini engeller. Gerçekçi bir sınır seçmek için paket boyutu hesaplayıcısını kullanın.

Hypothesis ile özellik tabanlı testler

Hypothesis stratejilerden girdiler üretir, her başarısızlığı en küçük örneğe indirger ve bir sonraki çalıştırma için hatırlar. Üç özellik bir decoder'ın büyük kısmını kapsar:

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

# 1. Rastgele baytlar: PayloadError dışında asla bir şey yok
@given(st.binary(max_size=8192))
def test_random_bytes_never_crash(payload):
    try:
        decode_reading(payload)
    except PayloadError:
        pass

# 2. Rastgele ama geçerli JSON: yine yalnızca 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. Gidiş-dönüş: encode ettiğimiz her şeyi decode edebiliriz
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

İkinci strateji birincisinden daha önemlidir. Rastgele baytlar neredeyse hiçbir zaman json.loads'u geçemez. Rastgele JSON ise alan kontrollerine ulaşır; bool'un int olması ve NaN hataları tam orada ortaya çıkar. pytest -q ile çalıştırın. Bir başarısızlık en küçük girdiyi yazdırır, örneğin {'device': '0', 'temp_c': True, 'ts': 0}.

Atheris ile kapsam güdümlü fuzzing

Atheris, Python için libFuzzer tabanlı bir fuzzer'dır. Baytları mutasyona uğratır ve yeni dallara ulaşan girdileri saklar; bu nedenle bir decoder'ın çok sayıda yolu olduğunda (birden fazla şema sürümü, ikili formatlar, sıkıştırma) rastgele üretimden daha derine iner.

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

Başka herhangi bir exception fuzzer'ı durdurur ve girdiyi içeren bir crash-<hash> dosyası yazar. Decoder'ı düzeltin, dosyayı bir regresyon testine ekleyin ve yeniden çalıştırın. corpus/ dizini ilginç girdileri saklar: commit edin ya da CI'da cache'leyin.

Corpus'u gerçek trafikle besleyin

Gerçek mesajlardan başlayan bir fuzzer, sıfırdan başlayana göre hataları daha hızlı bulur. Test topic'inizden bir payload örneği yakalayın:

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("kaydedildi", name, len(msg.payload), "bayt")

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

Aynı fikir JavaScript'te fast-check ile

MQTT.js 5 kullanan bir consumer için fast-check, Hypothesis'in rolünü üstlenir ve Vitest ya da Jest içinde çalışır:

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

test("decodeReading yalnızca PayloadError fırlatır", () => {
  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 },
  );
});

JavaScript'te JSON.parse NaN kabul etmez, ancak nesne beklediğiniz yerde sorunsuzca null, dizi ve string döndürür. Payload'ları mevcut nesnelerle birleştiriyorsanız __proto__ anahtarları da bir kontrolü hak eder.

Yalnızca decoder'ı değil, handler'ı da fuzz edin

Decoder kusursuz olabilir ve subscriber yine de çökebilir. paho-mqtt 2.x'te on_message içinde fırlatılan bir exception varsayılan olarak ağ döngüsünün dışına yayılır (client.suppress_exceptions = True ayarlamadıysanız) ve bu da loop_forever()'ı ya da loop_start() thread'ini durdurur. Tek bir kötü mesajın consumer'ı çökertememesi için handler'ı sarmalayın:

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

Ardından handler'ın kendisini sahte bir msg nesnesi (bir SimpleNamespace(topic=..., payload=...) yeterli) ve sahte bir istemciyle fuzz edin; tıpkı mock broker ile birim testlerinde olduğu gibi. Özellik şudur: handler her zaman geri döner ve reddedilen her payload dead-letter topic'ine düşer.

Çökme corpus'unu bir broker üzerinden yeniden oynatın

Birim seviyesindeki fuzzer'lar yeşile döndüğünde, kayıtlı corpus'u ve crash dosyalarını bir test topic'ine publish edin ve gerçek consumer'ın ayakta kaldığını izleyin:

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

Betik çalışırken online MQTT istemcisinde testmqtt/fuzz/# topic'ine abone olun: her payload'ı telemetri topic'inde bir kez, reddedilenleri dead-letter topic'inde görmeli ve sonunda consumer'ın hâlâ bağlı olduğunu doğrulamalısınız.

Replay'i tarayıcınızda izleyin

CI'a ekleyin

1. Hypothesis ve fast-check özelliklerini her commit'te çalıştırın. Birkaç saniye sürer.
2. Atheris'i gece çalışan bir job'da beş ila on beş dakika çalıştırın; corpus'u çalıştırmalar arasında cache'leyin.
3. Hatayı düzeltmeden önce her crash dosyasını isimlendirilmiş bir regresyon testine dönüştürün.
4. Belgelenmiş exception dışındaki herhangi bir exception tipinde build'i başarısız yapın.

Replay adımı için ayrıca bir broker ayağa kaldıran bir workflow için GitHub Actions'ta MQTT testleri yazısına bakın.

Fuzzing büyük bir kurulum gerektirmez. Tek bir belgelenmiş exception'a sahip bir decoder, üç özellik ve gece çalışan bir fuzzer; NaN'ı, boolean sıcaklığı ve 10.000 seviye iç içe geçmiş diziyi, hatalı bir cihaz bunları production'a göndermeden çok önce bulur.

Sıkça sorulan sorular