AI站点安全防护实战：如何抵御DDoS攻击

# ========== 1. 实时监控接口QPS（Shell脚本） ==========
#!/bin/bash
# 监控AI推理接口QPS，超过阈值告警
INTERFACE="/api/v1/chat/completions"
THRESHOLD=10000  # QPS阈值
LOG_FILE="/var/log/nginx/access.log"

# 实时统计QPS
while true; do
    QPS=$(tail -n 10000 $LOG_FILE | grep $INTERFACE | wc -l)
    if [ $QPS -gt $THRESHOLD ]; then
        # 触发告警（可对接企业微信/钉钉）
        curl -X POST "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的密钥" \
        -H "Content-Type: application/json" \
        -d '{
            "msgtype": "text",
            "text": {
                "content": "⚠️ AI接口QPS异常：当前QPS='${QPS}'，阈值='${THRESHOLD}'，疑似DDoS攻击"
            }
        }'
    fi
    sleep 5
done

# ========== 2. 识别异常IP（Python脚本） ==========
import pandas as pd
import numpy as np
from collections import Counter

# 读取Nginx日志
log_file = "/var/log/nginx/access.log"
df = pd.read_csv(log_file, sep=' ', header=None, 
                 names=['ip', '-', '-', 'time', 'request', 'status', 'size', 'referer', 'user_agent'])

# 统计每个IP的请求数
ip_counts = Counter(df['ip'])
# 识别异常IP（请求数超过平均值3倍）
mean_count = np.mean(list(ip_counts.values()))
abnormal_ips = [ip for ip, count in ip_counts.items() if count > mean_count * 3]

# 输出异常IP
print("异常IP列表：", abnormal_ips)
# 写入黑名单
with open("/etc/nginx/blacklist.conf", "a") as f:
    for ip in abnormal_ips:
        f.write(f"deny {ip};\n")

# 重载Nginx配置
import os
os.system("nginx -s reload")

# ========== 1. 隐藏源站IP（关键配置） ==========
# 1.1 开启CDN回源鉴权
# 配置回源密钥，仅允许CDN节点访问源站
# 阿里云CDN控制台：域名管理 → 回源配置 → 回源鉴权 → 开启
# 生成鉴权密钥，配置鉴权规则

# 1.2 源站仅允许CDN IP访问（Nginx配置）
# 获取CDN回源IP段：https://help.aliyun.com/document_detail/203096.html
# 在源站Nginx配置中添加：
location / {
    # 仅允许CDN回源IP访问
    allow 106.11.XX.XX/24;
    allow 123.58.XX.XX/24;
    deny all;
    
    # 其他配置
    proxy_pass http://127.0.0.1:8000;
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
}

# ========== 2. 开启CDN DDoS防护 ==========
# 2.1 登录阿里云CDN控制台，开启DDoS基础防护
# 路径：安全防护 → DDoS防护 → 开启
# 配置防护阈值：默认1Gbps，根据业务调整

# 2.2 开启CC防护
# 路径：安全防护 → CC防护 → 开启
# 配置规则：
# - 单IP QPS限制：100次/秒（根据业务调整）
# - 访问频率限制：5分钟内超过1000次触发拦截
# - 开启人机验证：对异常请求触发验证码

# ========== 3. 缓存静态资源，减少回源 ==========
# 3.1 配置缓存规则（Nginx）
location ~* \.(js|css|png|jpg|jpeg|gif|ico|svg)$ {
    expires 7d;  # 缓存7天
    add_header Cache-Control "public, max-age=604800";
    proxy_pass http://127.0.0.1:8000;
}

# 3.2 AI接口不缓存（关键）
location /api/v1/chat/completions {
    expires -1;  # 禁止缓存
    add_header Cache-Control "no-cache, no-store";
    proxy_pass http://127.0.0.1:8000;
}

# ========== 1. 高防IP配置 ==========
# 1.1 购买阿里云高防IP（按带宽计费）
# 推荐配置：保底带宽10Gbps，弹性带宽50Gbps
# 绑定AI站点域名，将流量引流到高防IP

# 1.2 配置清洗规则
# 登录阿里云高防IP控制台 → 防护配置 → 清洗策略
# 配置项：
# - 清洗阈值：默认2Gbps，攻击流量超过阈值自动清洗
# - 清洗模式：精准清洗（仅清洗攻击流量，保留合法流量）
# - 协议过滤：仅放行TCP/80、443端口，封禁其他端口

# ========== 2. 黑洞策略配置 ==========
# 2.1 配置黑洞触发阈值
# 路径：防护配置 → 黑洞策略
# - 触发阈值：10Gbps（攻击流量超过该值触发黑洞）
# - 黑洞时长：30分钟（可手动解除）
# - 白名单：添加核心业务IP，避免误黑洞

# 2.2 手动触发黑洞（应急脚本）
#!/bin/bash
# 手动触发黑洞，应对超大流量攻击
ALIYUN_ACCESS_KEY="你的AccessKey"
ALIYUN_SECRET="你的Secret"
IP="你的高防IP"
DURATION=30  # 黑洞时长（分钟）

# 调用阿里云API触发黑洞
curl -X POST "https://ddosbgp.aliyuncs.com/" \
-H "Content-Type: application/json" \
-d '{
    "Action": "BlackholeIp",
    "Ip": "'${IP}'",
    "Duration": '${DURATION}',
    "AccessKeyId": "'${ALIYUN_ACCESS_KEY}'",
    "Signature": "你的签名"
}'

