你的下载器可能是个 SSRF 跳板——视频下载工具的安全设计指南
你做了一个视频下载网站，用户粘贴一个链接，服务器去请求、解析、返回下载地址。功能很正常——直到有一天，有人在你的输入框里粘贴了
http://169.254.169.254/latest/meta-data/，你的服务器乖乖地去请求了 AWS 的元数据接口，然后把包含 AccessKey 的响应返回给了用户。这就是 SSRF（Server-Side Request Forgery），Web 下载工具最容易中的安全漏洞。这篇文章从实际攻击场景出发，讲清楚视频下载器面临的 5 类安全风险，以及每类风险的防御方案。TL;DR：下载器的核心风险：① SSRF（用户输入 URL → 服务器去请求内网地址）→ 用 URL 白名单 + 内网 IP 黑名单防御；② 任意文件读取（
file:///etc/passwd）→ 只允许 http/https scheme；③ 命令注入（URL 拼接到 shell 命令）→ 永远不要用 os.system 处理用户输入；④ 资源滥用（下载 100GB 文件耗尽磁盘）→ 限制文件大小和下载时长；⑤ Cookie 泄露（服务端存储的用户 Cookie 被日志/错误信息泄露）→ 敏感信息脱敏。
目录
- 一、SSRF：下载器最大的安全漏洞
- 二、命令注入：当 URL 被拼接到 shell 里
- 三、文件读取与路径穿越
- 四、资源滥用与 DoS
- 五、敏感信息泄露
- 六、一个安全的下载器架构
- 七、安全检查清单
- 八、合规与温馨提示
一、SSRF：下载器最大的安全漏洞
1.1 SSRF 攻击原理
正常流程：
用户 → 粘贴 https://www.bilibili.com/video/xxx
服务器 → 请求 https://www.bilibili.com/video/xxx
服务器 → 解析 → 返回视频地址
SSRF 攻击：
用户 → 粘贴 http://192.168.1.1/admin
服务器 → 请求 http://192.168.1.1/admin（内网地址！）
服务器 → 拿到路由器管理页面 → 返回给用户
更危险的例子：
攻击者输入：
http://169.254.169.254/latest/meta-data/iam/security-credentials/
服务器如果在 AWS 上，这个请求会返回临时 AccessKey。
攻击者拿到 AK 后可以完全控制你的 AWS 资源。
类似的内网探测地址：
阿里云：http://100.100.100.200/latest/meta-data/
腾讯云：http://metadata.tencentyun.com/latest/meta-data/
GCP：http://metadata.google.internal/computeMetadata/v1/
Docker：http://host.docker.internal:2375/containers/json
K8s：https://kubernetes.default.svc/api/v1/pods
本地：http://127.0.0.1:6379/（Redis）
本地：http://127.0.0.1:5432/（PostgreSQL）
1.2 防御方案
第一层：URL Scheme 白名单
from urllib.parse import urlparse
ALLOWED_SCHEMES = {'http', 'https'}
def validate_url_scheme(url):
parsed = urlparse(url)
if parsed.scheme not in ALLOWED_SCHEMES:
raise ValueError(f"不支持的协议: {parsed.scheme}")
# 阻止 file://、gopher://、dict:// 等危险协议
第二层：内网 IP 黑名单
import ipaddress
import socket
# 内网地址段
PRIVATE_NETWORKS = [
ipaddress.ip_network('10.0.0.0/8'),
ipaddress.ip_network('172.16.0.0/12'),
ipaddress.ip_network('192.168.0.0/16'),
ipaddress.ip_network('127.0.0.0/8'), # 本地回环
ipaddress.ip_network('169.254.0.0/16'), # 链路本地（AWS/阿里云元数据）
ipaddress.ip_network('0.0.0.0/8'), # 当前网络
ipaddress.ip_network('224.0.0.0/4'), # 组播
ipaddress.ip_network('240.0.0.0/4'), # 保留
]
# 特殊主机名
BLOCKED_HOSTS = {
'localhost',
'metadata.google.internal', # GCP
'metadata.tencentyun.com', # 腾讯云
'100.100.100.200', # 阿里云
}
def is_internal_host(hostname):
"""检查主机名是否指向内网"""
# 先检查主机名黑名单
if hostname.lower() in BLOCKED_HOSTS:
return True
# 解析 IP
try:
ip = ipaddress.ip_address(hostname)
except ValueError:
# 不是 IP 地址，是域名，需要 DNS 解析
try:
ip = ipaddress.ip_address(socket.gethostbyname(hostname))
except socket.gaierror:
raise ValueError(f"无法解析主机名: {hostname}")
# 检查是否属于内网地址段
for network in PRIVATE_NETWORKS:
if ip in network:
return True
return False
def validate_url(url):
"""完整的 URL 安全校验"""
parsed = urlparse(url)
# 1. Scheme 白名单
if parsed.scheme not in ALLOWED_SCHEMES:
raise ValueError(f"不支持的协议: {parsed.scheme}")
# 2. 必须有 hostname
if not parsed.hostname:
raise ValueError("URL 缺少主机名")
# 3. 阻止内网地址
if is_internal_host(parsed.hostname):
raise ValueError(f"禁止访问内网地址: {parsed.hostname}")
# 4. 阻止用户名密码（防止 http://user:pass@host 绕过）
if parsed.username or parsed.password:
raise ValueError("URL 不允许包含认证信息")
return True
第三层：DNS Rebinding 防御
即使 IP 检查通过了，攻击者还能用 DNS Rebinding 绕过：
攻击者的域名 evil.com 的 DNS 记录：
第一次解析 → 1.2.3.4（公网 IP，通过你的检查）
第二次解析 → 127.0.0.1（内网 IP，绕过检查）
防御方式：在整个请求生命周期中只做一次 DNS 解析，然后直接使用解析出的 IP：
import socket
from urllib.parse import urlparse
def resolve_once(url):
"""只解析一次 DNS，然后直接用 IP 建立连接"""
parsed = urlparse(url)
hostname = parsed.hostname
# 只解析一次
ip = socket.gethostbyname(hostname)
# 检查解析结果
if is_internal_host(ip):
raise ValueError(f"解析到内网地址: {ip}")
# 用 IP 替换主机名，后续不再解析
# （需要自己处理 SSL 证书验证）
return parsed._replace(netloc=parsed.netloc.replace(hostname, ip))
第四层：网络隔离（终极方案）
把下载器的请求逻辑放到一个独立的沙箱容器里，这个容器只能访问外网，完全无法访问内网：
# docker-compose.yml
services:
downloader:
build: .
networks:
- external_only # 只能访问外网
networks:
external_only:
driver: bridge
internal: false
# 通过 iptables 规则阻止访问内网
或者用 Kubernetes 的 NetworkPolicy：
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
name: deny-internal
spec:
podSelector:
matchLabels:
app: downloader
policyTypes:
- Egress
egress:
- to:
- ipBlock:
cidr: 0.0.0.0/0
except:
- 10.0.0.0/8
- 172.16.0.0/12
- 192.168.0.0/16
二、命令注入：当 URL 被拼接到 shell 里
2.1 危险代码示例
# ❌ 极其危险！永远不要这样做
import os
video_url = request.GET.get('url') # 用户输入
os.system(f"yt-dlp {video_url}") # 直接拼接到 shell 命令
# 攻击者输入：
# https://example.com/video.mp4; rm -rf /
# shell 实际执行：yt-dlp https://example.com/video.mp4; rm -rf /
2.2 正确的做法
# ✅ 使用 subprocess.run 的列表参数
import subprocess
video_url = validate_and_sanitize(request.GET.get('url'))
result = subprocess.run(
['yt-dlp', video_url], # 列表形式，不会触发 shell 解析
capture_output=True,
text=True,
timeout=300
)
subprocess.run 使用列表参数时，每个元素作为独立的参数传给程序，不会经过 shell 解析。;、|、$() 等 shell 特殊字符会作为普通字符串传递给 yt-dlp。
三、文件读取与路径穿越
3.1 攻击场景
# 用户输入的下载路径
output_path = request.GET.get('path') # 用户输入
# 攻击者输入：
# ../../../etc/passwd
# ../../root/.ssh/id_rsa
with open(f'/downloads/{output_path}', 'wb') as f:
f.write(video_data)
# → 写入 /etc/passwd，覆盖系统文件！
3.2 防御
import os
def safe_join_path(base_dir, user_path):
"""安全的路径拼接，防止路径穿越"""
# 解析为绝对路径
full_path = os.path.realpath(os.path.join(base_dir, user_path))
# 确保结果在 base_dir 之下
base_dir_real = os.path.realpath(base_dir)
if not full_path.startswith(base_dir_real + os.sep):
raise ValueError(f"路径穿越检测: {user_path}")
return full_path
# 使用
safe_path = safe_join_path('/downloads', user_filename)
with open(safe_path, 'wb') as f:
f.write(video_data)
四、资源滥用与 DoS
4.1 攻击场景
攻击 1：大文件耗尽磁盘
攻击者提交一个 100GB 的视频下载任务
→ 服务器磁盘写满 → 所有服务瘫痪
攻击 2：并发耗尽
攻击者同时提交 1000 个下载任务
→ CPU/内存/带宽耗尽 → 服务不可用
攻击 3：慢速攻击
攻击者提供一个极其慢的 URL（比如 1KB/s）
→ 一个连接占用一个 worker 很长时间
→ worker 池耗尽
4.2 防御
import time
class DownloadLimiter:
"""下载资源限制器"""
def __init__(self,
max_file_size=500 * 1024 * 1024, # 500MB
max_duration=600, # 10 分钟
max_concurrent=10):
self.max_file_size = max_file_size
self.max_duration = max_duration
self.semaphore = asyncio.Semaphore(max_concurrent)
self.total_downloaded = 0
self.start_time = time.time()
async def download(self, url, output_path):
async with self.semaphore:
downloaded = 0
start = time.time()
async with session.get(url) as resp:
# 检查 Content-Length
content_length = int(
resp.headers.get('content-length', 0)
)
if content_length > self.max_file_size:
raise ValueError(
f"文件过大: {content_length/1024/1024:.0f}MB"
)
with open(output_path, 'wb') as f:
async for chunk in resp.content.iter_chunked(8192):
downloaded += len(chunk)
# 大小检查
if downloaded > self.max_file_size:
raise ValueError("超过最大文件限制")
# 时长检查
if time.time() - start > self.max_duration:
raise TimeoutError("下载超时")
# 速度检查（太慢就放弃）
elapsed = time.time() - start
if elapsed > 30: # 给 30 秒缓冲
speed = downloaded / elapsed
if speed < 1024: # < 1KB/s
raise ConnectionError("下载速度过慢")
f.write(chunk)
return output_path
五、敏感信息泄露
5.1 常见的泄露途径
泄露途径 1：错误日志
用户提交了一个 URL，解析失败了
日志记录了完整的请求，包括 Cookie：
ERROR: Failed to parse https://example.com/video/123
Cookie: session_id=abc123; token=xyz789 ← 敏感信息！
泄露途径 2：调试输出
print(f"Downloading from {url} with headers {headers}")
→ headers 中包含 Cookie，输出到 stdout
泄露途径 3：异常堆栈
try:
parse(url)
except Exception as e:
return JsonResponse({'error': str(e)})
→ 异常消息可能包含 URL 和参数
5.2 防御
import re
import logging
class SensitiveDataFilter(logging.Filter):
"""过滤日志中的敏感信息"""
# 需要脱敏的字段
SENSITIVE_HEADERS = {
'cookie', 'authorization', 'x-api-key',
'set-cookie', 'x-csrf-token',
}
# 需要脱敏的 URL 参数
SENSITIVE_PARAMS = {
'token', 'sign', 'signature', 'access_key',
'secret', 'password', 'key',
}
def filter(self, record):
"""过滤日志记录"""
if hasattr(record, 'msg'):
msg = str(record.msg)
# 脱敏 URL 参数
msg = re.sub(
r'(token|sign|signature|key|secret|password)=[^&\s]+',
r'\1=***REDACTED***',
msg
)
# 脱敏 Cookie
msg = re.sub(
r'Cookie:\s*[^\n]+',
'Cookie: ***REDACTED***',
msg,
flags=re.IGNORECASE
)
record.msg = msg
return True
# 配置日志
logger = logging.getLogger('downloader')
logger.addFilter(SensitiveDataFilter())
六、一个安全的下载器架构
整合所有防御措施：
import asyncio
import hashlib
from urllib.parse import urlparse
from pathlib import Path
class SecureDownloader:
"""安全的视频下载器"""
def __init__(self, download_dir='/downloads',
max_size=500*1024*1024, max_duration=600):
self.download_dir = Path(download_dir)
self.max_size = max_size
self.max_duration = max_duration
self.limiter = DownloadLimiter(max_size, max_duration)
def validate(self, url):
"""入口校验——在任何网络请求之前"""
# 1. URL 格式检查
parsed = urlparse(url)
if parsed.scheme not in ('http', 'https'):
raise ValueError("仅支持 HTTP/HTTPS 协议")
if not parsed.hostname:
raise ValueError("无效的 URL")
# 2. 内网地址检查
if is_internal_host(parsed.hostname):
raise ValueError("不允许访问内网地址")
# 3. 域名白名单（可选，如果只支持特定平台）
# ALLOWED_DOMAINS = {'bilibili.com', 'youtube.com', ...}
# if not any(parsed.hostname.endswith(d) for d in ALLOWED_DOMAINS):
# raise ValueError("不支持的平台")
return parsed
def safe_filename(self, url, title='video'):
"""生成安全的文件名"""
# 用 URL 的哈希 + 标题生成文件名，避免特殊字符
url_hash = hashlib.md5(url.encode()).hexdigest()[:8]
safe_title = ''.join(
c for c in title if c.isalnum() or c in ' _-'
)[:100]
return f"{safe_title}_{url_hash}.mp4"
async def download(self, url, title='video'):
"""安全下载入口"""
# 校验
self.validate(url)
# 生成安全路径
filename = self.safe_filename(url, title)
output_path = self.download_dir / filename
output_path = safe_join_path(str(self.download_dir), filename)
# 带限流的下载
return await self.limiter.download(url, str(output_path))
七、安全检查清单
在发布你的下载工具之前，逐项检查：
[ ] URL Scheme 白名单（只允许 http/https）
[ ] 内网 IP 黑名单（阻止 10.x、172.16.x、192.168.x、127.x、169.254.x）
[ ] 阻止特殊主机名（localhost、metadata、100.100.100.200 等）
[ ] DNS Rebinding 防御（只解析一次 DNS）
[ ] 禁止 URL 中包含 @（防止 user:pass@host 绕过）
[ ] 使用 subprocess.run(list) 而非 os.system(str)
[ ] 路径穿越防御（os.path.realpath 检查）
[ ] 文件大小限制（Content-Length 检查 + 下载中累计检查）
[ ] 下载时长限制（超时中断）
[ ] 并发限制（Semaphore 控制）
[ ] 日志脱敏（Cookie/Token 自动打码）
[ ] 错误信息不返回内部细节给客户端
[ ] Cookie/Token 通过环境变量或安全存储读取
[ ] 定期更新依赖库（特别是 yt-dlp、requests）
八、合规与温馨提示
- 安全漏洞的发现和修复是软件开发的基本责任。如果你的下载工具有用户，你有义务保护用户数据安全
- 本文讨论的安全措施仅用于保护你自己的服务，请勿用于绕过他人的安全机制
- SSRF 攻击是严重的网络安全威胁，在未授权的情况下探测他人内网属于违法行为
- 更多讨论见 下载视频算侵权吗？聊聊个人备份与版权的那条线
做下载工具的人往往只关注"能不能下载"，忽略了"会不会被利用"。安全不是功能做完之后才考虑的事情——它应该从一开始就嵌入到架构里。上面这些检查项，花一个下午逐个过一遍，你的下载器就从"能用"变成了"能放心用"。
本文由 VidDown 技术博客原创发布。VidDown 的安全架构内置了上述所有防护措施——URL 白名单、内网隔离、资源限制、日志脱敏——确保你的下载安全可靠。访问 VidDown 了解更多。