Rate limiting en bescherming tegen botmisbruik
Implementeer distributed rate limiting en throttling met Redis om pieken op te vangen en misbruikende clients te blokkeren.
Rate limiting en bescherming tegen botmisbruik is een gratis Bootcamp backendontwikkeling met FastAPI-les op CoddyKit. Dit is les 2 van 4. Je kunt 3 lessen uit dit leerpad gratis volledig lezen — daarna ontgrendelt CoddyKit PRO alle lessen, plus praktische oefeningen met een ingebouwde code-editor en een AI-tutor die 24/7 beschikbaar is. Deze les maakt deel uit van het leertraject Bootcamp backendontwikkeling met FastAPI. Je voortgang wordt gesynchroniseerd op het web en in de CoddyKit-app. De cursus Bootcamp backendontwikkeling met FastAPI bevat in totaal 4 lessen.
Waarom gedistribueerde snelheidsbeperking nodig is
Een enkele FastAPI-worker die een teller voor snelheidsbeperking in procesgeheugen bijhoudt, werkt niet meer zodra je horizontaal schaalt. Met N Uvicorn-workers achter een load balancer krijgt een misbruikende client N keer zoveel ruimte, omdat elke worker onafhankelijk telt.
- Doel: één gedeelde teller die elke worker en elke pod kan zien.
- Hulpmiddel: Redis, een atomair in-memory gegevensarchief waarmee alle instanties verbinding maken.
- Doelstellingen: legitieme verkeerspieken opvangen en tegelijkertijd misbruikende bots afremmen of blokkeren.
In deze les bouwen we de algoritmen eerst in gewone Python en verbinden we ze daarna met afhankelijkheden in FastAPI.
De teller voor een vast tijdvenster
Het eenvoudigste algoritme: deel de tijd op in vaste vensters (bijvoorbeeld blokken van 60 seconden) en verhoog per client en per venster een teller. Zodra de teller de limiet overschrijdt, weiger je het verzoek.
Hieronder staat een simulatie in pure Python, zodat je de werking zonder Redis kunt bekijken. Let op het grensprobleem: een client kan aan het einde van het ene venster het volledige quotum versturen en aan het begin van het volgende opnieuw, waardoor de effectieve snelheid tijdelijk verdubbelt.
import time
class FixedWindow:
def __init__(self, limit, window_seconds):
self.limit = limit
self.window = window_seconds
self.buckets = {}
def allow(self, key, now):
win = int(now // self.window)
bucket_key = (key, win)
count = self.buckets.get(bucket_key, 0) + 1
self.buckets[bucket_key] = count
return count <= self.limit
limiter = FixedWindow(limit=3, window_seconds=60)
base = 1000.0
for i in range(5):
ok = limiter.allow('user:42', base + i)
print(f'request {i+1}: {"ALLOW" if ok else "BLOCK"}')Het logboek voor een schuivend tijdvenster
Om het grensprobleem op te lossen, houden we per client een logboek met tijdstempels bij en tellen we alleen de gebeurtenissen die binnen het voorafgaande venster ten opzichte van now vallen. Dit is exact, maar gebruikt veel geheugen: één item per verzoek.
Het onderstaande patroon vertalen we straks rechtstreeks naar een gesorteerde Redis-set, waarbij de score de tijdstempel is.
import time
from collections import deque
class SlidingLog:
def __init__(self, limit, window_seconds):
self.limit = limit
self.window = window_seconds
self.logs = {}
def allow(self, key, now):
dq = self.logs.setdefault(key, deque())
cutoff = now - self.window
while dq and dq[0] <= cutoff:
dq.popleft()
if len(dq) < self.limit:
dq.append(now)
return True
return False
limiter = SlidingLog(limit=2, window_seconds=10)
for t in [0, 1, 2, 11, 12]:
ok = limiter.allow('ip:1.2.3.4', float(t))
print(f't={t}s -> {"ALLOW" if ok else "BLOCK"}')Tokenbucket: pieken opvangen
Vaste en schuivende tijdvensters zijn strikt. Om pieken op te vangen en tegelijkertijd een langetermijngemiddelde af te dwingen, is de tokenbucket de standaardkeuze.
- Elke client heeft een bucket met een capaciteit (maximale piek) en een vulsnelheid (tokens per seconde).
- Elk verzoek verbruikt één token; als de bucket leeg is, wordt het verzoek afgeremd.
- Een piek van maximaal
capacityverzoeken wordt direct doorgelaten, waarna het verkeer wordt afgevlakt tot de vulsnelheid.
Dit is het algoritme dat we aanbevelen voor openbare API's met wisselend maar legitiem verkeer.
class TokenBucket:
def __init__(self, capacity, refill_per_sec):
self.capacity = capacity
self.refill = refill_per_sec
self.tokens = capacity
self.last = 0.0
def allow(self, now):
elapsed = now - self.last
self.tokens = min(self.capacity, self.tokens + elapsed * self.refill)
self.last = now
if self.tokens >= 1:
self.tokens -= 1
return True
return False
bucket = TokenBucket(capacity=5, refill_per_sec=1)
for t in [0, 0, 0, 0, 0, 0, 3]:
print(f't={t}: {"ALLOW" if bucket.allow(float(t)) else "THROTTLE"}')Waarom atomiciteit belangrijk is
Bij meerdere workers vormt het lezen, aanpassen en schrijven van een teller een klassieke raceconditie: twee workers lezen count=4, verhogen de waarde allebei, schrijven allebei 5 en laten daardoor een verzoek door dat geblokkeerd had moeten worden.
Redis lost dit op omdat elke opdracht atomair is, maar een beslissing over snelheidsbeperking vereist meestal meerdere opdrachten (verhogen, vervaltijd instellen, vergelijken). De robuuste oplossing is een Lua-script dat door EVAL wordt uitgevoerd: Redis voert het volledige script atomair uit, zonder dat andere clients tussendoor kunnen komen.
In de volgende scènes zie je het Redis-gebaseerde vaste tijdvenster en de tokenbucket met deze aanpak.
Vast Redis-tijdvenster met INCR + EXPIRE
De goedkoopste gedistribueerde snelheidsbeperker: één Redis-sleutel per client per venster. INCR retourneert de nieuwe tellerwaarde atomair; bij de eerste treffer stellen we een TTL in die gelijk is aan de vensterduur, zodat de sleutel zichzelf opruimt.
We verpakken beide opdrachten in een klein Lua-script, zodat het verhogen en het instellen van de vervaltijd één atomair geheel vormen. Dit is verbindingscode aan de FastAPI-kant, geen zelfstandig programma.
import redis.asyncio as redis
FIXED_WINDOW_LUA = """
local current = redis.call('INCR', KEYS[1])
if current == 1 then
redis.call('EXPIRE', KEYS[1], ARGV[1])
end
return current
"""
class RedisFixedWindow:
def __init__(self, client, limit, window):
self.client = client
self.limit = limit
self.window = window
self.script = client.register_script(FIXED_WINDOW_LUA)
async def allow(self, identifier: str) -> bool:
bucket = int(__import__('time').time()) // self.window
key = f'rl:fw:{identifier}:{bucket}'
count = await self.script(keys=[key], args=[self.window])
return count <= self.limit
# rl = RedisFixedWindow(redis.from_url('redis://localhost'), 100, 60)Redis-tokenbucket in Lua
Om pieken op te vangen, zetten we de tokenbucket om naar een Lua-script. De status staat in een Redis-hash met tokens en ts (tijdstip van de laatste vulling). Het script vult aan op basis van de verstreken tijd, probeert één token te verbruiken en schrijft de status terug — allemaal atomair.
Een retourwaarde 1 betekent toegestaan en 0 betekent afgeremd. We stellen ook een TTL in, zodat inactieve clients hun geheugenruimte weer vrijgeven.
TOKEN_BUCKET_LUA = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local state = redis.call('HMGET', key, 'tokens', 'ts')
local tokens = tonumber(state[1]) or capacity
local ts = tonumber(state[2]) or now
local delta = math.max(0, now - ts)
tokens = math.min(capacity, tokens + delta * refill)
local allowed = 0
if tokens >= 1 then
tokens = tokens - 1
allowed = 1
end
redis.call('HMSET', key, 'tokens', tokens, 'ts', now)
redis.call('EXPIRE', key, math.ceil(capacity / refill) * 2)
return allowed
"""
# Invoked with EVAL via redis-py: script(keys=[key], args=[cap, refill, time.time()])Een herbruikbare FastAPI-afhankelijkheid
We bieden de snelheidsbeperker aan als afhankelijkheid, zodat elke route ervoor kan kiezen deze te gebruiken. De afhankelijkheid bepaalt de identiteit van de client, voert de snelheidsbeperker uit en genereert HTTPException(429) wanneer het quotum op is.
Gebruik bij voorkeur een geauthenticeerde identiteit (API-sleutel, gebruikers-ID) in plaats van een onbewerkt IP-adres, omdat IP-adressen achter NAT worden gedeeld en bots ze eenvoudig kunnen wisselen. Gebruik alleen voor anoniem verkeer het IP-adres als terugvaloptie.
from fastapi import Request, HTTPException, Depends
def client_identity(request: Request) -> str:
api_key = request.headers.get('x-api-key')
if api_key:
return f'key:{api_key}'
return f'ip:{request.client.host}'
def rate_limit(limit: int, window: int):
async def dependency(request: Request):
ident = client_identity(request)
limiter = request.app.state.limiter
if not await limiter.allow(f'{ident}:{limit}:{window}'):
raise HTTPException(status_code=429, detail='Rate limit exceeded')
return dependency
# @app.get('/search', dependencies=[Depends(rate_limit(30, 60))])
# async def search(): ...Eerlijke 429-antwoorden: Retry-After en headers
Een correcte snelheidsbeperker is ook beleefd. Goed opgevoede clients houden rekening met standaardsignalen; door die te versturen verminder je nieuwe pogingen en supporttickets.
- 429 Too Many Requests is de enige juiste status; gebruik voor afremmen nooit 403 of 503.
Retry-Aftervertelt de client hoeveel seconden deze moet wachten.X-RateLimit-Limit,X-RateLimit-RemainingenX-RateLimit-Resethelpen clients om hun snelheid zelf aan te passen.
Laat je Lua-script ook de resterende tellerwaarde en de resetdatum retourneren, zodat je deze headers zonder extra roundtrips kunt vullen.
from fastapi import Request, HTTPException
async def enforce(request: Request, limit: int, window: int):
limiter = request.app.state.limiter
allowed, remaining, reset_in = await limiter.check(client_identity(request))
if not allowed:
raise HTTPException(
status_code=429,
detail='Rate limit exceeded',
headers={
'Retry-After': str(reset_in),
'X-RateLimit-Limit': str(limit),
'X-RateLimit-Remaining': '0',
'X-RateLimit-Reset': str(reset_in),
},
)Gelaagde limieten en escalatie bij misbruik
Eén globale limiet is bot. Productiesystemen combineren meerdere limieten:
- Limieten per route: een goedkope
GETkan veel meer verkeer verdragen dan een dure zoek- of loginroute. - Gelaagde identiteiten: anonieme IP-adressen krijgen een krap quotum, geauthenticeerde gebruikers meer en betaalde abonnementen het meest.
- Escalatie: wanneer een client herhaaldelijk 429 raakt, schrijf je deze naar een Redis-weigerlijst met een exponentiële TTL, zodat misbruikende bots steeds langer worden geblokkeerd.
De onderstaande helper berekent een verdubbelende blokkeerduur met een maximum van één uur.
def next_ban_seconds(strikes: int) -> int:
base = 60 # 1 minute
cap = 3600 # 1 hour
return min(cap, base * (2 ** strikes))
for s in range(8):
print(f'strike {s}: ban for {next_ban_seconds(s)}s')Bots detecteren zonder alleen te tellen
Snelheidsbeperking begrenst de hoeveelheid verkeer, maar geavanceerde bots blijven net onder de limiet. Combineer afremmen met goedkope gedragsignalen:
- Ontbrekende of onzinnige headers: een ontbrekende
User-Agent, of een UA die op een bekende slechte lijst staat. - Verhouding mislukte logins: een hoge verhouding tussen mislukte en geslaagde authenticaties per IP-adres wijst op credential stuffing.
- Padentropie: snelle verzoeken naar veel ongerelateerde endpoints wijzen op scraping.
- Proof of work / CAPTCHA als poort wanneer een score een drempel overschrijdt, in plaats van direct te blokkeren.
Voer deze signalen in een risicoscore per client in Redis in en verlaag de capaciteit van de tokenbucket dynamisch voor clients met een hoog risico.
Korte controle: het algoritme kiezen
Je openbare FastAPI-API draait over veel Uvicorn-workers en meerdere pods. Legitieme clients versturen soms korte legitieme pieken, maar je moet een stabiel langetermijngemiddelde afdwingen en de beslissing consistent houden over alle instanties. Welk ontwerp past het best?
Samenvatting en controlelijst voor productie
Je kunt nu gedistribueerde, misbruikbestendige afremming voor FastAPI bouwen:
- Centraliseer de status in Redis, zodat elke worker en pod één teller deelt.
- Kies het algoritme op basis van de behoefte: een vast venster voor eenvoudige limieten, een schuivend logboek voor exactheid en een tokenbucket om pieken op te vangen met een afgedwongen gemiddelde.
- Garandeer atomiciteit met Lua-scripts via
EVALom racecondities bij lezen-aanpassen-schrijven te voorkomen. - Bied snelheidsbegrenzers aan als FastAPI-afhankelijkheden, eerst gekoppeld aan de geauthenticeerde identiteit en met het IP-adres als terugvaloptie.
- Reageer eerlijk met 429,
Retry-AfterenX-RateLimit-*-headers. - Combineer verdedigingslagen: limieten per route en per niveau, weigerlijsten met exponentiële vertraging en gedragsgebaseerde botscores naast het simpelweg tellen.
Laat het systeem altijd zorgvuldig open falen: als Redis onbereikbaar is, beslis dan bewust of je verzoeken toestaat of blokkeert en geef daar een waarschuwing over.
Leer Bootcamp backendontwikkeling met FastAPI met een AI-tutor — gratis
Schrijf echte code en voer die uit in je browser, krijg direct hulp van een AI-tutor die 24/7 beschikbaar is en ga verder waar je gebleven bent op het web of in de app.
- Cursussen
- 21
- Lessen
- 84
Veelgestelde vragen
Is de les “Rate limiting en bescherming tegen botmisbruik” gratis?
Ja — je kunt hier op het web alle 3 lessen van het leerpad Bootcamp backendontwikkeling met FastAPI, waaronder “Rate limiting en bescherming tegen botmisbruik”, gratis volledig lezen. Daarna ontgrendelt CoddyKit PRO alle lessen, plus interactieve oefeningen met een ingebouwde code-editor en een AI-tutor die 24/7 beschikbaar is. De cursus Bootcamp backendontwikkeling met FastAPI bevat in totaal 4 lessen.
Wat leer ik in “Rate limiting en bescherming tegen botmisbruik”?
Implementeer distributed rate limiting en throttling met Redis om pieken op te vangen en misbruikende clients te blokkeren. Je oefent met Bootcamp backendontwikkeling met FastAPI door code rechtstreeks in de browser uit te voeren. Een AI-begeleider die 24/7 beschikbaar is beantwoordt je vragen terwijl je de les doorwerkt.
Heb ik ervaring nodig om met Bootcamp backendontwikkeling met FastAPI te beginnen?
Ervaring vooraf is niet nodig. Bootcamp backendontwikkeling met FastAPI op CoddyKit is opgebouwd voor beginners tot gevorderden, zodat je hier of bij het begin kunt starten en in je eigen tempo kunt leren. Dit is les 2 van 4.
Hoe lang duurt de les “Rate limiting en bescherming tegen botmisbruik”?
De meeste lessen van CoddyKit duren ongeveer 5–10 minuten. Elke les is kort en interactief, zodat je gestaag vooruitgaat en op het web en in de app precies verdergaat waar je was gebleven.
Kan ik code schrijven en uitvoeren in deze les over Bootcamp backendontwikkeling met FastAPI?
Ja. Elke les over Bootcamp backendontwikkeling met FastAPI bevat een ingebouwde code-editor, zodat je rechtstreeks in je browser echte code kunt schrijven en uitvoeren en direct feedback van AI krijgt — lokale installatie is niet nodig.
Alle lessen in deze cursus
- De OWASP API Security Top 10 beperken
- Rate limiting en bescherming tegen botmisbruik
- Secretsbeheer en sleutelrotatie
- CORS-, CSP- en veilige headerbeleidsregels