Python URL 抓取前的 SSRF 防护与 DNS 重绑定规避
问题场景
服务端提供“抓取 URL 预览”或“回调地址校验”时,用户可控制 URL。若直接把 URL 交给 requests/urllib,攻击者可构造指向 127.0.0.1、169.254.169.254、内网网段的地址,读取本机或云元数据服务。仅用正则判断字符串不够:域名可解析到私网地址,也可能在第一次解析返回公网、第二次连接时返回私网,即 DNS 重绑定。更稳妥的做法是:解析域名得到全部 A/AAAA 记录,逐条用 ipaddress 校验,拒绝私有、回环、链路本地、保留等地址;实际连接时固定使用已校验的 IP,不再让底层客户端重新解析域名。
最小实现
下面代码只依赖标准库,演示 HTTP/HTTPS 的单次 GET 与重定向逐跳校验。为了突出 SSRF 防护,未实现 gzip、chunked 解码和连接池;生产环境应使用成熟客户端并配合出口防火墙。
import ipaddress
import socket
import ssl
from urllib.parse import urlsplit, urljoin
def resolve_and_check(host, port):
infos = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
checked = []
for family, _, _, _, sockaddr in infos:
raw = sockaddr[0].split('%')[0]
ip = ipaddress.ip_address(raw)
if ip.version == 6 and ip.ipv4_mapped:
ip = ip.ipv4_mapped
if (not ip.is_global or ip.is_private or ip.is_loopback or
ip.is_link_local or ip.is_reserved or ip.is_multicast or
ip.is_unspecified):
raise ValueError('blocked address: %s' % ip)
checked.append(str(ip))
if not checked:
raise ValueError('no address for %s' % host)
return checked
def connect_fixed(host, port, ip, timeout):
sock = socket.create_connection((ip, port), timeout=timeout)
if port == 443:
ctx = ssl.create_default_context()
sock = ctx.wrap_socket(sock, server_hostname=host)
return sock
def http_get(host, port, path, ip, timeout=5):
sock = connect_fixed(host, port, ip, timeout)
req = ('GET %s HTTP/1.1\r\nHost: %s\r\nConnection: close\r\n'
'User-Agent: safe-fetch\r\n\r\n') % (path, host)
sock.sendall(req.encode('ascii'))
data = b''
while True:
chunk = sock.recv(65536)
if not chunk:
break
data += chunk
sock.close()
head, _, body = data.partition(b'\r\n\r\n')
lines = head.split(b'\r\n')
status = int(lines[0].split()[1])
headers = {}
for line in lines[1:]:
key, _, value = line.partition(b':')
headers[key.decode('latin1').lower()] = value.strip().decode('latin1')
return status, headers, body
def fetch(url, max_redirects=3, timeout=5):
for _ in range(max_redirects + 1):
parts = urlsplit(url)
if parts.scheme not in ('http', 'https'):
raise ValueError('scheme not allowed')
host = parts.hostname
if not host:
raise ValueError('missing host')
port = parts.port or (443 if parts.scheme == 'https' else 80)
if port not in (80, 443):
raise ValueError('port not allowed')
ips = resolve_and_check(host, port)
ip = ips[0]
path = parts.path or '/'
if parts.query:
path += '?' + parts.query
status, headers, body = http_get(host, port, path, ip, timeout)
if status in (301, 302, 303, 307, 308) and 'location' in headers:
url = urljoin(url, headers['location'])
continue
return status, headers, body
raise ValueError('too many redirects')
if __name__ == '__main__':
status, headers, body = fetch('http://example.com/')
print(status, headers.get('content-type'), len(body))
运行方法:保存为 safe_fetch.py,执行 python safe_fetch.py。将示例 URL 换成测试地址;在隔离环境可把 example.com 指向本地测试服务,验证 127.0.0.1 被拒绝。需要 Python 3.8+(ipaddress 的 ipv4_mapped 等属性可用)。注意 is_global 在不同 Python 版本对特殊地址段判断可能有差异,生产可显式补充网段列表并做单元测试。
为什么固定 IP 能规避 DNS 重绑定
resolve_and_check 返回全部解析结果并全部校验。http_get 用 socket.create_connection((ip, port)) 直接连 IP,不再把 host 交给解析器;HTTPS 通过 server_hostname=host 保留 SNI 和证书校验的主机名。这样校验的地址与实际连接的地址一致,攻击者无法在两次解析之间切换。重定向时每一跳都重新执行 fetch 的解析、校验和固定连接,避免“首跳公网、次跳内网”。
常见绕过用例
http://127.0.0.1/、http://[::1]/:hostname 得到回环地址,被拒。http://169.254.169.254/latest/meta-data/:链路本地地址,被拒。http://2130706433/、http://0x7f000001/:解析为 127.0.0.1 后按 IP 拒绝;不要只做字符串匹配。http://evil.example@127.0.0.1/:urlsplit 的 hostname 是 127.0.0.1,用户信息不会改变连接目标。http://example.com./或域名解析到内网:getaddrinfo 返回私网地址,被拒。- 重定向到
http://127.0.0.1/:逐跳校验会拒绝第二跳。 - IPv4-mapped IPv6:代码将
::ffff:127.0.0.1转成 IPv4 再判断。 - 非常用端口:代码限制 80/443,减少访问内网管理端口的风险。
边界与限制
代码是教学用最小实现。DNS 解析结果可能受本地 hosts、缓存和解析器影响;校验通过只表示当前解析到的地址在公网,不能替代网络层 egress 控制。HTTP 解析未处理 chunked、gzip、超时总时长、响应体大小限制和 TLS 证书链细节;生产应使用成熟 HTTP 客户端并设置总超时、最大响应体和重定向上限。固定 IP 连接对依赖虚拟主机和 SNI 的站点需要正确设置 Host 与 server_hostname。对于需要访问公网但 DNS 返回 CDN 多 IP 的场景,固定首个 IP 可能影响可用性,可在全部校验通过后按失败重试其他已校验 IP。最后,SSRF 防护应叠加云防火墙、元数据服务鉴权、禁止访问管理网段等纵深措施。
- 点赞
- 收藏
- 关注作者
评论(0)