Secure Webhooks by Verifying Request Signatures

# app.py (Python with Flask)
import hmac
import hashlib
import json
import os
from flask import Flask, request, abort

app = Flask(__name__)

# Load secret key from environment variables for security
WEBHOOK_SECRET = os.environ.get('WEBHOOK_SECRET')

@app.route('/webhook', methods=['POST'])
def webhook_receiver():
    if not WEBHOOK_SECRET:
        app.logger.error("WEBHOOK_SECRET environment variable not set.")
        abort(500, description="Server configuration error.")

    # Get the raw request body
    payload = request.data
    
    # Get the signature from the header provided by the webhook sender
    # The header name varies (e.g., 'X-Hub-Signature', 'X-Stripe-Signature', 'X-GitHub-Delivery')
    # For this example, let's assume 'X-Webhook-Signature'
    signature_header = request.headers.get('X-Webhook-Signature')

    if not signature_header:
        app.logger.warning("Missing X-Webhook-Signature header.")
        abort(400, description="Signature header missing.")

    # Calculate the expected signature
    # The hash algorithm (e.g., sha256) and signature format might vary
    # Example: 'sha256=<hex_digest>'
    try:
        method, received_signature = signature_header.split('=')
        if method != 'sha256':
            app.logger.warning(f"Unsupported signature method: {method}")
            abort(400, description="Unsupported signature method.")
    except ValueError:
        app.logger.warning(f"Invalid signature header format: {signature_header}")
        abort(400, description="Invalid signature header format.")

    # Convert secret to bytes
    secret_bytes = WEBHOOK_SECRET.encode('utf-8')

    # Calculate HMAC digest
    computed_hmac = hmac.new(secret_bytes, payload, hashlib.sha256).hexdigest()

    # Compare signatures using a constant-time comparison to prevent timing attacks
    if not hmac.compare_digest(computed_hmac, received_signature):
        app.logger.warning("Webhook signature mismatch.")
        abort(401, description="Invalid signature.")

    # If signature is valid, process the payload
    try:
        data = json.loads(payload)
        app.logger.info(f"Received valid webhook: {data}")
        # Process your webhook data here
        return {"status": "success", "message": "Webhook received and verified."}, 200
    except json.JSONDecodeError:
        app.logger.error("Invalid JSON payload.")
        abort(400, description="Invalid JSON payload.")

if __name__ == '__main__':
    # To run this:
    # 1. pip install Flask
    # 2. Set environment variable: export WEBHOOK_SECRET="your_secure_secret_key"
    #    (This key must match the one configured on the sender's side)
    # 3. python app.py
    # This example runs on http://127.0.0.1:5000/webhook
    app.run(debug=True)