Webhooks

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

const app = express();
const SECRET = process.env.PEEVE_WEBHOOK_SECRET; // whsec_…
const TOLERANCE_SECONDS = 300;

// express.raw — NOT express.json. The signature covers the bytes on the wire.
app.post("/hooks/peeve", express.raw({ type: "application/json" }), (req, res) => {
  const signature = req.get("Peeve-Signature") || "";
  const timestamp = req.get("Peeve-Timestamp") || "";
  const body = req.body; // Buffer

  // 1. Freshness — reject replays.
  const age = Math.abs(Math.floor(Date.now() / 1000) - Number(timestamp));
  if (!timestamp || Number.isNaN(age) || age > TOLERANCE_SECONDS) {
    return res.status(400).send("stale");
  }

  // 2. Recompute.
  const expected =
    "v1=" +
    crypto.createHmac("sha256", SECRET).update(`${timestamp}.${body}`, "utf8").digest("hex");

  // 3. Constant-time compare.
  const a = Buffer.from(signature);
  const b = Buffer.from(expected);
  if (a.length !== b.length || !crypto.timingSafeEqual(a, b)) {
    return res.status(401).send("bad signature");
  }

  const event = JSON.parse(body.toString("utf8"));
  if (!event.livemode) {
    // A test delivery. Log it, don't create real records from it.
  }

  // Answer fast; do the work afterwards.
  res.status(200).send("ok");
});

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

app = Flask(__name__)
SECRET = os.environ["PEEVE_WEBHOOK_SECRET"].encode()
TOLERANCE_SECONDS = 300

@app.post("/hooks/peeve")
def peeve_webhook():
    signature = request.headers.get("Peeve-Signature", "")
    timestamp = request.headers.get("Peeve-Timestamp", "")
    body = request.get_data()  # raw bytes

    try:
        age = abs(int(time.time()) - int(timestamp))
    except ValueError:
        abort(400)
    if age > TOLERANCE_SECONDS:
        abort(400)

    expected = "v1=" + hmac.new(
        SECRET, f"{timestamp}.".encode() + body, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        abort(401)

    event = json.loads(body)
    if not event["livemode"]:
        pass  # test delivery

    return "ok", 200