Basic UDP NAT rebinding simulator to test QUIC

udp_rebind_proxy.py

#!/usr/bin/env python3
import socket, threading

# Basic UDP NAT rebinding simulator to test QUIC
# Won't work with more than one client

local_port = 443 # Local UDP port to listen on
local_src_port = 65123 # Initial source port

# Destination host to forward traffic to
remote_host = "192.0.2.1"
remote_port = 443

def forwardData():
    ds = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    ds.setblocking(False)
    ds.bind(("0.0.0.0", local_port))
    print(f"Listening on UDP port {local_port}...")

    us = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    us.setblocking(False)
    us.bind(("0.0.0.0", local_src_port))

    while True:
        if local_src_port != us.getsockname()[1]:
            print("rebinding")
            us = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            us.setblocking(False)
            us.bind(("0.0.0.0", local_src_port))

        try:
            dsdata, dsaddr = ds.recvfrom(1500)
            print(f"Uploading {len(dsdata)} bytes from {dsaddr}")
            us.sendto(dsdata, (remote_host, remote_port))
        except BlockingIOError:
            pass

        try:
            usdata, usaddr = us.recvfrom(1500)
            print(f"Dnloading {len(usdata)} bytes from {usaddr}")
            ds.sendto(usdata, dsaddr)
        except BlockingIOError:
            pass

forwarder = threading.Thread(target=forwardData, daemon=True)
forwarder.start()

while True:
    print(f"Sourcing from port {local_src_port}, Enter to incrememt")
    input()
    local_src_port += 1