Rate limiting et abuse prevention

Pourquoi limiter les requêtes ?

Les applications IA sont des cibles privilégiées pour l’abus, et pour une raison qui n’existait pas dans les applications classiques : chaque appel au modèle coûte de l’argent. Une API REST traditionnelle soumise à un flot de requêtes ralentit et finit par tomber, mais elle ne vide pas un compte bancaire. Une application LLM, elle, continue de répondre parfaitement pendant que la facture grimpe. Sans rate limiting, un attaquant peut ruiner votre budget API en quelques minutes, lancer des attaques de prompt injection automatisées ou utiliser votre infrastructure pour générer du spam à vos frais.

Deux façons de compter

La fenêtre glissante est le mécanisme de base. Elle mémorise l’horodatage de chaque requête, écarte celles qui sont sorties de la fenêtre, et compare ce qui reste au plafond. Sa supériorité sur la fenêtre fixe tient à un détail qui compte en pratique : avec une fenêtre fixe réinitialisée chaque minute, un attaquant place son quota en fin de minute puis à nouveau au début de la suivante, et obtient le double du débit autorisé sur quelques secondes. La fenêtre glissante ne laisse pas cette ouverture.

La fonction ne renvoie d’ailleurs pas un simple booléen, mais un dictionnaire indiquant les requêtes restantes et le délai avant réinitialisation. Ces valeurs alimentent les en-têtes de réponse que vos clients légitimes utiliseront pour se réguler d’eux-mêmes — une intégration bien conçue attend plutôt que de marteler.

import time
from collections import defaultdict

class RateLimiter:
    """Rate limiter par fenêtre glissante."""

    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]:
        """Vérifie si une requête est autorisée pour cet identifiant."""
        maintenant = time.time()
        seuil = maintenant - self.fenetre

        # Nettoyer les anciennes requêtes
        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 par rôle
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),
}

Compter les requêtes ne suffit pourtant pas, parce que toutes ne se valent pas. Dix questions courtes coûtent une fraction de ce que coûte une seule requête accompagnée d’un document de cinquante pages. Un limiteur par requête laisserait passer la seconde et bloquerait les premières, exactement à l’envers de ce que dicte l’économie du service. Le budget en tokens rétablit la proportionnalité, avec une remise à zéro quotidienne et un enregistrement de l’usage réel après coup, puisque l’estimation faite avant l’appel ne connaît pas la longueur de la réponse.

class TokenBudgetLimiter:
    """Limite les dépenses en tokens par utilisateur."""

    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]:
        """Vérifie si l'utilisateur a encore du budget."""
        from datetime import date
        aujourd_hui = str(date.today())

        # Reset quotidien
        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):
        """Enregistre l'usage réel après un appel."""
        self.usage[user_id]["tokens_in"] += tokens_in
        self.usage[user_id]["tokens_out"] += tokens_out

# Exemple : 100K tokens/jour pour les utilisateurs gratuits
budget = TokenBudgetLimiter(
    budget_tokens_jour=100_000,
    cout_par_1k_input=0.005,
    cout_par_1k_output=0.015,
)

Reconnaître un automate

Un attaquant méthodique reste sous vos plafonds : il lui suffit de ralentir. Détecter l’abus demande donc de regarder la forme du trafic plutôt que son volume. Les quatre signaux implémentés ci-dessous se complètent bien. La répétition à l’identique trahit un script coincé dans une boucle. La rafale de requêtes en quelques secondes indique une cadence qu’aucun humain ne tient au clavier. Un message de plusieurs dizaines de milliers de caractères signale une tentative d’injection volumétrique. Le quatrième est le plus intéressant : des longueurs de messages rigoureusement identiques sur dix requêtes consécutives révèlent un fuzzer qui fait varier le contenu à structure constante, alors qu’un utilisateur réel écrit des phrases de tailles irrégulières.

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:
    """Détecte les patterns d'abus automatisé."""

    def __init__(self):
        self.profils: dict[str, ProfilUtilisateur] = {}

    def analyser(self, user_id: str, message: str) -> list[str]:
        """Analyse un message pour détecter des patterns d'abus."""
        if user_id not in self.profils:
            self.profils[user_id] = ProfilUtilisateur(user_id=user_id)

        profil = self.profils[user_id]
        alertes = []

        # Pattern 1 : messages identiques répétés
        if profil.requetes_recentes:
            derniers_messages = [r["message"] for r in profil.requetes_recentes[-5:]]
            if derniers_messages.count(message) >= 3:
                alertes.append("SPAM: message identique répété 3+ fois")

        # Pattern 2 : fréquence anormale
        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+ requêtes en 10 secondes")

        # Pattern 3 : messages très longs (tentative d'injection volumétrique)
        if len(message) > 10000:
            alertes.append(f"VOLUME: message de {len(message)} caractères")

        # Pattern 4 : variation systématique (fuzzing automatisé)
        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: messages de longueur identique (automate)")

        # Enregistrer
        profil.requetes_recentes.append({
            "message": message,
            "timestamp": maintenant,
            "alertes": alertes,
        })

        return alertes

