Setup

import express from 'express';
import { verifyWebhook } from 'babysea/webhooks';

// Use express.text() to get the raw body for signature verification.
// express.json() parses the body and may alter whitespace, breaking the signature.
app.post('/webhook', express.text({ type: 'application/json' }), async (req, res) => {
  const rawBody = req.body; // string, not parsed JSON
  const signature = req.headers['x-babysea-signature'] as string;

  let payload;
  try {
    payload = await verifyWebhook(rawBody, signature, process.env.BABYSEA_WEBHOOK_SECRET!);
  } catch {
    return res.status(400).send('Invalid signature');
  }

  // Acknowledge immediately
  res.status(200).send('OK');

  // Process in the background
  await queue.push(payload);
});

import hmac
import hashlib
import json
import time
from flask import Flask, request, abort

app = Flask(__name__)
WEBHOOK_SECRET = os.environ["BABYSEA_WEBHOOK_SECRET"]

@app.route("/webhook", methods=["POST"])
def webhook():
    raw_body = request.get_data(as_text=True)
    sig_header = request.headers.get("X-BabySea-Signature", "")

    parts = dict(p.split("=", 1) for p in sig_header.split(","))
    timestamp = parts.get("t", "")
    signature = parts.get("v1", "")

    # Check timestamp freshness against the default 5 minute tolerance window
    if abs(time.time() - int(timestamp)) > 300:
        abort(400, "Timestamp outside tolerance")

    # Verify HMAC-SHA256
    signed_payload = f"{timestamp}.{raw_body}"
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        signed_payload.encode(),
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(expected, signature):
        abort(400, "Invalid signature")

    payload = json.loads(raw_body)
    # Process async (e.g., enqueue to Celery)
    process_event.delay(payload)

    return "OK", 200