DDoS 防护与 WAF 实战

# 当攻击流量过大时，通过 BGP 宣告黑洞路由丢弃流量
# RTBH (Remotely Triggered Black Hole)

# 1. 源站向上游 ISP 发送 BGP 更新
# 宣告被攻击的 /32 IP 指向 NULL0
ip route 203.0.113.10 255.255.255.255 null0 tag 666

# 2. 上游 ISP 通过 iBGP 传播
# 所有邻居将目的为 203.0.113.10 的流量丢弃

# 3. 干净流量通过隧道/VPN 恢复
# 仅受信任的流量通过 GRE 隧道到达源站

# ModSecurity + OWASP CRS (Nginx)
modsecurity on;
modsecurity_rules_file /etc/nginx/modsecurity/modsecurity.conf;

# 自定义规则：防止 SQL 注入
SecRule REQUEST_COOKIES|REQUEST_COOKIES_NAMES|REQUEST_FILENAME|ARGS_NAMES|ARGS|XML:/* \
    "@rx (?i:(?:select\s*\*\s*from|(?:delete|drop|truncate)\s+table|union\s+select.*from|into\s+(?:outfile|dumpfile)))" \
    "id:1000001,phase:2,deny,status:403,msg:'SQL Injection Detected',logdata:'Matched Data: %{TX.0} found within %{MATCHED_VAR_NAME}: %{MATCHED_VAR}'"

# 自定义规则：速率限制
SecAction "id:1000002,phase:1,nolog,pass,setvar:ip.request_count=+1,expirevar:ip.request_count=60"
SecRule IP:REQUEST_COUNT "@gt 100" \
    "id:1000003,phase:1,deny,status:429,msg:'Rate Limit Exceeded'"

# Cloudflare 自定义防火墙规则 (Expression Language)

# 规则 1：阻止已知恶意 IP
(http.host eq "api.example.com" and ip.src in $blacklist)
or
(http.host eq "api.example.com" and ip.geoip.country in {"CN" "RU"})
→ Block

# 规则 2：API 速率限制
(http.host eq "api.example.com" and http.request.uri.path contains "/v1/")
→ Rate Limit: 100 requests / 10 seconds / IP
   Action: Block for 1 hour

# 规则 3：Bot 管理挑战
(http.host eq "api.example.com" and cf.bot_management.score lt 30)
→ Managed Challenge

# 规则 4：防止扫描工具
(http.user_agent contains "sqlmap" or http.user_agent contains "nikto" or http.user_agent contains "nmap")
→ Block

// Cloudflare Turnstile（无感知挑战）
// 替代传统 CAPTCHA，用户无感知

// 前端集成
<script src="https://challenges.cloudflare.com/turnstile/v0/api.js" async defer></script>
<div class="cf-turnstile" data-sitekey="your-site-key"></div>

// 后端验证
const verifyTurnstile = async (token) => {
  const response = await fetch('https://challenges.cloudflare.com/turnstile/v0/siteverify', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      secret: TURNSTILE_SECRET_KEY,
      response: token,
    }),
  });
  return response.json(); // { success: true, score: 0.9 }
};

import time
from threading import Lock

class TokenBucket:
    def __init__(self, capacity: int, refill_rate: float):
        self.capacity = capacity      # 桶容量
        self.tokens = capacity        # 当前令牌数
        self.refill_rate = refill_rate  # 每秒补充令牌数
        self.last_refill = time.time()
        self.lock = Lock()
    
    def allow_request(self, tokens_needed: int = 1) -> bool:
        with self.lock:
            now = time.time()
            # 补充令牌
            elapsed = now - self.last_refill
            self.tokens = min(
                self.capacity,
                self.tokens + elapsed * self.refill_rate
            )
            self.last_refill = now
            
            # 检查是否有足够令牌
            if self.tokens >= tokens_needed:
                self.tokens -= tokens_needed
                return True
            return False

# 使用：100 tokens/minute
bucket = TokenBucket(capacity=100, refill_rate=100/60)
if not bucket.allow_request():
    return 429  # Too Many Requests

import redis
import time

class RedisRateLimiter:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
    
    def sliding_window_check(self, key: str, window: int, limit: int) -> bool:
        """
        滑动窗口限流
        key: 限流标识（如 user:123:api:create）
        window: 窗口大小（秒）
        limit: 窗口内最大请求数
        """
        now = int(time.time())
        window_start = now - window
        
        pipe = self.redis.pipeline()
        
        # 移除窗口外的记录
        pipe.zremrangebyscore(key, 0, window_start)
        
        # 添加当前请求
        pipe.zadd(key, {str(now): now})
        
        # 统计窗口内请求数
        pipe.zcard(key)
        
        # 设置过期时间
        pipe.expire(key, window)
        
        results = pipe.execute()
        current_count = results[2]
        
        return current_count <= limit

# 使用
limiter = RedisRateLimiter(redis_client)
if not limiter.sliding_window_check(f"user:{user_id}:api", 60, 100):
    raise RateLimitExceeded("Too many requests")