进程仍持有引用时,删除名称不会释放整个网络命名空间。

网络命名空间隔离网络资源,名称并不拥有其中所有进程

Linux 网络命名空间分离接口、路由、邻居表、套接字端口和包过滤规则等网络资源。它与其他命名空间共享内核,不是独立操作系统。进程通常继承父进程的网络命名空间,工具也可以在另一个范围中执行进程。这种隔离针对网络资源,不会自动涵盖全部进程身份和文件系统访问,因此需要按实际机制说明边界。 坏掉的样子 运行 ip netns exec 不会自动获得完整容器隔离。PID、用户身份和主机名隔离属于其他机制。应用可能看到不同路由,却仍与父环境共享普通文件和主机名。如果把一个网络命名空间当成完整沙箱,就会赋予它从未配置过的保证。应区分实际改变了哪类资源,以及哪些属性仍然来自原有运行环境。 新命名空间也不会自动得到有用的外部连接。接口需要有意放置和配置,本地回环可能需要启用。移动接口会改变管理它的命名空间,原范围不再把该设备列为自己的接口。因此,从错误范围执行配置命令,可能报告接口不存在,尽管预期设备仍位于另一个命名空间。诊断前应先确认命令的执行上下文。 名称还有独立生命周期。ip netns 按约定在 /run/netns 暴露引用,通常也能通过 /var/run/netns 访问。删除名称会移除这项引用,但进程或其他引用仍可维持命名空间存活。如果清理流程把名称列表中消失,直接等同于所有服务终止,就可能留下网络资源和运行中的进程。名称只是管理入口,不是进程所有权的完整表达。 快速验证 将脚本保存为 check-network-namespace.py,准备 Docker 后运行 python3 check-network-namespace.py。固定环境使用 Python 3.11.16 和 iproute2 6.1.0,通过 NET_ADMIN 与 SYS_ADMIN 在一次性容器内进行嵌套命名空间操作。临时 proc 挂载支持其中的网络参数设置,不移动宿主接口,也不发布宿主端口;实验资源随容器清理: 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(['blue','green']) as (root,stack,setting): probe='import json,os,socket;print(json.dumps({"net":os.readlink("/proc/self/ns/net"),"pid":os.readlink("/proc/self/ns/pid"),"hostname":socket.gethostname()}))' parent=json.loads(run('python3','-c',probe).stdout) views={name:json.loads(ns(name,'python3','-c',probe).stdout) for name in ['blue','green']} assert len({parent['net'],views['blue']['net'],views['green']['net']})==3 assert all(v['pid']==parent['pid'] and v['hostname']==parent['hostname'] for v in views.values()) for name in ['blue','green']: ip(name,'link','add','dummy0','type','dummy');ip(name,'addr','add','192.0.2.10/24','dev','dummy0');ip(name,'link','set','dummy0','up') ip('blue','route','add','203.0.113.0/24','dev','dummy0') assert json.loads(ip('blue','-j','route','show','203.0.113.0/24').stdout) assert not json.loads(ip('green','-j','route','show','203.0.113.0/24').stdout) setting('blue','net.ipv4.ip_forward',0);setting('green','net.ipv4.ip_forward',1) forwarding={name:ns(name,'sysctl','-n','net.ipv4.ip_forward').stdout.strip() for name in ['blue','green']};assert forwarding=={'blue':'0','green':'1'} marker=root/'shared.txt';marker.write_text('shared filesystem') assert ns('blue','cat',str(marker)).stdout==ns('green','cat',str(marker)).stdout=='shared filesystem' daemon=server(stack,'blue','192.0.2.10',label='blue') named=request('blue','192.0.2.10');assert named['ok'] and named['body']['server']=='blue' run('ip','netns','delete','blue');assert not Path('/run/netns/blue').exists() assert daemon.poll() is None and os.readlink(f'/proc/{daemon.pid}/ns/net')==views['blue']['net'] by_pid=run('nsenter','-t',str(daemon.pid),'-n','python3','-c', 'import http.client,json; c=http.client.HTTPConnection("192.0.2.10",18080,timeout=1);c.request("GET","/");r=c.getresponse();print(json.dumps({"status":r.status,"body":json.loads(r.read())}));c.close()') surviving=json.loads(by_pid.stdout);assert surviving['status']==200 and surviving['body']['server']=='blue' print(json.dumps({'parent':parent,'namespace_views':views,'blue_only_route':True,'forwarding':forwarding, 'same_file_visible':True,'before_name_deleted':named,'name_deleted_process_alive':True,'after_name_deleted':surviving})) """ 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) 父环境、blue 和 green 报告三个不同网络命名空间标识,但 PID 命名空间标识与主机名保持相同。父进程创建的标记文件,也能被两个子范围读取。这些正向观察明确了本例边界:网络范围改变了,而所检查的这些进程和文件系统属性仍然共享。不能由网络标识不同,推导其他所有隔离也已经建立。 ...

二月 19, 2028 · 2 分钟 · XB Field Notes
只有一端处于管理启用状态,对端仍关闭时链路并不具备载波。

veth 对在两个端点之间搬运帧,路由决定这些帧承载的路径

veth 对由两个互联的虚拟以太网设备组成。从一端发送的帧到达另一端,再由接收网络栈或网桥处理。两端可以放在不同 Linux 网络命名空间,因此常用于容器网络。创建这一对设备只提供链路,地址、路由、桥接关系、转发和访问策略都需要另外配置。不能把接口已经出现,等同于完整网络已经准备好。 坏掉的样子 接口显示管理状态 UP,仍可能因为 veth 对端关闭而没有载波。只读取 UP 标志,会让尚未完成的配置看起来已经就绪。应检查两个端点,并区分管理状态与 LOWER_UP。路由也可能已经指向这个接口,但链路仍无法承载有效流量,所以路由列表本身不是连通性检查,更不能代替一次真实应用交换。 veth 对不会把 IP 载荷限制为对端接口自身地址。路由选择报文经过哪个链路,随后以太网转发或接收网络栈决定下一步处理。发往容器直接邻居之外的流量,也可以通过这一对设备到达路由器或网桥。声称“不发给对端的包都会留在本地”,混淆了链路交付与 IP 目的地选择,容易误导容器出口排障。 两端生命周期也相互关联。删除一个 veth 端点,会移除另一个端点,即使对端由另一个命名空间管理。因此宿主侧清理动作可能删除容器接口。影响活跃工作负载时,应有意识地安排停止流量、调整路由和删除设备的顺序。命名空间分离了管理视图,却没有让成对设备变成两项互不影响的资源。 快速验证 将脚本保存为 check-veth.py,准备 Docker 后执行 python3 check-veth.py。固定 Linux 容器使用 Python 3.11.16、iproute2 6.1.0,以及用于嵌套命名空间的 NET_ADMIN 和 SYS_ADMIN。它只创建一次性虚拟设备。左右命名空间各获得一个端点,分别配置 192.0.2.1/30 和 192.0.2.2/30,没有移动现有宿主网卡: 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(['left','right']) as (root,stack,setting): run('ip','link','add','left0','type','veth','peer','name','right0') run('ip','link','set','left0','netns','left');run('ip','link','set','right0','netns','right') ip('left','addr','add','192.0.2.1/30','dev','left0');ip('right','addr','add','192.0.2.2/30','dev','right0') ip('left','link','set','left0','up') before=json.loads(ip('left','-j','link','show','left0').stdout)[0] assert 'LOWER_UP' not in before['flags'] failed=ns('left','ping','-n','-c','1','-W','1','192.0.2.2',check=False);assert failed.returncode!=0 ip('right','link','set','right0','up') after={name:json.loads(ip(name,'-j','link','show',interface).stdout)[0] for name,interface in [('left','left0'),('right','right0')]} assert all('LOWER_UP' in v['flags'] for v in after.values()) ip('left','neigh','flush','dev','left0');ip('right','neigh','flush','dev','right0') successful=ns('left','ping','-n','-c','1','-W','1','192.0.2.2');assert successful.returncode==0 daemon=server(stack,'right','192.0.2.2',label='right');response=request('left','192.0.2.2');assert response['ok'] ip('right','addr','add','203.0.113.9/32','dev','lo');ip('left','route','add','203.0.113.9/32','via','192.0.2.2') server(stack,'right','203.0.113.9',label='beyond-link-address');routed_response=request('left','203.0.113.9');assert routed_response['ok'] ip('left','link','delete','left0');gone=ip('right','link','show','right0',check=False);assert gone.returncode!=0 and 'does not exist' in gone.stderr print(json.dumps({'one_end_up_flags':before['flags'],'one_end_up_ping_exit':failed.returncode, 'both_up_flags':{n:v['flags'] for n,v in after.items()},'both_up_ping_exit':successful.returncode,'http':response,'other_ip_destination':routed_response, 'deleting_one_removes_peer':True,'peer_absent_diagnostic':gone.stderr.strip()})) """ 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) 起初只启用 left0,它的标志包含 UP 和 NO-CARRIER,没有 LOWER_UP,因为 right0 仍关闭。单次 ping 失败。这里同时提供链路状态和实际失败交换,没有用“接口创建成功”来替代对端已经配置正确的证据。管理状态说明操作者希望设备工作,载波状态则反映配对链路当前能否成立。 ...

