AI API DDoS-Schutz und Rate-Limiting-Architekturdesign: Ein Komplett-Leitfaden für 2026

IP-Blacklisting und Anomalie-Erkennung

Basierend auf unserer HolySheep-Infrastruktur mit <50ms Latenz zeige ich Ihnen das DDoS-Erkennungssystem, das wir produktiv einsetzen:

"""
HolySheep AI - DDoS Detection und IP Blacklisting System
Verwendet sliding window averaging mit spike detection
"""
import asyncio
from collections import defaultdict, deque
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
import hashlib
import json

@dataclass
class IPMetrics:
    requests: deque = field(default_factory=deque)
    errors: deque = field(default_factory=deque)
    tokens: deque = field(default_factory=deque)
    first_seen: datetime = field(default_factory=datetime.now)
    blocked: bool = False
    block_reason: Optional[str] = None

class DDoSDetector:
    WINDOW_SIZE = 60  # Sekunden
    SPIKE_THRESHOLD = 5.0  # Faktor über Normalverbrauch
    ERROR_RATE_THRESHOLD = 0.3  # 30% Fehlerrate
    MIN_REQUESTS_FOR_ANALYSIS = 10
    
    def __init__(self):
        self.ip_data: Dict[str, IPMetrics] = defaultdict(IPMetrics)
        self.blacklist: Dict[str, datetime] = {}
        self.whitelist: set = {"127.0.0.1", "10.0.0.0/8"}  # Beispiel-Whitelist
    
    def record_request(self, ip: str, tokens: int, is_error: bool = False):
        now = datetime.now()
        cutoff = now - timedelta(seconds=self.WINDOW_SIZE)
        
        metrics = self.ip_data[ip]
        
        # Alte Einträge entfernen
        while metrics.requests and metrics.requests[0] < cutoff:
            metrics.requests.popleft()
            if metrics.errors:
                metrics.errors.popleft()
            if metrics.tokens:
                metrics.tokens.popleft()
        
        # Neue Metriken hinzufügen
        metrics.requests.append(now)
        metrics.errors.append(1 if is_error else 0)
        metrics.tokens.append(tokens)
        
        # Anomalie prüfen
        if len(metrics.requests) >= self.MIN_REQUESTS_FOR_ANALYSIS:
            self._check_anomaly(ip)
    
    def _calculate_stats(self, metrics: IPMetrics) -> Tuple[float, float, float]:
        """Berechne Durchschnittswerte"""
        if not metrics.requests:
            return 0.0, 0.0, 0.0
        
        request_rate = len(metrics.requests) / self.WINDOW_SIZE
        avg_tokens = sum(metrics.tokens) / len(metrics.tokens)
        
        error_count = sum(metrics.errors)
        error_rate = error_count / len(metrics.errors) if metrics.errors else 0
        
        return request_rate, avg_tokens, error_rate
    
    def _check_anomaly(self, ip: str):
        """Prüfe auf DDoS-typische Muster"""
        if ip in self.blacklist or ip in self.whitelist:
            return
            
        metrics = self.ip_data[ip]
        req_rate, avg_tokens, error_rate = self._calculate_stats(metrics)
        
        # Muster 1: Plötzlicher Anstieg
        if len(metrics.requests) > 5:
            recent_window = deque(list(metrics.requests)[-10:])
            older_window = deque(list(metrics.requests)[:-10])
            
            if older_window and len(older_window) >= 5:
                recent_rate = len(recent_window) / 10
                older_rate = len(older_window) / max(1, (60 - 10))
                
                if recent_rate > older_rate * self.SPIKE_THRESHOLD:
                    self._block_ip(ip, "Sudden traffic spike detected")
                    return
        
        # Muster 2: Hohe Fehlerrate (mögliche Credential Stuffing)
        if error_rate > self.ERROR_RATE_THRESHOLD:
            self._block_ip(ip, f"High error rate: {error_rate:.1%}")
            return
        
        # Muster 3: Unusual token consumption
        if avg_tokens > 50000:  # Ungewöhnlich hoher Token-Verbrauch
            self._block_ip(ip, f"Abnormal token usage: {avg_tokens}")
    
    def _block_ip(self, ip: str, reason: str, duration: int = 3600):
        """IP temporär blockieren"""
        self.ip_data[ip].blocked = True
        self.ip_data[ip].block_reason = reason
        self.blacklist[ip] = datetime.now() + timedelta(seconds=duration)
        
        # Log für Security Team
        print(f"[SECURITY] IP {ip} blocked: {reason}")
    
    def is_allowed(self, ip: str) -> Tuple[bool, Optional[str]]:
        """Prüfe ob IP erlaubt ist"""
        if ip in self.whitelist:
            return True, None
        
        if ip in self.blacklist:
            if datetime.now() < self.blacklist[ip]:
                return False, "IP is temporarily blocked"
            else:
                del self.blacklist[ip]
                self.ip_data[ip].blocked = False
        
        metrics = self.ip_data.get(ip)
        if metrics and metrics.blocked:
            return False, metrics.block_reason
        
        return True, None
    
    def get_threat_level(self, ip: str) -> str:
        """Gebe Bedrohungsstufe zurück"""
        if ip in self.whitelist:
            return "whitelisted"
        if ip in self.blacklist or (self.ip_data.get(ip) or IPMetrics()).blocked:
            return "blocked"
        
        metrics = self.ip_data.get(ip)
        if not metrics or len(metrics.requests) < 5:
            return "unknown"
        
        req_rate, _, error_rate = self._calculate_stats(metrics)
        
        if req_rate > 100 or error_rate > 0.2:
            return "high"
        elif req_rate > 50 or error_rate > 0.1:
            return "medium"
        return "low"


