调整 MSS 可以恢复一条 TCP 路径，却不会修复 MTU 反馈

from pathlib import Path
from tempfile import TemporaryDirectory
import subprocess,json,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 spawn(stack,command):
    process=subprocess.Popen(command,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)
    stack.callback(stop,process);return process
def line(process,timeout=3):
    assert select.select([process.stdout],[],[],timeout)[0],'child response deadline exceeded'
    value=process.stdout.readline();assert value,process.stderr.read();return value
@contextmanager
def lab():
    with TemporaryDirectory(prefix='mtu-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 ['client','r1','r2','target']:
            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 ['client','r1','r2','target']
            path=proc/'sys'/key.replace('.','/')
            ns(name,'python3','-c',f'from pathlib import Path; p=Path({str(path)!r});p.write_text({str(value)!r});assert p.read_text().strip()=={str(value)!r}')
        yield root,stack,setting
def topology(setting,ipv6=False):
    links=[('client','c0','r1','a0',1,2,1,1500),('r1','a1','r2','b0',2,1,2,1280),('r2','b1','target','t0',3,1,2,1500)]
    for left,left_if,right,right_if,network,left_host,right_host,mtu in links:
        run('ip','link','add',left_if,'type','veth','peer','name',right_if)
        for name,interface,host in [(left,left_if,left_host),(right,right_if,right_host)]:
            run('ip','link','set',interface,'netns',name);ip(name,'link','set',interface,'mtu',str(mtu))
            address=f'2001:db8:{network}::{host}/64' if ipv6 else f'10.210.{network}.{host}/30'
            ip(name,'addr','add',address,'dev',interface,*(['nodad'] if ipv6 else []));ip(name,'link','set',interface,'up')
    family='-6' if ipv6 else '-4'
    for name in ['r1','r2']:setting(name,'net.ipv6.conf.all.forwarding' if ipv6 else 'net.ipv4.ip_forward',1)
    routes=[('client','default','2001:db8:1::1' if ipv6 else '10.210.1.1'),('target','default','2001:db8:3::1' if ipv6 else '10.210.3.1'),
        ('r1','2001:db8:3::/64' if ipv6 else '10.210.3.0/30','2001:db8:2::2' if ipv6 else '10.210.2.2'),
        ('r2','2001:db8:1::/64' if ipv6 else '10.210.1.0/30','2001:db8:2::1' if ipv6 else '10.210.2.1')]
    for name,destination,gateway in routes:ip(name,family,'route','add',destination,'via',gateway)
    if ipv6:
        deadline=time.monotonic()+4
        while any('tentative' in ip(name,'-6','address','show').stdout for name in ['client','r1','r2','target']):
            assert time.monotonic()<deadline,'IPv6 link-local DAD did not finish';time.sleep(.02)
def nft(name,script):return ns(name,'nft','-f','-',input=script)
def counters(name):
    result={}
    for item in json.loads(ns(name,'nft','-j','list','ruleset').stdout)['nftables']:
        rule=item.get('rule',{})
        if 'comment' in rule:
            for expression in rule['expr']:
                if 'counter' in expression:result[rule['comment']]=expression['counter']['packets']
    return result
def capture(stack,root,name,interface,expression,label):
    path=root/(label+'.pcap')
    process=subprocess.Popen(['ip','netns','exec',name,'tcpdump','--immediate-mode','-Z','root','-U','-n','-i',interface,'-w',str(path),expression],stdout=subprocess.DEVNULL,stderr=subprocess.PIPE,text=True)
    stack.callback(stop,process);deadline=time.monotonic()+3
    while time.monotonic()<deadline:
        assert process.poll() is None,'capture exited'
        if select.select([process.stderr],[],[],.1)[0] and 'listening on' in process.stderr.readline():return process,path
    raise AssertionError('capture did not become ready')
def captured(process,path):
    process.send_signal(signal.SIGINT);_,diagnostic=process.communicate(timeout=3);assert process.returncode==0,diagnostic
    return run('tcpdump','-nn','-vv','-r',str(path)).stdout


def server(stack):
    code=r'''
import socketserver,hashlib
class Handler(socketserver.StreamRequestHandler):
    def handle(self):
        self.request.settimeout(3)
        header=self.rfile.readline()
        if not header:return
        length=int(header);data=self.rfile.read(length)
        if len(data)!=length:return
        self.wfile.write(hashlib.sha256(data).hexdigest().encode()+b'\n');self.wfile.flush()
class Server(socketserver.ThreadingTCPServer):allow_reuse_address=True;daemon_threads=True
service=Server(('10.210.3.2',18080),Handler);print('ready',flush=True);service.serve_forever()
'''
    process=spawn(stack,['ip','netns','exec','target','python3','-u','-c',code]);assert line(process).strip()=='ready'
def transfer(size):
    code=r'''
import socket,json,hashlib,sys,struct
size=int(sys.argv[1]);stream=socket.socket();stream.settimeout(1.5);connected=False
try:
    stream.connect(('10.210.3.2',18080));connected=True
    stream.sendall(str(size).encode()+b'\n'+b'x'*size)
    with stream.makefile('rb') as incoming:reply=incoming.readline().decode().strip()
    assert reply==hashlib.sha256(b'x'*size).hexdigest()
    print(json.dumps({'ok':True,'connected':True,'bytes':size,'sha256':reply,'snd_mss':stream.getsockopt(socket.IPPROTO_TCP,socket.TCP_MAXSEG)}))
except OSError as error:
    stream.setsockopt(socket.SOL_SOCKET,socket.SO_LINGER,struct.pack('ii',1,0))
    print(json.dumps({'ok':False,'connected':connected,'type':type(error).__name__,'errno':error.errno}))
finally:stream.close()
'''
    return json.loads(ns('client','python3','-c',code,str(size)).stdout)
def hide_fragmentation_feedback():
    nft('r1','''table ip hidden_mtu {
 chain output { type filter hook output priority 0; policy accept;
  icmp type destination-unreachable icmp code frag-needed counter drop comment "fragmentation_feedback_dropped"
 }
}
''')
with lab() as (root,stack,setting):
    topology(setting);setting('client','net.ipv4.tcp_mtu_probing',0);server(stack);hide_fragmentation_feedback()
    failed=transfer(65536);assert failed['connected'] and not failed['ok'] and failed['type']=='TimeoutError'
    initial_counter=counters('r1')['fragmentation_feedback_dropped'];assert initial_counter>0
    nft('r1','''table ip clamp {
 chain forward { type filter hook forward priority 0; policy accept;
  tcp flags & syn == syn tcp option maxseg size > 1240 tcp option maxseg size set 1240 counter comment "mss_rewritten"
 }
}
''')
    tap,path=capture(stack,root,'client','c0','tcp port 18080','mss')
    recovered=transfer(65536);assert recovered['ok'] and recovered['snd_mss']<=1240,recovered
    trace=captured(tap,path);assert 'mss 1240' in trace,trace
    after_transfer=counters('r1');assert after_transfer['mss_rewritten']>=2 and after_transfer['fragmentation_feedback_dropped']==initial_counter,after_transfer
    large_ping=ns('client','ping','-n','-M','do','-s','1400','-c','1','-W','1','10.210.3.2',check=False)
    assert large_ping.returncode!=0 and counters('r1')['fragmentation_feedback_dropped']>initial_counter
    print(json.dumps({'before_clamp':failed,'after_clamp':recovered,'counters_after_tcp':after_transfer,
        'icmp_echo_still_fails':{'returncode':large_ping.returncode,'stdout':large_ping.stdout,'stderr':large_ping.stderr},'capture':trace}))
"""
name='xbcatm-mtupath-'+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)