How to verify webhook signatures

import express from "express";
import Stripe from "stripe";

const stripe = new Stripe(process.env.STRIPE_SECRET_KEY);
const app = express();

// Raw body on this route: the signature covers the exact bytes.
app.post("/webhook", express.raw({ type: "application/json" }), (req, res) => {
  let event;
  try {
    event = stripe.webhooks.constructEvent(
      req.body, req.headers["stripe-signature"], process.env.STRIPE_WEBHOOK_SECRET);
  } catch (err) {
    return res.status(400).send(`Bad signature: ${err.message}`);
  }
  res.sendStatus(200);
  // …then handle event.type, and skip event.id values you've already seen
});

app.use(express.json());  // JSON parsing for every other route, after the webhook
app.listen(3000);

import os
import stripe
from flask import Flask, request

app = Flask(__name__)

@app.post("/webhook")
def webhook():
    payload = request.get_data()  # the raw bytes; request.json would be too late
    try:
        event = stripe.Webhook.construct_event(
            payload, request.headers.get("Stripe-Signature"),
            os.environ["STRIPE_WEBHOOK_SECRET"])
    except (ValueError, stripe.error.SignatureVerificationError):
        return "Bad signature", 400
    # …handle event["type"], and skip event["id"] values you've already seen
    return "", 200

import crypto from "node:crypto";
import express from "express";

const app = express();

// timingSafeEqual throws if the lengths differ, so check that first.
function safeEqual(a, b) {
  const x = Buffer.from(a), y = Buffer.from(b);
  return x.length === y.length && crypto.timingSafeEqual(x, y);
}

function verifyStripe(rawBody, header = "", secret, tolerance = 300) {
  const items = header.split(",").map((part) => part.split("="));
  const t = items.find(([k]) => k === "t")?.[1];
  const sigs = items.filter(([k]) => k === "v1").map(([, v]) => v);  // two during a secret roll
  if (!t || sigs.length === 0) return false;
  const age = Math.abs(Date.now() / 1000 - Number(t));
  if (!(age <= tolerance)) return false;  // also rejects a non-numeric t
  const expected = crypto.createHmac("sha256", secret)
    .update(`${t}.`).update(rawBody).digest("hex");
  return sigs.some((sig) => safeEqual(expected, sig));
}

app.post("/webhook", express.raw({ type: "application/json" }), (req, res) => {
  if (!Buffer.isBuffer(req.body) ||
      !verifyStripe(req.body, req.get("Stripe-Signature"), process.env.STRIPE_WEBHOOK_SECRET)) {
    return res.sendStatus(400);
  }
  const event = JSON.parse(req.body.toString("utf8"));  // parse only after the check
  res.sendStatus(200);
});

app.listen(3000);

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

app = Flask(__name__)

def verify_stripe(payload: bytes, header: str, secret: str, tolerance=300) -> bool:
    items = [part.split("=", 1) for part in (header or "").split(",") if "=" in part]
    t = next((v for k, v in items if k == "t"), "")
    sigs = [v for k, v in items if k == "v1"]  # two during a secret roll
    if not (t.isascii() and t.isdigit()) or not sigs or abs(time.time() - int(t)) > tolerance:
        return False
    expected = hmac.new(secret.encode(), t.encode() + b"." + payload, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected.encode(), sig.encode()) for sig in sigs)

@app.post("/webhook")
def webhook():
    payload = request.get_data()
    if not verify_stripe(payload, request.headers.get("Stripe-Signature"),
                         os.environ["STRIPE_WEBHOOK_SECRET"]):
        return "Bad signature", 400
    event = json.loads(payload)  # parse only after the check
    return "", 200

import crypto from "node:crypto";
import express from "express";

const app = express();
const secret = process.env.GITHUB_WEBHOOK_SECRET;

// timingSafeEqual throws if the lengths differ, so check that first.
function safeEqual(a, b) {
  const x = Buffer.from(a), y = Buffer.from(b);
  return x.length === y.length && crypto.timingSafeEqual(x, y);
}

// GitHub payloads can be up to 25 MB; express.raw stops at 100 kB by default.
app.post("/webhook", express.raw({ type: "*/*", limit: "25mb" }), (req, res) => {
  if (!Buffer.isBuffer(req.body)) return res.sendStatus(400);
  const expected = "sha256=" +
    crypto.createHmac("sha256", secret).update(req.body).digest("hex");
  if (!safeEqual(expected, req.get("X-Hub-Signature-256") ?? "")) {
    return res.sendStatus(401);
  }
  const event = JSON.parse(req.body.toString("utf8"));  // with content type application/json
  res.sendStatus(204);
});

app.listen(3000);

import hashlib
import hmac
import os
from flask import Flask, abort, request

app = Flask(__name__)
SECRET = os.environ["GITHUB_WEBHOOK_SECRET"].encode()

