alkcall/fuzz/gen_fuzz_seeds.py

#!/usr/bin/env python3
"""Regenerate the committed seed corpora for alkcall's fuzz targets.

Writes into fuzz/corpus/<target>/. Deterministic: fixed inputs only, no
randomness. Run from the repo root:

    python3 fuzz/gen_fuzz_seeds.py
"""

import json
import os
import struct

MAX_FRAME_SIZE = 64 * 1024 * 1024
MAX_CHUNK_LEN = 16 * 1024 * 1024

ENVELOPE_EVENT_TYPES = [
    "call.requested",
    "call.responded",
    "call.completed",
    "call.aborted",
    "call.error",
    "call.published",
    "totally.unknown.event",
    "",
]

SAMPLE_PAYLOADS = [
    {},
    None,
    0,
    "",
    "payload string",
    {"operationId": "/fs/readFile", "input": {"path": "/etc/hosts"}},
    {"output": {"ok": True}},
    {"input": [1, 2, 3]},
    {"code": "NOT_FOUND", "message": "missing", "retryable": False},
    {"nested": {"deep": {"deeper": [1, {"a": None}]}}},
    {"output": "x" * 4096},
]


def seed_name(target, i):
    return os.path.join("fuzz", "corpus", target, f"seed-{i:03d}")


def write_seed(target, i, data):
    path = seed_name(target, i)
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "wb") as f:
        f.write(data)


def envelope_seeds():
    i = 0

    for event_type in ENVELOPE_EVENT_TYPES:
        for payload in SAMPLE_PAYLOADS[:6]:
            body = json.dumps(
                {"type": event_type, "id": "req-1", "payload": payload},
                separators=(",", ":"),
            ).encode()
            write_seed("envelope_frame", i, struct.pack(">I", len(body)) + body)
            i += 1

    # Truncations at every prefix length of a valid frame.
    body = json.dumps(
        {
            "type": "call.requested",
            "id": "req-1",
            "payload": {"operationId": "/fs/readFile", "input": {"path": "/etc/hosts"}},
        },
        separators=(",", ":"),
    ).encode()
    frame = struct.pack(">I", len(body)) + body
    for cut in range(len(frame)):
        write_seed("envelope_frame", i, frame[:cut])
        i += 1

    # Length prefix edge cases.
    for name, length in [
        ("zero", 0),
        ("max", MAX_FRAME_SIZE),
        ("max-plus-1", MAX_FRAME_SIZE + 1),
        ("u32-max", 0xFFFFFFFF),
        ("huge-but-under-max", MAX_FRAME_SIZE - 1),
    ]:
        write_seed("envelope_frame", i, struct.pack(">I", length))
        i += 1

    # A length prefix claiming a small body but truncated at each offset.
    for claimed in (1, 2, 4, 16):
        for have in range(0, claimed):
            write_seed(
                "envelope_frame", i, struct.pack(">I", claimed) + b'{"a":1}'[:have]
            )
            i += 1

    # Invalid JSON bodies: bad UTF-8, wrong top-level type, missing fields,
    # wrong field types, JSON fragments.
    invalid_bodies = [
        b'{"type":"call.requested","id":"\xff\xfe","payload":null}',
        b'{"type":"call.requested","id":"\xc3","payload":null}',
        b"[1,2,3]",
        b'"just a string"',
        b"null",
        b"42",
        b'{"id":"req-1","payload":null}',
        b'{"type":"call.requested","payload":null}',
        b'{"type":"call.requested","id":"req-1"}',
        b'{"type":123,"id":"req-1","payload":null}',
        b'{"type":"call.requested","id":99,"payload":null}',
        b'{"type":"call.requested","id":"req-1","payload":',
        b'{"type":"call.requested","id":"req-1","payload":undefined}',
        b'{"type":"call.requested","id":"req-1","payload":NaN}',
        b"{",
        b"}",
        b'{"type":"call.requested","id":"req-1","payload":{}}extra',
    ]
    for body in invalid_bodies:
        write_seed("envelope_frame", i, struct.pack(">I", len(body)) + body)
        i += 1

    # Deep-ish nesting inside the payload (well under serde_json's 128-depth
    # recursion limit and libFuzzer's -max_len).
    depth = 64
    nested = ""
    for _ in range(depth):
        nested += '{"a":'
    nested += "1"
    for _ in range(depth):
        nested += "}"
    for depth in (2, 16, 64, 100, 127):
        nested = ""
        for _ in range(depth):
            nested += '{"a":'
        nested += "1"
        for _ in range(depth):
            nested += "}"
        body = json.dumps(
            {"type": "call.requested", "id": "deep", "payload": json.loads(nested)},
            separators=(",", ":"),
        ).encode()
        write_seed("envelope_frame", i, struct.pack(">I", len(body)) + body)
        i += 1

    # A structurally valid envelope with oversized claimed length and
    # empty stream (allocation-bound probe).
    write_seed("envelope_frame", i, struct.pack(">I", MAX_FRAME_SIZE) + b"{")
    i += 1


def chunk_header_seeds():
    i = 0

    def header(channel_id, length):
        return struct.pack(">II", channel_id, length)

    edge_lengths = [
        0,
        1,
        64,
        MAX_CHUNK_LEN,
        MAX_CHUNK_LEN + 1,
        0xFFFFFFFF,
    ]
    edge_ids = [0, 1, 2**31, 0xFFFFFFFF]

    for channel_id in edge_ids:
        for length in edge_lengths:
            write_seed("chunk_header", i, header(channel_id, length))
            i += 1

    # Truncations at every prefix length.
    full = header(0x01020304, 0x05060708)
    for cut in range(8):
        write_seed("chunk_header", i, full[:cut])
        i += 1

    # Oversized buffers (8 bytes is the minimum; longer inputs are legal,
    # parse_header must ignore the rest).
    write_seed("chunk_header", i, header(7, 3) + b"payload-bytes")
    i += 1


def main():
    envelope_seeds()
    chunk_header_seeds()
    for target in ("chunk_header", "envelope_frame"):
        d = os.path.join("fuzz", "corpus", target)
        n = len(os.listdir(d))
        print(f"{target}: {n} seeds")


if __name__ == "__main__":
    main()