Bootcamp i backendutvikling med FastAPI · leksjon

Begrensning av forespørselstakt og botbeskyttelse

Implementer distribuert begrensning av forespørselstakt og struping med Redis for å absorbere topper og blokkere misbrukende klienter.

Leksjon 2 av 413 trinn

Begrensning av forespørselstakt og botbeskyttelse er en gratis leksjon i Bootcamp i backendutvikling med FastAPI på CoddyKit. Dette er leksjon 2 av 4. Du kan lese valgfritt 3 leksjoner fra denne læringsstien gratis i sin helhet – deretter låser CoddyKit PRO opp alle leksjoner, samt praktisk øving med en innebygd kodeeditor og en AI-veileder som er tilgjengelig døgnet rundt. Den er en del av læringsløpet i Bootcamp i backendutvikling med FastAPI, og fremdriften din synkroniseres mellom nettet og CoddyKit-appen. Kurset i Bootcamp i backendutvikling med FastAPI inneholder totalt 4 leksjoner.

Hvorfor distribuert frekvensbegrensning

En enkelt FastAPI-worker som lagrer en teller for frekvensbegrensning i prosessminnet, slutter å fungere så snart du skalerer horisontalt. Med N Uvicorn-workere bak en lastbalanserer får en misbrukende klient N ganger så stor tillatt mengde, fordi hver worker teller uavhengig.

Mål: én delt teller som alle workere og pods ser.
Verktøy: Redis, et atomisk lager i minnet som alle instanser kobler seg til.
Målsettinger: håndtere legitime trafikkøkninger samtidig som misbrukende boter begrenses eller blokkeres.

Gjennom hele leksjonen bygger vi først algoritmene i ren Python og kobler dem deretter til FastAPI-avhengigheter.

Telleren for fast tidsvindu

Den enkleste algoritmen er å dele tiden inn i faste vinduer (for eksempel intervaller på 60 sekunder) og øke en teller per klient per vindu. Når telleren overskrider grensen, avvises forespørselen.

Nedenfor ser du en simulering i ren Python, slik at du kan se mekanikken uten Redis. Legg merke til grenseproblemet: En klient kan sende hele kvoten på slutten av ett vindu og på nytt i starten av det neste, slik at den effektive frekvensen kortvarig blir doblet.

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"}')

Logg for glidende tidsvindu

For å eliminere grenseproblemet sporer vi en logg over tidsstempler per klient og teller bare hendelsene som faller innenfor det etterfølgende vinduet relativt til now. Dette er nøyaktig, men krever mye minne: én oppføring per forespørsel.

Mønsteret nedenfor er nøyaktig det vi skal oversette til et sortert Redis-sett, der scoren er tidsstempelet.

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"}')

Token bucket: Håndtere trafikkøkninger

Faste og glidende tidsvinduer er strenge. For å håndtere plutselige trafikkøkninger samtidig som du håndhever et gjennomsnitt over lang tid, er token bucket standardvalget.

Hver klient har en bøtte med en kapasitet (maksimal burst) og en påfyllingshastighet (tokens per sekund).
Hver forespørsel bruker ett token. Hvis bøtten er tom, begrenses forespørselen.
En burst på opptil capacity forespørsler slipper gjennom umiddelbart, og trafikken jevnes deretter ut til påfyllingshastigheten.

Dette er algoritmen vi anbefaler for offentlige API-er som møter trafikk med legitime, men plutselige, økninger.

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"}')

Hvorfor atomisitet er viktig

På tvers av mange workere er lesing–endring–skriving av en teller en klassisk kappløpstilstand: To workere leser count=4, begge øker verdien, begge skriver 5, og en forespørsel som skulle ha blitt blokkert, slipper gjennom.

Redis løser dette fordi hver kommando er atomisk, men en avgjørelse om frekvensbegrensning trenger vanligvis flere kommandoer (øk, angi utløpstid, sammenlign). Den robuste løsningen er et Lua-skript som kjøres av EVAL: Redis kjører hele skriptet atomisk, uten at andre klienter kan komme imellom.

De neste scenene viser Redis-basert fast tidsvindu og token bucket ved hjelp av denne tilnærmingen.

