跳到主要内容

基于下载限制的对象存储OSS ddos防护

目标:

  • 用户点击一个链接(如 https://yourdomain.com/download?file=xxx.pdf)
  • 直接弹出浏览器下载框(无需登录、无需操作)
  • 每个 IP 地址每分钟最多下载 6 个文件,每小时最多下载 10000 个文件
  • 所有文件存储在阿里云私有 OSS Bucket 中
  • 服务器环境:Fedora 系统 + Nginx

一、整体架构图(先看懂流程)​

用户浏览器
│
▼
点击链接:https://yourdomain.com/download?file=report.pdf
│
▼
Nginx(反向代理,传递真实IP)
│
▼
Python 服务(运行在 127.0.0.1:5000)
├─ 检查该IP本分钟是否已下载 ≥6 次?
│ ├─ 是 → 返回 429 错误
│ └─ 否 → 检查该IP本小时是否已下载 ≥10000 次?
│ ├─ 是 → 返回 429 错误
│ └─ 否 → 生成5分钟有效的OSS预签名URL
│
▼
302 重定向到:https://bucket.oss-cn-xxx.aliyuncs.com/...?OSSAccessKeyId=xxx&Expires=xxx&Signature=xxx
│
▼
浏览器直接下载文件(用户无感)

二、前提条件(请先确认)​

  1. 你有一个域名(如 example.com),已解析到你的 Fedora 服务器 IP
  2. 你已安装并配置好 Nginx(能正常访问网站)
  3. 你有一个阿里云账号
  4. 你的 OSS Bucket 是私有的(不是"公共读")

如果还没做第4点,请先登录 阿里云 OSS 控制台:

  • 进入你的 Bucket → 基础设置 → 读写权限 → 改为"私有"
  • 删除任何允许 *(所有人)读取的 Bucket Policy

三、步骤详解(按顺序操作)​


第1步:创建阿里云子账号(安全!不要用主账号)​

耗时:3分钟 地点:阿里云控制台(网页)

  1. 打开 RAM 访问控制
  2. 点击 "创建用户"
    • 用户名:oss-download-service
    • 访问方式:✅ 勾选"AccessKey访问"
  3. 点击 "确定",然后点 "添加权限"
    • 授权范围:整个云账号
    • 权限策略:搜索并勾选 AliyunOSSReadOnlyAccess
  4. 点击 "确定"
  5. 点击刚创建的用户 → "安全信息" → "查看" AccessKey
    • 复制 AccessKey ID 和 AccessKey Secret(后面要用)

这个子账号只能读取 OSS 文件,不能删、不能改,安全!


第2步:在 Fedora 服务器上安装依赖​

地点:终端(SSH 登录你的服务器)

# 1. 安装基础软件
sudo dnf install -y python3 redis nginx git

# 2. 启动 Redis(用于计数)
sudo systemctl enable --now redis

# 3. 安装 Python 依赖
pip3 install flask oss2 redis --user

如果提示 pip3 不存在,先运行:sudo dnf install -y python3-pip


第3步:创建下载服务代码​

文件位置:/opt/oss-download-guard/app.py

# 创建目录
sudo mkdir -p /opt/oss-download-guard
cd /opt/oss-download-guard

# 创建代码文件
sudo nano app.py

粘贴以下完整代码(⚠️ 你需要修改 4 处配置!):

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
OSS 下载守门员服务
功能:限制每个 IP 每分钟最多下载 6 个文件,每小时最多下载 10000 个文件
作者:为你定制
"""

from flask import Flask, request, redirect, abort
import redis
from oss2 import Auth, Bucket
import time
import logging
import os

# ==============================
# 配置区(请务必修改!)
# ==============================
ACCESS_KEY_ID = 'LTAI5tXXXXXXXXXXXXXX' # ← 改成你的子账号 AccessKey ID
ACCESS_KEY_SECRET = 'your-access-key-secret' # ← 改成你的子账号 AccessKey Secret
BUCKET_NAME = 'your-oss-bucket-name' # ← 改成你的 OSS Bucket 名称(如 my-files)
REGION = 'oss-cn-hangzhou' # ← 改成你的 OSS 区域(如 cn-beijing, cn-shanghai)
MAX_PER_MINUTE = 6 # 每分钟最多下载次数
MAX_PER_HOUR = 10000 # 每小时最多下载次数
ALLOWED_PREFIX = 'public/' # 只允许下载 public/ 目录下的文件(防越权)
# ==============================

# 初始化
app = Flask(__name__)
app.logger.setLevel(logging.INFO)

# 连接 Redis
r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True, socket_connect_timeout=2)

# 初始化 OSS
auth = Auth(ACCESS_KEY_ID, ACCESS_KEY_SECRET)
bucket = Bucket(auth, f'https://{REGION}.aliyuncs.com', BUCKET_NAME)

@app.route('/download')
def download():
try:
# 获取用户真实 IP(支持 Nginx 代理)
ip = request.headers.get('X-Real-IP') or request.headers.get('X-Forwarded-For') or request.remote_addr
file_key = request.args.get('file', '').strip()

# 校验参数
if not file_key:
app.logger.warning(f"[{ip}] Missing 'file' parameter")
abort(400, "Missing 'file' parameter")

# 安全校验:只允许下载指定前缀的文件(防止 ../ 路径穿越)
if not file_key.startswith(ALLOWED_PREFIX) or '..' in file_key or file_key.startswith('/'):
app.logger.warning(f"[{ip}] Forbidden path: {file_key}")
abort(403, "Access denied")

# 获取当前时间
current_time = int(time.time())
minute = current_time // 60
hour = current_time // 3600

# 构造限流键
minute_rate_key = f"rate_minute:{ip}:{minute}"
hour_rate_key = f"rate_hour:{ip}:{hour}"

# 检查分钟限制
minute_count = r.get(minute_rate_key)
if minute_count is not None and int(minute_count) >= MAX_PER_MINUTE:
app.logger.warning(f"[{ip}] Minute rate limited. Current: {minute_count}")
abort(429, "Too many downloads. Please wait 1 minute.")

# 检查小时限制
hour_count = r.get(hour_rate_key)
if hour_count is not None and int(hour_count) >= MAX_PER_HOUR:
app.logger.warning(f"[{ip}] Hour rate limited. Current: {hour_count}")
abort(429, "Too many downloads. Please wait until next hour.")

# 生成预签名 URL(有效期 300 秒 = 5 分钟)
signed_url = bucket.sign_url('GET', file_key, 300)
app.logger.info(f"[{ip}] Generated signed URL for {file_key}")

# 分钟计数 +1
r.incr(minute_rate_key)
r.expire(minute_rate_key, 120) # 2分钟后自动清理

# 小时计数 +1
r.incr(hour_rate_key)
r.expire(hour_rate_key, 7200) # 2小时后自动清理

# 302 重定向 → 浏览器直接下载!
return redirect(signed_url)

except Exception as e:
app.logger.error(f"Error: {str(e)}")
abort(500, "Internal server error")

# 健康检查(可选)
@app.route('/health')
def health():
return "OK", 200

if __name__ == '__main__':
app.run(host='127.0.0.1', port=5000, debug=False)

你需要修改的 4 处(在代码开头):

  1. ACCESS_KEY_ID → 你的子账号 AccessKey ID
  2. ACCESS_KEY_SECRET → 你的子账号 AccessKey Secret
  3. BUCKET_NAME → 你的 OSS Bucket 名字(如 my-company-files)
  4. REGION → 你的 OSS 区域(如 oss-cn-beijing)

安全提示:

  • ALLOWED_PREFIX = 'public/' 表示只允许下载 OSS 中 public/ 文件夹下的文件
  • 例如:用户只能下 public/report.pdf,不能下 private/secret.docx
  • 如果你想允许根目录,改成 ALLOWED_PREFIX = '',但强烈不建议

保存文件:Ctrl+O → 回车 → Ctrl+X


第4步:创建 systemd 服务(开机自启)​

sudo nano /etc/systemd/system/oss-download-guard.service

粘贴以下内容:

[Unit]
Description=OSS Download Guard Service
After=network.target redis.service

[Service]
Type=simple
User=root
WorkingDirectory=/opt/oss-download-guard
ExecStart=/usr/bin/python3 /opt/oss-download-guard/app.py
Restart=always
RestartSec=5
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

启用服务:

sudo systemctl daemon-reload
sudo systemctl enable --now oss-download-guard

检查是否运行:

sudo systemctl status oss-download-guard
# 应该看到 "active (running)"

第5步:配置 Nginx(反向代理 + 传递真实 IP)​

编辑 Nginx 配置(假设你已有网站配置):

sudo nano /etc/nginx/conf.d/your-site.conf

如果你没有自定义配置,可以编辑默认配置:/etc/nginx/nginx.conf 或 /etc/nginx/conf.d/default.conf

在 server { ... } 块中添加以下 location:

# OSS 下载守门员
location /download {
proxy_pass http://127.0.0.1:5000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}

重载 Nginx:

sudo nginx -t && sudo systemctl reload nginx

第6步:测试服务是否正常​

测试 1:健康检查​

curl http://127.0.0.1:5000/health
# 应该返回 "OK"

测试 2:模拟下载(替换为你的真实文件)​

curl -v "http://127.0.0.1:5000/download?file=public/test.pdf"
  • 如果文件存在 → 应该返回 302 Found + 一个带签名的 OSS 链接
  • 如果文件不存在 → 返回 404
  • 如果路径非法(如 file=../etc/passwd)→ 返回 403

测试 3:通过域名访问​

在浏览器打开:

https://yourdomain.com/download?file=public/test.pdf

✅ 如果配置正确,浏览器会直接开始下载 test.pdf!


四、如何使用?(给用户什么链接)​

假设:

  • 你的域名:https://files.example.com
  • OSS 中文件路径:public/reports/2024/q1.pdf

那么用户下载链接就是:

https://files.example.com/download?file=public/reports/2024/q1.pdf

用户点击 → 直接下载 同一 IP 每分钟最多点 6 次,每小时最多点 10000 次 超过限制会看到:Too many downloads. Please wait.


五、安全加固建议​

  1. 限制文件类型(可选) 在 app.py 中加一行:

    if not file_key.lower().endswith(('.pdf', '.zip', '.mp4', '.jpg')):
    abort(403, "File type not allowed")
  2. 隐藏真实错误信息 当前代码在出错时会返回详细信息(如 404)。上线后可统一返回 404:

    abort(404) # 替代具体的错误信息
  3. 定期轮换 AccessKey 每 3 个月在阿里云 RAM 中更换子账号密钥


六、常见问题排查​

问题解决方法
点击链接显示 404检查 OSS 文件路径是否正确(区分大小写!)
显示 403 AccessDenied检查 Bucket 是否为"私有",子账号是否有 oss:GetObject 权限
显示 500 Internal Error查看日志:sudo journalctl -u oss-download-guard -f
下载很慢OSS 本身很快,检查服务器带宽或用户网络
限流不生效检查 Nginx 是否传递了 X-Real-IP,Redis 是否运行

七、附:完整文件清单​

文件路径作用
下载服务代码/opt/oss-download-guard/app.py核心逻辑
systemd 服务/etc/systemd/system/oss-download-guard.service开机自启
Nginx 配置/etc/nginx/conf.d/your-site.conf反向代理
日志查看sudo journalctl -u oss-download-guard -f调试用

最终效果验证​

  1. 打开浏览器,访问 https://yourdomain.com/download?file=public/test.pdf → 直接下载
  2. 快速刷新 6 次 → 第 7 次显示分钟限流提示
  3. 等 1 分钟 → 又能下载 6 次
  4. 累计下载接近 10000 次时 → 显示小时限流提示
  5. 尝试访问 https://your-bucket.oss-cn-xxx.aliyuncs.com/public/test.pdf → 403 Forbidden

恭喜!你已成功实现安全、双重限流、直接下载的 OSS 文件分发系统!