十一月 21, 2027 · 2 分钟 · XB Field Notes
进程不能写根文件系统或切换到零号用户,但可以使用显式允许写入的工作卷。

安全上下文需要运行时证据来验证

Kubernetes 的 securityContext 配置进程身份及部分操作系统约束,可以缩小被攻陷应用允许执行的操作,但本身不会建立完整安全边界。应先确定应用实际需要什么操作,再配置对应字段,并在最终容器中同时测试允许与拒绝的行为。只有一份看起来严格的清单,还不能说明运行中的进程已经按预期受到限制,也不能代替应用本身的权限设计。 把设置放在正确层级 Pod 级配置可以提供用户和组身份、共享卷的组权限行为,以及 seccomp 配置。容器级字段可以覆盖适用的默认值,并设置 capabilities、权限提升和只读根文件系统。需要核对字段真实 API 范围,把熟悉的名称写在错误位置,并不会获得等价约束。相同的 securityContext 名称之下,不代表所有字段都能互换放置。 runAsUser 选择数值用户身份,runAsNonRoot 则要求容器符合非 root 策略。这些设置补充镜像元数据,不能仅凭一个用户名看起来普通,就信任实际权限。卷权限也必须允许所选身份完成必要工作。下面的 emptyDir 通过 fsGroup 提供组访问,使应用能够写入指定工作目录,而不必因此获得对整个根文件系统的写权限。 Linux capabilities 把传统上广泛关联于 root 的权限拆成不同能力。删除 ALL 会移除容器配置中的这些能力。普通容器默认并不继承全部 Linux 能力,而容器内以 root 运行,也不等于已经成功逃逸到宿主机。应审阅真实有效权限,避免用这两种过度简化替代证据。减少权限很有价值,但其效果需要准确描述。 坏掉的样子 allowPrivilegeEscalation false 会为进程设置 no_new_privs,限制通过执行程序获取新权限,例如 set-user-ID 或文件能力引起的身份与权限提升。它不会撤销进程已经拥有的全部权限,也不代替业务授权。应检查实际进程标志和有效能力,不能把每一次被拒绝的操作都归因于这个字段,否则可能误解其他身份或文件权限设置的作用。 readOnlyRootFilesystem 让容器根文件系统只读,其他挂载仍可能可写,包括明确提供的工作卷。这并不是全面禁止执行代码:解释器仍可以读取并运行可写卷中的程序。它也不会阻止网络请求,或撤销对已经挂载数据的读取。只读存储约束的是一类操作,不能直接证明攻击无法继续,更不能把“不能改根目录”扩写成“不能写入任何内容”。 ...

