Kodeloka
  1. Kodeloka
  2. Kelas FastAPI
  3. Materi

Advanced & Deploy ยท Modul 18 dari 24

Rate Limiting & Perlindungan API

Membatasi laju permintaan dengan algoritma sliding window berbasis Redis, menetapkan kuota per pengguna, serta menambahkan header standar dan security header.

๐ŸŽฏ Poin-Poin Utama Modul Ini:
  • 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

rate_limit.py
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
๐Ÿ’ก Kenapa harus skrip Lua

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

middleware.py
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
๐Ÿšซ X-Forwarded-For bisa dipalsukan

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
}
โš ๏ธ Endpoint login butuh perlakuan khusus

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

security.py
@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"],
)
๐Ÿšซ Kombinasi berbahaya pada CORS

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.