Endpoint autentikasi merupakan target utama serangan brute-force dan distributed credential stuffing. Mengandalkan proteksi bawaan Django saja tidak cukup karena framework tidak menyertakan mekanisme pembatasan laju permintaan (rate limiting) terpasang pada layer autentikasi default. Solusi yang umum diambil adalah menginstal package pihak ketiga seperti django-ratelimit. Namun, menambahkan pustaka eksternal sering kali berlebihan jika kebutuhan proteksi dapat dipenuhi secara optimal menggunakan core Cache API bawaan Django.

Arsitektur Dual-Key: Mitigasi Brute Force dan Credential Stuffing

Membatasi request hanya berdasarkan alamat IP klien tidak memadai. Penyerang dengan botnet terdistribusi dapat merotasi ribuan alamat IP per menit untuk melakukan credential stuffing pada satu akun. Sebaliknya, membatasi request hanya berdasarkan target akun (username/email) membuka celah serangan Denial of Service (DoS), di mana penyerang sengaja mengunci akun pengguna sah dengan mengirimkan password acak secara berulang.

Pendekatan yang benar adalah menerapkan mekanisme dual-key secara independen:

  • Key IP (rl:ip:<ip_address>): Membatasi volume percobaan login total dari satu IP tertentu dalam jendela waktu yang ditentukan. Menahan serangan brute-force agresif dari sumber tunggal.
  • Key Identifier Akun (rl:user:<normalized_identifier>): Membatasi percobaan login terhadap target akun tertentu terlepas dari mana IP asalnya. Menahan serangan distributed credential stuffing.

Konfigurasi Backend Redis pada Django

Gunakan Redis sebagai cache backend terdistribusi untuk menjamin atomisitas operasi counter di lingkungan multi-worker (seperti Gunicorn atau Uvicorn). Mulai Django 4.0, backend Redis didukung secara native tanpa memerlukan library eksternal.

Tambahkan konfigurasi berikut pada berkas settings.py:

CACHES = {
    "default": {
        "BACKEND": "django.core.cache.backends.redis.RedisCache",
        "LOCATION": "redis://127.0.0.1:6379/1",
        "OPTIONS": {
            "pool_class": "redis.ConnectionPool",
        },
        "KEY_PREFIX": "auth_cache",
        "TIMEOUT": 300,
    }
}

# Parameter batas toleransi autentikasi
AUTH_RATE_LIMIT = {
    "IP_THRESHOLD": 10,       # Maksimum 10 percobaan per IP
    "IP_WINDOW": 300,         # Dalam 5 menit (300 detik)
    "USER_THRESHOLD": 5,      # Maksimum 5 percobaan per username
    "USER_WINDOW": 900,       # Dalam 15 menit (900 detik)
    "FAIL_CLOSED": False,     # Fail-open jika backend cache down
}

Implementasi Decorator Rate Limiting Menggunakan Cache API

Kunci dari implementasi rate limiter yang aman dari kondisi balapan (race condition) adalah operasi atomik cache.add() dan cache.incr(). Metode cache.add() hanya akan menulis data jika key belum ada di dalam cache, sekaligus menetapkan Time-to-Live (TTL). Jika key sudah ada, inkrementasi dilakukan secara atomik via cache.incr().

Buat decorator di dalam modul security/ratelimit.py:

import logging
from functools import wraps
from django.conf import settings
from django.core.cache import cache
from django.http import JsonResponse

logger = logging.getLogger(__name__)

def get_client_ip(request):
    x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR")
    if x_forwarded_for:
        return x_forwarded_for.split(",")[0].strip()
    return request.META.get("REMOTE_ADDR", "")

def _hit_rate_limit(key: str, limit: int, window: int) -> tuple[bool, int]:
    """
    Mencatat hit secara atomik dan mengembalikan status terblokir serta estimasi TTL.
    ponytail: Penghitungan TTL statis window. Upgrade ke cache.ttl() bila menggunakan django-redis.
    """
    # cache.add menetapkan TTL hanya pada inisiasi pertama
    is_new = cache.add(key, 1, timeout=window)
    if is_new:
        return False, window

    try:
        current_hits = cache.incr(key)
    except ValueError:
        # Key kedaluwarsa tepat di antara evaluasi add dan incr
        cache.set(key, 1, timeout=window)
        current_hits = 1

    is_blocked = current_hits > limit
    return is_blocked, window