十一月 20, 2027 · 3 分钟 · XB Field Notes
附着在容器镜像上的摘要与加密签名。

签名镜像可以证来源

容器镜像的引用是一个可变指针,摘要则不可变。签名把摘要和一把密钥绑在一起,于是校验方能在运行之前先问:这些字节出自谁手。这正是 registry 与清单层回答不了的问题——它们只能告诉你字节从哪里来,不能告诉你谁做的。 张力在于:签名只有在被校验时才是一项控制,而校验只有在信任根被管理好时才有意义。签名只是流水线里的几行;决定哪些密钥算数、一把密钥退役之后怎么办,才是真正的工程。 摘要是身份,标签是指针 摘要是对精确清单字节的哈希,镜像一变摘要就变。对摘要签名,就是把签名绑在内容上而不是名字上。用 EC P-256 密钥对、真实的 SHA-256 摘要和一份 cosign 形状的载荷,openssl dgst -verify 返回成功;把产物翻转一个字节,摘要随之改变,同一次校验返回失败。 标签的行为不同。三个标签指向同一个已签名摘要,全都能通过校验,因为标签从来不在被签名的载荷里。这正是设计:标签是指向摘要的指针,改指向是常规的 registry 操作。把标签挪到一个从未被签名的摘要上,对解析后摘要的校验就会失败——但前提是校验方先去解析标签,而不是直接相信你手上那个可变的引用。 对标签签名绑不住任何东西 如果签的是字符串 app:v2 而不是摘要,校验就会永远成功,因为这个字符串永远不变。把那个标签指向一个完全不同的镜像,签名照样通过:它对一个字符串说了真话,而这个字符串对字节什么都没说。所以签名必须签在摘要上,校验必须先解析标签再比对。 信任根是另一半。签名只证明「持有私钥的人产出了它」;公钥,或者一张由 Fulcio 签发、并配有一条 Rekor 日志条目的证书,才把它变成「这个身份构建了它」。分发和轮换这个锚点就是安全边界,而它完全在签名之外。 校验才是控制,签名不是 流水线签完名就直接部署、不做校验,买到的只是装饰。实测对比很直白:被篡改的产物在部署步骤做校验时被挡住;同一条产物在部署步骤只记录来源信息时顺利上线——而那份来源记录从此写着被篡改的摘要。强制点应该在准入,或者部署前的一道闸门,让未签名的镜像根本到不了节点。准入控制器在这里的价值,是它不依赖流水线的自觉:策略在对象进入集群的那一刻求值,跳过签名不是「忘了做」,而是直接失败。 撤销是签名内部完全没有答案的那部分。签名没有有效期、没有吊销列表:在它的密钥下永远有效。轮换密钥并不能撤销任何东西——旧签名在新密钥下被拒,在旧密钥下依旧被接受——所以任何持有已退役密钥的人都能继续产出可通过校验的产物。解法是配短时效证书和透明日志的信任根,再加上钉住身份的策略,而不只是检查「存在某个签名」。 坏掉的样子 最经典的是策略引擎从未检查过的未签名镜像。流水线签名,部署步骤按标签拉取并运行,而这个缺口在有人往同一标签推入不同字节之前一直不可见。审计记录从此为一份没人校验过的镜像声称来源,这比完全没有来源记录更糟。 第二种是签名能通过、但签名所用的密钥你已经不再信任。因为签名内部没有撤销机制,泄露或已退役的密钥会继续签出可接受的产物,而爆炸半径是那把密钥签过的每一个产物,且是回溯性的。在策略里钉住签名身份,让证书保持短时效,并把签名记进透明日志,这样一把意料之外的密钥会以一条日志条目出现,而不是一次无声通过。 快速验证 校验摘要,再确认标签仍然指向它: cosign verify --key cosign.pub registry.example.com/app@sha256:9ab5a134... cosign tree registry.example.com/app@sha256:9ab5a134... skopeo inspect --format '{{.Digest}}' docker://registry.example.com/app:latest cosign verify 必须打印出被校验的身份并以 0 退出,cosign tree 应列出挂在该摘要上的签名与证明。skopeo inspect 必须打印出你刚校验过的同一个摘要。错误的结果是:verify 报找不到匹配的签名,或者 verify 成功而 skopeo 报出另一个摘要——那意味着标签现在指向从未被签名的字节。 ...

