反向代理接收客户端请求,再与应用进行独立的上游交换。配置相应行为后,它可以集中路由、传输、请求头和准入策略,却不会自动隐藏所有后端接口、认证所有调用者或终止 TLS。真正有用的边界,是流量实际经过哪里、在那里执行什么策略,而不是代理在架构图上画在应用前面。描述能力时应对应配置和真实调用路径。
坏掉的样子 如果后端还能通过另一条路径到达,只放在代理上的控制就可能被绕过。不发布容器端口减少了一种暴露途径,但连接到相同网络的其他工作负载仍可能访问应用。如果后端授权完全依赖“所有流量已经经过公开代理”这个假设,内部直连请求就可能让假设失效。需要检查真正可达的调用者,而不是只检查公网端口列表。
URI 映射可能产生更隐蔽的绕过。代理阻止 /admin,并不代表宽泛的 /api/ 映射不会通过 /api/admin 暴露同一个后端处理器。应审阅最终上游 URI,而不只看客户端可见路径。公开一个明确端点,与在方便的前缀下公开所有处理器是不同决定,不能期待两个外部名称不同就天然对应不同权限边界。
转发元数据需要明确处理。客户端可能发送伪造地址或协议头,代理把它继续转发,并不会让它自动可信。应用需要知道可信一跳设置哪些字段,以及哪些调用者可以直接访问自己。浏览器响应头也不能替代请求授权或网络可达性控制。检查响应头存在,只能证明对应字段被发送,不能推断其他安全责任已经完成。
快速验证 将脚本保存为 check-proxy-entry.py,准备 Python、Docker 和固定镜像后运行。此前记录的环境使用 nginx 1.28.0,以及专用 Docker 网络里的可丢弃回显应用;本轮没有重跑。只有代理绑定宿主回环端口。实验使用 HTTP 检查路由与策略,并没有执行 TLS 部署;结束后清理自己的容器、镜像标签和网络,不改动已有服务:
from contextlib import contextmanager from pathlib import Path from tempfile import TemporaryDirectory import http.client,json,socket,subprocess,time,uuid NGINX='nginx@sha256:552e7481ca93ffccd046aa658dbbed22caefbc09c66fa7cd247cbb90b8a5c609' PYTHON='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 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'] 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('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','--network','none','-f',str(root/'backend.Dockerfile'),'-t',tags[1],str(root)) docker('run','-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).returncode==0) (root/'Dockerfile').write_text(f'FROM {NGINX}\nCOPY nginx.conf /etc/nginx/nginx.conf\nCOPY files /srv\n') docker('build','--network','none','-t',tags[0],str(root)) docker('run','--rm','--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','-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];assert binding['HostIp']=='127.0.0.1';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: docker('rm','-f',*containers,check=False);docker('image','rm',*tags,check=False);docker('network','rm',name,check=False) def config(locations,extra=''): return 'worker_processes 1;\nerror_log /tmp/error.log notice;\nevents {}\nhttp {\naccess_log off;\n'+extra+'\nserver { listen 8080; root /srv;\n'+locations+'\n}\n}\n' 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;' bad='events {} http { server { listen not-a-port; } }' 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);assert status==200 and data['path']=='/echo' and data['body']=='hello' and data['method']=='POST' assert data['host']=='edge.example' and data['forwarded_for']!='192.0.2.123' and data['forwarded_proto']=='http' assert h['x-content-type-options']=='nosniff';results={'proxied_request':data} status,h,_=request('/admin');assert status==403 and h['x-content-type-options']=='nosniff';results['admin_at_proxy']=status assert request('/api/echo',method='POST',body=b'x'*2048)[0]==413 assert request('/unknown')[0]==404 and request('/api/admin')[0]==404 results['size_and_route_rejections']=[413,404];results['admin_alias_rejected']=404 details=json.loads(docker('inspect',request.network+'-backend').stdout)[0] assert not details['HostConfig']['PortBindings'] code='import json,urllib.request;print(json.dumps(json.load(urllib.request.urlopen("http://backend:8081/admin"))))' internal=docker('run','--rm','--network',request.network,PYTHON,'python','-c',code) direct=json.loads(internal.stdout);assert direct['path']=='/admin' results['backend_has_host_port_binding']=False;results['internal_direct_path']=direct['path'] probe=docker('exec',proxy,'nginx','-t','-c','/srv/invalid.conf',check=False);assert probe.returncode!=0 assert request('/api/echo')[0]==200;results['invalid_candidate_rejected_live_config_survives']=True print(json.dumps(results,sort_keys=True)) POST 到精确公开路径 /api/echo 后,以原方法和正文到达后端 /echo。提供的 Host 按明确配置转发;实验没有验证主机名允许名单,不能单凭该字段把请求认定为某个可信租户。伪造 X-Forwarded-For 被替换为 nginx 观察到的套接字对端,伪造的 https 协议被替换为这个监听实际使用的 http。响应揭示后端真正收到的字段,而不是只根据指令存在就假定效果正确。
...