TIME-WAIT 保护连接实例，不是独占一个端口号

from pathlib import Path
from tempfile import TemporaryDirectory
import json,subprocess,uuid
recipe='FROM python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84\nRUN apt-get update && apt-get install -y --no-install-recommends iproute2=6.1.0-3 iputils-ping=3:20221126-1+deb12u1 nftables=1.0.6-2+deb12u2 tcpdump=4.99.3-1 traceroute=1:2.1.2-1 procps=2:4.0.2-3 && rm -rf /var/lib/apt/lists/*\nCMD ["python3"]\n'
program=r"""
from contextlib import ExitStack,contextmanager
from pathlib import Path
from tempfile import TemporaryDirectory
import json,os,select,signal,subprocess,time

def run(*args,check=True,input=None,timeout=8):
    result=subprocess.run(args,input=input,capture_output=True,text=True,timeout=timeout)
    if check:assert result.returncode==0,(args,result.returncode,result.stdout,result.stderr)
    return result
def ns(name,*args,**kwargs):return run('ip','netns','exec',name,*args,**kwargs)
def ip(name,*args,**kwargs):return ns(name,'ip',*args,**kwargs)
def stop(process):
    if process.poll() is None:
        process.terminate()
        try:process.wait(timeout=2)
        except subprocess.TimeoutExpired:process.kill();process.wait(timeout=2)
def line(process,timeout=3):
    assert select.select([process.stdout],[],[],timeout)[0],'child response deadline exceeded'
    value=process.stdout.readline();assert value,('child exited without a response',process.stderr.read())
    return value
def spawn(stack,command):
    process=subprocess.Popen(command,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)
    stack.callback(stop,process);return process
@contextmanager
def lab(names):
    with TemporaryDirectory(prefix='net-path-') as directory,ExitStack() as stack:
        root=Path(directory);proc=root/'proc';proc.mkdir();run('mount','-t','proc','proc',str(proc))
        stack.callback(run,'umount',str(proc))
        for name in names:
            run('ip','netns','add',name);stack.callback(run,'ip','netns','delete',name,check=False)
            ip(name,'link','set','lo','up')
        def setting(name,key,value):
            assert name in names
            path=proc/'sys'/key.replace('.','/')
            return ns(name,'python3','-c',f'from pathlib import Path; p=Path({str(path)!r}); p.write_text({str(value)!r}); print(p.read_text().strip())').stdout.strip()
        yield root,stack,setting
def wire(left,left_if,right,right_if,left_addresses,right_addresses):
    run('ip','link','add',left_if,'type','veth','peer','name',right_if)
    for name,interface,addresses in [(left,left_if,left_addresses),(right,right_if,right_addresses)]:
        run('ip','link','set',interface,'netns',name)
        for address in addresses:ip(name,'address','add',address,'dev',interface,*(['nodad'] if ':' in address else []))
        ip(name,'link','set',interface,'up')
def nft(name,script):return ns(name,'nft','-f','-',input=script)
def counters(name):
    data=json.loads(ns(name,'nft','-j','list','ruleset').stdout);found={}
    for entry in data['nftables']:
        rule=entry.get('rule',{})
        if 'comment' in rule:
            for expression in rule['expr']:
                if 'counter' in expression:found[rule['comment']]=expression['counter']['packets']
    return found


with lab(['client','target']) as (root,stack,setting):
    wire('client','c0','target','t0',['192.0.2.1/30'],['192.0.2.2/30'])
    setting('client','net.ipv4.tcp_tw_reuse',0)
    setting('client','net.ipv4.tcp_timestamps',0);setting('target','net.ipv4.tcp_timestamps',0)
    server_code=r'''
import socketserver,threading
class Handler(socketserver.BaseRequestHandler):
    def handle(self):
        while self.request.recv(4096):pass
class Server(socketserver.ThreadingTCPServer):allow_reuse_address=True;daemon_threads=True
servers=[Server(('192.0.2.2',port),Handler) for port in [18080,18081]]
for server in servers:threading.Thread(target=server.serve_forever,daemon=True).start()
print('ready',flush=True);threading.Event().wait()
'''
    service=spawn(stack,['ip','netns','exec','target','python3','-u','-c',server_code]);assert line(service).strip()=='ready'
    client_code=r'''
import json,socket
def exchange(destination):
    with socket.socket() as s:
        s.settimeout(2);s.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1);s.bind(('192.0.2.1',40200))
        s.connect(('192.0.2.2',destination));s.sendall(b'closed cleanly');s.shutdown(socket.SHUT_WR)
        assert s.recv(4096)==b''
exchange(18080)
try:exchange(18080);same={'ok':True}
except OSError as error:same={'ok':False,'errno':error.errno,'type':type(error).__name__}
assert not same['ok'] and same['errno']==99,same
exchange(18081)
print(json.dumps({'same_tuple':same,'same_source_port_new_destination_port':True}))
'''
    result=json.loads(ns('client','python3','-c',client_code).stdout)
    rows=ns('client','ss','-nHto','state','time-wait').stdout
    assert '192.0.2.1:40200' in rows and '192.0.2.2:18080' in rows and '192.0.2.2:18081' in rows,rows
    target=ns('target','ss','-nHto','state','time-wait').stdout;assert ':40200' not in target,target
    print(json.dumps({'client_active_close':result,'client_time_wait':rows.strip(),'target_time_wait':target.strip(),
        'timestamps_disabled_in_both_namespaces':True,'client_tcp_tw_reuse':ns('client','sysctl','-n','net.ipv4.tcp_tw_reuse').stdout.strip()}))
"""
name='xbcatm-tcpflow-'+uuid.uuid4().hex[:10]
def docker(*args,**kwargs):return subprocess.run(['docker',*args],capture_output=True,text=True,timeout=300,**kwargs)
try:
    with TemporaryDirectory(prefix=name+'-') as directory:
        root=Path(directory);(root/'Dockerfile').write_text(recipe)
        build=docker('build','-t',name,str(root));assert build.returncode==0,build.stderr
        result=docker('run','--rm','-i','--name',name,'--network','none','--cap-add','NET_ADMIN','--cap-add','SYS_ADMIN',
            '--memory','256m','--pids-limit','128',name,'python3','-',input=program)
        assert result.returncode==0,result.stderr
        print(json.dumps(json.loads(result.stdout),sort_keys=True))
finally:
    docker('rm','-f',name);docker('image','rm',name)