十月 21, 2027 · 1 分钟 · XB Field Notes
默认参数传给程序,包装脚本的进程结构决定谁收到停止信号。

ENTRYPOINT 和 CMD 定义可被运行者覆盖的启动默认值

镜像的 ENTRYPOINT 和 CMD 共同参与容器创建时的启动命令。使用 exec 形式 ENTRYPOINT 时,CMD 通常提供追加给入口的默认参数;没有入口时,CMD 可以提供可执行程序及其参数。这些是启动默认值,不是强制执行边界。有权限创建容器的运行者,可以替换入口或参数,因此不能把 ENTRYPOINT 描述成无论如何都不可改变的程序锁。 坏掉的样子 在 docker run image ... 中写到镜像名之后的参数,会替换 CMD,不是自动加到原 CMD 默认值之后。exec 形式入口仍然保留,除非显式覆盖。用户以为自己只是追加一个选项,却可能丢掉全部默认参数。因此,命令行工具需要说明哪些值可省略、哪些必须重新提供,以及自身参数解析器如何处理覆盖。 exec 形式使用 JSON 数组,不会隐式运行 shell。$DEMO 这样的字符串会原样传给程序,而不是自动展开环境变量。管道、重定向和通配符也需要真正的解释器,才会获得相应含义。shell 形式会改变参数和进程语义,所以两种形式之间的切换并不只是语法美化,不能只看命令表面似乎相同。 启动包装脚本还可能让服务藏在不转发停止信号的 shell 后面。Docker 通常向容器主进程发送停止信号,等待配置的超时,必要时强制终止。即使服务自身正确安装了 SIGTERM 处理函数,如果信号只到达另一个不会转发的进程,这个处理函数也无法发挥作用。排查优雅退出失败时,应先确认信号实际发给了谁。 快速验证 先取得固定基础镜像,再用宿主 Python 和 Docker 构建器运行脚本。实测为 Docker 29.3.1、Linux arm64,镜像内 Python 3.11.16。实验创建独立临时容器并关闭网络,等待每个服务打印就绪信息后测试停止,结束时删除自己的容器和镜像标签,不依赖启动后盲目等待固定时长来猜测服务已经准备好: ...

