Rate limiting and abuse prevention
On this page
Why limit requests?
AI applications are prime targets for abuse for a reason that didn’t exist with traditional apps: every model call costs money. A traditional REST API subjected to a flood of requests slows down and eventually crashes, but it doesn’t drain a bank account. An LLM application, however, continues to respond perfectly while the bill skyrockets. Without rate limiting, an attacker can ruin your API budget in minutes, launch automated prompt injection attacks, or use your infrastructure to generate spam at your expense.
Two ways to count
The sliding window is the fundamental mechanism. It records the timestamp of each request, discards those that have fallen out of the window, and compares the remainder against the ceiling. Its superiority over the fixed window comes down to a detail that matters in practice: with a fixed window that resets every minute, an attacker can place their quota at the end of one minute and again at the start of the next, effectively doubling the allowed throughput over a few seconds. The sliding window leaves no such opening.
Moreover, the function doesn’t return a simple boolean; it returns a dictionary indicating the remaining requests and the time until reset. These values populate the response headers that your legitimate clients will use to self-regulate—a well-designed integration waits rather than hammering the endpoint.
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),
}Counting requests isn’t enough, though, because not all requests are equal. Ten short questions cost a fraction of what a single request accompanied by a fifty-page document costs. A per-request limiter would let the latter through and block the former, which is exactly the opposite of what service economics dictate. A token budget restores proportionality, with a daily reset and recording of actual usage after the fact, since the estimate made before the call doesn’t know the length of the response.
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,
)Recognizing a bot
A methodical attacker stays under your ceilings: they just need to slow down. Detecting abuse therefore requires looking at the shape of traffic rather than its volume. The four signals implemented below complement each other well. Identical repetition betrays a script stuck in a loop. A burst of requests within a few seconds indicates a pace no human can maintain at the keyboard. A message tens of thousands of characters long signals a volumetric injection attempt. The fourth is the most interesting: rigorously identical message lengths over ten consecutive requests reveal a fuzzer varying content with constant structure, whereas a real user writes sentences of irregular lengths.
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()Tiered sanctions
Banning on the first alert would be a costly mistake, because none of these signals are infallible: a user nervously refreshing a page triggers the burst detector without being malicious. Progressive banning accumulates a score and triggers proportional responses—a warning, then throttling, then a temporary one-hour block, and finally twenty-four hours. Each tier gives the legitimate user a chance to get off the trajectory, while the automated script follows it to the end.
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 FalseThe order of checks matters
The final assembly chains the four mechanisms before calling the model, and this order obeys a simple rule: from least expensive to most expensive. The ban check is a dictionary lookup, rate limiting scans a short list, abuse detection analyzes history, and budget control comes just before the billed call. Reversing this sequence would mean paying for behavioral analysis for a user who is already banned. The last step is the one most often forgotten: recording actual usage from the usage field of the response, without which your budget would drift from the estimate made before the call.
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.contentKey takeaways
- Rate limiting protects your API budget and prevents automated attacks
- Combine per-request limits (sliding window) and per-cost limits (tokens)
- Abuse detection identifies patterns: spam, flood, fuzzing, volume
- Progressive banning (warn, throttle, block) avoids permanent false positives
- Every response should include rate limit headers to inform the user