# ========== 3. 流量清洗效果验证 ==========
# 查看清洗日志，验证攻击流量是否被清洗
# 路径：高防IP控制台 → 日志查询 → 清洗日志
# 关键指标：清洗流量、放行流量、攻击类型

# ========== 1. 自动收集恶意IP（Shell脚本） ==========
#!/bin/bash
# 每天更新恶意IP黑名单，集成多个威胁情报源

# 定义黑名单文件路径
BLACKLIST_FILE="/etc/nginx/blacklist.conf"
# 清空旧黑名单
echo "" > $BLACKLIST_FILE

# 从多个威胁情报源获取恶意IP
SOURCES=(
    "https://raw.githubusercontent.com/stamparm/ipsum/master/ipsum.txt"
    "https://lists.blocklist.de/lists/all.txt"
    "https://www.blocklist.de/downloads/ssh.txt"
)

# 下载并去重
for source in "${SOURCES[@]}"; do
    curl -s $source | grep -E '^([0-9]{1,3}\.){3}[0-9]{1,3}$' >> /tmp/ips.tmp
done

# 去重并添加到黑名单
sort -u /tmp/ips.tmp | while read ip; do
    echo "deny $ip;" >> $BLACKLIST_FILE
done

# 删除临时文件
rm -f /tmp/ips.tmp

# 重载Nginx配置
nginx -s reload

# ========== 2. Nginx集成黑名单 ==========
# 在nginx.conf中添加：
include /etc/nginx/blacklist.conf;

# ========== 3. 动态拦截异常IP（Python脚本） ==========
import requests
import json
import subprocess

# 从监控系统获取异常IP
def get_abnormal_ips():
    url = "http://monitor.example.com/api/abnormal_ips"
    response = requests.get(url)
    return response.json()["ips"]

# 添加IP到黑名单
def add_to_blacklist(ip):
    # 写入Nginx黑名单
    with open("/etc/nginx/blacklist.conf", "a") as f:
        f.write(f"deny {ip};\n")
    # 重载Nginx
    subprocess.run(["nginx", "-s", "reload"])
    # 添加到iptables
    subprocess.run(["iptables", "-A", "INPUT", "-s", ip, "-j", "DROP"])

# 主逻辑
if __name__ == "__main__":
    abnormal_ips = get_abnormal_ips()
    for ip in abnormal_ips:
        add_to_blacklist(ip)
    print(f"已拦截 {len(abnormal_ips)} 个异常IP")

# ========== 1. Nginx接口限流 ==========
# 在nginx.conf中配置限流
http {
    # 定义限流区域，基于IP限流
    limit_req_zone $binary_remote_addr zone=ai_api:10m rate=10r/s;
    
    server {
        # AI推理接口限流
        location /api/v1/chat/completions {
            # 限流配置：最多10个请求排队，超过返回503
            limit_req zone=ai_api burst=10 nodelay;
            
            # 并发连接数限制
            limit_conn_zone $binary_remote_addr zone=ai_conn:10m;
            limit_conn ai_conn 5;
            
            # 其他配置
            proxy_pass http://ai_server;
            proxy_set_header Host $host;
        }
    }
}

# ========== 2. AI接口鉴权配置 ==========
# Python FastAPI示例
from fastapi import FastAPI, Header, HTTPException, Depends
import jwt
import time

app = FastAPI()

# 接口密钥
SECRET_KEY = "你的接口密钥"

# 鉴权依赖
def verify_token(authorization: str = Header(None)):
    if not authorization:
        raise HTTPException(status_code=401, detail="未提供token")
    try:
        # 验证token
        token = authorization.split(" ")[1]
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        # 检查token过期
        if payload["exp"] < time.time():
            raise HTTPException(status_code=401, detail="token已过期")
        return payload["user_id"]
    except:
        raise HTTPException(status_code=401, detail="token无效")

# AI推理接口，需要鉴权
@app.post("/api/v1/chat/completions")
async def chat_completions(data: dict, user_id: str = Depends(verify_token)):
    # 按用户限流（每个用户每秒最多5次请求）
    redis_key = f"user_limit:{user_id}"
    current = redis.incr(redis_key)
    if current == 1:
        redis.expire(redis_key, 1)
    if current > 5:
        raise HTTPException(status_code=429, detail="请求频率过高，请稍后再试")
    
    # 处理推理请求
    return {"response": "AI推理结果"}

# ========== 3. 算力保护配置 ==========
# 限制单个请求的算力消耗
@app.post("/api/v1/chat/completions")
async def chat_completions(data: dict):
    # 限制输入文本长度
    if len(data.get("prompt", "")) > 1000:
        raise HTTPException(status_code=400, detail="输入文本过长")
    
    # 限制生成文本长度
    max_tokens = data.get("max_tokens", 500)
    if max_tokens > 1000:
        data["max_tokens"] = 1000
    
    # 处理推理请求
    return {"response": "AI推理结果"}

# ========== 1. Prometheus监控配置 ==========
# prometheus.yml
global:
  scrape_interval: 5s

scrape_configs:
  # 监控Nginx流量
  - job_name: 'nginx'
    static_configs:
      - targets: ['localhost:9113']  # nginx-exporter
    
  # 监控AI服务
  - job_name: 'ai_service'
    static_configs:
      - targets: ['ai_server:8000']
    
  # 监控服务器资源
  - job_name: 'node'
    static_configs:
      - targets: ['localhost:9100']  # node-exporter