十月 3, 2027 · 2 分钟 · XB Field Notes
whiteout 从合并视图隐藏文件,原层仍然保留其字节。

后续镜像层删除文件,并不会擦除先前层的原始字节

容器镜像层描述按顺序应用的文件系统变化。一层可以新增或替换文件,也可以记录某个先前路径应从合并文件系统中消失。普通容器进程看到的是最终合并视图,但这个视图不等于随镜像分发的全部字节历史。因此,仅检查运行中的容器,可能遗漏早期层仍然保留的数据,尤其容易误判删除操作的实际效果。 坏掉的样子 把凭据复制进镜像,再通过后续 RUN 删除,不会从早期不可变层中移除凭据。能够取得相关层的人,仍可能恢复原始字节。同样,把很大的构建压缩包放进某层,再在下一条指令里删除,最终路径虽然不存在,分发内容仍可能包含整个压缩包。目录看起来干净,并不能证明镜像已经没有这份数据。 镜像层是文件系统变更集,不一定是紧凑的二进制差分。替换大文件,可能把替换后的完整内容放入另一层。删除通过 whiteout 标记,在应用层时屏蔽原有路径。OCI 还规定了隐藏下层目录内容的不透明目录规则,所以不能把正确应用镜像层简化成反复普通解压 tar;删除和覆盖语义需要被专门处理。 不是每条 Dockerfile 指令都会创建文件系统层。ENV、LABEL 和 CMD 改变镜像配置,可以产生标记为 empty_layer 的历史条目。因此,Dockerfile 行数、构建历史条目数,都不能可靠地替代根文件系统层数。即使没有增加应用文件字节,这些配置变化仍然影响执行方式和产物身份,不应被当成无关信息。 快速验证 使用 Python、Docker 构建器和固定摘要的基础镜像运行脚本。实测为 Docker 29.3.1,采用 containerd 镜像存储,基础镜像内为 Linux arm64 的 Python 3.11.16。标记内容特意设为公开文本。脚本构建并保存一个临时镜像,直接读取归档成员,不把归档解压到宿主文件系统,结束时删除自己的标签: from pathlib import Path from tempfile import TemporaryDirectory import json,subprocess,tarfile,uuid BASE='python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84' def docker(*args,check=True): result=subprocess.run(["docker",*args],capture_output=True,text=True,timeout=120) if check:assert result.returncode==0,result.stderr return result def inspect(image):return json.loads(docker("image","inspect",image).stdout)[0] name='xbcatm-whiteout-'+uuid.uuid4().hex[:10];tag=name+':probe';marker=b'PUBLIC_DEMONSTRATION_BYTES\n' try: with TemporaryDirectory(prefix=name+'-') as directory: root=Path(directory);(root/'original.txt').write_bytes(marker) (root/'Dockerfile').write_text(f'FROM {BASE}\nCOPY original.txt /demo/original.txt\nRUN rm /demo/original.txt\nENV EXAMPLE_MODE=demo\nLABEL example.purpose="public fixture"\nCMD ["python", "-c", "from pathlib import Path;assert not Path(\'/demo/original.txt\').exists();print(\'absent\')"]\n') docker('build','--network','none','-t',tag,str(root)) assert docker('run','--rm','--network','none',tag).stdout.strip()=='absent' archive=root/'image.tar';docker('image','save','-o',str(archive),tag) with tarfile.open(archive) as outer: manifests=json.load(outer.extractfile('manifest.json'));assert len(manifests)==1 manifest=manifests[0];config=json.load(outer.extractfile(manifest['Config'])) assert len(manifest['Layers'])==len(config['rootfs']['diff_ids']) assert len(manifest['Layers'])==len(inspect(BASE)['RootFS']['Layers'])+2 found_data=False;found_whiteout=False for layer in manifest['Layers'][-2:]: with tarfile.open(fileobj=outer.extractfile(layer),mode='r:*') as contents: for entry in contents: path=entry.name.removeprefix('./') if path=='demo/original.txt':found_data=contents.extractfile(entry).read()==marker if path=='demo/.wh.original.txt':found_whiteout=True assert found_data and found_whiteout assert all(item.get('empty_layer') for item in config['history'][-3:]) print('the final container hides the removed file, but the older layer still contains the exact public marker') print('the deletion layer contains a whiteout; ENV, LABEL and CMD add metadata history without filesystem layers') finally:docker('image','rm',tag,check=False) Dockerfile 先把 original.txt 复制到 /demo,再在后续 RUN 删除。容器默认命令断言路径不存在,并输出 absent。这验证最终运行视图,却不能证明镜像归档里也没有对应字节,所以实验接下来单独检查保存后的表示。把两个观察拆开,是避免将“看不见”误判为“没有保存”的关键。 ...

