Verify webhook signatures

import crypto from 'node:crypto';
 
// rawBody: the request body exactly as received (Buffer or string), before JSON parsing.
export function isValidMaxmoveWebhook(headers, rawBody, secret) {
  const timestamp = headers['x-event-timestamp'];
  if (Math.abs(Date.now() - Date.parse(timestamp)) > 5 * 60 * 1000) return false;
 
  const bodyHash = crypto.createHash('sha256').update(rawBody).digest('hex');
  const base = [headers['x-event-id'], headers['x-delivery-id'], timestamp, bodyHash].join('.');
  const expected = 'sha256=' + crypto.createHmac('sha256', secret).update(base).digest('hex');
 
  const received = String(headers['x-signature'] ?? '');
  return (
    received.length === expected.length &&
    crypto.timingSafeEqual(Buffer.from(received), Buffer.from(expected))
  );
}

import hashlib
import hmac
from datetime import datetime, timezone
 
# raw_body: the request body exactly as received (bytes), before JSON parsing.
# headers: a case-insensitive mapping, for example Flask's request.headers.
def is_valid_maxmove_webhook(headers, raw_body: bytes, secret: str) -> bool:
    timestamp = headers.get("x-event-timestamp", "")
    try:
        sent_at = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
    except ValueError:
        return False
    if abs((datetime.now(timezone.utc) - sent_at).total_seconds()) > 5 * 60:
        return False
 
    body_hash = hashlib.sha256(raw_body).hexdigest()
    base = ".".join(
        [headers.get("x-event-id", ""), headers.get("x-delivery-id", ""), timestamp, body_hash]
    )
    expected = "sha256=" + hmac.new(secret.encode(), base.encode(), hashlib.sha256).hexdigest()
    return hmac.compare_digest(headers.get("x-signature", ""), expected)

<?php
// $rawBody: the request body exactly as received, for example file_get_contents('php://input').
// $headers: request headers, for example getallheaders().
function isValidMaxmoveWebhook(array $headers, string $rawBody, string $secret): bool
{
    $headers = array_change_key_case($headers, CASE_LOWER);
    $timestamp = $headers['x-event-timestamp'] ?? '';
    try {
        $sentAt = new DateTimeImmutable($timestamp);
    } catch (Exception $e) {
        return false;
    }
    if ($timestamp === '' || abs(time() - $sentAt->getTimestamp()) > 5 * 60) {
        return false;
    }
 
    $bodyHash = hash('sha256', $rawBody);
    $base = implode('.', [
        $headers['x-event-id'] ?? '',
        $headers['x-delivery-id'] ?? '',
        $timestamp,
        $bodyHash,
    ]);
    $expected = 'sha256=' . hash_hmac('sha256', $base, $secret);
 
    return hash_equals($expected, $headers['x-signature'] ?? '');
}

import express from 'express';
import { isValidMaxmoveWebhook } from './verify.js';
 
const app = express();
 
app.post('/maxmove/webhooks', express.raw({ type: 'application/json' }), (req, res) => {
  if (!isValidMaxmoveWebhook(req.headers, req.body, process.env.MAXMOVE_WEBHOOK_SECRET)) {
    return res.sendStatus(401);
  }
  const event = JSON.parse(req.body.toString('utf8'));
  res.sendStatus(200);
  // Process the event after answering, for example by putting it on a queue.
  queueEvent(event);
});

import json
import os
from flask import Flask, request
 
app = Flask(__name__)
 
@app.post("/maxmove/webhooks")
def maxmove_webhook():
    raw_body = request.get_data()
    if not is_valid_maxmove_webhook(request.headers, raw_body, os.environ["MAXMOVE_WEBHOOK_SECRET"]):
        return "", 401
    event = json.loads(raw_body)
    queue_event(event)  # Process asynchronously; answer within 5 seconds.
    return "", 200