Webhooks

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

const app = express();

app.post(
  "/webhooks/fortpesa",
  express.raw({ type: "application/json" }), // keep the raw body
  (req, res) => {
    const secret = process.env.FORTPESA_WEBHOOK_SECRET; // whsec_...
    const timestamp = req.header("X-Fortpesa-Timestamp");
    const signature = req.header("X-Fortpesa-Signature");

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

    const a = Buffer.from(signature ?? "");
    const b = Buffer.from(expected);
    const signatureOk = a.length === b.length && crypto.timingSafeEqual(a, b);
    const fresh = Math.abs(Date.now() / 1000 - Number(timestamp)) < 300;

    if (!signatureOk || !fresh) return res.sendStatus(400);

    const event = JSON.parse(req.body.toString("utf8"));
    // Re-fetch the transaction before fulfilling (see "Trust but verify").
    console.log(event.event, event.data.uuid);
    res.sendStatus(200); // respond 2xx quickly
  }
);

// app/api/webhooks/fortpesa/route.ts
import { createHmac, timingSafeEqual } from "crypto";

export async function POST(req: Request) {
  const raw = await req.text(); // raw bytes — do NOT use req.json() first
  const secret = process.env.FORTPESA_WEBHOOK_SECRET!;
  const timestamp = req.headers.get("X-Fortpesa-Timestamp")!;
  const signature = req.headers.get("X-Fortpesa-Signature")!;

  const expected =
    "sha256=" +
    createHmac("sha256", secret)
      .update(`${timestamp}.${raw}`)
      .digest("hex");

  const a = Buffer.from(signature);
  const b = Buffer.from(expected);
  const ok = a.length === b.length && timingSafeEqual(a, b);
  const fresh = Math.abs(Date.now() / 1000 - Number(timestamp)) < 300;

  if (!ok || !fresh) return new Response("Invalid signature", { status: 400 });

  const event = JSON.parse(raw);
  // Re-fetch GET /api/v1/transactions/{event.data.uuid} before fulfilling.
  return new Response(null, { status: 200 });
}

import hashlib
import hmac
import os
import time

from flask import Flask, abort, request

app = Flask(__name__)


@app.post("/webhooks/fortpesa")
def fortpesa_webhook():
    secret = os.environ["FORTPESA_WEBHOOK_SECRET"].encode()  # whsec_...
    raw = request.get_data()  # raw bytes — before any JSON parsing
    timestamp = request.headers["X-Fortpesa-Timestamp"].encode()
    signature = request.headers["X-Fortpesa-Signature"]

    expected = "sha256=" + hmac.new(
        secret, timestamp + b"." + raw, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        abort(400)
    if abs(time.time() - int(timestamp)) > 300:
        abort(400)

    event = request.get_json()
    # Re-fetch the transaction before fulfilling.
    return "", 200

use Illuminate\Http\Request;
use Illuminate\Support\Facades\Route;

Route::post('/webhooks/fortpesa', function (Request $request) {
    $secret = config('services.fortpesa.webhook_secret'); // whsec_...
    $raw = $request->getContent(); // raw bytes
    $timestamp = (string) $request->header('X-Fortpesa-Timestamp');
    $signature = (string) $request->header('X-Fortpesa-Signature');

    $expected = 'sha256=' . hash_hmac('sha256', $timestamp . '.' . $raw, $secret);

    if (! hash_equals($expected, $signature)) {
        abort(403, 'Invalid signature');
    }
    if (abs(time() - (int) $timestamp) > 300) {
        abort(403, 'Stale timestamp');
    }

    $event = $request->json()->all();
    // Re-fetch GET /api/v1/transactions/{uuid} before fulfilling.
    return response()->noContent();
});