防火墙在网络栈的特定位置执行报文策略。一条有用规则因此不只包含端口，还要明确地址族、挂钩、方向、连接状态和相对其他规则的位置。nftables 基础链按指定优先级接入 Netfilter 挂钩。报文实际经过的路径，决定它会遇到哪些裁决，而不是运维文档中描述的先后顺序。审阅策略必须保留这些执行上下文。

坏掉的样子

规则位于错误挂钩时，预期流量可能完全不受影响。INPUT 处理交付给当前命名空间本地的流量，FORWARD 处理经过它转发的流量，OUTPUT 面向本地产生的流量。宿主的容器转发路径，不一定就是本地输入路径。假定宿主级规则保护所有附近服务之前，应先查清实际流量经过哪个命名空间和挂钩。

地址族和方向错误也容易被掩盖。允许服务 TCP 目的端口，不同于匹配通常临时分配的客户端来源端口。只覆盖 IPv4 的策略不会自动覆盖 IPv6；IPv6 通信还依赖邻居发现等 ICMPv6 功能，丢弃所有非 TCP 报文，可能让 TCP 端口明明被允许的连接仍然失败。端口规则不能脱离链路和协议所需控制流量解释。

nftables 的 accept 不一定是所有基础链中的最终裁决。同一挂钩上更晚优先级的另一条链，仍能丢弃报文；drop 则会结束其遍历。只审阅一个表或一条链，可能看到允许计数器增长，却无法解释客户端为何失败。应查看全部适用规则集及其挂钩优先级，不能把第一条可见 accept 当成系统已经最终放行的证明。

快速验证

将脚本保存为 check-firewall.py，准备 Docker 后执行 python3 check-firewall.py。固定环境使用 nftables 1.0.6 和 Python 3.11.16。两个嵌套命名空间通过一次性 veth 同时进行 IPv4 与 IPv6 通信，容器获得实验所需 NET_ADMIN 和 SYS_ADMIN。规则不会修改宿主防火墙或现有服务，实验结束后随隔离环境移除：

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

def server(stack,name,address,port=18080,label='server'):
    code=r'''
from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
import json,socket,sys
address,port,label=sys.argv[1],int(sys.argv[2]),sys.argv[3]
class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        body=json.dumps({'server':label,'peer':self.client_address,'path':self.path}).encode()
        self.send_response(200);self.send_header('Content-Length',str(len(body)));self.end_headers();self.wfile.write(body)
    def log_message(self,*args):pass
class Server(ThreadingHTTPServer):
    address_family=socket.AF_INET6 if ':' in address else socket.AF_INET
    def server_bind(self):
        if self.address_family==socket.AF_INET6:self.socket.setsockopt(socket.IPPROTO_IPV6,socket.IPV6_V6ONLY,1)
        super().server_bind()
service=Server((address,port),Handler);print('ready',flush=True);service.serve_forever()
'''
    process=spawn(stack,['ip','netns','exec',name,'python3','-u','-c',code,address,str(port),label])
    assert line(process).strip()=='ready';return process
def request(name,address,port=18080):
    code=r'''
import http.client,json,sys
client=http.client.HTTPConnection(sys.argv[1],int(sys.argv[2]),timeout=.7)
try:
    client.request('GET','/lab-visible-message');response=client.getresponse();body=json.loads(response.read())
    assert response.status==200;print(json.dumps({'ok':True,'status':response.status,'body':body}))
except OSError as error:print(json.dumps({'ok':False,'type':type(error).__name__,'errno':error.errno}))
finally:client.close()
'''
    return json.loads(ns(name,'python3','-c',code,address,str(port)).stdout)

