反向代理接收客户端请求，再与应用进行独立的上游交换。配置相应行为后，它可以集中路由、传输、请求头和准入策略，却不会自动隐藏所有后端接口、认证所有调用者或终止 TLS。真正有用的边界，是流量实际经过哪里、在那里执行什么策略，而不是代理在架构图上画在应用前面。描述能力时应对应配置和真实调用路径。

坏掉的样子

如果后端还能通过另一条路径到达，只放在代理上的控制就可能被绕过。不发布容器端口减少了一种暴露途径，但连接到相同网络的其他工作负载仍可能访问应用。如果后端授权完全依赖“所有流量已经经过公开代理”这个假设，内部直连请求就可能让假设失效。需要检查真正可达的调用者，而不是只检查公网端口列表。

URI 映射可能产生更隐蔽的绕过。代理阻止 /admin，并不代表宽泛的 /api/ 映射不会通过 /api/admin 暴露同一个后端处理器。应审阅最终上游 URI，而不只看客户端可见路径。公开一个明确端点，与在方便的前缀下公开所有处理器是不同决定，不能期待两个外部名称不同就天然对应不同权限边界。

转发元数据需要明确处理。客户端可能发送伪造地址或协议头，代理把它继续转发，并不会让它自动可信。应用需要知道可信一跳设置哪些字段，以及哪些调用者可以直接访问自己。浏览器响应头也不能替代请求授权或网络可达性控制。检查响应头存在，只能证明对应字段被发送，不能推断其他安全责任已经完成。

HTTP 配置预期展示精确 /api/echo 映射、独立 403、404、413 拒绝、后端直连及显式转发头赋值。
精确反向代理路由：机制与证据边界。

快速验证

保存为 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 操作或并发变化。

以下描述夹具预期断言，并非本轮找回的运行观测。POST 到精确公开路径 /api/echo 后，应以原方法和正文到达后端 /echo。提供的 Host 按明确配置转发；实验没有验证主机名允许名单，不能单凭该字段把请求认定为某个可信租户。X-Forwarded-For 检查现在要求与 nginx 日志中的 $remote_addr 相等，通过响应请求 ID 及输入的伪造头关联同一次请求；证据缺失、重复或地址格式错误都会失败。这里观察的是代理入口对端，可能是 Docker 网关，不是后端的对端，也不是已经认证的用户身份。协议检查要求结果为 http，不能把地址对应关系当成授权证据。

在代理处，/admin 返回 403，/api/admin 返回 404。后一项对照阻止另一种 URI 映射到达同一后端处理器。2,048 字节的超大提交超过 1 KiB 限制，返回 413；未知路由返回 404。配置的 nosniff 头同时出现在成功响应和显式访问拒绝中。这些结果分别验证路由、大小与响应字段，不把其中一项成功当成全部策略生效。完整结果只有在已检查的清理过程返回后才会打印；有限的清单检查仍不等于生产资源生命周期证明。

脚本检查 Docker 后端配置，确认没有宿主端口绑定。随后，在同一专用网络里启动另一个可丢弃客户端，直接请求后端 /admin，断言要求请求成功。这个预期结果用于说明边界：代理的拒绝不会作用于根本没有经过代理的路径。专用网络内部可达与公网发布不同，但仍需要与应用的调用者信任模型对应。

候选配置故意包含一个不存在的指令。核对完整字节并检查当前有效配置后，要求退出码为 1，诊断精确包含指令、路径／行号及测试失败信息。Docker 失败、可执行文件缺失、超时或无关 nginx 错误不能通过。保留标准错误输出，并要求正在运行的 HTTP 仍返回 200。这是固定镜像的预期契约，本轮未执行；验证不会激活候选。reload、证书轮换与部署仍未验证，运行与端到端技术验收继续阻塞。

明确入口策略与应用责任

盘点每个后端的所有可达路径，包括内部工作负载和单独发布的监听。按照信任模型要求实施应用授权，并把网络访问限制给预期调用者。私有 bridge 是一种连接安排，不证明所有附着工作负载具有相同权限，也不证明后端流量经过加密。只有网络结构、身份与授权规则一致，入口策略才不会依赖错误的隐含前提。

URI 暴露范围和转发字段应成为应用契约的一部分。这个小实验用精确公开路由防止意外暴露处理器；更大应用需要经过审阅的路由家族，而不是任意复制测试中的路径清单。当多个外部 URL 可能命名同一个内部操作时，应验证近似路径与改写路径。检查必须沿映射直到真实处理器，不能停在第一层名称。

TLS 可以在代理终止、继续延伸到后端，或采用其他明确设计的边界。证书、上游验证、续期和协议选择需要各自检查。同样，即使前面连接很多健康应用，一个代理仍然是一个容量与故障组件。集中策略只有在配置和可用性被实际管理时，才真正减少重复工作；它不会免费提供所有传输和运维能力。

最后，既通过入口验证，也通过其他允许路径验证。检查应用收到什么、客户端看到什么，以及哪些请求根本不应到达应用。代理适合执行共同策略，但正确性来自这些可观察边界，而不是假定建筑只有图上标出的那一扇门。实际可达性与实际执行策略，应共同决定对外作出的保护承诺。

延伸阅读

来源