Webhook Signatures

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

const webhook = new Webhook(process.env.REQUESTY_WEBHOOK_SECRET);
const app = express();

app.post(
  "/requesty/webhooks",
  express.raw({ type: "application/json" }),
  (req, res) => {
    let event;
    try {
      event = webhook.verify(req.body, req.headers);
    } catch (err) {
      return res.status(400).send("Invalid signature");
    }

    // Handle the event, e.g. by event.type for alerts
    res.sendStatus(200);
  }
);

import os

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

webhook = Webhook(os.environ["REQUESTY_WEBHOOK_SECRET"])
app = Flask(__name__)


@app.post("/requesty/webhooks")
def requesty_webhook():
    try:
        event = webhook.verify(request.get_data(), dict(request.headers))
    except WebhookVerificationError:
        abort(400)

    # Handle the event, e.g. by event["type"] for alerts
    return "", 200

import crypto from "node:crypto";

const TOLERANCE_SECONDS = 5 * 60;

export function verifyRequestyWebhook(secret, headers, rawBody) {
  const id = headers["webhook-id"];
  const timestamp = headers["webhook-timestamp"];
  const signatures = headers["webhook-signature"];
  if (!id || !timestamp || !signatures) {
    throw new Error("Missing signature headers");
  }

  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - Number(timestamp)) > TOLERANCE_SECONDS) {
    throw new Error("Timestamp outside tolerance");
  }

  const key = Buffer.from(secret.replace(/^whsec_/, ""), "base64");
  const expected = crypto
    .createHmac("sha256", key)
    .update(`${id}.${timestamp}.${rawBody}`)
    .digest();

  const valid = signatures.split(" ").some((entry) => {
    const [version, signature] = entry.split(",");
    if (version !== "v1" || !signature) return false;
    const received = Buffer.from(signature, "base64");
    return (
      received.length === expected.length &&
      crypto.timingSafeEqual(received, expected)
    );
  });
  if (!valid) {
    throw new Error("No matching signature");
  }

  return JSON.parse(rawBody);
}

import base64
import hashlib
import hmac
import json
import time

TOLERANCE_SECONDS = 5 * 60


def verify_requesty_webhook(secret: str, headers: dict, raw_body: bytes) -> dict:
    headers = {k.lower(): v for k, v in headers.items()}
    msg_id = headers["webhook-id"]
    timestamp = headers["webhook-timestamp"]
    signatures = headers["webhook-signature"]

    if abs(time.time() - int(timestamp)) > TOLERANCE_SECONDS:
        raise ValueError("Timestamp outside tolerance")

    key = base64.b64decode(secret.removeprefix("whsec_"))
    signed_content = f"{msg_id}.{timestamp}.".encode() + raw_body
    expected = base64.b64encode(
        hmac.new(key, signed_content, hashlib.sha256).digest()
    ).decode()

    for entry in signatures.split(" "):
        version, _, signature = entry.partition(",")
        if version == "v1" and hmac.compare_digest(signature, expected):
            return json.loads(raw_body)

    raise ValueError("No matching signature")