tools/testing/selftests/bpf/generate_udp_fragments.py

   1 #!/bin/env python3
   2 # SPDX-License-Identifier: GPL-2.0
   3
   4 """
   5 This script helps generate fragmented UDP packets.
   6
   7 While it is technically possible to dynamically generate
   8 fragmented packets in C, it is much harder to read and write
   9 said code. `scapy` is relatively industry standard and really
  10 easy to read / write.
  11
  12 So we choose to write this script that generates a valid C
  13 header. Rerun script and commit generated file after any
  14 modifications.
  15 """
  16
  17 import argparse
  18 import os
  19
  20 from scapy.all import *
  21
  22
  23 # These constants must stay in sync with `ip_check_defrag.c`
  24 VETH1_ADDR = "172.16.1.200"
  25 VETH0_ADDR6 = "fc00::100"
  26 VETH1_ADDR6 = "fc00::200"
  27 CLIENT_PORT = 48878
  28 SERVER_PORT = 48879
  29 MAGIC_MESSAGE = "THIS IS THE ORIGINAL MESSAGE, PLEASE REASSEMBLE ME"
  30
  31
  32 def print_header(f):
  33     f.write("// SPDX-License-Identifier: GPL-2.0\n")
  34     f.write("/* DO NOT EDIT -- this file is generated */\n")
  35     f.write("\n")
  36     f.write("#ifndef _IP_CHECK_DEFRAG_FRAGS_H\n")
  37     f.write("#define _IP_CHECK_DEFRAG_FRAGS_H\n")
  38     f.write("\n")
  39     f.write("#include <stdint.h>\n")
  40     f.write("\n")
  41
  42
  43 def print_frags(f, frags, v6):
  44     for idx, frag in enumerate(frags):
  45         # 10 bytes per line to keep width in check
  46         chunks = [frag[i : i + 10] for i in range(0, len(frag), 10)]
  47         chunks_fmted = [", ".join([str(hex(b)) for b in chunk]) for chunk in chunks]
  48         suffix = "6" if v6 else ""
  49
  50         f.write(f"static uint8_t frag{suffix}_{idx}[] = {{\n")
  51         for chunk in chunks_fmted:
  52             f.write(f"\t{chunk},\n")
  53         f.write(f"}};\n")
  54
  55
  56 def print_trailer(f):
  57     f.write("\n")
  58     f.write("#endif /* _IP_CHECK_DEFRAG_FRAGS_H */\n")
  59
  60
  61 def main(f):
  62     # srcip of 0 is filled in by IP_HDRINCL
  63     sip = "0.0.0.0"
  64     sip6 = VETH0_ADDR6
  65     dip = VETH1_ADDR
  66     dip6 = VETH1_ADDR6
  67     sport = CLIENT_PORT
  68     dport = SERVER_PORT
  69     payload = MAGIC_MESSAGE.encode()
  70
  71     # Disable UDPv4 checksums to keep code simpler
  72     pkt = IP(src=sip,dst=dip) / UDP(sport=sport,dport=dport,chksum=0) / Raw(load=payload)
  73     # UDPv6 requires a checksum
  74     # Also pin the ipv6 fragment header ID, otherwise it's a random value
  75     pkt6 = IPv6(src=sip6,dst=dip6) / IPv6ExtHdrFragment(id=0xBEEF) / UDP(sport=sport,dport=dport) / Raw(load=payload)
  76
  77     frags = [f.build() for f in pkt.fragment(24)]
  78     frags6 = [f.build() for f in fragment6(pkt6, 72)]
  79
  80     print_header(f)
  81     print_frags(f, frags, False)
  82     print_frags(f, frags6, True)
  83     print_trailer(f)
  84
  85
  86 if __name__ == "__main__":
  87     dir = os.path.dirname(os.path.realpath(__file__))
  88     header = f"{dir}/ip_check_defrag_frags.h"
  89     with open(header, "w") as f:
  90         main(f)