mirror of
https://github.com/zhayujie/chatgpt-on-wechat.git
synced 2026-07-17 19:21:26 +08:00
ea47f3097e
web_fetch fetched any http/https URL a model emitted, checking only the scheme. It performed requests.get(..., allow_redirects=True) with no hostname resolution check and no private/loopback/link-local/cloud-metadata filtering, and never re-validated redirect targets. A model (including one under prompt injection) could make CowAgent fetch 127.0.0.1, RFC1918, 169.254.169.254 or other internal endpoints and return their bodies into the conversation; a public URL could also 302-bounce into a private target. The repo already shipped an SSRF validator for the vision tool (Vision._validate_url_safe). Extract that logic into a shared helper (agent/tools/utils/url_safety.py) and reuse it: - execute() now validates the URL before dispatching to either fetch path. - A new _safe_get() helper disables auto-redirect and follows redirects manually, re-validating every hop so a public URL cannot bounce into an internal address. Both the webpage and document fetch paths use it. - Vision._validate_url_safe now delegates to the shared helper (public API unchanged), so both URL-consuming tools share one guard. Stdlib only (ipaddress, socket, urllib.parse); no new dependency. Adds tests/test_security_ssrf_web_fetch.py covering loopback, cloud-metadata, RFC1918 and a public->loopback redirect. Sink: agent/tools/web_fetch/web_fetch.py (_fetch_webpage / _fetch_document). Signed-off-by: christop <825583681@qq.com>
67 lines
2.1 KiB
Python
67 lines
2.1 KiB
Python
"""
|
|
Shared SSRF guard utilities for tools that fetch model-supplied URLs.
|
|
|
|
A URL is only considered safe when it uses an http/https scheme, has a
|
|
hostname, that hostname resolves, and every resolved address is a public
|
|
(internet-routable) address. Loopback, private (RFC1918 / ULA), link-local
|
|
(incl. the 169.254.169.254 cloud-metadata endpoint) and otherwise reserved
|
|
addresses are rejected, for both IPv4 and IPv6.
|
|
"""
|
|
|
|
import ipaddress
|
|
import socket
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
def _is_blocked_ip(ip: "ipaddress._BaseAddress") -> bool:
|
|
"""Return True if the address is not safe to connect to (non-public)."""
|
|
return (
|
|
ip.is_private
|
|
or ip.is_loopback
|
|
or ip.is_link_local
|
|
or ip.is_reserved
|
|
or ip.is_multicast
|
|
or ip.is_unspecified
|
|
)
|
|
|
|
|
|
def assert_public_ip(ip_str: str) -> None:
|
|
"""Raise ValueError if the given literal IP is a non-public address.
|
|
|
|
Used to re-validate the concrete address a redirect resolved to.
|
|
"""
|
|
ip = ipaddress.ip_address(ip_str)
|
|
if _is_blocked_ip(ip):
|
|
raise ValueError(
|
|
f"URL resolves to a non-public address ({ip_str}), "
|
|
f"request blocked for security"
|
|
)
|
|
|
|
|
|
def validate_url_safe(url: str) -> None:
|
|
"""Reject URLs that target private/loopback/link-local addresses (SSRF guard).
|
|
|
|
Resolves the hostname to its IP address(es) and blocks any that fall
|
|
into non-public ranges. Also rejects URLs with no host, non-HTTP(S)
|
|
schemes, or hosts that fail DNS resolution.
|
|
|
|
Raises:
|
|
ValueError: if the URL targets a disallowed address.
|
|
"""
|
|
parsed = urlparse(url)
|
|
if parsed.scheme not in ("http", "https"):
|
|
raise ValueError(f"Unsupported URL scheme: {parsed.scheme}")
|
|
|
|
hostname = parsed.hostname
|
|
if not hostname:
|
|
raise ValueError("URL has no hostname")
|
|
|
|
try:
|
|
# Resolve all addresses for the hostname.
|
|
addr_infos = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
|
|
except socket.gaierror:
|
|
raise ValueError(f"Cannot resolve hostname: {hostname}")
|
|
|
|
for family, _, _, _, sockaddr in addr_infos:
|
|
assert_public_ip(sockaddr[0])
|