八月 16, 2027 · 2 分钟 · XB Field Notes
同一个固定基础镜像仍可对应不同构建、配置和容器可写状态。

固定基础镜像只固定一个输入,完整发布还需要其他记录

按摘要固定容器基础镜像,可以选择确定的镜像对象,避免依赖会移动的标签。这消除了一类意外构建变化,但不会让所有派生镜像或运行实例都完全相同。应用源码、构建参数、下载依赖、目标平台、运行配置和持久数据,仍可能改变构建结果及其行为。因此,基础镜像不可变是一个明确的输入约束,不是完整发布已经可复现的证明。 坏掉的样子 两个 Dockerfile 可以使用相同 FROM 摘要,却生成不同应用。即使 Dockerfile 一样,构建参数变化或未固定的下载取得新内容,也可能改变输出。所以,仅声明基础镜像不可变,无法证明完整构建是确定性的。决定这个性质的是全部输入,以及施加在输入上的转换过程;遗漏任何真正影响结果的部分,都会留下解释缺口。 完全相同的应用镜像,在不同环境变量下运行,也可能选择不同端点、功能行为或存储。可写绑定挂载还能替换镜像内已经测试过的文件。把配置留在镜像外通常有价值,但只要配置参与定义发布行为,它仍然需要版本管理和审阅。外置不等于无须控制,更不意味着配置变化不会影响同一镜像的实际表现。 普通容器还拥有可写文件系统层。紧急编辑可以改变一个实例,而镜像和其他实例保持原样。之后从未修改的镜像替换实例,编辑就会消失。因此,不可变部署实践需要把产物替换与受控运行状态结合起来;只在 FROM 后写摘要,无法自动强制团队遵守这种操作方式,也无法阻止容器内手工修补。 快速验证 使用宿主 Python、Docker 构建器和已取得的固定基础镜像运行脚本。实测环境为 Docker 29.3.1、Linux arm64,镜像内 Python 3.11.16。构建和容器关闭网络,因此结果不能归因于远端软件包变化。实验结束删除自己的临时容器和标签,不修改任何既有服务,以便把观察限定在明确的受控输入上: from pathlib import Path from tempfile import TemporaryDirectory import json,subprocess,uuid BASE='python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84' def docker(*args,check=True): result=subprocess.run(['docker',*args],capture_output=True,text=True,timeout=120) if check:assert result.returncode==0,result.stderr return result def inspect(image):return json.loads(docker('image','inspect',image).stdout)[0] name='xbcatm-pinning-'+uuid.uuid4().hex[:10];tags=[name+':one',name+':two'];container=name+'-running' try: with TemporaryDirectory(prefix=name+'-') as directory: root=Path(directory) (root/'Dockerfile').write_text(f'FROM {BASE}\nARG MESSAGE\nRUN printf "%s" "$MESSAGE" > /built-message\nENV MODE=default\nCOPY app.py /app.py\nCMD ["python", "/app.py"]\n') (root/'app.py').write_text('from pathlib import Path\nimport os\nprint(Path("/built-message").read_text(),os.environ["MODE"])\n') layers=inspect(BASE)['RootFS']['Layers'] for tag,message in zip(tags,['one','two']): docker('build','--network','none','--build-arg','MESSAGE='+message,'-t',tag,str(root)) assert inspect(tag)['RootFS']['Layers'][:len(layers)]==layers assert inspect(tags[0])['Id']!=inspect(tags[1])['Id'] assert docker('run','--rm','--network','none',tags[0]).stdout.strip()=='one default' assert docker('run','--rm','--network','none',tags[1]).stdout.strip()=='two default' assert docker('run','--rm','--network','none','-e','MODE=changed',tags[0]).stdout.strip()=='one changed' print('both builds use the same pinned base layers; different build arguments produce different artifacts') print('the identical application image behaves differently when runtime configuration changes') docker('run','-d','--name',container,'--network','none',tags[0],'python','-c','import time;time.sleep(60)') docker('exec',container,'python','-c','from pathlib import Path;Path("/built-message").write_text("edited")') assert docker('exec',container,'python','/app.py').stdout.strip()=='edited default' assert docker('run','--rm','--network','none',tags[0]).stdout.strip()=='one default' failed=docker('run','--rm','--network','none','--read-only',tags[0],'python','-c','from pathlib import Path;Path("/built-message").write_text("edited")',check=False) assert failed.returncode!=0 and 'Read-only file system' in failed.stderr print('a writable container can drift while its image stays unchanged; a read-only root rejects that particular write') finally: docker('rm','-f',container,check=False);docker('image','rm',*tags,check=False) 两次构建使用相同基础引用和应用源码,选择的构建差异只有 MESSAGE;Dockerfile 把它写入 /built-message。第一份镜像输出 one default,第二份输出 two default。完整镜像身份不同,但检查确认它们有序的基础文件系统层前缀完全一致。这里同时对照了公共输入和最终行为,避免把基础相同误当成全部相同。 ...

