Webhooks Security & Validation

import base64
import hashlib
import hmac
import gzip
from datetime import datetime, timezone, timedelta
def verify_catena_webhook(headers: dict, raw_body: bytes, secret: str) -> bool:
    """
    Verify a Catena webhook signature.
    Args:
        headers: Request headers dict
        raw_body: Raw request body as bytes (potentially compressed)
        secret: Your webhook secret
    Returns:
        True if signature is valid and timestamp is fresh
    """
    timestamp = headers.get("X-Catena-Timestamp")
    signature = headers.get("X-Catena-Signature")
    if not timestamp or not signature:
        return False
    # Check timestamp freshness (within 5 minutes)
    try:
        ts = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
        age = datetime.now(timezone.utc) - ts
        if age > timedelta(minutes=5):
            return False
    except ValueError:
        return False
    # Decompress payload (always gzip)
    try:
        payload = gzip.decompress(raw_body).decode("utf-8")
    except Exception:
        return False
    # Compute expected signature
    signed_payload = f"{timestamp}.{payload}".encode()
    expected = base64.b64encode(
        hmac.new(secret.encode(), signed_payload, hashlib.sha256).digest()
    ).decode()
    # Constant-time comparison
    return hmac.compare_digest(signature, expected)
# Usage in a Flask endpoint
from flask import Flask, request, jsonify
app = Flask(__name__)
WEBHOOK_SECRET = "your-webhook-secret"
@app.route("/webhooks/catena", methods=["POST"])
def handle_webhook():
    # Get raw body before Flask parses it
    raw_body = request.get_data()
    if not verify_catena_webhook(request.headers, raw_body, WEBHOOK_SECRET):
        return jsonify({"error": "Invalid signature"}), 401
    # Process the webhook
    event = request.get_json()
    event_type = request.headers.get("X-Event-Type")
    # Handle the event...
    return jsonify({"status": "received"}), 200