tools/testing/selftests/bpf/generate_udp_fragments.py

Source file repositories/reference/linux-study-clean/tools/testing/selftests/bpf/generate_udp_fragments.py

File Facts

System
Linux kernel
Corpus path
tools/testing/selftests/bpf/generate_udp_fragments.py
Extension
.py
Size
2640 bytes
Lines
91
Domain
Support Tooling And Documentation
Bucket
tools
Inferred role
Support Tooling And Documentation: tools
Status
atlas-only

Why This File Exists

Repository support layer: documentation, build tooling, samples, user-space helper tools, generated initramfs support, licenses, and validation utilities.

Dependency Surface

Detected Declarations

Annotated Snippet

#!/bin/env python3
# SPDX-License-Identifier: GPL-2.0

"""
This script helps generate fragmented UDP packets.

While it is technically possible to dynamically generate
fragmented packets in C, it is much harder to read and write
said code. `scapy` is relatively industry standard and really
easy to read / write.

So we choose to write this script that generates a valid C
header. Rerun script and commit generated file after any
modifications.
"""

import argparse
import os

from scapy.all import *


# These constants must stay in sync with `ip_check_defrag.c`
VETH1_ADDR = "172.16.1.200"
VETH0_ADDR6 = "fc00::100"
VETH1_ADDR6 = "fc00::200"
CLIENT_PORT = 48878
SERVER_PORT = 48879
MAGIC_MESSAGE = "THIS IS THE ORIGINAL MESSAGE, PLEASE REASSEMBLE ME"


def print_header(f):
    f.write("// SPDX-License-Identifier: GPL-2.0\n")
    f.write("/* DO NOT EDIT -- this file is generated */\n")
    f.write("\n")
    f.write("#ifndef _IP_CHECK_DEFRAG_FRAGS_H\n")
    f.write("#define _IP_CHECK_DEFRAG_FRAGS_H\n")
    f.write("\n")
    f.write("#include <stdint.h>\n")
    f.write("\n")


def print_frags(f, frags, v6):
    for idx, frag in enumerate(frags):
        # 10 bytes per line to keep width in check
        chunks = [frag[i : i + 10] for i in range(0, len(frag), 10)]
        chunks_fmted = [", ".join([str(hex(b)) for b in chunk]) for chunk in chunks]
        suffix = "6" if v6 else ""

        f.write(f"static uint8_t frag{suffix}_{idx}[] = {{\n")
        for chunk in chunks_fmted:
            f.write(f"\t{chunk},\n")
        f.write(f"}};\n")


def print_trailer(f):
    f.write("\n")
    f.write("#endif /* _IP_CHECK_DEFRAG_FRAGS_H */\n")


def main(f):
    # srcip of 0 is filled in by IP_HDRINCL
    sip = "0.0.0.0"
    sip6 = VETH0_ADDR6
    dip = VETH1_ADDR
    dip6 = VETH1_ADDR6
    sport = CLIENT_PORT
    dport = SERVER_PORT
    payload = MAGIC_MESSAGE.encode()

Annotation

Implementation Notes