Verify webhook signatures

# header: the value of BIRP-Signature; body.json: the body exactly as received.
if [ -z "$BIRP_WEBHOOK_SECRET" ]; then
  echo "invalid signature" # never check with an empty key
else
  t=$(printf '%s' "$header" | tr ',' '\n' | sed -n 's/^t=//p')
  now=$(date +%s)
  expected=$(printf '%s.' "$t" | cat - body.json | openssl dgst -sha256 -hmac "$BIRP_WEBHOOK_SECRET" | sed 's/^.*= //')
  if [ $((now - ${t:-0})) -le 300 ] && [ $((${t:-0} - now)) -le 300 ] && printf '%s' "$header" | tr ',' '\n' | grep -qx "v1=$expected"; then
    echo "valid signature"
  else
    echo "invalid signature"
  fi
fi

import crypto from "node:crypto";

// rawBody: the body exactly as received (a string or a Buffer), before any JSON parsing.
export function verifyBirpSignature(rawBody, header, secret, toleranceSeconds = 300) {
  if (!secret) return false; // never check with an empty key
  const parts = String(header).split(",").map((part) => part.trim().split("="));
  const timestamp = Number(parts.find(([key]) => key === "t")?.[1]);
  const signatures = parts.filter(([key]) => key === "v1").map(([, value]) => value ?? "");
  if (!Number.isInteger(timestamp) || signatures.length === 0) return false;
  if (Math.abs(Math.floor(Date.now() / 1000) - timestamp) > toleranceSeconds) return false;
  const expected = crypto.createHmac("sha256", secret).update(`${timestamp}.`).update(rawBody).digest();
  return signatures.some((signature) => {
    const received = Buffer.from(signature, "hex");
    return received.length === expected.length && crypto.timingSafeEqual(received, expected);
  });
}

// With Express: keep the raw body for the check, parse it after.
import express from "express";

// Your own storage, such as a table or a queue, keyed by event.id.
async function storeEvent(event) {}

const app = express();
app.post("/birp/webhooks", express.raw({ type: "application/json", limit: "2mb" }), async (req, res) => {
  // No raw body (not a JSON request): nothing to check.
  if (!Buffer.isBuffer(req.body) || !verifyBirpSignature(req.body, req.get("BIRP-Signature") ?? "", process.env.BIRP_WEBHOOK_SECRET)) {
    return res.sendStatus(400);
  }
  try {
    await storeEvent(JSON.parse(req.body));
  } catch {
    return res.sendStatus(500); // not stored: BIRP tries again later
  }
  res.sendStatus(200);
  // Handle the stored event in the background, once per event.id.
});

<?php
// $rawBody: the body exactly as received, before json_decode.
function verify_birp_signature(string $rawBody, string $header, string $secret, int $toleranceSeconds = 300): bool
{
    if ($secret === '') {
        return false; // never check with an empty key
    }
    $timestamp = null;
    $signatures = [];
    foreach (explode(',', $header) as $part) {
        $pair = explode('=', trim($part), 2);
        if (count($pair) !== 2) {
            continue;
        }
        if ($pair[0] === 't' && ctype_digit($pair[1])) {
            $timestamp = (int) $pair[1];
        } elseif ($pair[0] === 'v1') {
            $signatures[] = $pair[1];
        }
    }
    if ($timestamp === null || $signatures === [] || abs(time() - $timestamp) > $toleranceSeconds) {
        return false;
    }
    $expected = hash_hmac('sha256', $timestamp . '.' . $rawBody, $secret);
    foreach ($signatures as $signature) {
        if (hash_equals($expected, $signature)) {
            return true;
        }
    }
    return false;
}

// Your own storage, such as a table or a queue, keyed by $event['id'].
function store_event(array $event): void
{
}

$rawBody = file_get_contents('php://input');
$header = $_SERVER['HTTP_BIRP_SIGNATURE'] ?? '';
if (!verify_birp_signature($rawBody, $header, (string) getenv('BIRP_WEBHOOK_SECRET'))) {
    http_response_code(400);
    exit;
}
store_event(json_decode($rawBody, true));
http_response_code(200);
// Handle the stored event in the background, once per event id.

import hashlib
import hmac
import time


# raw_body: the body exactly as received, as bytes, before any JSON parsing.
def verify_birp_signature(raw_body: bytes, header: str, secret: str, tolerance_seconds: int = 300) -> bool:
    if not secret:
        return False  # never check with an empty key
    pairs = [part.strip().split("=", 1) for part in header.split(",") if "=" in part]
    timestamps = [value for key, value in pairs if key == "t" and value.isdigit()]
    signatures = [value for key, value in pairs if key == "v1"]
    if not timestamps or not signatures:
        return False
    timestamp = int(timestamps[0])
    if abs(time.time() - timestamp) > tolerance_seconds:
        return False
    signed = str(timestamp).encode() + b"." + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest().encode()
    return any(hmac.compare_digest(expected, signature.encode()) for signature in signatures)


# With Flask: request.get_data() is the raw body.
import os

from flask import Flask, abort, request

app = Flask(__name__)


def store_event(event):
    pass  # your own storage, such as a table or a queue, keyed by event["id"]


@app.post("/birp/webhooks")
def birp_webhook():
    header = request.headers.get("BIRP-Signature", "")
    if not verify_birp_signature(request.get_data(), header, os.environ.get("BIRP_WEBHOOK_SECRET", "")):
        abort(400)
    store_event(request.get_json())
    # Handle the stored event in the background, once per event["id"].
    return "", 200