Sécurité des webhooks : guide complet 2026

import hmac
import hashlib
import time
import json
import secrets
import requests
 
# Secret unique par receiver (généré à la création du webhook endpoint)
RECEIVER_SECRETS = {
    "https://partner1.example.test/webhooks/payment": "whsec_4eC39HqLyjWDarjtT1zdp7dc",
    "https://partner2.example.test/api/stripe-events": "whsec_7tF22MnHpTXBwxiyL2xfo5kz",
}
 
def send_webhook(receiver_url: str, event_type: str, data: dict):
    """Envoie un webhook signé HMAC."""
    secret = RECEIVER_SECRETS.get(receiver_url)
    if not secret:
        raise ValueError(f"No secret configured for {receiver_url}")
    
    # Construction du payload avec event_id et timestamp
    event = {
        "id": f"evt_{secrets.token_urlsafe(16)}",   # event_id unique
        "type": event_type,
        "created": int(time.time()),                  # timestamp Unix
        "data": data,
    }
    payload = json.dumps(event, separators=(",", ":"))
    timestamp = event["created"]
    
    # Signature HMAC SHA-256 sur "timestamp.payload"
    signed_payload = f"{timestamp}.{payload}"
    signature = hmac.new(
        secret.encode(),
        signed_payload.encode(),
        hashlib.sha256
    ).hexdigest()
    
    headers = {
        "Content-Type": "application/json",
        "X-Webhook-Id": event["id"],
        "X-Webhook-Timestamp": str(timestamp),
        "X-Webhook-Signature": f"sha256={signature}",
        "User-Agent": "MyService-Webhooks/1.0",
    }
    
    return requests.post(receiver_url, data=payload, headers=headers, timeout=10)

import time
from typing import Optional
 
# Délais de retry : 30s, 5min, 30min, 2h, 12h
RETRY_DELAYS_SECONDS = [30, 300, 1800, 7200, 43200]
 
def deliver_webhook_with_retry(
    receiver_url: str,
    event: dict,
    max_attempts: int = 5,
):
    """Délivre un webhook avec retry exponential backoff."""
    for attempt in range(max_attempts):
        try:
            response = send_webhook(receiver_url, event["type"], event["data"])
            
            if 200 <= response.status_code < 300:
                # Succès
                log_delivery_success(event["id"], receiver_url, attempt + 1)
                return True
            elif 400 <= response.status_code < 500:
                # Erreur permanente (4xx) : pas de retry
                log_delivery_failure(event["id"], receiver_url, response.status_code, "4xx_no_retry")
                return False
            else:
                # Erreur temporaire (5xx) : retry
                log_delivery_attempt(event["id"], receiver_url, response.status_code, attempt + 1)
        except requests.exceptions.RequestException as e:
            log_delivery_attempt(event["id"], receiver_url, "network_error", attempt + 1)
        
        # Attente avant retry suivant
        if attempt < max_attempts - 1:
            time.sleep(RETRY_DELAYS_SECONDS[attempt])
    
    # Tous les retries échoués
    log_delivery_failure(event["id"], receiver_url, "max_retries_exceeded", "alert")
    return False

import hmac
import hashlib
import time
from fastapi import FastAPI, Request, HTTPException, Header
 
app = FastAPI()
WEBHOOK_SECRET = "whsec_4eC39HqLyjWDarjtT1zdp7dc"  # depuis vault, pas hardcodé
 
def verify_webhook_signature(
    payload: bytes,
    timestamp: str,
    signature: str,
    secret: str,
    max_age_seconds: int = 300,
) -> bool:
    """Vérifie signature HMAC avec anti-replay."""
    # 1. Anti-replay : refuser timestamp trop ancien
    try:
        ts = int(timestamp)
    except ValueError:
        return False
    
    if abs(time.time() - ts) > max_age_seconds:
        return False
    
    # 2. Reconstruction signature attendue
    signed_payload = f"{ts}.{payload.decode()}"
    expected_signature = hmac.new(
        secret.encode(),
        signed_payload.encode(),
        hashlib.sha256
    ).hexdigest()
    
    # 3. Comparaison constant-time (anti-timing attack)
    received_signature = signature.replace("sha256=", "")
    return hmac.compare_digest(expected_signature, received_signature)
 
