网络地址转换会在数据包经过边界时改写地址信息。常见的多个客户端共用一个地址，实际采用网络地址和端口转换：路由器维护映射，区分同时存在的流量。远端 TCP 服务器可能看到相同源 IP，却看到不同源端口。这个共享地址代表转换边界，不一定代表同一台计算机、同一个账号或同一个人，不能直接替代身份判断。

坏掉的样子

转换不能代替路由。客户端仍需要到网关的路径，网关必须允许转发，转换后的地址也要具有返回路径。反过来，在能够路由相应前缀的内部网络中，使用私有地址并不必然要求 NAT。缺少返回路由和缺少转换规则都可能表现为超时，但两者对应不同配置决策。应先确定目标看到什么来源，再检查回复如何返回。

有状态 NAT 还会使规则变更中的已有连接与新连接表现不同。nftables 查询 NAT 链以建立流量映射，后续数据包复用保存的绑定。因此，移除 SNAT 规则不一定撤销已经建立的转换连接。这依赖连接跟踪继续工作、绑定仍然有效，不能把一次规则编辑理解成无条件清除全部会话。测试对象是否复用了旧连接，会直接影响结论。

端口映射与连接跟踪容量都是有限资源。资源耗尽、状态到期，或者故障切换设备缺少相关状态，都可能在对外地址未变时中断通信。可用转换源端口需要结合协议和相关连接元组判断，不能用一个对所有目的地、所有负载都通用的端口总数估算容量。NAT 与决定哪些流量被允许的防火墙策略，也属于不同职责。

两个客户端流量在共享转换地址下仍分别映射，返回路由与映射寿命是不同条件。
独立流量可共享转换地址。删除规则不同于映射与依赖寿命；最后一次交互用于观察，不要求必然失败。

快速验证

历史文章报告未找到对应原始集成日志。本轮没有重跑；特权操作继续暂停，也没有尝试替代路径。

将脚本保存为 check-nat.py，准备 Docker 后运行 python3 check-nat.py。它构建包含 Python 3.11.16 和 nftables 1.0.6 的一次性环境，NET_ADMIN 与 SYS_ADMIN 是创建四个嵌套网络命名空间及临时 proc 挂载的前提。历史文章报告 Linux 6.12.76；需启用 Python 断言，清理仅尽力执行。Docker 共享宿主内核，必须只在另行获准的一次性虚拟机里运行；这些权限与共享内核意味着仅用容器不足以保证隔离。实时软件源也不保证未来还能取得这些固定包版本：

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'
    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(['a','b','router','target']) as (root,stack,setting):
    wire('a','a0','router','ra',['10.203.1.2/30'],['10.203.1.1/30'])
    wire('b','b0','router','rb',['10.203.2.2/30'],['10.203.2.1/30'])
    wire('router','wan','target','t0',['192.0.2.1/30'],['192.0.2.2/30'])
    setting('router','net.ipv4.ip_forward',1)
    ip('a','route','add','default','via','10.203.1.1');ip('b','route','add','default','via','10.203.2.1')
    server_code=r'''
import json,socketserver
class Handler(socketserver.StreamRequestHandler):
    def handle(self):
        for raw in self.rfile:
            data=json.loads(raw);data['peer']=self.client_address
            self.wfile.write(json.dumps(data).encode()+b'\n');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()
'''
    daemon=spawn(stack,['ip','netns','exec','target','python3','-u','-c',server_code]);assert line(daemon).strip()=='ready'
    client_code=r'''
import json,socket,sys
label,port=sys.argv[1],int(sys.argv[2])
try:
    with socket.socket() as stream:
        stream.settimeout(.7);stream.bind(('0.0.0.0',port));stream.connect(('192.0.2.2',18080))
        with stream.makefile('rb') as incoming:
            def exchange(phase):
                stream.sendall(json.dumps({'label':label,'phase':phase}).encode()+b'\n')
                print(json.dumps({'ok':True,'reply':json.loads(incoming.readline())}),flush=True)
            exchange('initial')
            for command in sys.stdin:exchange(command.strip())
except OSError as error:print(json.dumps({'ok':False,'type':type(error).__name__,'errno':error.errno}),flush=True)
'''
    def once(name,port):return json.loads(ns(name,'python3','-u','-c',client_code,name,str(port),input='').stdout)
    before=once('a',39000);assert not before['ok'] and before['type']=='TimeoutError'
    nft('router','''table ip translation {
 chain postrouting { type nat hook postrouting priority srcnat; policy accept;
  ip saddr 10.203.0.0/16 oifname "wan" counter snat to 192.0.2.1 comment "new_mapping"
 }
}
table inet tracking {
 chain forward { type filter hook forward priority filter; policy accept;
  ct state established,related counter accept comment "established"
 }
}
''')
    clients={};initial={}
    for name in ['a','b']:
        clients[name]=spawn(stack,['ip','netns','exec',name,'python3','-u','-c',client_code,name,'40000'])
        initial[name]=json.loads(line(clients[name]));assert initial[name]['ok'] and initial[name]['reply']['label']==name
    peers=[value['reply']['peer'] for value in initial.values()]
    assert all(peer[0]=='192.0.2.1' for peer in peers) and peers[0][1]!=peers[1][1]
    mapped=counters('router');assert mapped['new_mapping']==2
    nft('router','flush chain ip translation postrouting')
    retained={}
    for name,process in clients.items():
        process.stdin.write('after-rule-removal\n');process.stdin.flush();retained[name]=json.loads(line(process))
        assert retained[name]['ok'] and retained[name]['reply']['peer']==initial[name]['reply']['peer'],(name,retained[name],initial[name])
    new_flow=once('a',40001);assert not new_flow['ok'] and new_flow['type']=='TimeoutError'
    ip('target','route','add','10.203.1.0/30','via','192.0.2.1')
    routed_without_nat=once('a',40003);assert routed_without_nat['ok'] and routed_without_nat['reply']['peer'][0]=='10.203.1.2'
    nft('router','delete table inet tracking')
    clients['a'].stdin.write('after-table-delete\n');clients['a'].stdin.flush()
    after_table_delete=json.loads(line(clients['a']))
    assert isinstance(after_table_delete.get('ok'),bool),after_table_delete
    if after_table_delete['ok']:
        reply=after_table_delete['reply']
        assert reply['label']=='a' and reply['phase']=='after-table-delete',reply
        assert reply['peer']==initial['a']['reply']['peer'],reply
    else:
        # Exact envelope emitted by this client's built-in OSError catch family.
        assert set(after_table_delete)=={'ok','type','errno'},after_table_delete
        assert type(after_table_delete['type']) is str and after_table_delete['type'] in {
            'OSError','BlockingIOError','ChildProcessError','ConnectionError',
            'BrokenPipeError','ConnectionAbortedError','ConnectionRefusedError','ConnectionResetError',
            'FileExistsError','FileNotFoundError','InterruptedError','IsADirectoryError',
            'NotADirectoryError','PermissionError','ProcessLookupError','TimeoutError'},after_table_delete
        assert after_table_delete['errno'] is None or type(after_table_delete['errno']) is int,after_table_delete
    print(json.dumps({'without_nat_or_return_route':before,'both_clients_bind_port':40000,'translated_connections':initial,
        'nat_first_packet_counter':mapped,'existing_after_rule_removal':retained,'new_after_rule_removal':new_flow,
        'explicit_return_route_without_nat':routed_without_nat,'after_tracking_table_delete':after_table_delete,
        'last_tracking_reference':'not_measured'}))
