Pular para o conteúdo

    Capítulo 29, Avançado

    Limitação de requisições

    Sem limite, um cliente com defeito (ou malicioso) consegue derrubar a sua API, esgotar uma cota paga ou testar milhões de senhas. A limitação de taxa é a defesa mais simples contra os três.

    A ideia

    Em vez de atender tudo, a API aceita um número máximo de requisições por janela de tempo, por cliente (por exemplo, 5 por minuto por endereço IP). Passou do limite, devolve 429 Too Many Requests, e, de preferência, o cabeçalho Retry-After, que diz quando tentar de novo.

    Com a biblioteca `slowapi`

    O slowapi é a biblioteca mais usada para isso no FastAPI. A chave get_remote_address identifica o cliente pelo IP, e o decorador @limiter.limit define o limite de cada rota. A função da rota precisa receber request: Request, que o slowapi usa para saber de quem é a chamada:

    Terminal
    uv add slowapi
    avancado/cap29_rate_limit.pylinhas 10 a 40
    from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient
from slowapi import Limiter
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter


@app.exception_handler(RateLimitExceeded)
async def limite_excedido(request: Request, exc: RateLimitExceeded):
    return JSONResponse(
        status_code=429,
        content={"detail": "Muitas requisições"},
        headers={"Retry-After": "60"},
    )


@app.get("/dados")
@limiter.limit("5/minute")
def dados(request: Request):
    return {"mensagem": "sucesso"}


cliente = TestClient(app)
respostas = [cliente.get("/dados") for _ in range(7)]
print([r.status_code for r in respostas])
print(respostas[-1].json(), respostas[-1].headers["retry-after"])
    Saída
    [200, 200, 200, 200, 200, 429, 429]
{'detail': 'Muitas requisições'} 60

    As cinco primeiras passaram, e da sexta em diante vieram 429. A ordem dos decoradores importa: @app.get por fora, @limiter.limit por dentro.

    Os algoritmos por trás

    Há três formas comuns de contar, e elas se comportam de modos diferentes nas bordas:

    AlgoritmoComo funcionaCaracterística
    Janela fixaConta por minuto "de relógio" e zeraSimples, mas permite uma rajada dupla na virada da janela
    Janela deslizanteConta os últimos 60 segundos a partir de agoraMais justo, guarda mais estado
    Balde de fichas (token bucket)Um balde recebe fichas a uma taxa fixa, e cada requisição gasta umaPermite pequenas rajadas e limita a média

    O balde de fichas é o mais usado, e cabe em poucas linhas. Como no cache, o relógio é um parâmetro, para o teste não precisar esperar:

    avancado/cap29_rate_limit.pylinhas 45 a 79
    import time
from typing import Callable


class BaldeDeFichas:
    def __init__(self, capacidade: float, fichas_por_segundo: float, relogio: Callable[[], float] = time.monotonic):
        self.capacidade = capacidade
        self.taxa = fichas_por_segundo
        self.relogio = relogio
        self.fichas = capacidade
        self.ultimo = relogio()

    def permitir(self, custo: float = 1.0) -> bool:
        agora = self.relogio()
        self.fichas = min(self.capacidade, self.fichas + (agora - self.ultimo) * self.taxa)
        self.ultimo = agora
        if self.fichas >= custo:
            self.fichas -= custo
            return True
        return False


class Relogio:
    def __init__(self):
        self.agora = 0.0

    def __call__(self) -> float:
        return self.agora


r = Relogio()
balde = BaldeDeFichas(capacidade=3, fichas_por_segundo=1, relogio=r)
print([balde.permitir() for _ in range(4)])
r.agora = 2.0
print([balde.permitir() for _ in range(3)])
    Saída
    [True, True, True, False]
[True, True, False]

    O balde começa com 3 fichas: três requisições passam, a quarta é negada. Depois de 2 segundos, ele recebeu 2 fichas (1 por segundo), e duas das três requisições seguintes passam.

    Um limite próprio, como dependência

    Com o balde, dá para escrever o limitador como uma dependência (capítulo 13), que vale só para as rotas que a pedirem, e que pode usar qualquer chave, não só o IP, como o usuário autenticado:

    avancado/cap29_rate_limit.pylinhas 84 a 105
    from fastapi import Depends, HTTPException

baldes: dict[str, BaldeDeFichas] = {}


def limitar_login(request: Request):
    chave = request.client.host if request.client else "desconhecido"
    balde = baldes.setdefault(chave, BaldeDeFichas(capacidade=3, fichas_por_segundo=3 / 60))
    if not balde.permitir():
        raise HTTPException(429, "Muitas tentativas de login", headers={"Retry-After": "20"})


app2 = FastAPI()


@app2.post("/login", dependencies=[Depends(limitar_login)])
def login():
    return {"ok": True}


c2 = TestClient(app2)
print([c2.post("/login").status_code for _ in range(5)])
    Saída
    [200, 200, 200, 429, 429]

    Limitar o login é uma das aplicações mais importantes: sem limite, um atacante testa milhares de senhas por minuto contra a rota do capítulo 20.

    Onde essa conta pode falhar

    Atrás de um proxy, o request.client.host é o endereço do proxy, e todos os clientes parecem um só. Configure o servidor para confiar nos cabeçalhos do proxy (--proxy-headers e --forwarded-allow-ips), e só deles: o cabeçalho X-Forwarded-For pode ser forjado por quem fala direto com o servidor. Com vários processos, cada um conta o seu próprio limite (um limite de 5 por minuto com 4 processos vira até 20). Para um limite global, o contador precisa estar em um armazenamento compartilhado, como o Redis, que o slowapi suporta por configuração.

    Exercício 1

    Três tentativas por minuto

    Crie um BaldeDeFichas de capacidade 3 e reposição de 1 ficha a cada 20 segundos, com um relógio falso. Confira que 3 tentativas passam, a 4ª é negada, e que depois de 20 segundos uma nova passa.