Webhook Security

X-DoDevWebhook-Signature: t=1700000000,v1=5a4b3c2d1e0f...

import crypto from "crypto";

function verifyWebhookSignature(
  rawBody: string,
  signatureHeader: string,
  signingSecret: string,
  toleranceSeconds = 300
): boolean {
  // Parse the signature header
  const parts = Object.fromEntries(
    signatureHeader.split(",").map((part) => {
      const [key, value] = part.split("=");
      return [key, value];
    })
  );

  const timestamp = parts.t;
  const signature = parts.v1;

  if (!timestamp || !signature) {
    throw new Error("Invalid signature header format");
  }

  // Check timestamp tolerance (replay protection)
  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - parseInt(timestamp)) > toleranceSeconds) {
    throw new Error("Webhook timestamp too old");
  }

  // Compute expected signature
  const signedPayload = `${timestamp}.${rawBody}`;
  const expected = crypto
    .createHmac("sha256", signingSecret)
    .update(signedPayload)
    .digest("hex");

  // Constant-time comparison
  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expected)
  );
}

import express from "express";

const app = express();

// Important: use raw body for signature verification
app.post("/webhooks/dodev", express.raw({ type: "application/json" }), (req, res) => {
  const rawBody = req.body.toString();
  const signature = req.headers["x-dodevwebhook-signature"] as string;

  try {
    const valid = verifyWebhookSignature(rawBody, signature, process.env.WEBHOOK_SECRET!);
    if (!valid) {
      return res.status(401).send("Invalid signature");
    }
  } catch (err) {
    return res.status(401).send("Signature verification failed");
  }

  const event = JSON.parse(rawBody);
  console.log(`Received: ${event.type}`);

  // Process the event...

  res.status(200).send("OK");
});

import crypto from "crypto";
import { NextResponse } from "next/server";

export async function POST(request: Request) {
  const rawBody = await request.text();
  const signature = request.headers.get("x-dodevwebhook-signature");

  if (!signature) {
    return NextResponse.json({ error: "Missing signature" }, { status: 401 });
  }

  const valid = verifyWebhookSignature(
    rawBody,
    signature,
    process.env.WEBHOOK_SECRET!
  );

  if (!valid) {
    return NextResponse.json({ error: "Invalid signature" }, { status: 401 });
  }

  const event = JSON.parse(rawBody);

  switch (event.type) {
    case "send.email.bounced":
      await handleBounce(event.data);
      break;
    case "send.email.complained":
      await handleComplaint(event.data);
      break;
  }

  return NextResponse.json({ received: true });
}

import hmac
import hashlib
import time

def verify_webhook(raw_body: str, signature_header: str, signing_secret: str, tolerance: int = 300) -> bool:
    # Parse signature header
    parts = dict(part.split("=", 1) for part in signature_header.split(","))
    timestamp = parts.get("t")
    signature = parts.get("v1")

    if not timestamp or not signature:
        raise ValueError("Invalid signature header")

    # Check timestamp tolerance
    if abs(time.time() - int(timestamp)) > tolerance:
        raise ValueError("Webhook timestamp too old")

    # Compute expected signature
    signed_payload = f"{timestamp}.{raw_body}"
    expected = hmac.new(
        signing_secret.encode(),
        signed_payload.encode(),
        hashlib.sha256,
    ).hexdigest()

    return hmac.compare_digest(signature, expected)

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/webhooks/dodev", methods=["POST"])
def handle_webhook():
    raw_body = request.get_data(as_text=True)
    signature = request.headers.get("X-DoDevWebhook-Signature")

    if not verify_webhook(raw_body, signature, WEBHOOK_SECRET):
        return jsonify({"error": "Invalid signature"}), 401

    event = request.get_json()
    print(f"Received: {event['type']}")

    return jsonify({"received": True}), 200

package main

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "fmt"
    "math"
    "strconv"
    "strings"
    "time"
)

func verifyWebhook(rawBody, signatureHeader, signingSecret string) (bool, error) {
    // Parse signature header
    parts := make(map[string]string)
    for _, part := range strings.Split(signatureHeader, ",") {
        kv := strings.SplitN(part, "=", 2)
        if len(kv) == 2 {
            parts[kv[0]] = kv[1]
        }
    }

    timestamp := parts["t"]
    signature := parts["v1"]

    // Check timestamp tolerance (5 minutes)
    ts, _ := strconv.ParseInt(timestamp, 10, 64)
    if math.Abs(float64(time.Now().Unix()-ts)) > 300 {
        return false, fmt.Errorf("webhook timestamp too old")
    }

    // Compute expected signature
    signedPayload := fmt.Sprintf("%s.%s", timestamp, rawBody)
    mac := hmac.New(sha256.New, []byte(signingSecret))
    mac.Write([]byte(signedPayload))
    expected := hex.EncodeToString(mac.Sum(nil))

    return hmac.Equal([]byte(signature), []byte(expected)), nil
}