Anatomi Masalah: Retry Agresif dan Serangan Replay

Penyedia layanan pihak ketiga (seperti Stripe, Xendit, atau GitHub) mengandalkan garansi pengiriman at-least-once. Jika server Anda terlambat merespons akibat beban kerja tinggi atau latensi jaringan melebihi batas waktu (timeout), penyedia akan otomatis mengirim ulang payload yang sama. Tanpa mitigasi, lonjakan request duplikat ini mengeksekusi logika bisnis berulang kali, seperti mutasi saldo ganda atau pembuatan invoice duplikat.

Selain masalah dobel eksekusi, webhook tanpa validasi waktu rentan terhadap replay attack, di mana penyerang yang berhasil mencegat payload absah mengirimkannya kembali ke server Anda untuk memicu efek samping yang sama. Masalah keamanan lainnya adalah timing attack pada verifikasi signature jika string pembanding diperiksa menggunakan operator kesetaraan standar (==).

Verifikasi Signature Bebas Timing Attack

Verifikasi signature harus selalu dihitung langsung dari request.body (raw bytes). Melakukan parsing JSON terlebih dahulu sebelum verifikasi merupakan celah fatal karena proses serialisasi ulang dapat mengubah spasi, urutan key, atau format angka, yang menyebabkan hash tidak cocok. Gunakan hmac.compare_digest untuk memastikan waktu evaluasi konstan guna mencegah eksploitasi berbasis waktu.

import hashlib
import hmac
import time
from django.conf import settings
from django.http import HttpResponseBadRequest, HttpResponseForbidden

WEBHOOK_TOLERANCE_SECONDS = 300  # 5 menit

def verify_webhook_signature(request, secret_key: str) -> bool:
    signature = request.headers.get("X-Signature")
    timestamp = request.headers.get("X-Timestamp")

    if not signature or not timestamp:
        return False

    # Mitigasi Replay Attack: Tolak timestamp yang melampaui ambang batas toleransi
    try:
        event_timestamp = int(timestamp)
        if abs(time.time() - event_timestamp) > WEBHOOK_TOLERANCE_SECONDS:
            return False
    except ValueError:
        return False

    # Susun payload bertanda tangan dari raw bytes
    signed_payload = f"{timestamp}.".encode("utf-8") + request.body
    expected_signature = hmac.new(
        secret_key.encode("utf-8"),
        signed_payload,
        hashlib.sha256
    ).hexdigest()

    # Mitigasi Timing Attack: Waktu perbandingan konstan
    return hmac.compare_digest(expected_signature, signature)

Pagar Idempotensi: Mengapa Pengecekan Level Aplikasi Gagal

Pengecekan sederhana seperti if WebhookEvent.objects.filter(event_id=id).exists(): gagal total saat menghadapi dua request identik yang masuk secara paralel (race condition). Kedua thread request akan mengevaluasi kondisi tersebut sebagai False secara bersamaan sebelum salah satu dari mereka sempat menyimpan record (Time-of-Check to Time-of-Use / TOCTOU).

Satu-satunya pertahanan yang konsisten adalah isolasi di tingkat basis data menggunakan UniqueConstraint. Basis data akan menjamin keunikan secara atomik melalui index b-tree.

Skema Model Idempotensi

Definisikan model untuk melacak status pemrosesan webhook. Terapkan UniqueConstraint gabungan antara kolom identitas penyedia (provider) dan identitas event unik dari penyedia (event_id).

from django.db import models

class WebhookEvent(models.Model):
    class Status(models.TextChoices):
        PENDING = "PENDING", "Pending"
        PROCESSED = "PROCESSED", "Processed"
        FAILED = "FAILED", "Failed"

    provider = models.CharField(max_length=50)
    event_id = models.CharField(max_length=255)
    payload = models.JSONField()
    status = models.CharField(
        max_length=20,
        choices=Status.choices,
        default=Status.PENDING
    )
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        constraints = [
            models.UniqueConstraint(
                fields=["provider", "event_id"],
                name="unique_provider_webhook_event"
            )
        ]
        indexes = [
            models.Index(fields=["provider", "event_id"]),
        ]

    def __str__(self):
        return f"{self.provider}:{self.event_id} - {self.status}"

Implementasi View Berbasis Django Murni

View harus menangani IntegrityError secara eksplisit. Saat terjadi benturan duplikat konkurensi, tangkap error tersebut dan segera kembalikan status HTTP 200 OK. Jika Anda mengembalikan status HTTP 4xx atau 5xx, sistem pihak ketiga akan menganggap pengiriman gagal dan terus mengulang request tersebut.