四月 30, 2027 · 1 分钟 · XB Field Notes
准备步骤失败后重试,再依次转换产物与启动应用;另一个 Never 策略 Pod 直接失败。

普通初始化容器建立启动前置条件

普通 init container 在 Pod 的应用容器启动之前执行准备工作。它有自己的命令、镜像和文件系统视图,也可以挂载应用稍后读取的共享卷。这让前置步骤与主进程分开,但不意味着任意初始化操作都会自动安全,也不意味着重试默认存在固定次数上限。应根据真实重试方式及产物生命周期设计这个步骤,而不是把启动顺序当成完整容错机制。 顺序约束只属于一个 Pod Kubelet 按声明顺序运行普通初始化容器。每一步必须成功结束,下一步才开始,应用容器则等待整个序列完成。检查 initContainerStatuses,才能区分正在等待的步骤、运行中的步骤和已经终止的某次尝试。仅查看 Pod 的阶段,无法说明哪条命令失败,也无法说明新的尝试是否正在进行;排查需要结合具体容器状态和日志。 共享存储需要显式挂载。emptyDir 可以在初始化容器与应用容器之间传递生成文件,并在同一 Pod 内的容器重启期间保留,但会随着 Pod 消失。不同容器不会自动共享镜像根文件系统,只写在初始化容器可写层中的文件,不会自然出现在应用容器里面。产物的写入位置和读取位置必须通过真实卷定义对应起来。 普通初始化容器需要运行到完成。Kubernetes 也支持在 initContainers 中通过容器级 restartPolicy Always 表示可重启边车,它们具有不同生命周期。因此,不能仅因为某个条目位于这个字段,就认定它必须退出后应用才能继续。本例只使用普通初始化容器,不把持续运行的边车混进顺序完成模型。 坏掉的样子 非零退出并不总会让整个 Pod 立即失败。在普通应用 Pod 采用 Always 策略时,失败的普通初始化容器会继续重试,直到成功;已经成功的初始化步骤则不会因为应用仍在运行就不断重启。退避会放慢连续尝试,却不是固定的重试次数限制。如果 Pod 的 restartPolicy 为 Never,初始化失败会使该 Pod 进入失败状态,恢复方式需要另行安排。 准备脚本应能够安全面对前一次留下的内容。中断的下载可能留下半个文件,emptyDir 中的标记也可能在失败进程退出后继续存在。应验证完整产物,再明确交付给消费者。下面用于记录次数的计数器只是诊断数据,不能作为生产逻辑,推断某个外部依赖在第二次尝试时就必然可用。真实条件必须由对应检查来判断。 ...

一月 3, 2027 · 3 分钟 · XB Field Notes