反向代理接收客户端请求，再与应用进行独立的上游交换。配置相应行为后，它可以集中路由、传输、请求头和准入策略，却不会自动隐藏所有后端接口、认证所有调用者或终止 TLS。真正有用的边界，是流量实际经过哪里、在那里执行什么策略，而不是代理在架构图上画在应用前面。描述能力时应对应配置和真实调用路径。
坏掉的样子 如果后端还能通过另一条路径到达，只放在代理上的控制就可能被绕过。不发布容器端口减少了一种暴露途径，但连接到相同网络的其他工作负载仍可能访问应用。如果后端授权完全依赖“所有流量已经经过公开代理”这个假设，内部直连请求就可能让假设失效。需要检查真正可达的调用者，而不是只检查公网端口列表。
URI 映射可能产生更隐蔽的绕过。代理阻止 /admin，并不代表宽泛的 /api/ 映射不会通过 /api/admin 暴露同一个后端处理器。应审阅最终上游 URI，而不只看客户端可见路径。公开一个明确端点，与在方便的前缀下公开所有处理器是不同决定，不能期待两个外部名称不同就天然对应不同权限边界。
转发元数据需要明确处理。客户端可能发送伪造地址或协议头，代理把它继续转发，并不会让它自动可信。应用需要知道可信一跳设置哪些字段，以及哪些调用者可以直接访问自己。浏览器响应头也不能替代请求授权或网络可达性控制。检查响应头存在，只能证明对应字段被发送，不能推断其他安全责任已经完成。
精确反向代理路由：机制与证据边界。 快速验证 保存为 check-proxy-entry.py；需要 Python、本机 Docker 和预先加载的固定镜像。作者报告使用 nginx 1.28.0 与可丢弃回显应用；精确历史输出不可用，本轮没有重跑。显式检查在优化模式下仍执行。只有代理绑定宿主回环，实验只涉及 HTTP。Docker 调用／构建分别限时 30／120 秒，就绪探测期限更短。清理检查自有容器、标签及网络；删除或 daemon 完成状态无法确认时都会失败：
from contextlib import contextmanager from pathlib import Path from tempfile import TemporaryDirectory import http.client,ipaddress,json,socket,subprocess,time,uuid def require(condition, message): if not condition: raise AssertionError(message) uncertain_commands = [] NGINX='nginx@sha256:552e7481ca93ffccd046aa658dbbed22caefbc09c66fa7cd247cbb90b8a5c609' PYTHON='python@sha256:528257d48c1da0dcecc2e725d1ae34498d60c965f1241e39cd6a85a8859bdf84' def docker(*args,check=True,timeout=30): try: result=subprocess.run(['docker',*args],capture_output=True,text=True,timeout=timeout) except subprocess.TimeoutExpired as error: uncertain_commands.append(args) raise RuntimeError('Docker command timed out: '+args[0]) from error except KeyboardInterrupt: uncertain_commands.append(args) raise if check:require(result.returncode==0,result.stderr) return result def command(*args): return docker(*args).stdout.strip() def cleanup_owned(groups): # Inspect exact owned names; absence is not inferred from a failed inspect. errors = [] for kind, names, format_string in groups: try: listing = command(kind, "ls", *( ["--all"] if kind == "container" else [] ), "--format", format_string).splitlines() present = sorted(set(names) & set(listing)) if present: command(kind, "rm", *( ["--force"] if kind == "container" else [] ), *present) except Exception as error: errors.append(f"{kind} removal: {error}") try: remaining = command(kind, "ls", *( ["--all"] if kind == "container" else [] ), "--format", format_string).splitlines() if set(names) & set(remaining): errors.append(f"{kind} remains: {sorted(set(names) & set(remaining))}") except Exception as error: errors.append(f"{kind} inventory unverified: {error}") if uncertain_commands: errors.append("A Docker command timed out or was interrupted; daemon-side completion remains uncertain") require(not errors, "Cleanup not verified: " + "; ".join(errors)) print(json.dumps({"cleanup": "owned names absent in final inventories"}), flush=True) def wait(predicate): deadline=time.monotonic()+15 while time.monotonic()<deadline: try: if predicate():return except OSError:pass time.sleep(.1) raise AssertionError('fixture did not become ready') @contextmanager def lab(configuration,files=None,backend=None,extra_ports=()): name='xbcatm-nginx-'+uuid.uuid4().hex[:10];tags=[name+':proxy',name+':backend'];containers=[name+'-proxy',name+'-backend',name+'-check',name+'-direct'] print(json.dumps({"owned_containers":containers,"owned_tags":tags,"owned_network":name}),flush=True) try: with TemporaryDirectory(prefix=name+'-') as directory: root=Path(directory);(root/'files').mkdir();(root/'nginx.conf').write_text(configuration) for path,content in (files or {}).items(): target=root/'files'/path;target.parent.mkdir(parents=True,exist_ok=True) target.write_bytes(content if isinstance(content,bytes) else content.encode()) docker('image','inspect',NGINX) if backend is not None:docker('image','inspect',PYTHON) docker('network','create',name) if backend is not None: (root/'backend.py').write_text(backend) (root/'backend.Dockerfile').write_text(f'FROM {PYTHON}\nCOPY backend.py /backend.py\nCMD ["python", "/backend.py"]\n') docker('build','--pull=false','--network','none','-f',str(root/'backend.Dockerfile'),'-t',tags[1],str(root),timeout=120) docker('run','--pull=never','-d','--name',containers[1],'--network',name,'--network-alias','backend',tags[1]) wait(lambda:docker('exec',containers[1],'python','-c','import socket;s=socket.create_connection(("127.0.0.1",8081),timeout=.5);s.close()',check=False,timeout=5).returncode==0) (root/'Dockerfile').write_text(f'FROM {NGINX}\nCOPY nginx.conf /etc/nginx/nginx.conf\nCOPY files /srv\n') docker('build','--pull=false','--network','none','-t',tags[0],str(root),timeout=120) docker('run','--pull=never','--rm','--name',containers[2],'--network',name,tags[0],'nginx','-t') published=[arg for value in (8080,*extra_ports) for arg in ('-p','127.0.0.1::'+str(value))] docker('run','--pull=never','-d','--name',containers[0],'--network',name,*published,tags[0]) details=json.loads(docker('inspect',containers[0]).stdout)[0] binding=details['NetworkSettings']['Ports']['8080/tcp'][0];require(binding['HostIp']=='127.0.0.1','non-loopback binding');port=int(binding['HostPort']) def ready(): connection=socket.create_connection(('127.0.0.1',port),timeout=.5);connection.close();return True wait(ready) def request(path,method='GET',headers=None,body=None): connection=http.client.HTTPConnection('127.0.0.1',port,timeout=5) try: connection.request(method,path,body=body,headers=headers or {}) response=connection.getresponse();return response.status,{k.lower():v for k,v in response.getheaders()},response.read() finally:connection.close() request.port=port;request.network=name request.ports={int(key.split('/')[0]):int(value[0]['HostPort']) for key,value in details['NetworkSettings']['Ports'].items() if value} yield containers[0],request finally: cleanup_owned((("container", containers, "{{.Names}}"), ("image", tags, "{{.Repository}}:{{.Tag}}"), ("network", (name,), "{{.Name}}"))) def config(locations,extra=''): return ("worker_processes 1;\nerror_log /tmp/error.log notice;\nevents {}\nhttp {\n" "log_format peer escape=json '{\"id\":\"$request_id\",\"peer\":\"$remote_addr\"," "\"supplied_xff\":\"$http_x_forwarded_for\"}';\n" "access_log /tmp/peer.log peer;\n"+extra+ "\nserver { listen 8080; root /srv;\n"+locations+"\n}\n}\n") def observed_peer(log_text,request_id,forged): require(bool(request_id), 'missing request ID') rows=[json.loads(line) for line in log_text.splitlines() if line.strip()] matches=[row for row in rows if row.get('id')==request_id] if not matches:return None require(len(matches)==1, 'duplicate request ID') row=matches[0] require(row.get('supplied_xff')==forged, 'wrong request matched') peer=str(ipaddress.ip_address(row['peer'])) require(peer!=forged, 'test peer collided with forged control') return peer backend=r''' from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer import json class Handler(BaseHTTPRequestHandler): protocol_version='HTTP/1.1' def log_message(self,*args):pass def respond(self,body=b''): data=json.dumps({'path':self.path,'method':self.command,'body':body.decode(), 'host':self.headers.get('Host'),'forwarded_for':self.headers.get('X-Forwarded-For'), 'forwarded_proto':self.headers.get('X-Forwarded-Proto')}).encode() self.send_response(200);self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(data))) self.end_headers();self.wfile.write(data) def do_GET(self):self.respond() def do_POST(self):self.respond(self.rfile.read(int(self.headers['Content-Length']))) ThreadingHTTPServer(('0.0.0.0',8081),Handler).serve_forever() ''' locations=''' location = /api/echo { proxy_set_header Host $host; proxy_set_header X-Forwarded-For $remote_addr; proxy_set_header X-Forwarded-Proto $scheme; proxy_pass http://backend:8081/echo; } location /admin { deny all; } location / { return 404; } ''' extra='client_max_body_size 1k; add_header X-Content-Type-Options nosniff always; add_header X-Lab-Request-Id $request_id always;' bad='events {} http { xbcatm_invalid_candidate; }' with lab(config(locations,extra),{'invalid.conf':bad},backend) as (proxy,request): status,h,b=request('/api/echo',method='POST',body=b'hello',headers={'Host':'edge.example', 'X-Forwarded-For':'192.0.2.123','X-Forwarded-Proto':'https'}) data=json.loads(b) require(status==200 and data['path']=='/echo' and data['body']=='hello' and data['method']=='POST','route/body mismatch') peer_rows=[] def peer_ready(): peer=observed_peer(docker('exec',proxy,'cat','/tmp/peer.log',timeout=5).stdout, h.get('x-lab-request-id'), '192.0.2.123') if peer is not None:peer_rows.append(peer);return True return False wait(peer_ready) require(data['host']=='edge.example' and data['forwarded_for']==peer_rows[0] and data['forwarded_proto']=='http','forwarded-header mismatch') require(h['x-content-type-options']=='nosniff','success header missing') results={'proxied_request':data,'proxy_observed_peer':peer_rows[0]} status,h,_=request('/admin') require(status==403 and h['x-content-type-options']=='nosniff','admin rejection mismatch') results['admin_at_proxy']=status require(request('/api/echo',method='POST',body=b'x'*2048)[0]==413,'body limit mismatch') require(request('/unknown')[0]==404 and request('/api/admin')[0]==404,'route rejection mismatch') results['size_and_route_rejections']=[413,404];results['admin_alias_rejected']=404 details=json.loads(docker('inspect',request.network+'-backend').stdout)[0] require(not details['HostConfig']['PortBindings'],'backend published to host') code=('import json,http.client; c=http.client.HTTPConnection("backend",8081,timeout=5); ' 'c.request("GET","/admin"); r=c.getresponse(); ' 'print(json.dumps({"status":r.status,"data":json.loads(r.read())})); c.close()') internal=docker('run','--pull=never','--rm','--name',request.network+'-direct', '--network',request.network,PYTHON,'python','-c',code) direct=json.loads(internal.stdout) require(direct['status']==200 and direct['data']['path']=='/admin','direct backend control failed') results['backend_has_host_port_binding']=False;results['internal_direct_path']=direct['data']['path'] require(docker('exec',proxy,'cat','/srv/invalid.conf').stdout==bad,'candidate bytes changed') docker('exec',proxy,'nginx','-t','-c','/etc/nginx/nginx.conf') probe=docker('exec',proxy,'nginx','-t','-c','/srv/invalid.conf',check=False) diagnostic='nginx: [emerg] unknown directive "xbcatm_invalid_candidate" in /srv/invalid.conf:1' require(probe.returncode==1 and diagnostic in probe.stderr.splitlines() and 'nginx: configuration file /srv/invalid.conf test failed' in probe.stderr.splitlines(), 'expected candidate-specific nginx rejection not established: '+probe.stderr) results['invalid_candidate_diagnostic']=probe.stderr require(request('/api/echo')[0]==200,'live config unavailable after candidate check') results['invalid_candidate_rejected_live_config_survives']=True # Results are printed only after the context manager's checked cleanup returns. print(json.dumps(results,sort_keys=True)) 被中断或超时的命令即使随后遇到空清单，状态仍然不确定。清理失败会通过异常链保留原始异常。再次中断或进程突然死亡可能阻止清理完成；最终清单无法排除延迟完成的 daemon 操作或并发变化。
...