"""
name='xbcatm-netpath-'+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)

客户端 a 与 b 位于路由器后的不同私网子网。目标起初只有通往路由器外侧接口 192.0.2.1 的直连路由。这里的文档保留地址始终位于实验内部，没有连接公网服务器。即使路由器已经启用转发，在既无转换、也无到客户端子网的明确返回路由时，第一次 TCP 交互仍然超时，证明启用转发并不足以保证往返。

脚本安装 SNAT 规则和匹配已建立状态的跟踪规则。两个客户端都绑定源端口 40000，并连接相同目标地址和端口。目标报告二者来源均为 192.0.2.1，但转换后端口不同。脚本还检查回复保留正确客户端标签，避免仅凭服务器接受两个套接字，就断言响应已经送回各自客户端。具体重分配端口由内核决定，不要求每次执行相同。

NAT 规则计数器记录两条新映射连接。随后脚本清空该规则，保留 NAT 基础链与独立跟踪规则。两个已有套接字仍使用原映射交换下一条消息，而新连接超时。这个对照直接说明，只检查早已打开的连接，可能漏掉新客户端无法建立会话的配置问题。计数器也不是经过转换的全部数据包总数，因为后续报文不重复查询该规则。

接着为目标添加返回路由，便允许来自 10.203.1.2 的新连接不经 SNAT 成功。删除跟踪表后，修订脚本记录交互成功或客户端报告的错误；成功时仍核对标签、阶段与对端，既不要求失败，也不把删除命名为释放最后引用。Linux 6.12.76 的源码支持有条件的引用释放：销毁 ct state 表达式会释放跟踪引用，只有相应协议族使用计数归零才撤销挂钩。空 NAT 基础链仍在，而夹具没有测量它或其他引用持有者。历史报告的 ConnectionResetError 没有找到原始日志；源码和这一观察都不能证明该错误、最后持有者或全部状态已被清除。若子进程没有回复或回复格式错误，测试程序会失败，但这不确立 NAT 因果关系。原有映射与新连接的对照仍保留，最后一步只是范围明确的观察，不能把任何失败结果当成内核引用计数的替代测量。

同时诊断映射和返回路径

跨网关关联流量时，应记录转换前后的源地址与端口、目的地元组、协议和时间戳。仅保存共享 IP 的服务器日志，无法区分本例的两个客户端。需要归属判断时，仍应使用认证和应用身份；转换日志提供网络关联线索，不直接证明用户身份。日志保留精度也要足以区分端口随时间被重新使用的情况。

变更或故障排查应同时测试已有连接和新建连接，检查两个方向的路由，以及实际适用的转换和过滤规则。维护期间要求连续性时，应有意保留相关状态。如果目标是终止会话，则应采用范围明确、且在部署系统上验证过行为的机制，不能从“规则已经删除”推导会话必定终止。修复结论需要匹配真正受到影响的连接生命周期阶段。

来源