fix(site-import): BLOCKED_NETS 补 v6 未指定/组播/保留段与 NAT64
- ::/128(v6 版 0.0.0.0,此前两层防线均放行)+ 224.0.0.0/4(组播) + 240.0.0.0/4(保留/广播,含 255.255.255.255)+ 64:ff9b::/96(NAT64) + 198.18.0.0/15(基准段) - 测试 +9:六段参数化钉位、http://[::]/x URL 层拒绝、重定向协议切换 (302→ftp://)第二跳拦截、DNS rebinding 端到端(真实 client 仅打桩 getaddrinfo 返 10.x,证明连接层校验先于连接发生) Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
@@ -28,8 +28,10 @@ from httpcore._exceptions import ConnectError, ConnectTimeout, map_exceptions
|
||||
|
||||
BLOCKED_NETS = [ipaddress.ip_network(n) for n in [
|
||||
"127.0.0.0/8", "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16",
|
||||
"169.254.0.0/16", "0.0.0.0/8", "100.64.0.0/10", "::1/128",
|
||||
"fc00::/7", "fe80::/10", "::ffff:0:0/96",
|
||||
"169.254.0.0/16", "0.0.0.0/8", "100.64.0.0/10",
|
||||
"224.0.0.0/4", "240.0.0.0/4", "198.18.0.0/15",
|
||||
"::1/128", "fc00::/7", "fe80::/10", "::ffff:0:0/96",
|
||||
"::/128", "64:ff9b::/96",
|
||||
]]
|
||||
ALLOWED_PORTS = {80, 443}
|
||||
MAX_REDIRECTS = 5
|
||||
|
||||
@@ -77,6 +77,14 @@ class TestIsBlockedIp:
|
||||
def test_私网段全拒(self, ip):
|
||||
assert safe_fetch.is_blocked_ip(ip) is True
|
||||
|
||||
@pytest.mark.parametrize("ip", [
|
||||
"::", "224.0.0.1", "240.0.0.1", "255.255.255.255",
|
||||
"64:ff9b::1.2.3.4", "198.18.0.1",
|
||||
])
|
||||
def test_补充禁段全拒(self, ip):
|
||||
"""v6 未指定/组播/保留广播/NAT64/基准段——补丁钉位(::/128 是 v6 版 0.0.0.0)。"""
|
||||
assert safe_fetch.is_blocked_ip(ip) is True
|
||||
|
||||
@pytest.mark.parametrize("ip", ["8.8.8.8", "1.1.1.1", "47.254.176.10", "2606:4700::1"])
|
||||
def test_公网放行(self, ip):
|
||||
assert safe_fetch.is_blocked_ip(ip) is False
|
||||
@@ -117,6 +125,11 @@ class TestFetch:
|
||||
with pytest.raises(safe_fetch.FetchError):
|
||||
await safe_fetch.fetch("file:///etc/passwd")
|
||||
|
||||
async def test_ipv6未指定地址URL拒绝(self):
|
||||
"""http://[::]/ —— v6 版 0.0.0.0,URL 层必须拒(urlsplit 剥括号得 "::")。"""
|
||||
with pytest.raises(safe_fetch.FetchError):
|
||||
await safe_fetch.fetch("http://[::]/x")
|
||||
|
||||
async def test_逐跳重定向每跳重校验(self, monkeypatch):
|
||||
"""302 → Location 指向私网 IP → 必须以 FetchError 终结(安全关键路径)。"""
|
||||
served = []
|
||||
@@ -131,6 +144,31 @@ class TestFetch:
|
||||
# 首跳(公网)发生、第二跳(私网 Location)在 URL 层即被拦,未发起
|
||||
assert served == ["http://example.com/start"]
|
||||
|
||||
async def test_重定向协议切换拒绝(self, monkeypatch):
|
||||
"""302 → Location 换协议(ftp://)→ 第二跳 URL 层 FetchError。"""
|
||||
served = []
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
served.append(str(request.url))
|
||||
return httpx.Response(302, headers={"Location": "ftp://evil/"})
|
||||
|
||||
monkeypatch.setattr(safe_fetch, "_make_client", _mock_client_factory(handler))
|
||||
with pytest.raises(safe_fetch.FetchError):
|
||||
await safe_fetch.fetch("http://example.com/start")
|
||||
assert served == ["http://example.com/start"]
|
||||
|
||||
async def test_连接层私网解析拒绝_真实链路(self, monkeypatch):
|
||||
"""DNS rebinding 端到端钉位:URL 层查不出的私网目标(域名解析返回 10.x),
|
||||
经真实 _make_client 的连接层必须拦下——只打桩 getaddrinfo,不打桩 connect_tcp,
|
||||
证明校验先于任何连接发生。"""
|
||||
async def fake_getaddrinfo(host, port, **kwargs):
|
||||
return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.66.0.9", port))]
|
||||
|
||||
monkeypatch.setattr(safe_fetch.anyio, "getaddrinfo", fake_getaddrinfo)
|
||||
with pytest.raises(safe_fetch.FetchError) as exc_info:
|
||||
await safe_fetch.fetch("http://rebinding.example.com/x")
|
||||
assert isinstance(exc_info.value.__cause__, safe_fetch.BlockedAddressError)
|
||||
|
||||
async def test_重定向上限5跳(self, monkeypatch):
|
||||
""">5 跳 302 链 → FetchError。"""
|
||||
calls = []
|
||||
|
||||
Reference in New Issue
Block a user