Flask Webhook Signature Verification Example
Webhooks are a cornerstone of modern distributed systems, enabling real-time communication between services. Whether you're integrating with payment gateways like Stripe, version control systems like GitHub, or communication platforms like Slack, webhooks provide an efficient way for external services to notify your application about events.
However, convenience often comes with security considerations. When your application exposes an endpoint to the public internet for webhooks, you're opening a potential vector for malicious activity. How do you ensure that an incoming webhook request genuinely originated from the service it claims to be from, and hasn't been tampered with along the way? The answer lies in webhook signature verification.
In this article, we'll dive deep into implementing robust webhook signature verification within a Flask application. We'll cover the core principles, walk through a practical example, and highlight common pitfalls and best practices to keep your integrations secure.
Why Webhook Signature Verification is Critical
Imagine your application processes orders based on webhooks from a payment provider. Without verification, an attacker could send a fake "payment successful" webhook, tricking your system into fulfilling an order that was never paid for. Or, they could replay an old webhook, causing duplicate actions.
Signature verification addresses these concerns by:
- Authenticity: Confirming the request originated from the legitimate sender.
- Integrity: Ensuring the payload hasn't been altered during transit.
- Replay Protection: Mitigating the risk of attackers re-sending old, valid requests.
Most reputable webhook providers implement a signature mechanism. Typically, this involves:
- A shared secret key known only to your application and the webhook sender.
- The sender computing a hash (or HMAC) of the request's payload (and sometimes headers or a timestamp) using this secret.
- The sender including this computed signature in a request header.
- Your application receiving the request, re-computing the signature using the same secret and method, and comparing it to the received signature.
If the signatures match, you can trust the request.
Setting Up a Basic Flask Webhook Endpoint
Before we get to verification, let's set up a minimal Flask application with an endpoint to receive webhooks.
First, ensure you have Flask installed:
pip install Flask python-dotenv
Then, create a file named app.py:
import os
from flask import Flask, request, abort
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
app = Flask(__name__)
@app.route('/webhook', methods=['POST'])
def handle_webhook():
# In a real application, you'd process the webhook here.
# For now, we'll just acknowledge receipt.
print(f"Received webhook: {request.json}")
return "Webhook received!", 200
if __name__ == '__main__':
app.run(debug=True, port=5000)
And a .env file to hold your secret later:
WEBHOOK_SECRET=your_super_secret_key_here
Run your app:
python app.py
Now, if you send a POST request to http://localhost:5000/webhook, it will print the JSON payload.
curl -X POST -H "Content-Type: application/json" -d '{"event": "test", "data": {"id": "123"}}' http://localhost:5000/webhook
This is a good start, but currently, anyone can send any data to this endpoint. We need to secure it.
The Core Logic: Signature Verification in Flask
The verification process involves several steps:
- Retrieve the Secret: Get your shared secret from a secure location (environment variables are best).
- Extract Signature and Timestamp (if applicable): These are usually in request headers.
- Get the Raw Request Body: This is crucial. You must use the raw, untampered request body that was used to generate the original signature.
- Compute Your Own Signature: Use the same hashing algorithm and secret as the sender.
- Compare Signatures: Perform a constant-time comparison to prevent timing attacks.
- Verify Timestamp (if applicable): Ensure the request isn't too old.
Let's integrate this into our Flask application. We'll use GitHub's X-Hub-Signature-256 header as a concrete example, which uses HMAC-SHA256. Many other services like Stripe and Shopify use similar HMAC-based schemes, often differing only in header names and specific payload construction for signing.
import os
import hmac
import hashlib
import time
from flask import Flask, request, abort
from dotenv import load_dotenv
load_dotenv()
app = Flask(__name__)
# Load your webhook secret from environment variables
WEBHOOK_SECRET = os.getenv('WEBHOOK_SECRET')
if not WEBHOOK_SECRET:
raise ValueError("WEBHOOK_SECRET environment variable not set.")
def verify_github_signature(payload_body, signature_header):
"""
Verifies a GitHub webhook signature.
Signature header format: "sha256=<signature>"
"""
if not signature_header:
abort(401, "Signature header not found.")
try:
# Expected format: "sha256=<signature>"
method, github_signature = signature_header.split('=')
if method != 'sha256':
abort(401, "Unsupported signature method.")
except ValueError:
abort(401, "Invalid signature header format.")
# Compute our own signature
# The secret must be bytes for hmac.new
secret_bytes = WEBHOOK_SECRET.encode('utf-8')
# The payload body must also be bytes
payload_bytes = payload_body
# Create an HMAC-SHA256 hash
our_signature = hmac.new(secret_bytes, payload_bytes, hashlib.sha256).hexdigest()
# Compare the signatures using a constant-time comparison
if not hmac.compare_digest(our_signature, github_signature):
abort(403, "Invalid signature.")
return True
@app.route('/webhook', methods=['POST'])
def handle_webhook():
# 1. Get the raw request body
# Crucial: DO NOT use request.json or request.get_json() directly for signature verification.
# These methods parse the JSON, which might alter whitespace or key order,
# leading to a different hash than the sender calculated.
# Always use the raw bytes from request.get_data() or request.data.
payload = request.get_data()
# 2. Extract the signature header
# GitHub uses 'X-Hub-Signature-256' for SHA256
github_signature_header = request.headers.get('X-Hub-Signature-256')
# 3. Verify the signature
try:
verify_github_signature(payload, github_signature_header)
except Exception as e:
app.logger.error(f"Signature verification failed: {e}")
abort(403, description="Signature verification failed.")
# If verification passes, process the webhook
# Now you can safely parse the JSON payload
try:
webhook_data = request.json
print(f"Successfully verified and received webhook: {webhook_data.get('event', 'unknown event')}")
# In a real app, queue this for background processing
except Exception as e:
app.logger.error(f"Error parsing JSON payload: {e}")
abort(400, description="Invalid JSON payload.")
return "Webhook received and verified!", 200
if __name__ == '__main__':
app.run(debug=True, port=5000)
Real-world Example: Stripe Signature Verification
Stripe's signature verification is a great example that includes a timestamp to prevent replay attacks. Their Stripe-Signature header typically looks like this: `t=1678886400,v