Rate limiting and abuse prevention
Auf dieser Seite
Warum Anfragen begrenzen?
KI-Anwendungen sind aus einem Grund, der bei herkömmlichen Anwendungen nicht existierte, besonders anfällig für Missbrauch: Jeder Modellaufruf kostet Geld. Eine traditionelle REST-API, die einem Ansturm von Anfragen ausgesetzt ist, verlangsamt sich und stürzt schließlich ab, aber sie entleert kein Bankkonto. Eine LLM-Anwendung hingegen antwortet weiterhin perfekt, während die Rechnung in die Höhe schießt. Ohne Rate Limiting kann ein Angreifer Ihr API-Budget innerhalb weniger Minuten ruinieren, automatisierte Prompt-Injection-Angriffe starten oder Ihre Infrastruktur nutzen, um Spam auf Ihre Kosten zu generieren.
Zwei Möglichkeiten zur Zählung
Das gleitende Fenster ist der grundlegende Mechanismus. Es protokolliert den Zeitstempel jeder Anfrage, verwirft diejenigen, die das Fenster verlassen haben, und vergleicht den Rest mit dem Limit. Der Vorteil gegenüber dem festen Fenster liegt in einem Detail, das in der Praxis zählt: Bei einem festen Fenster, das jede Minute zurückgesetzt wird, kann ein Angreifer sein Kontingent am Ende einer Minute und erneut am Anfang der nächsten Minute platzieren, wodurch die erlaubte Durchsatzrate über wenige Sekunden effektiv verdoppelt wird. Das gleitende Fenster lässt keinen solchen Spielraum zu.
Darüber hinaus gibt die Funktion nicht nur einen einfachen Boolean zurück, sondern ein Dictionary, das die verbleibenden Anfragen und die Zeit bis zum Reset angibt. Diese Werte füllen die Antwort-Header, die Ihre legitimen Clients zur Selbstregulierung nutzen werden – eine gut gestaltete Integration wartet, statt das Endpunkt zu bombardieren.
import time
from collections import defaultdict
class RateLimiter:
"""Sliding window rate limiter."""
def __init__(self, max_requetes: int, fenetre_secondes: int):
self.max_requetes = max_requetes
self.fenetre = fenetre_secondes
self.requetes: dict[str, list[float]] = defaultdict(list)
def autoriser(self, identifiant: str) -> tuple[bool, dict]:
"""Checks if a request is allowed for this identifier."""
maintenant = time.time()
seuil = maintenant - self.fenetre
# Clean up old requests
self.requetes[identifiant] = [
t for t in self.requetes[identifiant] if t > seuil
]
nb_requetes = len(self.requetes[identifiant])
if nb_requetes >= self.max_requetes:
prochain_slot = self.requetes[identifiant][0] + self.fenetre
return False, {
"restantes": 0,
"reset_dans": round(prochain_slot - maintenant, 1),
}
self.requetes[identifiant].append(maintenant)
return True, {
"restantes": self.max_requetes - nb_requetes - 1,
"reset_dans": round(self.fenetre - (maintenant - self.requetes[identifiant][0]), 1),
}
# Configuration by role
limiteurs = {
"gratuit": RateLimiter(max_requetes=10, fenetre_secondes=60),
"premium": RateLimiter(max_requetes=60, fenetre_secondes=60),
"api": RateLimiter(max_requetes=200, fenetre_secondes=60),
}Das Zählen von Anfragen reicht jedoch nicht aus, da nicht alle Anfragen gleich sind. Zehn kurze Fragen kosten einen Bruchteil dessen, was eine einzelne Anfrage mit einem 50-seitigen Dokument kostet. Ein pro-Anfrage-Limiter würde das letztere durchlassen und das erstere blockieren, was genau das Gegenteil dessen ist, was die Service-Ökonomie diktiert. Ein Token-Budget stellt die Proportionalität wieder her, mit einem täglichen Reset und einer Aufzeichnung des tatsächlichen Verbrauchs nachträglich, da die Schätzung vor dem Aufruf die Länge der Antwort nicht kennt.
class TokenBudgetLimiter:
"""Limits token spending per user."""
def __init__(self, budget_tokens_jour: int, cout_par_1k_input: float, cout_par_1k_output: float):
self.budget_jour = budget_tokens_jour
self.cout_input = cout_par_1k_input
self.cout_output = cout_par_1k_output
self.usage: dict[str, dict] = defaultdict(lambda: {"tokens_in": 0, "tokens_out": 0, "date": ""})
def verifier_budget(self, user_id: str, tokens_estimes: int) -> tuple[bool, dict]:
"""Checks if the user still has budget."""
from datetime import date
aujourd_hui = str(date.today())
# Daily reset
if self.usage[user_id]["date"] != aujourd_hui:
self.usage[user_id] = {"tokens_in": 0, "tokens_out": 0, "date": aujourd_hui}
total_utilise = self.usage[user_id]["tokens_in"] + self.usage[user_id]["tokens_out"]
restant = self.budget_jour - total_utilise
if tokens_estimes > restant:
return False, {"restant": restant, "demande": tokens_estimes}
return True, {"restant": restant - tokens_estimes}
def enregistrer_usage(self, user_id: str, tokens_in: int, tokens_out: int):
"""Records actual usage after a call."""
self.usage[user_id]["tokens_in"] += tokens_in
self.usage[user_id]["tokens_out"] += tokens_out
# Example: 100K tokens/day for free users
budget = TokenBudgetLimiter(
budget_tokens_jour=100_000,
cout_par_1k_input=0.005,
cout_par_1k_output=0.015,
)Erkennen eines Bots
Ein methodischer Angreifer bleibt unter Ihren Deckungen: Er muss nur langsamer werden. Das Erkennen von Missbrauch erfordert daher einen Blick auf die Form des Verkehrs, nicht auf sein Volumen. Die vier unten implementierten Signale ergänzen sich gut. Identische Wiederholungen verraten ein Skript, das in einer Schleife stecken ist. Eine Häufung von Anfragen innerhalb weniger Sekunden deutet auf ein Tempo hin, das kein Mensch am Tastatur halten kann. Eine Nachricht mit zehntausenden von Zeichen signalisiert einen volumetrischen Injektionsversuch. Das vierte Signal ist das interessanteste: Rigorös identische Nachrichtenlängen über zehn aufeinanderfolgende Anfragen offenbaren einen Fuzzer, der den Inhalt mit konstanter Struktur variiert, während ein echter Nutzer Sätze mit unregelmäßigen Längen schreibt.
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class ProfilUtilisateur:
user_id: str
requetes_recentes: list[dict] = field(default_factory=list)
alertes: list[str] = field(default_factory=list)
class DetecteurAbus:
"""Detects automated abuse patterns."""
def __init__(self):
self.profils: dict[str, ProfilUtilisateur] = {}
def analyser(self, user_id: str, message: str) -> list[str]:
"""Analyzes a message to detect abuse patterns."""
if user_id not in self.profils:
self.profils[user_id] = ProfilUtilisateur(user_id=user_id)
profil = self.profils[user_id]
alertes = []
# Pattern 1: repeated identical messages
if profil.requetes_recentes:
derniers_messages = [r["message"] for r in profil.requetes_recentes[-5:]]
if derniers_messages.count(message) >= 3:
alertes.append("SPAM: identical message repeated 3+ times")
# Pattern 2: abnormal frequency
maintenant = datetime.now()
recentes = [
r for r in profil.requetes_recentes
if (maintenant - r["timestamp"]).seconds < 10
]
if len(recentes) > 5:
alertes.append("FLOOD: 5+ requests in 10 seconds")
# Pattern 3: very long messages (volumetric injection attempt)
if len(message) > 10000:
alertes.append(f"VOLUME: message of {len(message)} characters")
# Pattern 4: systematic variation (automated fuzzing)
if len(profil.requetes_recentes) >= 10:
longueurs = [len(r["message"]) for r in profil.requetes_recentes[-10:]]
if max(longueurs) - min(longueurs) < 5 and len(set(longueurs)) == 1:
alertes.append("FUZZING: identical message length (bot)")
# Record
profil.requetes_recentes.append({
"message": message,
"timestamp": maintenant,
"alertes": alertes,
})
return alertes
detecteur = DetecteurAbus()Stufenweise Sanktionen
Ein Verbot bei der ersten Warnung wäre ein kostspieliger Fehler, da keines dieser Signale unfehlbar ist: Ein Nutzer, der nervös eine Seite neu lädt, löst den Burst-Detektor aus, ohne bösartig zu sein. Ein progressives Verbot akkumuliert einen Punktestand und löst proportionale Reaktionen aus – eine Warnung, dann Drosselung, dann eine einstündige Sperrung und schließlich eine 24-Stunden-Sperre. Jede Stufe gibt dem legitimen Nutzer eine Chance, aus der Bahn zu geraten, während das automatisierte Skript sie bis zum Ende verfolgt.
class GestionBannissement:
"""Progressive banning: warning, throttling, blocking."""
def __init__(self):
self.compteurs: dict[str, int] = defaultdict(int)
self.bans: dict[str, float] = {}
def signaler(self, user_id: str, severite: int = 1) -> str:
"""Reports abuse and applies the appropriate sanction."""
self.compteurs[user_id] += severite
score = self.compteurs[user_id]
if score >= 10:
self.bans[user_id] = time.time() + 86400 # Ban 24h
return "BAN_24H"
elif score >= 5:
self.bans[user_id] = time.time() + 3600 # Ban 1h
return "BAN_1H"
elif score >= 3:
return "RALENTIR" # Reduced rate limit
else:
return "AVERTIR"
def est_banni(self, user_id: str) -> bool:
"""Checks if a user is currently banned."""
if user_id in self.bans:
if time.time() < self.bans[user_id]:
return True
del self.bans[user_id]
return FalseDie Reihenfolge der Prüfungen ist entscheidend
Die finale Assemblierung verknüpft die vier Mechanismen, bevor das Modell aufgerufen wird, und diese Reihenfolge folgt einer einfachen Regel: vom kostengünstigsten zum teuersten. Die Ban-Prüfung ist ein Dictionary-Lookup, Rate Limiting scannt eine kurze Liste, Missbrauchserkennung analysiert die Historie, und die Budgetkontrolle erfolgt kurz vor dem abgerechneten Aufruf. Eine Umkehrung dieser Sequenz würde bedeuten, für Verhaltensanalysen zu zahlen, für einen Nutzer, der bereits gebannt ist. Der letzte Schritt ist der, der am häufigsten vergessen wird: Die Aufzeichnung des tatsächlichen Verbrauchs aus dem usage-Feld der Antwort, ohne die Ihr Budget vom vor dem Aufruf getroffenen Schätzung abweichen würde.
from openai import OpenAI
class APISecurisee:
"""API with integrated rate limiting and abuse detection."""
def __init__(self):
self.client = OpenAI()
self.rate_limiter = RateLimiter(max_requetes=20, fenetre_secondes=60)
self.budget = TokenBudgetLimiter(100_000, 0.005, 0.015)
self.detecteur = DetecteurAbus()
self.bans = GestionBannissement()
def traiter(self, user_id: str, message: str) -> str:
# Step 1: check ban status
if self.bans.est_banni(user_id):
return "Your access is temporarily suspended."
# Step 2: rate limiting
autorise, info = self.rate_limiter.autoriser(user_id)
if not autorise:
return f"Too many requests. Try again in {info['reset_dans']}s."
# Step 3: abuse detection
alertes = self.detecteur.analyser(user_id, message)
if alertes:
action = self.bans.signaler(user_id, severite=len(alertes))
if action.startswith("BAN"):
return "Abusive behavior detected. Access suspended."
# Step 4: check token budget
tokens_estimes = len(message) // 4 + 500
ok, budget_info = self.budget.verifier_budget(user_id, tokens_estimes)
if not ok:
return "Daily token budget exhausted."
# Step 5: call the model
response = self.client.chat.completions.create(
model="gpt-5.6-sol",
messages=[{"role": "user", "content": message}],
)
# Record actual usage
usage = response.usage
self.budget.enregistrer_usage(user_id, usage.prompt_tokens, usage.completion_tokens)
return response.choices[0].message.contentKernaussagen
- Rate Limiting schützt Ihr API-Budget und verhindert automatisierte Angriffe.
- Kombinieren Sie pro-Anfrage-Limits (gleitendes Fenster) und pro-Kosten-Limits (Tokens).
- Missbrauchserkennung identifiziert Muster: Spam, Flood, Fuzzing, Volumen.
- Progressives Verbot (warnen, drosseln, blockieren) vermeidet permanente Falschpositiv-Erkennungen.
- Jede Antwort sollte Rate-Limit-Header enthalten, um den Nutzer zu informieren.