TCP 三次握手建立字节流的两个方向。每端提供自己的初始序列号，并确认另一端的序列号。这是两个独立序列空间，不是共享墙上时钟；应用建立连接不需要两台机器时钟同步。之后的序列号用于标识字节，支持连接内部的排序、重复处理、确认和重传。把序列号与时间混为一谈，会误解握手真正建立的协议状态。

坏掉的样子

客户端发送带初始序列号的 SYN，服务器回复 SYN-ACK，给出自己的初始序列号并确认客户端 SYN，客户端再确认服务器响应。即使没有应用数据，SYN 也占用序列空间中的一个位置，所以对应确认号增加一。将这些数值当成时间戳，会错误解释报文，还可能把排障引向无关的时钟同步问题，忽略真正缺失的确认或路由。

两端不会在完全相同的时刻改变状态。客户端收到 SYN-ACK 后，connect 可能成功返回，但最后一个确认在到达服务器前丢失。服务器仍可能停留在 SYN-RECEIVED，Linux 工具显示为 SYN-RECV。后续携带适当确认的数据也能完成交换，因此单独第三个报文缺失，并不必然导致永久失败，不能把抓包中的固定三包模式当成唯一合法过程。

TCP 建立成功也不代表应用健康或对端身份可信。服务可能接受连接却无法生成有效响应，代理可能完成客户端这一侧连接，却无法建立另一个独立上游连接。初始序列号选择有助于区分连接实例、抵御预测攻击，但不是身份认证协议，也不会加密字节流。传输状态、应用结果和安全身份需要各自对应的验证，不能由一项成功代替全部。

快速验证

将脚本保存为 check-tcp-handshake.py，准备 Docker 后运行 python3 check-tcp-handshake.py。一次性容器使用 Python 3.11.16、iproute2 6.1.0、nftables 1.0.6 和 tcpdump 4.99.3，在 Linux 6.12.76 上验证。NET_ADMIN 与 SYS_ADMIN 支持嵌套网络命名空间。客户端和目标由新建 veth 相连，不移动宿主接口或发布端口：

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


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,'tcpdump exited before listening'
        if select.select([process.stderr],[],[],.1)[0] and 'listening on' in process.stderr.readline():return process,path
    raise AssertionError('tcpdump did not become ready')
def packets(process,path):
    import struct,socket
    process.send_signal(signal.SIGINT);_,diagnostic=process.communicate(timeout=3)
    assert process.returncode==0,diagnostic
    raw=path.read_bytes();assert raw[:4]==bytes.fromhex('d4c3b2a1') and struct.unpack_from('<I',raw,20)[0]==1
    offset=24;result=[]
    while offset<len(raw):
        seconds,micros,size,original=struct.unpack_from('<IIII',raw,offset);offset+=16
        frame=raw[offset:offset+size];offset+=size
        if frame[12:14]!=b'\x08\x00' or frame[23]!=6:continue
        ihl=(frame[14]&15)*4;length=struct.unpack_from('!H',frame,16)[0];tcp=14+ihl
        sport,dport,seq,ack=struct.unpack_from('!HHII',frame,tcp);header=(frame[tcp+12]>>4)*4
        opts={};cursor=tcp+20
        while cursor<tcp+header:
            kind=frame[cursor]
            if kind==0:break
            if kind==1:cursor+=1;continue
            option_length=frame[cursor+1];data=frame[cursor+2:cursor+option_length];assert option_length>=2
            if kind==2:opts['mss']=struct.unpack('!H',data)[0]
            elif kind==3:opts['wscale']=data[0]
            elif kind==4:opts['sack_ok']=True
            elif kind==5:opts['sack']=[list(pair) for pair in struct.iter_unpack('!II',data)]
            cursor+=option_length
        body=frame[tcp+header:14+length]
        result.append({'t':seconds+micros/1e6,'src':socket.inet_ntoa(frame[26:30]),'dst':socket.inet_ntoa(frame[30:34]),
            'sport':sport,'dport':dport,'seq':seq,'ack':ack,'flags':frame[tcp+13],'window':struct.unpack_from('!H',frame,tcp+14)[0],
            'length':len(body),'options':opts,'preview':body[:24].hex()})
    return result
