diff --git a/app/platform/site_import/__init__.py b/app/platform/site_import/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/platform/site_import/safe_fetch.py b/app/platform/site_import/safe_fetch.py new file mode 100644 index 0000000..af4f3f8 --- /dev/null +++ b/app/platform/site_import/safe_fetch.py @@ -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() diff --git a/tests/test_safe_fetch.py b/tests/test_safe_fetch.py new file mode 100644 index 0000000..d1c5f29 --- /dev/null +++ b/tests/test_safe_fetch.py @@ -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