@app.post("/webhooks/payment")
async def receive_webhook(
    request: Request,
    x_webhook_id: str = Header(...),
    x_webhook_timestamp: str = Header(...),
    x_webhook_signature: str = Header(...),
):
    payload = await request.body()
    
    # 1. Vérification signature + anti-replay timestamp
    if not verify_webhook_signature(
        payload, x_webhook_timestamp, x_webhook_signature, WEBHOOK_SECRET
    ):
        raise HTTPException(401, "Invalid signature")
    
    # 2. Déduplication via event_id
    if await is_already_processed(x_webhook_id):
        return {"status": "already_processed"}
    
    # 3. Traitement async (queue interne) pour répondre rapidement
    event = json.loads(payload)
    await enqueue_event_processing(event)
    
    # 4. Mark as processed pour déduplication future
    await mark_processed(x_webhook_id, ttl_seconds=86400 * 7)  # 7 jours
    
    # 5. Réponse 2xx rapide
    return {"status": "accepted"}

import redis
 
r = redis.Redis(host="redis.internal", decode_responses=True)
 
async def is_already_processed(event_id: str) -> bool:
    """Vérifie si un event_id a déjà été traité (dans les 7 derniers jours)."""
    key = f"webhook:processed:{event_id}"
    return await r.exists(key) > 0
 
async def mark_processed(event_id: str, ttl_seconds: int = 604800):
    """Marque un event_id comme traité avec TTL."""
    key = f"webhook:processed:{event_id}"
    await r.set(key, int(time.time()), ex=ttl_seconds)

# IDEMPOTENT : update conditionnel
def handle_payment_succeeded(event_data):
    payment_id = event_data["payment_id"]
    order = db.query(Order).filter_by(payment_id=payment_id).first()
    
    # Update conditionnel : ne fait rien si déjà 'paid'
    if order and order.status != "paid":
        order.status = "paid"
        order.paid_at = datetime.now(UTC)
        db.commit()
        send_confirmation_email(order)
 
# NON IDEMPOTENT : incrémenter un compteur
def handle_order_created(event_data):
    user = db.query(User).filter_by(id=event_data["user_id"]).first()
    user.orders_count += 1  # ← problème si event traité 2x
    db.commit()
 
# IDEMPOTENT : recalcul depuis la source de vérité
def handle_order_created_idempotent(event_data):
    user_id = event_data["user_id"]
    actual_count = db.query(Order).filter_by(user_id=user_id).count()
    db.query(User).filter_by(id=user_id).update({"orders_count": actual_count})
    db.commit()

# Configuration recommandée endpoint webhook receiver
@app.post(
    "/webhooks/{provider}",
    # Pas dans la doc OpenAPI publique (seul le sender connaît l'URL)
    include_in_schema=False,
    # Réponse minimale : ne pas exposer de stack trace
    response_class=JSONResponse,
)
async def receive_webhook(provider: str, request: Request):
    # ... validation et traitement
    pass
 
# Configuration nginx en frontage
# - HTTPS obligatoire
# - Rate limiting (refuse si > 100 req/min depuis IP unique)
# - body size limit (refuse si > 1 MB)
# - timeout strict (refuse après 10s)