Integration mit HolySheep API Gateway
class HolySheepGateway:
    def __init__(self):
        self.rate_limiter = HolySheepRateLimiter()
        self.ddos_detector = DDoSDetector()
        self.whitelist_ips = {"52.23.123.45", "54.87.234.12"}  # Kundenserver
    
    async def handle_request(self, ip: str, api_key: str, request_body: dict):
        # Schritt 1: DDoS-Prüfung
        allowed, reason = self.ddos_detector.is_allowed(ip)
        if not allowed:
            return {
                "error": "Access denied",
                "reason": reason,
                "status": 403
            }
        
        # Schritt 2: Rate-Limit prüfen
        tokens = request_body.get("max_tokens", 1000)
        allowed, headers = self.rate_limiter.check_rate_limit(api_key, tokens)
        
        if not allowed:
            self.ddos_detector.record_request(ip, tokens, is_error=True)
            return {
                "error": "Rate limit exceeded",
                "headers": headers,
                "status": 429
            }
        
        # Schritt 3: Request erlauben
        self.ddos_detector.record_request(ip, tokens)
        
        # Weiterleitung an Upstream...
        return {"status": "forwarding", "headers": headers}

# HolySheep AI - Nginx Rate Limiting Configuration
Optimiert für KI-API-Workloads

