- Rate limiting di memori rusak begitu ada lebih dari satu worker
- Sliding window dengan Sorted Set Redis lebih akurat daripada fixed window
- Balas 429 disertai header Retry-After dan RateLimit-*
- Batasi berdasarkan identitas pengguna, bukan hanya alamat IP
๐ Kenapa Rate Limiting Wajib
- Mencegah penyalahgunaan dan pengerukan data otomatis.
- Melindungi dari serangan tebak-password secara brute force.
- Menjaga biaya terkendali bila API Anda memanggil layanan berbayar.
- Menjamin satu klien yang rakus tidak menghabiskan kapasitas untuk semua orang.
โ๏ธ Fixed Window vs Sliding Window
FIXED WINDOW -- batas 100 permintaan per menit
10:00:59 โ 100 permintaan โ
diizinkan
10:01:01 โ 100 permintaan โ
diizinkan
Dalam 2 detik, 200 permintaan lolos. โ Ini disebut masalah tepi jendela.
SLIDING WINDOW -- selalu melihat 60 detik terakhir dari saat ini
10:01:01 โ memeriksa 10:00:01โ10:01:01
sudah ada 100 permintaan โ ditolak
๐ป Sliding Window dengan Redis
import time, uuid
from fastapi import HTTPException, Request
SKRIP_BATAS = """
local kunci = KEYS[1]
local batas = tonumber(ARGV[1])
local jendela = tonumber(ARGV[2])
local sekarang = tonumber(ARGV[3])
local id = ARGV[4]
redis.call('ZREMRANGEBYSCORE', kunci, 0, sekarang - jendela * 1000)
local jumlah = redis.call('ZCARD', kunci)
if jumlah >= batas then
return {0, jumlah}
end
redis.call('ZADD', kunci, sekarang, id)
redis.call('EXPIRE', kunci, jendela)
return {1, jumlah + 1}
"""
async def periksa_batas(r, kunci: str, batas: int, jendela: int):
sekarang = int(time.time() * 1000)
izin, jumlah = await r.eval(
SKRIP_BATAS, 1, kunci, batas, jendela, sekarang, str(uuid.uuid4())
)
return bool(izin), jumlah
Rangkaian "hitung dulu, lalu putuskan, lalu catat" mengandung balapan bila dijalankan sebagai perintah terpisah: dua permintaan bersamaan bisa sama-sama membaca jumlah 99 dan sama-sama lolos. Skrip Lua dijalankan Redis sebagai satu unit tanpa disela, sehingga keputusannya selalu benar.
๐งฑ Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
class MiddlewareBatas(BaseHTTPMiddleware):
def __init__(self, app, batas=100, jendela=60):
super().__init__(app)
self.batas, self.jendela = batas, jendela
async def dispatch(self, request: Request, call_next):
if request.url.path in ("/healthz", "/readyz", "/metrics"):
return await call_next(request)
pengenal = self._pengenal(request)
kunci = f"batas:{pengenal}:{request.url.path}"
izin, jumlah = await periksa_batas(
request.app.state.redis, kunci, self.batas, self.jendela
)
if not izin:
return JSONResponse(
status_code=429,
content={"detail": "Terlalu banyak permintaan"},
headers={
"Retry-After": str(self.jendela),
"RateLimit-Limit": str(self.batas),
"RateLimit-Remaining": "0",
},
)
respons = await call_next(request)
respons.headers["RateLimit-Limit"] = str(self.batas)
respons.headers["RateLimit-Remaining"] = str(max(0, self.batas - jumlah))
return respons
def _pengenal(self, request: Request) -> str:
# Pengguna terautentikasi dibatasi per identitas, bukan per IP
if pengguna := getattr(request.state, "pengguna", None):
return f"pengguna:{pengguna.id}"
return f"ip:{self._ip_asli(request)}"
def _ip_asli(self, request: Request) -> str:
# HANYA percaya header ini bila Anda mengendalikan proxy di depannya
if xff := request.headers.get("x-forwarded-for"):
return xff.split(",")[0].strip()
return request.client.host
Bila aplikasi Anda terjangkau langsung dari internet, siapa pun bisa mengirim X-Forwarded-For: 1.2.3.4 palsu dan melewati pembatasan cukup dengan mengacak nilainya. Header ini hanya boleh dipercaya bila ada reverse proxy tepercaya di depan yang menimpa nilainya, dan aplikasi tidak bisa dijangkau langsung dari luar.
๐๏ธ Kuota Berjenjang
KUOTA = {
"anonim": (20, 60), # 20 per menit
"gratis": (100, 60),
"pro": (1000, 60),
"mitra": (10000, 60),
}
# Batas berbeda untuk endpoint yang mahal
BATAS_ENDPOINT = {
"/api/cari": (30, 60),
"/api/ekspor": (5, 3600), # 5 per jam
"/api/auth/login": (5, 300), # anti brute force
}
Batasi per kombinasi IP dan nama pengguna, bukan hanya salah satunya. Membatasi hanya per IP membuat penyerang di belakang banyak IP tetap bisa menyerang satu akun; membatasi hanya per nama pengguna memungkinkan penyerang mengunci akun orang lain dengan sengaja gagal login berulang kali.
๐ก๏ธ Security Header
@app.middleware("http")
async def header_keamanan(request: Request, call_next):
r = await call_next(request)
r.headers["X-Content-Type-Options"] = "nosniff"
r.headers["X-Frame-Options"] = "DENY"
r.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
r.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
r.headers["Content-Security-Policy"] = "default-src 'self'"
return r
# CORS -- JANGAN pakai allow_origins=["*"] bersama credentials
app.add_middleware(
CORSMiddleware,
allow_origins=["https://aplikasi.contoh.com"],
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["Authorization", "Content-Type"],
)
allow_origins=["*"] bersama allow_credentials=True adalah kesalahan konfigurasi yang serius. Spesifikasi CORS sebenarnya melarang kombinasi ini, dan sebagian implementasi menanganinya dengan cara yang membuka celah: situs mana pun bisa mengirim permintaan dengan cookie pengguna Anda dan membaca responsnya. Selalu daftarkan origin secara eksplisit ketika memakai credentials.