Agent 读取工具遇到重定向：在下一跳发送前检查授权
允许读取一个 URL，不代表重定向后的每个目标都已获准。把检查放到下一次 send 之前，才能拦住尚未发生的访问。
1. 一个读取链接，为什么访问了另一个目标
给 Agent 一个联网读取工具时，最容易漏掉的动作，往往发生在工具函数内部。调用参数写着一个 URL，服务器回了 302，HTTP 客户端接着访问另一个 URL。模型没有再发起一次工具调用，工具外层的输入检查也就未必再执行。
本文只解决一个窄问题：在 HTTPX 准备发送下一跳之前，拿到它实际解析出的目标，按已经批准的范围作决定。我们不会让模型猜“这个跳转应该没事”，也不把库删除 Authorization 的行为当作访问许可。
实验使用真实 Python 3.12.14 和 HTTPX 0.28.1；每一个响应都由 MockTransport 合成。地址全部是 .test，header 与 body 全是假值，没有 DNS、监听端口、真实网络、浏览器或模型调用。你将看到客户端确实生成了什么请求，以及哪些目标从未进入传输层。
2. 先把“允许访问”写成可比较的范围
“可以读这个 URL”不是“可以读这个 host 的全部内容”。同一台服务器上的 /done、/admin 和 /done?export=1，可能代表完全不同的数据。端口和 scheme 变化也不能被忽略。
本文选择最容易审计的策略：批准集合里放精确 URL。比较前交给 HTTPX 解析并序列化，因而显式默认 HTTPS 端口 :443 与省略端口等价；我们不再做业务级的路径解码、查询参数排序或前缀扩展。/done 获批不会让 /done?export=1 自动获批。集合必须来自已经明确授予的权限，而不是从响应 Location 自动追加。
即使一个地址被写进集合，示例仍拒绝 HTTPX 解析后 userinfo 非空的地址、非 HTTPS 地址及 fragment 非空的地址。这两项字段检查并不是检查原始文本是否出现 @ 或 #：例如空 userinfo 的 https://@a.test/done 会被规范化成普通 URL，结尾只有 # 的地址，其 fragment 是空字符串，不触发该字段检查；它们仍须通过解析后完整 URL 的精确批准集合检查。非空 fragment 的拒绝是本例为了让审计记录保持简单的保守规则，不是在声称 fragment 会发送给服务器。如果业务批准的是路径范围，需要另写有边界的匹配规则；不要把本例的 URL 相等判断偷偷换成 startswith。
接下来的 approved 字段是本地实验预先声明的授权输入。它不代表某个真实用户批准了相应网站，也不承担从自然语言中推断权限的职责。
3. HTTPX 帮你处理跳转，但不知道任意 header 的秘密
HTTPX 0.28.1 默认不自动跟随重定向。返回 302 时，response.next_request 可以已经存在，而 history 仍为空。这个对象表示拟发送请求，不表示请求已经发出。显式开启 follow_redirects=True，才会在本例继续发送并产生 history。
原生行为矩阵先故意加入 Authorization: dummy-only 和 X-Private-Demo: dummy-only。跨 host、跨非默认端口和 HTTPS 降到 HTTP 的样本中，Authorization 被删除，自定义 header 仍被保留。同 origin 的样本则保留两者。还有一个不能省掉的例外：同 host 从默认 HTTP 端口升到默认 HTTPS 端口时，Authorization 保留；因此“所有跨 origin 都删除认证”也不是准确概括。
方法也会变化。这里 POST 遇 302 变成 GET，body 被丢掉；遇 307 或 308 则保留 POST 与 dummy-body。跨 host 的 307 可以同时出现“Authorization 没了，body 和自定义 header 还在”。HEAD 遇 302 仍是 HEAD。它们全部只走 MockTransport，不能据此获得向其他站点重放真实 POST 的许可。
这些现象说明两道检查不能合并：目标有没有批准，与请求内容能不能交给该目标，是不同的问题。本文的读取器索性不接收凭据，每跳新建一个干净 GET。需要登录态的生产工具，应另行绑定凭据与明确收件 origin，并审查 cookie、认证插件和客户端默认值。
4. 读取器的关键顺序：检查、发送、取下一跳
collect_get 接收起点、精确批准集合、注入的传输层和最大重定向数。入口先检查第一跳；随后用专用 Client，明确 follow_redirects=False、trust_env=False、auth=None。样例入口只注入 MockTransport，没有接受用户 URL 的命令行参数。
这里的 transport 是应用预先配置的可信依赖，本例固定为 MockTransport，不能来自模型或不可信调用者。trust_env=False 只关闭客户端从环境读取相关配置，不会禁用显式 transport 自己配置的代理、认证或额外发送行为。逐跳授权检查不能替代对这个依赖的信任。
发送时用 httpx.Request 新建 GET，而不是 client.build_request，也不直接把 next_request 继续送出去。区别很具体：后两种路径可能继承请求 header 或客户端 cookie 状态。示例让 HTTPX 负责解析相对 Location，只从实际 next_request 取目标和方法；授权通过之后，下一轮重新创建只有显式 Accept 与必要协议 header 的 GET。响应设置的假 cookie 也不会跟过去。
每条已发送记录包含来源 URL、状态码、解析后的下一跳和 allowed 决定；拒绝记录带目标、原因和 sent=false。这里的 URL 全是固定假样本。真实日志不能直接照抄：path、query 甚至 URL 本身都可能含私密信息，异常信息与响应内容也需要裁剪。
max_redirects=3 指最多跟随三次，连同初始请求，循环样本会发送四次后停止。零预算仍允许初始请求，但不允许下一跳。无 Location 的重定向状态被记为 missing_location；普通 404 是终止响应，complete 只表示不再跳转，不代表业务成功。HTTPX 解析或底层传输实际抛出的异常，此例不做容错封装，会原样向调用方传播；调用方不得把异常当作成功。部分非常规 Location 可能被 HTTPX 解析、规范化，因此仍须审核实际生成的 next_request.url，不能假定它们一律报错。
5. 在空目录保存三个文件，再执行一次
前提是现有环境已经具备 Python 3.12.14 与 HTTPX 0.28.1。本文不执行安装命令；版本不匹配会明确失败，不能拿另一个版本的运行结果冒充本次验证。新建空目录，把以下三个代码框分别保存为 redirect_lab.py、cases.json、expected.json，放在同一目录。源码不会读取浏览器、凭据、用户配置或外部 URL。
cases.json 是完整输入，expected.json 是独立写定的期望，不由 observed.json 反推。源码只读取前两者，写出本次 observed.json；它不会覆写预期文件。断言采用显式比较和抛错，不依赖可被 python -O 移除的 assert。
python -c "import platform, httpx; print(platform.python_version(), httpx.__version__)"
python redirect_lab.py第一条应显示 3.12.14 0.28.1。第二条的标准输出如下，完整逐请求记录写入当前脚本目录中的 observed.json。
第二条命令的完整标准输出（完整内容）
Python 3.12.14 | HTTPX 0.28.1
PASS 14 native + 19 collector = 33 cases
Transport: MockTransport only; DNS/TLS/proxy/real network NOT TESTED
Details: observed.json
redirect_lab.py（完整内容）
"""Fixed .test fixtures only. Real HTTPX client; exclusively MockTransport."""
import json
import platform
from pathlib import Path
import httpx
HERE = Path(__file__).resolve().parent
REDIRECTS = {301, 302, 303, 307, 308}
DUMMY = {"Authorization": "dummy-only", "X-Private-Demo": "dummy-only"}
def snapshot(request):
return {"url": str(request.url), "method": request.method,
"authorization": "authorization" in request.headers,
"private": "x-private-demo" in request.headers,
"cookie": "cookie" in request.headers,
"body": request.content.decode("ascii")}
def target_decision(url, approved):
# Equality uses HTTPX's parsed/serialized URL, not prefix/host matching.
if url.userinfo:
return "userinfo"
if url.scheme != "https":
return "not_https"
if url.fragment:
return "fragment"
if str(url) not in approved:
return "not_approved_url"
return "allowed"
def collect_get(start, approved_urls, transport, max_redirects=3):
"""GET-only, no caller credentials. Caller must supply approved scope.
This is a transport-injected control-flow example, not SSRF protection.
The runner below supplies only a synthetic MockTransport.
"""
if type(max_redirects) is not int or max_redirects < 0:
raise ValueError("max_redirects must be a nonnegative integer")
approved = frozenset(str(httpx.URL(u)) for u in approved_urls)
trace = []
current = httpx.URL(start)
followed = 0
with httpx.Client(transport=transport, follow_redirects=False,
trust_env=False, auth=None) as client:
while True:
verdict = target_decision(current, approved)
if verdict != "allowed":
trace.append({"target": str(current), "decision": verdict,
"sent": False})
return {"outcome": "blocked", "reason": verdict, "trace": trace}
# Request(), unlike client.build_request(), does not merge the
# client's cookie jar/default headers. Never reuse next_request headers.
request = httpx.Request("GET", current, headers={"Accept": "text/plain"})
response = client.send(request, follow_redirects=False, auth=None)
try:
candidate = response.next_request
target = candidate.url if candidate is not None else None
status = response.status_code
trace.append({"url": str(current), "status": status,
"next_target": str(target) if target else None,
"decision": "allowed", "sent": True})
finally:
response.close()
if candidate is None:
if status in REDIRECTS:
return {"outcome": "blocked", "reason": "missing_location", "trace": trace}
# 'complete' means terminal HTTP response, including 404/500.
return {"outcome": "complete", "reason": "terminal_response",
"status": status, "trace": trace}
# Inspect the actual next_request produced by this HTTPX version.
verdict = target_decision(target, approved)
if verdict == "allowed" and candidate.method != "GET":
verdict = "not_get"
if verdict == "allowed" and followed >= max_redirects:
verdict = "redirect_limit"
if verdict != "allowed":
trace.append({"target": str(target), "decision": verdict, "sent": False})
return {"outcome": "blocked", "reason": verdict, "trace": trace}
current = target
followed += 1
def run_native(case):
seen = []
def handler(request):
seen.append(snapshot(request))
if len(seen) == 1:
return httpx.Response(case["status"], headers={"Location": case["location"]})
return httpx.Response(200)
options = {} if case["follow"] is None else {"follow_redirects": case["follow"]}
with httpx.Client(transport=httpx.MockTransport(handler), trust_env=False, **options) as client:
response = client.request(case["method"], case["start"], headers=DUMMY,
content=b"dummy-body" if case["method"] == "POST" else b"")
result = {"sent": seen, "status": response.status_code,
"history": len(response.history),
"next": str(response.next_request.url) if response.next_request else None}
response.close()
return result
def run_guard(case):
sent = []
def handler(request):
sent.append(snapshot(request))
code, location, cookie = case["routes"][str(request.url)]
headers = {}
if location is not None:
headers["Location"] = location
if cookie is not None:
headers["Set-Cookie"] = cookie
return httpx.Response(code, headers=headers)
result = collect_get(case["start"], case["approved"], httpx.MockTransport(handler),
case["limit"])
return {"sent": sent, **result}
def require(actual, expected, name):
if actual != expected:
raise AssertionError(f"{name}: expected {expected!r}; got {actual!r}")
def main():
require(platform.python_version(), "3.12.14", "Python version")
require(httpx.__version__, "0.28.1", "HTTPX version")
fixtures = json.loads((HERE / "cases.json").read_text())
expected = json.loads((HERE / "expected.json").read_text())
results = {}
for case in fixtures["native"]:
name = case["name"]
result = run_native(case)
results[name] = result
want = expected[name]
observed = {"calls": len(result["sent"]), "status": result["status"],
"history": result["history"], "next": result["next"],
"last": result["sent"][-1]}
require(observed, want, name)
for case in fixtures["guard"]:
name = case["name"]
result = run_guard(case)
results[name] = result
observed = {"outcome": result["outcome"], "reason": result["reason"],
"sent_urls": [x["url"] for x in result["sent"]]}
require(observed, expected[name], name)
for request in result["sent"]:
require({k: request[k] for k in ("method", "authorization", "private", "cookie", "body")},
{"method": "GET", "authorization": False, "private": False,
"cookie": False, "body": ""}, name + " clean GET")
require(set(results), set(expected), "case coverage")
(HERE / "observed.json").write_text(json.dumps(results, ensure_ascii=False, indent=2) + "\n")
print(f"Python {platform.python_version()} | HTTPX {httpx.__version__}")
print(f"PASS {len(fixtures['native'])} native + {len(fixtures['guard'])} collector = {len(results)} cases")
print("Transport: MockTransport only; DNS/TLS/proxy/real network NOT TESTED")
print("Details: observed.json")
if __name__ == "__main__":
main()
cases.json（完整内容）
{
"native": [
{
"name": "default_no_follow",
"start": "https://a.test/start",
"location": "/done",
"status": 302,
"method": "GET",
"follow": null
},
{
"name": "same_origin",
"start": "https://a.test/start",
"location": "/done",
"status": 302,
"method": "GET",
"follow": true
},
{
"name": "cross_host",
"start": "https://a.test/start",
"location": "https://b.test/done",
"status": 302,
"method": "GET",
"follow": true
},
{
"name": "cross_port",
"start": "https://a.test/start",
"location": "https://a.test:8443/done",
"status": 302,
"method": "GET",
"follow": true
},
{
"name": "default_port_equivalent",
"start": "https://a.test/start",
"location": "https://a.test:443/done",
"status": 302,
"method": "GET",
"follow": true
},
{
"name": "https_to_http",
"start": "https://a.test/start",
"location": "http://a.test/done",
"status": 302,
"method": "GET",
"follow": true
},
{
"name": "http_to_https_exception",
"start": "http://a.test/start",
"location": "https://a.test/done",
"status": 302,
"method": "GET",
"follow": true
},
{
"name": "http_to_https_nonstandard_port",
"start": "http://a.test:8000/start",
"location": "https://a.test/done",
"status": 302,
"method": "GET",
"follow": true
},
{
"name": "scheme_relative",
"start": "https://a.test/start",
"location": "//b.test/done",
"status": 302,
"method": "GET",
"follow": true
},
{
"name": "post_302",
"start": "https://a.test/start",
"location": "/done",
"status": 302,
"method": "POST",
"follow": true
},
{
"name": "post_307",
"start": "https://a.test/start",
"location": "/done",
"status": 307,
"method": "POST",
"follow": true
},
{
"name": "post_308",
"start": "https://a.test/start",
"location": "/done",
"status": 308,
"method": "POST",
"follow": true
},
{
"name": "post_cross_307",
"start": "https://a.test/start",
"location": "https://b.test/done",
"status": 307,
"method": "POST",
"follow": true
},
{
"name": "head_302",
"start": "https://a.test/start",
"location": "/done",
"status": 302,
"method": "HEAD",
"follow": true
}
],
"guard": [
{
"name": "relative_allowed",
"routes": {
"https://a.test/start": [
302,
"/done",
null
],
"https://a.test/done": [
200,
null,
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
},
{
"name": "allowed_cross_host",
"routes": {
"https://a.test/start": [
302,
"https://b.test/done",
null
],
"https://b.test/done": [
200,
null,
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
},
{
"name": "second_hop_denied",
"routes": {
"https://a.test/start": [
302,
"https://b.test/mid",
null
],
"https://b.test/mid": [
302,
"https://c.test/done",
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
},
{
"name": "denied_initial",
"routes": {},
"start": "https://c.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
},
{
"name": "denied_port",
"routes": {
"https://a.test/start": [
302,
"https://a.test:8443/done",
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
},
{
"name": "denied_downgrade",
"routes": {
"https://a.test/start": [
302,
"http://a.test/done",
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
},
{
"name": "denied_userinfo",
"routes": {
"https://a.test/start": [
302,
"https://dummy@b.test/done",
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
},
{
"name": "redirect_loop",
"routes": {
"https://a.test/start": [
302,
"/start",
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
},
{
"name": "denied_path",
"routes": {
"https://a.test/start": [
302,
"/admin",
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
},
{
"name": "denied_query",
"routes": {
"https://a.test/start": [
302,
"/done?export=1",
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
},
{
"name": "only_one_url_granted",
"routes": {
"https://a.test/start": [
302,
"/done",
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start"
],
"limit": 3
},
{
"name": "denied_lookalike",
"routes": {
"https://a.test/start": [
302,
"https://a.test.evil.test/done",
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
},
{
"name": "scheme_relative_allowed",
"routes": {
"https://a.test/start": [
302,
"//b.test/done",
null
],
"https://b.test/done": [
200,
null,
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
},
{
"name": "cookie_not_forwarded",
"routes": {
"https://a.test/start": [
302,
"/done",
"session=dummy-only; Path=/"
],
"https://a.test/done": [
200,
null,
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
},
{
"name": "default_port_allowed",
"routes": {
"https://a.test/start": [
302,
"https://a.test:443/done",
null
],
"https://a.test/done": [
200,
null,
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
},
{
"name": "missing_location",
"routes": {
"https://a.test/start": [
302,
null,
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
},
{
"name": "zero_redirect_budget",
"routes": {
"https://a.test/start": [
302,
"/done",
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 0
},
{
"name": "fragment_denied",
"routes": {
"https://a.test/start": [
302,
"/done#part",
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
},
{
"name": "terminal_404",
"routes": {
"https://a.test/start": [
404,
null,
null
]
},
"start": "https://a.test/start",
"approved": [
"https://a.test/start",
"https://a.test/done",
"https://b.test/done",
"https://b.test/mid"
],
"limit": 3
}
]
}
expected.json（完整内容）
{
"default_no_follow": {
"calls": 1,
"status": 302,
"history": 0,
"next": "https://a.test/done",
"last": {
"url": "https://a.test/start",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
}
},
"same_origin": {
"calls": 2,
"status": 200,
"history": 1,
"next": null,
"last": {
"url": "https://a.test/done",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
}
},
"cross_host": {
"calls": 2,
"status": 200,
"history": 1,
"next": null,
"last": {
"url": "https://b.test/done",
"method": "GET",
"authorization": false,
"private": true,
"cookie": false,
"body": ""
}
},
"cross_port": {
"calls": 2,
"status": 200,
"history": 1,
"next": null,
"last": {
"url": "https://a.test:8443/done",
"method": "GET",
"authorization": false,
"private": true,
"cookie": false,
"body": ""
}
},
"default_port_equivalent": {
"calls": 2,
"status": 200,
"history": 1,
"next": null,
"last": {
"url": "https://a.test/done",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
}
},
"https_to_http": {
"calls": 2,
"status": 200,
"history": 1,
"next": null,
"last": {
"url": "http://a.test/done",
"method": "GET",
"authorization": false,
"private": true,
"cookie": false,
"body": ""
}
},
"http_to_https_exception": {
"calls": 2,
"status": 200,
"history": 1,
"next": null,
"last": {
"url": "https://a.test/done",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
}
},
"http_to_https_nonstandard_port": {
"calls": 2,
"status": 200,
"history": 1,
"next": null,
"last": {
"url": "https://a.test/done",
"method": "GET",
"authorization": false,
"private": true,
"cookie": false,
"body": ""
}
},
"scheme_relative": {
"calls": 2,
"status": 200,
"history": 1,
"next": null,
"last": {
"url": "https://b.test/done",
"method": "GET",
"authorization": false,
"private": true,
"cookie": false,
"body": ""
}
},
"post_302": {
"calls": 2,
"status": 200,
"history": 1,
"next": null,
"last": {
"url": "https://a.test/done",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
}
},
"post_307": {
"calls": 2,
"status": 200,
"history": 1,
"next": null,
"last": {
"url": "https://a.test/done",
"method": "POST",
"authorization": true,
"private": true,
"cookie": false,
"body": "dummy-body"
}
},
"post_308": {
"calls": 2,
"status": 200,
"history": 1,
"next": null,
"last": {
"url": "https://a.test/done",
"method": "POST",
"authorization": true,
"private": true,
"cookie": false,
"body": "dummy-body"
}
},
"post_cross_307": {
"calls": 2,
"status": 200,
"history": 1,
"next": null,
"last": {
"url": "https://b.test/done",
"method": "POST",
"authorization": false,
"private": true,
"cookie": false,
"body": "dummy-body"
}
},
"head_302": {
"calls": 2,
"status": 200,
"history": 1,
"next": null,
"last": {
"url": "https://a.test/done",
"method": "HEAD",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
}
},
"relative_allowed": {
"outcome": "complete",
"reason": "terminal_response",
"sent_urls": [
"https://a.test/start",
"https://a.test/done"
]
},
"allowed_cross_host": {
"outcome": "complete",
"reason": "terminal_response",
"sent_urls": [
"https://a.test/start",
"https://b.test/done"
]
},
"second_hop_denied": {
"outcome": "blocked",
"reason": "not_approved_url",
"sent_urls": [
"https://a.test/start",
"https://b.test/mid"
]
},
"denied_initial": {
"outcome": "blocked",
"reason": "not_approved_url",
"sent_urls": []
},
"denied_port": {
"outcome": "blocked",
"reason": "not_approved_url",
"sent_urls": [
"https://a.test/start"
]
},
"denied_downgrade": {
"outcome": "blocked",
"reason": "not_https",
"sent_urls": [
"https://a.test/start"
]
},
"denied_userinfo": {
"outcome": "blocked",
"reason": "userinfo",
"sent_urls": [
"https://a.test/start"
]
},
"redirect_loop": {
"outcome": "blocked",
"reason": "redirect_limit",
"sent_urls": [
"https://a.test/start",
"https://a.test/start",
"https://a.test/start",
"https://a.test/start"
]
},
"denied_path": {
"outcome": "blocked",
"reason": "not_approved_url",
"sent_urls": [
"https://a.test/start"
]
},
"denied_query": {
"outcome": "blocked",
"reason": "not_approved_url",
"sent_urls": [
"https://a.test/start"
]
},
"only_one_url_granted": {
"outcome": "blocked",
"reason": "not_approved_url",
"sent_urls": [
"https://a.test/start"
]
},
"denied_lookalike": {
"outcome": "blocked",
"reason": "not_approved_url",
"sent_urls": [
"https://a.test/start"
]
},
"scheme_relative_allowed": {
"outcome": "complete",
"reason": "terminal_response",
"sent_urls": [
"https://a.test/start",
"https://b.test/done"
]
},
"cookie_not_forwarded": {
"outcome": "complete",
"reason": "terminal_response",
"sent_urls": [
"https://a.test/start",
"https://a.test/done"
]
},
"default_port_allowed": {
"outcome": "complete",
"reason": "terminal_response",
"sent_urls": [
"https://a.test/start",
"https://a.test/done"
]
},
"missing_location": {
"outcome": "blocked",
"reason": "missing_location",
"sent_urls": [
"https://a.test/start"
]
},
"zero_redirect_budget": {
"outcome": "blocked",
"reason": "redirect_limit",
"sent_urls": [
"https://a.test/start"
]
},
"fragment_denied": {
"outcome": "blocked",
"reason": "fragment",
"sent_urls": [
"https://a.test/start"
]
},
"terminal_404": {
"outcome": "complete",
"reason": "terminal_response",
"sent_urls": [
"https://a.test/start"
]
}
}
6. 通过的重点不是 blocked 字样，而是没发出去
原有 14 项原生行为与 8 项逐跳检查全部保留，另加 11 项读取器边界，总计 33 项。新增项包括同 host 未批准路径、未批准 query、只批准一个 URL、相似域名、scheme-relative 跳转、cookie 不传播、显式默认端口、缺少 Location、零跳预算、fragment 和终止 404。
原生样本对最后一个实际请求的方法、body、两个假 header、cookie、调用次数、终止状态、history 和 next_request 作固定比较。读取器样本逐一比较 outcome、停止原因和完整 sent_urls 序列，并检查每个请求都是无 body、无 Authorization、无自定义私密 header、无 Cookie 的 GET。不是只数“通过了多少项”。
看 second_hop_denied：输入让 a.test/start 指向 b.test/mid，再让 b 指向 c.test/done。固定预期只允许调用列表出现 a、b。结果文件里的最后一条 trace 指向 c、决定为 not_approved_url、sent=false。若程序先发 c 再宣告 blocked，完整列表比较仍会失败；模拟路由也没有给 c 配响应。这比只检查函数最后返回 blocked 更接近真正要守住的时刻。
下面公开本次完整输出文件，便于与本地运行逐项比较。它是运行产物，不是期望来源。
observed.json（完整内容）
{
"default_no_follow": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
}
],
"status": 302,
"history": 0,
"next": "https://a.test/done"
},
"same_origin": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
},
{
"url": "https://a.test/done",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
}
],
"status": 200,
"history": 1,
"next": null
},
"cross_host": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
},
{
"url": "https://b.test/done",
"method": "GET",
"authorization": false,
"private": true,
"cookie": false,
"body": ""
}
],
"status": 200,
"history": 1,
"next": null
},
"cross_port": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
},
{
"url": "https://a.test:8443/done",
"method": "GET",
"authorization": false,
"private": true,
"cookie": false,
"body": ""
}
],
"status": 200,
"history": 1,
"next": null
},
"default_port_equivalent": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
},
{
"url": "https://a.test/done",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
}
],
"status": 200,
"history": 1,
"next": null
},
"https_to_http": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
},
{
"url": "http://a.test/done",
"method": "GET",
"authorization": false,
"private": true,
"cookie": false,
"body": ""
}
],
"status": 200,
"history": 1,
"next": null
},
"http_to_https_exception": {
"sent": [
{
"url": "http://a.test/start",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
},
{
"url": "https://a.test/done",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
}
],
"status": 200,
"history": 1,
"next": null
},
"http_to_https_nonstandard_port": {
"sent": [
{
"url": "http://a.test:8000/start",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
},
{
"url": "https://a.test/done",
"method": "GET",
"authorization": false,
"private": true,
"cookie": false,
"body": ""
}
],
"status": 200,
"history": 1,
"next": null
},
"scheme_relative": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
},
{
"url": "https://b.test/done",
"method": "GET",
"authorization": false,
"private": true,
"cookie": false,
"body": ""
}
],
"status": 200,
"history": 1,
"next": null
},
"post_302": {
"sent": [
{
"url": "https://a.test/start",
"method": "POST",
"authorization": true,
"private": true,
"cookie": false,
"body": "dummy-body"
},
{
"url": "https://a.test/done",
"method": "GET",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
}
],
"status": 200,
"history": 1,
"next": null
},
"post_307": {
"sent": [
{
"url": "https://a.test/start",
"method": "POST",
"authorization": true,
"private": true,
"cookie": false,
"body": "dummy-body"
},
{
"url": "https://a.test/done",
"method": "POST",
"authorization": true,
"private": true,
"cookie": false,
"body": "dummy-body"
}
],
"status": 200,
"history": 1,
"next": null
},
"post_308": {
"sent": [
{
"url": "https://a.test/start",
"method": "POST",
"authorization": true,
"private": true,
"cookie": false,
"body": "dummy-body"
},
{
"url": "https://a.test/done",
"method": "POST",
"authorization": true,
"private": true,
"cookie": false,
"body": "dummy-body"
}
],
"status": 200,
"history": 1,
"next": null
},
"post_cross_307": {
"sent": [
{
"url": "https://a.test/start",
"method": "POST",
"authorization": true,
"private": true,
"cookie": false,
"body": "dummy-body"
},
{
"url": "https://b.test/done",
"method": "POST",
"authorization": false,
"private": true,
"cookie": false,
"body": "dummy-body"
}
],
"status": 200,
"history": 1,
"next": null
},
"head_302": {
"sent": [
{
"url": "https://a.test/start",
"method": "HEAD",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
},
{
"url": "https://a.test/done",
"method": "HEAD",
"authorization": true,
"private": true,
"cookie": false,
"body": ""
}
],
"status": 200,
"history": 1,
"next": null
},
"relative_allowed": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
},
{
"url": "https://a.test/done",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "complete",
"reason": "terminal_response",
"status": 200,
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://a.test/done",
"decision": "allowed",
"sent": true
},
{
"url": "https://a.test/done",
"status": 200,
"next_target": null,
"decision": "allowed",
"sent": true
}
]
},
"allowed_cross_host": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
},
{
"url": "https://b.test/done",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "complete",
"reason": "terminal_response",
"status": 200,
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://b.test/done",
"decision": "allowed",
"sent": true
},
{
"url": "https://b.test/done",
"status": 200,
"next_target": null,
"decision": "allowed",
"sent": true
}
]
},
"second_hop_denied": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
},
{
"url": "https://b.test/mid",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "blocked",
"reason": "not_approved_url",
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://b.test/mid",
"decision": "allowed",
"sent": true
},
{
"url": "https://b.test/mid",
"status": 302,
"next_target": "https://c.test/done",
"decision": "allowed",
"sent": true
},
{
"target": "https://c.test/done",
"decision": "not_approved_url",
"sent": false
}
]
},
"denied_initial": {
"sent": [],
"outcome": "blocked",
"reason": "not_approved_url",
"trace": [
{
"target": "https://c.test/start",
"decision": "not_approved_url",
"sent": false
}
]
},
"denied_port": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "blocked",
"reason": "not_approved_url",
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://a.test:8443/done",
"decision": "allowed",
"sent": true
},
{
"target": "https://a.test:8443/done",
"decision": "not_approved_url",
"sent": false
}
]
},
"denied_downgrade": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "blocked",
"reason": "not_https",
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": "http://a.test/done",
"decision": "allowed",
"sent": true
},
{
"target": "http://a.test/done",
"decision": "not_https",
"sent": false
}
]
},
"denied_userinfo": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "blocked",
"reason": "userinfo",
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://dummy@b.test/done",
"decision": "allowed",
"sent": true
},
{
"target": "https://dummy@b.test/done",
"decision": "userinfo",
"sent": false
}
]
},
"redirect_loop": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
},
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
},
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
},
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "blocked",
"reason": "redirect_limit",
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://a.test/start",
"decision": "allowed",
"sent": true
},
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://a.test/start",
"decision": "allowed",
"sent": true
},
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://a.test/start",
"decision": "allowed",
"sent": true
},
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://a.test/start",
"decision": "allowed",
"sent": true
},
{
"target": "https://a.test/start",
"decision": "redirect_limit",
"sent": false
}
]
},
"denied_path": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "blocked",
"reason": "not_approved_url",
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://a.test/admin",
"decision": "allowed",
"sent": true
},
{
"target": "https://a.test/admin",
"decision": "not_approved_url",
"sent": false
}
]
},
"denied_query": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "blocked",
"reason": "not_approved_url",
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://a.test/done?export=1",
"decision": "allowed",
"sent": true
},
{
"target": "https://a.test/done?export=1",
"decision": "not_approved_url",
"sent": false
}
]
},
"only_one_url_granted": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "blocked",
"reason": "not_approved_url",
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://a.test/done",
"decision": "allowed",
"sent": true
},
{
"target": "https://a.test/done",
"decision": "not_approved_url",
"sent": false
}
]
},
"denied_lookalike": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "blocked",
"reason": "not_approved_url",
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://a.test.evil.test/done",
"decision": "allowed",
"sent": true
},
{
"target": "https://a.test.evil.test/done",
"decision": "not_approved_url",
"sent": false
}
]
},
"scheme_relative_allowed": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
},
{
"url": "https://b.test/done",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "complete",
"reason": "terminal_response",
"status": 200,
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://b.test/done",
"decision": "allowed",
"sent": true
},
{
"url": "https://b.test/done",
"status": 200,
"next_target": null,
"decision": "allowed",
"sent": true
}
]
},
"cookie_not_forwarded": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
},
{
"url": "https://a.test/done",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "complete",
"reason": "terminal_response",
"status": 200,
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://a.test/done",
"decision": "allowed",
"sent": true
},
{
"url": "https://a.test/done",
"status": 200,
"next_target": null,
"decision": "allowed",
"sent": true
}
]
},
"default_port_allowed": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
},
{
"url": "https://a.test/done",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "complete",
"reason": "terminal_response",
"status": 200,
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://a.test/done",
"decision": "allowed",
"sent": true
},
{
"url": "https://a.test/done",
"status": 200,
"next_target": null,
"decision": "allowed",
"sent": true
}
]
},
"missing_location": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "blocked",
"reason": "missing_location",
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": null,
"decision": "allowed",
"sent": true
}
]
},
"zero_redirect_budget": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "blocked",
"reason": "redirect_limit",
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://a.test/done",
"decision": "allowed",
"sent": true
},
{
"target": "https://a.test/done",
"decision": "redirect_limit",
"sent": false
}
]
},
"fragment_denied": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "blocked",
"reason": "fragment",
"trace": [
{
"url": "https://a.test/start",
"status": 302,
"next_target": "https://a.test/done#part",
"decision": "allowed",
"sent": true
},
{
"target": "https://a.test/done#part",
"decision": "fragment",
"sent": false
}
]
},
"terminal_404": {
"sent": [
{
"url": "https://a.test/start",
"method": "GET",
"authorization": false,
"private": false,
"cookie": false,
"body": ""
}
],
"outcome": "complete",
"reason": "terminal_response",
"status": 404,
"trace": [
{
"url": "https://a.test/start",
"status": 404,
"next_target": null,
"decision": "allowed",
"sent": true
}
]
}
}
7. 33 项通过之后，还不能说什么
这次验证到客户端的发送边界为止。MockTransport 能证明真实 HTTPX 在这些合成响应下如何改写请求，以及我们有没有把越界目标交给传输层；它不能证明真实服务器返回相同 Location，也不能证明允许的 host 最终连接到公网 IP。
本文没有实现或验收 DNS 重绑定防御、解析地址与实际连接的绑定、私网及环回 IP 检查、代理路径、TLS、网络级出站规则、响应体大小与解压上限。send 的默认读取行为也没有变成流式体积限制。超时、重试、异步客户端、认证插件和完整 cookie 域策略均不在本次测试范围内。没有浏览器或模型参与，更没有检验模型能否正确识别用户授权。
因此它是一段可复用的 GET 逐跳检查控制流，不是完整 SSRF 防护，也不适用于上传、付款或其他写入工具。接入真实传输之前，应在连接层、数据交付层和资源限制层补齐独立设计与测试。最值得带走的一条习惯是：读到 Location 之后，把“下一跳是什么”和“下一跳能不能发”分成两步，后一问必须在 send 之前得到答案。
8. 补充：只看响应头，send 仍可能先去读响应体
2026-10-13 补充。前面的 collect_get 只返回状态与跳转记录，却没有显式设置 stream。HTTPX 0.28.1 的 Client.send 默认 stream=False，会在把响应交回调用者之前执行 response.read()。原实验用空响应体，因而看不出这一步对控制流的影响。
把第一跳固定为 https://a.test/start，返回 302 与 Location: https://b.test/done；批准集合仍只有起点。现在让响应体在第一次迭代时抛出合成 ReadError。原 collect_get 的 send 会先抛异常，代码还没有走到自己的下一跳批准检查，就结束了本次调用。HTTPX 在这条异常路径上关闭响应。b.test 并未进入传输层，原来的发送边界没有被越过；这是一项响应体读取顺序的补充实验，不构成新发现的授权绕过或安全漏洞。
如果这个工具的目标确实只有响应头和跳转记录，可以显式传入 stream=True，同时保留 follow_redirects=False，让调用者拿到响应后自行决定是否继续。下文保留每一跳的精确 URL 批准规则，以及每跳重建干净 GET 的做法。读取完必要的元数据后，finally 仍须调用 response.close()。
这个顺序可直接对照 HTTPX 0.28.1 的 Client.send 源码。自动跟随还有自己的读取步骤：_send_handling_redirects 会在跟随中间响应时调用 read。因此，单独打开 stream=True，再把自动跟随也打开，仍会在本次合成 302 上触发 ReadError。
9. 保留原批准规则，只给发送调用增加 stream=True
补充实验另存一份 original_redirect_lab.py，内容与上文 redirect_lab.py 的完整源码逐字一致。它仅被导入，不会运行其 main。下面的 streaming_reader.py 从原模块复用 target_decision 与 REDIRECTS，把函数命名为 collect_headers_get，以便与原函数并排运行。除函数名外，函数体只在 client.send 增加 stream=True；原来的 finally 关闭逻辑保持原样。测试会比较两份函数的语法树，防止把批准规则或其他行为悄悄改掉。
这里的 complete/terminal_response 只表示拿到了无需继续跳转的终止响应头。响应体没有被读取或校验，状态也可能是 404 或 500。调用方如果需要正文，就不能把这个返回值当作正文下载完成；应另行定义允许读取哪些内容、何时读取、怎样限制资源，并给那个读取阶段补上测试。
streaming_reader.py（完整内容）
"""Header-only variant. Policy and cleanup are copied from the original.
The only execution change is stream=True on Client.send.
complete/terminal_response does not validate or return response body content.
"""
import httpx
from original_redirect_lab import REDIRECTS, target_decision
def collect_headers_get(start, approved_urls, transport, max_redirects=3):
"""GET-only, no caller credentials. Caller must supply approved scope.
This is a transport-injected control-flow example, not SSRF protection.
The runner below supplies only a synthetic MockTransport.
"""
if type(max_redirects) is not int or max_redirects < 0:
raise ValueError("max_redirects must be a nonnegative integer")
approved = frozenset(str(httpx.URL(u)) for u in approved_urls)
trace = []
current = httpx.URL(start)
followed = 0
with httpx.Client(transport=transport, follow_redirects=False,
trust_env=False, auth=None) as client:
while True:
verdict = target_decision(current, approved)
if verdict != "allowed":
trace.append({"target": str(current), "decision": verdict,
"sent": False})
return {"outcome": "blocked", "reason": verdict, "trace": trace}
# Request(), unlike client.build_request(), does not merge the
# client's cookie jar/default headers. Never reuse next_request headers.
request = httpx.Request("GET", current, headers={"Accept": "text/plain"})
response = client.send(request, stream=True, follow_redirects=False, auth=None)
try:
candidate = response.next_request
target = candidate.url if candidate is not None else None
status = response.status_code
trace.append({"url": str(current), "status": status,
"next_target": str(target) if target else None,
"decision": "allowed", "sent": True})
finally:
response.close()
if candidate is None:
if status in REDIRECTS:
return {"outcome": "blocked", "reason": "missing_location", "trace": trace}
# 'complete' means terminal HTTP response, including 404/500.
return {"outcome": "complete", "reason": "terminal_response",
"status": status, "trace": trace}
# Inspect the actual next_request produced by this HTTPX version.
verdict = target_decision(target, approved)
if verdict == "allowed" and candidate.method != "GET":
verdict = "not_get"
if verdict == "allowed" and followed >= max_redirects:
verdict = "redirect_limit"
if verdict != "allowed":
trace.append({"target": str(target), "decision": verdict, "sent": False})
return {"outcome": "blocked", "reason": verdict, "trace": trace}
current = target
followed += 1
original_redirect_lab.py（完整内容）
"""Fixed .test fixtures only. Real HTTPX client; exclusively MockTransport."""
import json
import platform
from pathlib import Path
import httpx
HERE = Path(__file__).resolve().parent
REDIRECTS = {301, 302, 303, 307, 308}
DUMMY = {"Authorization": "dummy-only", "X-Private-Demo": "dummy-only"}
def snapshot(request):
return {"url": str(request.url), "method": request.method,
"authorization": "authorization" in request.headers,
"private": "x-private-demo" in request.headers,
"cookie": "cookie" in request.headers,
"body": request.content.decode("ascii")}
def target_decision(url, approved):
# Equality uses HTTPX's parsed/serialized URL, not prefix/host matching.
if url.userinfo:
return "userinfo"
if url.scheme != "https":
return "not_https"
if url.fragment:
return "fragment"
if str(url) not in approved:
return "not_approved_url"
return "allowed"
def collect_get(start, approved_urls, transport, max_redirects=3):
"""GET-only, no caller credentials. Caller must supply approved scope.
This is a transport-injected control-flow example, not SSRF protection.
The runner below supplies only a synthetic MockTransport.
"""
if type(max_redirects) is not int or max_redirects < 0:
raise ValueError("max_redirects must be a nonnegative integer")
approved = frozenset(str(httpx.URL(u)) for u in approved_urls)
trace = []
current = httpx.URL(start)
followed = 0
with httpx.Client(transport=transport, follow_redirects=False,
trust_env=False, auth=None) as client:
while True:
verdict = target_decision(current, approved)
if verdict != "allowed":
trace.append({"target": str(current), "decision": verdict,
"sent": False})
return {"outcome": "blocked", "reason": verdict, "trace": trace}
# Request(), unlike client.build_request(), does not merge the
# client's cookie jar/default headers. Never reuse next_request headers.
request = httpx.Request("GET", current, headers={"Accept": "text/plain"})
response = client.send(request, follow_redirects=False, auth=None)
try:
candidate = response.next_request
target = candidate.url if candidate is not None else None
status = response.status_code
trace.append({"url": str(current), "status": status,
"next_target": str(target) if target else None,
"decision": "allowed", "sent": True})
finally:
response.close()
if candidate is None:
if status in REDIRECTS:
return {"outcome": "blocked", "reason": "missing_location", "trace": trace}
# 'complete' means terminal HTTP response, including 404/500.
return {"outcome": "complete", "reason": "terminal_response",
"status": status, "trace": trace}
# Inspect the actual next_request produced by this HTTPX version.
verdict = target_decision(target, approved)
if verdict == "allowed" and candidate.method != "GET":
verdict = "not_get"
if verdict == "allowed" and followed >= max_redirects:
verdict = "redirect_limit"
if verdict != "allowed":
trace.append({"target": str(target), "decision": verdict, "sent": False})
return {"outcome": "blocked", "reason": verdict, "trace": trace}
current = target
followed += 1
def run_native(case):
seen = []
def handler(request):
seen.append(snapshot(request))
if len(seen) == 1:
return httpx.Response(case["status"], headers={"Location": case["location"]})
return httpx.Response(200)
options = {} if case["follow"] is None else {"follow_redirects": case["follow"]}
with httpx.Client(transport=httpx.MockTransport(handler), trust_env=False, **options) as client:
response = client.request(case["method"], case["start"], headers=DUMMY,
content=b"dummy-body" if case["method"] == "POST" else b"")
result = {"sent": seen, "status": response.status_code,
"history": len(response.history),
"next": str(response.next_request.url) if response.next_request else None}
response.close()
return result
def run_guard(case):
sent = []
def handler(request):
sent.append(snapshot(request))
code, location, cookie = case["routes"][str(request.url)]
headers = {}
if location is not None:
headers["Location"] = location
if cookie is not None:
headers["Set-Cookie"] = cookie
return httpx.Response(code, headers=headers)
result = collect_get(case["start"], case["approved"], httpx.MockTransport(handler),
case["limit"])
return {"sent": sent, **result}
def require(actual, expected, name):
if actual != expected:
raise AssertionError(f"{name}: expected {expected!r}; got {actual!r}")
def main():
require(platform.python_version(), "3.12.14", "Python version")
require(httpx.__version__, "0.28.1", "HTTPX version")
fixtures = json.loads((HERE / "cases.json").read_text())
expected = json.loads((HERE / "expected.json").read_text())
results = {}
for case in fixtures["native"]:
name = case["name"]
result = run_native(case)
results[name] = result
want = expected[name]
observed = {"calls": len(result["sent"]), "status": result["status"],
"history": result["history"], "next": result["next"],
"last": result["sent"][-1]}
require(observed, want, name)
for case in fixtures["guard"]:
name = case["name"]
result = run_guard(case)
results[name] = result
observed = {"outcome": result["outcome"], "reason": result["reason"],
"sent_urls": [x["url"] for x in result["sent"]]}
require(observed, expected[name], name)
for request in result["sent"]:
require({k: request[k] for k in ("method", "authorization", "private", "cookie", "body")},
{"method": "GET", "authorization": False, "private": False,
"cookie": False, "body": ""}, name + " clean GET")
require(set(results), set(expected), "case coverage")
(HERE / "observed.json").write_text(json.dumps(results, ensure_ascii=False, indent=2) + "\n")
print(f"Python {platform.python_version()} | HTTPX {httpx.__version__}")
print(f"PASS {len(fixtures['native'])} native + {len(fixtures['guard'])} collector = {len(results)} cases")
print("Transport: MockTransport only; DNS/TLS/proxy/real network NOT TESTED")
print("Details: observed.json")
if __name__ == "__main__":
main()
10. 用五个固定输入重跑，再核对关闭动作
环境仍是已经安装好的 Python 3.12.14 与 HTTPX 0.28.1，无需也不会安装依赖。在空目录保存本补充给出的六个输入文件：original_redirect_lab.py、streaming_reader.py、streaming_lab.py、test_streaming.py、streaming_cases.json、streaming_expected.json。它们不需要原文的 cases.json 或 expected.json。执行下面的命令，程序会另写 streaming_observed.json。版本不一致时会明确失败。
python streaming_lab.py
python -m unittest -v test_streaming
python -O streaming_lab.py
python -O -m unittest -v test_streamingstreaming_expected.json 是预先写定的五条期望，运行器只读它，不会由实测结果更新它。测试中还逐项写死了返回结果、发送 URL、响应体迭代次数和关闭次数。对照项故意保留默认读取，以及“流式发送但自动跟随”的路径；两项都必须得到 ReadError，才说明测试确实碰到了这次要观察的读取步骤。
每个合成响应体都记录 __iter__ 与 close 的调用次数，测试同时检查 HTTPX Response.is_closed。手动流式的拒绝、允许继续和缺少 Location 三种情况，都应在未迭代 body 的情况下关闭响应；两跳均批准时，终止响应的 body 也仍然未读。测试另覆盖入口未批准、零跳预算、循环预算、URL query 越界以及后续批准检查抛异常时的关闭状态。
streaming_cases.json（完整内容）
[
{"name":"original_unapproved","mode":"original","approved":["https://a.test/start"],"location":"https://b.test/done"},
{"name":"streamed_unapproved","mode":"manual_stream","approved":["https://a.test/start"],"location":"https://b.test/done"},
{"name":"streamed_both_approved","mode":"manual_stream","approved":["https://a.test/start","https://b.test/done"],"location":"https://b.test/done"},
{"name":"streamed_missing_location","mode":"manual_stream","approved":["https://a.test/start"],"location":null},
{"name":"streamed_automatic_follow","mode":"automatic_stream","approved":["https://a.test/start","https://b.test/done"],"location":"https://b.test/done"}
]
streaming_expected.json（完整内容）
[
{"case":"original_unapproved","result":"ReadError","sent_urls":["https://a.test/start"],"body_iterations":[1],"body_close_calls":[1],"response_closed":[true]},
{"case":"streamed_unapproved","result":"blocked/not_approved_url","sent_urls":["https://a.test/start"],"body_iterations":[0],"body_close_calls":[1],"response_closed":[true]},
{"case":"streamed_both_approved","result":"complete/terminal_response","sent_urls":["https://a.test/start","https://b.test/done"],"body_iterations":[0,0],"body_close_calls":[1,1],"response_closed":[true,true]},
{"case":"streamed_missing_location","result":"blocked/missing_location","sent_urls":["https://a.test/start"],"body_iterations":[0],"body_close_calls":[1],"response_closed":[true]},
{"case":"streamed_automatic_follow","result":"ReadError","sent_urls":["https://a.test/start"],"body_iterations":[1],"body_close_calls":[1],"response_closed":[true]}
]
streaming_lab.py（完整内容）
"""Five fixed, offline response-body probes. No user URL input, no installs."""
from contextlib import ExitStack, contextmanager
import json
from pathlib import Path
import platform
import socket
from unittest.mock import patch
import httpx
from original_redirect_lab import collect_get
from streaming_reader import collect_headers_get
HERE = Path(__file__).resolve().parent
START = "https://a.test/start"
NEXT = "https://b.test/done"
class SyntheticReadFailure(httpx.SyncByteStream):
"""This fixture fails before yielding any bytes; it is not a real server."""
def __init__(self):
self.iterations = 0
self.close_calls = 0
def __iter__(self):
self.iterations += 1
raise httpx.ReadError("synthetic body unavailable")
yield b"" # Keep __iter__ a generator; no bytes are yielded.
def close(self):
self.close_calls += 1
def forbid_network(*args, **kwargs):
raise AssertionError("DNS/socket access is forbidden in this fixed lab")
@contextmanager
def offline_guard():
# Defense against accidental changes, not an OS/network sandbox claim.
with ExitStack() as stack:
for target, attr in [(socket, "getaddrinfo"), (socket, "create_connection"),
(socket.socket, "connect"), (socket.socket, "connect_ex")]:
stack.enter_context(patch.object(target, attr, forbid_network))
yield
def require_equal(actual, expected, label):
# Explicit checks still execute under python -O.
if actual != expected:
raise AssertionError(f"{label}: expected {expected!r}; got {actual!r}")
def check_versions():
require_equal(platform.python_version(), "3.12.14", "Python version")
require_equal(httpx.__version__, "0.28.1", "HTTPX version")
def run_case(case):
sent, bodies, responses = [], [], []
def handler(request):
url = str(request.url)
# No route is synthesized for an unexpected or third request.
require_equal(url, START if not sent else NEXT, "fixed route")
if len(sent) >= 2:
raise AssertionError("unexpected third request")
require_equal(request.method, "GET", "method")
require_equal(request.content, b"", "request body")
for name in ("authorization", "x-private-demo", "cookie"):
require_equal(name in request.headers, False, "clean GET " + name)
sent.append(url)
body = SyntheticReadFailure()
bodies.append(body)
first = len(sent) == 1
headers = {"Location": case["location"]} if first and case["location"] is not None else {}
response = httpx.Response(302 if first else 200, headers=headers, stream=body)
responses.append(response)
return response
with offline_guard():
try:
transport = httpx.MockTransport(handler)
if case["mode"] == "automatic_stream":
# Deliberate counterexample: no per-hop approval loop here.
# Both URLs are approved in the fixed input; do not reuse this
# branch as an authorization implementation.
with httpx.Client(transport=transport, trust_env=False, auth=None) as client:
response = client.send(httpx.Request("GET", START), stream=True,
follow_redirects=True, auth=None)
try:
result = "unexpected_return"
finally:
response.close()
else:
readers = {"original": collect_get, "manual_stream": collect_headers_get}
got = readers[case["mode"]](START, case["approved"], transport)
result = got["outcome"] + "/" + got["reason"]
except httpx.ReadError:
result = "ReadError"
return {"case": case["name"], "result": result, "sent_urls": sent,
"body_iterations": [b.iterations for b in bodies],
"body_close_calls": [b.close_calls for b in bodies],
"response_closed": [r.is_closed for r in responses]}
def load_cases():
return json.loads((HERE / "streaming_cases.json").read_text())
def main():
check_versions()
cases = load_cases()
expected = json.loads((HERE / "streaming_expected.json").read_text())
observed = [run_case(case) for case in cases]
require_equal(observed, expected, "all five fixed cases")
report = {"python": platform.python_version(), "httpx": httpx.__version__,
"transport": "MockTransport only", "results": observed}
(HERE / "streaming_observed.json").write_text(json.dumps(report, indent=2) + "\n")
print(f"Python {platform.python_version()} | HTTPX {httpx.__version__}")
print("PASS 5 fixed streaming cases; exact expected output matched")
print("All 6 synthetic response streams closed exactly once")
print("No DNS/socket/server/model; no response-body validation")
print("Details: streaming_observed.json")
if __name__ == "__main__":
main()
test_streaming.py（完整内容）
"""Independent fixed assertions, policy checks, and cleanup checks (stdlib)."""
import ast
import copy
import inspect
import json
from pathlib import Path
import socket
import unittest
from unittest.mock import patch
import httpx
import original_redirect_lab as original
import streaming_reader as streamed
import streaming_lab as lab
HERE = Path(__file__).resolve().parent
class StreamingTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
lab.check_versions()
def setUp(self):
self.cases = lab.load_cases()
self.guard = lab.offline_guard()
self.guard.__enter__()
self.addCleanup(self.guard.__exit__, None, None, None)
def test_five_exact_expected_records(self):
expected = json.loads((HERE / "streaming_expected.json").read_text())
self.assertEqual([lab.run_case(c) for c in self.cases], expected)
def test_literal_counterexample_expectations(self):
# These literal expectations are independent of the JSON comparator.
results = [lab.run_case(c) for c in self.cases]
self.assertEqual([r["result"] for r in results], [
"ReadError", "blocked/not_approved_url", "complete/terminal_response",
"blocked/missing_location", "ReadError"])
self.assertEqual([r["sent_urls"] for r in results], [
[lab.START], [lab.START], [lab.START, lab.NEXT], [lab.START], [lab.START]])
self.assertEqual([r["body_iterations"] for r in results], [[1], [0], [0, 0], [0], [1]])
self.assertEqual([r["body_close_calls"] for r in results], [[1], [1], [1, 1], [1], [1]])
self.assertEqual([r["response_closed"] for r in results], [[True], [True], [True, True], [True], [True]])
def test_policy_code_diff_is_only_stream_flag(self):
old = ast.parse(inspect.getsource(original.collect_get))
new = ast.parse(inspect.getsource(streamed.collect_headers_get))
old.body[0].name = "collect_headers_get"
sends = [n for n in ast.walk(old) if isinstance(n, ast.Call)
and isinstance(n.func, ast.Attribute) and n.func.attr == "send"]
self.assertEqual(len(sends), 1)
sends[0].keywords.insert(0, ast.keyword(arg="stream", value=ast.Constant(value=True)))
self.assertEqual(ast.dump(old, include_attributes=False), ast.dump(new, include_attributes=False))
self.assertIs(streamed.target_decision, original.target_decision)
def test_corrupted_expectation_is_rejected(self):
observed = lab.run_case(self.cases[1])
wrong = copy.deepcopy(observed)
wrong["sent_urls"].append(lab.NEXT)
with self.assertRaises(AssertionError):
lab.require_equal(observed, wrong, "negative comparator")
def test_default_send_is_not_header_only(self):
result = lab.run_case(self.cases[0])
self.assertEqual(result["body_iterations"], [1])
self.assertNotEqual(result["result"], "blocked/not_approved_url")
def test_auto_follow_still_consumes_intermediate(self):
result = lab.run_case(self.cases[4])
self.assertEqual(result["body_iterations"], [1])
self.assertEqual(result["result"], "ReadError")
self.assertEqual(result["sent_urls"], [lab.START])
def collect(self, *, approved, location=lab.NEXT, status=302, limit=3):
sent, bodies, responses = [], [], []
def handler(request):
sent.append(str(request.url))
body = lab.SyntheticReadFailure()
bodies.append(body)
headers = {"Location": location} if location is not None else {}
response = httpx.Response(status, headers=headers, stream=body)
responses.append(response)
return response
result = streamed.collect_headers_get(lab.START, approved, httpx.MockTransport(handler), limit)
return result, sent, bodies, responses
def check_unread_closed(self, bodies, responses):
self.assertEqual([b.iterations for b in bodies], [0] * len(bodies))
self.assertEqual([b.close_calls for b in bodies], [1] * len(bodies))
self.assertTrue(all(r.is_closed for r in responses))
def test_unapproved_start_never_enters_transport(self):
result, sent, bodies, responses = self.collect(approved=[])
self.assertEqual(result["reason"], "not_approved_url")
self.assertEqual((sent, bodies, responses), ([], [], []))
def test_zero_budget_closes_first_response(self):
result, sent, bodies, responses = self.collect(approved=[lab.START, lab.NEXT], limit=0)
self.assertEqual(result["reason"], "redirect_limit")
self.assertEqual(sent, [lab.START])
self.check_unread_closed(bodies, responses)
def test_same_exact_url_policy_rejects_query_http_userinfo_fragment(self):
samples = [(lab.NEXT + "?export=1", "not_approved_url"),
("http://b.test/done", "not_https"),
("https://dummy@b.test/done", "userinfo"),
(lab.NEXT + "#part", "fragment")]
for target, reason in samples:
with self.subTest(target=target):
result, sent, bodies, responses = self.collect(approved=[lab.START, lab.NEXT], location=target)
self.assertEqual(result["reason"], reason)
self.assertEqual(sent, [lab.START])
self.check_unread_closed(bodies, responses)
def test_loop_budget_closes_four_responses(self):
result, sent, bodies, responses = self.collect(approved=[lab.START], location=lab.START)
self.assertEqual(result["reason"], "redirect_limit")
self.assertEqual(sent, [lab.START] * 4)
self.check_unread_closed(bodies, responses)
def test_terminal_404_is_complete_without_validating_body(self):
result, sent, bodies, responses = self.collect(approved=[lab.START], location=None, status=404)
self.assertEqual((result["outcome"], result["status"]), ("complete", 404))
self.assertEqual(sent, [lab.START])
self.check_unread_closed(bodies, responses)
def test_invalid_budget_rejected(self):
for invalid in (-1, True, 1.5, "3"):
with self.subTest(limit=invalid), self.assertRaises(ValueError):
self.collect(approved=[lab.START], limit=invalid)
def test_cleanup_before_later_policy_exception(self):
bodies, responses = [], []
def handler(request):
body = lab.SyntheticReadFailure()
response = httpx.Response(302, headers={"Location": lab.NEXT}, stream=body)
bodies.append(body)
responses.append(response)
return response
with patch.object(streamed, "target_decision", side_effect=["allowed", RuntimeError("injected policy error")]):
with self.assertRaisesRegex(RuntimeError, "injected policy error"):
streamed.collect_headers_get(lab.START, [lab.START, lab.NEXT], httpx.MockTransport(handler))
self.check_unread_closed(bodies, responses)
def test_socket_tripwire_rejects_attempt(self):
with self.assertRaisesRegex(AssertionError, "forbidden"):
socket.getaddrinfo("a.test", 443)
if __name__ == "__main__":
unittest.main(verbosity=2)
11. 结果里的 0 次迭代，只适用于这次合成响应流
五条固定输出全部匹配。原函数与自动跟随对照各迭代一次并抛出 ReadError；手动流式的三个输入分别返回 blocked/not_approved_url、complete/terminal_response 与 blocked/missing_location。两跳都批准的输入发送 a、b 两次，其余四条都只发送 a。六个被创建的响应流均关闭一次，六个 Response 的 is_closed 均为 true。完整输出保留在下面，便于逐项对照。
python streaming_lab.py 的完整标准输出
Python 3.12.14 | HTTPX 0.28.1
PASS 5 fixed streaming cases; exact expected output matched
All 6 synthetic response streams closed exactly once
No DNS/socket/server/model; no response-body validation
Details: streaming_observed.json
streaming_observed.json（完整内容）
{
"python": "3.12.14",
"httpx": "0.28.1",
"transport": "MockTransport only",
"results": [
{
"case": "original_unapproved",
"result": "ReadError",
"sent_urls": [
"https://a.test/start"
],
"body_iterations": [
1
],
"body_close_calls": [
1
],
"response_closed": [
true
]
},
{
"case": "streamed_unapproved",
"result": "blocked/not_approved_url",
"sent_urls": [
"https://a.test/start"
],
"body_iterations": [
0
],
"body_close_calls": [
1
],
"response_closed": [
true
]
},
{
"case": "streamed_both_approved",
"result": "complete/terminal_response",
"sent_urls": [
"https://a.test/start",
"https://b.test/done"
],
"body_iterations": [
0,
0
],
"body_close_calls": [
1,
1
],
"response_closed": [
true,
true
]
},
{
"case": "streamed_missing_location",
"result": "blocked/missing_location",
"sent_urls": [
"https://a.test/start"
],
"body_iterations": [
0
],
"body_close_calls": [
1
],
"response_closed": [
true
]
},
{
"case": "streamed_automatic_follow",
"result": "ReadError",
"sent_urls": [
"https://a.test/start"
],
"body_iterations": [
1
],
"body_close_calls": [
1
],
"response_closed": [
true
]
}
]
}
14 个 unittest 的完整输出
test_auto_follow_still_consumes_intermediate (test_streaming.StreamingTests.test_auto_follow_still_consumes_intermediate) ... ok
test_cleanup_before_later_policy_exception (test_streaming.StreamingTests.test_cleanup_before_later_policy_exception) ... ok
test_corrupted_expectation_is_rejected (test_streaming.StreamingTests.test_corrupted_expectation_is_rejected) ... ok
test_default_send_is_not_header_only (test_streaming.StreamingTests.test_default_send_is_not_header_only) ... ok
test_five_exact_expected_records (test_streaming.StreamingTests.test_five_exact_expected_records) ... ok
test_invalid_budget_rejected (test_streaming.StreamingTests.test_invalid_budget_rejected) ... ok
test_literal_counterexample_expectations (test_streaming.StreamingTests.test_literal_counterexample_expectations) ... ok
test_loop_budget_closes_four_responses (test_streaming.StreamingTests.test_loop_budget_closes_four_responses) ... ok
test_policy_code_diff_is_only_stream_flag (test_streaming.StreamingTests.test_policy_code_diff_is_only_stream_flag) ... ok
test_same_exact_url_policy_rejects_query_http_userinfo_fragment (test_streaming.StreamingTests.test_same_exact_url_policy_rejects_query_http_userinfo_fragment) ... ok
test_socket_tripwire_rejects_attempt (test_streaming.StreamingTests.test_socket_tripwire_rejects_attempt) ... ok
test_terminal_404_is_complete_without_validating_body (test_streaming.StreamingTests.test_terminal_404_is_complete_without_validating_body) ... ok
test_unapproved_start_never_enters_transport (test_streaming.StreamingTests.test_unapproved_start_never_enters_transport) ... ok
test_zero_budget_closes_first_response (test_streaming.StreamingTests.test_zero_budget_closes_first_response) ... ok
----------------------------------------------------------------------
Ran 14 tests in 0.009s
OK
本次普通运行与 python -O 运行都通过五条固定比较和 14 个 unittest；连续重跑得到相同的 JSON。负向校验会故意把未批准目标加入预期发送列表，并确认比较器确实报错。函数体语法树检查只允许新增 stream=True，避免把“看起来修好了”建立在批准策略变动之上。
这些计数来自 MockTransport 的自定义 SyncByteStream。实验没有 DNS 查询、套接字连接、真实服务器、浏览器或模型调用。真实传输层可能缓冲或预读数据；这里的 0 次迭代不能换算成网络上读了 0 字节，也没有测出固定内存上限、正文大小上限或解压限制。stream=True 本身不承担 SSRF 防护。原文列出的连接层、代理和资源限制仍需单独设计与验收。
关闭未读响应与验证正文是不同的工作。Response.read 负责消费响应体，Response.close 负责关闭流。本例有意不进入正文消费阶段；若正文损坏，终止响应头已拿到也不能为其内容背书。
Authorization 被删掉，就可以自动跟随重定向吗？
不能。目标仍需处于明确批准范围；自定义 header 和 307/308 的 body 也可能继续传播。
批准一个 URL 是否等于批准整个 host？
不是。本文使用精确 URL 集合，未批准的路径、query 或端口都不会自动放行。
MockTransport 的 33 项通过是否证明防住了 SSRF？
不能。它只验证固定输入下的客户端控制流，未检查 DNS、私网、实际连接地址、代理或出站网络限制。