with lab(['client','target']) as (root,stack,setting):
    wire('client','c0','target','t0',['192.0.2.1/30'],['192.0.2.2/30'])
    server_code=r'''
import socketserver
class Handler(socketserver.StreamRequestHandler):
    def handle(self):
        data=self.rfile.readline();self.wfile.write(b'echo:'+data);self.wfile.flush()
class Server(socketserver.ThreadingTCPServer):allow_reuse_address=True;daemon_threads=True
server=Server(('192.0.2.2',18080),Handler);print('ready',flush=True);server.serve_forever()
'''
    service=spawn(stack,['ip','netns','exec','target','python3','-u','-c',server_code]);assert line(service).strip()=='ready'
    tap,path=capture(stack,root,'client','c0','tcp port 18080','handshake')
    normal=ns('client','python3','-c',
        'import socket; s=socket.socket();s.settimeout(2);s.bind(("192.0.2.1",40100));s.connect(("192.0.2.2",18080));s.sendall(b"normal\\n");print(s.makefile("rb").readline().decode().strip());s.close()')
    assert normal.stdout.strip()=='echo:normal'
    nft('client','''table ip missing_ack {
 chain output { type filter hook output priority 0; policy accept;
  tcp dport 18080 tcp flags & (syn|ack|rst|fin) == ack counter drop comment "missing_final_ack"
 }
}
''')
    client_code=r'''
import json,socket,sys
s=socket.socket();s.settimeout(.25);s.bind(('192.0.2.1',40101));s.connect(('192.0.2.2',18080))
print(json.dumps({'connect_returned':True}),flush=True);s.sendall(b'deferred\n')
try:s.recv(64);raise AssertionError('unexpected application response')
except TimeoutError:print(json.dumps({'application_timeout':True}),flush=True)
assert sys.stdin.readline().strip()=='restore';s.settimeout(5)
received=b''
while not received.endswith(b'\n'):
    chunk=s.recv(64);assert chunk,'premature EOF';received+=chunk
assert received==b'echo:deferred\n',received
print(json.dumps({'recovered':received.decode().strip()}),flush=True);s.close()
'''
    client=spawn(stack,['ip','netns','exec','client','python3','-u','-c',client_code])
    connected=json.loads(line(client));failure=json.loads(line(client));assert connected['connect_returned'] and failure['application_timeout']
    pending=ns('target','ss','-nHt','state','syn-recv').stdout;assert '192.0.2.1:40101' in pending,pending
    blocked=counters('client');assert blocked['missing_final_ack']>0
    nft('client','delete table ip missing_ack');client.stdin.write('restore\n');client.stdin.flush()
    recovery=json.loads(line(client,6));assert recovery['recovered']=='echo:deferred'
    client.wait(timeout=2);assert client.returncode==0
    observed=packets(tap,path);flow=[p for p in observed if 40100 in (p['sport'],p['dport'])]
    syn=next(p for p in flow if p['flags']&0x12==0x02)
    synack=next(p for p in flow if p['flags']&0x12==0x12)
    third=next(p for p in flow if p['sport']==40100 and p['flags']&0x12==0x10)
    assert synack['ack']==(syn['seq']+1)%2**32
    assert third['seq']==(syn['seq']+1)%2**32 and third['ack']==(synack['seq']+1)%2**32
    print(json.dumps({'three_messages':[syn,synack,third],'normal_reply':normal.stdout.strip(),
        'client_connect_returned':connected['connect_returned'],'application_timeout':failure['application_timeout'],
        'server_still_syn_recv':pending.strip(),'drop_counters':blocked,'after_ack_filter_removed':recovery}))
"""
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)

第一条连接使用客户端端口 40100 和目标端口 18080，发送换行结束的请求，收到预期回显。抓包提供真实 SYN、SYN-ACK 和最终 ACK 数值。断言检查服务器确认号等于客户端序列号加一，第三条消息确认号等于服务器序列号加一，并对三十二位序列空间使用取模运算。这样验证的是协议关系，而不是要求每次随机数相同。

第二条连接使用客户端端口 40101。客户端命名空间输出规则丢弃带 ACK、但不带 SYN、FIN 或 RST 的报文，包含最后确认与后续数据。阻断后续数据很重要，否则新的数据段可能确认 SYN-ACK，修复刻意丢失的第三条消息。实验由此稳定暴露两端不同状态，而不是依赖短暂时序巧合，或错误地假定一个 ACK 丢失后永远无法恢复。

客户端报告 connect 已成功返回，但读取应用响应直到截止时间仍失败。同一观察点，服务器套接字列表显示该客户端四元组处于 SYN-RECV，规则计数器也确认发生丢弃。三个独立观察分别对应客户端本地 API 结果、服务器协议状态和应用交换失败，证明它们不能混成同一项“连接正常”。故障归因需要同时保存这些上下文。

移除规则后，同一连接可以继续完成交换并交付已排队请求。客户端收到 echo:deferred，没有新建替代套接字来掩盖原连接失败。带边界的读取会累积响应，不假定一次 recv 总能返回完整应用消息。测试同时验证临时失败，以及受控原因移除之后的恢复；恢复来自 TCP 后续传输与确认，不是应用重开连接造成的另一个成功样本。

分别诊断建立过程和有效服务

记录四元组、命名空间、捕获位置与两端状态。重置、反复 SYN、反复 SYN-ACK，以及握手完成后的应用超时，会指向不同的下一步检查。条件允许时捕获两个方向：本地看到一个确认离开，不代表远端已经收到。本例通过规则计数和远端状态补足这个缺口，避免只看单侧包迹就宣布服务器一定进入已建立状态。

验证服务时，应在传输连接之后发送有意义的请求并检查响应。验证身份和机密性时，应使用 TCP 上层适当的认证协议。本例覆盖没有 Fast Open 或同时打开的普通连接，不声称每个合法连接都以三个分离可见的报文开始。关键是序列号与状态的关系，而不是简单数包，更不能把握手成功当作业务操作已经完成的证明。

来源

  • RFC 9293 TCP，序列号、连接状态与三次握手，已核验 2026-09-12。
  • Linux tcp(7)，TCP 套接字行为与 Linux 控制选项，已核验 2026-09-12。
  • Linux ss(8)，查看已建立与 SYN-RECV 套接字，已核验 2026-09-12。