2
月
在社区成长 2 月
LEVEL 1
1 XP
一、背景
服务端请求伪造（SSRF）常见于网页预览、Webhook 调试、图片代理、文档转换、远程文件导入等功能。应用接收用户提供的 URL，再由服务器发起请求。如果只检查字符串是否以 http 开头，攻击者就可能借助服务器的网络位置访问回环地址、内网管理接口或其他原本不可达的服务。
本文只在本机授权靶场中复现问题：使用 127.0.0.1:9000 模拟内部服务，使用 127.0.0.1:8000 模拟带 URL 抓取功能的业务服务。重点不是收集绕过载荷，而是建立一套可验证的防御流程：规范化解析、域名解析、IP 分类、重定向复检、出口限制、响应约束和审计告警。
二、漏洞原理与关键边界
SSRF 的本质是“用户影响了服务端请求的目标”。服务器通常拥有不同于外部用户的网络权限，因此一次普通 HTTP 请求可能跨越安全边界。
防御中容易忽略以下问题：
1. URL 文本不等于最终连接目标。域名经过 DNS 解析后可能指向私网或回环地址。
2. 只检查第一次请求不够。外部地址可能通过 30x 重定向跳转到受限地址。
3. 黑名单容易遗漏 IPv6、整数形式 IP、混合编码、大小写及解析器差异。
4. 校验与连接之间存在时间差。先解析再让 HTTP 库重新解析域名，可能引入 DNS 变化风险。
5. 应用层校验不是唯一边界。即使代码存在缺陷，网络出口策略也应阻止访问敏感网段。
6. 响应本身也可能造成风险，例如超大文件、无限流、慢速响应或压缩炸弹。
因此，应优先采用业务域名白名单；确实需要抓取任意公网 URL 时，再实施严格的 IP 判定和纵深限制。
三、前置条件
实验环境：
• Linux、macOS 或 WSL
• Python 3.10 以上
• 仅监听本机地址，不映射到公网
• 已获得当前设备和测试环境的授权
创建隔离目录与虚拟环境：
mkdir ssrf-lab && cd ssrf-lab
python3 -m venv .venv
source .venv/bin/activate
pip install flask requests
四、步骤一：搭建模拟内部服务
创建 internal.py：
from flask import Flask, jsonify
app = Flask(__name__)
@app.get("/internal/status")
def status():
return jsonify({"service": "mock-admin", "environment": "lab", "secret": "LAB_ONLY_TOKEN"})
app.run(host="127.0.0.1", port=9000)
启动服务：
python internal.py
另开终端验证：
curl http://127.0.0.1:9000/internal/status
这里的 LAB_ONLY_TOKEN 只是虚构测试数据，不应放入任何真实凭据。
五、步骤二：复现存在缺陷的 URL 抓取器
创建 vulnerable.py：
from flask import Flask, request, jsonify
import requests
app = Flask(__name__)
@app.get("/fetch")
def fetch():
target = request.args.get("url", "")
if not target.startswith(("http://", "https://")):
return jsonify({"error": "unsupported scheme"}), 400
response = requests.get(target, timeout=3)
return jsonify({"status": response.status_code, "body": response.text[:1000]})
app.run(host="127.0.0.1", port=8000)
启动：
python vulnerable.py
在本地靶场中请求：
curl --get --data-urlencode "url=http://127.0.0.1:9000/internal/status" http://127.0.0.1:8000/fetch
如果返回模拟内部服务内容，说明简单的协议检查没有建立任何网络边界。该验证只能用于本地或明确授权环境，不应替换为真实组织的内网地址。
六、步骤三：实现规范化解析与地址分类
创建 hardened.py。下面的实现拒绝 URL 用户信息、非 HTTP 协议、异常端口，以及解析到非全局地址的主机；同时关闭自动重定向，并对每一跳重新校验。
from flask import Flask, request, jsonify
from urllib.parse import urlsplit, urljoin
import socket
import ipaddress
import requests
app = Flask(__name__)
MAX_REDIRECTS = 3
MAX_BODY = 1024 * 1024
ALLOWED_PORTS = {80, 443}
class UnsafeTarget(Exception):
pass
def validate_url(raw_url):
try:
parsed = urlsplit(raw_url)
except ValueError as exc:
raise UnsafeTarget("invalid URL") from exc
if parsed.scheme.lower() not in {"http", "https"}:
raise UnsafeTarget("scheme is not allowed")
if not parsed.hostname:
raise UnsafeTarget("hostname is required")
if parsed.username is not None or parsed.password is not None:
raise UnsafeTarget("userinfo is not allowed")
try:
port = parsed.port or (443 if parsed.scheme.lower() == "https" else 80)
except ValueError as exc:
raise UnsafeTarget("invalid port") from exc
if port not in ALLOWED_PORTS:
raise UnsafeTarget("port is not allowed")
try:
records = socket.getaddrinfo(parsed.hostname, port, type=socket.SOCK_STREAM)
except socket.gaierror as exc:
raise UnsafeTarget("DNS resolution failed") from exc
addresses = {item[4][0] for item in records}
if not addresses:
raise UnsafeTarget("no address resolved")
for value in addresses:
ip = ipaddress.ip_address(value)
if not ip.is_global:
raise UnsafeTarget("non-global address is blocked")
return parsed.geturl(), addresses
def safe_fetch(raw_url):
current = raw_url
session = requests.Session()
session.trust_env = False
for hop in range(MAX_REDIRECTS + 1):
checked_url, resolved = validate_url(current)
response = session.get(
checked_url,
allow_redirects=False,
timeout=(2, 5),
stream=True,
headers={"User-Agent": "SafeFetcher-Lab/1.0"}
)
if response.is_redirect or response.is_permanent_redirect:
location = response.headers.get("Location")
response.close()
if not location:
raise UnsafeTarget("redirect without Location")
if hop >= MAX_REDIRECTS:
raise UnsafeTarget("too many redirects")
current = urljoin(checked_url, location)
continue
content_length = response.headers.get("Content-Length")
if content_length and int(content_length) > MAX_BODY:
response.close()
raise UnsafeTarget("response is too large")
body = bytearray()
for chunk in response.iter_content(16384):
body.extend(chunk)
if len(body) > MAX_BODY:
response.close()
raise UnsafeTarget("response exceeded limit")
result = {
"url": checked_url,
"resolved_addresses": sorted(resolved),
"status": response.status_code,
"content_type": response.headers.get("Content-Type", ""),
"body_preview": bytes(body[:1000]).decode("utf-8", errors="replace")
}
response.close()
return result
raise UnsafeTarget("request did not complete")
@app.get("/fetch")
def fetch():
target = request.args.get("url", "")
try:
return jsonify(safe_fetch(target))
except UnsafeTarget as exc:
return jsonify({"error": str(exc)}), 400
except requests.RequestException:
return jsonify({"error": "upstream request failed"}), 502
app.run(host="127.0.0.1", port=8001)
启动并验证本地地址被拒绝：
python hardened.py
curl --get --data-urlencode "url=http://127.0.0.1:9000/internal/status" http://127.0.0.1:8001/fetch
预期结果是 HTTP 400，并提示 non-global address is blocked。
七、步骤四：补充自动化测试
安装 pytest：
pip install pytest
创建 test_validation.py：
import pytest
nfrom hardened import validate_url, UnsafeTarget
@pytest.mark.parametrize("url", [
"http://127.0.0.1/",
"http://localhost/",
"http://[::1]/",
"file:///etc/passwd",
"ftp://example.com/file",
"http://user:[email protected]/"
])
def test_rejected_targets(url):
with pytest.raises(UnsafeTarget):
validate_url(url)
运行：
pytest -q
注意：上面示例中的第二行应为“from hardened import validate_url, UnsafeTarget”；如果直接复制文件，请去掉开头多余的字母 n。保留这个检查点也能提醒团队：安全代码必须经过实际执行和审查，不能只依赖文档表面正确。
建议继续加入以下测试用例：大小写混合协议、空主机、非法端口、IPv4 与 IPv6 回环地址、私网地址、链路本地地址、超长 URL、重定向到受限地址、响应超过大小限制、连接超时与读取超时。
八、步骤五：增加网络层出口限制
代码校验可能因依赖库升级、DNS 变化或解析器差异失效。生产环境应把抓取组件放入独立容器或网络命名空间，只允许访问必要的 DNS、代理和公网出口，显式拒绝回环、私网、链路本地、管理网段以及云环境敏感端点。
更稳妥的架构是：业务应用不直接访问任意 URL，而是把请求交给专用抓取代理；代理执行域名白名单、地址固定、重定向复检、内容类型限制、下载上限和统一审计。防火墙规则应结合实际网卡、容器网络、IPv6 状态和运维通道设计，避免直接复制一组未经验证的规则导致业务中断。
九、日志与检测建议
至少记录：请求发起用户、原始 URL 的脱敏摘要、规范化后的主机、解析结果、端口、重定向链、阻断原因、响应状态、字节数和耗时。不要记录 URL 中的完整令牌、Cookie 或响应正文。
可针对以下模式告警：短时间大量不同主机、连续命中受限地址、频繁 DNS 失败、重定向链异常、访问非常用端口、请求耗时持续接近超时阈值。日志只能作为检测手段，不能替代阻断。
十、风险与注意事项
1. 示例使用 ipaddress.is_global 作为保守基线，但生产策略应按业务定义允许范围，不能把某个库函数当作永久不变的安全规范。
2. “先解析再请求”仍可能存在校验结果与实际连接地址不一致的问题。高安全场景需要将已验证 IP 固定到连接过程，同时正确处理 HTTPS 的 SNI 与证书主机名校验，或统一经过受控代理。
3. 禁止自动继承 HTTP_PROXY、HTTPS_PROXY 等环境变量，除非代理本身是设计中的安全边界。
4. 应同时配置连接超时、读取超时、总任务时限、并发上限和响应大小上限。
5. 不要把上游错误、堆栈、DNS 细节或内部 IP 原样返回给普通用户。
6. 如果业务只需访问少量合作方，域名与端口白名单明显优于“允许整个公网”。
7. 靶场服务必须只监听本机或隔离网络，测试完成后应停止进程并删除虚构敏感数据。
十一、常见问题
问：禁止 127.0.0.1 是否就足够？
答：不够。还要考虑 IPv6 回环、私网、链路本地、保留地址、多个 DNS 结果和重定向后的新目标。
问：为什么不只用正则表达式检查 URL？
答：URL 语法复杂，不同组件可能采用不同解析规则。应使用成熟解析器拆分 scheme、hostname、port 等字段，再对解析结果实施策略。
问：允许 HTTPS 是否天然安全？
答：不是。HTTPS 保护传输过程，不决定目标是否属于内网，也不会自动消除重定向、超大响应和资源消耗风险。
问：返回内容截断到 1000 字符是否足够？
答：不够。必须在读取过程中限制总字节数。先完整下载再截断，仍会消耗带宽和内存。
问：最优先的改造是什么？
答：先确认业务是否真的需要任意 URL；能使用白名单就不要开放全公网。随后加入逐跳校验、响应限制、出口 ACL 和审计。
十二、总结
SSRF 防御不是增加一个字符串黑名单，而是管理“服务端代表用户发起网络请求”这一高风险能力。可靠方案应覆盖 URL 解析、DNS 结果、IP 分类、每次重定向、实际连接目标、响应资源限制和网络出口策略。通过本地靶场可以明确验证：协议前缀检查会失败，单次目标检查也不完整；只有应用层与网络层共同设限，并配合自动化回归测试和可审计日志，才能把 URL 抓取功能控制在清晰、可维护的安全边界内。
服务端请求伪造（SSRF）常见于网页预览、Webhook 调试、图片代理、文档转换、远程文件导入等功能。应用接收用户提供的 URL，再由服务器发起请求。如果只检查字符串是否以 http 开头，攻击者就可能借助服务器的网络位置访问回环地址、内网管理接口或其他原本不可达的服务。
本文只在本机授权靶场中复现问题：使用 127.0.0.1:9000 模拟内部服务，使用 127.0.0.1:8000 模拟带 URL 抓取功能的业务服务。重点不是收集绕过载荷，而是建立一套可验证的防御流程：规范化解析、域名解析、IP 分类、重定向复检、出口限制、响应约束和审计告警。
二、漏洞原理与关键边界
SSRF 的本质是“用户影响了服务端请求的目标”。服务器通常拥有不同于外部用户的网络权限，因此一次普通 HTTP 请求可能跨越安全边界。
防御中容易忽略以下问题：
1. URL 文本不等于最终连接目标。域名经过 DNS 解析后可能指向私网或回环地址。
2. 只检查第一次请求不够。外部地址可能通过 30x 重定向跳转到受限地址。
3. 黑名单容易遗漏 IPv6、整数形式 IP、混合编码、大小写及解析器差异。
4. 校验与连接之间存在时间差。先解析再让 HTTP 库重新解析域名，可能引入 DNS 变化风险。
5. 应用层校验不是唯一边界。即使代码存在缺陷，网络出口策略也应阻止访问敏感网段。
6. 响应本身也可能造成风险，例如超大文件、无限流、慢速响应或压缩炸弹。
因此，应优先采用业务域名白名单；确实需要抓取任意公网 URL 时，再实施严格的 IP 判定和纵深限制。
三、前置条件
实验环境：
• Linux、macOS 或 WSL
• Python 3.10 以上
• 仅监听本机地址，不映射到公网
• 已获得当前设备和测试环境的授权
创建隔离目录与虚拟环境：
mkdir ssrf-lab && cd ssrf-lab
python3 -m venv .venv
source .venv/bin/activate
pip install flask requests
四、步骤一：搭建模拟内部服务
创建 internal.py：
from flask import Flask, jsonify
app = Flask(__name__)
@app.get("/internal/status")
def status():
return jsonify({"service": "mock-admin", "environment": "lab", "secret": "LAB_ONLY_TOKEN"})
app.run(host="127.0.0.1", port=9000)
启动服务：
python internal.py
另开终端验证：
curl http://127.0.0.1:9000/internal/status
这里的 LAB_ONLY_TOKEN 只是虚构测试数据，不应放入任何真实凭据。
五、步骤二：复现存在缺陷的 URL 抓取器
创建 vulnerable.py：
from flask import Flask, request, jsonify
import requests
app = Flask(__name__)
@app.get("/fetch")
def fetch():
target = request.args.get("url", "")
if not target.startswith(("http://", "https://")):
return jsonify({"error": "unsupported scheme"}), 400
response = requests.get(target, timeout=3)
return jsonify({"status": response.status_code, "body": response.text[:1000]})
app.run(host="127.0.0.1", port=8000)
启动：
python vulnerable.py
在本地靶场中请求：
curl --get --data-urlencode "url=http://127.0.0.1:9000/internal/status" http://127.0.0.1:8000/fetch
如果返回模拟内部服务内容，说明简单的协议检查没有建立任何网络边界。该验证只能用于本地或明确授权环境，不应替换为真实组织的内网地址。
六、步骤三：实现规范化解析与地址分类
创建 hardened.py。下面的实现拒绝 URL 用户信息、非 HTTP 协议、异常端口，以及解析到非全局地址的主机；同时关闭自动重定向，并对每一跳重新校验。
from flask import Flask, request, jsonify
from urllib.parse import urlsplit, urljoin
import socket
import ipaddress
import requests
app = Flask(__name__)
MAX_REDIRECTS = 3
MAX_BODY = 1024 * 1024
ALLOWED_PORTS = {80, 443}
class UnsafeTarget(Exception):
pass
def validate_url(raw_url):
try:
parsed = urlsplit(raw_url)
except ValueError as exc:
raise UnsafeTarget("invalid URL") from exc
if parsed.scheme.lower() not in {"http", "https"}:
raise UnsafeTarget("scheme is not allowed")
if not parsed.hostname:
raise UnsafeTarget("hostname is required")
if parsed.username is not None or parsed.password is not None:
raise UnsafeTarget("userinfo is not allowed")
try:
port = parsed.port or (443 if parsed.scheme.lower() == "https" else 80)
except ValueError as exc:
raise UnsafeTarget("invalid port") from exc
if port not in ALLOWED_PORTS:
raise UnsafeTarget("port is not allowed")
try:
records = socket.getaddrinfo(parsed.hostname, port, type=socket.SOCK_STREAM)
except socket.gaierror as exc:
raise UnsafeTarget("DNS resolution failed") from exc
addresses = {item[4][0] for item in records}
if not addresses:
raise UnsafeTarget("no address resolved")
for value in addresses:
ip = ipaddress.ip_address(value)
if not ip.is_global:
raise UnsafeTarget("non-global address is blocked")
return parsed.geturl(), addresses
def safe_fetch(raw_url):
current = raw_url
session = requests.Session()
session.trust_env = False
for hop in range(MAX_REDIRECTS + 1):
checked_url, resolved = validate_url(current)
response = session.get(
checked_url,
allow_redirects=False,
timeout=(2, 5),
stream=True,
headers={"User-Agent": "SafeFetcher-Lab/1.0"}
)
if response.is_redirect or response.is_permanent_redirect:
location = response.headers.get("Location")
response.close()
if not location:
raise UnsafeTarget("redirect without Location")
if hop >= MAX_REDIRECTS:
raise UnsafeTarget("too many redirects")
current = urljoin(checked_url, location)
continue
content_length = response.headers.get("Content-Length")
if content_length and int(content_length) > MAX_BODY:
response.close()
raise UnsafeTarget("response is too large")
body = bytearray()
for chunk in response.iter_content(16384):
body.extend(chunk)
if len(body) > MAX_BODY:
response.close()
raise UnsafeTarget("response exceeded limit")
result = {
"url": checked_url,
"resolved_addresses": sorted(resolved),
"status": response.status_code,
"content_type": response.headers.get("Content-Type", ""),
"body_preview": bytes(body[:1000]).decode("utf-8", errors="replace")
}
response.close()
return result
raise UnsafeTarget("request did not complete")
@app.get("/fetch")
def fetch():
target = request.args.get("url", "")
try:
return jsonify(safe_fetch(target))
except UnsafeTarget as exc:
return jsonify({"error": str(exc)}), 400
except requests.RequestException:
return jsonify({"error": "upstream request failed"}), 502
app.run(host="127.0.0.1", port=8001)
启动并验证本地地址被拒绝：
python hardened.py
curl --get --data-urlencode "url=http://127.0.0.1:9000/internal/status" http://127.0.0.1:8001/fetch
预期结果是 HTTP 400，并提示 non-global address is blocked。
七、步骤四：补充自动化测试
安装 pytest：
pip install pytest
创建 test_validation.py：
import pytest
nfrom hardened import validate_url, UnsafeTarget
@pytest.mark.parametrize("url", [
"http://127.0.0.1/",
"http://localhost/",
"http://[::1]/",
"file:///etc/passwd",
"ftp://example.com/file",
"http://user:[email protected]/"
])
def test_rejected_targets(url):
with pytest.raises(UnsafeTarget):
validate_url(url)
运行：
pytest -q
注意：上面示例中的第二行应为“from hardened import validate_url, UnsafeTarget”；如果直接复制文件，请去掉开头多余的字母 n。保留这个检查点也能提醒团队：安全代码必须经过实际执行和审查，不能只依赖文档表面正确。
建议继续加入以下测试用例：大小写混合协议、空主机、非法端口、IPv4 与 IPv6 回环地址、私网地址、链路本地地址、超长 URL、重定向到受限地址、响应超过大小限制、连接超时与读取超时。
八、步骤五：增加网络层出口限制
代码校验可能因依赖库升级、DNS 变化或解析器差异失效。生产环境应把抓取组件放入独立容器或网络命名空间，只允许访问必要的 DNS、代理和公网出口，显式拒绝回环、私网、链路本地、管理网段以及云环境敏感端点。
更稳妥的架构是：业务应用不直接访问任意 URL，而是把请求交给专用抓取代理；代理执行域名白名单、地址固定、重定向复检、内容类型限制、下载上限和统一审计。防火墙规则应结合实际网卡、容器网络、IPv6 状态和运维通道设计，避免直接复制一组未经验证的规则导致业务中断。
九、日志与检测建议
至少记录：请求发起用户、原始 URL 的脱敏摘要、规范化后的主机、解析结果、端口、重定向链、阻断原因、响应状态、字节数和耗时。不要记录 URL 中的完整令牌、Cookie 或响应正文。
可针对以下模式告警：短时间大量不同主机、连续命中受限地址、频繁 DNS 失败、重定向链异常、访问非常用端口、请求耗时持续接近超时阈值。日志只能作为检测手段，不能替代阻断。
十、风险与注意事项
1. 示例使用 ipaddress.is_global 作为保守基线，但生产策略应按业务定义允许范围，不能把某个库函数当作永久不变的安全规范。
2. “先解析再请求”仍可能存在校验结果与实际连接地址不一致的问题。高安全场景需要将已验证 IP 固定到连接过程，同时正确处理 HTTPS 的 SNI 与证书主机名校验，或统一经过受控代理。
3. 禁止自动继承 HTTP_PROXY、HTTPS_PROXY 等环境变量，除非代理本身是设计中的安全边界。
4. 应同时配置连接超时、读取超时、总任务时限、并发上限和响应大小上限。
5. 不要把上游错误、堆栈、DNS 细节或内部 IP 原样返回给普通用户。
6. 如果业务只需访问少量合作方，域名与端口白名单明显优于“允许整个公网”。
7. 靶场服务必须只监听本机或隔离网络，测试完成后应停止进程并删除虚构敏感数据。
十一、常见问题
问：禁止 127.0.0.1 是否就足够？
答：不够。还要考虑 IPv6 回环、私网、链路本地、保留地址、多个 DNS 结果和重定向后的新目标。
问：为什么不只用正则表达式检查 URL？
答：URL 语法复杂，不同组件可能采用不同解析规则。应使用成熟解析器拆分 scheme、hostname、port 等字段，再对解析结果实施策略。
问：允许 HTTPS 是否天然安全？
答：不是。HTTPS 保护传输过程，不决定目标是否属于内网，也不会自动消除重定向、超大响应和资源消耗风险。
问：返回内容截断到 1000 字符是否足够？
答：不够。必须在读取过程中限制总字节数。先完整下载再截断，仍会消耗带宽和内存。
问：最优先的改造是什么？
答：先确认业务是否真的需要任意 URL；能使用白名单就不要开放全公网。随后加入逐跳校验、响应限制、出口 ACL 和审计。
十二、总结
SSRF 防御不是增加一个字符串黑名单，而是管理“服务端代表用户发起网络请求”这一高风险能力。可靠方案应覆盖 URL 解析、DNS 结果、IP 分类、每次重定向、实际连接目标、响应资源限制和网络出口策略。通过本地靶场可以明确验证：协议前缀检查会失败，单次目标检查也不完整；只有应用层与网络层共同设限，并配合自动化回归测试和可审计日志，才能把 URL 抓取功能控制在清晰、可维护的安全边界内。
