Hook Mesh
Log inStart Free
Back to Blog
Hook Mesh Team

Receive and Verify Webhooks in Python: A Complete Guide

Learn how to securely receive and verify webhooks in Python using Flask and FastAPI. Covers HMAC signature verification, timestamp validation, and common security pitfalls to avoid.

Receive and Verify Webhooks in Python: A Complete Guide

Receive and Verify Webhooks in Python

Receiving webhooks securely requires more than exposing an endpoint. This guide walks you through building secure receivers in Python using Flask and FastAPI, with signature verification, replay attack prevention, and common pitfall avoidance.

Why Verification Matters

Your webhook endpoint is a public URL. Anyone who discovers it can send requests pretending to be a legitimate service. Without verification, an attacker could send a fake "payment completed" event and grant themselves access to paid features, or trigger a "user deleted" event to corrupt your data.

Webhook signature verification solves this problem. The sender signs each payload with a shared secret, and your application verifies that signature before trusting the data. If the signature does not match, you reject the request.

Setting Up a Flask Webhook Endpoint

Install dependencies:

pip install flask
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhooks', methods=['POST'])
def handle_webhook():
    payload = request.get_json()

    # Process the webhook
    print(f"Received event: {payload.get('event')}")

    return jsonify({'status': 'received'}), 200

if __name__ == '__main__':
    app.run(port=3000)

This works, but it is dangerously insecure. Anyone can send requests to this endpoint. Let us add proper verification.

Setting Up a FastAPI Webhook Endpoint

FastAPI offers async support and automatic documentation:

pip install fastapi uvicorn
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()

@app.post('/webhooks')
async def handle_webhook(request: Request):
    payload = await request.json()

    # Process the webhook
    print(f"Received event: {payload.get('event')}")

    return JSONResponse({'status': 'received'}, status_code=200)

# Run with: uvicorn main:app --port 3000

Again, this accepts any request. Time to add security.

Accessing the Raw Request Body

Verify raw bytes, not parsed JSON—parsing alters whitespace/ordering and invalidates signatures.

Flask: Getting the Raw Body

@app.route('/webhooks', methods=['POST'])
def handle_webhook():
    # Get raw bytes BEFORE parsing JSON
    raw_body = request.get_data()

    # Later, parse for processing
    payload = request.get_json()

FastAPI: Getting the Raw Body

@app.post('/webhooks')
async def handle_webhook(request: Request):
    # Get raw bytes BEFORE parsing JSON
    raw_body = await request.body()

    # Later, parse for processing
    import json
    payload = json.loads(raw_body)

Order matters: read raw body, verify signature, then parse JSON.

HMAC Signature Verification

Here's how to implement HMAC-SHA256 verification in Python:

import hmac
import hashlib

