Webhook Security: HMAC Validation, Replay Prevention, and Secret Rotation

A webhook receiver is a public HTTP endpoint that processes events from a third-party service — a payment provider notifying you of a charge completion, a source control platform delivering a push event, a SaaS tool triggering an automation. The endpoint must be reachable from the internet. The problem is that “reachable from the internet” means reachable by anyone, including attackers who want to inject fabricated events, replay legitimate events, or probe your business logic with crafted payloads.

The solution is HMAC-based request authentication. Every major webhook provider — Stripe, GitHub, Shopify, Twilio, Slack — implements this pattern. Many applications consume these webhooks correctly because the provider’s SDK handles verification. The risk is webhook endpoints you control on both sides: internal services, custom integrations, and third-party platforms that let you define the signing mechanism yourself.

What Correct Webhook Authentication Requires

A correctly implemented webhook receiver needs four properties:

1. Signature authenticity: The request was signed with a secret only the sender and receiver know
2. Replay resistance: A valid request captured and replayed by an attacker is rejected
3. Timing-safe comparison: The signature comparison does not leak the correct value through timing
4. Rotatable secrets: The signing secret can be changed without downtime

HMAC-SHA256 Signature Validation

The standard signing mechanism is HMAC-SHA256. The sender computes HMAC-SHA256(secret, payload) and includes the result in a request header. The receiver recomputes the same HMAC over the received payload using the shared secret and compares the two values.

Python Implementation

import hmac
import hashlib
import time
from functools import wraps
from flask import Flask, request, abort

app = Flask(__name__)

WEBHOOK_SECRET = b"your-webhook-signing-secret"
TIMESTAMP_TOLERANCE_SECONDS = 300  # 5 minutes

def verify_webhook(secret: bytes):
    """Decorator that enforces HMAC signature and replay protection."""
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            # 1. Extract timestamp and signature headers
            timestamp = request.headers.get("X-Webhook-Timestamp")
            signature = request.headers.get("X-Webhook-Signature")
            
            if not timestamp or not signature:
                abort(401, "Missing signature headers")
            
            # 2. Replay protection: reject stale timestamps
            try:
                request_time = int(timestamp)
            except ValueError:
                abort(400, "Invalid timestamp format")
            
            current_time = int(time.time())
            if abs(current_time - request_time) > TIMESTAMP_TOLERANCE_SECONDS:
                abort(401, "Request timestamp outside tolerance window")
            
            # 3. Reconstruct the signed payload (timestamp + body)
            payload = request.get_data()
            signed_content = f"{timestamp}.".encode() + payload
            
            # 4. Compute expected HMAC
            expected_sig = hmac.new(secret, signed_content, hashlib.sha256).hexdigest()
            
            # 5. Timing-safe comparison — NEVER use == for signature comparison
            if not hmac.compare_digest(f"sha256={expected_sig}", signature):
                abort(401, "Signature verification failed")
            
            return f(*args, **kwargs)
        return wrapper
    return decorator


@app.route("/webhook/payment", methods=["POST"])
@verify_webhook(WEBHOOK_SECRET)
def handle_payment_event():
    event = request.get_json()
    # Process event — at this point the request is authenticated
    return "", 200

The signed_content = f"{timestamp}.".encode() + payload construction mirrors how Stripe and GitHub format their signed content. Including the timestamp in the signed payload is what provides replay protection: an attacker who captures a valid request cannot replay it after the tolerance window closes, even though they have both the timestamp and signature. Changing the timestamp to make the request appear fresh would invalidate the signature.

Go Implementation

package main

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "fmt"
    "io"
    "math"
    "net/http"
    "strconv"
    "strings"
    "time"
)

const (
    webhookSecret        = "your-webhook-signing-secret"
    timestampTolerance   = 5 * time.Minute
)

func verifyWebhook(secret []byte, next http.HandlerFunc) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        timestamp := r.Header.Get("X-Webhook-Timestamp")
        signature := r.Header.Get("X-Webhook-Signature")

        if timestamp == "" || signature == "" {
            http.Error(w, "Missing signature headers", http.StatusUnauthorized)
            return
        }

        // Parse and validate timestamp
        ts, err := strconv.ParseInt(timestamp, 10, 64)
        if err != nil {
            http.Error(w, "Invalid timestamp", http.StatusBadRequest)
            return
        }

        age := time.Since(time.Unix(ts, 0))
        if math.Abs(float64(age)) > float64(timestampTolerance) {
            http.Error(w, "Timestamp outside tolerance window", http.StatusUnauthorized)
            return
        }

        // Read body — must be done before computing HMAC
        body, err := io.ReadAll(r.Body)
        if err != nil {
            http.Error(w, "Failed to read body", http.StatusBadRequest)
            return
        }

        // Reconstruct signed content
        signedContent := fmt.Sprintf("%s.", timestamp) + string(body)

        // Compute expected HMAC
        mac := hmac.New(sha256.New, secret)
        mac.Write([]byte(signedContent))
        expectedSig := "sha256=" + hex.EncodeToString(mac.Sum(nil))

        // Timing-safe comparison
        if !hmac.Equal([]byte(expectedSig), []byte(signature)) {
            http.Error(w, "Signature verification failed", http.StatusUnauthorized)
            return
        }

        next(w, r)
    }
}

