Network Address Translation changes addressing information as packets cross a boundary. The familiar many-clients-to-one-address arrangement uses network address and port translation: the router keeps mappings that distinguish simultaneous flows. A remote TCP server may see one source IP while seeing different source ports. That shared address identifies a translation boundary, not necessarily one computer, account, or person.

When it breaks

Translation does not replace routing. A client still needs a route toward the gateway, the gateway must forward packets, and the translated address must support a return path. Conversely, private addressing does not inherently require NAT inside a network that routes those prefixes. A missing return route and a missing translation rule can produce similar timeouts but require different configuration decisions.

Stateful NAT also makes existing and new connections behave differently during rule changes. nftables consults a NAT chain to establish a flow’s mapping; later packets use the stored binding. Removing an SNAT rule therefore does not necessarily revoke established translated connections. This depends on connection tracking remaining active and the binding remaining valid, so treating a ruleset edit as an unconditional session purge is unsafe.

Port mappings and connection-tracking capacity are finite resources. Exhaustion, expiry, or a failover device lacking the relevant state can interrupt communication even when the public address remains unchanged. A usable translated source port is considered within the relevant protocol and connection tuple; avoid estimating capacity as one universal port count for every destination and workload. NAT is also separate from the firewall policy deciding permitted traffic.

Check it quickly

Save this as check-nat.py and run python3 check-nat.py with Docker. It builds a disposable environment with Python 3.11.16 and nftables 1.0.6. NET_ADMIN and SYS_ADMIN enable four nested network namespaces and their temporary proc mount. The example was executed on Linux 6.12.76; Docker shares its host kernel. No host interface or existing firewall is changed:

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)

Clients a and b occupy separate private subnets behind a router. The target has only its connected route to the router’s outward interface, 192.0.2.1. These documentation addresses remain inside the laboratory. Without translation or an explicit return route to the client subnet, the first TCP exchange times out even though the router has forwarding enabled.

The script installs an SNAT rule and an established-state tracking rule. Both clients bind source port 40000 and contact the same target address and port. The target reports source address 192.0.2.1 for both, with different translated ports. The script checks that replies retain the correct client label; merely observing two accepted sockets would not establish that responses were delivered to the intended clients.

The NAT rule counter records two newly mapped connections. The script then flushes that rule while leaving the NAT base chain and separate tracking rule in place. Both existing sockets exchange another message using their original mappings, while a new connection times out. This directly demonstrates why checking only a previously opened connection can miss a broken configuration for new clients.

Next, adding a return route on the target allows another new connection without SNAT. Its reported peer is the actual private address, 10.203.1.2. Finally, removing the last tracking expression breaks an existing translated socket; this run reported ConnectionResetError. The matching kernel source unregisters tracking hooks when the final namespace reference is released. That is a different operation from deleting one NAT rule, and does not justify claiming that every stored entry was erased.

Diagnose the mapping and the return path together

Record original and translated source addresses and ports, destination tuple, protocol, and timestamps when correlating traffic across a gateway. Server logs containing only the shared IP cannot distinguish the clients in this example. Authentication and application identities remain necessary where attribution matters; a translation log provides network correlation rather than proof of a user’s identity.

For a change or incident, test both an established connection and a newly opened one. Check routing in both directions and inspect the applicable translation and filtering rules. Preserve state intentionally during maintenance when continuity is required. If sessions must be terminated, use an explicit, scoped mechanism whose behavior is verified for the deployed system; do not infer termination from a removed rule alone.

Sources