with lab(['client','target']) as (root,stack,setting):
    wire('client','c0','target','t0',['192.0.2.1/30','2001:db8:1::1/64'],['192.0.2.2/30','2001:db8:1::2/64'])
    addresses=['192.0.2.2','2001:db8:1::2']
    for address in addresses:
        for port in [18080,18081,18082]:server(stack,'target',address,port)
    before={a:[request('client',a,p) for p in [18080,18081,18082]] for a in addresses}
    assert all(v['ok'] for values in before.values() for v in values)
    nft('target','''table inet gate {
 chain input { type filter hook input priority 0; policy drop;
  iifname "lo" accept
  ct state established,related accept
  meta l4proto ipv6-icmp accept
  ip protocol icmp accept
  ip saddr 192.0.2.1 tcp dport 18080 counter accept comment "allow4"
  ip6 saddr 2001:db8:1::1 tcp dport 18080 counter accept comment "allow6"
  tcp dport 18081 counter reject with tcp reset comment "reject"
  tcp dport 18082 counter drop comment "drop"
 }
}
''')
    after={a:[request('client',a,p) for p in [18080,18081,18082]] for a in addresses}
    for values in after.values():
        assert values[0]['ok'] and not values[1]['ok'] and values[1]['errno']==111
        assert not values[2]['ok'] and values[2]['type']=='TimeoutError'
    first=counters('target');assert all(first[n]>0 for n in ['allow4','allow6','reject','drop'])
    nft('target','''table inet later {
 chain input { type filter hook input priority 10; policy accept;
  tcp dport 18080 counter drop comment "later_drop"
 }
}
''')
    overridden=request('client','192.0.2.2');assert not overridden['ok'] and overridden['type']=='TimeoutError'
    later=counters('target');assert later['allow4']>first['allow4'] and later['later_drop']>0
    nft('target','delete table inet later')
    recovered=request('client','192.0.2.2');assert recovered['ok']
    print(json.dumps({'before_all_ports_healthy':True,'with_rules':after,'first_counters':first,
        'later_chain_overrides_accept':overridden,'later_counters':later,'later_chain_removed':recovered}))
"""
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)

目标在两个地址族分别启动一万八千零八十、一万八千零八十一和一万八千零八十二端口的真实 HTTP 监听。过滤之前，六种请求都成功。这个基线很重要，后续失败不能简单归因于缺少监听或某个地址族尚未配置。然后程序安装默认丢弃的 inet 输入链，同时保留回环、已建立流量和选定的 ICMP 处理。

两个地址族分别用明确来源地址规则允许目的端口一万八千零八十；一万八千零八十一通过 TCP reset 拒绝，一万八千零八十二静默丢弃。两种地址族的探测都符合预期：依次获得 HTTP 二百、错误号一百一十一的 ConnectionRefusedError，以及 TimeoutError。超时由本例客户端截止时间决定，不承诺所有应用等待相同长度。

规则计数器确认报文经过允许、拒绝和丢弃分支。拒绝连接与等待超时都表现为失败，但具体语义不同。计数器补充客户端结果，配置文本或单独一个计数器都不能证明应用交换了预期响应。本例的成功请求还核对状态并记录服务身份，把规则评估与最终应用行为联系起来，避免只看到计数增加便宣布功能已经正确。

脚本再增加优先级十的输入基础链，位于原优先级零的链之后，并丢弃一万八千零八十端口。新的 IPv4 请求超时，同时较早的允许计数与较晚的丢弃计数均增长。只移除后续链，HTTP 就恢复成功。这个实验用真实报文验证链之间的相互作用，没有根据单条规则外观推断最终通行结果。

用代表性流量审阅生效策略

通过可审阅配置流程一起维护访问意图和生成规则。应用变更前检查语法，之后检查生效状态，并执行允许与拒绝探测。已建立状态处理可能维持现有会话，而新连接表现不同，因此策略更新测试应包含新建连接。本实验每个比较都创建新客户端连接，避免旧会话继续工作掩盖新的接入规则。

依据工作负载真实依赖与运维要求定义出口策略。它与入站监听是不同决策，不能仅从开放的服务器端口推导统一出口清单。DNS、目标地址变化、相关控制流量和转发连接，都需要对应上下文。本实验验证目标输入行为，没有声称检查生产出口或所有容器运行时的转发规则。范围应按实际经过的路径说明。

允许的服务失败时，把客户端错误与地址族、来源和目的地、命名空间、挂钩、命中计数及后续链对应起来。监听本身需要另外确认，路径仍不清楚时再补充报文证据。简洁策略只有在生效行为与这些检查一致时才可靠，其中也包括拒绝请求应当以何种具体方式失败。可审计性来自规则与实际结果能够相互解释。

延伸阅读

来源