detecteur = DetecteurAbus()

Sanctionner par paliers

Bannir dès la première alerte serait une erreur coûteuse, car aucun de ces signaux n’est infaillible : un utilisateur qui rafraîchit nerveusement une page déclenche le détecteur de rafale sans être malveillant. Le bannissement progressif accumule un score et déclenche des réponses proportionnées — un avertissement, puis un ralentissement, puis un blocage temporaire d’une heure, enfin de vingt-quatre heures. Chaque palier laisse à l’utilisateur légitime l’occasion de sortir de la trajectoire, tandis que le script automatisé, lui, la suit jusqu’au bout.

class GestionBannissement:
    """Bannissement progressif : avertissement, ralentissement, blocage."""

    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:
        """Signale un abus et applique la sanction appropriée."""
        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"  # Rate limit réduit
        else:
            return "AVERTIR"

    def est_banni(self, user_id: str) -> bool:
        """Vérifie si un utilisateur est actuellement banni."""
        if user_id in self.bans:
            if time.time() < self.bans[user_id]:
                return True
            del self.bans[user_id]
        return False

L’ordre des vérifications n’est pas neutre

L’assemblage final enchaîne les quatre mécanismes avant d’appeler le modèle, et cet ordre obéit à une règle simple : du moins coûteux au plus coûteux. La vérification de bannissement est une consultation de dictionnaire, le rate limiting parcourt une courte liste, la détection d’abus analyse l’historique, et le contrôle de budget arrive juste avant l’appel facturé. Inverser cette séquence reviendrait à payer l’analyse comportementale pour un utilisateur déjà banni. Le dernier geste est celui qu’on oublie le plus souvent : l’enregistrement de l’usage réel à partir du champ usage de la réponse, sans lequel votre budget dériverait de l’estimation faite avant l’appel.

from openai import OpenAI

class APISecurisee:
    """API avec rate limiting et détection d'abus intégrés."""

    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:
        # Étape 1 : vérifier le bannissement
        if self.bans.est_banni(user_id):
            return "Votre accès est temporairement suspendu."

        # Étape 2 : rate limiting
        autorise, info = self.rate_limiter.autoriser(user_id)
        if not autorise:
            return f"Trop de requêtes. Réessayez dans {info['reset_dans']}s."

        # Étape 3 : détection d'abus
        alertes = self.detecteur.analyser(user_id, message)
        if alertes:
            action = self.bans.signaler(user_id, severite=len(alertes))
            if action.startswith("BAN"):
                return "Comportement abusif détecté. Accès suspendu."

        # Étape 4 : vérifier le budget tokens
        tokens_estimes = len(message) // 4 + 500
        ok, budget_info = self.budget.verifier_budget(user_id, tokens_estimes)
        if not ok:
            return "Budget quotidien de tokens épuisé."

        # Étape 5 : appel au modèle
        response = self.client.chat.completions.create(
            model="gpt-5.6-sol",
            messages=[{"role": "user", "content": message}],
        )

        # Enregistrer l'usage réel
        usage = response.usage
        self.budget.enregistrer_usage(user_id, usage.prompt_tokens, usage.completion_tokens)

        return response.choices[0].message.content

Points clés à retenir

• Le rate limiting protège votre budget API et empêche les attaques automatisées
• Combinez limite par requête (fenêtre glissante) et limite par coût (tokens)
• La détection d’abus identifie les patterns : spam, flood, fuzzing, volume
• Le bannissement progressif (avertir, ralentir, bloquer) évite les faux positifs définitifs
• Chaque réponse doit inclure les headers de rate limit pour informer l’utilisateur