def verify_signature(payload: bytes, signature: str, secret: str) -> bool:
    """
    Verify an HMAC-SHA256 webhook signature.

    Args:
        payload: Raw request body bytes
        signature: Signature from the webhook header
        secret: Your webhook signing secret

    Returns:
        True if signature is valid, False otherwise
    """
    expected_signature = hmac.new(
        secret.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()

    # CRITICAL: Use timing-safe comparison
    return hmac.compare_digest(signature, expected_signature)

Use hmac.compare_digest()—naive comparison reveals signatures byte-by-byte through timing differences. Timing-safe comparison always takes the same duration.

Hook Mesh Signature Verification

Hook Mesh format: t=timestamp,v1=signature with signed payload as timestamp.body.

Complete verification function:

import hmac
import hashlib
import time

def verify_hookmesh_signature(
    payload: bytes,
    signature_header: str,
    secret: str,
    tolerance_seconds: int = 300
) -> bool:
    """
    Verify a Hook Mesh webhook signature with timestamp validation.

    Args:
        payload: Raw request body bytes
        signature_header: Value of X-HookMesh-Signature header
        secret: Your Hook Mesh webhook secret
        tolerance_seconds: Maximum age of webhook (default 5 minutes)

    Returns:
        True if signature is valid and timestamp is within tolerance

    Raises:
        ValueError: If signature format is invalid or verification fails
    """
    # Parse the signature header
    elements = {}
    for element in signature_header.split(','):
        key, value = element.split('=', 1)
        elements[key] = value

    timestamp = elements.get('t')
    received_signature = elements.get('v1')

    if not timestamp or not received_signature:
        raise ValueError('Invalid signature format')

    # Validate timestamp to prevent replay attacks
    current_time = int(time.time())
    webhook_time = int(timestamp)

    if abs(current_time - webhook_time) > tolerance_seconds:
        raise ValueError('Webhook timestamp outside tolerance window')

    # Reconstruct the signed payload
    signed_payload = f"{timestamp}.".encode('utf-8') + payload

    # Compute expected signature
    expected_signature = hmac.new(
        secret.encode('utf-8'),
        signed_payload,
        hashlib.sha256
    ).hexdigest()

    # Timing-safe comparison
    if not hmac.compare_digest(received_signature, expected_signature):
        raise ValueError('Signature verification failed')

    return True

Complete Flask Implementation

import hmac
import hashlib
import time
import os
import json
from flask import Flask, request, jsonify

app = Flask(__name__)

WEBHOOK_SECRET = os.environ['WEBHOOK_SECRET']

def verify_hookmesh_signature(payload, signature_header, secret, tolerance=300):
    elements = dict(e.split('=', 1) for e in signature_header.split(','))
    timestamp = elements.get('t')
    received_sig = elements.get('v1')

    if not timestamp or not received_sig:
        raise ValueError('Invalid signature format')

    if abs(int(time.time()) - int(timestamp)) > tolerance:
        raise ValueError('Timestamp outside tolerance')

    signed_payload = f"{timestamp}.".encode() + payload
    expected_sig = hmac.new(
        secret.encode(), signed_payload, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(received_sig, expected_sig):
        raise ValueError('Invalid signature')

    return True

@app.route('/webhooks', methods=['POST'])
def handle_webhook():
    # Step 1: Get raw body BEFORE any parsing
    raw_body = request.get_data()

    # Step 2: Extract signature header
    signature = request.headers.get('X-HookMesh-Signature')
    if not signature:
        return jsonify({'error': 'Missing signature'}), 401

    # Step 3: Verify signature
    try:
        verify_hookmesh_signature(raw_body, signature, WEBHOOK_SECRET)
    except ValueError as e:
        return jsonify({'error': str(e)}), 401

    # Step 4: Parse JSON only after verification
    payload = json.loads(raw_body)

    # Step 5: Respond quickly, process asynchronously
    # Queue the event for background processing
    # process_webhook_async(payload)

    return jsonify({'status': 'received'}), 200

if __name__ == '__main__':
    app.run(port=3000)

Complete FastAPI Implementation

import hmac
import hashlib
import time
import os
import json
from fastapi import FastAPI, Request, HTTPException, Header
from fastapi.responses import JSONResponse

app = FastAPI()

WEBHOOK_SECRET = os.environ['WEBHOOK_SECRET']

def verify_hookmesh_signature(payload, signature_header, secret, tolerance=300):
    elements = dict(e.split('=', 1) for e in signature_header.split(','))
    timestamp = elements.get('t')
    received_sig = elements.get('v1')

    if not timestamp or not received_sig:
        raise ValueError('Invalid signature format')

    if abs(int(time.time()) - int(timestamp)) > tolerance:
        raise ValueError('Timestamp outside tolerance')

    signed_payload = f"{timestamp}.".encode() + payload
    expected_sig = hmac.new(
        secret.encode(), signed_payload, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(received_sig, expected_sig):
        raise ValueError('Invalid signature')

    return True

@app.post('/webhooks')
async def handle_webhook(
    request: Request,
    x_hookmesh_signature: str = Header(None)
):
    # Step 1: Get raw body BEFORE any parsing
    raw_body = await request.body()

    # Step 2: Check for signature header
    if not x_hookmesh_signature:
        raise HTTPException(status_code=401, detail='Missing signature')

    # Step 3: Verify signature
    try:
        verify_hookmesh_signature(raw_body, x_hookmesh_signature, WEBHOOK_SECRET)
    except ValueError as e:
        raise HTTPException(status_code=401, detail=str(e))

    # Step 4: Parse JSON only after verification
    payload = json.loads(raw_body)

    # Step 5: Respond quickly, process asynchronously
    # await process_webhook_async(payload)

    return JSONResponse({'status': 'received'}, status_code=200)

# Run with: uvicorn main:app --port 3000

Common Mistakes to Avoid

Parsing JSON Before Verification

Never call request.get_json() before verification.

# WRONG - JSON parsed before getting raw body
@app.route('/webhooks', methods=['POST'])
def handle_webhook():
    payload = request.get_json()  # This consumes the body
    raw_body = request.get_data()  # Now empty or re-serialized

Using Standard String Comparison

Never use ==—use hmac.compare_digest() to prevent timing attacks.

# WRONG - vulnerable to timing attacks
if received_signature == expected_signature:

# CORRECT - timing-safe comparison
if hmac.compare_digest(received_signature, expected_signature):

Slow Webhook Processing

Respond within 5-30 seconds or providers retry, causing duplicates.

# WRONG - slow synchronous processing
@app.route('/webhooks', methods=['POST'])
def handle_webhook():
    # This takes 30 seconds
    process_large_data(payload)
    send_notifications(payload)
    update_external_systems(payload)
    return jsonify({'status': 'ok'})

# CORRECT - respond quickly, process later
@app.route('/webhooks', methods=['POST'])
def handle_webhook():
    # Queue for background processing
    task_queue.enqueue(process_webhook, payload)
    return jsonify({'status': 'received'}), 200

Missing Signature Verification Entirely

Always verify signatures, even in staging environments.

Ignoring Timestamp Validation

Without timestamp checks, attackers replay captured webhooks indefinitely. For more security patterns, see our webhook security best practices.

Responding Quickly

Decouple receiving from processing: verify signature, store in queue, return 200, process asynchronously. To prevent duplicates on retry, implement idempotent webhook handlers.

Conclusion

Secure webhook handling in Python requires: raw body reading, timing-safe comparison, timestamp validation, and fast responses. The code here gives you a solid foundation for Flask and FastAPI.

Hook Mesh includes timestamps in every signature with consistent header formats. Combined with retry logic and monitoring, you focus on processing rather than infrastructure.

To send webhooks from Python, see our sending webhooks with Python guide. For Django, check out our Django webhook integration guide. For more tutorials, explore our webhook implementation guides.

Related Posts

Send Webhooks with Python: A Complete Tutorial

Learn how to send webhooks with Python using the requests library. This hands-on tutorial covers payload structure, HMAC signatures, error handling, retry logic with exponential backoff, and async options.

HMAC-SHA256 Webhook Signatures: Implementation Guide

Learn how to implement secure webhook signature verification using HMAC-SHA256. Complete guide with code examples for signing and verifying webhook payloads in Node.js and Python.

Webhook Security Best Practices: The Complete Guide

Learn how to secure your webhook implementations with HMAC signature verification, replay attack prevention, SSRF mitigation, and more. Includes code examples in Node.js and Python.

Building Webhooks with Django and Hook Mesh

A complete tutorial for Django developers on implementing webhooks—sending events from views and signals, receiving webhooks securely, signature verification, and integrating with Hook Mesh for reliable delivery.

Webhook Idempotency: Why It Matters and How to Implement It

A comprehensive technical guide to implementing idempotency for webhooks. Learn about idempotency keys, deduplication strategies, and implementation patterns with Node.js and Python code examples.