import json
from django.http import JsonResponse, HttpResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from django.db import IntegrityError, transaction
from .models import WebhookEvent
from .security import verify_webhook_signature

WEBHOOK_SECRET = "whsec_live_example_secret_key_12345"

@csrf_exempt
@require_POST
def payment_webhook_view(request):
    if not verify_webhook_signature(request, WEBHOOK_SECRET):
        return JsonResponse({"error": "Invalid signature or expired timestamp"}, status=403)

    try:
        payload = json.loads(request.body.decode("utf-8"))
        event_id = payload.get("id")
        if not event_id:
            return JsonResponse({"error": "Missing event ID"}, status=400)
    except json.JSONDecodeError:
        return JsonResponse({"error": "Invalid JSON"}, status=400)

    # ponytail: eksekusi sinkron langsung dalam view. Alihkan ke Celery/worker jika proses bisnis > 200ms.
    try:
        with transaction.atomic():
            event = WebhookEvent.objects.create(
                provider="payment_gateway",
                event_id=event_id,
                payload=payload,
                status=WebhookEvent.Status.PENDING
            )
            
            # Logika bisnis inti dijalankan di sini
            _process_payment(payload)
            
            event.status = WebhookEvent.Status.PROCESSED
            event.save(update_fields=["status", "updated_at"])
            
    except IntegrityError:
        # Request duplikat terdeteksi oleh database constraint. Bypass pemrosesan ulang.
        return HttpResponse("Event already received", status=200)

    return HttpResponse("Event processed successfully", status=200)

def _process_payment(payload: dict):
    # Eksekusi mutasi bisnis riil
    pass

Automated Test Skenario Duplikasi dan Replay

Gunakan Django TestCase untuk menguji validasi signature, pencegahan replay attack, dan perilaku idempotensi ketika request duplikat diterima secara berturut-turut.

import hashlib
import hmac
import json
import time
from django.test import TestCase, Client
from django.urls import reverse
from .models import WebhookEvent

class WebhookIdempotencyTests(TestCase):
    def setUp(self):
        self.client = Client()
        self.secret = "whsec_live_example_secret_key_12345"
        self.url = reverse("payment_webhook")

    def _generate_headers(self, body_bytes: bytes, timestamp: int):
        signed_payload = f"{timestamp}.".encode("utf-8") + body_bytes
        signature = hmac.new(
            self.secret.encode("utf-8"),
            signed_payload,
            hashlib.sha256
        ).hexdigest()
        return {
            "HTTP_X_SIGNATURE": signature,
            "HTTP_X_TIMESTAMP": str(timestamp),
        }

    def test_idempotent_retry_returns_200_without_duplicate_row(self):
        payload = {"id": "evt_test_123456", "type": "payment.succeeded"}
        body = json.dumps(payload).encode("utf-8")
        headers = self._generate_headers(body, int(time.time()))

        # Request pertama: Eksekusi normal
        resp1 = self.client.post(self.url, data=body, content_type="application/json", **headers)
        self.assertEqual(resp1.status_code, 200)
        self.assertEqual(WebhookEvent.objects.filter(event_id="evt_test_123456").count(), 1)

        # Request kedua (retry): Harus tetap 200 OK via IntegrityError bypass
        resp2 = self.client.post(self.url, data=body, content_type="application/json", **headers)
        self.assertEqual(resp2.status_code, 200)
        self.assertEqual(WebhookEvent.objects.filter(event_id="evt_test_123456").count(), 1)

    def test_rejects_expired_timestamp_replay_attack(self):
        payload = {"id": "evt_test_replay", "type": "payment.succeeded"}
        body = json.dumps(payload).encode("utf-8")
        expired_timestamp = int(time.time()) - 600  # 10 menit lalu
        headers = self._generate_headers(body, expired_timestamp)

        response = self.client.post(self.url, data=body, content_type="application/json", **headers)
        self.assertEqual(response.status_code, 403)
        self.assertEqual(WebhookEvent.objects.filter(event_id="evt_test_replay").count(), 0)
Catatan Operasional: Jika beban webhook sangat masif atau logika bisnis membutuhkan integrasi I/O lambat (seperti panggilan API pihak ketiga lain), geser pemrosesan bisnis ke background worker (misalnya Celery atau Redis Queue). Cukup simpan event berstatus PENDING di dalam view lalu balas HTTP 200 secara instan.