网络地址转换会在数据包经过边界时改写地址信息。常见的多个客户端共用一个地址，实际采用网络地址和端口转换：路由器维护映射，区分同时存在的流量。远端 TCP 服务器可能看到相同源 IP，却看到不同源端口。这个共享地址代表转换边界，不一定代表同一台计算机、同一个账号或同一个人，不能直接替代身份判断。
坏掉的样子
转换不能代替路由。客户端仍需要到网关的路径，网关必须允许转发，转换后的地址也要具有返回路径。反过来，在能够路由相应前缀的内部网络中，使用私有地址并不必然要求 NAT。缺少返回路由和缺少转换规则都可能表现为超时，但两者对应不同配置决策。应先确定目标看到什么来源，再检查回复如何返回。
有状态 NAT 还会使规则变更中的已有连接与新连接表现不同。nftables 查询 NAT 链以建立流量映射，后续数据包复用保存的绑定。因此，移除 SNAT 规则不一定撤销已经建立的转换连接。这依赖连接跟踪继续工作、绑定仍然有效，不能把一次规则编辑理解成无条件清除全部会话。测试对象是否复用了旧连接，会直接影响结论。
端口映射与连接跟踪容量都是有限资源。资源耗尽、状态到期，或者故障切换设备缺少相关状态，都可能在对外地址未变时中断通信。可用转换源端口需要结合协议和相关连接元组判断，不能用一个对所有目的地、所有负载都通用的端口总数估算容量。NAT 与决定哪些流量被允许的防火墙策略，也属于不同职责。
快速验证
历史集成示例保持原样，本次没有重跑；当前不能使用 Docker、网络命名空间、特权挂载或防火墙变更。
将脚本保存为 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 上执行；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-tracking-removal\n');clients['a'].stdin.flush()
tracking_removed=json.loads(line(clients['a']));assert not tracking_removed['ok'],tracking_removed
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,'last_tracking_reference_removed':tracking_removed}))
"""
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 成功。历史运行在删除跟踪表后记录了 ConnectionResetError。Linux 6.12.76 的源码支持更窄的机制：ct state 表达式取得网络命名空间的跟踪引用，销毁时释放引用；只有相应协议族的使用计数降为零，才撤销该协议族的跟踪挂钩。这两个文件既不能证明本实验当时没有其他引用持有者，也不保证一定出现上述套接字错误。输出字段 last_tracking_reference_removed 只是标签，并未测量实际引用计数。删除规则与释放跟踪依赖不同；这些源码路径也不足以证明所有保存的连接状态都已被擦除。
同时诊断映射和返回路径
跨网关关联流量时，应记录转换前后的源地址与端口、目的地元组、协议和时间戳。仅保存共享 IP 的服务器日志，无法区分本例的两个客户端。需要归属判断时，仍应使用认证和应用身份；转换日志提供网络关联线索，不直接证明用户身份。日志保留精度也要足以区分端口随时间被重新使用的情况。
变更或故障排查应同时测试已有连接和新建连接，检查两个方向的路由，以及实际适用的转换和过滤规则。维护期间要求连续性时，应有意保留相关状态。如果目标是终止会话，则应采用范围明确、且在部署系统上验证过行为的机制，不能从“规则已经删除”推导会话必定终止。修复结论需要匹配真正受到影响的连接生命周期阶段。
来源
- RFC 3022 Traditional NAT，地址端口转换、映射与会话限制，已核验 2026-09-12。
- nftables NAT documentation，首包规则查询与连接映射的复用，已核验 2026-09-12。
- Linux 6.12.76 connection-tracking hooks，各协议族引用计数及计数归零时撤销协议挂钩，已核验 2026-10-15。
- Linux 6.12.76 nftables ct expressions，ct 表达式取得和释放连接跟踪引用，已核验 2026-10-15。
