xinyouhe/ai-kefu-admin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(site-import): SSRF 连接层校验 fetch 原语

Browse Source 
- ValidatingBackend：connect_tcp 处 anyio.getaddrinfo 解析→禁段校验→
  只连已验 IP 字面量（anyio 对 IP 字面量跳过 DNS，闭合 rebinding TOCTOU；
  TLS/SNI 由 httpcore 以原 hostname 包，证书校验不受影响）
- 双层防线：fetch URL 层预检（scheme/host/端口白名单+IP 字面量禁段）
  + 连接层逐跳 DNS 后校验；手动重定向 ≤5 跳每跳重校验
- 私有口（_pool 替换/AnyIOStream/异常映射）由版本探针测试钉位
- tests/test_safe_fetch.py 29 条：禁段参数化/连接层拒绝/重定向私网拦截/
  上限边界/请求真实路由经校验后端（anyio 层打桩离线全链路）

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-10-20 18:17:19 +08:00
parent 0d6a82ba14
commit 9f556ebe95
3 changed files with 379 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/platform/site_import/__init__.py
View File
app/platform/site_import/safe_fetch.py
+192
View File
@@ -0,0 +1,192 @@
"""site_import 安全抓取原语（spec §3.1）。
连接层 SSRF 校验：自定义 httpcore 后端在 connect_tcp 处做
anyio.getaddrinfo 解析（异步——单进程零同步 IO 纪律）→ IP 禁段校验 →
只连已验 IP 字面量。DNS rebinding 的 TOCTOU 由「解析与连接使用同一
后端的同一结果」闭合。禁止改写 URL host 为 IP（会破坏 TLS 证书校验）——
SNI/证书由 httpcore 以原 hostname 包 TLS 保持正确
（httpcore connection.py: start_tls(server_hostname=origin.host)）。
私有 API 依赖（httpx 0.28 / httpcore 1.0.9 钉位，tests/test_safe_fetch.py
版本探针守卫，升级破坏时立即红）：
- httpx.AsyncHTTPTransport._pool 属性替换注入校验连接池
- httpcore.AsyncConnectionPool(network_backend=...)
- httpcore._backends.anyio.AnyIOStream（anyio 流 → httpcore 流包装）
- httpcore._exceptions.ConnectError / ConnectTimeout / map_exceptions
"""
import ipaddress
import socket
import ssl
import urllib.parse
import anyio
import certifi
import httpcore
import httpx
from httpcore._backends.anyio import AnyIOStream
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",
]]
ALLOWED_PORTS = {80, 443}
MAX_REDIRECTS = 5
USER_AGENT = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
)
class BlockedAddressError(Exception):
"""目标解析到禁段（或端口非法）——连接前拒绝。"""
class FetchError(Exception):
"""抓取失败（URL 非法 / 端口禁 / 重定向超限 / 请求层错误）。"""
def is_blocked_ip(ip: str) -> bool:
"""禁段判定；非法 IP 串一律视为禁（fail-closed）。"""
try:
addr = ipaddress.ip_address(ip)
except ValueError:
return True
return any(addr in net for net in BLOCKED_NETS)
def _is_blocked_literal(hostname: str) -> bool:
"""URL 层的 host 禁段预检——只拒「IP 字面量落在禁段」的 host。
普通域名（非 IP 字面量）此处放行，交由连接层解析后按 is_blocked_ip
校验（is_blocked_ip 对非 IP 串 fail-closed 为 True，那是给解析结果
用的语义，不能套在域名上）。
"""
try:
addr = ipaddress.ip_address(hostname)
except ValueError:
return False
return is_blocked_ip(str(addr))
def _sockaddr_ip(raw) -> str:
"""getaddrinfo 的 sockaddr[0] 兼容 str / 打包 bytes 两种形态（IPv6 bytes）。"""
if isinstance(raw, (bytes, bytearray)):
return str(ipaddress.ip_address(bytes(raw)))
return str(raw)
class ValidatingBackend(httpcore.AsyncNetworkBackend):
"""connect_tcp 钩子：anyio.getaddrinfo 解析 → 逐候选禁段校验 → 只连已验 IP 字面量。
评审折入的两个实现陷阱：
1. anyio 不重导出 socket 常量——AF_UNSPEC/SOCK_STREAM 用 socket 模块；
2. 默认后端 anyio.connect_tcp(remote_host=host) 会对 hostname 内部重新做
DNS 解析——把 rebinding TOCTOU 重新打开。本实现只把「已验 IP 字面量」
递给连接调用（anyio 对 IP 字面量跳过 DNS 直连），原 hostname 不再进
任何会做 DNS 的路径。connect_unix_socket 不实现（基类默认拒绝）。
"""
async def connect_tcp(
self,
host: str,
port: int,
timeout: float | None = None,
local_address: str | None = None,
socket_options=None,
):
if port not in ALLOWED_PORTS:
raise BlockedAddressError(
f"port {port} not in allowed {sorted(ALLOWED_PORTS)}")
if socket_options is None:
socket_options = []
exc_map = {
TimeoutError: ConnectTimeout,
OSError: ConnectError,
anyio.BrokenResourceError: ConnectError,
}
with map_exceptions(exc_map):
infos = await anyio.getaddrinfo(
host, port, family=socket.AF_UNSPEC, type=socket.SOCK_STREAM)
validated: str | None = None
for _family, _type, _proto, _canon, sockaddr in infos:
ip = _sockaddr_ip(sockaddr[0])
if is_blocked_ip(ip):
raise BlockedAddressError(
f"{host!r} resolves to blocked address {ip}")
if validated is None:
validated = ip
if validated is None: # pragma: no cover - getaddrinfo 空结果防御
raise BlockedAddressError(f"{host!r} resolved to no addresses")
with anyio.fail_after(timeout):
stream = await anyio.connect_tcp(
remote_host=validated, remote_port=port, local_host=local_address)
for option in socket_options:
stream._raw_socket.setsockopt(*option) # type: ignore[attr-defined]
return AnyIOStream(stream)
async def sleep(self, seconds: float) -> None:
await anyio.sleep(seconds)
def _make_client(timeout: float = 10.0) -> httpx.AsyncClient:
"""校验后端注入的 client（私有口 _pool 替换——版本探针测试钉住）。"""
transport = httpx.AsyncHTTPTransport()
transport._pool = httpcore.AsyncConnectionPool(
ssl_context=ssl.create_default_context(cafile=certifi.where()),
network_backend=ValidatingBackend(),
)
return httpx.AsyncClient(
transport=transport,
timeout=timeout,
follow_redirects=False,
headers={"User-Agent": USER_AGENT},
)
def _validate_url(url: str) -> None:
"""URL 层预检：scheme 白名单 + host 禁段 + 端口白名单（连接层校验的前置快速失败）。"""
try:
parsed = urllib.parse.urlsplit(url)
port = parsed.port
except ValueError as exc:
raise FetchError(f"invalid URL: {url!r}") from exc
if parsed.scheme not in ("http", "https"):
raise FetchError(f"scheme not allowed: {parsed.scheme!r} ({url!r})")
if not parsed.hostname:
raise FetchError(f"URL without host: {url!r}")
if _is_blocked_literal(parsed.hostname):
raise FetchError(f"URL host in blocked range: {parsed.hostname}")
if port is None:
port = 443 if parsed.scheme == "https" else 80
if port not in ALLOWED_PORTS:
raise FetchError(f"port {port} not allowed: {url!r}")
async def fetch(url: str, timeout: float = 10.0) -> httpx.Response:
"""单页安全抓取：URL 预检 + 逐跳手动重定向（≤5 跳，每跳重过完整校验链）。
双层防线：本函数的 URL 层预检挡住字面量私网/非标端口；ValidatingBackend
在连接层对每一跳的 DNS 解析结果做禁段校验（防 rebinding）。网络层错误
统一包装为 FetchError（保留 __cause__）。
"""
_validate_url(url)
client = _make_client(timeout)
current = url
redirects = 0
try:
while True:
response = await client.get(current)
if not response.has_redirect_location:
return response
redirects += 1
if redirects > MAX_REDIRECTS:
raise FetchError(
f"too many redirects (>{MAX_REDIRECTS}) at {current!r}")
current = urllib.parse.urljoin(current, response.headers["location"])
_validate_url(current)
except (httpx.HTTPError, httpx.InvalidURL, BlockedAddressError, OSError) as exc:
raise FetchError(f"fetch failed for {url!r}: {exc}") from exc
finally:
await client.aclose()
tests/test_safe_fetch.py
+187
View File
@@ -0,0 +1,187 @@
"""safe_fetch：连接层 SSRF 校验（spec §3.1——防 DNS rebinding）。"""
import inspect
import socket
import httpx
import pytest
from app.platform.site_import import safe_fetch
def _mock_client_factory(handler):
"""redirect 类测试的打桩口：换掉 _make_client，注入 MockTransport client。"""
def factory(timeout: float = 10.0) -> httpx.AsyncClient:
return httpx.AsyncClient(
transport=httpx.MockTransport(handler), follow_redirects=False)
return factory
class _FakeRawSocket:
def setsockopt(self, *args, **kwargs):
pass
class _FakeAnyioStream:
"""anyio 流替身：吐一条完整 HTTP/1.1 200 响应（经 AnyIOStream 包装喂 httpcore）。"""
def __init__(self) -> None:
self._buf = (
b"HTTP/1.1 200 OK\r\n"
b"Content-Length: 2\r\n"
b"Connection: close\r\n"
b"\r\n"
b"ok"
)
self._raw_socket = _FakeRawSocket()
async def receive(self, max_bytes: int) -> bytes:
chunk, self._buf = self._buf[:max_bytes], self._buf[max_bytes:]
return chunk
async def send(self, item: bytes) -> None:
pass
async def aclose(self) -> None:
pass
def extra(self, *args, **kwargs):
return None
class TestVersionPin:
def test_httpx私有口存在(self):
"""httpx/httpcore 升级破坏私有 API 时立即红（spec 要求的版本钉位）——真断言，禁永真。"""
import httpcore
t = httpx.AsyncHTTPTransport()
assert hasattr(t, "_pool")
assert "network_backend" in inspect.signature(
httpcore.AsyncConnectionPool.__init__).parameters
def test_httpcore私有导出口存在(self):
"""safe_fetch 消费的 httpcore 私有符号（AnyIOStream 流包装 / 异常映射口）逐个在位。"""
from httpcore._backends.anyio import AnyIOStream
from httpcore._exceptions import ConnectError, ConnectTimeout, map_exceptions
assert callable(map_exceptions)
assert issubclass(ConnectTimeout, Exception)
assert issubclass(ConnectError, Exception)
class TestIsBlockedIp:
@pytest.mark.parametrize("ip", [
"127.0.0.1", "10.1.2.3", "172.16.0.1", "172.31.255.255",
"192.168.1.1", "169.254.1.1", "0.0.0.0", "100.64.0.1",
"::1", "fd00::1", "fe80::1", "::ffff:127.0.0.1", "::ffff:10.0.0.1",
])
def test_私网段全拒(self, ip):
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
def test_非法IP串视为拒(self):
assert safe_fetch.is_blocked_ip("not-an-ip") is True
class TestValidatingBackend:
async def test_connect_tcp_拒绝私网IP字面量(self):
backend = safe_fetch.ValidatingBackend()
with pytest.raises(safe_fetch.BlockedAddressError):
await backend.connect_tcp("127.0.0.1", 80)
with pytest.raises(safe_fetch.BlockedAddressError):
await backend.connect_tcp("10.0.0.1", 443)
# （连接层校验逻辑用 IP 字面量直接钉，不做真实域名 DNS 测试）
async def test_非法端口拒绝(self):
backend = safe_fetch.ValidatingBackend()
with pytest.raises(safe_fetch.BlockedAddressError):
await backend.connect_tcp("8.8.8.8", 8080)
class TestFetch:
async def test_非法端口拒绝(self):
with pytest.raises(safe_fetch.FetchError):
await safe_fetch.fetch("http://example.com:8080/")
async def test_私网URL拒绝(self):
with pytest.raises(safe_fetch.FetchError):
await safe_fetch.fetch("http://127.0.0.1/x")
with pytest.raises(safe_fetch.FetchError):
await safe_fetch.fetch("http://10.0.0.1/x")
async def test_非http协议拒绝(self):
with pytest.raises(safe_fetch.FetchError):
await safe_fetch.fetch("ftp://example.com/x")
with pytest.raises(safe_fetch.FetchError):
await safe_fetch.fetch("file:///etc/passwd")
async def test_逐跳重定向每跳重校验(self, monkeypatch):
"""302 → Location 指向私网 IP → 必须以 FetchError 终结（安全关键路径）。"""
served = []
def handler(request: httpx.Request) -> httpx.Response:
served.append(str(request.url))
return httpx.Response(302, headers={"Location": "http://10.0.0.2/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")
# 首跳（公网）发生、第二跳（私网 Location）在 URL 层即被拦，未发起
assert served == ["http://example.com/start"]
async def test_重定向上限5跳(self, monkeypatch):
""">5 跳 302 链 → FetchError。"""
calls = []
def handler(request: httpx.Request) -> httpx.Response:
calls.append(request.url.path)
return httpx.Response(302, headers={"Location": "/next"})
monkeypatch.setattr(safe_fetch, "_make_client", _mock_client_factory(handler))
with pytest.raises(safe_fetch.FetchError):
await safe_fetch.fetch("http://example.com/start")
# 首跳 + 5 次跟随 = 6 个请求；第 6 个 Location 不再跟
assert len(calls) == 6
async def test_五跳内重定向正常返回(self, monkeypatch):
"""恰好 5 跳 302 + 终点 200 → 正常返回（上限边界正向钉位）。"""
locations = iter(["/r1", "/r2", "/r3", "/r4", "/final"])
n_redirects = 0
def handler(request: httpx.Request) -> httpx.Response:
nonlocal n_redirects
loc = next(locations, None)
if loc is None:
return httpx.Response(200, text="ok")
n_redirects += 1
return httpx.Response(302, headers={"Location": loc})
monkeypatch.setattr(safe_fetch, "_make_client", _mock_client_factory(handler))
resp = await safe_fetch.fetch("http://example.com/start")
assert resp.status_code == 200
assert resp.text == "ok"
assert n_redirects == 5
async def test_请求真实走校验后端(self, monkeypatch):
"""_make_client 的请求确实路由经 ValidatingBackend（连接层钩子被调用）。"""
events = []
async def fake_getaddrinfo(host, port, **kwargs):
events.append(("resolve", host))
return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", port))]
async def fake_connect_tcp(remote_host, remote_port, **kwargs):
events.append(("connect", remote_host, remote_port))
return _FakeAnyioStream()
monkeypatch.setattr(safe_fetch.anyio, "getaddrinfo", fake_getaddrinfo)
monkeypatch.setattr(safe_fetch.anyio, "connect_tcp", fake_connect_tcp)
resp = await safe_fetch.fetch("http://example.com/")
assert resp.status_code == 200
assert resp.text == "ok"
assert ("resolve", "example.com") in events
# 连接目标是已验 IP 字面量而非原 hostname（DNS rebinding TOCTOU 闭合点）
assert ("connect", "93.184.216.34", 80) in events