Redis: Fast tidsvindu med INCR + EXPIRE

Den enkleste distribuerte begrenseren: én Redis-nøkkel per klient per tidsvindu. INCR returnerer den nye telleren atomisk. Ved første treff angir vi en TTL som tilsvarer vinduet, slik at nøkkelen slettes automatisk.

Vi pakker begge kommandoene inn i et lite Lua-skript, slik at økningen og utløpstiden utgjør én atomisk enhet. Dette er limkode på FastAPI-siden, ikke et frittstående program.

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-token bucket i Lua

For å håndtere plutselige trafikkøkninger overfører vi token bucket til et Lua-skript. Tilstanden ligger i en Redis-hash som inneholder tokens og ts (tidspunktet for siste påfylling). Skriptet fyller på basert på tiden som har gått, forsøker å bruke ett token og skriver tilstanden tilbake, alt atomisk.

Returverdien 1 betyr tillatt, mens 0 betyr begrenset. Vi angir også en TTL, slik at inaktive klienter frigjør minnet de bruker.

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()])

En gjenbrukbar FastAPI-avhengighet

Vi eksponerer begrenseren som en avhengighet, slik at alle ruter kan velge å bruke den. Avhengigheten utleder klientidentiteten, kjører begrenseren og utløser HTTPException(429) når kvoten er brukt opp.

Foretrekk en autentisert identitet (API-nøkkel, bruker-ID) fremfor rå IP-adresse, fordi IP-adresser deles bak NAT og enkelt kan roteres av boter. Bruk IP-adresse som reserve bare for anonym trafikk.

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(): ...

Ærlige 429-svar: Retry-After og headere

En korrekt begrenser er også en høflig begrenser. Velfungerende klienter respekterer standardiserte signaler, og ved å sende dem reduserer du antall nye forsøk og henvendelser til brukerstøtten.

429 Too Many Requests er den eneste riktige statusen. Bruk aldri 403 eller 503 for frekvensbegrensning.
Retry-After forteller klienten hvor mange sekunder den skal vente.
X-RateLimit-Limit, X-RateLimit-Remaining og X-RateLimit-Reset lar klientene regulere hastigheten selv.

La Lua-skriptet også returnere gjenværende antall og tidspunktet for tilbakestilling, slik at du kan fylle ut disse headerne uten ekstra rundturer.

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),
            },
        )

Nivådelt begrensning og eskalering av misbruk

Én global grense er lite presis. Systemer i produksjon legger flere grenser oppå hverandre:

Grenser per rute: En billig GET tåler langt mer trafikk enn et kostbart søke- eller innloggingsendepunkt.
Nivådelt identitet: Anonyme IP-adresser får en streng kvote, autentiserte brukere får mer, og betalte abonnementer får mest.
Eskalering: Når en klient gjentatte ganger får 429, skriver du den til en Redis-nekteliste med en eksponentiell TTL, slik at misbrukende boter får stadig lengre utestengelser.

Hjelpefunksjonen nedenfor beregner en dobling av utestengelsesvarigheten, begrenset til én time.

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')

Oppdage boter utover telling

Frekvensbegrensning begrenser mengden trafikk, men avanserte boter holder seg like under grensen. Kombiner begrensning med rimelige atferdssignaler:

Manglende eller suspekte headere: manglende User-Agent eller en UA som står på en kjent blokkeringsliste.
Forholdet mellom mislykkede og vellykkede innlogginger: Et høyt forhold mellom mislykket og vellykket autentisering per IP kan tyde på credential stuffing.
Stientropi: Raske treff på mange urelaterte endepunkter kan tyde på scraping.
Proof of work / CAPTCHA som en kontroll når en score overskrider en terskel, i stedet for en fullstendig blokkering.

Bruk disse signalene i en risikoscore per klient i Redis, og reduser token bucket-kapasiteten dynamisk for klienter med høy risiko.

Hurtigsjekk: Velge algoritme

Det offentlige FastAPI-API-et ditt kjører på mange Uvicorn-workere og flere pods. Legitime klienter sender av og til korte legitime bursts, men du må håndheve et jevnt gjennomsnitt over lang tid og sørge for at avgjørelsen er konsekvent på tvers av alle instanser. Hvilket design passer best?