func handlePaymentEvent(w http.ResponseWriter, r *http.Request) {
    // Process authenticated webhook event
    w.WriteHeader(http.StatusOK)
}

func main() {
    secret := []byte(webhookSecret)
    http.HandleFunc("/webhook/payment", verifyWebhook(secret, handlePaymentEvent))
    http.ListenAndServe(":8080", nil)
}

Idempotency and Event ID Deduplication

Replay protection via timestamp tolerance handles the case where an attacker captures and resends a request. But legitimate infrastructure can also deliver the same webhook event more than once — retries on failure, duplicate delivery due to network issues, or provider-side retry logic.

The correct pattern is to track event IDs and reject duplicates:

import redis

redis_client = redis.Redis(host="localhost", port=6379, db=0)
EVENT_DEDUP_TTL = 86400  # 24 hours

def deduplicate_event(event_id: str) -> bool:
    """Returns True if this is a new event, False if already processed."""
    key = f"webhook:seen:{event_id}"
    # SET NX (only set if not exists) is atomic — safe under concurrent requests
    result = redis_client.set(key, "1", nx=True, ex=EVENT_DEDUP_TTL)
    return result is True  # True = key was set (new event), None = key existed (duplicate)

@app.route("/webhook/payment", methods=["POST"])
@verify_webhook(WEBHOOK_SECRET)
def handle_payment_event():
    event = request.get_json()
    event_id = event.get("id")
    
    if not event_id:
        abort(400, "Missing event ID")
    
    if not deduplicate_event(event_id):
        # Already processed — return 200 to stop provider retries
        return "", 200
    
    # Process event exactly once
    process_payment_event(event)
    return "", 200

Return HTTP 200 for duplicate events. Returning a 4xx response will typically cause the sending service to retry indefinitely, creating more duplicates.

Secret Rotation Without Downtime

Webhook secrets need to be rotatable. When a secret is compromised or a regular rotation policy requires a change, you cannot immediately invalidate the old secret without potentially rejecting legitimate in-flight requests from the sender that haven’t yet received the new secret.

The solution is a brief dual-validation window:

import os

class WebhookSecretManager:
    """Supports zero-downtime secret rotation with dual validation."""
    
    def __init__(self):
        self.current_secret = os.environ["WEBHOOK_SECRET_CURRENT"].encode()
        self.previous_secret = os.environ.get("WEBHOOK_SECRET_PREVIOUS", "").encode()
    
    def verify(self, timestamp: str, payload: bytes, signature: str) -> bool:
        signed_content = f"{timestamp}.".encode() + payload
        
        for secret in [self.current_secret, self.previous_secret]:
            if not secret:
                continue
            expected = "sha256=" + hmac.new(
                secret, signed_content, hashlib.sha256
            ).hexdigest()
            if hmac.compare_digest(expected, signature):
                return True
        
        return False

Rotation procedure:

1. Generate new secret
2. Set WEBHOOK_SECRET_CURRENT to new secret, move old value to WEBHOOK_SECRET_PREVIOUS
3. Update the sending service with the new secret
4. After the sender confirms it is using the new secret, clear WEBHOOK_SECRET_PREVIOUS

The dual-validation window lasts only as long as the cutover period — typically minutes to hours depending on how quickly the sender propagates the new secret.

Transport and Network Controls

HMAC signature validation is your primary control. These additional measures apply defense in depth:

Enforce HTTPS only. Redirect HTTP webhook delivery to HTTPS. HMAC protects authenticity but not confidentiality — event payloads delivered over HTTP are readable to network observers.

IP allowlisting as a secondary control. Most webhook providers publish static IP ranges. GitHub, Stripe, and Twilio all publish their webhook IP ranges via API. An IP allowlist rule upstream (WAF, load balancer, or application) reduces the population of requests that reach your HMAC validation logic. This is a secondary control — IP ranges change and providers can add IPs without notice, so the HMAC check must remain your primary authentication mechanism.

Body size limits. Webhook payloads should be small (kilobytes, not megabytes). Set an explicit body size limit at the application layer or reverse proxy (e.g., client_max_body_size 1m in nginx) to prevent request flooding via oversized payloads.

Rate limiting. Even authenticated webhook sources should not be able to trigger unlimited event processing. Rate limit by sender IP or event type at the application or WAF layer.

Common Implementation Mistakes

Using == for signature comparison is the most dangerous mistake. Python’s == on strings short-circuits on the first differing character, which creates a timing side channel that allows an attacker to brute-force the correct signature byte by byte. Always use hmac.compare_digest() in Python or hmac.Equal() in Go.

Skipping the timestamp check removes replay protection entirely. The HMAC alone proves the request was signed by someone with the secret but does not prove it was signed recently.

Putting the raw secret in code instead of environment variables creates a secret management problem: the secret ends up in source control, build artifacts, and logs. Load secrets from environment variables or a secrets manager at application startup.

Logging the full request body when signature validation fails can expose event payloads from legitimate retries in your log storage. Log the source IP, timestamp, and failure reason without the payload body on validation failures.