Verify signatures

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

const app = express();
const secret = process.env.SAASYBILL_WEBHOOK_SECRET; // whsec_…

// Read the body as raw bytes. Don't use express.json() on this route.
app.post("/saasybill-webhooks", express.raw({ type: "application/json" }), (req, res) => {
  const body = req.body.toString("utf8");

  if (!verifySignature(body, req.get("Saasybill-Signature") ?? "", secret)) {
    return res.sendStatus(400);
  }

  const event = JSON.parse(body);
  // Skip repeats using event.id, then handle event.type here.

  res.sendStatus(200);
});

function verifySignature(body, header, secret, toleranceSeconds = 300) {
  let timestamp = null;
  const signatures = [];

  for (const part of header.split(",")) {
    const [key, value] = part.trim().split("=");
    if (key === "t" && /^\d+$/.test(value ?? "")) timestamp = Number(value);
    if (key === "v1" && value) signatures.push(value);
  }

  if (timestamp === null || signatures.length === 0) return false;
  if (Math.abs(Date.now() / 1000 - timestamp) > toleranceSeconds) return false;

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

  return signatures.some((signature) => {
    const actual = Buffer.from(signature, "hex");
    return actual.length === expected.length && crypto.timingSafeEqual(actual, expected);
  });
}

app.listen(3000);

import hashlib
import hmac
import os
import time

from flask import Flask, request

app = Flask(__name__)
SECRET = os.environ["SAASYBILL_WEBHOOK_SECRET"]  # whsec_…


def verify_signature(body: bytes, header: str, secret: str, tolerance: int = 300) -> bool:
    timestamp, signatures = None, []

    for part in header.split(","):
        key, _, value = part.strip().partition("=")
        if key == "t" and value.isdigit():
            timestamp = int(value)
        elif key == "v1" and value:
            signatures.append(value)

    if timestamp is None or not signatures:
        return False
    if abs(time.time() - timestamp) > tolerance:
        return False

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

    return any(hmac.compare_digest(expected, signature) for signature in signatures)


@app.post("/saasybill-webhooks")
def saasybill_webhooks():
    body = request.get_data()  # the raw bytes

    if not verify_signature(body, request.headers.get("Saasybill-Signature", ""), SECRET):
        return "", 400

    event = request.get_json()
    # Skip repeats using event["id"], then handle event["type"] here.

    return "", 200

<?php

function verifySignature(string $body, string $header, string $secret, int $tolerance = 300): bool
{
    $timestamp = null;
    $signatures = [];

    foreach (explode(',', $header) as $part) {
        [$key, $value] = array_pad(explode('=', trim($part), 2), 2, '');
        if ($key === 't' && ctype_digit($value)) {
            $timestamp = (int) $value;
        }
        if ($key === 'v1' && $value !== '') {
            $signatures[] = $value;
        }
    }

    if ($timestamp === null || count($signatures) === 0) {
        return false;
    }
    if (abs(time() - $timestamp) > $tolerance) {
        return false;
    }

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

    foreach ($signatures as $signature) {
        if (hash_equals($expected, $signature)) {
            return true;
        }
    }

    return false;
}

$body = file_get_contents('php://input'); // the raw body
$header = $_SERVER['HTTP_SAASYBILL_SIGNATURE'] ?? '';

if (!verifySignature($body, $header, getenv('SAASYBILL_WEBHOOK_SECRET'))) {
    http_response_code(400);
    exit;
}

$event = json_decode($body, true);
// Skip repeats using $event['id'], then handle $event['type'] here.

http_response_code(200);