Besvar den i den interaktive leksjonen

Oppsummering og sjekkliste for produksjon

Du kan nå bygge distribuert, misbruksbestandig frekvensbegrensning for FastAPI:

Sentraliser tilstanden i Redis, slik at alle workere og pods deler én teller.
Velg algoritme etter behov: fast tidsvindu for enkle grenser, glidende logg for nøyaktighet, og token bucket for å håndtere bursts med et håndhevet gjennomsnitt.
Garanter atomisitet med Lua-skript via EVAL for å unngå kappløpstilstander ved lesing–endring–skriving.
Eksponer begrensere som FastAPI-avhengigheter med autentisert identitet som førstevalg og IP som reserve.
Svar ærlig med 429, Retry-After og X-RateLimit-*-headere.
Legg forsvarslag oppå hverandre: grenser per rute og nivå, nektelister med eksponentiell tilbakeventing, og atferdsbasert bot-scoring utover ren telling.

Feil alltid kontrollert åpent: Hvis Redis ikke er tilgjengelig, må du bevisst avgjøre om forespørsler skal tillates eller blokkeres, og varsle om situasjonen.

Gratis å komme i gang

Lær deg Bootcamp i backendutvikling med FastAPI med en AI-veileder – gratis

Skriv og kjør ekte kode i nettleseren, få umiddelbar hjelp fra en AI-veileder som er tilgjengelig døgnet rundt, og fortsett der du slapp – på nettet eller i appen.

Kurs: 21
Leksjoner: 84

Start læringen gratis Se hele kurset

Ofte stilte spørsmål

Er leksjonen «Begrensning av forespørselstakt og botbeskyttelse» gratis?

Ja – du kan lese valgfritt 3 av leksjonene i læringsstien Bootcamp i backendutvikling med FastAPI, inkludert «Begrensning av forespørselstakt og botbeskyttelse», gratis i sin helhet her på nettet. Deretter låser CoddyKit PRO opp alle leksjoner, samt interaktiv øving med en innebygd kodeeditor og en AI-veileder som er tilgjengelig døgnet rundt. Kurset i Bootcamp i backendutvikling med FastAPI inneholder totalt 4 leksjoner.

Hva lærer jeg i «Begrensning av forespørselstakt og botbeskyttelse»?

Implementer distribuert begrensning av forespørselstakt og struping med Redis for å absorbere topper og blokkere misbrukende klienter. Du øver på Bootcamp i backendutvikling med FastAPI med praktisk kode som du kjører direkte i nettleseren, mens en AI-veileder som er tilgjengelig døgnet rundt, svarer på spørsmålene dine mens du jobber deg gjennom leksjonen.

Trenger jeg erfaring for å begynne med Bootcamp i backendutvikling med FastAPI?

Ingen tidligere erfaring er nødvendig. Bootcamp i backendutvikling med FastAPI på CoddyKit er lagt opp for både nybegynnere og viderekomne, så De kan begynne her eller helt fra start og lære i Deres eget tempo. Dette er leksjon 2 av 4.

Hvor lang tid tar leksjonen «Begrensning av forespørselstakt og botbeskyttelse»?

De fleste CoddyKit-leksjoner tar omtrent 5–10 minutter. Hver leksjon er kort og interaktiv, slik at De gjør jevne fremskritt og kan fortsette akkurat der De slapp – både på nettet og i appen.

Kan jeg skrive og kjøre kode i denne Bootcamp i backendutvikling med FastAPI-leksjonen?

Ja. Alle Bootcamp i backendutvikling med FastAPI-leksjoner har en innebygd kodeeditor, slik at De kan skrive og kjøre ekte kode direkte i nettleseren og få umiddelbar tilbakemelding fra AI – uten lokal konfigurering.

Alle leksjonene i dette kurset

Redusere risikoen fra OWASP API Security Top 10
Begrensning av forespørselstakt og botbeskyttelse
Håndtering av hemmeligheter og nøkkelrotasjon
CORS-, CSP- og sikre header-policyer

← Tilbake til Bootcamp i backendutvikling med FastAPI