Let's play with a simple TCP server with SO_REUSEPORT

simple_tcp_server_with_SO_REUSEPORT.py

import socket
import sys

from contextlib import contextmanager

import click

so_reuseport = socket.SO_REUSEPORT

@contextmanager
def reserve_sock_addr(reuseport: bool, predefined_port: int=0):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        if reuseport:
            print('enable SO_REUSEPORT')
            sock.setsockopt(socket.SOL_SOCKET, so_reuseport, 1)
        
        sock.bind(("", predefined_port if predefined_port else 0))
        _, port = sock.getsockname()
        print(f'reserve following socket {socket.getfqdn()} {port}')
        yield (socket.getfqdn(), port)
        print('closing the first socket')


# https://pymotw.com/2/socket/tcp.html
def simple_tcp_echo(port: int, reuseport: bool):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        if reuseport:
            print('enable SO_REUSEPORT')
            sock.setsockopt(socket.SOL_SOCKET, so_reuseport, 1)

        sock.bind(("", port))
        _, port = sock.getsockname()
        print(f'listening to {socket.getfqdn()} {port}')

        # Listen for incoming connections
        sock.listen()
        
        while True:
            # Wait for a connection
            print('waiting for a connection')
            connection, client_address = sock.accept()

            with connection:
                try:
                    print('connection from', client_address)

                    # Receive the data in small chunks and retransmit it
                    while True:
                        data = connection.recv(8)
                        print(f'received "{data}"')
                        if data:
                            print('sending data back to the client')
                            connection.sendall(data)
                        else:
                            print('no more data from')
                            break
                        
                finally:
                    # Clean up the connection
                    connection.close()
                    print('clean up the connection')

    print('closing the second socket')



@click.command()
@click.option("--port", default=0, type=int,
              help="TCP port to use")
@click.option("--reuseport", is_flag=True, type=bool,
              help="Enable SO_REUSEPORT")
def start_server(port: int, reuseport: bool):
    if not port:
        with reserve_sock_addr(reuseport) as (_, port):
            port = port
    
    simple_tcp_echo(port, reuseport)

    print('this is the end')


@click.command()
@click.option("--predefined_port", default=0, type=int,
              help="TCP port to use")
@click.option("--target_port", type=int,
              help="target TCP port to hijack")
@click.option("--reuseport", is_flag=True, type=bool,
              help="Enable SO_REUSEPORT")
def tcp_bind_loop(predefined_port: int,
                  target_port: int,
                  reuseport: bool):
    
    from collections import Counter

    all_ports = Counter()
    n_attempts = 0

    while(True):
        n_attempts += 1

        with reserve_sock_addr(reuseport, predefined_port=predefined_port) as (_, port):
            port = port
            all_ports[port] += 1
            
        if port < target_port:
            print(f'binding to {socket.getfqdn()} {port} less then {target_port}')
        elif target_port == port:
            print(f'hijacking successful! {socket.getfqdn()} {port}')
            break
        elif port > target_port:
            print(f'binding to {socket.getfqdn()} {port} greater then {target_port}')

    print(f'there were {len(all_ports)} unique ports after {n_attempts} attempts')
    print(f'the top-25 most common were {all_ports.most_common(25)}')    

    print('sad but true')


@click.command()
@click.option("--reuseport", is_flag=True, type=bool,
              help="Enable SO_REUSEPORT")
def create_reuseport_and_wait(reuseport: bool):
    with reserve_sock_addr(reuseport) as (_, port):
        port = port
        input("#1 Press Enter to continue...")

    
    input("#2 Press Enter to continue...")

    print('this is the end')


@click.command()
@click.option("--reuseport", is_flag=True, type=bool,
              help="Enable SO_REUSEPORT")
def create_reuseport_and_reuseagain(reuseport: bool):
    with reserve_sock_addr(reuseport) as (_, port):
        port = port
        input("#1 Press Enter to continue...")

    with reserve_sock_addr(reuseport, predefined_port=port) as (_, port):
        port = port
        input("#1.1 Press Enter to continue...")

    input("#2 Press Enter to continue...")

    print('this is the end')


@click.command()
@click.option("--reuseport", is_flag=True, type=bool,
              help="Enable SO_REUSEPORT")
def create_reuseport_and_explicit_exit(reuseport: bool):
    context_manager_instance = reserve_sock_addr(reuseport)

    (_, port) = context_manager_instance.__enter__()
    explicit_close = lambda: context_manager_instance.__exit__(None, None, None)

    input("#1 Press Enter to continue...")

    explicit_close()

    with reserve_sock_addr(reuseport, predefined_port=port) as (_, port):
        print('reserved 2nd time')
        port = port
        input("#1.1 Press Enter to continue...")

    input("#2 Press Enter to continue...")

    print('this is the end')



@click.group()
def cli():
    """simple TCP server to reproduce port binding issues"""

cli.add_command(start_server)
cli.add_command(tcp_bind_loop)
cli.add_command(create_reuseport_and_wait)
cli.add_command(create_reuseport_and_reuseagain)
cli.add_command(create_reuseport_and_explicit_exit)


if __name__ == "__main__":
    cli()