@app.post("/webhook")
def github_webhook():
    body = request.get_data()  # raw bytes: read them before request.json or request.form
    expected = "sha256=" + hmac.new(SECRET, body, hashlib.sha256).hexdigest()
    received = request.headers.get("X-Hub-Signature-256", "")
    if not hmac.compare_digest(expected.encode(), received.encode()):
        abort(401)
    event = request.get_json()  # with content type application/json
    return "", 204

import crypto from "node:crypto";
import express from "express";

const app = express();
const secret = process.env.SLACK_SIGNING_SECRET;

// timingSafeEqual throws if the lengths differ, so check that first.
function safeEqual(a, b) {
  const x = Buffer.from(a), y = Buffer.from(b);
  return x.length === y.length && crypto.timingSafeEqual(x, y);
}

app.post("/slack/events", express.raw({ type: "*/*" }), (req, res) => {
  if (!Buffer.isBuffer(req.body)) return res.sendStatus(400);
  const ts = req.get("X-Slack-Request-Timestamp") ?? "";
  const age = Math.abs(Date.now() / 1000 - Number(ts));
  if (!(age <= 300)) return res.sendStatus(401);  // older than 5 minutes, or not a number
  const expected = "v0=" + crypto.createHmac("sha256", secret)
    .update(`v0:${ts}:`).update(req.body).digest("hex");
  if (!safeEqual(expected, req.get("X-Slack-Signature") ?? "")) {
    return res.sendStatus(401);
  }
  res.sendStatus(200);
  // …then parse: JSON for the Events API, a form for slash commands
});

app.listen(3000);

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

app = Flask(__name__)
SECRET = os.environ["SLACK_SIGNING_SECRET"].encode()

@app.post("/slack/events")
def slack_events():
    body = request.get_data()  # first: once request.form has run, the bytes are gone
    ts = request.headers.get("X-Slack-Request-Timestamp", "")
    if not (ts.isascii() and ts.isdigit()) or abs(time.time() - int(ts)) > 300:
        abort(401)  # older than 5 minutes, or not a number
    base = b"v0:" + ts.encode() + b":" + body
    expected = "v0=" + hmac.new(SECRET, base, hashlib.sha256).hexdigest()
    received = request.headers.get("X-Slack-Signature", "")
    if not hmac.compare_digest(expected.encode(), received.encode()):
        abort(401)
    return "", 200

import crypto from "node:crypto";
import express from "express";

const app = express();
const secret = process.env.SHOPIFY_CLIENT_SECRET;

// timingSafeEqual throws if the lengths differ, so check that first.
function safeEqual(a, b) {
  const x = Buffer.from(a), y = Buffer.from(b);
  return x.length === y.length && crypto.timingSafeEqual(x, y);
}

app.post("/webhooks", express.raw({ type: "*/*" }), (req, res) => {
  if (!Buffer.isBuffer(req.body)) return res.sendStatus(400);
  const expected = crypto.createHmac("sha256", secret).update(req.body).digest("base64");
  if (!safeEqual(expected, req.get("X-Shopify-Hmac-Sha256") ?? "")) {
    return res.sendStatus(401);
  }
  res.sendStatus(200);
  // …then skip X-Shopify-Webhook-Id values you've already handled
});

app.listen(3000);

import base64
import hashlib
import hmac
import os
from flask import Flask, abort, request

app = Flask(__name__)
SECRET = os.environ["SHOPIFY_CLIENT_SECRET"].encode()

@app.post("/webhooks")
def shopify_webhook():
    body = request.get_data()
    digest = hmac.new(SECRET, body, hashlib.sha256).digest()
    expected = base64.b64encode(digest).decode()  # base64, not hex
    received = request.headers.get("X-Shopify-Hmac-Sha256", "")
    if not hmac.compare_digest(expected.encode(), received.encode()):
        abort(401)
    return "", 200

import express from "express";
import { Webhook } from "standardwebhooks";

const app = express();
// Takes the "whsec_…" string: strips the prefix and base64-decodes the rest.
const wh = new Webhook(process.env.WEBHOOK_SECRET);

app.post("/webhook", express.raw({ type: "*/*" }), (req, res) => {
  let event;
  try {
    event = wh.verify(req.body, req.headers);  // checks signature and 5-minute window
  } catch (err) {
    return res.sendStatus(401);
  }
  res.sendStatus(204);
});

app.listen(3000);

import os
from flask import Flask, abort, request
from standardwebhooks.webhooks import Webhook, WebhookVerificationError

app = Flask(__name__)
# Takes the "whsec_…" string: strips the prefix and base64-decodes the rest.
wh = Webhook(os.environ["WEBHOOK_SECRET"])

@app.post("/webhook")
def webhook():
    try:
        event = wh.verify(request.get_data(), dict(request.headers))
    except WebhookVerificationError:
        abort(401)
    return "", 204