def rate_limit_login(view_func):
    @wraps(view_func)
    def _wrapped_view(request, *args, **kwargs):
        if request.method != "POST":
            return view_func(request, *args, **kwargs)

        cfg = settings.AUTH_RATE_LIMIT
        client_ip = get_client_ip(request)
        identifier = request.POST.get("username", "").strip().lower()

        ip_key = f"rl:ip:{client_ip}"
        user_key = f"rl:user:{identifier}" if identifier else None

        try:
            # Evaluasi limit IP
            ip_blocked, ip_ttl = _hit_rate_limit(ip_key, cfg["IP_THRESHOLD"], cfg["IP_WINDOW"])
            if ip_blocked:
                response = JsonResponse(
                    {"error": "Terlalu banyak percobaan login dari alamat IP ini."},
                    status=429
                )
                response["Retry-After"] = str(ip_ttl)
                return response

            # Evaluasi limit Akun
            if user_key:
                user_blocked, user_ttl = _hit_rate_limit(user_key, cfg["USER_THRESHOLD"], cfg["USER_WINDOW"])
                if user_blocked:
                    response = JsonResponse(
                        {"error": "Terlalu banyak percobaan masuk untuk akun ini. Coba lagi nanti."},
                        status=429
                    )
                    response["Retry-After"] = str(user_ttl)
                    return response

        except Exception as exc:
            logger.error("Kegagalan sistem cache pada rate limiter: %s", exc, exc_info=True)
            if cfg.get("FAIL_CLOSED", False):
                return JsonResponse(
                    {"error": "Layanan autentikasi sementara tidak tersedia."},
                    status=503
                )
            # Fail-open: biarkan request masuk jika cache mati demi ketersediaan layanan

        return view_func(request, *args, **kwargs)

    return _wrapped_view

Integrasi pada View Autentikasi

Penerapan decorator pada function-based view atau class-based view dilakukan secara langsung pada endpoint penerima kredensial:

from django.contrib.auth import authenticate, login
from django.http import JsonResponse
from django.views.decorators.http import require_POST
from .security.ratelimit import rate_limit_login

@require_POST
@rate_limit_login
def login_view(request):
    username = request.POST.get("username")
    password = request.POST.get("password")
    
    user = authenticate(request, username=username, password=password)
    if user is not None:
        login(request, user)
        # Hapus riwayat rate limiting jika otentikasi berhasil
        client_ip = request.META.get("REMOTE_ADDR", "")
        cache.delete(f"rl:user:{username.strip().lower()}")
        cache.delete(f"rl:ip:{client_ip}")
        return JsonResponse({"message": "Autentikasi berhasil."})

    return JsonResponse({"error": "Kredensial tidak valid."}, status=401)
Catatan: Menghapus counter via cache.delete() saat login berhasil adalah langkah krusial untuk mencegah pemblokiran pengguna sah yang sebelumnya salah memasukkan password beberapa kali.

Pertimbangan Arsitektur: Fail-Open vs Fail-Closed

Ketika node Redis mengalami gangguan (down, kehabisan memori, atau terputus secara jaringan), sistem harus menentukan strategi mitigasi:

  • Fail-Open (Ketersediaan Diutamakan): Error dari backend cache ditangkap dan dicatat ke log, request login diizinkan masuk ke database. Risiko: endpoint rentan terhadap serangan brute force saat Redis mati. Cocok untuk aplikasi consumer/SaaS umum.
  • Fail-Closed (Keamanan Diutamakan): Jika backend cache tidak merespons, sistem langsung mengembalikan status HTTP 503 Service Unavailable. Tidak ada autentikasi yang dapat diproses tanpa rate limiter aktif. Cocok untuk perbankan, portal pemerintahan, atau sistem berisiko finansial tinggi.

Pengujian Unit (Unit Testing)

Unit test berikut memvalidasi threshold limit, kemunculan header Retry-After, dan pemulihan kunci setelah cache dibersihkan:

from django.test import TestCase, RequestFactory
from django.core.cache import cache
from django.http import JsonResponse
from .views import login_view

class DynamicRateLimitTestCase(TestCase):
    def setUp(self):
        cache.clear()
        self.factory = RequestFactory()
        self.endpoint = "/api/login/"

    def tearDown(self):
        cache.clear()

    def test_threshold_enforcement_returns_429(self):
        payload = {"username": "victim_user", "password": "wrong_pass"}
        
        # Eksekusi 5 kali gagal sesuai batas USER_THRESHOLD
        for _ in range(5):
            request = self.factory.post(self.endpoint, payload, REMOTE_ADDR="192.168.1.50")
            response = login_view(request)
            self.assertEqual(response.status_code, 401)

        # Percobaan ke-6 harus ditolak oleh rate limiter (HTTP 429)
        blocked_request = self.factory.post(self.endpoint, payload, REMOTE_ADDR="192.168.1.50")
        response = login_view(blocked_request)

        self.assertEqual(response.status_code, 429)
        self.assertIn("Retry-After", response.headers)
        self.assertEqual(response.headers["Retry-After"], "900")

    def test_ttl_reset_allows_requests_again(self):
        payload = {"username": "reset_user", "password": "wrong_pass"}
        
        for _ in range(5):
            request = self.factory.post(self.endpoint, payload, REMOTE_ADDR="10.0.0.1")
            login_view(request)

        # Pastikan akun terblokir
        request = self.factory.post(self.endpoint, payload, REMOTE_ADDR="10.0.0.1")
        self.assertEqual(login_view(request).status_code, 429)

        # Simulasikan kedaluwarsa TTL dengan membersihkan cache
        cache.clear()

        # Request baru harus kembali diizinkan masuk ke pengecekan autentikasi
        new_request = self.factory.post(self.endpoint, payload, REMOTE_ADDR="10.0.0.1")
        response = login_view(new_request)
        self.assertEqual(response.status_code, 401)

Implementasi di atas mengamankan endpoint autentikasi dengan overhead komputasi minimal, menjaga struktur dependensi proyek tetap bersih, serta memberikan respons standar sesuai spesifikasi HTTP RFC 6585.