# Stripe : header "Stripe-Signature" avec format "t=<timestamp>,v1=<signature>"
def verify_stripe_webhook(payload: bytes, signature_header: str, secret: str) -> bool:
    elements = {}
    for item in signature_header.split(","):
        key, value = item.split("=", 1)
        elements[key] = value
    
    timestamp = elements.get("t")
    signature_v1 = elements.get("v1")
    
    if not timestamp or not signature_v1:
        return False
    
    # Vérification timestamp (5 min max)
    if abs(time.time() - int(timestamp)) > 300:
        return False
    
    # Reconstruction signature
    signed_payload = f"{timestamp}.{payload.decode()}"
    expected = hmac.new(
        secret.encode(), signed_payload.encode(), hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(expected, signature_v1)
 
# Usage :
# Stripe envoie header :
# Stripe-Signature: t=1714118400,v1=abc123...,v0=oldformat

# GitHub : header "X-Hub-Signature-256" avec format "sha256=<signature>"
def verify_github_webhook(payload: bytes, signature_header: str, secret: str) -> bool:
    if not signature_header.startswith("sha256="):
        return False
    
    received_signature = signature_header[len("sha256="):]
    expected = hmac.new(
        secret.encode(), payload, hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(expected, received_signature)
 
# GitHub n'inclut pas timestamp dans signature (anti-replay au niveau event_id GUID)
# Dedup via header X-GitHub-Delivery (GUID unique par event)

# Slack : headers "X-Slack-Request-Timestamp" et "X-Slack-Signature"
def verify_slack_webhook(payload: bytes, timestamp: str, signature: str, secret: str) -> bool:
    # Anti-replay 5 minutes
    if abs(time.time() - int(timestamp)) > 300:
        return False
    
    # Format : v0=<signature>, signed_payload = "v0:timestamp:body"
    signed_payload = f"v0:{timestamp}:{payload.decode()}"
    expected = "v0=" + hmac.new(
        secret.encode(), signed_payload.encode(), hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(expected, signature)

# Standard Webhooks : 4 headers webhook-id, webhook-timestamp, webhook-signature
# Format signature : "v1,<base64-encoded-signature>"
import base64
 
def verify_standard_webhook(
    payload: bytes,
    msg_id: str,
    timestamp: str,
    signature_header: str,
    secret: str,
) -> bool:
    # Anti-replay
    if abs(time.time() - int(timestamp)) > 300:
        return False
    
    # Décodage secret base64 (format svix)
    if secret.startswith("whsec_"):
        secret_bytes = base64.b64decode(secret[len("whsec_"):])
    else:
        secret_bytes = secret.encode()
    
    # Signature : HMAC-SHA256 de "msg_id.timestamp.payload"
    signed_payload = f"{msg_id}.{timestamp}.{payload.decode()}"
    expected_sig = base64.b64encode(
        hmac.new(secret_bytes, signed_payload.encode(), hashlib.sha256).digest()
    ).decode()
    
    # Format header : "v1,sig1 v1,sig2" (peut contenir plusieurs signatures pour rotation)
    for sig_pair in signature_header.split(" "):
        version, sig = sig_pair.split(",", 1)
        if version == "v1" and hmac.compare_digest(expected_sig, sig):
            return True
    
    return False

1. Attaquant crée compte sur votre service.
2. Attaquant configure l'URL de webhook : http://169.254.169.254/latest/meta-data/iam/security-credentials/
3. Votre service tente d'envoyer un webhook vers cette URL.
4. La réponse contient les credentials IAM de l'instance EC2.
5. Si les logs deliveries sont visibles à l'attaquant (dashboard), exfiltration des credentials.

import socket
import ipaddress
from urllib.parse import urlparse
 
PRIVATE_NETWORKS = [
    ipaddress.ip_network("10.0.0.0/8"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.168.0.0/16"),
    ipaddress.ip_network("127.0.0.0/8"),
    ipaddress.ip_network("169.254.0.0/16"),  # link-local, IMDS
    ipaddress.ip_network("0.0.0.0/8"),
    ipaddress.ip_network("::1/128"),         # IPv6 localhost
    ipaddress.ip_network("fc00::/7"),        # IPv6 ULA
    ipaddress.ip_network("fe80::/10"),       # IPv6 link-local
]
 
def is_safe_webhook_url(url: str) -> bool:
    """Valide qu'une URL webhook ne pointe pas vers infrastructure interne."""
    parsed = urlparse(url)
    
    # 1. HTTPS uniquement
    if parsed.scheme != "https":
        return False
    
    # 2. Hostname valide
    if not parsed.hostname:
        return False
    
    # 3. Résoudre en IP et vérifier
    try:
        addresses = socket.getaddrinfo(parsed.hostname, None)
    except socket.gaierror:
        return False
    
    for family, _, _, _, sockaddr in addresses:
        ip_str = sockaddr[0]
        try:
            ip = ipaddress.ip_address(ip_str)
        except ValueError:
            continue
        
        # Refuser IPs privées
        for network in PRIVATE_NETWORKS:
            if ip in network:
                return False
        
        # Refuser IPs réservées / multicast
        if ip.is_reserved or ip.is_multicast or ip.is_loopback:
            return False
    
    return True
 
# Usage à la configuration du webhook par l'utilisateur
def configure_webhook(url: str, user: User):
    if not is_safe_webhook_url(url):
        raise HTTPException(400, "Webhook URL must point to a public HTTPS endpoint")
    # ... save webhook configuration

# MAUVAIS : comparaison directe de strings
if expected_signature == received_signature:  # vulnerable timing attack
    process()
 
# CORRECT : constant-time
if hmac.compare_digest(expected_signature, received_signature):
    process()

SENSITIVE_FIELDS_PATHS = ["data.email", "data.phone", "data.card.number"]
 
def mask_sensitive(payload: dict) -> dict:
    masked = copy.deepcopy(payload)
    for path in SENSITIVE_FIELDS_PATHS:
        keys = path.split(".")
        target = masked
        for key in keys[:-1]:
            target = target.get(key, {})
        if keys[-1] in target:
            target[keys[-1]] = "****"
    return masked
 
logger.info("Webhook received", extra={"event": mask_sensitive(payload)})