Manage PeerPesa webhooks and verify event signatures

curl -X POST https://api.peerpesa.co/v1/webhooks \
  -H "Authorization: Bearer $ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://example.com/peerpesa/webhooks",
    "events": ["transaction.completed", "transaction.failed", "deposit.confirmed"],
    "description": "Production payouts"
  }'

const res = await fetch("https://api.peerpesa.co/v1/webhooks", {
  method: "POST",
  headers: {
    Authorization: `Bearer ${accessToken}`,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    url: "https://example.com/peerpesa/webhooks",
    events: ["transaction.completed", "transaction.failed", "deposit.confirmed"],
    description: "Production payouts",
  }),
});
const webhook = await res.json();

import crypto from "crypto";
import express from "express";

const app = express();
const WEBHOOK_KEY = process.env.PEERPESA_WEBHOOK_KEY;
const TOLERANCE_MS = 5 * 60 * 1000;

app.post("/peerpesa/webhooks", express.raw({ type: "application/json" }), (req, res) => {
  const signature = req.get("X-Signature");
  const timestamp = req.get("X-Timestamp");
  if (!signature || !timestamp) return res.status(400).send("Missing signature");

  // Reject old or replayed requests (timestamp is in milliseconds)
  if (Math.abs(Date.now() - Number(timestamp)) > TOLERANCE_MS) {
    return res.status(400).send("Stale timestamp");
  }

  const rawBody = req.body.toString("utf8");
  const expected = crypto
    .createHmac("sha256", WEBHOOK_KEY)
    .update(`${timestamp}.${rawBody}`)
    .digest("hex");

  const valid =
    expected.length === signature.length &&
    crypto.timingSafeEqual(Buffer.from(expected), Buffer.from(signature));

  if (!valid) return res.status(401).send("Invalid signature");

  const event = JSON.parse(rawBody);
  res.sendStatus(200); // acknowledge first
  handleEvent(event);  // then process asynchronously
});

import hashlib
import hmac
import os
import time

from flask import Flask, abort, request

app = Flask(__name__)
WEBHOOK_KEY = os.environ["PEERPESA_WEBHOOK_KEY"].encode()
TOLERANCE_MS = 5 * 60 * 1000

@app.post("/peerpesa/webhooks")
def peerpesa_webhook():
    signature = request.headers.get("X-Signature")
    timestamp = request.headers.get("X-Timestamp")
    if not signature or not timestamp:
        abort(400)

    if abs(time.time() * 1000 - int(timestamp)) > TOLERANCE_MS:
        abort(400)

    raw_body = request.get_data(as_text=True)
    expected = hmac.new(
        WEBHOOK_KEY, f"{timestamp}.{raw_body}".encode(), hashlib.sha256
    ).hexdigest()

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

    event = request.get_json()
    # process the event asynchronously
    return "", 200