Webhooks

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

const app = express();

export function isFromReedapt(rawBody, header, secret) {
  const parts = Object.fromEntries(header.split(",").map((p) => p.split("=")));
  const expected = crypto
    .createHmac("sha256", secret)
    .update(`${parts.t}.${rawBody}`)
    .digest("hex");
  const fresh = Math.abs(Date.now() / 1000 - Number(parts.t)) < 300;
  const given = Buffer.from(parts.v1 ?? "");
  return fresh && given.length === expected.length && crypto.timingSafeEqual(Buffer.from(expected), given);
}

// express.raw keeps the body exactly as it was sent.
app.post("/reedapt/webhooks", express.raw({ type: "application/json" }), (req, res) => {
  const header = req.get("Reedapt-Signature") ?? "";
  if (!isFromReedapt(req.body.toString("utf8"), header, process.env.REEDAPT_WEBHOOK_SECRET)) {
    return res.sendStatus(400);
  }
  const event = JSON.parse(req.body);
  // Queue event for processing, then answer quickly.
  res.sendStatus(200);
});

import hashlib
import hmac
import os
import time

from flask import Flask, abort, request

app = Flask(__name__)


def is_from_reedapt(raw_body: bytes, header: str, secret: str) -> bool:
    parts = dict(p.split("=", 1) for p in header.split(",") if "=" in p)
    signed = parts.get("t", "").encode() + b"." + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
    fresh = parts.get("t", "").isdigit() and abs(time.time() - int(parts["t"])) < 300
    return fresh and hmac.compare_digest(expected, parts.get("v1", ""))


@app.post("/reedapt/webhooks")
def reedapt_webhook():
    header = request.headers.get("Reedapt-Signature", "")
    if not is_from_reedapt(request.get_data(), header, os.environ["REEDAPT_WEBHOOK_SECRET"]):
        abort(400)
    event = request.get_json()
    # Queue event for processing, then answer quickly.
    return "", 200