http {
    # Limit Zones definieren
    limit_req_zone $binary_remote_addr zone=api_global:10m rate=10r/s;
    limit_req_zone $http_authorization zone=api_key:10m rate=100r/s;
    limit_conn_zone $binary_remote_addr zone=addr:10m;
    
    # Token Bucket für Burst-Handling
    limit_req_zone $binary_remote_addr zone=burst:10m burst=50 nodelay;
    
    # Logging
    log_format rate_limit '$remote_addr - $remote_user [$time_local] '
                          '"$request" $status $body_bytes_sent '
                          '"$http_referer" "$http_user_agent" '
                          'limit_req_status: $limit_req_status';
    
    upstream holysheep_api {
        server api.holysheep.ai:443;
        keepalive 32;
        keepalive_requests 1000;
        keepalive_timeout 60s;
    }
    
    server {
        listen 443 ssl http2;
        server_name your-gateway.com;
        
        ssl_certificate /etc/ssl/certs/gateway.crt;
        ssl_certificate_key /etc/ssl/private/gateway.key;
        
        # Rate Limiting
        limit_req zone=api_global burst=20 nodelay;
        limit_req zone=burst burst=50 nodelay;
        
        # Connection Limiting
        limit_conn addr 10;
        
        location /v1/chat/completions {
            # API Key Extraktion aus Authorization Header
            set $api_key $http_authorization;
            
            # Premium-Kunden haben höhere Limits (via Header)
            if ($http_x_customer-tier = "enterprise") {
                limit_req zone=burst burst=200 nodelay;
            }
            
            # Proxy mit Connection Pooling
            proxy_pass https://holysheep_api/chat/completions;
            proxy_http_version 1.1;
            proxy_set_header Host api.holysheep.ai;
            proxy_set_header Authorization $http_authorization;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            
            # Timeouts (KI-APIs brauchen länger)
            proxy_connect_timeout 10s;
            proxy_send_timeout 300s;
            proxy_read_timeout 300s;
            
            # Caching deaktiviert für Chat-API
            proxy_buffering off;
            
            # Limit Status in Response Header
            add_header X-RateLimit-Status $limit_req_status;
        }
        
        location /v1/embeddings {
            # Embeddings haben andere Rate-Limits
            limit_req zone=burst burst=100 nodelay;
            
            proxy_pass https://holysheep_api/embeddings;
            proxy_http_version 1.1;
            proxy_set_header Host api.holysheep.ai;
            proxy_set_header Authorization $http_authorization;
        }
        
        # Health Check Endpunkt (keine Limitierung)
        location /health {
            access_log off;
            return 200 'OK';
            add_header Content-Type text/plain;
        }
        
        # Rate Limit Error Pages
        error_page 429 = @rate_limit_exceeded;
        location @rate_limit_exceeded {
            default_type application/json;
            return 429 '{"error": {"message": "Rate limit exceeded", "type": "rate_limit_error", "code": "rate_limit_exceeded"}}';
        }
    }
}

# Fehlerhafte Konfiguration (NICHT SO)
redis_client = redis.Redis(host='localhost', port=6379)  # Neue Verbindung pro Request

Korrekte Konfiguration
from redis import ConnectionPool

pool = ConnectionPool(
    host='localhost',
    port=6379,
    max_connections=100,
    decode_responses=True,
    socket_timeout=5,
    socket_connect_timeout=5,
    retry_on_timeout=True
)
redis_client = redis.Redis(connection_pool=pool)

Bessere Lösung: HolySheep nutzen (kein Redis-Management nötig)
base_url="https://api.holysheep.ai/v1"
Inkludiert bereits DDoS-Schutz und Rate-Limiting

# FEHLER: Non-atomare Operation (Race Condition möglich)
current = redis.get(f"counter:{key}")
current = int(current) + 1
redis.set(f"counter:{key}", current)

LÖSUNG: Lua Script für Atomarität
ATOMIC_INCREMENT_SCRIPT = """
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local current = tonumber(redis.call('GET', key) or 0)

if current >= limit then
    return 0  -- Limit erreicht
end

local new_val = redis.call('INCR', key)
if new_val == 1 then
    redis.call('EXPIRE', key, 60)
end
return 1  -- Erfolgreich
"""

def check_and_increment(redis_client, key, limit):
    return redis_client.eval(
        ATOMIC_INCREMENT_SCRIPT,
        1,
        key,
        limit
    )

# Lösung: TTL immer setzen + Cleanup Job
CLEANUP_SCRIPT = """
local pattern = ARGV[1]
local max_age = tonumber(ARGV[2])
local cursor = 0
local deleted = 0

repeat
    cursor, keys = redis.call('SCAN', cursor, 'MATCH', pattern, 'COUNT', 100)
    for _, key in ipairs(keys) do
        local ttl = redis.call('TTL', key)
        if ttl == -1 then
            redis.call('EXPIRE', key, max_age)
            deleted = deleted + 1
        end
    end
until cursor == 0

return deleted
"""

Regelmäßig ausführen (z.B. stündlich via Cron)
def cleanup_old_keys(redis_client, pattern="ratelimit:*", max_age_seconds=3600):
    deleted = redis_client.eval(
        CLEANUP_SCRIPT,
        0,
        pattern,
        max_age_seconds
    )
    print(f"Cleaned up {deleted} orphaned keys")
    return deleted

Alternative: HolySheep verwenden - automatische Key-Rotation inklusive